Add configurable SimpleDae class

Author: Tom-Willemsen

doc/simpledae/controllers.md

@@ -0,0 +1,46 @@
+# SimpleDae controllers
+
+The `Controller` class is responsible for starting and stopping acquisitions, in a generic
+way.
+
+# Predefined controllers
+
+Some controllers have been predefined in the 
+`ibex_bluesky_core.devices.simpledae.controllers` module.
+
+## RunPerPointController
+
+This controller starts and stops a new DAE run for each scan point. It can be configured to 
+either end runs or abort them on completion.
+
+This controller causes the following signals to be published by `SimpleDae`:
+
+- `controller.run_number` - The run number into which data was collected. Only published 
+  if runs are being saved.
+
+## PeriodPerPointController
+
+This controller begins a single DAE run at the start of a scan, and then counts into a new
+DAE period for each individual scan point.
+
+The DAE must be configured with enough periods in advance. This is possible to do from a
+plan as follows:
+
+```python
+import bluesky.plan_stubs as bps
+import bluesky.plans as bp
+from ibex_bluesky_core.devices.simpledae import SimpleDae
+from ibex_bluesky_core.devices.block import BlockRw
+
+
+def plan():
+    dae: SimpleDae = ...
+    block: BlockRw = ...
+    num_points = 20
+    yield from bps.mv(dae.number_of_periods, num_points)
+    yield from bp.scan([dae], block, 0, 10, num=num_points)
+```
+
+The controller causes the following signals to be published by `SimpleDae`:
+
+- `simpledae.period_num` - the period number into which this scan point was counted.

doc/simpledae/reducers.md

@@ -0,0 +1,36 @@
+# Reducers
+
+A `Reducer` for a `SimpleDae` is responsible for publishing any data derived from the raw
+DAE signals. For example, normalizing intensities are implemented as a reducer.
+
+A reducer may produce any number of reduced signals.
+
+## GoodFramesNormalizer
+
+This normalizer sums a set of user-defined detector spectra, and then divides by the number
+of good frames.
+
+Published signals:
+- `simpledae.good_frames` - the number of good frames reported by the DAE
+- `reducer.det_counts` - summed detector counts for all of the user-provided spectra
+- `reducer.intensity` - normalized intensity (`det_counts / good_frames`)
+
+## PeriodGoodFramesNormalizer
+
+Equivalent to the `GoodFramesNormalizer` above, but uses good frames only from the current
+period. This should be used if a controller which counts into multiple periods is being used.
+
+Published signals:
+- `simpledae.period.good_frames` - the number of good frames reported by the DAE
+- `reducer.det_counts` - summed detector counts for all of the user-provided spectra
+- `reducer.intensity` - normalized intensity (`det_counts / good_frames`)
+
+## DetectorMonitorNormalizer
+
+This normalizer sums a set of user-defined detector spectra, and then divides by the sum
+of a set of user-defined monitor spectra.
+
+Published signals:
+- `reducer.det_counts` - summed detector counts for the user-provided detector spectra
+- `reducer.mon_counts` - summed monitor counts for the user-provided monitor spectra
+- `reducer.intensity` - normalized intensity (`det_counts / mon_counts`)

doc/simpledae/simpledae.md

