Implement the agreed serialized-magnet accessor convention

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)
 

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
 
 

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):

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):

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