[TASK/Snapshot] read data from database

src/dt4acc/core/command.py

@@ -1,11 +1,7 @@
 from bact_twin_architecture.data_model.command import Command, BehaviourOnError
 from bact_twin_architecture.interfaces.command_rewritter import CommandRewriterBase
-from bact_twin_bessyii_impl.bl.io.pytac_repositories import PyTACRepository
-from bact_twin_bessyii_impl.bl.command_rewritter import CommandRewriter
-from bact_twin_bessyii_impl.bl.translation_service import TranslationService
 
 from .accelerators.accelerator_manager import AcceleratorManager
-from .accelerators.pyat_accelerator import setup_accelerator
 from .update_context_manager import UpdateContext
 
 

src/dt4acc/core/model/elementmodel.py

@@ -6,17 +6,19 @@
 class MagnetElementSetup:
     type: str
     name: str
-    hw2phys: float
-    phys2hw: float
-    energy: float
-    magnetic_strength: float
-    electron_rest_mass: float
-    speed_of_light: float
-    brho: float
-    edf: float
+    #: power converter it is connected to
     pc: str
+    magnetic_strength: float
+    hw2phys: Optional[float] = None
     k: Optional[float] = None
 
+
+    def order(self) -> int:
+        """return order in European Convention
+        """
+        magnet_order = dict(bend=1, quadrupole=2, sextupole=3, octupole=4, steerer=1)
+        return magnet_order[self.type.lower()]
+
 @dataclass
 class PowerConverterElementSetup:
     type: str

src/dt4acc/custom_epics/ioc/handlers.py

@@ -1,6 +1,5 @@
 from bact_twin_bessyii_impl.bl.command_rewritter import CommandRewriter
-from bact_twin_bessyii_impl.bl.io.pytac_repositories import PyTACRepository
-from bact_twin_bessyii_impl.bl.translation_service import TranslationService
+from dt4acc.custom_epics.ioc.liasion_translation_manager import build_managers
 from p4p.client.asyncio import Context
 
 from ...core.accelerators.pyat_accelerator import setup_accelerator
@@ -12,12 +11,14 @@
 ctx = Context("pva")  # Create a context for EPICS PVA (PV Access)
 
 #: todo replace soon by database service
-repo = PyTACRepository()
+
+lm, tm = build_managers()
 
 # todo: should this be part of the controller
 update_manager = UpdateManager(
     command_rewritter=CommandRewriter(
-        TranslationService(conversion_info=repo.state_conversion_repo)
+        liasion_manager=lm,
+        translation_service=tm
     ),
     acc_mgr=setup_accelerator()
 )

src/dt4acc/custom_epics/ioc/liasion_translation_manager.py

