Feature implement bpm device

pyaml/bpm/bpm.py

@@ -0,0 +1,76 @@
+from pyaml.lattice.element import Element, ElementConfigModel
+from pyaml.lattice.abstract_impl import RBpmArray, RWBpmOffsetArray, RWBpmTiltScalar
+from ..control.deviceaccess import DeviceAccess
+from ..control import abstract
+from typing import Self
+from pyaml.bpm.bpm_model import BPMModel
+
+PYAMLCLASS = "BPM"
+
+class ConfigModel(ElementConfigModel):
+
+        model: BPMModel | None = None
+        """Object in charge of BPM modeling"""
+
+
+
+class BPM(Element):
+    """
+    Class providing access to one BPM of a physical or simulated lattice
+    """
+
+    def __init__(self, cfg: ConfigModel):
+        """
+        Construct a BPM
+
+        Parameters
+        ----------
+        name : str
+            Element name
+        hardware : DeviceAccess
+            Direct access to a hardware (bypass the BPM model)
+        model : BPMModel
+            BPM model in charge of computing beam position
+        """
+
+        super().__init__(cfg.name)
+
+        self.__model = cfg.model if hasattr(cfg, "model") else None
+        self._cfg = cfg
+        self.__positions = None
+        self.__offset = None
+        self.__tilt = None
+
+    @property
+    def model(self) -> BPMModel:
+         return self.__model
+
+    @property
+    def positions(self) -> RBpmArray:
+        if self.__positions is None:
+            raise Exception(f"{str(self)} has no attached positions") 
+        return self.__positions
+
+    @property
+    def offset(self) -> RWBpmOffsetArray:
+        if self.__offset is None:
+            raise Exception(f"{str(self)} has no attached offset")
+        return self.__offset
+
+    @property
+    def tilt(self) -> RWBpmTiltScalar:
+        if self.__tilt is None:
+            raise Exception(f"{str(self)} has no attached tilt")
+        return self.__tilt
+
+    def attach(self, positions: RBpmArray , offset: RWBpmOffsetArray,
+               tilt: RWBpmTiltScalar) -> Self:
+        # Attach positions, offset and tilt attributes and returns a new
+        # reference
+        obj = self.__class__(self._cfg)
+        obj.__model = self.__model
+        obj.__positions = positions
+        obj.__offset = offset
+        obj.__tilt = tilt
+        return obj
+

pyaml/bpm/bpm_model.py

@@ -0,0 +1,70 @@
+from abc import ABCMeta, abstractmethod
+import numpy as np
+from numpy.typing import NDArray
+
+class BPMModel(metaclass=ABCMeta):
+    """
+    Abstract class providing interface to accessing BPM positions, offsets,
+    tilts.
+    """
+    @abstractmethod
+    def read_position(self) -> NDArray[np.float64]:
+        """
+        Read horizontal and vertical positions from a BPM.
+        Returns
+        -------
+        NDArray[np.float64]
+            Array of shape (2,) containing the horizontal and vertical
+            positions
+        """
+        pass
+
+    @abstractmethod
+    def read_tilt(self) -> float:
+        """
+        Read the tilt value from a BPM.
+        Returns
+        -------
+        float
+            The tilt value of the BPM
+        """
+        pass
+
+    @abstractmethod
+    def read_offset(self) -> NDArray:
+        """
+        Read the offset values from a BPM.
+        Returns
+        -------
+        NDArray[np.float64]
+            Array of shape (2,) containing the horizontal and vertical
+            offsets
+        """
+        pass
+
+    @abstractmethod
+    def set_tilt(self, tilt: float):
+        """
+        Set the tilt value of a BPM.
+        Parameters
+        ----------
+        tilt : float
+            The tilt value to set for the BPM
+        Returns
+        -------
+        None
+        """
+        pass
+
+    @abstractmethod
+    def set_offset(self, offset: NDArray[np.float64]):
+        """
+        Set the offset values of a BPM
+        Parameters
+        ----------
+        offset_values : NDArray[np.float64]
+            Array of shape (2,) containing the horizontal and vertical
+            offsets to set for the BPM
+        """
+        pass
+