@@ -0,0 +1,51 @@
+# Simple Dae
+
+The `SimpleDae` class is designed to be a configurable DAE object, which will cover the
+majority of DAE use-cases within bluesky.
+
+This class uses several objects to configure its behaviour:
+- The `Controller` is responsible for beginning and ending acquisitions.
+- The `Waiter` is responsible for waiting for an acquisition to be "complete".
+- The `Reducer` is responsible for publishing data from an acquisition that has 
+  just been completed.
+
+This means that `SimpleDae` is generic enough to cope with most typical DAE use-casess, for
+example running using either one DAE run per scan point, or one DAE period per scan point.
+
+For complex use-cases, particularly those where the DAE may need to start and stop multiple 
+acquisitions per scan point (e.g. polarization measurements), `SimpleDae` is unlikely to be 
+suitable; instead the `Dae` class should be subclassed directly to allow for finer control.
+
+## Mapping to bluesky device model
+
+### Start of scan (`stage`)
+
+`SimpleDae` will call `controller.setup()` to allow any pre-scan setup to be done.
+
+For example, this is where the period-per-point controller object will begin a DAE run.
+
+### Each scan point (`trigger`)
+
+`SimpleDae` will call:
+- `controller.start_counting()` to begin counting for a single scan point.
+- `waiter.wait()` to wait for that acquisition to complete
+- `controller.stop_counting()` to finish counting for a single scan point.
+- `reducer.reduce_data()` to do any necessary post-processing on 
+  the raw DAE data (e.g. normalization)
+
+### Each scan point (`read`)
+
+Any signals marked as "interesting" by the controller, reducer or waiter will be published
+in the top-level documents published when `read()`ing the `SimpleDae` object.
+
+These may correspond to EPICS signals directly from the DAE (e.g. good frames), or may be 
+soft signals derived at runtime (e.g. normalized intensity).
+
+This means that the `SimpleDae` object is suitable for use as a detector in most bluesky
+plans, and will make an appropriate set of data available in the emitted documents.
+
+### End of scan (`unstage`)
+
+`SimpleDae` will call `controller.teardown()` to allow any post-scan teardown to be done.
+
+For example, this is where the period-per-point controller object will end a DAE run.

doc/simpledae/waiters.md

@@ -0,0 +1,41 @@
+# Waiters
+
+A `waiter` defines an arbitrary strategy for how long to count at each point.
+
+Some waiters may be very simple, such as waiting for a fixed amount of time or for a number
+of good frames or microamp-hours. However, it is also possible to define much more 
+sophisticated waiters, for example waiting until sufficient statistics have been collected.
+
+## GoodUahWaiter
+
+Waits for a user-specified number of microamp-hours.
+
+Published signals:
+- `simpledae.good_uah` - actual good uAh for this run.
+
+## GoodFramesWaiter
+
+Waits for a user-specified number of good frames (in total for the entire run)
+
+Published signals:
+- `simpledae.good_frames` - actual good frames for this run.
+
+## GoodFramesWaiter
+
+Waits for a user-specified number of good frames (in the current period)
+
+Published signals:
+- `simpledae.period.good_frames` - actual period good frames for this run.
+
+## MEventsWaiter
+
+Waits for a user-specified number of millions of events
+
+Published signals:
+- `simpledae.m_events` - actual period good frames for this run.
+
+## TimeWaiter
+
+Waits for a user-specified time duration, irrespective of DAE state.
+
+Does not publish any additional signals.

pyproject.toml

@@ -44,7 +44,8 @@ dependencies = [
   "bluesky",
   "ophyd-async[ca]",
   "matplotlib",
-  "numpy"
+  "numpy",
+  "scipp",
 ]
 
 [project.optional-dependencies]
