From 6f3d0f5838ebd31922a424cf7f26342a0b7fb8b5 Mon Sep 17 00:00:00 2001 From: Alexis Gamelin Date: Thu, 1 Oct 2026 11:17:47 +0200 Subject: [PATCH] Fix SerializedMagnets attachment to a control system (#458) All magnets of a serialized group share one power supply, so resolve it once instead of indexing one device per magnet (IndexError). Virtual magnet strengths now carry their position in the group so that CSStrengthScalarAggregator maps each one to the right strength of the serialized model; previously every magnet hit slot 0 and only 1/N of a change reached the power supply. Co-Authored-By: Claude Opus 5.5 --- pyaml/control/abstract_impl.py | 41 ++++++++++++- pyaml/control/controlsystem.py | 13 ++--- tests/config/sr_serialized_magnets_cs.yaml | 57 ++++++++++++++++++ tests/magnet/test_serialized_magnets_cs.py | 67 ++++++++++++++++++++++ 4 files changed, 170 insertions(+), 8 deletions(-) create mode 100644 tests/config/sr_serialized_magnets_cs.yaml create mode 100644 tests/magnet/test_serialized_magnets_cs.py diff --git a/pyaml/control/abstract_impl.py b/pyaml/control/abstract_impl.py index 077a819d..58169413 100644 --- a/pyaml/control/abstract_impl.py +++ b/pyaml/control/abstract_impl.py @@ -347,7 +347,10 @@ def add_magnet(self, magnet: Magnet, devs: list[DeviceAccess]): devs : list[DeviceAccess] Hardware devices associated with the magnet model. """ - strengthIndex = magnet.strength.index() if isinstance(magnet.strength, abstract.RWMapper) else 0 + # Magnets of a serialized group share the serialized model (one power supply for N strengths): + # the power-supply current is computed by the model from the N strengths (mean of the implied currents). + is_indexed = isinstance(magnet.strength, (abstract.RWMapper, RWSerializedStrengthScalar)) + strengthIndex = magnet.strength.index() if is_indexed else 0 if magnet.model not in self.__models: index = len(self.__models) self.__models.append(magnet.model) @@ -628,6 +631,42 @@ def set_magnet_rigidity(self, brho: np.double): self.__model.set_magnet_rigidity(brho) +class RWSerializedStrengthScalar(RWStrengthScalar): + """ + Expose the strength of one magnet of a serialized group. + + All magnets of a serialized group share one power supply. This accessor + converts with the magnet's own sub-model and remembers the magnet position + in the group so that aggregators can map it to the right strength of the + serialized model. + + Parameters + ---------- + model : MagnetModel + Sub-model of the magnet, used for strength conversion and units. + dev : DeviceAccess + Power-supply device shared by the serialized group. + idx : int + Zero-based position of the magnet in the serialized group. + + Methods + ------- + index() + Return the position of the magnet in the serialized group. + """ + + def __init__(self, model: MagnetModel, dev: DeviceAccess, idx: int): + """ + Initialize the RWSerializedStrengthScalar. + """ + super().__init__(model, dev) + self.__idx = idx + + def index(self) -> int: + """Return the position of the magnet in the serialized group.""" + return self.__idx + + # ------------------------------------------------------------------------------ diff --git a/pyaml/control/controlsystem.py b/pyaml/control/controlsystem.py index 9199f98d..e7145404 100644 --- a/pyaml/control/controlsystem.py +++ b/pyaml/control/controlsystem.py @@ -28,6 +28,7 @@ RWRFFrequencyScalar, RWRFPhaseScalar, RWRFVoltageScalar, + RWSerializedStrengthScalar, RWStrengthArray, RWStrengthScalar, ) @@ -226,16 +227,14 @@ def _fill_combined_function_magnet(self, magnet: CombinedFunctionMagnet) -> None self.magnet.add(virtual_magnet) def _fill_serialized_magnets(self, magnets: SerializedMagnets) -> None: - devices = self.get_devices_access(magnets.model.get_device_names()) + # All magnets of the group share the same power supply + device = self.get_device_access(magnets.model.get_device_names()[0]) currents = [] strengths = [] for index in range(magnets.get_nb_magnets()): - currents.append( - RWHardwareScalar(magnets.model.get_sub_model(index), devices[index]) if magnets.model.has_hardware() else None - ) - strengths.append( - RWStrengthScalar(magnets.model.get_sub_model(index), devices[index]) if magnets.model.has_physics() else None - ) + sub_model = magnets.model.get_sub_model(index) + currents.append(RWHardwareScalar(sub_model, device) if magnets.model.has_hardware() else None) + strengths.append(RWSerializedStrengthScalar(sub_model, device, index) if magnets.model.has_physics() else None) attached_magnets = magnets.attach(self, strengths, currents) self.serialized_magnet.add(attached_magnets[0]) for magnet in attached_magnets[1:]: diff --git a/tests/config/sr_serialized_magnets_cs.yaml b/tests/config/sr_serialized_magnets_cs.yaml new file mode 100644 index 00000000..2db7546f --- /dev/null +++ b/tests/config/sr_serialized_magnets_cs.yaml @@ -0,0 +1,57 @@ +type: pyaml.accelerator +facility: ESRF +machine: sr +energy: 6e9 +simulators: + - type: pyaml.lattice.simulator + lattice: sr/lattices/ebs.mat + name: design +controls: + - type: tango.pyaml.controlsystem + tango_host: ebs-simu-3:10000 + name: live + catalog: + type: tango.pyaml.tango_catalog +data_folder: /data/store +arrays: + - type: pyaml.arrays.magnet + name: ALL_SERIALIZED + elements: + - QF1A-C* + - QD2A-C* +devices: + - type: pyaml.magnet.serialized_magnet + name: QF1A + function: B1 + elements: + - QF1A-C05 + - QF1A-C06 + - QF1A-C07 + - QF1A-C08 + model: + type: pyaml.magnet.linear_serialized_model + calibration_factors: [1.0, 1.01, 0.99, 1.02] + calibration_offsets: 0.0 + unit: m-1 + hardware_unit: A + curves: + type: pyaml.magnet.csvcurve + file: sr/magnet_models/QF1_strength.csv + powerconverter: srmag/ps-qf1a/series/current + - type: pyaml.magnet.serialized_magnet + name: QD2A + function: B1 + elements: + - QD2A-C05 + - QD2A-C06 + - QD2A-C07 + model: + type: pyaml.magnet.linear_serialized_model + calibration_factors: 1.0 + calibration_offsets: 0.0 + unit: m-1 + hardware_unit: A + curves: + type: pyaml.magnet.csvcurve + file: sr/magnet_models/QD2_strength.csv + powerconverter: srmag/ps-qd2a/series/current diff --git a/tests/magnet/test_serialized_magnets_cs.py b/tests/magnet/test_serialized_magnets_cs.py new file mode 100644 index 00000000..cd4f2998 --- /dev/null +++ b/tests/magnet/test_serialized_magnets_cs.py @@ -0,0 +1,67 @@ +import numpy as np +import pytest + +from pyaml.accelerator import Accelerator +from pyaml.magnet.serialized_magnet import SerializedMagnets + +CONFIG = "tests/config/sr_serialized_magnets_cs.yaml" +TANGO_DUMMY = [{"name": "tango-pyaml", "path": "tests/dummy_cs/tango-pyaml"}] + + +def _sub_strengths(family: SerializedMagnets, current: float) -> np.ndarray: + return np.array([family.model.get_sub_model(i).compute_strengths([current])[0] for i in range(family.get_nb_magnets())]) + + +@pytest.mark.parametrize("install_test_package", TANGO_DUMMY, indirect=True) +def test_serialized_magnets_attach_to_control_system(install_test_package): + sr: Accelerator = Accelerator.load(CONFIG) + family: SerializedMagnets = sr.live.serialized_magnet.get("QF1A") + magnets = family.get_magnets() + + assert family.get_nb_magnets() == 4 + assert [m.get_name() for m in magnets] == ["QF1A-C05", "QF1A-C06", "QF1A-C07", "QF1A-C08"] + for m in magnets: + assert sr.live.magnet.get(m.get_name()) is m + + +@pytest.mark.parametrize("install_test_package", TANGO_DUMMY, indirect=True) +def test_serialized_magnets_share_power_supply(install_test_package): + sr: Accelerator = Accelerator.load(CONFIG) + family: SerializedMagnets = sr.live.serialized_magnet.get("QF1A") + magnets = family.get_magnets() + + magnets[2].hardware.set(80.0) + assert np.allclose([m.hardware.get() for m in magnets], 80.0) + assert np.allclose([m.strength.get() for m in magnets], _sub_strengths(family, 80.0)) + + # Setting the strength of one magnet drives the shared power supply with that magnet's own sub-model + magnets[1].strength.set(0.5) + current = family.model.get_sub_model(1).compute_hardware_values([0.5])[0] + assert np.allclose([m.hardware.get() for m in magnets], current) + assert magnets[1].strength.get() == pytest.approx(0.5) + + # The other family is on its own power supply + assert np.allclose([m.hardware.get() for m in sr.live.serialized_magnet.get("QD2A").get_magnets()], 0.0) + + +@pytest.mark.parametrize("install_test_package", TANGO_DUMMY, indirect=True) +def test_serialized_magnets_aggregator(install_test_package): + sr: Accelerator = Accelerator.load(CONFIG) + qf1a: SerializedMagnets = sr.live.serialized_magnet.get("QF1A") + qd2a: SerializedMagnets = sr.live.serialized_magnet.get("QD2A") + qf1a.get_magnets()[0].hardware.set(30.0) + qd2a.get_magnets()[0].hardware.set(20.0) + + array = sr.live.magnets.get("ALL_SERIALIZED") + expected = np.concatenate([_sub_strengths(qf1a, 30.0), _sub_strengths(qd2a, 20.0)]) + assert np.allclose(array.strengths.get(), expected) + assert np.allclose(array.hardwares.get(), [30.0] * 4 + [20.0] * 3) + + # Doubling every magnet of a series moves its power supply to the current of the doubled strength (no 1/N dilution) + array.strengths.set(2 * array.strengths.get()) + assert np.allclose(array.strengths.get(), 2 * expected) + qf1a_current = qf1a.model.get_sub_model(0).compute_hardware_values([2 * expected[0]])[0] + assert np.allclose(array.hardwares.get()[:4], qf1a_current) + + array.hardwares.set(array.hardwares.get()) + assert np.allclose(array.strengths.get(), 2 * expected)