pyaml/bpm/bpm_simple_model.py

@@ -0,0 +1,83 @@
+from pyaml.bpm.bpm_model import BPMModel
+from pydantic import BaseModel,ConfigDict
+import numpy as np
+from ..control.deviceaccess import DeviceAccess
+from numpy.typing import NDArray
+# Define the main class name for this module
+PYAMLCLASS = "BPMSimpleModel"
+
+class ConfigModel(BaseModel):
+
+    model_config = ConfigDict(arbitrary_types_allowed=True,extra="forbid")
+
+    x_pos: DeviceAccess
+    """Horizontal position"""
+    y_pos: DeviceAccess
+    """Vertical position"""
+
+class BPMSimpleModel(BPMModel):
+    """
+    Concrete implementation of BPMModel that simulates a BPM with tilt and
+    offset values.
+    """
+    def __init__(self, cfg: ConfigModel):
+        self._cfg = cfg 
+
+        self.__x_pos = cfg.x_pos
+        self.__y_pos = cfg.y_pos
+
+    def read_position(self) -> NDArray:
+        """
+        Simulate reading the position values from a BPM.
+        Returns
+        -------
+        np.ndarray
+            Array of shape (2,) containing the horizontal and vertical
+            positions
+        """
+        return np.array([self.__x_pos.get(), self.__y_pos.get()])
+
+    def read_tilt(self) -> float:
+        """
+        Simulate reading the tilt value from a BPM.
+        Returns
+        -------
+        float
+            The tilt value of the BPM
+        """
+        raise NotImplementedError("Tilt reading not implemented in this model.")
+
+    def read_offset(self) -> NDArray:
+        """
+        Simulate reading the offset values from a BPM.
+        Returns
+        -------
+        np.ndarray
+            Array of shape (2,) containing the horizontal and vertical
+            offsets
+        """
+        raise NotImplementedError("Offset reading not implemented in this model.")
+    def set_tilt(self, tilt: float):
+        """
+        Simulate setting the tilt value of a BPM.
+        Parameters
+        ----------
+        tilt : float
+            The tilt value to set for the BPM
+        Returns
+        -------
+        None
+        """
+        raise NotImplementedError("Tilt setting not implemented in this model.")
+
+    def set_offset(self, offset_values: np.ndarray):
+        """
+        Simulate setting the offset values of a BPM
+        Parameters
+        ----------
+        offset_values : np.ndarray
+            Array of shape (2,) containing the horizontal and vertical
+            offsets to set for the BPM
+        """
+        raise NotImplementedError("Offset setting not implemented in this model.")
+

pyaml/bpm/bpm_tiltoffset_model.py

@@ -0,0 +1,83 @@
+from pyaml.bpm.bpm_model import BPMModel
+from pyaml.bpm.bpm_simple_model import BPMSimpleModel
+from pydantic import BaseModel,ConfigDict
+import numpy as np
+from ..control.deviceaccess import DeviceAccess
+from numpy.typing import NDArray
+# Define the main class name for this module
+PYAMLCLASS = "BPMTiltOffsetModel"
+
+class ConfigModel(BaseModel):
+
+    model_config = ConfigDict(arbitrary_types_allowed=True,extra="forbid")
+
+    x_pos: DeviceAccess
+    """Horizontal position"""
+    y_pos: DeviceAccess
+    """Vertical position"""
+    x_offset: DeviceAccess
+    """Horizontal BPM offset"""
+    y_offset: DeviceAccess
+    """Vertical BPM offset"""
+    tilt: DeviceAccess
+    """BPM tilt"""
+
+class BPMTiltOffsetModel(BPMSimpleModel):
+    """
+    Concrete implementation of BPMModel that simulates a BPM with tilt and
+    offset values.
+    """
+    def __init__(self, cfg: ConfigModel):
+        super().__init__(cfg) 
+        self.__x_pos = cfg.x_pos
+        self.__y_pos = cfg.y_pos
+        self.__x_offset = cfg.x_offset
+        self.__y_offset = cfg.y_offset
+        self.__tilt = cfg.tilt
+
+    def read_tilt(self) -> float:
+        """
+        Simulate reading the tilt value from a BPM.
+        Returns
+        -------
+        float
+            The tilt value of the BPM
+        """
+        return self.__tilt.get()
+
+    def read_offset(self) -> NDArray:
+        """
+        Simulate reading the offset values from a BPM.
+        Returns
+        -------
+        np.ndarray
+            Array of shape (2,) containing the horizontal and vertical
+            offsets
+        """
+        return np.array([self.__x_offset.get(), self.__y_offset.get()])
+
+    def set_tilt(self, tilt: float):
+        """
+        Simulate setting the tilt value of a BPM.
+        Parameters
+        ----------
+        tilt : float
+            The tilt value to set for the BPM
+        Returns
+        -------
+        None
+        """
+        self.__tilt.set(tilt)
+
+    def set_offset(self, offset_values: np.ndarray):
+        """
+        Simulate setting the offset values of a BPM
+        Parameters
+        ----------
+        offset_values : np.ndarray
+            Array of shape (2,) containing the horizontal and vertical
+            offsets to set for the BPM
+        """
+        self.__x_offset.set(offset_values[0])
+        self.__y_offset.set(offset_values[1])
+

