Added tests

Author: JeanLucPons

pyaml/arrays/bpm_array.py

@@ -9,22 +9,22 @@ class RWBPMPosition(ReadFloatArray):
     def __init__(self, name:str, bpms:list[BPM]):
         self.__bpms = bpms
         self.__name = name
-        self.aggregator:DeviceAccessList = None
+        self.__aggregator:DeviceAccessList = None
 
     # Gets the values
     def get(self) -> np.array:
-        if not self.aggregator:
+        if not self.__aggregator:
             return np.array([b.positions.get() for b in self.__bpms])
         else:  
-            return self.aggregator.get().reshape(len(self.__bpms),2)
+            return self.__aggregator.get().reshape(len(self.__bpms),2)
 
     # Gets the unit of the values
     def unit(self) -> list[str]:
         return [b.positions.unit() for b in self.__bpms]
 
     # Set the aggregator (Control system only)
     def set_aggregator(self,agg:DeviceAccessList):
-        self.aggregator = agg
+        self.__aggregator = agg
 
 
 class RWBPMSinglePosition(ReadFloatArray):
@@ -33,22 +33,22 @@ def __init__(self, name:str, bpms:list[BPM],idx: int):
         self.__bpms = bpms
         self.__name = name
         self.__idx = idx
-        self.aggregator:DeviceAccessList = None
+        self.__aggregator:DeviceAccessList = None
 
     # Gets the values
     def get(self) -> np.array:
-        if not self.aggregator:
+        if not self.__aggregator:
             return np.array([b.positions.get()[self.__idx] for b in self.__bpms])
         else:
-            return self.aggregator.get()
+            return self.__aggregator.get()
 
     # Gets the unit of the values
     def unit(self) -> list[str]:
         return [b.positions.unit() for b in self.__bpms]
 
     # Set the aggregator (Control system only)
     def set_aggregator(self,agg:DeviceAccessList):
-        self.aggregator = agg
+        self.__aggregator = agg
 
 
 
@@ -57,7 +57,7 @@ class BPMArray(list[BPM]):
     Class that implements access to a BPM array
     """
 
-    def __init__(self,arrayName:str,bpms:list[BPM],holder):
+    def __init__(self,arrayName:str,bpms:list[BPM],holder = None):
         """
         Construct a BPM array
 
@@ -68,34 +68,35 @@ def __init__(self,arrayName:str,bpms:list[BPM],holder):
         bpms: list[BPM]
             BPM iterator
         holder : Element holder
-            Holder that contains element of this array (Simulator or Control System)
+            Holder (Simulator or Control System) that contains element of this array used for aggregator
         """
         super().__init__(i for i in bpms)
         self.__name = arrayName
         self.__hvpos = RWBPMPosition(arrayName,bpms)
         self.__hpos = RWBPMSinglePosition(arrayName,bpms,0)
         self.__vpos = RWBPMSinglePosition(arrayName,bpms,1)
 
-        aggs = holder.create_bpm_aggregators(bpms)
-        self.__hvpos.set_aggregator(aggs[0])
-        self.__hpos.set_aggregator(aggs[1])
-        self.__vpos.set_aggregator(aggs[2])
+        if holder is not None:    
+            aggs = holder.create_bpm_aggregators(bpms)
+            self.__hvpos.set_aggregator(aggs[0])
+            self.__hpos.set_aggregator(aggs[1])
+            self.__vpos.set_aggregator(aggs[2])
 
-    @property   
+    @property
     def positions(self) -> RWBPMPosition:
         """
         Give access to bpm posttions of each bpm of this array
         """
         return self.__hvpos
 
-    @property   
+    @property
     def h(self) -> RWBPMSinglePosition:
         """
         Give access to bpm H posttions of each bpm of this array
         """
         return self.__hpos
 
-    @property   
+    @property
     def v(self) -> RWBPMSinglePosition:
         """
         Give access to bpm V posttions of each bpm of this array

pyaml/arrays/magnet_array.py

