2 handle serialized magnets

pyaml/arrays/magnet_array.py

@@ -111,7 +111,7 @@ def __init__(self, arrayName: str, magnets: list[Magnet], use_aggregator=True):
 
         if use_aggregator and len(magnets) > 0:
             aggs = self.get_peer().create_magnet_strength_aggregator(magnets)
-            aggh = self.get_peer().create_magnet_harddware_aggregator(magnets)
+            aggh = self.get_peer().create_magnet_hardware_aggregator(magnets)
             self.__rwstrengths.set_aggregator(aggs)
             self.__rwhardwares.set_aggregator(aggh)
 

pyaml/arrays/serialized_magnet.py

@@ -0,0 +1,32 @@
+from ..common.element_holder import ElementHolder
+from .array import ArrayConfig, ArrayConfigModel
+
+# Define the main class name for this module
+PYAMLCLASS = "SerializedMagnets"
+
+
+class ConfigModel(ArrayConfigModel): ...
+
+
+class SerializedMagnets(ArrayConfig):
+    """
+    Serialized magnets array configuration
+
+    Example
+    -------
+
+    A magnet array configuration can also be created by code using
+    the following example::
+
+        from pyaml.arrays.serialized_magnet import SerializedMagnets
+        from pyaml.arrays.serialized_magnet import ConfigModel as SerializedMagnetConfigModel
+        magArray = SerializedMagnets(
+                     SerializedMagnetConfigModel(name="mySerializedMagnets", elements=["mag1","mag2"])
+                   )
+    """
+
+    def __init__(self, cfg: ArrayConfigModel):
+        super().__init__(cfg)
+
+    def fill_array(self, holder: ElementHolder):
+        holder.fill_serialized_magnet_array(self._cfg.name, self._cfg.elements)

pyaml/arrays/serialized_magnet_array.py

@@ -0,0 +1,124 @@
+import numpy as np
+
+from ..common.abstract import ReadWriteFloatArray
+from ..common.exception import PyAMLException
+from ..magnet.serialized_magnet import SerializedMagnets
+from .element_array import ElementArray
+
+# TODO handle aggregator for CFM
+
+
+class RWMagnetStrengths(ReadWriteFloatArray):
+    def __init__(self, name: str, magnets: list[SerializedMagnets]):
+        self.__name = name
+        self.__magnets = magnets
+        self.__nb = sum(m.get_nb_magnets() for m in magnets)
+
+    # Gets the values
+    def get(self) -> np.array:
+        r = np.zeros(self.__nb)
+        idx = 0
+        for m in self.__magnets:
+            r[idx : idx + m.get_nb_magnets()] = m.strengths.get()
+            idx += m.get_nb_magnets()
+        return r
+
+    # Sets the values
+    def set(self, value: np.array):
+        nvalue = np.ones(self.__nb) * value if isinstance(value, float) else value
+        for idx, m in enumerate(self.__magnets):
+            m.strengths.set(nvalue[idx])
+            idx += m.get_nb_magnets()
+
+    # Sets the values and waits that the read values reach their setpoint
+    def set_and_wait(self, value: np.array):
+        raise NotImplementedError("Not implemented yet.")
+
+    # Gets the unit of the values
+    def unit(self) -> list[str]:
+        r = []
+        for m in self.__magnets:
+            r.extend(m.strengths.unit())
+        return r
+
+
+class RWMagnetHardwares(ReadWriteFloatArray):
+    def __init__(self, name: str, magnets: list[SerializedMagnets]):
+        self.__name = name
+        self.__magnets = magnets
+        self.__nb = sum(m.get_nb_magnets() for m in magnets)
+
+    # Gets the values
+    def get(self) -> np.array:
+        r = np.zeros(self.__nb)
+        idx = 0
+        for m in self.__magnets:
+            r[idx : idx + m.get_nb_magnets()] = m.hardwares.get()
+            idx += m.get_nb_magnets()
+        return r
+
+    # Sets the values
+    def set(self, value: np.array):
+        nvalue = np.ones(self.__nb) * value if isinstance(value, float) else value
+        for idx, m in enumerate(self.__magnets):
+            m.hardwares.set(nvalue[idx])
+
+    # Sets the values and waits that the read values reach their setpoint
+    def set_and_wait(self, value: np.array):
+        raise NotImplementedError("Not implemented yet.")
+
+    # Gets the unit of the values
+    def unit(self) -> list[str]:
+        r = []
+        for m in self.__magnets:
+            r.extend(m.hardwares.unit())
+        return r
+
+
+class SerializedMagnetsArray(ElementArray):
+    """
+    Class that implements access to a combined function magnet array
+
+    Parameters
+    ----------
+    arrayName : str
+        Array name
+    magnets: list[Magnet]
+        Magnet list, all elements must be attached to the same instance of
+        either a Simulator or a ControlSystem.
+    use_aggregator : bool
+        Use aggregator to increase performance by using paralell
+        access to underlying devices.
+    """
+
+    def __init__(
+        self,
+        arrayName: str,
+        magnets: list[SerializedMagnets],
+        use_aggregator=False,
+    ):
+        super().__init__(arrayName, magnets, use_aggregator)
+
+        self.__rwstrengths = RWMagnetStrengths(arrayName, magnets)
+        self.__rwhardwares = RWMagnetHardwares(arrayName, magnets)
+
+        if use_aggregator:
+            raise (
+                PyAMLException(
+                    "Aggregator not implemented for CombinedFunctionMagnetArray"
+                )
+            )
+
+    @property
+    def strengths(self) -> RWMagnetStrengths:
+        """
+        Give access to strength of each magnet of this array
+        """
+        return self.__rwstrengths
+
+    @property
+    def hardwares(self) -> RWMagnetHardwares:
+        """
+        Give access to hardware value of each magnet of this array
+        """
+        return self.__rwhardwares

