First catalog version with first reference in a test.

Author: gupichon-soleil

pyaml/accelerator.py

@@ -9,6 +9,7 @@
 from .arrays.array import ArrayConfig
 from .common.element import Element
 from .common.exception import PyAMLConfigException
+from .configuration.catalog import Catalog
 from .configuration.factory import Factory
 from .configuration.fileloader import load, set_root_folder
 from .control.controlsystem import ControlSystem
@@ -44,6 +45,8 @@ class ConfigModel(BaseModel):
         Acceleration description
     devices : list[Element]
         Element list
+    control_system_catalog : Catalog
+        catalog of DeviceAccess objects
     """
 
     model_config = ConfigDict(arbitrary_types_allowed=True, extra="forbid")
@@ -57,6 +60,7 @@ class ConfigModel(BaseModel):
     description: str | None = None
     arrays: list[ArrayConfig] = Field(default=None, repr=False)
     devices: list[Element] = Field(repr=False)
+    control_system_catalog: Catalog = Field(repr=False)
 
 
 class Accelerator(object):
@@ -69,6 +73,7 @@ def __init__(self, cfg: ConfigModel):
 
         if cfg.controls is not None:
             for c in cfg.controls:
+                c.set_catalog(cfg.control_system_catalog)
                 if c.name() == "live":
                     self.__live = c
                 else:

pyaml/configuration/catalog.py

@@ -0,0 +1,155 @@
+import re
+from typing import Pattern
+
+from pydantic import BaseModel, ConfigDict, model_validator
+
+from pyaml import PyAMLException
+from pyaml.configuration.catalog_entry import CatalogEntry, CatalogValue
+from pyaml.configuration.catalog_entry import ConfigModel as CatalogEntryConfigModel
+from pyaml.control.deviceaccess import DeviceAccess
+
+# Define the main class name for this module
+PYAMLCLASS = "Catalog"
+
+
+class ConfigModel(BaseModel):
+    """
+    Configuration model for a value catalog.
+
+    Attributes
+    ----------
+    refs : list[CatalogEntryConfigModel]
+        List of catalog entries.
+    """
+
+    model_config = ConfigDict(arbitrary_types_allowed=True, extra="forbid")
+
+    refs: list[CatalogEntry]
+
+    @model_validator(mode="after")
+    def _validate_unique_references(self) -> "ConfigModel":
+        """
+        Ensure that all references are unique within the catalog.
+        """
+        seen: set[str] = set()
+        for entry in self.refs:
+            if entry.get_reference() in seen:
+                raise ValueError(
+                    f"Duplicate catalog reference: '{entry.get_reference()}'."
+                )
+            seen.add(entry.get_reference())
+        return self
+
+
+class Catalog:
+    """
+    A simple registry mapping reference keys to DeviceAccess objects.
+
+    The catalog is intentionally minimal:
+    - It resolves references to DeviceAccess or list[DeviceAccess]
+    - It does NOT expose any DeviceAccess-like interface (no get/set/readback/etc.)
+    """
+
+    def __init__(self, cfg: ConfigModel):
+        self._entries: dict[str, CatalogValue] = {}
+        for ref in cfg.refs:
+            self.add(ref.get_reference(), ref.get_value())
+
+    # ------------------------------------------------------------------
+
+    def add(self, reference: str, value: CatalogValue):
+        """
+        Register a reference in the catalog.
+
+        Raises
+        ------
+        PyAMLException
+            If the reference already exists.
+        """
+        if reference in self._entries:
+            raise PyAMLException(f"Duplicate catalog reference: '{reference}'")
+        self._entries[reference] = value
+
+    # ------------------------------------------------------------------
+
+    def get(self, reference: str) -> CatalogValue:
+        """
+        Resolve a reference key.
+
+        Returns
+        -------
+        DeviceAccess | list[DeviceAccess]
+
+        Raises
+        ------
+        PyAMLException
+            If the reference does not exist.
+        """
+        try:
+            return self._entries[reference]
+        except KeyError as exc:
+            raise PyAMLException(f"Catalog reference '{reference}' not found.") from exc
+
+    # ------------------------------------------------------------------
+
+    def get_one(self, reference: str) -> DeviceAccess:
+        """
+        Resolve a reference and ensure it corresponds to a single DeviceAccess.
+
+        Raises
+        ------
+        PyAMLException
+            If the reference does not exist or is multi-device.
+        """
+        value = self.get(reference)
+
+        if isinstance(value, list):
+            raise PyAMLException(
+                f"Catalog reference '{reference}' is multi-device; use get_many()."
+            )
+
+        return value
+
+    # ------------------------------------------------------------------
+
+    def get_many(self, reference: str) -> list[DeviceAccess]:
+        """
+        Resolve a reference and ensure it corresponds to multiple DeviceAccess.
+
+        Returns
+        -------
+        list[DeviceAccess]
+
+        Raises
+        ------
+        PyAMLException
+            If the reference does not exist or is single-device.
+        """
+        value = self.get(reference)
+
+        if not isinstance(value, list):
+            raise PyAMLException(
+                f"Catalog reference '{reference}' is single-device; use get_one()."
+            )
+
+        return value
+
+    # ------------------------------------------------------------------
+
+    def find(self, pattern: str) -> dict[str, CatalogValue]:
+        """
+        Resolve references matching a regular expression.
+
+        Returns
+        -------
+        dict[str, DeviceAccess | list[DeviceAccess]]
+            Mapping {reference -> value}.
+        """
+        regex: Pattern[str] = re.compile(pattern)
+        return {k: v for k, v in self._entries.items() if regex.search(k)}
+
+    # ------------------------------------------------------------------
+
+    def keys(self) -> list[str]:
+        """Return all catalog reference keys."""
+        return list(self._entries.keys())

pyaml/configuration/catalog_entry.py

@@ -0,0 +1,74 @@
+from typing import Union
+
+from pydantic import BaseModel, ConfigDict, model_validator
+
+from pyaml.control.deviceaccess import DeviceAccess
+
+# Define the main class name for this module
+PYAMLCLASS = "CatalogEntry"
+
+
+class ConfigModel(BaseModel):
+    """
+    Configuration model for a single catalog entry.
+
+    Exactly one of 'device' or 'devices' must be provided.
+
+    Attributes
+    ----------
+    reference : str
+        Unique key used to identify the value in the catalog.
+    device : dict | None
+        Factory configuration dict for a single DeviceAccess.
+    devices : list[DeviceAccess] | None
+        Factory configuration dicts for multiple DeviceAccess objects.
+    """
+
+    model_config = ConfigDict(arbitrary_types_allowed=True, extra="forbid")
+
+    reference: str
+    device: DeviceAccess = None
+    devices: list[DeviceAccess] = None
+
+    @model_validator(mode="after")
+    def _validate_one_of(self) -> "ConfigModel":
+        """
+        Ensure exactly one of (device, devices) is provided and properly shaped.
+        """
+        has_device = self.device is not None
+        has_devices = self.devices is not None and len(self.devices) > 0
+
+        # both True or both False -> invalid
+        if has_device == has_devices:
+            raise ValueError(
+                "Catalog entry must define exactly one of 'device' or 'devices'."
+            )
+
+        if has_device and not isinstance(self.device, DeviceAccess):
+            raise ValueError("'device' must be a DeviceAccess.")
+
+        if has_devices:
+            if not isinstance(self.devices, list) or len(self.devices) == 0:
+                raise ValueError("'devices' must be a non-empty list.")
+            for i, d in enumerate(self.devices):
+                if not isinstance(d, DeviceAccess):
+                    raise ValueError(f"'devices[{i}]' must be a DeviceAccess.")
+
+        return self
+
+
+CatalogValue = Union[DeviceAccess, list[DeviceAccess]]
+
+
+class CatalogEntry:
+    def __init__(self, cfg: ConfigModel):
+        self._cfg: ConfigModel = cfg
+        self._value: CatalogValue = (
+            cfg.device if cfg.device is not None else cfg.devices
+        )
+
+    def get_reference(self) -> str:
+        return self._cfg.reference
+
+    def get_value(self) -> CatalogValue:
+        return self._value

pyaml/control/controlsystem.py

@@ -8,6 +8,7 @@
 from ..common.element import Element
 from ..common.element_holder import ElementHolder
 from ..common.exception import PyAMLException
+from ..configuration.catalog import Catalog
 from ..configuration.factory import Factory
 from ..control.abstract_impl import (
     CSBPMArrayMapper,
@@ -47,6 +48,10 @@ class ControlSystem(ElementHolder, metaclass=ABCMeta):
 
     def __init__(self):
         ElementHolder.__init__(self)
+        self._catalog: Catalog | None = None
+
+    def set_catalog(self, catalog: Catalog):
+        self._catalog = catalog
 
     @abstractmethod
     def attach(self, dev: list[DeviceAccess]) -> list[DeviceAccess]:

tests/config/bpms.yaml

@@ -85,3 +85,30 @@ devices:
     - [A0, SH1A-C01-V]
     - [A1, SH1A-C01-SQ]
   model: sr/magnet_models/SH1AC01.yaml
+control_system_catalog:
+  type: pyaml.configuration.catalog
+  refs:
+    - type: pyaml.configuration.catalog_entry
+      reference: BPM_C01-02
+      devices:
+        - type: tango.pyaml.attribute_read_only
+          attribute: srdiag/bpm/c01-02/SA_HPosition
+          unit: mm
+        - type: tango.pyaml.attribute_read_only
+          attribute: srdiag/bpm/c01-02/SA_HPosition
+          unit: mm
+    - type: pyaml.configuration.catalog_entry
+      reference: BPM_C01-03
+      devices:
+        - type: tango.pyaml.attribute_read_only
+          attribute: srdiag/bpm/c01-03/SA_HPosition
+          unit: mm
+        - type: tango.pyaml.attribute_read_only
+          attribute: srdiag/bpm/c01-03/SA_VPosition
+          unit: mm
+    - type: pyaml.configuration.catalog_entry
+      reference: BPM_C01-04
+      device:
+        type: tango.pyaml.attribute_read_only
+        attribute: srdiag/bpm/c01-04/Position
+        unit: mm