@@ -0,0 +1,173 @@
+from collections.abc import Mapping
+from typing import Dict, Sequence
+
+from bact_twin_architecture.data_model.identifiers import LatticeElementPropertyID, DevicePropertyID, ConversionID
+from bact_twin_architecture.interfaces.liaison_manager import LiaisonManagerBase
+from bact_twin_architecture.interfaces.state_conversion import StateConversion
+from bact_twin_architecture.interfaces.translator_service import TranslatorServiceBase
+from bact_twin_architecture.utils.unit_conversion import LinearUnitConversion
+from bact_twin_bessyii_impl.bl.bessyii_nomen_clature import name_matches_horizontal_steerer_name, \
+    name_matches_vertical_steerer_name, name_matches_steerer_name
+
+from ..data.querries import get_magnets
+from ..data.constants import DEFAULTS
+from ...core.model.elementmodel import MagnetElementSetup
+
+
+class LiaisonManager(LiaisonManagerBase):
+    def __init__(
+        self,
+        forward_lut: Mapping[LatticeElementPropertyID, Sequence[DevicePropertyID]],
+        inverse_lut: Mapping[DevicePropertyID, Sequence[LatticeElementPropertyID]],
+    ):
+        self.forward_lut = forward_lut
+        self.inverse_lut = inverse_lut
+
+    def forward(self, id_: LatticeElementPropertyID) -> Sequence[DevicePropertyID]:
+        return self.forward_lut[id_]
+
+    def inverse(self, id_: DevicePropertyID) -> Sequence[LatticeElementPropertyID]:
+        return self.inverse_lut[id_]
+
+
+class TranslatorService(TranslatorServiceBase):
+    def __init__(self, lut: Mapping[ConversionID, StateConversion]):
+        self.lut = lut
+
+    def get(self, id_: ConversionID) -> StateConversion:
+        return self.lut[id_]
+
+
+def remove_id(d: Dict) -> Dict:
+    nd = d.copy()
+    del d
+    del nd["_id"]
+    return nd
+
+
+def magnet_infos_from_db() -> Sequence[MagnetElementSetup]:
+    return [MagnetElementSetup(**remove_id(info)) for info in get_magnets().to_list()]
+
+
+def build_managers() -> (LiaisonManagerBase, TranslatorServiceBase):
+    """A first poor mans implementation of the database
+
+    Todo:
+        Which info is already in database and better obtained from database?
+    """
+    infos = magnet_infos_from_db()
+
+    magnet_types = set([info.type for info in infos])
+    # Make sure that names are unique ... everything down the list depends on it
+    magnet_names = set([info.name for info in infos])
+    if len(list(magnet_names)) != len(infos):
+        raise AssertionError("Magnet names seem not to be unique, but is assumption of all further processing")
+
+    power_converter_names = set([info.pc for info in infos])
+    power_converter_feeds = {
+        pc_name: [info.name for info in infos if info.pc == pc_name] for pc_name in power_converter_names
+    }
+
+    magnet_lut = {info.name: info for info in infos}
+    # todo: check if property must be different for the different magnets ...
+
+
+
+    # first for steerers : for AT these are angles applied to the host magnet
+    # I use that I know one pc goes to one steerer
+    inverse_lut = {
+        DevicePropertyID(device_name=info.pc, property="set_current"):
+            (LatticeElementPropertyID(element_name=info.name[1:], property="x_kick"),)
+        for info in infos if name_matches_horizontal_steerer_name(info.name)
+
+    }
+    inverse_lut.update({
+        DevicePropertyID(device_name=info.pc, property="set_current"):
+            (LatticeElementPropertyID(element_name=info.name[1:], property="y_kick"),)
+        for info in infos if name_matches_vertical_steerer_name(info.name)
+
+    })
+
+    # Test that steerer power converters only feed one before going to the next step
+    for key in inverse_lut:
+        corr_pc = key.device_name
+        magnet_names = power_converter_feeds[corr_pc]
+        if len(magnet_names) != 1:
+            raise AssertionError(f"Found {magnet_names} magnets on assumed corrector power supply {corr_pc}")
+
+    # quadrupoles and sextupoles
+    steerer_pc_names = [key.device_name for key in inverse_lut]
+    inverse_lut.update({
+        DevicePropertyID(device_name=pc_name, property="set_current"):
+            tuple([LatticeElementPropertyID(element_name=magnet_name, property="K") for magnet_name in magnet_names])
+        for pc_name, magnet_names in power_converter_feeds.items() if pc_name not in steerer_pc_names
+    })
+
+    # Add lut for quadrupoles and sextupole axes
+    quad_updates = dict()
+    for axis_name  in "x", "y":
+        quad_updates.update({
+            DevicePropertyID(device_name=info.name, property=axis_name):
+                (LatticeElementPropertyID(element_name=info.name, property=axis_name),)
+            for info in infos if info.type in ["Sextupole", "Quadrupole"]
+        })
+    inverse_lut.update(quad_updates)
+
+
+    forward_lut = None
+    lm = LiaisonManager(forward_lut=forward_lut, inverse_lut=inverse_lut)
+
+    def element_method(element_name):
+        device_name = "HS4M2D1R"
+        if element_name == device_name:
+            pass
+        if name_matches_horizontal_steerer_name(element_name):
+            return "x_kick"
+        elif name_matches_vertical_steerer_name(element_name):
+            return "y_kick"
+        else:
+            return "K"
+
+    def extract_host_element_name(element_name: str) -> str:
+        if name_matches_steerer_name(element_name):
+            return element_name[1:]
+        return element_name
+
+    def construct_linear_conversion(slope: float) -> LinearUnitConversion:
+        if slope is None:
+            raise AssertionError("Refusing creating linear unit conversion without slope")
+        return LinearUnitConversion(slope=slope, intercept=0.0)
+
+    # start to build it for the magnets ... power converter feed
+    translator_lut = {
+        # todo: replace it with an energy independent version
+        ConversionID(
+            LatticeElementPropertyID(
+                element_name=extract_host_element_name(info.name),
+                property=element_method(info.name)
+            ),
+            DevicePropertyID(device_name=info.pc, property="set_current")
+        ):
+            # todo: check for the correct conversion
+            construct_linear_conversion(slope=info.magnetic_strength)
+        for info in infos
+    }
+
+    axis_updates = dict()
+    for axis_name  in "x", "y":
+        axis_updates.update({
+            ConversionID(
+                lattice_property_id=LatticeElementPropertyID(element_name=info.name, property=axis_name),
+                device_property_id=DevicePropertyID(device_name=info.name, property=axis_name)
+            )
+            : LinearUnitConversion(slope=1.0, intercept=0.0)
+            for info in infos if info.type in ["Sextupole", "Quadrupole"]
+        })
+    translator_lut.update(axis_updates)
+    # start to build it for quadrupoles and sextupoles axes
+
+    tm = TranslatorService(translator_lut)
+    return lm, tm
+
+if __name__ == "__main__":
+    build_managers()

