Merge pull request #6 from hz-b/dev/feature/periodic-publisher

[TASK] Split view into data that produced by calculation and data republished as it is also done by the machine

Author: Sulimankhail

src/dt4acc/core/accelerators/accelerator_manager.py

@@ -81,5 +81,4 @@ def setup_event_subscriptions(self):
         self.accelerator.on_new_twiss.subscribe(self.view.push_twiss)
         self.accelerator.on_new_orbit.subscribe(self.view.push_orbit)
 
-        self.accelerator.on_new_orbit.subscribe(self.view.push_bpms)
         self.accelerator.on_changed_value.subscribe(self.view.push_value)

src/dt4acc/core/bl/delay_execution.py

@@ -39,7 +39,7 @@ async def request_execution(self):
             self.pending_task.cancel()
 
         # Schedule a new execution task after the delay
-        self.pending_task = asyncio.create_task(self._delayed_execution())
+        self.pending_task = asyncio.create_task(self._delayed_execution(), name="delayed-execution")
 
     async def _delayed_execution(self):
         """

src/dt4acc/core/interfaces/view_interface.py

@@ -0,0 +1,9 @@
+from abc import ABCMeta, abstractmethod
+
+
+class ViewInterface(metaclass=ABCMeta):
+    @abstractmethod
+    async def push(self, data):
+        """
+        """
+        raise NotImplementedError("use base class instead")

src/dt4acc/core/model/twiss.py

@@ -40,4 +40,11 @@ def get_plane(self, plane: Planes):
         elif plane == Planes.y:
             return self.y
         else:
-            raise AssertionError("How could I end up here")
+            raise AssertionError("How could I end up here")
+
+@dataclass
+class TuneData:
+    """extracted tune data for the tune device
+    """
+    x: float
+    y: float

src/dt4acc/core/utils/periodic_publisher.py

@@ -0,0 +1,41 @@
+import asyncio
+import itertools
+
+from .logger import get_logger
+from ..interfaces.view_interface import ViewInterface
+
+#: use a "singleton", even if the name is not unique it should still work
+counter = itertools.count()
+logger = get_logger()
+
+class PeriodicPublisher:
+    """
+
+    Todo:
+        find a better name for it?
+        Its rather a periodic publisher proxy
+
+        It is expected to be triggered from outside
+    """
+    def __init__(self, view: ViewInterface, name: str):
+        """
+
+        Todo:
+            view should cohere to a protocol or interface
+        """
+        self.view = view
+        self.data = None
+        self.name = name
+
+    def set_data(self, data):
+        self.data = data
+
+    async def publish(self):
+        """
+        Expect that publish is called in parallel, thus just return stat object
+        """
+        #  name=f"publish-data-{self.name}-{next(counter)}"
+        if self.data is None:
+            logger.warning(f"{self.__class__.__name__}(name={self.name}), self.data is None, thus not publishing data")
+            return
+        return await self.view.push(self.data)

src/dt4acc/core/views/shared_view.py

@@ -21,9 +21,11 @@ def get_view_instance():
     server_type = os.environ.get("server", "epics").lower()
 
     if server_type == "tango":
-        from ...custom_tango.views.calculation_result_view import ResultView
+        # todo: needs to be implemented, sorry for breaking
+        #       tango implementation Pierre
+        from ...custom_tango.views.result_view import ResultView
     elif server_type == "epics":
-        from ...custom_epics.views.calculation_result_view import ResultView
+        from ...custom_epics.views.result_view import ResultView
     else:
         raise ValueError(f"Unsupported server type: {server_type}")
 

src/dt4acc/custom_epics/ioc/handlers.py

@@ -1,15 +1,15 @@
 from bact_twin_architecture.bl.command_rewriter import CommandRewriter
 # from dt4acc.custom_epics.ioc.liasion_translation_manager import build_managers
-from p4p.client.asyncio import Context
 
 from .liasion_translation_manager import build_managers
+from ..utils.context_proxy import ContextProxy
 from ...core.accelerators.pyat_accelerator import setup_accelerator
 from ...core.command import UpdateManager
 from ...core.utils.logger import get_logger
 
 logger = get_logger()
 
-ctx = Context("pva")  # Create a context for EPICS PVA (PV Access)
+ctx = ContextProxy("pva")  # Create a context for EPICS PVA (PV Access)
 
 #: todo replace soon by database service
 

src/dt4acc/custom_epics/ioc/pv_setup.py

@@ -3,7 +3,6 @@
     LatticeElementPropertyID,
     DevicePropertyID,
 )
-from p4p.asLib.yacc import start
 
 from .handlers import handle_device_update, update_manager
 from ..data.constants import config, special_pvs, cavity_names
@@ -116,7 +115,7 @@ def add_pc_pvs(builder, pc_name, prefix):
     rdbk = builder.aOut(f"{pc_name}:rdbk", initial_value=start_val, PREC=2)
 
     async def handle_pc_update(device_id: str, property_id: str, value: float):
-        logger.debug("%s:%s updating setpoint val=%s", device_id, property_id, value)
+        logger.warning("%s:%s updating setpoint val=%s", device_id, property_id, value)
         r = await handle_device_update(
             device_id=device_id, property_id=property_id, value=value
         )
@@ -148,6 +147,12 @@ def initialize_orbit_pvs(builder):
     builder.aOut(f"beam:orbit:found", initial_value=0)
 
 
+def initialize_tune_pvs(builder):
+    for axis in ["x", "y"]:
+        builder.aOut(f"TUNECC:{axis}", initial_value=0.0, PREC=9)
+    builder.longOut(f"TUNECC:count", initial_value=0)
+
+
 def initialize_twiss_pvs(builder):
     """
     Initializes PVs for Twiss parameters, which describe beam optics.

