From 49a49a55fc4e0b30fa1f79b3a3da14b721b78719 Mon Sep 17 00:00:00 2001 From: Alexis Gamelin Date: Thu, 1 Oct 2026 10:46:36 +0200 Subject: [PATCH] Allow for BPM with a single position Solve #437 Add a new CSSparseScalarAggregator class to allow for aggregator with missing values. --- pyaml/control/abstract_impl.py | 103 +++++++++++++++++++++++++--- pyaml/control/controlsystem.py | 11 +-- tests/bpm/test_bpm_controlsystem.py | 103 ++++++++++++++++++++++++++++ 3 files changed, 204 insertions(+), 13 deletions(-) diff --git a/pyaml/control/abstract_impl.py b/pyaml/control/abstract_impl.py index 077a819d..96024762 100644 --- a/pyaml/control/abstract_impl.py +++ b/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,20 +884,22 @@ 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. """ @@ -822,14 +907,14 @@ def __init__(self, hDev: DeviceAccess, vDev: DeviceAccess): 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() # ------------------------------------------------------------------------------ diff --git a/pyaml/control/controlsystem.py b/pyaml/control/controlsystem.py index 9199f98d..23876d77 100644 --- a/pyaml/control/controlsystem.py +++ b/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) diff --git a/tests/bpm/test_bpm_controlsystem.py b/tests/bpm/test_bpm_controlsystem.py index 13321ca1..4dfea2a0 100644 --- a/tests/bpm/test_bpm_controlsystem.py +++ b/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)