Allow for BPM to read only a single position

pyaml/control/abstract_impl.py

@@ -294,6 +294,89 @@ def nb_device(self) -> int:
 # ------------------------------------------------------------------------------
 
 
+class CSSparseScalarAggregator(CSScalarAggregator):
+    """
+    Scalar aggregator whose slots may have no device.
+
+    Only existing devices are added to the backend aggregator, so they are
+    still read in one grouped call. Values are scattered back to their slots
+    and missing slots read NaN.
+
+    Parameters
+    ----------
+    devs : DeviceAccessList
+        Backend aggregator holding the existing devices.
+
+    Methods
+    -------
+    add_devices(devices)
+        Add one or more slots, ``None`` for a slot without device.
+    set(value)
+        Write the values of the slots that have a device.
+    set_and_wait(value)
+        Write the values of the slots that have a device and wait.
+    get()
+        Read all slots, NaN for slots without device.
+    readback()
+        Read all slot readbacks, NaN for slots without device.
+    nb_device()
+        Return the number of slots.
+    """
+
+    def __init__(self, devs: DeviceAccessList):
+        """
+        Initialize the CSSparseScalarAggregator.
+        """
+        super().__init__(devs)
+        self._nb_slots = 0
+        self._slots: list[int] = []
+
+    def add_devices(self, devices: DeviceAccess | None | list[DeviceAccess | None]):
+        """
+        Add one or more slots, ``None`` for a slot without device.
+
+        Parameters
+        ----------
+        devices : DeviceAccess | None | list[DeviceAccess | None]
+            Control-system device or devices to manage.
+        """
+        for d in devices if isinstance(devices, list) else [devices]:
+            if d is not None:
+                self._devs.add_devices(d)
+                self._slots.append(self._nb_slots)
+            self._nb_slots += 1
+
+    def _scatter(self, values: NDArray[np.float64]) -> NDArray[np.float64]:
+        out = np.full(self._nb_slots, np.nan)
+        out[self._slots] = values
+        return out
+
+    def set(self, value: NDArray[np.float64]):
+        """Write the values of the slots that have a device."""
+        if self._slots:
+            self._devs.set(np.asarray(value)[self._slots])
+
+    def set_and_wait(self, value: NDArray[np.float64]):
+        """Write the values of the slots that have a device and wait."""
+        if self._slots:
+            self._devs.set_and_wait(np.asarray(value)[self._slots])
+
+    def get(self) -> NDArray[np.float64]:
+        """Read all slots, NaN for slots without device."""
+        return self._scatter(self._devs.get() if self._slots else [])
+
+    def readback(self) -> np.array:
+        """Read all slot readbacks, NaN for slots without device."""
+        return self._scatter(self._devs.readback() if self._slots else [])
+
+    def nb_device(self) -> int:
+        """Return the number of slots."""
+        return self._nb_slots
+
+
+# ------------------------------------------------------------------------------
+
+
 class CSStrengthScalarAggregator(CSScalarAggregator):
     """
     Aggregate magnet strengths while avoiding duplicate hardware writes.
@@ -801,35 +884,37 @@ class RBpmArray(abstract.ReadFloatArray):
 
     Parameters
     ----------
-    hDev : DeviceAccess
-        Device providing the horizontal BPM position.
-    vDev : DeviceAccess
-        Device providing the vertical BPM position.
+    hDev : DeviceAccess | None
+        Device providing the horizontal BPM position, ``None`` for a BPM
+        without horizontal plane.
+    vDev : DeviceAccess | None
+        Device providing the vertical BPM position, ``None`` for a BPM
+        without vertical plane.
 
     Methods
     -------
     get()
-        Return horizontal and vertical BPM positions.
+        Return horizontal and vertical BPM positions (NaN for a missing plane).
     unit()
         Return the unit reported by the BPM device.
     """
 
-    def __init__(self, hDev: DeviceAccess, vDev: DeviceAccess):
+    def __init__(self, hDev: DeviceAccess | None, vDev: DeviceAccess | None):
         """
         Initialize the RBpmArray.
         """
         self._hDev = hDev
         self._vDev = vDev
 
     def get(self) -> np.array:
-        """Return horizontal and vertical BPM positions."""
-        return np.array([self._hDev.get(), self._vDev.get()])
+        """Return horizontal and vertical BPM positions (NaN for a missing plane)."""
+        return np.array([d.get() if d is not None else np.nan for d in (self._hDev, self._vDev)], dtype=float)
 
     # Gets the unit of the value Assume that x and y, offsets and positions
     # have the same unit
     def unit(self) -> str:
         """Return the unit reported by the BPM device."""
-        return self._hDev.unit()
+        return (self._hDev if self._hDev is not None else self._vDev).unit()
 
 
 # ------------------------------------------------------------------------------

pyaml/control/controlsystem.py

