Config layer suggestion with registry

examples/mls_example/mls_config.py

@@ -0,0 +1,57 @@
+"""Example to create configuration for MLS using config classes."""
+
+from accml.core.config import AcceleratorConfig, EPICSConfig, SimulatorConfig, MagnetConfig, EPICSType
+from accml.core.registry.utils import WildcardDict
+
+#%% Facility and machine name
+facility = 'MLS'
+machine = 'storage_ring'
+
+#%% Control system modes
+live = EPICSConfig(access_type=EPICSType.CA)
+twin = EPICSConfig(access_type=EPICSType.CA,pv_prefix='twin')
+
+#%% Simulator modes
+design = SimulatorConfig(type='pyat',model='design_lattice.json')
+error = SimulatorConfig(type='pyat',model='error_lattice.json')
+measured = SimulatorConfig(type='pyat',model='measured_lattice.json')
+
+#%% Devices
+
+# Quadrupoles
+quads = []
+families = ['Q1','Q2','Q3']
+cells = ['1','2']
+sections = ['K1','L2','K3','L4']
+
+for quad in families:
+    for cell in cells:
+        for section in sections:
+            quads.append(MagnetConfig(name=f'{quad}M{cell}{section}RP',type='quadrupole',power_supply=f'{quad}P{cell}{section}RP'))
+
+# Generate a dictionary of the quadrupole configurations
+quads_by_name = {q.name: q for q in quads}
+
+#%% Families
+
+Q1 = [device.name for device in WildcardDict(quads_by_name)['Q1*']]
+Q2 = [device.name for device in WildcardDict(quads_by_name)['Q2*']]
+Q3 = [device.name for device in WildcardDict(quads_by_name)['Q3*']]
+TuneCorrectors = Q1 + Q3
+
+families = {'Q1': Q1,
+            'Q2': Q2,
+            'Q3': Q3,
+            'TuneCorrectors': TuneCorrectors}
+
+#%% Accelerator
+
+config = AcceleratorConfig(facility=facility, machine=machine,
+                           controls = {'live': live, 'twin': twin},
+                           simulators = {'design': design,
+                                         'error': error,
+                                         'measured': measured
+                                         },
+                           devices = quads_by_name,
+                           families = families
+                           )

examples/mls_example/mls_registry.py

@@ -0,0 +1,18 @@
+"""Example how to use registry for MLS."""
+
+from mls_config import config
+from accml.core.registry.accelerator import Accelerator
+
+
+#%% Create the accelerator object
+
+mls = Accelerator(config)
+
+#%% Extract information from the registry
+
+print(f'Facility: {mls.facility}\n')
+print(f'Machine: {mls.machine}\n')
+print(f'Controls: {mls.controls}\n')
+print(f'Simulators: {mls.simulators}\n')
+print(f'Devices: {mls.devices}\n')
+print(f'Families: {mls.families}\n')

src/accml/core/config/__init__.py

@@ -0,0 +1 @@
+from .data_models import *

src/accml/core/config/data_models/__init__.py

@@ -0,0 +1,3 @@
+from .accelerator import *
+from .control_system import *
+from .device import *

src/accml/core/config/data_models/accelerator.py

@@ -0,0 +1,24 @@
+""" Configuration class for accelerator and families."""
+
+from pydantic import BaseModel
+from typing import Dict, List, Optional, Hashable
+from .control_system import ControlSystemConfig
+from .simulator import SimulatorConfig
+from .device import DeviceConfig
+
+class AcceleratorConfig(BaseModel):
+
+    facility: Optional[str]
+    machine: str
+    # TODO: data_storage
+    controls: Optional[Dict[Hashable, ControlSystemConfig]] = None
+    simulators: Optional[Dict[Hashable, SimulatorConfig]] = None
+    devices: Optional[Dict[Hashable, DeviceConfig]] = None
+    families: Optional[Dict[Hashable, List[Hashable]]] = None
+    # TODO: operational_modes
+
+
+class FamilyConfig(BaseModel):
+    name: Hashable
+    devices: list[Hashable]
+

src/accml/core/config/data_models/control_system.py

@@ -0,0 +1,29 @@
+""" Configuration classes for control system."""
+
+from abc import ABC
+from pydantic import BaseModel
+from typing import Optional
+from enum import Enum
+
+
+class ControlSystemConfig(BaseModel, ABC):
+    pass
+
+
+class TANGOConfig(ControlSystemConfig):
+    host: str
+
+
+class EPICSType(Enum):
+
+    CA = 'CA' # Channel Access
+    PV = 'PV' # PV Access
+
+
+class EPICSConfig(ControlSystemConfig):
+    access_type: EPICSType
+    pv_prefix: Optional[str] = ""
+
+
+class DOOCSConfig(ControlSystemConfig):
+    pass

