20 feature implement tunemonitor

pyaml/common/element_holder.py

@@ -21,6 +21,7 @@ def __init__(self):
         self.__RFPLANT: dict = {}
         self.__RFTRANSMITTER: dict = {}
         self.__OTHERS: dict = {}
+        self.__DIAG: dict = {}
 
         # Array handle
         self.__MAGNET_ARRAYS: dict = {}
@@ -105,4 +106,18 @@ def get_rf_trasnmitter(self,name:str) -> RFTransmitter:
 
 
 
-
+    def get_bpm(self,name:str) -> Element:
+      if name not in self.__BPMS:
+         raise Exception(f"BPM {name} not defined")
+      return self.__BPMS[name]
+
+    def add_bpm(self,name:str,bpm:Element):
+        self.__BPMS[name] = bpm
+
+    def get_betatron_tune_monitor(self, name:str) -> Element:
+        if name not in self.__DIAG:
+            raise Exception(f"Diagnostic devices array does not contain {name}")
+        return self.__DIAG[name]
+
+    def add_betatron_tune_monitor(self, name:str, tune_monitor:Element):
+        self.__DIAG[name] = tune_monitor

pyaml/control/abstract_impl.py

@@ -7,7 +7,6 @@
 from ..rf.rf_plant import RFPlant
 from ..rf.rf_transmitter import RFTransmitter
 from ..common.abstract_aggregator import ScalarAggregator
-
 from numpy import double
 import numpy as np
 from numpy.typing import NDArray
@@ -350,3 +349,21 @@ def set_and_wait(self, value:float):
     def unit(self) -> str:
         return self.__rf._cfg.masterclock.unit()
 
+#------------------------------------------------------------------------------
+
+class RBetatronTuneArray(abstract.ReadFloatScalar):
+    """
+    Class providing read write access to betatron tune of a control system.
+    """
+
+    def __init__(self, tune_monitor):
+        self.__tune_monitor = tune_monitor
+
+    def get(self) -> NDArray:
+        # Return horizontal and vertical betatron tunes as a NumPy array
+        return np.array([self.__tune_monitor._cfg.tune_h.get(), 
+               self.__tune_monitor._cfg.tune_v.get()])
+
+    def unit(self) -> str:
+        return self.__tune_monitor._cfg.tune_v.unit()
+

pyaml/control/controlsystem.py

@@ -4,10 +4,12 @@
 from ..lattice.element import Element
 from ..control.abstract_impl import RWHardwareScalar,RWHardwareArray,RWStrengthScalar,RWStrengthArray
 from ..bpm.bpm import BPM
+from ..diagnostics.tune_monitor import BetatronTuneMonitor
 from ..control.abstract_impl import RWBpmTiltScalar,RWBpmOffsetArray, RBpmArray
 from ..control.abstract_impl import RWRFFrequencyScalar,RWRFVoltageScalar,RWRFPhaseScalar
 from ..control.abstract_impl import CSScalarAggregator,CSStrengthScalarAggregator
 from ..common.abstract_aggregator import ScalarAggregator
+from ..control.abstract_impl import RBetatronTuneArray
 from ..magnet.magnet import Magnet
 from ..magnet.cfm_magnet import CombinedFunctionMagnet
 from ..rf.rf_plant import RFPlant,RWTotalVoltage
@@ -134,3 +136,7 @@ def fill_device(self,elements:list[Element]):
              voltage = RWTotalVoltage(attachedTrans)
              ne = e.attach(frequency,voltage)
              self.add_rf_plant(ne.get_name(),ne)
+          elif isinstance(e,BetatronTuneMonitor):
+              betatron_tune = RBetatronTuneArray(e)
+              e = e.attach(betatron_tune)
+              self.add_betatron_tune_monitor(e.get_name(), e)

pyaml/diagnostics/tune_monitor.py

@@ -0,0 +1,48 @@
+
+from ..lattice.element import Element, ElementConfigModel
+from ..lattice.abstract_impl import RBetatronTuneArray
+from ..control.deviceaccess import DeviceAccess
+from typing import Self
+from pydantic import ConfigDict
+
+PYAMLCLASS = "BetatronTuneMonitor"
+
+class ConfigModel(ElementConfigModel):
+
+    model_config = ConfigDict(arbitrary_types_allowed=True,extra="forbid")
+
+    tune_h: DeviceAccess
+    """Horizontal betatron tune"""
+    tune_v: DeviceAccess
+    """Vertical betatron tune"""
+
+class BetatronTuneMonitor(Element):
+    """
+    Class providing access to a betatron tune monitor of a physical or simulated lattice.
+    The monitor provides horizontal and vertical betatron tune measurements.
+    """
+
+    def __init__(self, cfg: ConfigModel):
+        """
+        Construct a BetatronTuneMonitor.
+        Parameters
+        ----------
+        cfg : ConfigModel
+            Configuration for the BetatronTuneMonitor, including 
+            device access for horizontal and vertical tunes.
+        """
+
+        super().__init__(cfg.name)
+        self._cfg = cfg
+        self.__tune = None
+
+    @property
+    def tune(self) -> RBetatronTuneArray:
+        return self.__tune
+
+    def attach(self, betatron_tune: RBetatronTuneArray) -> Self:
+
+        obj = self.__class__(self._cfg)
+        obj.__tune = betatron_tune
+        return obj
+

