Changes for schema registry for magnets

pyaml/magnet/cfm_magnet.py

@@ -2,11 +2,12 @@
 
 from ..common import abstract
 from ..common.abstract import RWMapper
-from ..common.element import Element, ElementConfigModel, __pyaml_repr__
+from ..common.element import Element, __pyaml_repr__
 from ..common.exception import PyAMLException
 from ..configuration.factory import ELEMENT_REGISTRY
+from ..validation import DynamicValidation, register_schema
 from .hcorrector import HCorrector
-from .magnet import Magnet, MagnetConfigModel
+from .magnet import Magnet
 from .model import MagnetModel
 from .octupole import Octupole
 from .quadrupole import Quadrupole
@@ -31,39 +32,64 @@
 PYAMLCLASS = "CombinedFunctionMagnet"
 
 
-class ConfigModel(ElementConfigModel):
-    mapping: list[list[str]]
-    """Name mapping for multipoles
-    (i.e. [[B0,C01A-H],[A0,C01A-H],[B2,C01A-S]])"""
-    model: MagnetModel | None = None
-    """Object in charge of converting magnet strenghts to currents"""
-
-
-class CombinedFunctionMagnet(Element):
-    """CombinedFunctionMagnet class"""
-
-    def __init__(self, cfg: ConfigModel, peer=None):
-        super().__init__(cfg.name)
-        self._cfg = cfg
-        self.model = cfg.model
+@register_schema
+class CombinedFunctionMagnet(Element, DynamicValidation):
+    """
+    Combined function magnet made up of several virtual single-function magnets.
+
+    This class represents a magnet whose effect is described by multiple multipole
+    components, such as a corrector, quadrupole, sextupole, or octupole family.
+    Each entry in ``mapping`` creates a virtual magnet backed by the same
+    underlying magnet model, and the virtual magnets are exposed as individual
+    elements while still belonging to the same combined-function object.
+
+    Parameters
+    ----------
+    name : str
+        Name of the combined-function magnet.
+    mapping : list[list[str]]
+        List of ``[multipole, magnet_name]`` pairs. The first entry selects the
+        virtual magnet type, and the second entry gives the name of the virtual
+        magnet.
+    model : MagnetModel | None, optional
+        Magnet model used to convert strengths to hardware values and vice versa.
+    description : str | None, optional
+        Human-readable description of the magnet.
+    peer : object, optional
+        Control-system or simulator peer used when attaching the magnet.
+
+    Raises
+    ------
+    PyAMLException
+        If the mapping is invalid, if an unsupported multipole is requested, or
+        if the model does not provide the required multipole information.
+    """
+
+    def __init__(
+        self, name: str, mapping: list[list[str]], model: MagnetModel | None = None, description: str | None = None, peer=None
+    ):
+        super().__init__(name, None, description)
+
+        self._mapping = mapping
+        self.model = model
         self.__virtuals: list[Magnet] = []
-        self.__strengths: abstract.ReadWriteFloatArray = None
-        self.__hardwares: abstract.ReadWriteFloatArray = None
+        self.__strengths: abstract.ReadWriteFloatArray | None = None
+        self.__hardwares: abstract.ReadWriteFloatArray | None = None
 
         if peer is None:
             # Configuration part
-            if self.model is not None and not hasattr(self.model._cfg, "multipoles"):
-                raise PyAMLException(f"{cfg.name} model: mutipolesfield required for combined function magnet")
+            if self.model is not None and not hasattr(self.model, "multipoles"):
+                raise PyAMLException(f"{name} model: mutipoles field required for combined function magnet")
 
             idx = 0
             self.polynoms = []
-            for _idx, m in enumerate(cfg.mapping):
+            for _idx, m in enumerate(self._mapping):
                 # Check mapping validity
                 if len(m) != 2:
                     raise PyAMLException("Invalid CombinedFunctionMagnet mapping for {m}")
                 if m[0] not in _fmap:
                     raise PyAMLException(m[0] + " not implemented for combined function magnet")
-                if m[0] not in self.model._cfg.multipoles:
+                if m[0] not in self.model.multipoles:
                     raise PyAMLException(m[0] + " not found in underlying magnet model")
                 self.polynoms.append(_fmap[m[0]].polynom)
                 # Create the virtual magnet for the correspoding multipole
@@ -81,16 +107,16 @@ def get_model_name(self) -> str:
         """
         Returns the model name of this magnet
         """
-        return self._cfg.name
+        return self._name
 
     def __create_virutal_manget(self, name: str, idx: int) -> Magnet:
         args = {"name": name, "model": self.model}
-        mVirtual: Magnet = _fmap[idx](MagnetConfigModel(**args))
+        mVirtual: Magnet = _fmap[idx](**args)
         mVirtual.set_model_name(self.get_name())
         return mVirtual
 
     def nb_multipole(self) -> int:
