Repository: bluesky/bluesky-darkframes

Issue:
Support multiple DarkFramePreprocessors
I believe (but have not rigorously verified) that if two `DarkFramePreprocessors` are applied, they will interfere. The outer one will see the `trigger` messages added by the inner one, and try to take a dark frame for a dark frame (if that makes sense...).

We briefly tried this at CSX and that theory seemed to be correct; at least we can say for sure that trying two nested preprocessors failed.

The fix I have in mind is to let `DarkFramePreprocessors` recognize `trigger` messages that came from itself by using a special convention in the `group`---that is instead of `'trigger-<uuid>'`, do `'trigger-dark-frame-preprocessor-<uuid>'`. The content of the group has no meaning to any other part of our system.

Discussion:

Patch:
diff --git a/bluesky_darkframes/__init__.py b/bluesky_darkframes/__init__.py
index ddb3fac..6603ab1 100644
--- a/bluesky_darkframes/__init__.py
+++ b/bluesky_darkframes/__init__.py
@@ -8,6 +8,7 @@ import event_model
 from frozendict import frozendict
 import bluesky.preprocessors
 import bluesky.plan_stubs as bps
+from bluesky.utils import short_uid
 import numpy
 from ophyd import Device
 
@@ -138,6 +139,9 @@ class _SnapshotShell:
         return getattr(self.__snapshot, key)
 
 
+GROUP_PREFIX = 'bluesky-darkframes-trigger'
+
+
 class DarkFramePreprocessor:
     """
     A plan preprocessor that ensures each Run records a dark frame.
@@ -312,18 +316,29 @@ class DarkFramePreprocessor:
                 # bluesky will always see the "Device" self._current_snapshot
                 # here, and it will be satisfied.
                 yield from bps.stage(self._current_snapshot)
-                yield from bps.trigger_and_read([self._current_snapshot],
-                                                name=self.stream_name)
+                yield from trigger_and_read(
+                    [self._current_snapshot],
+                    name=self.stream_name,
+                    group=short_uid(GROUP_PREFIX))
                 yield from bps.unstage(self._current_snapshot)
             self._latch = False
             if msg is not None:
                 return (yield msg)
 
         def maybe_insert_dark_frame(msg):
-            if msg.command == 'trigger' and msg.obj is self.detector and not self._latch:
+            # * Is this a 'trigger' message?
+            # * Does it refer to the detector that this preprocessor cares about?
+            # * Is it *not* from another DarkFramePreprocessor. (There can be
+            # multiple nested DarkFramePreprocessors watching the same detector
+            # by applying different dark_plans / recording Events in different
+            # streams.)
+            if (msg.command == 'trigger'
+                    and msg.obj is self.detector
+                    and not msg.kwargs['group'].startswith(GROUP_PREFIX)
+                    and not self._latch):
                 force_read = self._force_read_before_next_event
                 self._force_read_before_next_event = False
-                self._latch = True
+                self._latch = True  # prevents infinite recursion
                 return insert_dark_frame(force_read=force_read, msg=msg), None
             elif msg.command == 'open_run':
                 # Make sure we get a new Event because we have just started a
@@ -486,3 +501,56 @@ class NoDarkFrame(RuntimeError, BlueskyDarkframesException):
 
 class NoMatchingSnapshot(KeyError, BlueskyDarkframesException):
     ...
+
+
+# Vendored from bluesky.plan_stubs, added `group` parameter
+
+
+def trigger_and_read(devices, name='primary', group=None):
+    """
+    Trigger and read a list of detectors and bundle readings into one Event.
+
+    Parameters
+    ----------
+    devices : iterable
+        devices to trigger (if they have a trigger method) and then read
+    name : string, optional
+        event stream name, a convenient human-friendly identifier; default
+        name is 'primary'
+    group : string (or any hashable object), optional
+        identifier used by 'trigger' and 'wait'
+
+    Yields
+    ------
+    msg : Msg
+        messages to 'trigger', 'wait' and 'read'
+    """
+    # If devices is empty, don't emit 'create'/'save' messages.
+    if not devices:
+        yield from bps.null()
+    devices = bps.separate_devices(devices)  # remove redundant entries
+    rewindable = bps.all_safe_rewind(devices)  # if devices can be re-triggered
+
+    def inner_trigger_and_read():
+        nonlocal group
+        if group is None:
+            group = short_uid('trigger')
+        no_wait = True
+        for obj in devices:
+            if hasattr(obj, 'trigger'):
+                no_wait = False
+                yield from bps.trigger(obj, group=group)
+        # Skip 'wait' if none of the devices implemented a trigger method.
+        if not no_wait:
+            yield from bps.wait(group=group)
+        yield from bps.create(name)
+        ret = {}  # collect and return readings to give plan access to them
+        for obj in devices:
+            reading = (yield from bps.read(obj))
+            if reading is not None:
+                ret.update(reading)
+        yield from bps.save()
+        return ret
+    from bluesky.preprocessors import rewindable_wrapper
+    return (yield from rewindable_wrapper(inner_trigger_and_read(),
+                                          rewindable))
diff --git a/docs/source/usage.rst b/docs/source/usage.rst
index 7071e1a..cb89373 100644
--- a/docs/source/usage.rst
+++ b/docs/source/usage.rst
@@ -110,8 +110,10 @@ for whether/how dark frames can be reused.)
        yield from bps.unstage(detector)
        yield from bps.stage(detector)
        yield from bps.mv(shutter, 'closed')
-       yield from bps.trigger(detector, group='darkframe-trigger')
-       yield from bps.wait('darkframe-trigger')
+       # The `group` parameter passed to trigger MUST start with
+       # bluesky-darkframes-trigger.
+       yield from bps.trigger(detector, group='bluesky-darkframes-trigger')
+       yield from bps.wait('bluesky-darkframes-trigger')
        snapshot = bluesky_darkframes.SnapshotDevice(detector)
        yield from bps.mv(shutter, 'open')
        # Restage.


Test patch:
diff --git a/bluesky_darkframes/tests/tests.py b/bluesky_darkframes/tests/tests.py
index 98ddcbb..31bf2ea 100644
--- a/bluesky_darkframes/tests/tests.py
+++ b/bluesky_darkframes/tests/tests.py
@@ -21,7 +21,7 @@ def dark_plan(detector):
     yield from bps.unstage(detector)
     yield from bps.mv(shutter, 'closed')
     yield from bps.stage(detector)
-    yield from bps.trigger(detector, group='darkframe-trigger')
+    yield from bps.trigger(detector, group='bluesky-darkframes-trigger')
     yield from bps.wait('darkframe-trigger')
     snapshot = bluesky_darkframes.SnapshotDevice(detector)
     yield from bps.unstage(detector)
@@ -277,3 +277,19 @@ def test_no_dark_frames(RE, tmp_path):
 
     with pytest.raises(bluesky_darkframes.NoDarkFrame):
         RE(count([det]))
+
+
+def test_nested_preprocessors(RE):
+    N = 3
+    for i in range(N):
+        dark_frame_preprocessor = bluesky_darkframes.DarkFramePreprocessor(
+            dark_plan=dark_plan, detector=det, max_age=0,
+            stream_name=f'dark_{i}')
+        RE.preprocessors.append(dark_frame_preprocessor)
+
+    def verify_event_count(name, doc):
+        if name == 'stop':
+            for i in range(N):
+                assert doc['num_events'][f'dark_{i}'] == 3
+
+    RE(count([det], 3), verify_event_count)