signal walking added for debugging (#917)

* signal walking added for debuggin@

* respond to review

* add test walk devices

* update tests as expected

* another test for signal sources

* lint

* make an explicit test of signal_walk

* fix imports

Author: stan-dot

src/ophyd_async/core/__init__.py

@@ -56,7 +56,9 @@
     soft_signal_rw,
     wait_for_value,
     walk_config_signals,
+    walk_devices,
     walk_rw_signals,
+    walk_signal_sources,
 )
 from ._signal_backend import (
     Array1D,
@@ -147,6 +149,8 @@
     "set_and_wait_for_other_value",
     "walk_rw_signals",
     "walk_config_signals",
+    "walk_devices",
+    "walk_signal_sources",
     # Readable
     "StandardReadable",
     "StandardReadableFormat",

src/ophyd_async/core/_signal.py

@@ -688,7 +688,7 @@ async def set_and_wait_for_value(
     )
 
 
-def walk_rw_signals(device: Device, path_prefix: str = "") -> dict[str, SignalRW[Any]]:
+def walk_rw_signals(device: Device) -> dict[str, SignalRW[Any]]:
     """Retrieve all SignalRWs from a device.
 
     Stores retrieved signals with their dotted attribute paths in a dictionary. Used as
@@ -700,48 +700,67 @@ def walk_rw_signals(device: Device, path_prefix: str = "") -> dict[str, SignalRW
         A dictionary matching the string attribute path of a SignalRW with the
         signal itself.
     """
-    signals: dict[str, SignalRW[Any]] = {}
-
-    for attr_name, attr in device.children():
-        dot_path = f"{path_prefix}{attr_name}"
-        if type(attr) is SignalRW:
-            signals[dot_path] = attr
-        attr_signals = walk_rw_signals(attr, path_prefix=dot_path + ".")
-        signals.update(attr_signals)
-    return signals
+    all_devices = walk_devices(device)
+    return {path: dev for path, dev in all_devices.items() if type(dev) is SignalRW}
 
 
 async def walk_config_signals(
-    device: Device, path_prefix: str = ""
+    device: Device,
 ) -> dict[str, SignalRW[Any]]:
     """Retrieve all configuration signals from a device.
 
     Stores retrieved signals with their dotted attribute paths in a dictionary. Used as
     part of saving and loading a device.
 
     :param device: Device to retrieve configuration signals from.
-    :param path_prefix: For internal use, leave blank when calling the method.
     :return:
         A dictionary matching the string attribute path of a SignalRW with the
         signal itself.
     """
-    signals: dict[str, SignalRW[Any]] = {}
     config_names: list[str] = []
     if isinstance(device, Configurable):
         configuration = device.read_configuration()
         if inspect.isawaitable(configuration):
             configuration = await configuration
         config_names = list(configuration.keys())
-    for attr_name, attr in device.children():
-        dot_path = f"{path_prefix}{attr_name}"
-        if isinstance(attr, SignalRW) and attr.name in config_names:
-            signals[dot_path] = attr
-        signals.update(await walk_config_signals(attr, path_prefix=dot_path + "."))
 
-    return signals
+    all_devices = walk_devices(device)
+    return {
+        path: dev
+        for path, dev in all_devices.items()
+        if isinstance(dev, SignalRW) and dev.name in config_names
+    }
 
 
 class Ignore:
     """Annotation to ignore a signal when connecting a device."""
 
     pass
+
+
+def walk_devices(device: Device, path_prefix: str = "") -> dict[str, Device]:
+    """Recursively retrieve all Devices from a device tree.
+
+    :param device: Root device to start from.
+    :param path_prefix: For internal use, leave blank when calling the method.
+    :return: A dictionary mapping dotted attribute paths to Device instances.
+    """
+    devices: dict[str, Device] = {}
+    for attr_name, attr in device.children():
+        dot_path = f"{path_prefix}{attr_name}"
+        devices[dot_path] = attr
+        devices.update(walk_devices(attr, path_prefix=dot_path + "."))
+    return devices
+
+
+def walk_signal_sources(device: Device) -> dict[str, str]:
+    """Recursively gather the `source` field from every Signal in a device tree.
+
+    :param device: Root device to start from.
+    :param path_prefix: For internal use, leave blank when calling the method.
+    :return: A dictionary mapping dotted attribute paths to Signal source strings.
+    """
+    all_devices = walk_devices(device)
+    return {
+        path: dev.source for path, dev in all_devices.items() if isinstance(dev, Signal)
+    }

tests/core/test_signal.py

@@ -27,9 +27,15 @@
     soft_signal_r_and_setter,
     soft_signal_rw,
     wait_for_value,
+    walk_devices,
+    walk_signal_sources,
+)
+from ophyd_async.core import (
+    StandardReadableFormat as Format,
+)
+from ophyd_async.core._signal import (
+    _SignalCache,  # noqa: PLC2701
 )
-from ophyd_async.core import StandardReadableFormat as Format
-from ophyd_async.core._signal import _SignalCache  # noqa: PLC2701
 from ophyd_async.epics.core import epics_signal_r, epics_signal_rw
 from ophyd_async.epics.core._signal import get_signal_backend_type  # noqa: PLC2701
 from ophyd_async.testing import (
@@ -1021,3 +1027,44 @@ def test_remove_non_existing_listener():
     signal_rw = soft_signal_rw(int, initial_value=4)
     cbs = []
     assert signal_rw.clear_sub(cbs.append) is None
+
+
+async def test_walk_devices_returns_all_devices(mock_readable: DummyReadableArray):
+    """
+    Test that walk_devices returns all child devices with correct dotted paths.
+    """
+
+    # Get all devices in the tree
+    devices = walk_devices(mock_readable)
+
+    assert devices == {
+        "int_value": mock_readable.int_value,
+        "int_array": mock_readable.int_array,
+        "float_array": mock_readable.float_array,
+        "str_value": mock_readable.str_value,
+        "strictEnum_value": mock_readable.strictEnum_value,
+    }
+
+    # All returned objects should be Device instances
+    for dev in devices.values():
+        assert isinstance(dev, Device)
+
+
+async def test_walk_signal_sources_returns_signal_sources(
+    mock_readable: DummyReadableArray,
+):
+    """
+    Test that walk_signal_sources returns correct mapping
+    of dotted paths to Signal sources.
+    """
+    sources = walk_signal_sources(mock_readable)
+
+    expected_sources = {
+        "int_value": mock_readable.int_value.source,
+        "int_array": mock_readable.int_array.source,
+        "float_array": mock_readable.float_array.source,
+        "str_value": mock_readable.str_value.source,
+        "strictEnum_value": mock_readable.strictEnum_value.source,
+    }
+
+    assert sources == expected_sources