-        return len(self._cfg.mapping)
+        return len(self._mapping)
 
     def attach(
         self,
@@ -100,13 +126,13 @@ def attach(
     ) -> list[Magnet]:
         l = []
         # Attached the CombinedFunctionMagnet itself
-        nCFM = CombinedFunctionMagnet(self._cfg, peer)
+        nCFM = CombinedFunctionMagnet(self._name, self._mapping, self.model, self._description, peer)
         nCFM.__strengths = strengths
         nCFM.__hardwares = hardwares
         l.append(nCFM)
         # Construct a single function magnet for each multipole
         # of this combined function magnet
-        for idx, _m in enumerate(self._cfg.mapping):
+        for idx, _m in enumerate(self._mapping):
             strength = RWMapper(strengths, idx)
             hardware = RWMapper(hardwares, idx) if self.model.has_hardware() else None
             l.append(self.__virtuals[idx].attach(peer, strength, hardware))

pyaml/magnet/corrector.py

@@ -8,7 +8,7 @@ class RWCorrectorAngle(abstract.ReadWriteFloatScalar):
     """
     Set the angle of a horizontal or vertical corrector.
     KickAngle sign convention is defined the a global PyAML constant
-    (see pyaml.common.constant.HORIZONATL_KICK_SIGN).
+    (see pyaml.common.constant.HORIZONTAL_KICK_SIGN).
     To change the convention, you have execute the code below prior to everything:
     import pyaml.common.constants
     pyaml.common.constants.HORIZONTAL_KICK_SIGN = -1.0

pyaml/magnet/csvcurve.py

@@ -1,51 +1,75 @@
-from pathlib import Path
-
 import numpy as np
-from pydantic import BaseModel, ConfigDict
+from numpy.typing import NDArray
 
+from ..common.element import __pyaml_repr__
 from ..common.exception import PyAMLException
 from ..configuration.fileloader import ROOT
+from ..validation import DynamicValidation, register_schema
 from .curve import Curve
 
 # Define the main class name for this module
 PYAMLCLASS = "CSVCurve"
 
 
-class ConfigModel(BaseModel):
+@register_schema
+class CSVCurve(Curve, DynamicValidation):
     """
-    Configuration model for CSV curve
+    Curve loaded from a CSV file.
 
-    Parameters
-    ----------
-    file : str
-        CSV file that contains the curve (n rows, 2 columns)
-    """
+    This class reads a CSV file containing a two-column numeric dataset
+    representing ``(x, y)`` coordinate pairs. The file is loaded during
+    initialization and stored internally as a NumPy array.
 
-    model_config = ConfigDict(arbitrary_types_allowed=True, extra="forbid")
+    The CSV file must:
 
-    file: str
+    - contain exactly two columns,
+    - use commas as the delimiter,
+    - contain values that can be converted to ``float``.
 
+    Parameters
+    ----------
+    file : str
+        Path to the CSV file. Relative paths are resolved using the
+        project's configured root directory.
 
-class CSVCurve(Curve):
-    """
-    Class for load CSV (x,y) curve
+    Attributes
+    ----------
+    file : str
+        Path to the CSV file provided during initialization.
+
+    Raises
+    ------
+    PyAMLException
+        If the file cannot be parsed as a numeric CSV or if the loaded
+        data does not have shape ``(n, 2)``.
+
+    Notes
+    -----
+    The loaded curve is stored internally as a NumPy array with shape
+    ``(n, 2)``, where the first column contains x-values and the second
+    column contains y-values. The data can be accessed using
+    :meth:`get_curve`.
     """
 
-    def __init__(self, cfg: ConfigModel):
-        self._cfg = cfg
+    def __init__(self, file: str):
+        self._file = file
 
         # Load CSV curve
-        path = ROOT.expand_path(cfg.file)
+        path = ROOT.expand_path(self._file)
         try:
             self._curve = np.genfromtxt(path, delimiter=",", dtype=float, loose=False)
         except ValueError as e:
-            raise PyAMLException(f"CSVCurve(file='{cfg.file}',dtype=float): {str(e)}") from None
+            raise PyAMLException(f"CSVCurve(file='{self._file}',dtype=float): {str(e)}") from None
 
         _s = np.shape(self._curve)
         if len(_s) != 2 or _s[1] != 2:
-            raise PyAMLException(f"CSVCurve(file='{cfg.file}',dtype=float):wrong shape (2,2) expected but got {str(_s)}")
+            raise PyAMLException(f"CSVCurve(file='{self._file}',dtype=float):wrong shape (2,2) expected but got {str(_s)}")
+
+    @property
+    def file(self):
+        return self._file
 
-    def get_curve(self) -> np.array:
+    def get_curve(self) -> NDArray[np.float64]:
         """
         Get the curve data.
 
@@ -57,4 +81,4 @@ def get_curve(self) -> np.array:
         return self._curve
 
     def __repr__(self):
-        return repr(self._cfg).replace("ConfigModel", self.__class__.__name__)
+        return __pyaml_repr__(self)

pyaml/magnet/csvmatrix.py

@@ -1,46 +1,51 @@
-from pathlib import Path
-
 import numpy as np
-from pydantic import BaseModel, ConfigDict
+from numpy.typing import NDArray
 
+from ..common.element import __pyaml_repr__
 from ..common.exception import PyAMLException
 from ..configuration.fileloader import ROOT
+from ..validation import DynamicValidation, register_schema
 from .matrix import Matrix
 
 # Define the main class name for this module
 PYAMLCLASS = "CSVMatrix"
 
 
-class ConfigModel(BaseModel):
+@register_schema
+class CSVMatrix(Matrix, DynamicValidation):
     """
-    Configuration model for CSV matrix
+    Matrix loaded from a CSV file.
+
+    This class reads a CSV file containing numeric values and stores the
+    resulting matrix as a NumPy array.
 
     Parameters
     ----------
     file : str
-        CSV file that contains the matrix
-    """
-
-    model_config = ConfigDict(arbitrary_types_allowed=True, extra="forbid")
-
-    file: str
+        Path to the CSV file. Relative paths are resolved using the
+        project's configured root directory.
 
-
-class CSVMatrix(Matrix):
-    """
-    Class for loading CSV matrix
+    Raises
+    ------
+    PyAMLException
+        If the CSV file cannot be parsed as a numeric array.
     """
 
-    def __init__(self, cfg: ConfigModel):
-        self._cfg = cfg
+    def __init__(self, file: str):
+        self._file = file
+
         # Load CSV matrix
-        path = ROOT.expand_path(cfg.file)
+        path = ROOT.expand_path(self._file)
         try:
             self._mat = np.genfromtxt(path, delimiter=",", dtype=float, loose=False)
         except ValueError as e:
-            raise PyAMLException(f"CSVMatrix(file='{cfg.file}',dtype=float): {str(e)}") from None
+            raise PyAMLException(f"CSVMatrix(file='{self._file}',dtype=float): {str(e)}") from None
+
+    @property
+    def file(self):
+        return self._file
 
-    def get_matrix(self) -> np.array:
+    def get_matrix(self) -> NDArray[np.float64]:
         """
         Get the matrix data.
 
@@ -52,4 +57,4 @@ def get_matrix(self) -> np.array:
         return self._mat
 
     def __repr__(self):
-        return repr(self._cfg).replace("ConfigModel", self.__class__.__name__)
+        return __pyaml_repr__(self)

pyaml/magnet/hcorrector.py

@@ -1,30 +1,28 @@
+import copy
+from typing import Self
+
 from ..common import abstract
 from ..common.constants import HORIZONTAL_KICK_SIGN
 from ..lattice.polynom_info import PolynomInfo
+from ..validation import DynamicValidation, register_schema
 from .corrector import RWCorrectorAngle
-from .magnet import Magnet, MagnetConfigModel
+from .magnet import Magnet
+from .model import MagnetModel
 
 # Define the main class name for this module
 PYAMLCLASS = "HCorrector"
 
 
-class ConfigModel(MagnetConfigModel):
-    """Configuration model for Horizontal Corrector magnet."""
-
-    ...
-
-
-class HCorrector(Magnet):
+@register_schema
+class HCorrector(Magnet, DynamicValidation):
     """Horizontal Corrector class"""
 
     polynom = PolynomInfo("PolynomB", 0, HORIZONTAL_KICK_SIGN)
 
-    def __init__(self, cfg: ConfigModel):
-        super().__init__(
-            cfg.name,
-            cfg.model if hasattr(cfg, "model") else None,
-        )
-        self._cfg = cfg
+    def __init__(
+        self, name: str, model: MagnetModel | None = None, lattice_names: str | None = None, description: str | None = None
+    ):
+        super().__init__(name, model, lattice_names, description)
         self.__angle = RWCorrectorAngle(self)
 
     @property
@@ -33,3 +31,17 @@ def angle(self) -> abstract.ReadWriteFloatScalar:
         Set the kick angle.
         """
         return self.__angle
+
+    def attach(
+        self,
+        peer,
+        strength: abstract.ReadWriteFloatScalar,
+        hardware: abstract.ReadWriteFloatScalar,
+    ) -> Self:
+        """
+        Create a new reference to attach this magnet to a simulator
+        or a control systemand.
+        """
+        obj = super().attach(peer, strength, hardware)
+        obj.__angle = RWCorrectorAngle(obj)
+        return obj