src/dt4acc/custom_epics/ioc/server.py

@@ -13,17 +13,16 @@
     initialize_cavity_pvs
 )
 
-# Create an asyncio dispatcher to handle asynchronous PV updates
-dispatcher = asyncio_dispatcher.AsyncioDispatcher()
-# Retrieve the device name prefix from the environment, defaulting to "Anonym" if not set
-prefix = os.environ.get("DT4ACC_PREFIX", "Anonym")
-builder.SetDeviceName(prefix)
-
 
 def main():
     """
     Main function to initialize all the process variables (PVs) and start the IOC server.
     """
+    # Create an asyncio dispatcher to handle asynchronous PV updates
+    dispatcher = asyncio_dispatcher.AsyncioDispatcher()
+    # Retrieve the device name prefix from the environment, defaulting to "Anonym" if not set
+    prefix = os.environ.get("DT4ACC_PREFIX", "Anonym")
+    builder.SetDeviceName(prefix)
     # Initialize PVs for various accelerator components
     initialize_power_converter_pvs(builder, prefix)  # Initialize power converters and linked magnets
     initialize_orbit_pvs(builder)  # Initialize orbit-related PVs

tests/test_command_update.py

@@ -1,21 +1,26 @@
+"""
+Todo:
+    Tests show that a readback is required without layer violations
+
+"""
 import asyncio
 import pytest
 
+
 from src.dt4acc.core.accelerators.element_proxies import estimate_shift
-from src.dt4acc.core.command import update_manager, acc
+from src.dt4acc.custom_epics.ioc.handlers import update_manager
 
 
 # pytest_plugins = ('pytest_asyncio',)
 
-
 @pytest.mark.asyncio(scope="session")
 async def test_update_quadrupole_x():
     device_id = "Q5M2T5R"
     for val in [-1e-4, 2e-4, 0]:
         await update_manager.update(device_id=device_id, property_name="x", value=val)
         # check that the result arrived
         # todo: reading  interaction should be improved
-        proxy = await acc.acc_mgr.get_element(device_id)
+        proxy = await update_manager.acc_mgr.accelerator.get_element(device_id)
         element, = proxy._obj
         shift =  estimate_shift(element)
         assert shift[0] == pytest.approx(-val, abs=1e-6)