@@ -7,48 +7,26 @@
 class RWMagnetStrength(ReadWriteFloatArray):
 
     def __init__(self, name:str, magnets:list[Magnet]):
-        self.__magnets = magnets
         self.__name = name
-        self.aggregator:ScalarAggregator = None
+        self.__magnets = magnets
+        self.__nb = len(self.__magnets)
+        self.__aggregator:ScalarAggregator = None
 
     # Gets the values
     def get(self) -> np.array:
-        if not self.aggregator:
+        if not self.__aggregator:
             return np.array([m.strength.get() for m in self.__magnets])
         else:
-            allHardwareValues = self.aggregator.get() # Read all hardware setpoints
-            allStrength = np.zeros(len(self.__magnets))
-            mIdx = 0
-            idx = 0
-            for m in self.__magnets:
-                nbDev = len(m.model.get_devices())
-                allStrength[mIdx] = m.model.compute_strengths(allHardwareValues[idx:idx+nbDev])[m.strength.index()]
-                mIdx += 1
-                idx += nbDev
-            return allStrength
+            return self.__aggregator.get()
 
     # Sets the values
     def set(self, value:np.array):
-        if not self.aggregator:
+        nvalue = np.ones(self.__nb) * value if isinstance(value,float) else value        
+        if not self.__aggregator:
             for idx,m in enumerate(self.__magnets):
-                m.strength.set(value[idx])
+                m.strength.set(nvalue[idx])
         else:
-            # TODO: if the array does not contains mappings to combined function 
-            # magnets, the algorithm below can be optimized
-            allHardwareValues = self.aggregator.get() # Read all hardware setpoints
-            newHardwareValues = np.zeros(len(self.aggregator))
-            mIdx = 0
-            idx = 0
-            for m in self.__magnets:
-                # m is a single function magnet or a mapping to a 
-                # combined function magnet (RWMapper)
-                nbDev = len(m.model.get_devices())
-                mStrengths = m.model.compute_strengths( allHardwareValues[idx:idx+nbDev] )
-                mStrengths[m.strength.index()] = value[mIdx]
-                newHardwareValues[idx:idx+nbDev] = m.model.compute_hardware_values(mStrengths)
-                mIdx += 1
-                idx += nbDev
-            self.aggregator.set(newHardwareValues)
+            self.__aggregator.set(nvalue)
 
     # Sets the values and waits that the read values reach their setpoint
     def set_and_wait(self, value:np.array):
@@ -60,36 +38,31 @@ def unit(self) -> list[str]:
 
     # Set the aggregator (Control system only)
     def set_aggregator(self,agg:ScalarAggregator):
-        self.aggregator = agg
+        self.__aggregator = agg
 
 class RWMagnetHardware(ReadWriteFloatArray):
 
     def __init__(self, name:str, magnets:list[Magnet]):
         self.__name = name
         self.__magnets = magnets
-        self.aggregator:ScalarAggregator = None
-        self.hasHardwareMapping = True
+        self.__nb = len(self.__magnets)
+        self.__aggregator:ScalarAggregator = None
 
     # Gets the values
     def get(self) -> np.array:
-        if not self.aggregator:
+        if not self.__aggregator:
             return np.array([m.hardware.get() for m in self.__magnets])
         else:
-            if not self.hasHardwareMapping:
-                raise Exception(f"Array {self.__name} contains elements that that do not support hardware units")
-            else:
-                return self.aggregator.get()
+            return self.__aggregator.get()
 
     # Sets the values
     def set(self, value:np.array):
-        if not self.aggregator:
+        nvalue = np.ones(self.__nb) * value if isinstance(value,float) else value        
+        if not self.__aggregator:
             for idx,m in enumerate(self.__magnets):
                 m.hardware.set(value[idx])
         else:
-            if not self.hasHardwareMapping:
-                raise Exception(f"Array {self.__name} contains elements that that do not support hardware units")
-            else:
-                self.aggregator.set(value)
+            self.__aggregator.set(value)
 
     # Sets the values and waits that the read values reach their setpoint
     def set_and_wait(self, value:np.array):