pyaml/lattice/abstract_impl.py

@@ -378,3 +378,19 @@ def set_and_wait(self, value:float):
     def unit(self) -> str:
         return self.__rf._cfg.masterclock.unit()
 
+#------------------------------------------------------------------------------
+
+class RBetatronTuneArray(abstract.ReadFloatScalar):
+    """
+    Class providing read-only access to the betatron tune of a ring.
+    """
+
+    def __init__(self, ring: at.Lattice):
+        self.__ring = ring
+
+    def get(self) -> float:
+        return self.__ring.get_tune()[:2]
+
+    def unit(self) -> str:
+        return '1'
+

pyaml/lattice/simulator.py

@@ -5,6 +5,7 @@
 from pathlib import Path
 from ..magnet.magnet import Magnet
 from pyaml.bpm.bpm import BPM
+from pyaml.diagnostics.tune_monitor import BetatronTuneMonitor
 from ..magnet.cfm_magnet import CombinedFunctionMagnet
 from ..rf.rf_plant import RFPlant,RWTotalVoltage
 from ..rf.rf_transmitter import RFTransmitter
@@ -13,6 +14,7 @@
 from ..lattice.abstract_impl import RWRFFrequencyScalar,RWRFVoltageScalar,RWRFPhaseScalar
 from ..common.element_holder import ElementHolder
 from ..common.abstract_aggregator import ScalarAggregator
+from ..lattice.abstract_impl import RBetatronTuneArray
 from ..lattice.abstract_impl import RWBpmTiltScalar,RWBpmOffsetArray, RBpmArray
 from ..lattice.abstract_impl import BPMHScalarAggregator,BPMScalarAggregator,BPMVScalarAggregator
 from ..common.exception import PyAMLException
@@ -140,6 +142,10 @@ def fill_device(self,elements:list[Element]):
              voltage = RWTotalVoltage(attachedTrans)
              ne = e.attach(frequency,voltage)
              self.add_rf_plant(ne.get_name(),ne)
+          elif isinstance(e, BetatronTuneMonitor):
+             betatron_tune = RBetatronTuneArray(self.ring)
+             e = e.attach(betatron_tune)
+             self.add_betatron_tune_monitor(e.get_name(), e)
 
 
     def get_at_elems(self,element:Element) -> list[at.Element]:

tests/config/EBSTune.yaml

@@ -1877,3 +1877,14 @@ instruments:
           type: tango.pyaml.attribute
           attribute: srmag/vps-qd2/c03-a/current
           unit: A
+    - type: pyaml.diagnostics.tune_monitor
+      name: BETATRON_TUNE
+      tune_h:
+        type: tango.pyaml.attribute_read_only
+        attribute: srdiag/tune/tune_h
+        unit: mm
+      tune_v:
+        type: tango.pyaml.attribute_read_only
+        attribute: srdiag/tune/tune_v
+        unit: mm
+

tests/config/tune_monitor.yaml

@@ -0,0 +1,25 @@
+type: pyaml.pyaml
+instruments:
+  - type: pyaml.instrument
+    name: sr
+    energy: 6e9
+    simulators:
+      - type: pyaml.lattice.simulator
+        lattice: sr/lattices/ebs.mat
+        name: design
+    controls:
+      - type: tango.pyaml.controlsystem
+        tango_host: ebs-simu-3:10000
+        name: live
+    data_folder: /data/store
+    devices:
+    - type: pyaml.diagnostics.tune_monitor
+      name: BETATRON_TUNE
+      tune_h:
+        type: tango.pyaml.attribute_read_only
+        attribute: srdiag/tune/tune_h
+        unit: mm
+      tune_v:
+        type: tango.pyaml.attribute_read_only
+        attribute: srdiag/tune/tune_v
+        unit: mm

tests/test_tune.py

@@ -2,7 +2,6 @@
 from pyaml.instrument import Instrument
 from pyaml.configuration.factory import Factory
 import numpy as np