@@ -18,6 +18,7 @@
 from ..configuration.unbound_element import UnboundElement
 from ..control.abstract_impl import (
     CSScalarAggregator,
+    CSSparseScalarAggregator,
     CSStrengthScalarAggregator,
     RBetatronTuneArray,
     RBpmArray,
@@ -197,12 +198,14 @@ def create_bpm_aggregators(self, bpms: list[BPM]) -> list[ScalarAggregator | Non
         -------
         list[ScalarAggregator | None]
             Aggregators for combined, horizontal, and vertical positions.
+            A plane without device (e.g. the horizontal plane of a
+            vertical-only XBPM) reads NaN, the other devices are still read
+            in one grouped call.
         """
-        agg = self._create_scalar_aggregator()
-        aggh = self._create_scalar_aggregator()
-        aggv = self._create_scalar_aggregator()
-        if agg is None or aggh is None or aggv is None:
+        aggs = [self.get_aggregator() for _ in range(3)]
+        if any(a is None for a in aggs):
             return [None, None, None]
+        agg, aggh, aggv = (CSSparseScalarAggregator(a) for a in aggs)
         for b in bpms:
             devs = self.get_devices_access(b.get_pos_devices())
             agg.add_devices(devs)

tests/bpm/test_bpm_controlsystem.py

@@ -94,3 +94,106 @@ def test_controlsystem_bpm_position_indexed(install_test_package):
     bpm = sr.live.bpm.get("BPM_C01-04")
 
     assert np.allclose(bpm.positions.get(), np.array([0.0, 1.0]))
+
+
+def _single_plane_bpm_config() -> dict:
+    """Two full BPMs and one vertical-only XBPM, grouped in one BPM array."""
+    entries = []
+    for key in ["c01-01/H", "c01-01/V", "c01-02/H", "c01-02/V", "xbpm/V"]:
+        entries.append(
+            {
+                "type": "tango.pyaml.static_catalog_entry",
+                "key": key,
+                "device": {"type": "tango.pyaml.attribute_read_only", "attribute": f"sr/bpm/{key}", "unit": "mm"},
+            }
+        )
+    return {
+        "type": "pyaml.accelerator",
+        "facility": "ESRF",
+        "machine": "sr",
+        "energy": 6e9,
+        "data_folder": "/data/store",
+        "controls": [
+            {
+                "type": "tango.pyaml.controlsystem",
+                "tango_host": "ebs-simu-3:10000",
+                "name": "live",
+                "catalog": {"type": "tango.pyaml.static_catalog", "entries": entries},
+            }
+        ],
+        "arrays": [{"type": "pyaml.arrays.bpm", "name": "BPM", "elements": ["BPM_1", "BPM_2", "XBPM"]}],
+        "devices": [
+            {"type": "pyaml.bpm.bpm", "name": "BPM_1", "x_pos": "c01-01/H", "y_pos": "c01-01/V"},
+            {"type": "pyaml.bpm.bpm", "name": "BPM_2", "x_pos": "c01-02/H", "y_pos": "c01-02/V"},
+            {"type": "pyaml.bpm.bpm", "name": "XBPM", "y_pos": "xbpm/V"},
+        ],
+    }
+
+
+@pytest.mark.parametrize(
+    "install_test_package",
+    [{"name": "tango-pyaml", "path": "tests/dummy_cs/tango-pyaml"}],
+    indirect=True,
+)
+def test_controlsystem_bpm_single_plane(install_test_package):
+    from tango.pyaml.attribute_store import set_attribute
+
+    set_attribute("sr/bpm/xbpm/V", 0.3, unit="mm")
+    sr = Accelerator.from_dict(_single_plane_bpm_config())
+    xbpm = sr.live.bpm.get("XBPM")
+
+    pos = xbpm.positions.get()
+    assert np.isnan(pos[0])
+    assert pos[1] == pytest.approx(0.3)
+    assert xbpm.positions.unit() == "mm"
+
+
+@pytest.mark.parametrize(
+    "install_test_package",
+    [{"name": "tango-pyaml", "path": "tests/dummy_cs/tango-pyaml"}],
+    indirect=True,
+)
+def test_controlsystem_bpm_array_with_single_plane_bpm(install_test_package):
+    from tango.pyaml.attribute_store import set_attribute
+
+    for name, value in [("c01-01/H", 0.1), ("c01-01/V", -0.1), ("c01-02/H", 0.2), ("c01-02/V", -0.2), ("xbpm/V", 0.3)]:
+        set_attribute(f"sr/bpm/{name}", value, unit="mm")
+    sr = Accelerator.from_dict(_single_plane_bpm_config())
+    bpms = sr.live.bpms.get("BPM")
+
+    pos = bpms.positions.get()
+    assert pos.shape == (3, 2)
+    assert np.isnan(pos[2, 0])
+    assert np.allclose(pos[:2], [[0.1, -0.1], [0.2, -0.2]])
+    assert pos[2, 1] == pytest.approx(0.3)
+
+    h = bpms.h.get()
+    assert np.allclose(h[:2], [0.1, 0.2])
+    assert np.isnan(h[2])
+    assert np.allclose(bpms.v.get(), [-0.1, -0.2, 0.3])
+
+    # The missing plane must not disable the grouped reads
+    agg, aggh, aggv = sr.live.create_bpm_aggregators(list(bpms))
+    assert [agg.nb_device(), aggh.nb_device(), aggv.nb_device()] == [6, 3, 3]
+    assert np.allclose(agg.get(), pos.flatten(), equal_nan=True)
+    assert np.allclose(aggh.get(), h, equal_nan=True)
+    assert np.allclose(aggv.get(), [-0.1, -0.2, 0.3])
+
+
+@pytest.mark.parametrize(
+    "install_test_package",
+    [{"name": "tango-pyaml", "path": "tests/dummy_cs/tango-pyaml"}],
+    indirect=True,
+)
+def test_controlsystem_bpm_array_without_horizontal_plane(install_test_package):
+    from tango.pyaml.attribute_store import set_attribute
+
+    set_attribute("sr/bpm/xbpm/V", 0.3, unit="mm")
+    config = _single_plane_bpm_config()
+    config["arrays"][0]["elements"] = ["XBPM"]
+    sr = Accelerator.from_dict(config)
+    bpms = sr.live.bpms.get("BPM")
+
+    assert np.isnan(bpms.h.get()).all()
+    assert np.allclose(bpms.v.get(), [0.3])
+    assert bpms.positions.get().shape == (1, 2)