pyaml/control/abstract_impl.py

@@ -1,11 +1,11 @@
 from numpy import double
-import numpy as np
-
-from ..control import abstract
-from ..magnet.model import MagnetModel
+from pyaml.control import abstract
+from pyaml.magnet.model import MagnetModel
+from pyaml.bpm.bpm_model import BPMModel
 from ..rf.rf_plant import RFPlant
 from ..rf.rf_transmitter import RFTransmitter
-
+import numpy as np
+from numpy.typing import NDArray
 #------------------------------------------------------------------------------
 
 class RWHardwareScalar(abstract.ReadWriteFloatScalar):
@@ -116,6 +116,66 @@ def set_and_wait(self, value:np.array):
     def unit(self) -> list[str]:
         return self.__model.get_strength_units()
 
+#------------------------------------------------------------------------------
+
+class RBpmArray(abstract.ReadFloatArray):
+    """
+    Class providing read access to a BPM array of a control system
+    """
+    def __init__(self, model:BPMModel):
+        self.__model = model
+
+    # Gets the value
+    def get(self) -> np.array:
+        return self.__model.read_position()
+
+    # Gets the unit of the value
+    def unit(self) -> list[str]:
+        return [self.__model.__x_pos.unit(), self.__model.__y_pos.unit()]
+
+#------------------------------------------------------------------------------
+
+class RWBpmTiltScalar(abstract.ReadFloatScalar):
+    """
+    Class providing read access to a BPM tilt of a control system
+    """
+    def __init__(self, model:BPMModel):
+        self.__model = model
+
+    # Gets the value
+    def get(self) -> float:
+        return self.__model.read_tilt()
+
+    def set(self, value:float):
+        self.__model.set_tilt(value)
+
+    def set_and_wait(self, value: NDArray[np.float64]):
+        raise NotImplementedError("Not implemented yet.")
+    # Gets the unit of the value
+    def unit(self) -> str:
+        return self.__model.__tilt.unit()
+
+#------------------------------------------------------------------------------
+
+class RWBpmOffsetArray(abstract.ReadWriteFloatArray):
+    """
+    Class providing read write access to a BPM offset of a control system
+    """
+    def __init__(self, model: BPMModel):
+        self.__model = model
+
+    # Gets the value
+    def get(self) -> NDArray[np.float64]:
+        return self.__model.read_offset()
+
+    # Sets the value
+    def set(self, value: NDArray[np.float64]):
+        self.__model.set_offset(value) 
+    def set_and_wait(self, value: NDArray[np.float64]):
+        raise NotImplementedError("Not implemented yet.")
+    # Gets the unit of the value
+    def unit(self) -> str:
+        return self.__model.__x_offset.unit()
 
 #------------------------------------------------------------------------------