pyaml/common/element_holder.py

@@ -8,11 +8,13 @@
 from ..arrays.cfm_magnet_array import CombinedFunctionMagnetArray
 from ..arrays.element_array import ElementArray
 from ..arrays.magnet_array import MagnetArray
+from ..arrays.serialized_magnet_array import SerializedMagnetsArray
 from ..bpm.bpm import BPM
 from ..common.exception import PyAMLException
 from ..diagnostics.tune_monitor import BetatronTuneMonitor
 from ..magnet.cfm_magnet import CombinedFunctionMagnet
 from ..magnet.magnet import Magnet
+from ..magnet.serialized_magnet import SerializedMagnets
 from ..rf.rf_plant import RFPlant
 from ..rf.rf_transmitter import RFTransmitter
 from .element import Element
@@ -32,6 +34,7 @@ def __init__(self):
         # Device handle
         self.__MAGNETS: dict = {}
         self.__CFM_MAGNETS: dict = {}
+        self.__SERIALIZED_MAGNETS: dict = {}
         self.__BPMS: dict = {}
         self.__RFPLANT: dict = {}
         self.__RFTRANSMITTER: dict = {}
@@ -42,6 +45,7 @@ def __init__(self):
         # Array handle
         self.__MAGNET_ARRAYS: dict = {}
         self.__CFM_MAGNET_ARRAYS: dict = {}
+        self.__SERIALIZED_MAGNETS_ARRAYS: dict = {}
         self.__BPM_ARRAYS: dict = {}
         self.__ELEMENT_ARRAYS: dict = {}
 
@@ -101,7 +105,7 @@ def fill_element_array(self, arrayName: str, elementNames: list[str]):
     def get_element(self, name: str) -> Element:
         return self.__get("Element", name, self.__ALL)
 
-    def get_elemens(self, name: str) -> ElementArray:
+    def get_elements(self, name: str) -> ElementArray:
         return self.__get("Element array", name, self.__ELEMENT_ARRAYS)
 
     def get_all_elements(self) -> list[Element]:
@@ -151,6 +155,31 @@ def get_cfm_magnets(self, name: str) -> CombinedFunctionMagnetArray:
     def get_all_cfm_magnets(self) -> list[CombinedFunctionMagnet]:
         return [value for key, value in self.__CFM_MAGNETS.items()]
 
+    # Serialized magnets
+
+    def fill_serialized_magnet_array(self, arrayName: str, elementNames: list[str]):
+        self.fill_array(
+            arrayName,
+            elementNames,
+            self.get_serialized_magnet,
+            SerializedMagnetsArray,
+            self.__SERIALIZED_MAGNETS_ARRAYS,
+        )
+
+    def get_serialized_magnet(self, name: str) -> Magnet:
+        return self.__get("SerializedMagnets", name, self.__SERIALIZED_MAGNETS)
+
+    def add_serialized_magnet(self, m: Magnet):
+        self.__add(self.__SERIALIZED_MAGNETS, m)
+
+    def get_serialized_magnets(self, name: str) -> SerializedMagnetsArray:
+        return self.__get(
+            "SerializedMagnets array", name, self.__SERIALIZED_MAGNETS_ARRAYS
+        )
+
+    def get_all_serialized_magnets(self) -> list[SerializedMagnets]:
+        return [value for key, value in self.__SERIALIZED_MAGNETS.items()]
+
     # BPMs
 
     def fill_bpm_array(self, arrayName: str, elementNames: list[str]):

pyaml/configuration/inline_curve.py

