From a87e7d7826dcaf6030995a9d1ea4d6e042a9dc86 Mon Sep 17 00:00:00 2001 From: Alexis Gamelin Date: Tue, 13 Oct 2026 11:55:00 +0200 Subject: [PATCH] Implement the agreed serialized-magnet accessor convention (#473) A serialized family (N magnets on one power supply) has a single degree of freedom, the PS current I. The accessors now follow that in design and live: - family.hardware returns I (was N x I) and sets I. - family.strength returns the average strength sum(f_i(I))/N and sets the I that gives it (exact inverse of the piecewise-linear curves). - magnet[i].strength sets I = f_i^-1(x) with the magnet's own submodel; the other magnets follow. Design no longer splits the value by length. - Array writes give the PS the mean of the currents asked by the series members present in the array, not by all N, so writing one member no longer moves the PS by 1/N. Inconsistent requests log a warning. Design arrays (no aggregator) apply the same rule as live. Combined-function magnets keep their current behaviour: the strength aggregator now calls MagnetModel.compute_hardware_values_partial(), whose default is the previous logic; LinearSerializedMagnetModel overrides it. Also: - Build the serialized submodels once, at configuration. Attaching a family used to rebuild them, so the per-magnet accessors kept stale submodels that never received the magnetic rigidity. - SerializedMagnetsArray.unit() no longer splits unit strings into characters. - Remove ReadWriteSerializedStrengths.elements / get_elements(); use SerializedMagnets.get_magnets(). Tests: new design/live tests for idempotence, family values, per-magnet writes, design/live parity and array writes; tests that relied on the sum and length-sharing convention are updated. Refs #461 Co-Authored-By: Claude Opus 5.5 --- pyaml/arrays/magnet_array.py | 13 +- pyaml/arrays/serialized_magnet_array.py | 4 +- pyaml/control/abstract_impl.py | 10 +- pyaml/lattice/abstract_impl.py | 86 +++++------- pyaml/magnet/linear_serialized_model.py | 108 ++++++++++++++- pyaml/magnet/model.py | 42 ++++++ pyaml/magnet/serialized_magnet.py | 147 +++++++++++---------- tests/lattice/test_magnet_accessors.py | 26 ++-- tests/magnet/test_serialized_magnets.py | 15 ++- tests/magnet/test_serialized_magnets_cs.py | 118 +++++++++++++++++ 10 files changed, 411 insertions(+), 158 deletions(-) diff --git a/pyaml/arrays/magnet_array.py b/pyaml/arrays/magnet_array.py index b24a956f..7f05f35a 100644 --- a/pyaml/arrays/magnet_array.py +++ b/pyaml/arrays/magnet_array.py @@ -66,8 +66,19 @@ def set(self, value: np.array): """ nvalue = np.ones(self.__nb) * value if isinstance(value, float) else value if not self.__aggregator: + # Magnets of a series share one power supply: it is written once, from the strengths of the + # magnets of the series present in the array + series: dict[int, tuple[Magnet, dict[int, float]]] = {} for idx, m in enumerate(self.__magnets): - m.strength.set(nvalue[idx]) + if m.model is not None and m.model.shares_power_supply(): + series.setdefault(id(m.model), (m, {}))[1][m.strength.index()] = nvalue[idx] + else: + m.strength.set(nvalue[idx]) + for m, requested in series.values(): + if len(requested) == 1: + m.strength.set(next(iter(requested.values()))) + else: + m.hardware.set(m.model.compute_hardware_values_partial(None, requested)[0]) else: self.__aggregator.set(nvalue) diff --git a/pyaml/arrays/serialized_magnet_array.py b/pyaml/arrays/serialized_magnet_array.py index c891e5c6..3022e2ff 100644 --- a/pyaml/arrays/serialized_magnet_array.py +++ b/pyaml/arrays/serialized_magnet_array.py @@ -86,7 +86,7 @@ def unit(self) -> list[str]: """Return the strength unit of every serialized group.""" r = [] for m in self.__magnets: - r.extend(m.strength.unit()) + r.append(m.strength.unit()) return r @@ -162,7 +162,7 @@ def unit(self) -> list[str]: """Return the hardware unit of every serialized group.""" r = [] for m in self.__magnets: - r.extend(m.hardware.unit()) + r.append(m.hardware.unit()) return r diff --git a/pyaml/control/abstract_impl.py b/pyaml/control/abstract_impl.py index 5b110562..ffea3234 100644 --- a/pyaml/control/abstract_impl.py +++ b/pyaml/control/abstract_impl.py @@ -431,7 +431,7 @@ def add_magnet(self, magnet: Magnet, devs: list[DeviceAccess]): Hardware devices associated with the magnet model. """ # 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). + # the power-supply current is computed by the model from the strengths of the magnets present in the array. is_indexed = isinstance(magnet.strength, (abstract.RWMapper, RWSerializedStrengthScalar)) strengthIndex = magnet.strength.index() if is_indexed else 0 if magnet.model not in self.__models: @@ -458,10 +458,10 @@ def set(self, value: NDArray[np.float64]): hardwareIndex = 0 for modelIndex, model in enumerate(self.__models): nbDev = len(model.get_device_names()) - mStrengths = model.compute_strengths(allHardwareValues[hardwareIndex : hardwareIndex + nbDev]) - for valueIdx, strengthIdx in self.__modelToMagnet[modelIndex]: - mStrengths[strengthIdx] = value[valueIdx] - newHardwareValues[hardwareIndex : hardwareIndex + nbDev] = model.compute_hardware_values(mStrengths) + requested = {strengthIdx: value[valueIdx] for valueIdx, strengthIdx in self.__modelToMagnet[modelIndex]} + newHardwareValues[hardwareIndex : hardwareIndex + nbDev] = model.compute_hardware_values_partial( + allHardwareValues[hardwareIndex : hardwareIndex + nbDev], requested + ) hardwareIndex += nbDev dev_range = self._devs.get_range() if not check_range(newHardwareValues, dev_range): diff --git a/pyaml/lattice/abstract_impl.py b/pyaml/lattice/abstract_impl.py index a734a587..b506d4bc 100644 --- a/pyaml/lattice/abstract_impl.py +++ b/pyaml/lattice/abstract_impl.py @@ -277,7 +277,10 @@ def get_model(self) -> MagnetModel: class RWSerializedHardware(abstract.ReadWriteFloatScalar): """ - RWSerializedHardware configuration or runtime object. + Hardware value of one magnet of a serialized group. + + All the magnets of the group share one power supply: writing the hardware + value of one magnet writes it to every magnet of the group. Parameters ---------- @@ -288,18 +291,18 @@ class RWSerializedHardware(abstract.ReadWriteFloatScalar): Methods ------- - get_element_length() - Return the total length of the represented element. - get_total_length() - Return the total length of the represented elements. + index() + Return the position of the magnet in the serialized group. get() Return the current value. set(value) - Set the current value. + Set the current value of the whole serialized group. set_and_wait(value) Set the value and wait for readback convergence. unit() Return the value unit. + get_model() + Return the submodel of the magnet. set_magnet_rigidity(brho) Set the magnetic rigidity used for conversion. """ @@ -310,17 +313,10 @@ def __init__(self, elements: list[RWHardwareScalar], element_index: int): """ self.__elements = elements self.__element_index = element_index - self.__total_length = 0 - for e in elements: - self.__total_length += e.get_length() - def get_element_length(self) -> float: - """Return the total length of the represented element.""" - return self.__elements[self.__element_index].get_length() - - def get_total_length(self) -> float: - """Return the total length of the represented elements.""" - return self.__total_length + def index(self) -> int: + """Return the position of the magnet in the serialized group.""" + return self.__element_index # Gets the value def get(self) -> float: @@ -330,7 +326,7 @@ def get(self) -> float: # Sets the value def set(self, value: float): """ - Set the current value. + Set the current value of the whole serialized group. Parameters ---------- @@ -361,6 +357,10 @@ def unit(self) -> str: """Return the value unit.""" return self.__elements[self.__element_index].unit() + def get_model(self) -> MagnetModel: + """Return the submodel of the magnet.""" + return self.__elements[self.__element_index].get_model() + def set_magnet_rigidity(self, brho: np.double): """ Set the magnetic rigidity used for conversion. @@ -375,7 +375,11 @@ def set_magnet_rigidity(self, brho: np.double): class RWSerializedStrength(abstract.ReadWriteFloatScalar): """ - RWSerializedStrength configuration or runtime object. + Strength of one magnet of a serialized group. + + All the magnets of the group share one power supply. Writing the strength + ``x`` of magnet ``i`` sets the shared hardware value to ``I = f_i^-1(x)``, + using the submodel of the magnet, and the other magnets follow as ``f_j(I)``. Parameters ---------- @@ -388,10 +392,8 @@ class RWSerializedStrength(abstract.ReadWriteFloatScalar): Methods ------- - get_element_length() - Return the total length of the represented element. - get_total_length() - Return the total length of the represented elements. + index() + Return the position of the magnet in the serialized group. get() Return the current value. set(value) @@ -414,33 +416,17 @@ def __init__( Initialize the RWSerializedStrength. """ self.__element = elements_strength[element_index] - self.__elements_strength = elements_strength self.__elements_hardware = elements_hardware self.__element_index = element_index - # The strength is shared between the serialized magnets in proportion to their (effective) length: thick - # magnets share by length, thin magnets share equally. Mixing both has no meaningful share rule. - thin_flags = [e.is_thin() for e in elements_hardware] - if any(thin_flags) and not all(thin_flags): - raise PyAMLException( - "Serialized magnets must be either all thin (zero length) or all thick; " - f"got {sum(thin_flags)} thin out of {len(thin_flags)} magnets" - ) - self.__total_length = 0 - for e in self.__elements_hardware: - self.__total_length += e.get_length() - - def get_element_length(self) -> float: - """Return the total length of the represented element.""" - return self.__element.get_element_length() - def get_total_length(self) -> float: - """Return the total length of the represented elements.""" - return self.__total_length + def index(self) -> int: + """Return the position of the magnet in the serialized group.""" + return self.__element_index # Gets the value def get(self) -> float: """Return the current value.""" - return self.__elements_strength[self.__element_index].get() + return self.__element.get() # Sets the value def set(self, value: float): @@ -452,19 +438,9 @@ def set(self, value: float): value : float Strength to apply to the magnet, converted to the shared hardware setpoint. """ - elements_values = [value * e.get_length() / self.get_total_length() for e in self.__elements_hardware] - self.__element.set(elements_values[self.__element_index]) - - # compute the local hardware value - hardware_value = self.__elements_hardware[self.__element_index].get() - - # compute the total hardware value - total_hardware = hardware_value * self.get_total_length() / self.get_element_length() - - # dispatch this value - for index, element in enumerate(self.__elements_hardware): - if index != self.__element_index: - element.set(total_hardware * element.get_length() / self.get_total_length()) + model = self.__elements_hardware[self.__element_index].get_model() + hardware_value = model.compute_hardware_values([value])[0] + [element.set(hardware_value) for element in self.__elements_hardware] # Sets the value and wait that the read value reach the setpoint def set_and_wait(self, value: float): diff --git a/pyaml/magnet/linear_serialized_model.py b/pyaml/magnet/linear_serialized_model.py index 1e912c11..6915c862 100644 --- a/pyaml/magnet/linear_serialized_model.py +++ b/pyaml/magnet/linear_serialized_model.py @@ -4,6 +4,8 @@ This model applies linear calibration to a configurable sequence of serialized magnet elements and hardware channels. """ +import logging + import numpy as np from ..common.element import __pyaml_repr__ @@ -17,6 +19,11 @@ # Define the main class name for this module PYAMLCLASS = "LinearSerializedMagnetModel" +logger = logging.getLogger(__name__) + +# Relative spread above which the currents requested by members of a series are considered inconsistent +_SPREAD_RTOL = 1e-6 + def _get_length(elem) -> int: """ @@ -132,6 +139,11 @@ class LinearSerializedMagnetModel(MagnetModel, DynamicValidation): hardware values. Conversely, a hardware current is mapped back to a strength value for each sub-model. + The serialized magnets have a single degree of freedom, the power-supply + current ``I``. Magnet ``i`` has the strength ``k_i = f_i(I)``, where ``f_i`` + is its sub-model. The family strength is the average strength + ``sum(f_i(I)) / N``. + Parameters ---------- curves : Curve or list[Curve] @@ -162,6 +174,14 @@ class LinearSerializedMagnetModel(MagnetModel, DynamicValidation): Convert magnet strengths to hardware values. compute_strengths(currents) Convert hardware values to magnet strengths. + compute_hardware_values_partial(current_hardware, strengths_by_index) + Convert the strengths of some magnets of the series to the shared hardware value. + compute_mean_strength(current) + Return the average strength of the series for a hardware value. + compute_hardware_from_mean_strength(strength) + Return the hardware value giving an average strength of the series. + shares_power_supply() + Return True, all the magnets of the series share one power supply. get_strength_units() Return the units of magnet strengths. get_hardware_units() @@ -312,7 +332,7 @@ def compute_hardware_values(self, strengths: np.array) -> np.array: Returns ------- np.array - Hardware value obtained from the submodels. + Hardware value obtained from the submodels: the mean of the hardware values implied by each strength. """ currents = [model.compute_hardware_values([s])[0] for s, model in zip(strengths, self.__sub_models, strict=True)] return np.array([np.mean(currents)]) @@ -334,6 +354,92 @@ def compute_strengths(self, currents: np.array) -> np.array: current = currents[0] return np.array([model.compute_strengths([current])[0] for model in self.__sub_models]) + def compute_hardware_values_partial( + self, current_hardware: np.ndarray | None, strengths_by_index: dict[int, float] + ) -> np.ndarray: + """ + Convert the strengths of some magnets of the series to the shared hardware value. + + Each requested strength implies a hardware value through its own submodel. The power supply gets the mean of + the hardware values implied by the requested magnets only: the magnets that are not requested do not take part. + A warning is logged when the requested strengths are not consistent with a single hardware value. + + Parameters + ---------- + current_hardware : np.array | None + Current hardware value of the series. Not used: the requested strengths define the hardware value. + strengths_by_index : dict[int, float] + Requested strengths, indexed by the position of the magnet in the series. + + Returns + ------- + np.array + Hardware value of the power supply. + """ + currents = np.array( + [self.__sub_models[idx].compute_hardware_values([strength])[0] for idx, strength in strengths_by_index.items()] + ) + current = np.mean(currents) + if np.ptp(currents) > _SPREAD_RTOL * max(abs(current), 1.0): + logger.warning( + f"Inconsistent strengths requested for magnets {list(strengths_by_index.keys())} of a series on " + f"{self._powerconverter}: implied hardware values {currents}, the mean {current} is applied" + ) + return np.array([current]) + + def compute_mean_strength(self, current: float) -> float: + """ + Return the average strength of the series for a hardware value. + + Parameters + ---------- + current : float + Hardware value shared by the serialized magnets. + + Returns + ------- + float + Average strength ``sum(f_i(I)) / N``. + """ + return float(np.mean(self.compute_strengths([current]))) + + def compute_hardware_from_mean_strength(self, strength: float) -> float: + """ + Return the hardware value giving an average strength of the series. + + The submodels are piecewise linear in the hardware value, so the average strength is piecewise linear on the + union of the curve breakpoints and is inverted exactly by interpolation. + + Parameters + ---------- + strength : float + Requested average strength of the series. + + Returns + ------- + float + Hardware value ``I`` such that ``sum(f_i(I)) / N == strength``. + + Raises + ------ + PyAMLException + If the average strength is not monotonic in the hardware value. + """ + currents = np.unique(np.concatenate([curve.get_curve()[:, 0] for curve in self.__curves])) + strengths = np.array([self.compute_mean_strength(current) for current in currents]) + steps = np.diff(strengths) + if np.all(steps <= 0): + currents, strengths = currents[::-1], strengths[::-1] + elif not np.all(steps >= 0): + raise PyAMLException( + f"Average strength of the series on {self._powerconverter} is not monotonic, it cannot be inverted" + ) + return float(np.interp(strength, strengths, currents)) + + def shares_power_supply(self) -> bool: + """Return True, all the magnets of the series share one power supply.""" + return True + def get_strength_units(self) -> list[str]: """Return the units of magnet strengths.""" return [self._unit] * self.__nbMagnets diff --git a/pyaml/magnet/model.py b/pyaml/magnet/model.py index 47ffe366..cfcdd88a 100644 --- a/pyaml/magnet/model.py +++ b/pyaml/magnet/model.py @@ -34,6 +34,10 @@ class MagnetModel(metaclass=ABCMeta): Tells if the model allows to work in hardware unit. has_physics() Tells if the model allows to work in physics unit. + shares_power_supply() + Tells if all the strengths of the model are driven by one power supply. + compute_hardware_values_partial(current_hardware, strengths_by_index) + Compute hardware value(s) when only some strengths are requested. """ @abstractmethod @@ -130,3 +134,41 @@ def has_physics(self) -> bool: True if the model supports physics unit """ return True + + def shares_power_supply(self) -> bool: + """ + Tells if all the strengths of the model are driven by one power supply. + + Returns + ---------- + bool + True if the model has a single degree of freedom shared by all its strengths (serialized magnets) + """ + return False + + def compute_hardware_values_partial( + self, + current_hardware: npt.NDArray[np.float64] | None, + strengths_by_index: dict[int, float], + ) -> npt.NDArray[np.float64]: + """ + Compute hardware value(s) when only some strengths of the model are requested. + + The strengths that are not requested keep the value computed from the current hardware values. + + Parameters + ---------- + current_hardware : npt.NDArray[np.float64] | None + Current hardware values of the model. + strengths_by_index : dict[int, float] + Requested strengths, indexed by their position in the model. + + Returns + ------- + npt.NDArray[np.float64] + Array of hardware values (i.e. currents or voltages). + """ + strengths = self.compute_strengths(current_hardware) + for idx, value in strengths_by_index.items(): + strengths[idx] = value + return self.compute_hardware_values(strengths) diff --git a/pyaml/magnet/serialized_magnet.py b/pyaml/magnet/serialized_magnet.py index ddfe1733..44cf55c1 100644 --- a/pyaml/magnet/serialized_magnet.py +++ b/pyaml/magnet/serialized_magnet.py @@ -25,69 +25,66 @@ PYAMLCLASS = "SerializedMagnets" -class ReadWriteSerializedStrengths(abstract.ReadWriteFloatScalar): +class ReadWriteSerializedHardware(abstract.ReadWriteFloatScalar): """ - Read/write aggregate for serialized-magnet strengths. + Read/write access to the hardware value of a serialized-magnet family. + + The family has a single degree of freedom, the hardware value ``I`` of its + power supply, shared by every magnet of the series. Parameters ---------- - elements : list[abstract.ReadWriteFloatScalar] - Per-magnet accessors sharing the group setpoint. - model : MagnetModel | None - Magnet model used to convert between strength and hardware value. Optional for a hardware-only group. + hardwares : list[abstract.ReadWriteFloatScalar] + Per-magnet hardware accessors. Writing one of them moves the whole series. + model : MagnetModel + Serialized magnet model used for units and rigidity. Methods ------- get() - Return the sum of the serialized element strengths. + Return the shared hardware value ``I``. set(value) - Set the shared serialized-magnet strength. + Set the shared hardware value ``I``. set_and_wait(value) - Set the shared strength and wait for convergence. + Set the shared hardware value and wait for convergence. unit() - Return the physical strength unit. + Return the hardware-value unit. get_model() Return the magnet conversion model. - get_elements() - Return the underlying scalar element accessors. set_magnet_rigidity(brho) Set the magnetic rigidity used for conversion. """ - def __init__( - self, - elements: list[abstract.ReadWriteFloatScalar], - model: MagnetModel | None = None, - ): + def __init__(self, hardwares: list[abstract.ReadWriteFloatScalar], model: MagnetModel): """ - Initialize a shared strength accessor for serialized elements. + Initialize a family hardware accessor for serialized magnets. """ - self.elements = elements - self.model = model + self._hardwares = hardwares + self._model = model def get(self) -> float: - """Return the sum of the serialized element strengths.""" - return sum([elem.get() for elem in self.elements]) + """Return the shared hardware value ``I``.""" + return self._hardwares[0].get() def set(self, value: float): """ - Set the shared serialized-magnet strength. + Set the shared hardware value ``I``. Parameters ---------- value : float - Strength applied to every magnet of the group. + Hardware value applied to the power supply of the series. """ - self.elements[0].set(value) + self._hardwares[0].set(value) def set_and_wait(self, value: float): """ - Set the shared strength and wait for convergence. + Set the shared hardware value and wait for convergence. Parameters ---------- value : float - Strength applied to every magnet of the group. + Hardware value applied to the power supply of the series. Raises ------ @@ -97,16 +94,12 @@ def set_and_wait(self, value: float): raise NotImplementedError("Not implemented yet.") def unit(self) -> str: - """Return the physical strength unit.""" - return self.model.get_strength_units()[0] + """Return the hardware-value unit.""" + return self._model.get_hardware_units()[0] def get_model(self) -> MagnetModel: """Return the magnet conversion model.""" - return self.model - - def get_elements(self): - """Return the underlying scalar element accessors.""" - return self.elements + return self._model def set_magnet_rigidity(self, brho: np.double): """ @@ -117,52 +110,51 @@ def set_magnet_rigidity(self, brho: np.double): brho : np.double Magnetic rigidity in tesla metres, forwarded to the magnet model. """ - [element.set_magnet_rigidity(brho) for element in self.elements] + self._model.set_magnet_rigidity(brho) -class ReadWriteSerializedHardwares(ReadWriteSerializedStrengths): +class ReadWriteSerializedStrength(ReadWriteSerializedHardware): """ - Read/write aggregate for serialized-magnet hardware values. + Read/write access to the average strength of a serialized-magnet family. + + Magnet ``i`` of the series has the strength ``k_i = f_i(I)``. The family + strength is the average strength ``sum(f_i(I)) / N``. Parameters ---------- - elements : list[abstract.ReadWriteFloatScalar] - Per-magnet accessors sharing the group setpoint. - model : MagnetModel | None - Magnet model used to convert between strength and hardware value. Optional for a hardware-only group. + hardwares : list[abstract.ReadWriteFloatScalar] + Per-magnet hardware accessors. Writing one of them moves the whole series. + model : MagnetModel + Serialized magnet model converting between average strength and hardware value. Methods ------- + get() + Return the average strength of the series. + set(value) + Set the hardware value giving the requested average strength. unit() - Return the hardware-value unit. - set_magnet_rigidity(brho) - Set the magnetic rigidity used for conversion. + Return the physical strength unit. """ - def __init__( - self, - elements: list[abstract.ReadWriteFloatScalar], - model: MagnetModel | None = None, - ): - """ - Initialize a shared hardware accessor for serialized elements. - """ - super().__init__(elements, model) - - def unit(self) -> str: - """Return the hardware-value unit.""" - return self.model.get_hardware_units()[0] + def get(self) -> float: + """Return the average strength of the series.""" + return self._model.compute_mean_strength(self._hardwares[0].get()) - def set_magnet_rigidity(self, brho: np.double): + def set(self, value: float): """ - Set the magnetic rigidity used for conversion. + Set the hardware value giving the requested average strength. Parameters ---------- - brho : np.double - Magnetic rigidity in tesla metres, forwarded to the magnet model. + value : float + Average strength of the series. """ - [element.set_magnet_rigidity(brho) for element in self.elements] + self._hardwares[0].set(self._model.compute_hardware_from_mean_strength(value)) + + def unit(self) -> str: + """Return the physical strength unit.""" + return self._model.get_strength_units()[0] @register_schema @@ -193,9 +185,9 @@ class SerializedMagnets(Element, DynamicValidation): Attributes ---------- strength - Gives access to the strengths of those magnets in physics unit + Average strength of the magnets of the series, in physics unit hardware - Gives access to the strengths of this those magnets in hardware unit when possible + Hardware value of the power supply shared by the series Methods ------- @@ -220,7 +212,19 @@ class SerializedMagnets(Element, DynamicValidation): ----- The serialized group stores a virtual magnet for each underlying element. When attached, each virtual magnet is bound to the same peer and the group exposes - aggregate strength and hardware accessors. + family strength and hardware accessors. + + The series has a single degree of freedom, the power-supply hardware value ``I``. + Magnet ``i`` has the strength ``k_i = f_i(I)``: + + ====================== =============================== ================================================== + Accessor ``get()`` ``set(x)`` + ====================== =============================== ================================================== + ``family.hardware`` ``I`` write ``I = x`` + ``family.strength`` average ``sum(f_i(I)) / N`` solve ``sum(f_i(I)) / N = x`` for ``I``, write it + ``magnet[i].hardware`` ``I`` write ``I = x``, all the magnets follow + ``magnet[i].strength`` ``k_i = f_i(I)`` write ``I = f_i^-1(x)``, the others follow + ====================== =============================== ================================================== Configuration ------------- @@ -260,9 +264,10 @@ def __init__( self.__hardwares = None self.__virtuals: list[Magnet] = [] self.__elements = elements if isinstance(elements, list) else [elements] - self.model.set_number_of_magnets(len(self.__elements)) if peer is None: # Configuration part + # The submodels are built once here: the attached copies share them with the per-magnet accessors + self.model.set_number_of_magnets(len(self.__elements)) self.polynom = function_map[self.function].polynom if self.function not in function_map: raise PyAMLException(self.function + " not implemented for serialized magnet") @@ -335,8 +340,10 @@ def attach( """ l = [] n_ser_mag = SerializedMagnets(self._name, self.function, self.__elements, self.model, self.description, peer) - n_ser_mag.__strengths = ReadWriteSerializedStrengths(strengths, self.model) - n_ser_mag.__hardwares = ReadWriteSerializedHardwares(hardwares, self.model) + if self.model.has_hardware(): + n_ser_mag.__hardwares = ReadWriteSerializedHardware(hardwares, self.model) + if self.model.has_physics(): + n_ser_mag.__strengths = ReadWriteSerializedStrength(hardwares, self.model) l.append(n_ser_mag) # Construct a single magnet for each magnet. sub_magnets: list[Magnet] = [] @@ -351,7 +358,7 @@ def attach( @property def strength(self) -> abstract.ReadWriteFloatScalar: """ - Gives access to the strengths of those magnets in physics unit + Average strength of the magnets of the series, in physics unit """ self.check_peer() if self.__strengths is None: @@ -361,7 +368,7 @@ def strength(self) -> abstract.ReadWriteFloatScalar: @property def hardware(self) -> abstract.ReadWriteFloatScalar: """ - Gives access to the strengths of this those magnets in hardware unit when possible + Hardware value of the power supply shared by the series """ self.check_peer() if self.__hardwares is None: diff --git a/tests/lattice/test_magnet_accessors.py b/tests/lattice/test_magnet_accessors.py index e585b37e..28c7a924 100644 --- a/tests/lattice/test_magnet_accessors.py +++ b/tests/lattice/test_magnet_accessors.py @@ -2,7 +2,6 @@ import numpy as np import pytest -from pyaml import PyAMLException from pyaml.lattice.abstract_impl import ( RWHardwareArray, RWHardwareScalar, @@ -75,8 +74,10 @@ def test_zero_length_combined_function_magnet_stores_integrated_strengths(access np.testing.assert_allclose(accessor.get(), [1.0e-6, -2.0e-6]) -def test_zero_length_serialized_magnets_share_strength_equally(): - elements = [_thin_corrector("COR_A"), _thin_corrector("COR_B")] +def test_serialized_magnets_follow_the_strength_of_one_magnet(): + # Serialized magnets share one power supply: setting the strength of one magnet sets the shared hardware value, + # so every magnet with the same submodel gets the same strength (no length weighting). + elements = [_thin_corrector("COR_A"), at.Corrector("COR_B", 0.2, [0.0, 0.0], PolynomA=[0.0], PolynomB=[0.0])] model = IdentityMagnetModel(physics="COR", unit="rad") strengths = [RWStrengthScalar([e], HCorrector.polynom, model) for e in elements] currents = [RWHardwareScalar([e], HCorrector.polynom, model) for e in elements] @@ -84,17 +85,8 @@ def test_zero_length_serialized_magnets_share_strength_equally(): serialized[0].set(1.0e-6) - assert serialized[0].get() == pytest.approx(0.5e-6) - assert serialized[1].get() == pytest.approx(0.5e-6) - assert elements[0].PolynomB[0] == pytest.approx(-0.5e-6) - assert elements[1].PolynomB[0] == pytest.approx(-0.5e-6) - - -def test_serialized_magnets_mixing_thin_and_thick_is_rejected(): - elements = [_thin_corrector("COR_A"), at.Corrector("COR_B", 0.2, [0.0, 0.0], PolynomA=[0.0], PolynomB=[0.0])] - model = IdentityMagnetModel(physics="COR", unit="rad") - strengths = [RWStrengthScalar([e], HCorrector.polynom, model) for e in elements] - currents = [RWHardwareScalar([e], HCorrector.polynom, model) for e in elements] - - with pytest.raises(PyAMLException, match="all thin .* or all thick"): - RWSerializedStrength(strengths, currents, 0) + assert serialized[0].get() == pytest.approx(1.0e-6) + assert serialized[1].get() == pytest.approx(1.0e-6) + assert elements[0].PolynomB[0] == pytest.approx(-1.0e-6) + assert elements[1].PolynomB[0] * elements[1].Length == pytest.approx(-1.0e-6) + assert serialized[1].index() == 1 diff --git a/tests/magnet/test_serialized_magnets.py b/tests/magnet/test_serialized_magnets.py index b362f7b7..564ab6c1 100644 --- a/tests/magnet/test_serialized_magnets.py +++ b/tests/magnet/test_serialized_magnets.py @@ -74,8 +74,8 @@ def test_magnet_modification(sr_file): print() print("Reading strengths from serialized magnets") - for ii in range(len(sm.strength.elements)): - print(element_names[ii], sm.strength.elements[ii].get()) + for ii, magnet in enumerate(sm.get_magnets()): + print(element_names[ii], magnet.strength.get()) print() strength0 = sm.strength.get() @@ -96,8 +96,8 @@ def test_magnet_modification(sr_file): print() print("Reading strengths from serialized magnets") - for ii in range(len(sm.strength.elements)): - print(element_names[ii], sm.strength.elements[ii].get()) + for ii, magnet in enumerate(sm.get_magnets()): + print(element_names[ii], magnet.strength.get()) @pytest.mark.parametrize( @@ -194,10 +194,11 @@ def test_strength_computation(sr_file): sr: Accelerator = Accelerator.load(sr_file, include_locations=False, ignore_external=True) sm: SerializedMagnets = sr.design.serialized_magnet.get("QF1A") assert sm.get_nb_magnets() == 31 - sm.strength.set(24.0) - assert abs(sm.strength.get() - 24.0) < 1e-3 + sm.strength.set(0.8) + assert abs(sm.strength.get() - 0.8) < 1e-6 magnets_strengths = [m.strength.get() for m in sm.get_magnets()] magnets_from_lattice_strengths = get_strengths_from_lattice(sr, sm) - assert abs(sum(magnets_from_lattice_strengths) - 24.0) < 1e-3 + # The family strength is the average strength of the magnets + assert abs(np.mean(magnets_from_lattice_strengths) - 0.8) < 1e-6 assert magnets_strengths == magnets_from_lattice_strengths diff --git a/tests/magnet/test_serialized_magnets_cs.py b/tests/magnet/test_serialized_magnets_cs.py index cd4f2998..1cc7a3e3 100644 --- a/tests/magnet/test_serialized_magnets_cs.py +++ b/tests/magnet/test_serialized_magnets_cs.py @@ -65,3 +65,121 @@ def test_serialized_magnets_aggregator(install_test_package): array.hardwares.set(array.hardwares.get()) assert np.allclose(array.strengths.get(), 2 * expected) + + +MODES = ["design", "live"] + + +def _family(sr: Accelerator, mode: str, name: str = "QF1A") -> SerializedMagnets: + return getattr(sr, mode).serialized_magnet.get(name) + + +@pytest.mark.parametrize("mode", MODES) +@pytest.mark.parametrize("install_test_package", TANGO_DUMMY, indirect=True) +def test_family_hardware_is_the_power_supply_current(install_test_package, mode): + sr: Accelerator = Accelerator.load(CONFIG) + family = _family(sr, mode) + family.hardware.set(80.0) + assert family.hardware.get() == pytest.approx(80.0) + assert np.allclose([m.hardware.get() for m in family.get_magnets()], 80.0) + + +@pytest.mark.parametrize("mode", MODES) +@pytest.mark.parametrize("install_test_package", TANGO_DUMMY, indirect=True) +def test_set_of_get_does_not_change_the_machine(install_test_package, mode): + sr: Accelerator = Accelerator.load(CONFIG) + family = _family(sr, mode) + magnet = family.get_magnets()[2] + family.hardware.set(80.0) + for accessor in [family.hardware, family.strength, magnet.hardware, magnet.strength]: + accessor.set(accessor.get()) + assert family.hardware.get() == pytest.approx(80.0) + + +@pytest.mark.parametrize("mode", MODES) +@pytest.mark.parametrize("install_test_package", TANGO_DUMMY, indirect=True) +def test_family_strength_is_the_average_strength(install_test_package, mode): + sr: Accelerator = Accelerator.load(CONFIG) + family = _family(sr, mode) + family.hardware.set(80.0) + assert family.strength.get() == pytest.approx(np.mean(_sub_strengths(family, 80.0))) + + # QF1A has a different calibration factor per magnet: the requested value is the average strength + family.strength.set(0.79) + assert family.strength.get() == pytest.approx(0.79) + assert np.mean([m.strength.get() for m in family.get_magnets()]) == pytest.approx(0.79) + assert family.hardware.get() == pytest.approx(family.model.compute_hardware_from_mean_strength(0.79)) + + +@pytest.mark.parametrize("mode", MODES) +@pytest.mark.parametrize("install_test_package", TANGO_DUMMY, indirect=True) +def test_magnet_strength_drives_the_series_with_its_own_submodel(install_test_package, mode): + sr: Accelerator = Accelerator.load(CONFIG) + family = _family(sr, mode) + magnets = family.get_magnets() + magnets[1].strength.set(0.5) + current = family.model.get_sub_model(1).compute_hardware_values([0.5])[0] + assert family.hardware.get() == pytest.approx(current) + assert magnets[1].strength.get() == pytest.approx(0.5) + assert np.allclose([m.strength.get() for m in magnets], _sub_strengths(family, current)) + + +@pytest.mark.parametrize("install_test_package", TANGO_DUMMY, indirect=True) +def test_design_and_live_agree(install_test_package): + sr: Accelerator = Accelerator.load(CONFIG) + design, live = _family(sr, "design"), _family(sr, "live") + design.hardware.set(80.0) + live.hardware.set(80.0) + assert design.strength.get() == pytest.approx(live.strength.get()) + assert np.allclose( + [m.strength.get() for m in design.get_magnets()], + [m.strength.get() for m in live.get_magnets()], + ) + + +@pytest.mark.parametrize("mode", MODES) +@pytest.mark.parametrize("install_test_package", TANGO_DUMMY, indirect=True) +def test_array_with_one_magnet_of_a_series_moves_the_whole_series(install_test_package, mode): + from pyaml.arrays.magnet_array import MagnetArray + + sr: Accelerator = Accelerator.load(CONFIG) + family = _family(sr, mode) + family.hardware.set(80.0) + array = MagnetArray("ONE", [family.get_magnets()[1]]) + array.strengths.set(np.array([0.5])) + current = family.model.get_sub_model(1).compute_hardware_values([0.5])[0] + assert family.hardware.get() == pytest.approx(current) + assert array.strengths.get() == pytest.approx([0.5]) + + +@pytest.mark.parametrize("mode", MODES) +@pytest.mark.parametrize("install_test_package", TANGO_DUMMY, indirect=True) +def test_array_with_consistent_magnets_of_a_series(install_test_package, mode): + sr: Accelerator = Accelerator.load(CONFIG) + qf1a, qd2a = _family(sr, mode, "QF1A"), _family(sr, mode, "QD2A") + qf1a.hardware.set(80.0) + qd2a.hardware.set(20.0) + array = getattr(sr, mode).magnets.get("ALL_SERIALIZED") + array.strengths.set(array.strengths.get()) + assert qf1a.hardware.get() == pytest.approx(80.0) + assert qd2a.hardware.get() == pytest.approx(20.0) + + +@pytest.mark.parametrize("mode", MODES) +@pytest.mark.parametrize("install_test_package", TANGO_DUMMY, indirect=True) +def test_array_with_inconsistent_magnets_of_a_series(install_test_package, mode, caplog): + sr: Accelerator = Accelerator.load(CONFIG) + qf1a, qd2a = _family(sr, mode, "QF1A"), _family(sr, mode, "QD2A") + qf1a.hardware.set(80.0) + qd2a.hardware.set(20.0) + array = getattr(sr, mode).magnets.get("ALL_SERIALIZED") + strengths = array.strengths.get() + requested = np.array([0.78, 0.79, 0.80, 0.81]) + strengths[:4] = requested + with caplog.at_level("WARNING"): + array.strengths.set(strengths) + # The power supply gets the mean of the currents asked by the magnets of the series + expected = np.mean([qf1a.model.get_sub_model(i).compute_hardware_values([s])[0] for i, s in enumerate(requested)]) + assert qf1a.hardware.get() == pytest.approx(expected) + assert qd2a.hardware.get() == pytest.approx(20.0) + assert "Inconsistent strengths" in caplog.text