@@ -101,16 +74,14 @@ def unit(self) -> list[str]:
 
     # Set the aggregator
     def set_aggregator(self,agg:ScalarAggregator):
-        self.aggregator = agg
-        for m in self.__magnets:
-            self.hasHardwareMapping |= m.model.has_hardware()
+        self.__aggregator = agg
 
 class MagnetArray(list[Magnet]):
     """
     Class that implements access to a magnet array
     """
 
-    def __init__(self,arrayName:str,magnets:list[Magnet],holder):
+    def __init__(self,arrayName:str,magnets:list[Magnet],holder = None):
         """
         Construct a magnet array
 
@@ -121,16 +92,18 @@ def __init__(self,arrayName:str,magnets:list[Magnet],holder):
         magnets: list[Magnet]
             Magnet iterator
         holder : Element holder
-            Holder that contains element of this array (Simulator or Control System)
+            Holder (Simulator or Control System) that contains element of this array used for aggregator
         """
         super().__init__(i for i in magnets)
         self.__name = arrayName
         self.__rwstrengths = RWMagnetStrength(arrayName,magnets)
         self.__rwhardwares = RWMagnetHardware(arrayName,magnets)
 
-        agg = holder.create_magnet_aggregator(magnets)
-        self.__rwstrengths.set_aggregator(agg)
-        self.__rwhardwares.set_aggregator(agg)
+        if holder is not None:
+            aggs = holder.create_magnet_strength_aggregator(magnets)
+            aggh = holder.create_magnet_harddware_aggregator(magnets)
+            self.__rwstrengths.set_aggregator(aggs)
+            self.__rwhardwares.set_aggregator(aggh)
 
     @property        
     def strengths(self) -> RWMagnetStrength:

pyaml/control/abstract_impl.py

@@ -2,6 +2,7 @@
 from ..control.deviceaccesslist import DeviceAccessList
 from ..control.deviceaccess import DeviceAccess
 from ..magnet.model import MagnetModel
+from ..magnet.magnet import Magnet
 from ..bpm.bpm_model import BPMModel
 from ..rf.rf_plant import RFPlant
 from ..rf.rf_transmitter import RFTransmitter
@@ -15,29 +16,107 @@
 
 class CSScalarAggregator(ScalarAggregator):
     """
-    Control system aggregator for a scalar value
+    Basic control system aggregator for a list of scalar values
     """
 
     def __init__(self, devs:DeviceAccessList):
-        self.__devs = devs
+        self._devs = devs
 
     def add_devices(self, devices:DeviceAccess | list[DeviceAccess] ):
-        self.__devs.add_devices(devices)
+        self._devs.add_devices(devices)
 
     def set(self, value: NDArray[np.float64]):
-        self.__devs.set(value)
+        self._devs.set(value)
 
     def set_and_wait(self, value: NDArray[np.float64]):
-        self.__devs.set_and_wait(value)
+        self._devs.set_and_wait(value)
 
     def get(self) -> NDArray[np.float64]:
-        return self.__devs.get()
+        return self._devs.get()
 
     def readback(self) -> np.array:
-        return self.__devs.readback()
+        return self._devs.readback()
 
     def unit(self) -> str:
