ENH: add the ability to subscribe to a child in SubscriptionStatus

ophyd/status.py

@@ -762,6 +762,10 @@ class SubscriptionStatus(DeviceStatus):
 
     run: bool, optional
         Run the callback now
+
+    subscribe_child : str, optional
+        The name of the child to actually subscribe to.  May be dotted to go
+        down more than one generation.
     """
 
     def __init__(
@@ -772,6 +776,8 @@ def __init__(
         timeout=None,
         settle_time=None,
         run=True,
+        *,
+        subscribe_child=None,
     ):
         # Store device and attribute information
         self.device = device
@@ -781,7 +787,16 @@ def __init__(
         super().__init__(device, timeout=timeout, settle_time=settle_time)
 
         # Subscribe callback and run initial check
-        self.device.subscribe(self.check_value, event_type=event_type, run=run)
+        if subscribe_child is None:
+            self.device.subscribe(self.check_value, event_type=event_type, run=run)
+            self._sub_obj = self.device
+
+        else:
+            obj = device
+            for k in subscribe_child.split("."):
+                obj = getattr(obj, k)
+            obj.subscribe(self.check_value, event_type=event_type, run=run)
+            self._sub_obj = obj
 
     def check_value(self, *args, **kwargs):
         """
@@ -808,7 +823,7 @@ def set_finished(self):
         Status object, but only by the object that created and returned it.
         """
         # Clear callback
-        self.device.clear_sub(self.check_value)
+        self._sub_obj.clear_sub(self.check_value)
         # Run completion
         super().set_finished()
 

ophyd/tests/test_status.py

@@ -3,7 +3,7 @@
 
 import pytest
 
-from ophyd import Device
+from ophyd import Device, Component
 from ophyd.signal import EpicsSignalRO
 from ophyd.status import (
     DeviceStatus,
@@ -117,9 +117,7 @@ def test_direct_done_setting():
         st.done = True  # but for now no-ops warn
 
 
-def test_subscription_status():
-    # Arbitrary device
-    d = Device("Tst:Prefix", name="test")
+def _sub_staus_helper(device, obj, name):
     # Mock callback
     m = Mock()
 
@@ -130,20 +128,49 @@ def cb(*args, done=False, **kwargs):
         # Return finished or not
         return done
 
-    status = SubscriptionStatus(d, cb, event_type=d.SUB_ACQ_DONE)
+    status = SubscriptionStatus(
+        device, cb, event_type=obj.SUB_ACQ_DONE, subscribe_child=name
+    )
 
     # Run callbacks but do not mark as complete
-    d._run_subs(sub_type=d.SUB_ACQ_DONE, done=False)
+    obj._run_subs(sub_type=obj.SUB_ACQ_DONE, done=False)
     time.sleep(0.1)  # Wait for callbacks to run.
     assert m.called
     assert not status.done and not status.success
 
     # Run callbacks and mark as complete
-    d._run_subs(sub_type=d.SUB_ACQ_DONE, done=True)
+    obj._run_subs(sub_type=obj.SUB_ACQ_DONE, done=True)
     time.sleep(0.1)  # Wait for callbacks to run.
     assert status.done and status.success
 
 
+def test_subscription_status():
+    # Arbitrary device
+    d = Device("Tst:Prefix", name="test")
+    _sub_staus_helper(d, d, None)
+
+
+def test_subscription_status_1gen():
+    # Arbitrary device
+    class Dev(Device):
+        a = Component(Device)
+
+    d = Dev("Tst:Prefix", name="test")
+    _sub_staus_helper(d, d.a, "a")
+
+
+def test_subscription_status_2gen():
+    # Arbitrary device
+    class Child(Device):
+        a = Component(Device)
+
+    class Dev(Device):
+        b = Component(Child)
+
+    d = Dev("Tst:Prefix", name="test")
+    _sub_staus_helper(d, d.b.a, "b.a")
+
+
 def test_subscription_status_does_not_try_and_stop_ro_device():
     # Arbitrary device
     d = EpicsSignalRO("Tst:Prefix", name="test")