Serialized magnets: mixed-function families, strengths/hardwares arrays and partial series writes

pyaml/arrays/magnet_array.py

@@ -67,18 +67,16 @@ 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
+            # magnets of the series present in the array, the others following the resulting hardware value
             series: dict[int, tuple[Magnet, dict[int, float]]] = {}
             for idx, m in enumerate(self.__magnets):
                 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])
+                current_hardware = np.array([m.hardware.get()])
+                m.hardware.set(m.model.compute_hardware_values_partial(current_hardware, requested)[0])
         else:
             self.__aggregator.set(nvalue)
 

pyaml/control/abstract_impl.py

@@ -431,7 +431,8 @@ 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 strengths of the magnets present in the array.
+        # the power-supply current is computed by the model from the strengths of the magnets present in the array,
+        # the others following the resulting current.
         is_indexed = isinstance(magnet.strength, (abstract.RWMapper, RWSerializedStrengthScalar))
         strengthIndex = magnet.strength.index() if is_indexed else 0
         if magnet.model not in self.__models:
@@ -719,36 +720,66 @@ 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.
+    reads with the magnet's own sub-model, writes through the model of the
+    series, 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.
+        Sub-model of the magnet, used for strength reading and units.
     dev : DeviceAccess
         Power-supply device shared by the serialized group.
     idx : int
         Zero-based position of the magnet in the serialized group.
+    series_model : MagnetModel
+        Model of the series, used to convert the strength of the magnet to the shared hardware value.
 
     Methods
     -------
     index()
         Return the position of the magnet in the serialized group.
+    set(value)
+        Set the shared hardware value giving the requested strength of the magnet.
     """
 
-    def __init__(self, model: MagnetModel, dev: DeviceAccess, idx: int):
+    def __init__(self, model: MagnetModel, dev: DeviceAccess, idx: int, series_model: MagnetModel):
         """
         Initialize the RWSerializedStrengthScalar.
         """
         super().__init__(model, dev)
+        self.__dev = dev
         self.__idx = idx
+        self.__series_model = series_model
 
     def index(self) -> int:
         """Return the position of the magnet in the serialized group."""
         return self.__idx
 
+    def set(self, value: float):
+        """
+        Set the shared hardware value giving the requested strength of the magnet.
+
+        The model of the series solves for the hardware value from the present one, so that the power supply does not
+        move when it already gives the requested strength (for example on a plateau of the excitation curve).
+
+        Parameters
+        ----------
+        value : float
+            Strength of the magnet.
+
+        Raises
+        ------
+        PyAMLException
+            If the converted hardware value is outside the device's configured range.
+        """
+        current_hardware = np.array([self.__dev.get()])
+        current = self.__series_model.compute_hardware_values_partial(current_hardware, {self.__idx: value})[0]
+        dev_range = self.__dev.get_range()
+        if not check_range(current, dev_range):
+            raise PyAMLException(format_out_of_range_message(current, self.__dev))
+        self.__dev.set(current)
+
 
 # ------------------------------------------------------------------------------
 

pyaml/control/controlsystem.py

@@ -238,7 +238,9 @@ def _fill_serialized_magnets(self, magnets: SerializedMagnets) -> None:
         for index in range(magnets.get_nb_magnets()):
             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)
+            strengths.append(
+                RWSerializedStrengthScalar(sub_model, device, index, magnets.model) 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:]:

pyaml/lattice/abstract_impl.py

@@ -378,8 +378,11 @@ class RWSerializedStrength(abstract.ReadWriteFloatScalar):
     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)``.
+    ``x`` of magnet ``i`` sets the shared hardware value ``I`` solving
+    ``f_i(I) = x``, and the other magnets follow as ``f_j(I)``. With the model of
+    the series, the solution starts from the present hardware value, so that the
+    power supply does not move when it already gives ``x`` (for example on a
+    plateau of the excitation curve).
 
     Parameters
     ----------
@@ -389,6 +392,9 @@ class RWSerializedStrength(abstract.ReadWriteFloatScalar):
         Hardware accessors of the same magnets, used to apply the shared setpoint.
     element_index : int
         Index of the magnet this accessor represents within the serialized group.
+    series_model : MagnetModel, optional
+        Model of the series, used to convert the strength of the magnet to the shared hardware value. If omitted, the
+        submodel of the magnet is inverted.
 
     Methods
     -------
@@ -411,13 +417,15 @@ def __init__(
         elements_strength: list[RWStrengthScalar],
         elements_hardware: list[RWHardwareScalar],
         element_index: int,
+        series_model: MagnetModel | None = None,
     ):
         """
         Initialize the RWSerializedStrength.
         """
         self.__element = elements_strength[element_index]
         self.__elements_hardware = elements_hardware
         self.__element_index = element_index
+        self.__series_model = series_model
 
     def index(self) -> int:
         """Return the position of the magnet in the serialized group."""
@@ -438,8 +446,14 @@ def set(self, value: float):
         value : float
             Strength to apply to the magnet, converted to the shared hardware setpoint.
         """
