New tuning and measurement tools architecture

examples/ESRF_ORM_example/measure_ideal_ORM_and_disp.py

@@ -6,9 +6,7 @@
 from pyaml.accelerator import Accelerator
 
 parent_folder = Path(__file__).parent
-config_path = parent_folder.parent.parent.joinpath(
-    "tests", "config", "EBSOrbit.yaml"
-).resolve()
+config_path = parent_folder.parent.parent.joinpath("tests", "config", "EBSOrbit.yaml").resolve()
 sr = Accelerator.load(config_path)
 
 ebs = sr.design
@@ -19,18 +17,9 @@
 ebs.dispersion.measure()
 dispersion_data = ebs.dispersion.get()
 
-rf_response = (
-    dispersion_data["frequency_response_x"] + dispersion_data["frequency_response_y"]
-)
-
-ideal_ORM_data = {
-    "type": "pyaml.tuning_tools.response_matrix",
-    "matrix": orm_data["matrix"],
-    "input_names": orm_data["input_names"],
-    "output_names": orm_data["output_names"],
-    "input_planes": orm_data["input_planes"],
-    "output_planes": orm_data["output_planes"],
-    "rf_response": rf_response,
-}
-
-json.dump(ideal_ORM_data, open("ideal_orm_disp.json", "w"))
+rf_response = dispersion_data["frequency_response_x"] + dispersion_data["frequency_response_y"]
+
+orm_data["rf_response"] = rf_response
+orm_data["type"] = "pyaml.tuning_tools.orbit_response_matrix_data"
+
+json.dump(orm_data, open("ideal_orm_disp.json", "w"))

pyaml/apidoc/gen_api.py

@@ -10,6 +10,7 @@
 
 modules = [
     "pyaml.accelerator",
+    "pyaml.apidoc.gen_api",
     "pyaml.arrays.array",
     "pyaml.arrays.bpm",
     "pyaml.arrays.bpm_array",
@@ -80,8 +81,12 @@
     "pyaml.tuning_tools.dispersion",
     "pyaml.tuning_tools.orbit",
     "pyaml.tuning_tools.orbit_response_matrix",
-    "pyaml.tuning_tools.response_matrix",
+    "pyaml.tuning_tools.measurement_tool",
     "pyaml.tuning_tools.tune",
+    "pyaml.tuning_tools.orbit_response_matrix_data",
+    "pyaml.tuning_tools.response_matrix_data",
+    "pyaml.tuning_tools.tune_response_matrix",
+    "pyaml.tuning_tools.tuning_tool",
     "pyaml.yellow_pages",
 ]
 

pyaml/common/constants.py

@@ -5,25 +5,21 @@
 
     Horizontal kick sign convention for both kick an kick angle (default: -1).
 
-.. data:: ACTION_APPLY
-
-    Action identifier for callback in measurement tools
-    Triggered just after exitation has been sent
+"""
 
-.. data:: ACTION_RESTORE
+from enum import Enum
 
-    Action identifier for callback in measurement tools
-    Triggered just after exitation has been restored
+HORIZONTAL_KICK_SIGN: float = -1.0
 
-.. data:: ACTION_MEASURE
 
+class Action(Enum):
+    """
     Action identifier for callback in measurement tools
-    Triggered just after measurement
-
-"""
-
-HORIZONTAL_KICK_SIGN: float = -1.0
-
-ACTION_APPLY: int = 0
-ACTION_RESTORE: int = 1
-ACTION_MEASURE: int = 2
+    """
+
+    APPLY = 0
+    "Triggered imediatly after actuator exitation"
+    RESTORE = 1
+    "Triggered imediatly after actuator has been restored to initial value"
+    MEASURE = 2
+    "Triggered imediatly after measurement"

pyaml/common/element.py

@@ -103,19 +103,13 @@ def check_peer(self):
         to a simulator or to a control system
         """
         if self._peer is None:
-            raise PyAMLException(
-                f"{str(self)} is not attachedto a control system or the a simulator"
-            )
+            raise PyAMLException(f"{str(self)} is not attachedto a control system or the a simulator")
 
     def get_peer(self) -> str:
         """
         Returns a string representation of peer simulator or control system
         """
-        return (
-            "None"
-            if self._peer is None
-            else f"{self._peer.__class__.__name__}:{self._peer.name()}"
-        )
+        return "None" if self._peer is None else f"{self._peer.__class__.__name__}:{self._peer.name()}"
 
     def post_init(self):
         """

pyaml/common/element_holder.py

@@ -23,10 +23,13 @@
 from .element import Element
 
 if TYPE_CHECKING:
+    from ..tuning_tools.chromaticity import Chromaticity
+    from ..tuning_tools.chromaticity_response_matrix import ChromaticityResponseMatrix
     from ..tuning_tools.dispersion import Dispersion
     from ..tuning_tools.orbit import Orbit
     from ..tuning_tools.orbit_response_matrix import OrbitResponseMatrix
     from ..tuning_tools.tune import Tune
+    from ..tuning_tools.tune_response_matrix import TuneResponseMatrix
 
 
 class ElementHolder(object):
@@ -258,53 +261,68 @@ def get_betatron_tune_monitor(self, name: str) -> BetatronTuneMonitor:
     def add_betatron_tune_monitor(self, tune_monitor: Element):
         self.__add(self.__DIAG, tune_monitor)
 
-    # Chromaticity monitor
+    # Tuning/Measurement tools
+
+    def add_tool(self, tool: Element):
+        self.__add(self.__TUNING_TOOLS, tool)
+
+    # ---- Chromaticity -------------------------------------------------
 
     def get_chromaticity_monitor(self, name: str) -> ChomaticityMonitor:
-        obj = self.__get("Diagnostic", name, self.__DIAG)
+        obj = self.__get("Chomaticity monitor", name, self.__TUNING_TOOLS)
         return obj
 
-    def add_chromaticity_monitor(self, chromaticity_monitor: Element):
-        self.__add(self.__DIAG, chromaticity_monitor)
+    def get_chromaticity_tuning(self, name: str) -> "Chromaticity":
+        return self.__get("Chromaticity tool", name, self.__TUNING_TOOLS)
+
+    def get_crm_tuning(self, name: str) -> "ChromaticityResponseMatrix":
+        return self.__get("ChromaticityResponseMatrix tool", name, self.__TUNING_TOOLS)
+
+    @property
+    def chromaticity(self) -> "Chromaticity":
+        return self.get_chromaticity_tuning("DEFAULT_CHROMATICITY_CORRECTION")
+
+    @property
+    def crm(self) -> "ChromaticityResponseMatrix":
+        return self.get_crm_tuning("DEFAULT_CHROMATICITY_RESPONSE_MATRIX")
 
-    # Tuning tools
+    # ---- Tune ---------------------------------------------------------
 
     def get_tune_tuning(self, name: str) -> "Tune":
         return self.__get("Tune tuning tool", name, self.__TUNING_TOOLS)
 
-    def add_tune_tuning(self, tune: Element):
-        self.__add(self.__TUNING_TOOLS, tune)
-
     @property
     def tune(self) -> "Tune":
         return self.get_tune_tuning("DEFAULT_TUNE_CORRECTION")
 
+    def get_trm_tuning(self, name: str) -> "TuneResponseMatrix":
+        return self.__get("TuneResponseMatrix tool", name, self.__TUNING_TOOLS)
+
+    @property
+    def trm(self) -> "TuneResponseMatrix":
+        return self.get_trm_tuning("DEFAULT_TUNE_RESPONSE_MATRIX")
+
+    # ---- Orbit --------------------------------------------------------
+
     def get_orbit_tuning(self, name: str) -> "Orbit":
         return self.__get("Orbit tuning tool", name, self.__TUNING_TOOLS)
 
-    def add_orbit_tuning(self, orbit: Element):
-        self.__add(self.__TUNING_TOOLS, orbit)
-
     @property
     def orbit(self) -> "Orbit":
         return self.get_orbit_tuning("DEFAULT_ORBIT_CORRECTION")
 
     def get_orm_tuning(self, name: str) -> "OrbitResponseMatrix":
         return self.__get("OrbitResponseMatrix tool", name, self.__TUNING_TOOLS)
 
-    def add_orm_tuning(self, orm: Element):
-        self.__add(self.__TUNING_TOOLS, orm)
-
     @property
     def orm(self) -> "OrbitResponseMatrix":
         return self.get_orm_tuning("DEFAULT_ORBIT_RESPONSE_MATRIX")
 
+    # ---- Dispersive orbit --------------------------------------------
+
     def get_dispersion_tuning(self, name: str) -> "Dispersion":
         return self.__get("Dispersion tool", name, self.__TUNING_TOOLS)
 
-    def add_dispersion_tuning(self, dispersion: Element):
-        self.__add(self.__TUNING_TOOLS, dispersion)
-
     @property
     def dispersion(self) -> "Dispersion":
         return self.get_dispersion_tuning("DEFAULT_DISPERSION")

pyaml/control/abstract_impl.py

@@ -66,10 +66,7 @@ def check_range(values: Any, dev_range: Any) -> bool:
         mins = np.full(n, mins_obj[0], dtype=object)
         maxs = np.full(n, maxs_obj[0], dtype=object)
     else:
-        raise ValueError(
-            f"Inconsistent sizes: {n} value(s) for {k} range(s). "
-            f"Supported: N==K, N==1, or K==1."
-        )
+        raise ValueError(f"Inconsistent sizes: {n} value(s) for {k} range(s). Supported: N==K, N==1, or K==1.")
 
     # ---- Replace None bounds with -inf / +inf (NumPy-safe) ----
     mins_is_none = np.equal(mins, None)
@@ -99,11 +96,7 @@ def _iter_devices_and_ranges(devs: DeviceAccess | DeviceAccessList):
       - DeviceAccessList: yields N items based on get_devices() and get_range() flattening
     """
     # Single device
-    if (
-        hasattr(devs, "get")
-        and hasattr(devs, "get_range")
-        and not hasattr(devs, "get_devices")
-    ):
+    if hasattr(devs, "get") and hasattr(devs, "get_range") and not hasattr(devs, "get_devices"):
         r = devs.get_range()
         if r is None:
             r = [None, None]
@@ -160,13 +153,10 @@ def format_out_of_range_message(
         vv = v
         pairs = [dev_pairs[0]] * n
     else:
-        raise ValueError(
-            f"Inconsistent sizes: {n} value(s) for {k} device(s). "
-            f"Supported: N==K, N==1, or K==1."
-        )
+        raise ValueError(f"Inconsistent sizes: {n} value(s) for {k} device(s). Supported: N==K, N==1, or K==1.")
 
     lines = [header]
-    for val, (dev, r) in zip(vv, pairs):
+    for val, (dev, r) in zip(vv, pairs, strict=True):
         if not check_range(val, r):
             unit = dev.unit() if hasattr(dev, "unit") else ""
             name = str(dev)
@@ -233,11 +223,7 @@ def add_magnet(self, magnet: Magnet, devs: list[DeviceAccess]):
         # 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
-        )
+        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)
@@ -254,20 +240,14 @@ def set(self, value: NDArray[np.float64]):
         hardwareIndex = 0
         for modelIndex, model in enumerate(self.__models):
             nbDev = len(model.get_devices())
-            mStrengths = model.compute_strengths(
-                allHardwareValues[hardwareIndex : hardwareIndex + nbDev]
-            )
+            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)
-            )
+            newHardwareValues[hardwareIndex : hardwareIndex + nbDev] = model.compute_hardware_values(mStrengths)
             hardwareIndex += nbDev
         dev_range = self._devs.get_range()
         if not check_range(newHardwareValues, dev_range):
-            raise PyAMLException(
-                format_out_of_range_message(newHardwareValues, self._devs)
-            )
+            raise PyAMLException(format_out_of_range_message(newHardwareValues, self._devs))
         self._devs.set(newHardwareValues)
 
     def set_and_wait(self, value: NDArray[np.float64]):
@@ -279,9 +259,7 @@ def get(self) -> NDArray[np.float64]:
         hardwareIndex = 0
         for modelIndex, model in enumerate(self.__models):
             nbDev = len(model.get_devices())
-            mStrengths = model.compute_strengths(
-                allHardwareValues[hardwareIndex : hardwareIndex + nbDev]
-            )
+            mStrengths = model.compute_strengths(allHardwareValues[hardwareIndex : hardwareIndex + nbDev])
             for valueIdx, strengthIdx in self.__modelToMagnet[modelIndex]:
                 allStrength[valueIdx] = mStrengths[strengthIdx]
             hardwareIndex += nbDev
@@ -293,9 +271,7 @@ def readback(self) -> np.array:
         hardwareIndex = 0
         for modelIndex, model in enumerate(self.__models):
             nbDev = len(model.get_devices())
-            mStrengths = model.compute_strengths(
-                allHardwareValues[hardwareIndex : hardwareIndex + nbDev]
-            )
+            mStrengths = model.compute_strengths(allHardwareValues[hardwareIndex : hardwareIndex + nbDev])
             for valueIdx, strengthIdx in self.__modelToMagnet[modelIndex]:
                 allStrength[valueIdx] = mStrengths[strengthIdx]
             hardwareIndex += nbDev
@@ -706,23 +682,3 @@ def unit(self) -> str:
 
 
 # ------------------------------------------------------------------------------
-
-
-class RChromaticityArray(abstract.ReadFloatArray):
-    """
-    Class providing read write access to chromaticity of a control system.
-    """
-
-    def __init__(self, chromaticity_monitor):
-        self.__chromaticity_monitor = chromaticity_monitor
-
-    def _update_chromaticity_monitor(self, chromaticity_monitor):
-        """Use to attach the proper chromaticity_monitor and not the one used to create this instance"""
-        self.__chromaticity_monitor = chromaticity_monitor
-
-    def get(self) -> NDArray:
-        # Return horizontal and vertical chromaticity as a NumPy array
-        return self.__chromaticity_monitor._last_measured
-
-    def unit(self) -> str:
-        return "1"