Move magnet model device to name (to be solved by catalog)

pyaml/control/abstract_impl.py

@@ -238,7 +238,7 @@ def set(self, value: NDArray[np.float64]):
         newHardwareValues = np.zeros(self.nb_device())
         hardwareIndex = 0
         for modelIndex, model in enumerate(self.__models):
-            nbDev = len(model.get_devices())
+            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]
@@ -257,7 +257,7 @@ def get(self) -> NDArray[np.float64]:
         allStrength = np.zeros(self.__nbMagnet)
         hardwareIndex = 0
         for modelIndex, model in enumerate(self.__models):
-            nbDev = len(model.get_devices())
+            nbDev = len(model.get_device_names())
             mStrengths = model.compute_strengths(allHardwareValues[hardwareIndex : hardwareIndex + nbDev])
             for valueIdx, strengthIdx in self.__modelToMagnet[modelIndex]:
                 allStrength[valueIdx] = mStrengths[strengthIdx]
@@ -269,7 +269,7 @@ def readback(self) -> np.array:
         allStrength = np.zeros(self.__nbMagnet)
         hardwareIndex = 0
         for modelIndex, model in enumerate(self.__models):
-            nbDev = len(model.get_devices())
+            nbDev = len(model.get_device_names())
             mStrengths = model.compute_strengths(allHardwareValues[hardwareIndex : hardwareIndex + nbDev])
             for valueIdx, strengthIdx in self.__modelToMagnet[modelIndex]:
                 allStrength[valueIdx] = mStrengths[strengthIdx]

pyaml/control/controlsystem.py

@@ -101,7 +101,7 @@ def create_magnet_strength_aggregator(self, magnets: list[Magnet]) -> ScalarAggr
             return None
         magg = CSStrengthScalarAggregator(agg)
         for m in magnets:
-            devs = self.attach(m.model.get_devices())
+            devs = self.get_devices_access(m.model.get_device_names())
             magg.add_magnet(m, devs)
         return magg
 
@@ -116,7 +116,7 @@ def create_magnet_hardware_aggregator(self, magnets: list[Magnet]) -> ScalarAggr
             if not m.model.has_hardware():
                 return None
             psIndex = m.hardware.index() if isinstance(m.hardware, RWMapper) else 0
-            agg.add_devices(self.attach([m.model.get_devices()[psIndex]])[0])
+            agg.add_devices(self.get_devices_access([m.model.get_device_names()[psIndex]])[0])
         return agg
 
     def create_bpm_aggregators(self, bpms: list[BPM]) -> list[ScalarAggregator | None]:
