Merge branch 'main' into 551_use_pvi_information

Author: shihab-dls

docs/tutorials.md

@@ -10,4 +10,5 @@ tutorials/installation
 tutorials/using-devices
 tutorials/implementing-devices
 tutorials/writing-tests-for-devices
+tutorials/implementing-detectors
 ```

docs/tutorials/implementing-detectors.md

@@ -0,0 +1,211 @@
+# Implementing File Writing Detectors
+
+In [](./implementing-devices.md) we learned how to create Devices that talk to a control system to implement a particular behavior in bluesky plans. This behavior was based around the verbs from the [](#bluesky.protocols.Movable) and [](#bluesky.protocols.Readable) protocols, allowing us to use these Devices in a typical scan: moving them to a position, then acquiring data via the control system. We will now explore the [](#bluesky.protocols.WritesExternalAssets) protocol, and how it would be implemented for a File Writing Detector.
+
+## Run the demo
+
+We will return to our [](#ophyd_async.sim) devices we saw in [](./using-devices.md) for this tutorial, and dig a little deeper into what [Event Model Documents](inv:event-model#data_model) they produce. Let's run up our ipython shell again:
+
+```
+$ ipython -i -m ophyd_async.sim
+Python 3.11.11 (main, Dec  4 2024, 20:38:25) [GCC 12.2.0]
+Type 'copyright', 'credits' or 'license' for more information
+IPython 8.30.0 -- An enhanced Interactive Python. Type '?' for help.
+
+In [1]: 
+```
+
+## Run a grid scan and investigate the documents
+
+Now let's run a grid scan on the point detector, and pass a callback to the RunEngine so it prints the documents that are emitted:
+```{eval-rst}
+.. ipython:: python
+    :suppress:
+
+    from ophyd_async.sim.__main__ import *
+    # Make the moves faster so docs build don't take too long
+    RE(bps.mv(stage.x.velocity, 1000, stage.y.velocity, 1000))
+
+.. ipython:: python
+  
+    RE(bp.grid_scan([pdet], stage.x, 1, 2, 2, stage.y, 2, 3, 2), print)
+```
+We see a series of documents being emitted:
+- A [](#event_model.RunStart) document that tells us a scan is starting and what sort of scan it is, along with the names of the motors that will be moved.
+- An [](#event_model.EventDescriptor) document that tells us that the motor readbacks and detector channels will be all be read together in a single stream. It is used to make the column headings, but it contains more metadata about the Devices too, like their configuration.
+- For each point in the scan:
+  - An [](#event_model.Event) document, containing the motor readbacks and detector channels with their timestamps. It is used to make each row of the table.
+- A [](#event_model.RunStop) document that tells us the scan has stopped, and gives us its status.
+
+Now let's try the same thing, but this time with the blob detector:
+
+```{eval-rst}
+.. ipython:: python
+    :okwarning:
+    
+    RE(bp.grid_scan([bdet], stage.x, 1, 2, 2, stage.y, 2, 3, 2), print)
+```
+This time we see some different documents:
+- The same [](#event_model.RunStart) document
+- A similar [](#event_model.EventDescriptor) document, but with `'external': 'STREAM:'` on the detector column headings.
+- A couple of [](#event_model.StreamResource) documents for each of those detector column headings giving an HDF file name and dataset within it where data will be written.
+- For each point in the scan:
+  - A couple of [](#event_model.StreamDatum) documents with a range of indices that have been written to an HDF dataset referenced in the StreamResource document.
+  - An [](#event_model.Event) document, containing the motor readbacks and timestamps. It is used to make each row of the table.
+- The same [](#event_model.RunStop) document
+
+And we can run the plan with both detectors to see a document stream that combines both the previous example:
+
+```{eval-rst}
+.. ipython:: python
+    :okwarning:
+    
+    RE(bp.grid_scan([bdet, pdet], stage.x, 1, 2, 2, stage.y, 2, 3, 2), print)
+```
+
+## Simplify the plan to just use the detector
+
+The above examples show what happens if you `trigger()` a detector at each point of a scan, in this case taking a single frame each time. Let's write our own simple plan that only triggers and reads from the detector, using the utility [`bps` (`bluesky.plan_stubs`)](inv:bluesky#stub_plans) and [`bpp` (`bluesky.preprocessors`)](inv:bluesky#preprocessors):
+
+```{eval-rst}
+.. ipython:: python
+
+    @bpp.stage_decorator([bdet])
+    @bpp.run_decorator()
+    def my_count_plan():
+        for i in range(2):
+            yield from bps.trigger_and_read([bdet])
+
+.. ipython:: python
+    :okwarning:
+
+    RE(my_count_plan(), print)
+```
+
+Here we see the same sort of documents as above, but with only detector information in it.
+
+Note that on each trigger, only a single image is taken, at the default exposure of `0.1s`. If we would like a different exposure time, we can specify with a [](#TriggerInfo):
+
+```{eval-rst}
+.. ipython:: python
+
+    from ophyd_async.core import TriggerInfo
+  
+    @bpp.stage_decorator([bdet])
+    @bpp.run_decorator()
+    def my_count_plan_with_prepare():
+        yield from bps.prepare(bdet, TriggerInfo(livetime=0.001), wait=True)
+        for i in range(2):
+            yield from bps.trigger_and_read([bdet])
+
+.. ipython:: python
+    :okwarning:
+
+    RE(my_count_plan_with_prepare(), print)
+```
+
+This also moves the work of setting up the detector from the first call of `trigger()` to the `prepare()` call. We can also move the creation of the descriptor earlier, so there is no extra work to do on the first call to `trigger()`:
+
+```{eval-rst}
+.. ipython:: python
+
+    from ophyd_async.core import TriggerInfo
+  
+    @bpp.stage_decorator([bdet])
+    @bpp.run_decorator()
+    def my_count_plan_with_prepare():
+        yield from bps.prepare(bdet, TriggerInfo(), wait=True)
+        yield from bps.declare_stream(bdet, name="primary")
+        for i in range(2):
+            yield from bps.trigger_and_read([bdet])
+
+.. ipython:: python
+    :okwarning:
+
+    RE(my_count_plan_with_prepare(), print)
+```
+
+## Run a fly scan and investigate the documents
+
+The above demonstrates the detector portion of a step scan, letting the things you want to scan settle before taking data from the detector, and doing this at every point of the scan. Our filewriting detector also supports the ability to fly scan it, taking data while you are scanning other things. To do this, it implements the [](#bluesky.protocols.Flyable) protocol, which allows us to `kickoff()` a series of images, then wait until it is `complete()`:
+
+```{eval-rst}
+.. ipython:: python
+  
+    @bpp.stage_decorator([bdet])
+    @bpp.run_decorator()
+    def fly_plan():
+        yield from bps.prepare(bdet, TriggerInfo(number_of_triggers=7), wait=True)
+        yield from bps.declare_stream(bdet, name="primary")
+        yield from bps.kickoff(bdet, wait=True)
+        yield from bps.collect_while_completing(flyers=[bdet], dets=[bdet], flush_period=0.5)
+
+.. ipython:: python
+    :okwarning:
+
+    RE(fly_plan(), print)
+```
+
+As before, we see the start, descriptor, and pair of stream resources, but this time we don't see any event documents. Also, even though we asked for 7 frames from each of the 2 streams, we only got 2 stream datums for each stream. 
+
+What is happening is that instead of triggering, waiting, and publishing a single frame, we are setting up the detector to take 7 frames without stopping, then at the `flush_period` of 0.5s emitting a stream datum with the frames that have been captured. If we inspect the stream datum documents for each stream we see that:
+- The first has `'indices': {'start': 0, 'stop': 4}`
+- The second has `'indices': {'start': 4, 'stop': 7}`
+
+This behavior allows us to scale up the framerate of the detector without scaling up the number of documents emitted: whether the detector goes at 10Hz or 10MHz it still only emits one stream datum per stream per flush period, just with different numbers in the `indices` field.
+
+## Look at the Device implementations
+
+Now we'll have a look at the code to see how we implement one of these detectors:
+
+### `SimBlobDetector`
+
+```{literalinclude} ../../src/ophyd_async/sim/_blob_detector.py
+:language: python
+```
+
+It derives from [](#StandardDetector) which is a utility baseclass that implements the protocols we have mentioned so far in this tutorial. It uses a pair of logic classes to provide behavior for each protocol verb:
+- [](#DetectorController) to setup the exposure and trigger mode of the detector, arm it, and wait for it to complete
+- [](#DetectorWriter) to tell the detector to open a file, describe the datasets it will write, and wait for it to have written a given number of frames
+
+In this case, we have a Controller and Writer class written just for this simulation, both taking a reference to the pattern generator that provides methods for both detector control and file writing. In other cases the detector control and filewriting may be handled by different sub-devices that talk to different parts of the control system. The job of the top level detector class is to take the arguments that the controller and writer need and distribute them, passing the instances to the superclass init.
+
+Now let's look at the underlying classes that define the detector behavior:
+
+### `BlobDetectorControl`
+
+First we have `BlobDetectorController`, a [](#DetectorController) subclass:
+
+```{literalinclude} ../../src/ophyd_async/sim/_blob_detector_controller.py
+:language: python
+```
+
+It's job is to control the acquisition process on the detector, starting and stopping the collection of data:
+- `prepare()` takes a [](#TriggerInfo) which details everything the detector needs to know about the upcoming acquisition. In this case we just store it for later use.
+- `arm()` starts the acquisition process that has been prepared. In this case we create a background task that will write our simulation images to file.
+- `wait_for_idle()` waits for that acquisition process to be complete.
+- `disarm()` interrupts the acquisition process, and then waits for it to complete.
+
+### `BlobDetectorWriter`
+
+Then we have `BlobDetectorWriter`, a [](#DetectorWriter) subclass:
+
+```{literalinclude} ../../src/ophyd_async/sim/_blob_detector_writer.py
+:language: python
+```
+
+Its job is to control the file writing process on the detector, which may or may not be coupled to the acquisition process:
+- `open()` tells the detector to open a file, and returns information about the datasets that it will write
+- `hints` gives a list of the datasets that are interesting to plot
+- `get_indices_written()` returns the last index written
+- `observe_indices_written()` repeatedly yields the last index written then waits for the next one to be written
+- `collect_stream_docs()` uses the [](#HDFDocumentComposer) to publish a document per dataset that says which frames have been written since the last call
+- `close()` tells the detector to close the file
+
+## Conclusion
+
+We have seen how to make a [](#StandardDetector) and how the [](#DetectorController) and [](#DetectorWriter) allow us to customise its behavior.
+
+```{seealso}
+[](../how-to/implement-ad-detector.md) for writing an implementation of `StandardDetector` for an EPICS areaDetector
+```

pyproject.toml

@@ -187,7 +187,8 @@ type = "forbidden"
 forbidden_modules = ["ophyd_async.testing", "ophyd_async.sim"]
 source_modules = [
     "ophyd_async.plan_stubs",
-    "ophyd_async.fastcs",
-    "ophyd_async.epics",
-    "ophyd_async.tango",
+    "ophyd_async.fast.*",
+    "ophyd_async.epics.*",
+    "ophyd_async.tango.*",
 ]
+ignore_imports = ["ophyd_async.tango.testing.* -> ophyd_async.testing"]

src/ophyd_async/core/__init__.py

@@ -35,6 +35,7 @@
 )
 from ._settings import Settings, SettingsProvider
 from ._signal import (
+    Ignore,
     Signal,
     SignalConnector,
     SignalR,
@@ -48,6 +49,7 @@
     soft_signal_r_and_setter,
     soft_signal_rw,
     wait_for_value,
+    walk_config_signals,
     walk_rw_signals,
 )
 from ._signal_backend import (
@@ -130,6 +132,7 @@
     "set_and_wait_for_value",
     "set_and_wait_for_other_value",
     "walk_rw_signals",
+    "walk_config_signals",
     # Readable
     "StandardReadable",
     "StandardReadableFormat",
@@ -179,4 +182,5 @@
     # Back compat - delete before 1.0
     "ConfigSignal",
     "HintedSignal",
+    "Ignore",
 ]

src/ophyd_async/core/_detector.py

@@ -51,7 +51,7 @@ class DetectorTrigger(Enum):
 class TriggerInfo(BaseModel):
     """Minimal set of information required to setup triggering on a detector."""
 
-    number_of_triggers: NonNegativeInt | list[NonNegativeInt]
+    number_of_triggers: NonNegativeInt | list[NonNegativeInt] = Field(default=1)
     """Number of triggers that will be sent, (0 means infinite).
 
     Can be:
@@ -135,16 +135,23 @@ async def open(self, multiplier: int = 1) -> dict[str, DataKey]:
         :return: Output for ``describe()``
         """
 
-    @abstractmethod
-    def observe_indices_written(
-        self, timeout=DEFAULT_TIMEOUT
-    ) -> AsyncGenerator[int, None]:
-        """Yield the index of each frame (or equivalent data point) as it is written."""
+    @property
+    def hints(self) -> Hints:
+        """The hints to be used for the detector."""
+        return {}
 
     @abstractmethod
     async def get_indices_written(self) -> int:
         """Get the number of indices written."""
 
+    # Note: this method is really async, but if we make it async here then we
+    # need to give it a body with a redundant yield statement, which is a bit
+    # awkward. So we just leave it as a regular method and let the user
+    # implement it as async.
+    @abstractmethod
+    def observe_indices_written(self, timeout: float) -> AsyncGenerator[int, None]:
+        """Yield the index of each frame (or equivalent data point) as it is written."""
+
     @abstractmethod
     def collect_stream_docs(self, indices_written: int) -> AsyncIterator[StreamAsset]:
         """Create Stream docs up to given number written."""
@@ -153,11 +160,6 @@ def collect_stream_docs(self, indices_written: int) -> AsyncIterator[StreamAsset
     async def close(self) -> None:
         """Close writer, blocks until I/O is complete."""
 
-    @property
-    def hints(self) -> Hints:
-        """The hints to be used for the detector."""
-        return {}
-
 
 # Add type var for controller so we can define
 # StandardDetector[KinetixController, ADWriter] for example
@@ -272,8 +274,8 @@ async def trigger(self) -> None:
                 )
             )
 
-        self._trigger_info = _ensure_trigger_info_exists(self._trigger_info)
-        if self._trigger_info.trigger is not DetectorTrigger.INTERNAL:
+        trigger_info = _ensure_trigger_info_exists(self._trigger_info)
+        if trigger_info.trigger is not DetectorTrigger.INTERNAL:
             msg = "The trigger method can only be called with INTERNAL triggering"
             raise ValueError(msg)
 
@@ -284,9 +286,7 @@ async def trigger(self) -> None:
         end_observation = indices_written + 1
 
         async for index in self._writer.observe_indices_written(
-            DEFAULT_TIMEOUT
-            + (self._trigger_info.livetime or 0)
-            + self._trigger_info.deadtime
+            DEFAULT_TIMEOUT + (trigger_info.livetime or 0) + trigger_info.deadtime
         ):
             if index >= end_observation:
                 break
@@ -312,24 +312,26 @@ async def prepare(self, value: TriggerInfo) -> None:
             raise ValueError(msg)
         elif not value.deadtime:
             value.deadtime = self._controller.get_deadtime(value.livetime)
-        self._trigger_info = value
         self._number_of_triggers_iter = iter(
-            self._trigger_info.number_of_triggers
-            if isinstance(self._trigger_info.number_of_triggers, list)
-            else [self._trigger_info.number_of_triggers]
+            value.number_of_triggers
+            if isinstance(value.number_of_triggers, list)
+            else [value.number_of_triggers]
         )
         self._describe, _ = await asyncio.gather(
             self._writer.open(value.multiplier), self._controller.prepare(value)
         )
         self._initial_frame = await self._writer.get_indices_written()
         if value.trigger != DetectorTrigger.INTERNAL:
             await self._controller.arm()
-            self._fly_start = time.monotonic()
+        self._trigger_info = value
 
     @AsyncStatus.wrap
     async def kickoff(self):
         if self._trigger_info is None or self._number_of_triggers_iter is None:
             raise RuntimeError("Prepare must be called before kickoff!")
+        if self._trigger_info.trigger == DetectorTrigger.INTERNAL:
+            await self._controller.arm()
+        self._fly_start = time.monotonic()
         try:
             self._frames_to_complete = next(self._number_of_triggers_iter)
             self._completable_frames += self._frames_to_complete
@@ -341,13 +343,13 @@ async def kickoff(self):
 
     @WatchableAsyncStatus.wrap
     async def complete(self):
-        self._trigger_info = _ensure_trigger_info_exists(self._trigger_info)
+        trigger_info = _ensure_trigger_info_exists(self._trigger_info)
         indices_written = self._writer.observe_indices_written(
-            self._trigger_info.frame_timeout
+            trigger_info.frame_timeout
             or (
                 DEFAULT_TIMEOUT
-                + (self._trigger_info.livetime or 0)
-                + (self._trigger_info.deadtime or 0)
+                + (trigger_info.livetime or 0)
+                + (trigger_info.deadtime or 0)
             )
         )
         try:
@@ -367,7 +369,7 @@ async def complete(self):
                     break
         finally:
             await indices_written.aclose()
-            if self._completable_frames >= self._trigger_info.total_number_of_triggers:
+            if self._completable_frames >= trigger_info.total_number_of_triggers:
                 self._completable_frames = 0
                 self._frames_to_complete = 0
                 self._number_of_triggers_iter = None