src/dt4acc/custom_epics/ioc/server.py

@@ -4,6 +4,7 @@
 
 from softioc import softioc, builder, asyncio_dispatcher
 
+from ...core.utils.logger import get_logger
 from .tasks import monitor_heartbeat
 from .pv_setup import (
     initialize_power_converter_pvs,
@@ -13,9 +14,12 @@
     initialize_bpm_pvs,
     initialize_orbit_object_pvs,
     initialize_twiss_pvs,
+    initialize_tune_pvs,
     initialize_other_pvs,
 )
 
+logger = get_logger()
+
 # Create an asyncio dispatcher to handle asynchronous PV updates
 dispatcher = asyncio_dispatcher.AsyncioDispatcher()
 
@@ -39,14 +43,16 @@ def startup():
     initialize_bpm_pvs(builder)  # Initialize Beam Position Monitor PVs
     initialize_orbit_object_pvs(builder)  # Initialize PV's of the new orbit object ... collection of bpms
     initialize_twiss_pvs(builder)  # Initialize Twiss parameter PVs
+    initialize_tune_pvs(builder)
+    logger.warning("All pvs set up")
 
     # Load the database of PVs defined above into the SoftIOC server
     builder.LoadDatabase()
     # Start the SoftIOC server to handle PV interactions
     softioc.iocInit(dispatcher)
 
     # Start monitoring the heartbeat to ensure the server is running correctly
-    asyncio.create_task(monitor_heartbeat())
+    asyncio.create_task(monitor_heartbeat(), name="server-heartbeat-loop")
 
 
 def main():

src/dt4acc/custom_epics/ioc/tasks.py

@@ -1,12 +1,12 @@
 import asyncio
 import getpass
+from datetime import datetime
 
-from p4p.client.asyncio import Context
-
+from ..utils.context_proxy import ContextProxy
 from ...core.utils.logger import get_logger
 from ...core.views.shared_view import get_view_instance
 
-ctx = Context("pva")
+ctx = ContextProxy("pva")
 logger = get_logger()
 view = get_view_instance()
 heartbeat_task = None  # Global variable to store the heartbeat task reference
@@ -25,7 +25,7 @@ async def monitor_heartbeat():
             await ctx.put(f'{getpass.getuser()}:VS3P2T5R:set', 1e-3)
             await ctx.put(f'{getpass.getuser()}:VS3P2T5R:set', 0e-3)
             logger.warning("Heartbeat loop initial start or it was terminated unexpectedly. Restarting...")
-            heartbeat_task = asyncio.create_task(heartbeat_loop())
+            heartbeat_task = asyncio.create_task(heartbeat_loop(), name="heart-beat-loop")
         await asyncio.sleep(1)
 
 
@@ -36,12 +36,14 @@ async def heartbeat_loop():
     Runs indefinitely and logs any errors encountered.
     """
     while True:
+        logger.info(f"Heart beat loop executing at  {datetime.now()}")
         try:
             await asyncio.sleep(1)  # Run every second
-            # logger.warning("Heartbeat loop was running ...")
+            logger.debug(f"Heart beat loop executed at {datetime.now()}")
             await view.heart_beat()
         except asyncio.CancelledError:
             logger.warning("Heartbeat loop was cancelled.")
             break  # Exit cleanly on cancellation
         except Exception as exc:
             logger.error(f"Heartbeat encountered an error: {exc}")
+            raise exc