Isis standard cb

doc/tutorial/overview.md

@@ -182,6 +182,8 @@ def my_plan(det_block_name: str, mot_block_name: str, start: float, stop: float,
 The above will show a `LiveTable` by default, any time `my_plan` is executed. The same mechanism can
 be used for example to always configure a particular scan with plots and a fit with a specific type.
 
+As this is fairly common functionality for most plans, we have created a "standard callbacks" collection which should suit the needs of most plans. This includes the ability to fit, plot, add human-readable file output and show a live table of scanned fields. See {py:obj}`ibex_bluesky_core.callbacks.ISISCallbacks` for API reference on how to use this.
+
 For more information on callbacks, see
 [bluesky callbacks documentation](https://blueskyproject.io/bluesky/main/callbacks.html).
 

manual_system_tests/dae_scan.py

@@ -8,16 +8,11 @@
 import bluesky.plans as bp
 import matplotlib
 import matplotlib.pyplot as plt
-from bluesky.callbacks import LiveFitPlot, LiveTable
-from bluesky.preprocessors import subs_decorator
 from bluesky.utils import Msg
 from ophyd_async.plan_stubs import ensure_connected
 
-from ibex_bluesky_core.callbacks.file_logger import HumanReadableFileCallback
-from ibex_bluesky_core.callbacks.fitting import LiveFit
+from ibex_bluesky_core.callbacks import ISISCallbacks
 from ibex_bluesky_core.callbacks.fitting.fitting_utils import Linear
-from ibex_bluesky_core.callbacks.fitting.livefit_logger import LiveFitLogger
-from ibex_bluesky_core.callbacks.plotting import LivePlot
 from ibex_bluesky_core.devices import get_pv_prefix
 from ibex_bluesky_core.devices.block import block_rw_rbv
 from ibex_bluesky_core.devices.simpledae import SimpleDae
@@ -28,7 +23,6 @@
     GoodFramesNormalizer,
 )
 from ibex_bluesky_core.devices.simpledae.waiters import GoodFramesWaiter
-from ibex_bluesky_core.plan_stubs import call_qt_aware
 from ibex_bluesky_core.run_engine import get_run_engine
 
 NUM_POINTS: int = 3
@@ -73,66 +67,40 @@ def dae_scan_plan() -> Generator[Msg, None, None]:
     controller.run_number.set_name("run number")
     reducer.intensity.set_name("normalized counts")
 
-    _, ax = yield from call_qt_aware(plt.subplots)
-
-    lf = LiveFit(
-        Linear.fit(), y=reducer.intensity.name, x=block.name, yerr=reducer.intensity_stddev.name
-    )
-
     yield from ensure_connected(block, dae, force_reconnect=True)
 
-    @subs_decorator(
-        [
-            HumanReadableFileCallback(
-                [
-                    block.name,
-                    controller.run_number.name,
-                    reducer.intensity.name,
-                    reducer.det_counts.name,
-                    dae.good_frames.name,
-                ],
-                output_dir=Path("C:\\")
-                / "instrument"
-                / "var"
-                / "logs"
-                / "bluesky"
-                / "output_files",
-            ),
-            LiveFitPlot(livefit=lf, ax=ax),
-            LivePlot(
-                y=reducer.intensity.name,
-                x=block.name,
-                marker="x",
-                linestyle="none",
-                ax=ax,
-                yerr=reducer.intensity_stddev.name,
-            ),
-            LiveTable(
-                [
-                    block.name,
-                    controller.run_number.name,
-                    reducer.intensity.name,
-                    reducer.intensity_stddev.name,
-                    reducer.det_counts.name,
-                    reducer.det_counts_stddev.name,
-                    dae.good_frames.name,
-                ]
-            ),
-            LiveFitLogger(
-                lf,
-                y=reducer.intensity.name,
-                x=block.name,
-                output_dir=Path("C:\\Instrument\\Var\\logs\\bluesky\\fitting"),
-                yerr=reducer.intensity_stddev.name,
-                postfix="manual_test",
-            ),
-        ]
+    icc = ISISCallbacks(
+        x=block.name,
+        y=reducer.intensity.name,
+        yerr=reducer.intensity_stddev.name,
+        fit=Linear.fit(),
+        measured_fields=[
+            controller.run_number.name,
+            reducer.det_counts.name,
+            reducer.det_counts_stddev.name,
+            dae.good_frames.name,
+        ],
+        human_readable_file_output_dir=Path("C:\\")
+        / "instrument"
+        / "var"
+        / "logs"
+        / "bluesky"
+        / "output_files",
+        live_fit_logger_output_dir=Path("C:\\")
+        / "instrument"
+        / "var"
+        / "logs"
+        / "bluesky"
+        / "fitting",
     )
+
+    @icc
     def _inner() -> Generator[Msg, None, None]:
         yield from bps.mv(dae.number_of_periods, NUM_POINTS)  # type: ignore
         # Pyright does not understand as bluesky isn't typed yet
         yield from bp.scan([dae], block, 0, 10, num=NUM_POINTS)
-        print(lf.result.fit_report())
+        print(icc.live_fit.result.fit_report())
+        print(f"COM: {icc.peak_stats['com']}")
 
     yield from _inner()
 

src/ibex_bluesky_core/callbacks/__init__.py

@@ -1 +1,234 @@
 """Bluesky callbacks which may be attached to the RunEngine."""
+
+import logging
+import threading
+from collections.abc import Generator
+from os import PathLike
+from pathlib import Path
+from typing import Any, Callable
+
+import bluesky.preprocessors as bpp
+import matplotlib.pyplot as plt
+from bluesky.callbacks import CallbackBase, LiveFitPlot, LiveTable
+from bluesky.callbacks.fitting import PeakStats
+from bluesky.callbacks.mpl_plotting import QtAwareCallback
+from bluesky.utils import Msg, make_decorator
+from event_model import RunStart
+from matplotlib.axes import Axes
+
+from ibex_bluesky_core.callbacks.file_logger import (
+    DEFAULT_PATH as DEFAULT_PATH_HRF,
+)
+from ibex_bluesky_core.callbacks.file_logger import (
+    HumanReadableFileCallback,
+)
+from ibex_bluesky_core.callbacks.fitting import FitMethod, LiveFit
+from ibex_bluesky_core.callbacks.fitting.livefit_logger import (
+    DEFAULT_PATH as DEFAULT_PATH_LFL,
+)
+from ibex_bluesky_core.callbacks.fitting.livefit_logger import (
+    LiveFitLogger,
+)
+from ibex_bluesky_core.callbacks.plotting import LivePlot
+
+logger = logging.getLogger(__name__)
+
+# ruff: noqa: PLR0913, PLR0912, PLR0917
+
+
+class ISISCallbacks:
+    """ISIS standard callbacks for use within plans."""
+
+    def __init__(
+        self,
+        *,
+        x: str,
+        y: str,
+        yerr: str | None = None,
+        measured_fields: list[str] | None = None,
+        add_table_cb: bool = True,
+        fields_for_live_table: list[str] | None = None,
+        add_human_readable_file_cb: bool = True,
+        fields_for_hr_file: list[str] | None = None,
+        human_readable_file_output_dir: str | PathLike[str] | None = None,
+        add_plot_cb: bool = True,
+        ax: Axes | None = None,
+        fit: FitMethod | None = None,
+        show_fit_on_plot: bool = True,
+        add_peak_stats: bool = True,
+        add_live_fit_logger: bool = True,
+        live_fit_logger_output_dir: str | PathLike[str] | None = None,
+        live_fit_logger_postfix: str = "isc",
+    ) -> None:
+        """A collection of ISIS standard callbacks for use within plans.
+
+        By default, this adds:
+
+        - HumanReadableFileCallback
+
+        - LiveTable
+
+        - PeakStats
+
+        - LiveFit
+
+        - LiveFitPlot
+
+        - LivePlot
+
+        Results can be accessed from the `live_fit` and `peak_stats` properties.
+
+        This is to be used as a member and then as a decorator if results are needed ie::
+
+            def dae_scan_plan():
+                ...
+                icc = ISISCallbacks(
+                    x=block.name,
+                    y=reducer.intensity.name,
+                    yerr=reducer.intensity_stddev.name,
+                    fit=Linear.fit(),
+                    ...
+                )
+                ...
+
+                @icc
+                def _inner():
+                    yield from ...
+                    ...
+                    print(icc.live_fit.result.fit_report())
+                    print(f"COM: {icc.peak_stats['com']}")
+
+        Args:
+            x: The signal name to use for X within plots and fits.
+            y: The signal name to use for Y within plots and fits.
+            yerr: The signal name to use for the Y uncertainty within plots and fits.
+            measured_fields: the fields to use for both the live table and human-readable file.
+            add_table_cb: whether to add a table callback.
+            fields_for_live_table: the fields to measure for the live table (in addition to `measured_fields`).
+            add_human_readable_file_cb: whether to add a human-readable file callback.
+            fields_for_hr_file: the fields to measure for the human-readable file (in addition to `measured_fields`).
+            human_readable_file_output_dir: the output directory for human-readable files. can be blank and will default.
+            add_plot_cb: whether to add a plot callback.
+            ax: An optional axes object to use for plotting.
+            fit: The fit method to use when fitting.
+            show_fit_on_plot: whether to show fit on plot.
+            add_peak_stats: whether to add a peak stats callback.
+            add_live_fit_logger: whether to add a live fit logger.
+            live_fit_logger_output_dir: the output directory for live fit logger.
+            live_fit_logger_postfix: the postfix to add to live fit logger.
+        """  # noqa
+        self._subs = []
+        self._peak_stats = None
+        self._live_fit = None
+        if measured_fields is None:
+            measured_fields = []
+        if fields_for_live_table is None:
+            fields_for_live_table = []
+        if fields_for_hr_file is None:
+            fields_for_hr_file = []
+
+        measured_fields.append(x)
+        measured_fields.append(y)
+        if yerr is not None:
+            measured_fields.append(yerr)
+
+        if add_human_readable_file_cb:
+            combined_hr_fields = measured_fields + fields_for_hr_file
+            self._subs.append(
+                HumanReadableFileCallback(
+                    fields=combined_hr_fields,
+                    output_dir=Path(human_readable_file_output_dir)
+                    if human_readable_file_output_dir
+                    else DEFAULT_PATH_HRF,
+                ),
+            )
+
+        if add_table_cb:
+            combined_lt_fields = measured_fields + fields_for_live_table
+            self._subs.append(
+                LiveTable(combined_lt_fields),
+            )
+
+        if add_peak_stats:
+            self._peak_stats = PeakStats(x=x, y=y)
+            self._subs.append(self._peak_stats)
+
+        if (add_plot_cb or show_fit_on_plot) and not ax:
+            logger.debug("No axis provided, creating a new one")
+            fig, ax, exc, result = None, None, None, None
+            done_event = threading.Event()
+
+            class _Cb(QtAwareCallback):
+                def start(self, doc: RunStart) -> None:
+                    nonlocal result, exc, fig, ax
+                    try:
+                        plt.close("all")
+                        fig, ax = plt.subplots()
+                    finally:
+                        done_event.set()
+
+            cb = _Cb()
+            cb("start", {"time": 0, "uid": ""})
+            done_event.wait(10.0)
+
+        if fit is not None:
+            self._live_fit = LiveFit(fit, y=y, x=x, yerr=yerr)
+            if show_fit_on_plot:
+                self._subs.append(LiveFitPlot(livefit=self._live_fit, ax=ax))
+            else:
+                self._subs.append(self._live_fit)
+            if add_live_fit_logger:
+                self._subs.append(
+                    LiveFitLogger(
+                        livefit=self._live_fit,
+                        x=x,
+                        y=y,
+                        yerr=yerr,
+                        output_dir=DEFAULT_PATH_LFL
+                        if live_fit_logger_output_dir is None
+                        else live_fit_logger_output_dir,
+                        postfix=live_fit_logger_postfix,
+                    )
+                )
+
+        if add_plot_cb or show_fit_on_plot:
+            self._subs.append(
+                LivePlot(
+                    y=y,
+                    x=x,
+                    marker="x",
+                    linestyle="none",
+                    ax=ax,
+                    yerr=yerr,
+                )
+            )
+
+    @property
+    def live_fit(self) -> LiveFit:
+        """The live fit object containing fitting results."""
+        if self._live_fit is None:
+            raise ValueError("live_fit was not added as a callback.")
+        return self._live_fit
+
+    @property
+    def peak_stats(self) -> PeakStats:
+        """The peak stats object containing statistics ie. centre of mass."""
+        if self._peak_stats is None:
+            raise ValueError("peak stats was not added as a callback.")
+        return self._peak_stats
+
+    @property
+    def subs(self) -> list[CallbackBase]:
+        """The list of subscribed callbacks."""
+        return self._subs
+
+    def _icbc_wrapper(self, plan: Generator[Msg, None, None]) -> Generator[Msg, None, None]:
+        @bpp.subs_decorator(self.subs)
+        def _inner() -> Generator[Msg, None, None]:
+            return (yield from plan)
+
+        return (yield from _inner())
+
+    def __call__(self, f: Callable[..., Any]) -> Callable[..., Any]:
+        """Make a decorator to wrap the plan and subscribe to all callbacks."""
+        return make_decorator(self._icbc_wrapper)()(f)

src/ibex_bluesky_core/callbacks/fitting/livefit_logger.py

@@ -37,7 +37,7 @@ def __init__(
         y: str,
         x: str,
         postfix: str,
-        output_dir: Path = DEFAULT_PATH,
+        output_dir: str | os.PathLike[str] = DEFAULT_PATH,
         yerr: str | None = None,
     ) -> None:
         """Initialise LiveFitLogger callback.
@@ -55,7 +55,7 @@ def __init__(
         super().__init__()
         self.livefit = livefit
         self.postfix = postfix
-        self.output_dir = output_dir
+        self.output_dir = Path(output_dir)
         self.current_start_document: Optional[str] = None
 
         self.x = x