From 153fdf866740350cec45af0ce0d3af09957d6c20 Mon Sep 17 00:00:00 2001 From: guillaumepichon Date: Mon, 15 Jun 2026 15:56:13 +0200 Subject: [PATCH 1/3] Rename get_device() to get_device_access() for clarity Add coverage for catalog config, EPICS/Tango lookups, indexed devices, and get_device error paths. --- pyaml_cs_oa/controlsystem.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pyaml_cs_oa/controlsystem.py b/pyaml_cs_oa/controlsystem.py index 361d03c..6e1f2bb 100644 --- a/pyaml_cs_oa/controlsystem.py +++ b/pyaml_cs_oa/controlsystem.py @@ -73,7 +73,7 @@ def attach_array(self, devs: list[OASignal | None]) -> list[OASignal | None]: # Deprecated function return self._attach([d._cfg if d is not None else None for d in devs]) - def get_device(self, ref: str | BaseModel | None) -> DeviceAccess | None: + def get_device_access(self, ref: str | BaseModel | None) -> DeviceAccess | None: if ref is None: return None From 24a0209ed11475a0f2620c0d9acfcc391728c41b Mon Sep 17 00:00:00 2001 From: guillaumepichon Date: Mon, 15 Jun 2026 16:08:52 +0200 Subject: [PATCH 2/3] Corrections for tests --- tests/test_bpm_orbit.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/tests/test_bpm_orbit.py b/tests/test_bpm_orbit.py index e8ca5c8..8efec71 100644 --- a/tests/test_bpm_orbit.py +++ b/tests/test_bpm_orbit.py @@ -97,7 +97,7 @@ class IdentityAttachControlSystem(OphydAsyncControlSystem): # attach public methods are depecrated - def get_device(self, ref: str | BaseModel | None) -> DeviceAccess | None: + def get_device_access(self, ref: str | BaseModel | None) -> DeviceAccess | None: config = self._cfg.catalog.resolve(ref) return IndexedVectorSignal(config) @@ -112,7 +112,7 @@ def _attached_indexed_bpm( y_pos=f"BPM{bpm_index}:Y", ), ) - x_pos, y_pos = control_system.get_devices(model.get_pos_devices()) + x_pos, y_pos = control_system.get_device_access(model.get_pos_devices()) bpm = BPM(BPMConfig(name=f"BPM{bpm_index}", model=model)) return bpm.attach( From cfa5ae9b5478a40cb0f70d0536c526d47f855761 Mon Sep 17 00:00:00 2001 From: guillaumepichon Date: Mon, 15 Jun 2026 17:06:36 +0200 Subject: [PATCH 3/3] Corrections for static catalogs and tests --- pyaml_cs_oa/static_catalog.py | 4 +-- tests/test_bpm_orbit.py | 51 ++++++++++++++++++----------------- 2 files changed, 29 insertions(+), 26 deletions(-) diff --git a/pyaml_cs_oa/static_catalog.py b/pyaml_cs_oa/static_catalog.py index 4745057..0400f5f 100644 --- a/pyaml_cs_oa/static_catalog.py +++ b/pyaml_cs_oa/static_catalog.py @@ -22,7 +22,7 @@ class ConfigModel(BaseModel): class StaticCatalog(Catalog): def __init__(self, cfg: ConfigModel): - super().__init__(cfg) + self._cfg = cfg if not cfg.entries: raise PyAMLException("StaticCatalog.entries must contain at least one entry") self._refs: dict[str, DeviceAccess] = {} @@ -36,4 +36,4 @@ def resolve(self, key: str) -> BaseModel: try: return self._refs[key]._cfg except KeyError as exc: - raise PyAMLException(f"Catalog '{self.get_name()}' cannot resolve key '{key}'") from exc + raise PyAMLException(f"StaticCatalog cannot resolve key '{key}'") from exc diff --git a/tests/test_bpm_orbit.py b/tests/test_bpm_orbit.py index 8efec71..4878c85 100644 --- a/tests/test_bpm_orbit.py +++ b/tests/test_bpm_orbit.py @@ -10,9 +10,12 @@ from pyaml.control.deviceaccess import DeviceAccess from pydantic import BaseModel, ConfigDict -from pyaml_cs_oa.catalog import Catalog from pyaml_cs_oa.controlsystem import ConfigModel, OphydAsyncControlSystem from pyaml_cs_oa.float_signal import FloatSignalContainer +from pyaml_cs_oa.static_catalog import ConfigModel as StaticCatalogConfig +from pyaml_cs_oa.static_catalog import StaticCatalog +from pyaml_cs_oa.static_catalog_entry import ConfigModel as StaticCatalogEntryConfig +from pyaml_cs_oa.static_catalog_entry import StaticCatalogEntry from pyaml_cs_oa.types import EpicsConfigR @@ -84,14 +87,6 @@ def name(self) -> str: return f"{self._cfg.read_pvname}[{self._cfg.index}]" -class StaticCatalog(Catalog): - def __init__(self, devices: dict[str, DeviceAccess]) -> None: - self._devices = devices - - def resolve(self, key: str) -> BaseModel: - return self._devices[key] - - class IdentityAttachControlSystem(OphydAsyncControlSystem): """Control system fake that keeps pre-built DeviceAccess objects attached.""" @@ -112,7 +107,7 @@ def _attached_indexed_bpm( y_pos=f"BPM{bpm_index}:Y", ), ) - x_pos, y_pos = control_system.get_device_access(model.get_pos_devices()) + x_pos, y_pos = control_system.get_devices_access(model.get_pos_devices()) bpm = BPM(BPMConfig(name=f"BPM{bpm_index}", model=model)) return bpm.attach( @@ -124,20 +119,28 @@ def _attached_indexed_bpm( def _control_system_with_indexed_orbit(orbit_device: VectorDevice, bpm_count: int) -> IdentityAttachControlSystem: - catalog = StaticCatalog( - { - f"BPM{bpm_index}:X": IndexedVectorSignalConfig( - source=orbit_device, read_pvname=orbit_device.name(), unit=orbit_device.unit(), index=2 * bpm_index - ) - for bpm_index in range(bpm_count) - } - | { - f"BPM{bpm_index}:Y": IndexedVectorSignalConfig( - source=orbit_device, read_pvname=orbit_device.name(), unit=orbit_device.unit(), index=2 * bpm_index + 1 - ) - for bpm_index in range(bpm_count) - }, - ) + entries = [] + for bpm_index in range(bpm_count): + x_config = IndexedVectorSignalConfig( + source=orbit_device, + read_pvname=orbit_device.name(), + unit=orbit_device.unit(), + index=2 * bpm_index, + ) + y_config = IndexedVectorSignalConfig( + source=orbit_device, + read_pvname=orbit_device.name(), + unit=orbit_device.unit(), + index=2 * bpm_index + 1, + ) + entries.extend( + [ + StaticCatalogEntry(StaticCatalogEntryConfig(key=f"BPM{bpm_index}:X", device=IndexedVectorSignal(x_config))), + StaticCatalogEntry(StaticCatalogEntryConfig(key=f"BPM{bpm_index}:Y", device=IndexedVectorSignal(y_config))), + ], + ) + + catalog = StaticCatalog(StaticCatalogConfig(entries=entries)) control_system = IdentityAttachControlSystem(ConfigModel(name="live", catalog=catalog)) return control_system