@@ -76,6 +77,12 @@ omit = [
 
 [tool.coverage.report]
 fail_under = 100
+exclude_lines = [
+  "pragma: no cover",
+  "if TYPE_CHECKING:",
+  "if typing.TYPE_CHECKING:",
+  "@abstractmethod",
+]
 
 [tool.coverage.html]
 directory = "coverage_html_report"

src/ibex_bluesky_core/demo_plan.py

@@ -3,17 +3,25 @@
 from typing import Generator
 
 import bluesky.plan_stubs as bps
+import bluesky.plans as bp
 import matplotlib
 import matplotlib.pyplot as plt
 from bluesky.callbacks import LiveTable
-from bluesky.preprocessors import run_decorator, subs_decorator
+from bluesky.preprocessors import subs_decorator
 from bluesky.utils import Msg
 from ophyd_async.plan_stubs import ensure_connected
 
 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.dae.dae import Dae
+from ibex_bluesky_core.devices.simpledae import SimpleDae
+from ibex_bluesky_core.devices.simpledae.controllers import (
+    RunPerPointController,
+)
+from ibex_bluesky_core.devices.simpledae.reducers import (
+    GoodFramesNormalizer,
+)
+from ibex_bluesky_core.devices.simpledae.waiters import GoodFramesWaiter
 from ibex_bluesky_core.run_engine import get_run_engine
 
 __all__ = ["demo_plan"]
@@ -23,27 +31,45 @@ def demo_plan() -> Generator[Msg, None, None]:
     """Demonstration plan which moves a block and reads the DAE."""
     prefix = get_pv_prefix()
     block = block_rw_rbv(float, "mot")
-    dae = Dae(prefix)
+
+    controller = RunPerPointController(save_run=True)
+    waiter = GoodFramesWaiter(500)
+    reducer = GoodFramesNormalizer(
+        prefix=prefix,
+        detector_spectra=[i for i in range(1, 100)],
+    )
+
+    dae = SimpleDae(
+        prefix=prefix,
+        controller=controller,
+        waiter=waiter,
+        reducer=reducer,
+    )
+
+    # Demo giving some signals more user-friendly names
+    controller.run_number.set_name("run number")
+    reducer.intensity.set_name("normalized counts")
 
     yield from ensure_connected(block, dae, force_reconnect=True)
 
     @subs_decorator(
         [
-            LivePlot(y="DAE-good_uah", x=block.name, marker="x", linestyle="none"),
-            LiveTable([block.name, "DAE-good_uah"]),
+            LivePlot(y=reducer.intensity.name, x=block.name, marker="x", linestyle="none"),
+            LiveTable(
+                [
+                    block.name,
+                    controller.run_number.name,
+                    reducer.intensity.name,
+                    reducer.det_counts.name,
+                    dae.good_frames.name,
+                ]
+            ),
         ]
     )
-    @run_decorator(md={})
     def _inner() -> Generator[Msg, None, None]:
-        # Acquisition showing arbitrary DAE control to support complex use-cases.
-        yield from bps.abs_set(block, 2.0, wait=True)
-        yield from bps.trigger(dae.controls.begin_run, wait=True)
-        yield from bps.sleep(5)  # ... some complicated logic ...
-        yield from bps.trigger(dae.controls.end_run, wait=True)
-        yield from bps.create()  # Create a bundle of readings
-        yield from bps.read(block)
-        yield from bps.read(dae.good_uah)
-        yield from bps.save()
+        num_points = 20
+        yield from bps.mv(dae.number_of_periods, num_points)
+        yield from bp.scan([dae], block, 0, 10, num=num_points)
 
     yield from _inner()
 

src/ibex_bluesky_core/devices/dae/dae.py

@@ -56,8 +56,16 @@ def __init__(self, prefix: str, name: str = "DAE") -> None:
         self.good_frames: SignalR[int] = epics_signal_r(int, f"{dae_prefix}GOODFRAMES")
         self.raw_frames: SignalR[int] = epics_signal_r(int, f"{dae_prefix}RAWFRAMES")
         self.total_counts: SignalR[int] = epics_signal_r(int, f"{dae_prefix}TOTALCOUNTS")
-        self.run_number: SignalR[int] = epics_signal_r(int, f"{dae_prefix}IRUNNUMBER")
-        self.run_number_str: SignalR[str] = epics_signal_r(str, f"{dae_prefix}RUNNUMBER")
+
+        # Beware that this increments just after a run is ended. So it is generally not correct to
+        # read this just after a DAE run has been ended().
+        self.current_or_next_run_number: SignalR[int] = epics_signal_r(
+            int, f"{dae_prefix}IRUNNUMBER"
+        )
+        self.current_or_next_run_number_str: SignalR[str] = epics_signal_r(
+            str, f"{dae_prefix}RUNNUMBER"
+        )
+
         self.cycle_number: SignalR[str] = epics_signal_r(str, f"{dae_prefix}ISISCYCLE")
         self.inst_name: SignalR[str] = epics_signal_r(str, f"{dae_prefix}INSTNAME")
         self.run_start_time: SignalR[str] = epics_signal_r(str, f"{dae_prefix}STARTTIME")

src/ibex_bluesky_core/devices/dae/dae_controls.py

@@ -43,4 +43,4 @@ def __init__(self, dae_prefix: str, name: str = "") -> None:
     @AsyncStatus.wrap
     async def set(self, value: BeginRunExBits) -> None:
         """Start a run with the specified bits - See BeginRunExBits."""
-        await self._raw_begin_run_ex.set(value, wait=True)
+        await self._raw_begin_run_ex.set(value, wait=True, timeout=None)

src/ibex_bluesky_core/devices/dae/dae_spectra.py

@@ -2,6 +2,8 @@
 
 import asyncio
 
+import scipp as sc
+from event_model.documents.event_descriptor import DataKey
 from numpy import float32
 from numpy.typing import NDArray
 from ophyd_async.core import SignalR, StandardReadable
@@ -22,6 +24,15 @@ def __init__(self, dae_prefix: str, *, spectra: int, period: int, name: str = ""
             int, f"{dae_prefix}SPEC:{period}:{spectra}:X.NORD"
         )
 
+        # x-axis; time-of-flight.
+        # These are bin-edge coordinates, with a size one more than the corresponding data.
+        self.tof_edges: SignalR[NDArray[float32]] = epics_signal_r(
+            NDArray[float32], f"{dae_prefix}SPEC:{period}:{spectra}:XE"
+        )
+        self.tof_edges_size: SignalR[int] = epics_signal_r(
+            int, f"{dae_prefix}SPEC:{period}:{spectra}:XE.NORD"
+        )
+
         # y-axis; counts / tof
         # This is the number of counts in a ToF bin, normalized by the width of
         # that ToF bin.
@@ -57,10 +68,46 @@ async def read_tof(self) -> NDArray[float32]:
         """Read a correctly-sized time-of-flight (x) array representing bin centres."""
         return await self._read_sized(self.tof, self.tof_size)
 
+    async def read_tof_edges(self) -> NDArray[float32]:
+        """Read a correctly-sized time-of-flight (x) array representing bin edges."""
+        return await self._read_sized(self.tof_edges, self.tof_edges_size)
+
     async def read_counts(self) -> NDArray[float32]:
         """Read a correctly-sized array of counts."""
         return await self._read_sized(self.counts, self.counts_size)
 
     async def read_counts_per_time(self) -> NDArray[float32]:
         """Read a correctly-sized array of counts divided by bin width."""
         return await self._read_sized(self.counts_per_time, self.counts_per_time_size)
+
+    async def read_spectrum_dataarray(self) -> sc.DataArray:
+        """Get a scipp DataArray containing the current data from this spectrum.
+
+        Variances are set to the counts - i.e. the standard deviation is sqrt(N), which is typical
+        for counts data.
+
+        Data is returned along dimension "tof", which has bin-edge coordinates and units set from
+        the units of the underlying PVs.
+        """
+        tof_edges, tof_edges_descriptor, counts = await asyncio.gather(
+            self.read_tof_edges(),
+            self.tof_edges.describe(),
+            self.read_counts(),
+        )
+
+        if tof_edges.size != counts.size + 1:
+            raise ValueError(
+                "Time-of-flight edges must have size one more than the data. "
+                "You may be trying to read too many time channels. "
+                f"Edges size was {tof_edges.size}, counts size was {counts.size}."
+            )
+
+        datakey: DataKey = tof_edges_descriptor[self.tof_edges.name]
+        unit = datakey.get("units", None)
+        if unit is None:
+            raise ValueError("Could not determine engineering units of tof edges.")
+
+        return sc.DataArray(
+            data=sc.Variable(dims=["tof"], values=counts, variances=counts, unit=sc.units.counts),
+            coords={"tof": sc.array(dims=["tof"], values=tof_edges, unit=sc.Unit(unit))},
+        )