@@ -145,15 +145,15 @@ def fill_device(self, elements: list[Element]):
         """
         for e in elements:
             if isinstance(e, Magnet):
-                dev = self.attach(e.model.get_devices())[0]
+                dev = self.get_device_access(e.model.get_device_names()[0])
                 current = RWHardwareScalar(e.model, dev) if e.model.has_hardware() else None
                 strength = RWStrengthScalar(e.model, dev) if e.model.has_physics() else None
                 # Create a unique ref for this control system
                 m = e.attach(self, strength, current)
                 self.add_magnet(m)
 
             elif isinstance(e, CombinedFunctionMagnet):
-                devs = self.attach(e.model.get_devices())
+                devs = self.get_devices_access(e.model.get_device_names())
                 currents = RWHardwareArray(e.model, devs)
                 strengths = RWStrengthArray(e.model, devs)
                 # Create unique refs the cfm and
@@ -164,7 +164,7 @@ def fill_device(self, elements: list[Element]):
                     self.add_magnet(m)
 
             elif isinstance(e, SerializedMagnets):
-                devs = self.attach(e.model.get_devices())
+                devs = self.get_devices_access(e.model.get_device_names())
                 currents = []
                 strengths = []
                 # Create unique refs the series and each of its function for this

pyaml/magnet/identity_cfm_model.py

@@ -29,8 +29,8 @@ class ConfigModel(BaseModel):
     model_config = ConfigDict(arbitrary_types_allowed=True, extra="forbid")
 
     multipoles: list[str]
-    powerconverters: list[DeviceAccess | None] | None = None
-    physics: list[DeviceAccess | None] | None = None
+    powerconverters: list[str | None] | None = None
+    physics: list[str | None] | None = None
     units: list[str]
 
 
@@ -79,7 +79,7 @@ def get_strength_units(self) -> list[str]:
     def get_hardware_units(self) -> list[str]:
         return self._cfg.units
 
-    def get_devices(self) -> list[DeviceAccess | None]:
+    def get_device_names(self) -> list[str | None]:
         return self.__devices
 
     def set_magnet_rigidity(self, brho: np.double):

pyaml/magnet/identity_model.py

@@ -26,8 +26,8 @@ class ConfigModel(BaseModel):
 
     model_config = ConfigDict(arbitrary_types_allowed=True, extra="forbid")
 
-    powerconverter: DeviceAccess | None = None
-    physics: DeviceAccess | None = None
+    powerconverter: str | None = None
+    physics: str | None = None
     unit: str
 
 
@@ -62,7 +62,7 @@ def get_strength_units(self) -> list[str]:
     def get_hardware_units(self) -> list[str]:
         return [self.__unit]
 
-    def get_devices(self) -> list[DeviceAccess]:
+    def get_device_names(self) -> list[str | None]:
         return [self.__device]
 
     def set_magnet_rigidity(self, brho: np.double):

pyaml/magnet/linear_cfm_model.py

@@ -34,14 +34,16 @@ class ConfigModel(BaseModel):
     pseudo_offsets : list[float], optional
         Offsets applied to 'pseudo currents', 1 offset per function.
         Default: zeros
-    powerconverters : list[DeviceAccess]
+    powerconverters : list[str]
         List of power converter devices to apply currents (can be different
         from number of functions)
     matrix : Matrix, optional
         n x m matrix (n rows for n functions, m columns for m currents)
         to handle multipoles separation. Default: Identity
     units : list[str]
         List of strength units (i.e. ['rad', 'm-1', 'm-2'])
+    hardware_units : list[str]
+        List of hardware units (i.e. 'A' , 'V')
     """
 
     model_config = ConfigDict(arbitrary_types_allowed=True, extra="forbid")
@@ -52,9 +54,10 @@ class ConfigModel(BaseModel):
     calibration_offsets: list[float] = None
     pseudo_factors: list[float] = None
     pseudo_offsets: list[float] = None
-    powerconverters: list[DeviceAccess | None]
+    powerconverters: list[str | None]
     matrix: Matrix = None
     units: list[str]
+    hardware_units: list[str]
 
 
 class LinearCFMagnetModel(MagnetModel):
@@ -97,6 +100,7 @@ def __init__(self, cfg: ConfigModel):
         self.__check_len(self.__pf, "pseudo_factors", self.__nbFunction)
         self.__check_len(self.__po, "pseudo_offsets", self.__nbFunction)
         self.__check_len(cfg.units, "units", self.__nbFunction)
+        self.__check_len(cfg.hardware_units, "hardware_units", self.__nbPS)
         self.__check_len(cfg.curves, "curves", self.__nbFunction)
 
         if cfg.matrix is None:
@@ -149,9 +153,9 @@ def get_strength_units(self) -> list[str]:
         return self._cfg.units
 
     def get_hardware_units(self) -> list[str]:
-        return np.array([p.unit() for p in self._cfg.powerconverters])
+        return self._cfg.hardware_units
 
-    def get_devices(self) -> list[DeviceAccess]:
+    def get_device_names(self) -> list[str | None]:
         return self._cfg.powerconverters
 
     def set_magnet_rigidity(self, brho: np.double):

pyaml/magnet/linear_model.py

@@ -2,7 +2,6 @@
 from pydantic import BaseModel, ConfigDict
 
 from ..common.element import __pyaml_repr__