@@ -29,7 +34,7 @@ async def test_update_quadrupole_y():
         await update_manager.update(device_id=device_id, property_name="y", value=val)
         # check that the result arrived
         # todo: reading  interaction should be improved
-        proxy = await acc.acc_mgr.get_element(device_id)
+        proxy = await update_manager.acc_mgr.accelerator.get_element(device_id)
         element, = proxy._obj
         shift =  estimate_shift(element)
         assert shift[0] == pytest.approx(0.0, abs=1e-6)
@@ -38,7 +43,6 @@ async def test_update_quadrupole_y():
 
 @pytest.mark.asyncio(scope="session")
 async def test_update_steerer_pc_current():
-    acc
     device_id = "HS4P1D1R"
     lattice_id = device_id[1:].replace("P", "M")
 
@@ -48,7 +52,7 @@ async def test_update_steerer_pc_current():
         # check that the result arrived
         # todo: reading  interaction should be improved
         lattice_id = device_id[1:].replace("P", "M")
-        proxy = await acc.acc_mgr.get_element(lattice_id)
+        proxy = await update_manager.acc_mgr.accelerator.get_element(lattice_id)
         element, = proxy._obj
         x_kick, y_kick = element.KickAngle
 
@@ -63,5 +67,5 @@ async def test_update_steerer_pc_current():
 async def test_update_master_clock():
     device_id = "MCLKHX251C"
     frequency = 500e6
-    with pytest.raises(AssertionError):
+    with pytest.raises(KeyError):
         await update_manager.update(device_id=device_id, property_name="reference_frequency", value=frequency)

tests/test_liasion_manager.py

@@ -0,0 +1,28 @@
+from bact_twin_architecture.data_model.identifiers import ConversionID, DevicePropertyID, LatticeElementPropertyID
+
+from src.dt4acc.custom_epics.ioc.liasion_translation_manager import build_managers
+
+
+def test_liasion_service():
+    lm, _ = build_managers()
+
+    device_name = "HS4P2D1R"
+    r, = lm.inverse(DevicePropertyID(device_name=device_name, property="set_current"))
+    assert r.property == "x_kick"
+    assert r.element_name == device_name.replace("P", "M")
+
+    device_name = "VS2P2D1R"
+    r, = lm.inverse(DevicePropertyID(device_name=device_name, property="set_current"))
+    assert r.property == "y_kick"
+    assert r.element_name == device_name.replace("P", "M")
+
+
+    device_name = "S4PD1R"
+    # warning: no warrenty which one is first
+    down_stream, up_stream = lm.inverse(DevicePropertyID(device_name=device_name, property="set_current"))
+    r = up_stream
+    assert r.property == "K"
+    assert r.element_name == device_name[:2] + "M1" + device_name[3:]
+    r = down_stream
+    assert r.property == "K"
+    assert r.element_name == device_name[:2] + "M2" + device_name[3:]

tests/test_translation_service.py

@@ -0,0 +1,28 @@
+from bact_twin_architecture.data_model.identifiers import ConversionID, LatticeElementPropertyID, DevicePropertyID
+
+from src.dt4acc.custom_epics.ioc.liasion_translation_manager import build_managers
+
+
+def test_translation_service():
+    _, tm = build_managers()
+    r = tm.get(
+        ConversionID(
+            lattice_property_id=LatticeElementPropertyID(element_name="Q1M1D1R", property="K"),
+            device_property_id=DevicePropertyID(device_name="Q1PDR", property="set_current")
+        )
+    )
+
+def test_translation_service_as_issued_by_command_test():
+    _, tm = build_managers()
+
+    # make selection to simplify searching for the cause
+    d = {
+        key: item for key, item in tm.lut.items() if key.lattice_property_id.element_name in ["S4M1D1R", "HS4M1D1R"]
+     }
+    r = tm.get(
+        ConversionID(
+            lattice_property_id=LatticeElementPropertyID(element_name='S4M1D1R', property='x_kick'),
+            device_property_id=DevicePropertyID(device_name='HS4P1D1R', property='set_current')
+        )
+    )
+