-        return self.__devs.unit()
+        return self._devs.unit()
+    
+    def nb_device(self) -> int:
+        return self._devs.__len__()
+
+#------------------------------------------------------------------------------
+
+class CSStrengthScalarAggregator(CSScalarAggregator):
+    """
+    Control system aggregator for a list of magnet strengths.
+    This aggregator is in charge of computing hardware setpoints and applying them without overlap.
+    When virtual magnets exported from combined function mangets are present (RWMapper), 
+    the aggregator prevents to apply several times the same power supply setpoint.
+    """
+
+    def __init__(self, peer:CSScalarAggregator):
+        CSScalarAggregator.__init__(self,peer._devs)
+        self.__models: list[MagnetModel] = []                   # List of magnet model
+        self.__modelToMagnet: list[list[tuple[int,int]]] = []   # strengths indexing
+        self.__nbMagnet = 0                                     # Number of magnet strengths
+
+    def add_magnet(self, magnet:Magnet):
+        # Incoming magnet can be a magnet exported from a CombinedFunctionMagnet or simple magnet.
+        # All magnets exported from a same CombinedFunctionMagnet share the same model
+        # TODO: check that strength is supported (m.strength may be None)
+        strengthIndex = magnet.strength.index() if isinstance(magnet.strength,abstract.RWMapper) else 0
+        if magnet.model not in self.__models:
+            index = len(self.__models)
+            self.__models.append(magnet.model)
+            self.__modelToMagnet.append([(self.__nbMagnet,strengthIndex)])
+            self._devs.add_devices(magnet.model.get_devices())
+        else:
+            index = self.__models.index(magnet.model)
+            self.__modelToMagnet[index].append((self.__nbMagnet,strengthIndex))
+        self.__nbMagnet += 1
+
+    def set(self, value: NDArray[np.float64]):
+        allHardwareValues = self._devs.get() # Read all hardware setpoints
+        newHardwareValues = np.zeros(self.nb_device())
+        hardwareIndex = 0
+        for modelIndex,model in enumerate(self.__models):
+            nbDev = len(model.get_devices())
+            mStrengths = model.compute_strengths( allHardwareValues[hardwareIndex:hardwareIndex+nbDev] )
+            for (valueIdx,strengthIdx) in self.__modelToMagnet[modelIndex]:
+                mStrengths[strengthIdx] = value[valueIdx]
+            newHardwareValues[hardwareIndex:hardwareIndex+nbDev] = model.compute_hardware_values(mStrengths)
+            hardwareIndex += nbDev
+        self._devs.set(newHardwareValues)
+
+    def set_and_wait(self, value: NDArray[np.float64]):
+        raise NotImplementedError("Not implemented yet.")
+
+    def get(self) -> NDArray[np.float64]:
+        allHardwareValues = self._devs.get() # Read all hardware setpoints
+        allStrength = np.zeros(self.__nbMagnet)
+        hardwareIndex = 0
+        for modelIndex,model in enumerate(self.__models):
+            nbDev = len(model.get_devices())
+            mStrengths = model.compute_strengths( allHardwareValues[hardwareIndex:hardwareIndex+nbDev] )
+            for (valueIdx,strengthIdx) in self.__modelToMagnet[modelIndex]:
+                allStrength[valueIdx] = mStrengths[strengthIdx]
+            hardwareIndex += nbDev
+        return allStrength
+
+    def readback(self) -> np.array:
+        allHardwareValues = self._devs.readback() # Read all hardware readback
+        allStrength = np.zeros(self.__nbMagnet)
+        hardwareIndex = 0
+        for modelIndex,model in enumerate(self.__models):
+            nbDev = len(model.get_devices())
+            mStrengths = model.compute_strengths( allHardwareValues[hardwareIndex:hardwareIndex+nbDev] )
+            for (valueIdx,strengthIdx) in self.__modelToMagnet[modelIndex]:
+                allStrength[valueIdx] = mStrengths[strengthIdx]
+            hardwareIndex += nbDev
+        return allStrength
+
+    def unit(self) -> str:
+        return self._devs.unit()
+
 
 #------------------------------------------------------------------------------
 
@@ -61,9 +140,6 @@ def set_and_wait(self, value: double):
     def unit(self) -> str:
         return self.__model.get_hardware_units()[0]
 
-    def index(self) -> int:
-        return 0
-
 #------------------------------------------------------------------------------
 
 class RWStrengthScalar(abstract.ReadWriteFloatScalar):
@@ -92,9 +168,6 @@ def set_and_wait(self, value:float):
     def unit(self) -> str:
         return self.__model.get_strength_units()[0]
 
-    def index(self) -> int:
-        return 0
-
 #------------------------------------------------------------------------------
 
 class RWHardwareArray(abstract.ReadWriteFloatArray):