WIP: copy plugins of same type

ophyd/areadetector/plugins.py

@@ -25,13 +25,14 @@
 import re
 import time as ttime
 from collections import OrderedDict
+from typing import Optional
 
 import numpy as np
 
 from ..device import Component, Device
 from ..device import FormattedComponent as FCpt
 from ..device import GenerateDatumInterface
-from ..signal import ArrayAttributeSignal, EpicsSignal, EpicsSignalRO
+from ..signal import ArrayAttributeSignal, EpicsSignal, EpicsSignalRO, Signal
 from ..utils import enum
 from ..utils.errors import DestroyedError, PluginMisconfigurationError, UnprimedPlugin
 from .base import ADBase
@@ -1122,7 +1123,9 @@ class PluginBase_V34(PluginBase_V33, version=(3, 4), version_of=PluginBase):
 # --- NDFile ---
 
 
-class FilePlugin_V20(PluginBase_V20, FilePlugin, version=(2, 0), version_of=FilePlugin):
+class FilePlugin_V20(
+    PluginBase_V20, FilePlugin, version=(2, 0), version_of=FilePlugin
+):
     ...
 
 
@@ -1218,7 +1221,9 @@ class ColorConvPlugin_V34(
 # --- NDFileHDF5 ---
 
 
-class HDF5Plugin_V20(FilePlugin_V20, HDF5Plugin, version=(2, 0), version_of=HDF5Plugin):
+class HDF5Plugin_V20(
+    FilePlugin_V20, HDF5Plugin, version=(2, 0), version_of=HDF5Plugin
+):
     ...
 
 
@@ -1417,7 +1422,9 @@ class ImagePlugin_V34(
 # --- NDFileJPEG ---
 
 
-class JPEGPlugin_V20(FilePlugin_V20, JPEGPlugin, version=(2, 0), version_of=JPEGPlugin):
+class JPEGPlugin_V20(
+    FilePlugin_V20, JPEGPlugin, version=(2, 0), version_of=JPEGPlugin
+):
     ...
 
 
@@ -1850,6 +1857,7 @@ class ROIPlugin_V34(
 @register_plugin
 class ROIStatPlugin(Device, version_type="ADCore"):
     "Serves as a base class for other versions"
+
     _default_suffix = "ROIStat1:"
     _suffix_re = r"ROIStat\d:"
     _plugin_type = "NDPluginROIStat"
@@ -1908,6 +1916,7 @@ class ROIStatPlugin_V34(
 
 class ROIStatNPlugin(Device, version_type="ADCore"):
     "Serves as a base class for other versions"
+
     ...
 
 
@@ -2033,7 +2042,9 @@ class StatsPlugin_V34(
 # --- NDFileTIFF ---
 
 
-class TIFFPlugin_V20(FilePlugin_V20, TIFFPlugin, version=(2, 0), version_of=TIFFPlugin):
+class TIFFPlugin_V20(
+    FilePlugin_V20, TIFFPlugin, version=(2, 0), version_of=TIFFPlugin
+):
     ...
 
 
@@ -2162,6 +2173,7 @@ class TransformPlugin_V34(
 @register_plugin
 class PvaPlugin(Device, version_type="ADCore"):
     "Serves as a base class for other versions"
+
     _default_suffix = "Pva1:"
     _suffix_re = r"Pva\d:"
     _plugin_type = "NDPluginPva"
@@ -2201,6 +2213,7 @@ class PvaPlugin_V34(
 @register_plugin
 class FFTPlugin(Device, version_type="ADCore"):
     "Serves as a base class for other versions"
+
     ...
     _default_suffix = "FFT1:"
     _suffix_re = r"FFT\d:"
@@ -2263,6 +2276,7 @@ class FFTPlugin_V34(
 @register_plugin
 class ScatterPlugin(Device, version_type="ADCore"):
     "Serves as a base class for other versions"
+
     _default_suffix = "Scatter1:"
     _suffix_re = r"Scatter\d:"
     _plugin_type = "NDPluginScatter"
@@ -2276,7 +2290,9 @@ class ScatterPlugin_V31(
     )
 
 
-class ScatterPlugin_V32(ScatterPlugin_V31, version=(3, 2), version_of=ScatterPlugin):
+class ScatterPlugin_V32(
+    ScatterPlugin_V31, version=(3, 2), version_of=ScatterPlugin
+):
     ...
 
 
@@ -2298,6 +2314,7 @@ class ScatterPlugin_V34(
 @register_plugin
 class PosPlugin(Device, version_type="ADCore"):
     "Serves as a base class for other versions"
+
     _default_suffix = "Pos1:"
     _suffix_re = r"Pos\d:"
     _plugin_type = "NDPosPlugin"
@@ -2353,6 +2370,7 @@ class PosPluginPlugin_V34(
 @register_plugin
 class CircularBuffPlugin(Device, version_type="ADCore"):
     "Serves as a base class for other versions"
+
     _default_suffix = "CB1:"
     _suffix_re = r"CB\d:"
     _plugin_type = "NDPluginCircularBuff"
@@ -2439,6 +2457,7 @@ class CircularBuffPlugin_V34(
 
 class AttributeNPlugin(Device, version_type="ADCore"):
     "Serves as a base class for other versions"
+
     ...
 
 
@@ -2462,6 +2481,7 @@ class AttributeNPlugin_V26(
 
 class AttrPlotPlugin(Device, version_type="ADCore"):
     "Serves as a base class for other versions"
+
     _plugin_type = "NDAttrPlot"
 
 
@@ -2489,6 +2509,7 @@ class AttrPlotPlugin_V34(
 
 class TimeSeriesNPlugin(Device, version_type="ADCore"):
     "Serves as a base class for other versions"
+
     ...
 
 
@@ -2505,6 +2526,7 @@ class TimeSeriesNPlugin_V25(
 @register_plugin
 class TimeSeriesPlugin(Device, version_type="ADCore"):
     "Serves as a base class for other versions"
+
     _plugin_type = "NDPluginTimeSeries"
 
 
@@ -2563,6 +2585,7 @@ class TimeSeriesPlugin_V34(
 @register_plugin
 class CodecPlugin(Device, version_type="ADCore"):
     "Serves as a base class for other versions"
+
     _plugin_type = "NDPluginCodec"
 
 
@@ -2595,6 +2618,7 @@ class CodecPlugin_V34(
 @register_plugin
 class AttributePlugin(Device, version_type="ADCore"):
     "Serves as a base class for other versions"
+
     _default_suffix = "Attr1:"
     _suffix_re = r"Attr\d:"
     _plugin_type = "NDPluginAttribute"
@@ -2733,7 +2757,7 @@ def plugin_from_pvname(pv):
 
 
 def get_areadetector_plugin_class(prefix, timeout=2.0):
-    """Get an areadetector plugin class by supplying its PV prefix
+    """Get an areadetector plugin class by supplying its PV prefix.
 
     Uses `plugin_from_pvname` first, but falls back on using epics channel
     access to determine the plugin type.
@@ -2754,26 +2778,26 @@ def get_areadetector_plugin_class(prefix, timeout=2.0):
     if cls is not None:
         return cls
 
-    type_rbv = prefix + "PluginType_RBV"
+    type_rbv = prefix + 'PluginType_RBV'
     type_ = cl.caget(type_rbv, timeout=timeout)
 
     if type_ is None:
-        raise ValueError("Unable to determine plugin type (caget timed out)")
+        raise ValueError('Unable to determine plugin type (caget timed out)')
 
     # HDF5 includes version number, remove it
-    type_ = type_.split(" ")[0]
+    type_ = type_.split(' ')[0]
 
     try:
         return _plugin_class[type_]
     except KeyError:
         raise ValueError(
-            "Unable to determine plugin type (PluginType={})" "".format(type_)
+            'Unable to determine plugin type (PluginType={})' ''.format(type_)
         )
 
 
 def get_areadetector_plugin(prefix, **kwargs):
-    """Get an instance of an areadetector plugin by supplying its PV prefix
-    and any kwargs for the constructor.
+    """
+    Get an instance of an areadetector plugin by supplying its PV prefix and any kwargs for the constructor.
 
     Uses `plugin_from_pvname` first, but falls back on using
     epics channel access to determine the plugin type.
@@ -2788,9 +2812,72 @@ def get_areadetector_plugin(prefix, **kwargs):
     ValueError
         If the plugin type can't be determined
     """
-
     cls = get_areadetector_plugin_class(prefix)
     if cls is None:
-        raise ValueError("Unable to determine plugin type")
+        raise ValueError('Unable to determine plugin type')
 
     return cls(prefix, **kwargs)
+
+
+def _resolve_dotted_attr(obj, dotted_name):
+    """Resolve a dotted attribute name on an object.
+
+    Args:
+        obj (Object): The object on which to resolve the attribute.
+        dotted_name (str): The dotted attribute name to resolve.
+
+    Returns:
+        Any: The resolved attribute value, or None if not found.
+    """
+    for part in dotted_name.split('.'):
+        obj = getattr(obj, part, None)
+        if obj is None:
+            return None
+    return obj
+
+
+def copy_plugin(
+    source: PluginBase,
+    target: PluginBase,
+    include: Optional[set[Signal]] = None,
+    exclude: Optional[set[Signal]] = None,
+):
+    """Copy signals from one plugin to another.
+
+    Args:
+        source (PluginBase): source plugin from which to copy signals
+        target (PluginBase): target plugin to which signals will be copied
+        include (list, optional): list of source signals to include. Defaults to None.
+        exclude (list, optional): list of source signals to exclude. Defaults to None.
+
+    Raises:
+        TypeError: If source and target are not the same type
+        ValueError: If both include and exclude lists are specified, only one should be used
+    """
+    if not isinstance(source, PluginBase) or not isinstance(target, PluginBase):
+        raise TypeError('Source and target must be instances of PluginBase')
+
+    if type(source) is not type(target):
+        raise TypeError(
+            f'Source plugin and target plugin must be of the same type, '
+            f'got {type(source)} and {type(target)}'
+        )
+
+    if include is not None and exclude is not None:
+        raise ValueError('Cannot specify both include and exclude lists, choose one.')
+
+    for walk in source.walk_signals():
+        src_sig: Signal = walk.item
+        if exclude and src_sig in exclude:
+            continue
+        if include and src_sig not in include:
+            continue
+        tgt_sig = _resolve_dotted_attr(target, walk.dotted_name)
+
+        print(f'{type(tgt_sig)=}')
+        print(f'{isinstance(tgt_sig, Signal)=}')
+        print(f'{tgt_sig.write_access=}')
+        # Only copy if tgt_sig is a real ophyd Signal and is writable
+        if isinstance(tgt_sig, Signal) and tgt_sig.write_access:
+            value = src_sig.get()
+            tgt_sig.put(value)

ophyd/signal.py

@@ -2373,5 +2373,9 @@ class ArrayAttributeSignal(AttributeSignal):
     how to store the data into metadatastore.
     """
 
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+        self._metadata.update(write_access=False)
+
     def get(self, **kwargs):
         return np.asarray(super().get(**kwargs))

ophyd/tests/test_copy.py

@@ -0,0 +1,81 @@
+from ophyd.areadetector.plugins import StatsPlugin, copy_plugin
+from ophyd.sim import make_fake_device
+
+
+def test_copy_plugin_values_all():
+    # Create a fake StatsPlugin device class
+    FakeStats1 = make_fake_device(StatsPlugin)
+    FakeStats2 = make_fake_device(StatsPlugin)
+
+    # Instantiate two fake plugin devices
+    src = FakeStats1("SRC:", name="src")
+    tgt = FakeStats2("TGT:", name="tgt")
+
+    # Set a value on the source device
+    src.compute_centroid.put(1)
+    src.hist_max.put(123)
+    src.hist_min.put(10)
+
+    # Target should have different values before copy
+    tgt.compute_centroid.put(0)
+    tgt.hist_max.put(0)
+    tgt.hist_min.put(0)
+
+    # Copy all values from src to tgt
+    copy_plugin(src, tgt)
+
+    # Assert that values have been copied
+    assert tgt.compute_centroid.get() == "1"
+    assert tgt.hist_max.get() == 123
+    assert tgt.hist_min.get() == 10
+
+
+def test_copy_plugin_include():
+    FakeStats_INC = make_fake_device(StatsPlugin)
+
+    # Instantiate two fake plugin devices
+    src_inc = FakeStats_INC("SRC:", name="src")
+    tgt_inc = FakeStats_INC("TGT:", name="tgt")
+
+    # Set a value on the source device
+    src_inc.compute_centroid.sim_put(1)
+    src_inc.hist_max.sim_put(123)
+    src_inc.hist_min.sim_put(10)
+
+    # Target should have different values before copy
+    tgt_inc.compute_centroid.sim_put(0)
+    tgt_inc.hist_max.sim_put(0)
+    tgt_inc.hist_min.sim_put(0)
+
+    # Test include/exclude
+    src_inc.hist_max.sim_put(555)
+    copy_plugin(src_inc, tgt_inc, include={src_inc.hist_max})
+    assert tgt_inc.hist_max.get() == 555
+    # hist_min and compute_centroid should remain unchanged
+    assert tgt_inc.hist_min.get() == 0
+    assert tgt_inc.compute_centroid.get() == "0"
+
+
+def test_copy_plugin_exclude():
+    FakeStat2 = make_fake_device(StatsPlugin)
+
+    # Instantiate two fake plugin devices
+    src_exc = FakeStat2("SRC:", name="src")
+    tgt_exc = FakeStat2("TGT:", name="tgt")
+
+    # Set a value on the source device
+    src_exc.compute_centroid.sim_put(1)
+    src_exc.hist_max.sim_put(123)
+    src_exc.hist_min.sim_put(999)
+
+    # Target should have different values before copy
+    tgt_exc.compute_centroid.sim_put(0)
+    tgt_exc.hist_max.sim_put(0)
+    tgt_exc.hist_min.sim_put(0)
+
+    copy_plugin(src_exc, tgt_exc, exclude={src_exc.hist_max})
+    # hist_max should remain unchanged
+    assert tgt_exc.hist_max.get() == 0
+    # hist_min and compute_centroid should be updated
+    assert tgt_exc.hist_min.get() == 999
+    assert tgt_exc.compute_centroid.get() == "1"