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