Initialization

Author: Sulimankhail

.gitignore

@@ -0,0 +1,45 @@
+# PyCharm
+.idea/
+
+# Emacs auto save files
+*~
+
+# build files
+*.o
+
+# Library files
+*.so
+*.a
+
+# Byte-compiled / optimized / DLL files
+__pycache__/
+*.py[cod]
+*$py.class
+
+# C extensions
+*.so
+
+# Directories created by the build process
+bin/
+lib/
+include/
+dbd/
+db/
+configure/O.linux-x86_64/
+vaccelApp/Db/O.linux-x86_64/
+vaccelApp/Db/O.Common/
+vaccelApp/src/O.Common/
+vaccelApp/src/O.linux-x86_64/
+
+# Files created by magnet_info_from_db
+vaccelApp/src/*.inc
+*.inc
+
+# envPaths: built during first make after make clean
+iocBoot/iocvaccel/envPaths
+
+# python related stuff
+src/dist/
+*.pyc
+*.pyo
+__pycache__/

pyproject.toml

@@ -0,0 +1,29 @@
+[build-system]
+requires = [ "poetry-core>=1.0.0",]
+build-backend = "poetry.core.masonry.api"
+
+[tool.poetry]
+name = "lat2db"
+version = "0.0.1"
+description = "Storing accelerator lattices as json in mongo"
+authors = [  "Waheedullah Sulaiman Khail", "Pierre Schnizer", "Teresia Olsson"]
+license = "GPL-3.0"
+homepage = "https://github.com/hz-b/bact-bessyii-ophyd"
+documentation = "https://github.com/hz-b/bact-bessyii-ophyd"
+repository = "https://github.com/hz-b/bact-bessyii-ophyd.git"
+keywords = []
+
+[tool.poetry.dependencies]
+python = ">=3.7"
+fastapi="*"
+pymongo="*"
+pydantic="*"
+jsons="*"
+uuid="*"
+pytest="*"
+attrs="*"
+numpy="*"
+uvicorn="*"
+httpx="*"
+lark="*"
+accelerator-toolbox="*"

requirements.txt

@@ -0,0 +1,7 @@
+pymongo~=4.10.1
+setuptools~=71.1.0
+
+pydantic~=2.10.5
+softioc~=4.5.0
+numpy~=2.2.1
+p4p~=4.2.0

setup.cfg

@@ -0,0 +1,24 @@
+# -*- coding: utf-8 -*-
+[metadata]
+name = lat2db
+fullname = lattice to database
+version = 0.1.0
+url = https://gitlab.helmholtz-berlin.de/acc-tools/lat2db
+author = Waheedullah Sulaiman Khail, Pierre Schnizer
+author_email = [email protected], [email protected]
+description =
+long_description = file: README.md
+classifiers =
+	 Development Status :: 4 - Beta
+     Intended Audience :: Science/Research
+     License :: OSI Approved :: GNU General Public License (GPL)
+     Programming Language :: Python :: 3
+     Topic :: Scientific/Engineering :: Physics
+[build-system]
+requires = setup_tools
+
+
+[options]
+packages = find:
+include_package_data = True
+zip_safe = False

setup.py

@@ -0,0 +1,3 @@
+from setuptools import setup
+
+setup()

src/__init__.py

@@ -0,0 +1,5 @@
+from src.config.db_config import mongodb_url as _mongodb_
+
+mongodb_ = _mongodb_(__name__)
+
+__all__ = ["mongodb_"]

src/config/config.cfg

@@ -0,0 +1,14 @@
+[default]
+mongodb = localhost
+
+[localhost]
+server = localhost
+port = 27017
+database_name = bessyii
+
+[BESSYII]
+server = mongodb.bessy.de
+port = 28018
+;username = notadmin
+;password = admin123
+database_name = bessyii

src/config/db_config.py

@@ -0,0 +1,51 @@
+from configparser import ConfigParser
+from importlib.resources import files
+from urllib.parse import quote_plus
+import os
+
+from src.core.utils.logger import get_logger
+
+logger = get_logger()
+
+
+def config(mod_name):
+    parser = ConfigParser()
+    config_path = files(mod_name).joinpath("config/config.cfg")
+    read_files = parser.read(config_path)
+
+    if not read_files:
+        raise FileNotFoundError(f"Configuration file not found: {config_path}")
+
+    return parser
+
+def mongodb_url_from_config(mod_name):
+    cfg = config(mod_name)
+    mongodb = cfg["default"]["mongodb"]
+    dbsec = cfg[mongodb]
+    server = dbsec['server']
+    port = dbsec['port']
+    try:
+        username = dbsec['username']
+    except KeyError:
+        username = None   # Add username from config
+    try:
+        password = dbsec['password']
+    except KeyError:
+        password = None
+    db_name = dbsec['database_name']
+    if(username):
+        # Encode username and password for inclusion in URL
+        encoded_username = quote_plus(username)
+        encoded_password = quote_plus(password)
+        return f"mongodb://{encoded_username}:{encoded_password}@{server}:{port}/{db_name}"
+    else: return f"mongodb://{server}:{port}/{db_name}"
+
+def mongodb_url(mod_name):
+    default_url = mongodb_url_from_config(mod_name)
+    try:
+        return os.environ["MONGODB_URL"]
+    except KeyError:
+        txt = f'Environment variable MONGODB_URL is not defined, using default: {default_url}'
+        print(txt)
+        logger.warning(txt)
+        return default_url

src/core/accelerators/accelerator_impl.py

@@ -0,0 +1,86 @@
+from typing import Union
+
+from ..bl.delay_execution import DelayExecution
+from ..bl.event import Event
+from ..interfaces.accelerator_interface import AcceleratorInterface
+from ..interfaces.element_interface import ElementInterface
+
+
+class AcceleratorImpl(AcceleratorInterface):
+    def __init__(self, acc, proxy_factory, twiss_calculator, orbit_calculator, *, delay=0.1):
+        """
+        Initializes the accelerator implementation.
+        This manages the accelerator state, interactions, and calculation flow.
+        Registers event handlers and calculations.
+
+        Args:
+            acc: Accelerator model.
+            proxy_factory: Factory for creating element proxies.
+            twiss_calculator: Calculator for twiss parameters.
+            orbit_calculator: Calculator for orbit calculations.
+            delay (float): Delay for calculation executions.
+        """
+        super().__init__()
+        self.acc = acc
+        self.proxy_factory = proxy_factory
+        self.twiss_calculator = twiss_calculator
+        self.orbit_calculator = orbit_calculator
+
+        # Store calculation results
+        self.twiss = twiss_calculator.calculate()
+        self.orbit = orbit_calculator.calculate()
+
+        # Events for notification
+        self.on_new_orbit = Event()
+        self.on_new_twiss = Event()
+
+        async def cb_twiss():
+            """Internal method to calculate twiss parameters."""
+            self.twiss = self.twiss_calculator.calculate()
+            await self.on_new_twiss.trigger(self.twiss)
+
+        async def cb_orbit():
+            """Internal method to calculate orbit parameters."""
+            self.orbit = self.orbit_calculator.calculate()
+            await self.on_new_orbit.trigger(self.orbit)
+
+        self.twiss_calculation_delay = DelayExecution(callback=cb_twiss, delay=delay)
+        self.orbit_calculation_delay = DelayExecution(callback=cb_orbit, delay=delay)
+
+        self.on_changed_value = Event()
+
+        self.on_orbit_calculation_request = self.orbit_calculation_delay.on_calculation_requested
+        self.on_orbit_calculation = self.orbit_calculation_delay.on_calculation
+        self.on_twiss_calculation_request = self.twiss_calculation_delay.on_calculation_requested
+        self.on_twiss_calculation = self.twiss_calculation_delay.on_calculation
+
+    def set_delay(self, delay: Union[float, None]):
+        """How much to delay twiss and orbit calculation after last received command"""
+        self.twiss_calculation_delay.set_delay(delay)
+        self.orbit_calculation_delay.set_delay(delay)
+
+    async def get_element(self, element_id) -> ElementInterface:
+        """
+        Retrieves an element and subscribes to its changes.
+
+        Args:
+            element_id (str): The element identifier.
+
+        Returns:
+            ElementInterface: Proxy object for the element.
+        """
+        proxy = self.proxy_factory.get(element_id)
+        proxy.on_changed_value.subscribe(self.on_changed_value.trigger)
+
+        async def cb(unused):
+            await self.orbit_calculation_delay.request_execution()
+            await self.twiss_calculation_delay.request_execution()
+
+        proxy.on_update_finished.subscribe(cb)
+        return proxy
+
+    def calculate_twiss(self):
+        return self.twiss_calculator.calculate()
+
+    def calculate_orbit(self):
+        return self.orbit_calculator.calculate()

src/core/accelerators/accelerator_manager.py

@@ -0,0 +1,85 @@
+from src.core.utils.logger import get_logger
+from src.custom_epics.utils.bpm_mimicry import BPMMimicry
+from .proxy_factory import PyATProxyFactory
+from ..accelerators.accelerator_impl import AcceleratorImpl
+from ..calculations.pyat_calculator import PyAtTwissCalculator, PyAtOrbitCalculator
+from ...custom_epics.views.shared_view import get_view_instance
+
+logger = get_logger()
+
+
+class AcceleratorManager:
+    """
+    Manages the initialization and event subscription of the accelerator.
+    Orchestrates interactions, manages configurations,
+    and serves as a bridge between core accelerator logic and higher-level application needs.
+
+    Attributes:
+        prefix (str): The prefix for PVs (Process Variables).
+        accelerator (AcceleratorImpl): Instance of the accelerator implementation.
+        view (ResultView): Instance of the result view to handle user interface.
+        bpm_mimicry (BPMMimicry): Object to manage BPM data mimicking.
+    """
+
+    def __init__(self, prefix):
+        """
+        Initializes the AcceleratorManager with a given prefix.
+
+        Args:
+            prefix (str): The prefix for the PVs in the EPICS system.
+            todo: What about tango? do we have/need usage of prefix? we will findout
+        """
+        self.prefix = prefix
+        self.accelerator = None  # Will be initialized in the `initialize` method
+        self.view = get_view_instance()  # Shared view instance for displaying results
+        self.bpm_mimicry = None  # Placeholder for BPM mimicry instance
+
+    def initialize(self):
+        """
+        Initializes the accelerator and its required components.
+
+        This method sets up the accelerator, proxies, calculators, and BPMs.
+        It also subscribes to relevant events.
+
+        Raises:
+            Exception: If initialization fails.
+        """
+        try:
+            from lat2db.model.accelerator import Accelerator
+            acc_model = Accelerator()
+
+            # Initialize the accelerator with required components
+            self.accelerator = AcceleratorImpl(
+                acc_model.ring,
+                PyATProxyFactory(lattice_model=None, at_lattice=acc_model.ring),
+                PyAtTwissCalculator(acc_model),
+                PyAtOrbitCalculator(acc_model.ring)
+            )
+
+            # Extract BPM elements from the accelerator and create BPM mimicry
+            bpm_names = [elem.FamName for elem in self.accelerator.acc if elem.FamName.startswith("BPM")]
+            self.bpm_mimicry = BPMMimicry(prefix=self.prefix, bpm_names=bpm_names)
+            # Inject the BPM mimicry into the view instance
+            self.view.set_bpm_mimicry(self.bpm_mimicry)
+            self.setup_event_subscriptions()
+
+            logger.warning("AcceleratorManager initialized successfully.")
+        except Exception as e:
+            logger.error(f"Failed to initialize AcceleratorManager: {e}")
+            raise
+
+    def setup_event_subscriptions(self):
+        """
+        Subscribe to accelerator events and bind them to view update methods.
+
+        Ensures that the view receives updates whenever the accelerator produces new data.
+        """
+        if not self.accelerator:
+            logger.error("Accelerator must be initialized before subscribing to events.")
+
+        # Subscriptions
+        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)