@@ -0,0 +1,41 @@
+from pathlib import Path
+
+import numpy as np
+from pydantic import BaseModel, ConfigDict
+
+from ..common.exception import PyAMLException
+from ..configuration import get_root_folder
+from .curve import Curve
+
+# Define the main class name for this module
+PYAMLCLASS = "InlineCurve"
+
+
+class ConfigModel(BaseModel):
+    model_config = ConfigDict(arbitrary_types_allowed=True, extra="forbid")
+
+    mat: list[list[float]]
+    """CSV file that contains the curve (n rows,2 columns)"""
+
+
+class InlineCurve(Curve):
+    """
+    Class for load CSV (x,y) curve
+    """
+
+    def __init__(self, cfg: ConfigModel):
+        self._cfg = cfg
+        # Load the curve
+        self._curve = np.array(self._cfg.mat)
+
+        _s = np.shape(self._curve)
+        if len(_s) != 2 or _s[1] != 2:
+            raise PyAMLException(
+                f"InlineCurve(mat='{cfg.mat}',dtype=float): wrong shape (2,2) expected but got {str(_s)}"
+            )
+
+    def get_curve(self) -> np.array:
+        return self._curve
+
+    def __repr__(self):
+        return repr(self._cfg).replace("ConfigModel", self.__class__.__name__)

pyaml/configuration/inline_matrix.py

@@ -0,0 +1,31 @@
+import numpy as np
+from pydantic import BaseModel, ConfigDict
+
+from .matrix import Matrix
+
+# Define the main class name for this module
+PYAMLCLASS = "InlineMatrix"
+
+
+class ConfigModel(BaseModel):
+    model_config = ConfigDict(arbitrary_types_allowed=True, extra="forbid")
+
+    mat: list[list[float]]
+    """The matrix"""
+
+
+class InlineMatrix(Matrix):
+    """
+    Class for loading CSV matrix
+    """
+
+    def __init__(self, cfg: ConfigModel):
+        self._cfg = cfg
+        # Load the matrix
+        self._mat = np.array(self._cfg.mat)
+
+    def get_matrix(self) -> np.array:
+        return self._mat
+
+    def __repr__(self):
+        return repr(self._cfg).replace("ConfigModel", self.__class__.__name__)

pyaml/control/abstract_impl.py

@@ -160,6 +160,9 @@ def set_and_wait(self, value: double):
     def unit(self) -> str:
         return self.__model.get_hardware_units()[0]
 
+    def set_magnet_rigidity(self, brho: np.double):
+        self.__model.set_magnet_rigidity(brho)
+
 
 # ------------------------------------------------------------------------------
 
@@ -191,6 +194,9 @@ def set_and_wait(self, value: float):
     def unit(self) -> str:
         return self.__model.get_strength_units()[0]
 
+    def set_magnet_rigidity(self, brho: np.double):
+        self.__model.set_magnet_rigidity(brho)
+
 
 # ------------------------------------------------------------------------------
 

pyaml/control/controlsystem.py

@@ -24,6 +24,7 @@
 from ..diagnostics.tune_monitor import BetatronTuneMonitor
 from ..magnet.cfm_magnet import CombinedFunctionMagnet
 from ..magnet.magnet import Magnet
+from ..magnet.serialized_magnet import SerializedMagnets
 from ..rf.rf_plant import RFPlant, RWTotalVoltage
 from ..rf.rf_transmitter import RFTransmitter
 from ..tuning_tools.orbit import Orbit
@@ -74,7 +75,7 @@ def create_magnet_strength_aggregator(
             agg.add_magnet(m, devs)
         return agg
 
-    def create_magnet_harddware_aggregator(
+    def create_magnet_hardware_aggregator(
         self, magnets: list[Magnet]
     ) -> ScalarAggregator:
         """When working in hardware space, 1 single power
@@ -147,6 +148,29 @@ def fill_device(self, elements: list[Element]):
                 for m in ms[1:]:
                     self.add_magnet(m)
 
+            elif isinstance(e, SerializedMagnets):
+                devs = self.attach(e.model.get_devices())
+                currents = []
+                strengths = []
+                # Create unique refs the series and each of its function for this control system
+                for i in range(e.get_nb_magnets()):
+                    current = (
+                        RWHardwareScalar(e.model.get_sub_model(i), devs[i])
+                        if e.model.has_hardware()
+                        else None
+                    )
+                    strength = (
+                        RWStrengthScalar(e.model.get_sub_model(i), devs[i])
+                        if e.model.has_physics()
+                        else None
+                    )
+                    currents.append(current)
+                    strengths.append(strength)
+                ms = e.attach(self, strengths, currents)
+                self.add_serialized_magnet(ms[0])
+                for m in ms[1:]:
+                    self.add_magnet(m)
+
             elif isinstance(e, BPM):
                 tiltDev = self.attach([e.model.get_tilt_device()])[0]
                 offsetsDevs = self.attach(e.model.get_offset_devices())