-import at
 import pytest
 
 @pytest.mark.parametrize("install_test_package", [{
@@ -17,16 +16,16 @@ def test_tune(install_test_package):
 
     quadForTuneDesign = sr.design.get_magnets("QForTune")
     quadForTuneLive = sr.live.get_magnets("QForTune")
-
+    tune_monitor = sr.design.get_betatron_tune_monitor("BETATRON_TUNE")
     # Build tune response matrix
-    tune = sr.design.get_lattice().get_tune()
+    tune = tune_monitor.tune.get()
     print(tune)
     tunemat = np.zeros((len(quadForTuneDesign),2))
 
     for idx,m in enumerate(quadForTuneDesign):
         str = m.strength.get()
         m.strength.set(str+1e-4)
-        dq = sr.design.get_lattice().get_tune() - tune
+        dq = tune_monitor.tune.get() - tune
         tunemat[idx] = dq*1e4
         m.strength.set(str)
 
@@ -37,8 +36,9 @@ def test_tune(install_test_package):
     strs = quadForTuneDesign.strengths.get()
     strs += np.matmul(correctionmat,[0.1,0.05]) # Ask for correction [dqx,dqy]
     quadForTuneDesign.strengths.set(strs)
-    newTune = sr.design.get_lattice().get_tune()
+    newTune = tune_monitor.tune.get()
     diffTune = newTune-tune
+
     print(diffTune)
     assert( np.abs(diffTune[0]-0.1) < 1e-3 )
     assert( np.abs(diffTune[1]-0.05) < 1e-3 )

tests/test_tune_hardware.py

@@ -2,7 +2,6 @@
 from pyaml.instrument import Instrument
 from pyaml.configuration.factory import Factory
 import numpy as np
-import at
 import pytest
 
 @pytest.mark.parametrize("install_test_package", [{
@@ -18,15 +17,17 @@ def test_tune(install_test_package):
     quadForTuneDesign = sr.design.get_magnets("QForTune")
 
     # Build tune response matrix (hardware units)
-    tune = sr.design.get_lattice().get_tune()
+
+    tune_monitor = sr.design.get_betatron_tune_monitor("BETATRON_TUNE")
+    tune = tune_monitor.tune.get()
     print(tune)
     tunemat = np.zeros((len(quadForTuneDesign),2))
 
     idx = 0
     for m in quadForTuneDesign:
         current = m.hardware.get()
         m.hardware.set(current+1e-6)
-        dq = sr.design.get_lattice().get_tune() - tune
+        dq = tune_monitor.tune.get() - tune
         tunemat[idx] = dq*1e6
         m.hardware.set(current)
         idx += 1
@@ -38,7 +39,7 @@ def test_tune(install_test_package):
     currents = quadForTuneDesign.hardwares.get()
     currents += np.matmul(correctionmat,[0.1,0.05]) # Ask for correction [dqx,dqy]
     quadForTuneDesign.hardwares.set(currents)
-    newTune = sr.design.get_lattice().get_tune()
+    newTune = tune_monitor.tune.get()
     units = quadForTuneDesign.hardwares.unit()
     diffTune = newTune - tune
     assert( np.abs(diffTune[0]-0.1) < 1e-3 )

tests/test_tune_monitor.py

@@ -0,0 +1,35 @@
+
+from pyaml.pyaml import PyAML, pyaml
+from pyaml.instrument import Instrument
+from pyaml.configuration.factory import Factory
+import pytest
+
+@pytest.mark.parametrize("install_test_package", [{
+    "name": "tango-pyaml",
+    "path": "tests/dummy_cs/tango-pyaml"
+}], indirect=True)
+def test_simulator_tune_monitor(install_test_package):
+
+    ml:PyAML = pyaml("tests/config/tune_monitor.yaml")
+    sr:Instrument = ml.get('sr')
+    sr.design.get_lattice().disable_6d()
+    tune_monitor = sr.design.get_betatron_tune_monitor("BETATRON_TUNE")
+    assert tune_monitor.tune.get()[0] == sr.design.get_lattice().get_tune()[0]
+    assert tune_monitor.tune.get()[1] == sr.design.get_lattice().get_tune()[1]
+
+    Factory.clear()
+
+@pytest.mark.parametrize("install_test_package", [{
+    "name": "tango-pyaml",
+    "path": "tests/dummy_cs/tango-pyaml"
+}], indirect=True)
+def test_controlsystem_tune_monitor(install_test_package):
+
+    ml:PyAML = pyaml("tests/config/tune_monitor.yaml")
+    sr:Instrument = ml.get('sr')
+    tune_monitor = sr.live.get_betatron_tune_monitor("BETATRON_TUNE")
+    assert tune_monitor.tune.get()[0] == 0.0
+    assert tune_monitor.tune.get()[1] == 0.0
+
+    Factory.clear()
+