src/accml/core/config/data_models/conversions.py

@@ -0,0 +1,22 @@
+""" Configuration classes for conversions."""
+
+# TODO: add different type of conversion models here
+
+from abc import ABC
+from pydantic import BaseModel
+
+class ConversionConfig(BaseModel, ABC):
+    pass
+
+
+class EnergyDependentConversionModel(ConversionConfig):
+    """Energy-dependent conversion model for magnetic objects.
+
+    Todo:
+        - Add metadata describing the units
+    """
+
+    intercept: float  # y-intercept of the calibration curve
+    slope: float  # slope of the calibration curve
+    conversion_type: str  # e.g., 'linear'
+

src/accml/core/config/data_models/data_storage.py

@@ -0,0 +1,3 @@
+""" Configuration classes for data storage."""
+
+# TODO: define data models for both storing as files and database.

src/accml/core/config/data_models/device.py

@@ -0,0 +1,52 @@
+""" Configuration classes for devices."""
+
+from pydantic import BaseModel, Field
+from abc import ABC
+from typing import Hashable, Optional
+from .conversions import ConversionConfig
+
+# TODO: figure out what is required for a device and what is same/different
+# depending on the control system
+# It needs to also be possible to configure devices from external modules
+# in the same way.
+
+class DeviceConfig(BaseModel, ABC):
+    name: Hashable # e.g., 'QF1C01A'
+    # TODO: what is common for all devices?
+
+
+class MagnetConfig(DeviceConfig):
+    magnet_type: str = Field(alias="type")
+    device_id: Optional[Hashable] = None# Engineering id, can be used for example to
+    # link to specific excitation curve or magnet model.
+    power_supply: Optional[Hashable] = None
+    conversion: Optional[ConversionConfig] = None
+
+
+# class ResponseModel(BaseModel):
+#     """General response model for a device reacting to a control input change
+
+#     Timeout: whithin this time the device has to answer
+#     Settle time: after this time the device is expected to be in a stable state
+#     """
+
+#     #: seconds
+#     timeout: float
+#     # seconds
+#     settle_time: float
+
+
+# class PowerConverterInterface(BaseModel):
+#     #:  e.g., 'CHANNEL:QF1C01A:SP'
+#     setpoint: str
+#     #: e.g., 'CHANNEL:QF1C01A:RB'
+#     readback: str
+
+
+# class PowerConverter(BaseModel):
+#     id: Hashable
+#     interface: PowerConverterInterface
+#     response: ResponseModel
+
+#     def get_current(self):
+#         return 0.0

src/accml/core/config/data_models/simulator.py

@@ -0,0 +1,12 @@
+""" Configuration classes for simulators."""
+
+from pydantic import BaseModel
+from enum import Enum
+
+class SimulationEngine(Enum):
+    PYAT = 'pyat'
+
+
+class SimulatorConfig(BaseModel):
+    type: SimulationEngine
+    model: str  # Path to the lattice model.

src/accml/core/registry/accelerator.py

@@ -0,0 +1,56 @@
+""" Module for the AO and registry functionality."""
+
+from accml.core.config import AcceleratorConfig
+from accml.core.registry.utils import WildcardDict
+
+class Accelerator():
+
+    def __init__(self, config: AcceleratorConfig):
+
+        self._facility = config.facility
+        self._machine = config.machine
+        self._controls = config.controls
+        self._simulators = config.simulators
+        self._devices = WildcardDict(config.devices) # Dict which can use wildcards
+        self._families = config.families
+
+
+    @property
+    def facility(self):
+        return self._facility
+
+    @property
+    def machine(self):
+        return self._machine
+
+    @property
+    def controls(self):
+        if self._controls:
+            return self._controls
+        else:
+            print('No controls have been configured.')
+
+    @property
+    def simulators(self):
+        if self._simulators:
+            return self._simulators
+        else:
+            print('No simulators have been configured.')
+
+    @property
+    def devices(self):
+        if self._devices:
+            return self._devices
+        else:
+            print('No devices have been configured.')
+
+    @property
+    def families(self):
+        if self._families:
+            return self._families
+        else:
+            print('No families have been configured.')
+
+    def __str__(self):
+        """Pretty printing of the accelerator configuration."""
+        pass

src/accml/core/registry/utils.py

@@ -0,0 +1,13 @@
+""" Module for registry utility functionality. """
+
+from collections import UserDict
+import fnmatch
+from typing import List
+
+class WildcardDict(UserDict):
+    """
+    Dictionary where keys can be found using wildcards.
+    """
+
+    def __getitem__(self, pattern: str) -> List:
+        return [v for k, v in self.data.items() if fnmatch.fnmatch(str(k), pattern)]