-from ..control.deviceaccess import DeviceAccess
 from .curve import Curve
 from .model import MagnetModel
 
@@ -19,7 +18,7 @@ class ConfigModel(BaseModel):
     curve : Curve or None, optional
         Curve object used for interpolation. By default,
         identity curve is used.
-    powerconverter : DeviceAccess or None, optional
+    powerconverter : str or None, optional
         Power converter device to apply currrent
     calibration_factor : float, optional
         Correction factor applied to the curve. Default: 1.0
@@ -29,17 +28,19 @@ class ConfigModel(BaseModel):
         Crosstalk factor. Default: 1.0
     unit : str
         Unit of the strength (i.e. 1/m or m-1)
-
+    hardware_unit : str
+        Hardware units (i.e. 'A' , 'V')
     """
 
     model_config = ConfigDict(arbitrary_types_allowed=True, extra="forbid")
 
     curve: Curve | None = None
-    powerconverter: DeviceAccess | None
+    powerconverter: str | None
     calibration_factor: float = 1.0
     calibration_offset: float = 0.0
     crosstalk: float = 1.0
     unit: str
+    hardware_unit: str
 
 
 class LinearMagnetModel(MagnetModel):
@@ -60,14 +61,8 @@ def __init__(self, cfg: ConfigModel):
             self.__g = cfg.calibration_factor * cfg.crosstalk
             self.__o = cfg.calibration_offset
         self.__strength_unit = cfg.unit
-
-        if cfg.powerconverter is not None:
-            self.__hardware_unit = cfg.powerconverter.unit()
-            self.__ps = cfg.powerconverter
-        else:
-            self.__hardware_unit = None
-            self.__ps = None
-
+        self.__hardware_unit = cfg.hardware_unit
+        self.__ps = cfg.powerconverter
         self.__brho = np.nan
 
     def compute_hardware_values(self, strengths: np.array) -> np.array:
@@ -90,7 +85,7 @@ def get_strength_units(self) -> list[str]:
     def get_hardware_units(self) -> list[str]:
         return [self.__hardware_unit] if self.__hardware_unit is not None else [""]
 
-    def get_devices(self) -> list[DeviceAccess]:
+    def get_device_names(self) -> list[str | None]:
         return [self.__ps]
 
     def set_magnet_rigidity(self, brho: np.double):

pyaml/magnet/linear_serialized_model.py

@@ -32,11 +32,13 @@ class ConfigModel(BaseModel):
         Default: zeros
     crosstalk : float or list[float], optional
         Crosstalk factors. Default: 1.0
-    powerconverter : DeviceAccess
+    powerconverter : str
         The hardware can be a single power supply or a list of power supplies.
         If a list is provided, the same value will be affected to all of them
     unit : str
         Strength unit: rad, m-1, m-2
+    hardware_unit : str
+        Hardware unit (i.e. 'A' , 'V')
     """
 
     model_config = ConfigDict(arbitrary_types_allowed=True, extra="forbid")
@@ -45,8 +47,9 @@ class ConfigModel(BaseModel):
     calibration_factors: float | list[float] = None
     calibration_offsets: float | list[float] = None
     crosstalk: float | list[float] = 1.0
-    powerconverter: DeviceAccess
+    powerconverter: str | None
     unit: str
+    hardware_unit: str
 
 
 def _get_length(elem) -> int:
@@ -136,6 +139,7 @@ def __initialize(self):
                 crosstalk=self.__crosstalk[magnet_idx],
                 powerconverter=self._cfg.powerconverter,
                 unit=self._cfg.unit,
+                hardware_unit=self._cfg.hardware_unit,
             )
             self.__sub_models.append(LinearMagnetModel(sub_model))
 
@@ -158,9 +162,9 @@ def get_strength_units(self) -> list[str]:
         return [self._cfg.unit] * self.__nbMagnets
 
     def get_hardware_units(self) -> list[str]:
-        return [self.__sub_models[0].get_hardware_units()[0]]
+        return [self._cfg.hardware_unit]
 
-    def get_devices(self) -> list[DeviceAccess]:
+    def get_device_names(self) -> list[str | None]:
         return [self._cfg.powerconverter]
 
     def set_magnet_rigidity(self, brho: np.double):

pyaml/magnet/model.py

@@ -62,27 +62,14 @@ def get_strength_units(self) -> list[str]:
         pass
 
     @abstractmethod
-    def get_hardware_units(self) -> list[str]:
+    def get_device_names(self) -> list[str | None]:
         """
-        Get hardware units
+        Get device names
 
         Returns
         -------
-        list[str]
-            Array of hardware units. For a single multipole,
-            returns a list of 1 item
-        """
-        pass
-
-    @abstractmethod
-    def get_devices(self) -> list[DeviceAccess | None]:
-        """
-        Get device handles
-
-        Returns
-        -------
-        list[DeviceAccess]
-            Array of DeviceAcess
+        list[DevstriceAccess]
+            Array of device name
         """
         pass
 

pyaml/magnet/serialized_magnet.py

@@ -170,5 +170,5 @@ def set_energy(self, energy: float):
     def __repr__(self):
         return __pyaml_repr__(self)
 
-    def get_devices(self) -> list[DeviceAccess]:
-        return self.model.get_devices()
+    def get_device_names(self) -> list[str | None]:
+        return self.model.get_device_names()

pyaml/magnet/spline_model.py

@@ -29,6 +29,8 @@ class ConfigModel(BaseModel):
         Crosstalk factor. Default: 1.0
     unit : str
         Unit of the strength (i.e. 1/m or m-1)
+    hardware_unit : str
+        Hardware units (i.e. 'A' , 'V')
     alpha : float, optional
         Regularization parameter (alpha >= 0). alpha = 0 means the interpolation
         passes through all the points of the curve. Default: 0.0
@@ -37,11 +39,12 @@ class ConfigModel(BaseModel):
     model_config = ConfigDict(arbitrary_types_allowed=True, extra="forbid")
 
     curve: Curve
-    powerconverter: DeviceAccess | None
+    powerconverter: str | None
     calibration_factor: float = 1.0
     calibration_offset: float = 0.0
     crosstalk: float = 1.0
     unit: str
+    hardware_unit: str
     alpha: float = 0.0
 
 
@@ -57,7 +60,7 @@ def __init__(self, cfg: ConfigModel):
         self.__curve[:, 1] = self.__curve[:, 1] * cfg.calibration_factor * cfg.crosstalk + cfg.calibration_offset
         rcurve = Curve.inverse(self.__curve)
         self.__strength_unit = cfg.unit
-        self.__hardware_unit = cfg.powerconverter.unit()
+        self.__hardware_unit = cfg.hardware_unit
         self.__brho = np.nan
         self.__ps = cfg.powerconverter
         self.__spl = make_smoothing_spline(self.__curve[:, 0], self.__curve[:, 1], lam=cfg.alpha)
@@ -77,7 +80,7 @@ def get_strength_units(self) -> list[str]:
     def get_hardware_units(self) -> list[str]:
         return [self.__hardware_unit] if self.__hardware_unit is not None else [""]
 
-    def get_devices(self) -> list[DeviceAccess]:
+    def get_device_names(self) -> list[str | None]:
         return [self.__ps]
 
     def set_magnet_rigidity(self, brho: np.double):

tests/arrays/test_ranges_array.py

@@ -10,7 +10,7 @@
     indirect=True,
 )
 def test_ranges_array(install_test_package):
-    sr: Accelerator = Accelerator.load("tests/config/EBSTune-range.yaml")
+    sr: Accelerator = Accelerator.load("tests/config/EBSTune.yaml")
     mag_cur = sr.live.get_magnets("QForTune").hardwares
     mag_cur.set(mag_cur.get() + 50)