-        model = self.__elements_hardware[self.__element_index].get_model()
-        hardware_value = model.compute_hardware_values([value])[0]
+        hardware = self.__elements_hardware[self.__element_index]
+        if self.__series_model is None:
+            hardware_value = hardware.get_model().compute_hardware_values([value])[0]
+        else:
+            current_hardware = np.array([hardware.get()])
+            hardware_value = self.__series_model.compute_hardware_values_partial(
+                current_hardware, {self.__element_index: value}
+            )[0]
         [element.set(hardware_value) for element in self.__elements_hardware]
 
     # Sets the value and wait that the read value reach the setpoint

pyaml/lattice/simulator.py

@@ -271,12 +271,12 @@ def _fill_serialized_magnets(self, magnets: SerializedMagnets) -> None:
         strengths = []
         for index, magnet in enumerate(magnets.get_magnets()):
             current = (
-                RWHardwareScalar(self.get_at_elems(magnet), magnets.polynom, magnets.model.get_sub_model(index))
+                RWHardwareScalar(self.get_at_elems(magnet), magnets.polynoms[index], magnets.model.get_sub_model(index))
                 if magnets.model.has_hardware()
                 else None
             )
             strength = (
-                RWStrengthScalar(self.get_at_elems(magnet), magnets.polynom, magnets.model.get_sub_model(index))
+                RWStrengthScalar(self.get_at_elems(magnet), magnets.polynoms[index], magnets.model.get_sub_model(index))
                 if magnets.model.has_physics()
                 else None
             )
@@ -286,7 +286,9 @@ def _fill_serialized_magnets(self, magnets: SerializedMagnets) -> None:
         linked_strengths = []
         for index in range(magnets.get_nb_magnets()):
             linked_currents.append(RWSerializedHardware(currents, index) if magnets.model.has_hardware() else None)
-            linked_strengths.append(RWSerializedStrength(strengths, currents, index) if magnets.model.has_physics() else None)
+            linked_strengths.append(
+                RWSerializedStrength(strengths, currents, index, magnets.model) if magnets.model.has_physics() else None
+            )
         attached_magnets = magnets.attach(self, linked_strengths, linked_currents)
         self.serialized_magnet.add(attached_magnets[0])
         for magnet in attached_magnets[1:]:

pyaml/magnet/linear_model.py

@@ -96,7 +96,7 @@ def __init__(
         Initialize the LinearMagnetModel.
         """
         if curve:
-            self.__curve = curve.get_curve()
+            self.__curve = curve.get_curve().copy()
             self.__curve[:, 1] = self.__curve[:, 1] * calibration_factor * crosstalk + calibration_offset
             self.__rcurve = Curve.inverse(self.__curve)
         else: