Release 0.3.0

.pre-commit-config.yaml

@@ -12,10 +12,10 @@ repos:
         args: [--fix]
       - id: ruff-format
 
-#  - repo: https://github.com/pre-commit/mirrors-mypy
-#    rev: v1.18.2
-#    hooks:
-#      - id: mypy
-#        additional_dependencies: [
-#          "pydantic>=2.0",
-#        ]
+  - repo: https://github.com/pre-commit/mirrors-mypy
+    rev: v1.18.2
+    hooks:
+      - id: mypy
+        additional_dependencies: [
+          "pydantic>=2.0",
+        ]

pyaml_cs_oa/container.py

@@ -61,7 +61,6 @@ async def _run_get(self) -> SignalDatatypeT:
         """Connect and fetch the backend's current value."""
         await self._r_sig.connect()
         backend = self._r_sig._connector.backend
-        print(f"Read {self._r_sig.name}")
         return await backend.get_value()
 
     async def async_get(self) -> SignalDatatypeT:

pyaml_cs_oa/controlsystem.py

@@ -2,11 +2,13 @@
 
 import logging
 
+from pyaml.common.element import __pyaml_repr__
 from pyaml.common.exception import PyAMLException
 from pyaml.control.controlsystem import ControlSystem
 from pyaml.control.deviceaccess import DeviceAccess
 from pyaml.control.deviceaccesslist import DeviceAccessList
-from pydantic import BaseModel, ConfigDict
+from pyaml.validation import DynamicValidation, register_schema
+from pydantic import BaseModel
 
 from . import __version__
 from .aggregator import OAAggregator
@@ -23,48 +25,48 @@
 logger = logging.getLogger(__name__)
 
 
-class ConfigModel(BaseModel):
-    """
-    Configuration model for an OA Control System.
-
-    Attributes
-    ----------
-    name : str
-        Name of the control system.
-    prefix : str
-        Prefix added to the PV or attribute name. It can be a
-        for instance, TANGO_HOST, or a PV prefix.
-    catalog : Catalog | None
-        Catalog instance or catalog name used to resolve PyAML device keys.
-        If None specified a dynamic catalog is used.
-    debug_level : str
-        Debug verbosity level.
-    """
-
-    model_config = ConfigDict(arbitrary_types_allowed=True, extra="forbid")
-
-    name: str
-    prefix: str = ""
-    catalog: Catalog | None = None
-    debug_level: str | None = None
-
-
-class OphydAsyncControlSystem(ControlSystem):
+@register_schema
+class OphydAsyncControlSystem(ControlSystem, DynamicValidation):
     """Generic PyAML control system using an ophyd-async backend."""
 
-    def __init__(self, cfg: ConfigModel):
+    def __init__(
+        self,
+        name: str,
+        prefix: str = "",
+        catalog: Catalog | None = None,
+        debug_level: str | int | None = None,
+    ):
+        """Create an ophyd-async control-system interface.
+
+        Parameters
+        ----------
+        name : str
+            Name used to identify the control system.
+        prefix : str, optional
+            Prefix added to the PV or attribute name. It can be a
+            for instance, TANGO_HOST, or a PV prefix.
+        catalog : Catalog or None, optional
+            Catalog instance or catalog name used to resolve PyAML device keys.
+            If None specified a dynamic catalog is used.
+        debug_level : str | int | None, optional
+            Debug verbosity level. Such as INFO, DEBUG, WARNING, ERROR, CRITICAL. Or 10, 20, 30, 40, 50.
+        """
+
         super().__init__()
-        self._cfg = cfg
+        self._name = name
+        self._prefix = prefix
+        self._catalog = catalog
+        self._debug_level = debug_level
         self._devices: dict[str, DeviceAccess] = {}  # Dict containing all attached DeviceAccess
 
-        if self._cfg.debug_level:
-            log_level = getattr(logging, self._cfg.debug_level, logging.WARNING)
+        if self._debug_level:
+            log_level = getattr(logging, self._debug_level, logging.WARNING)
             logger.setLevel(log_level)
 
         logger.log(
             logging.WARNING,
-            f"PyAML OA control system binding ({__version__}) initialized with name '{self._cfg.name}'"
-            f" and prefix='{self._cfg.prefix}'",
+            f"PyAML OA control system binding ({__version__}) initialized with name '{self._name}'"
+            f" and prefix='{self._prefix}'",
         )
 
     def attach(self, devs: list[OASignal | None]) -> list[OASignal | None]:
@@ -84,10 +86,10 @@ def get_device_access(self, ref: str | BaseModel | None) -> DeviceAccess | None:
 
         if isinstance(ref, str):
             # Retrieve a config from a key using using a Catalog
-            if self._cfg.catalog is None:
+            if self._catalog is None:
                 raise PyAMLException(f"Control system '{self.name()}' has no catalog when trying to resolve '{ref}'")
             try:
-                ref = self._cfg.catalog.resolve(ref)
+                ref = self._catalog.resolve(ref)
             except AttributeError as exc:
                 raise PyAMLException(f"Control system '{self.name()}' catalog cannot resolve key '{ref}'") from exc
 
@@ -111,30 +113,30 @@ def _attach(self, configs: list[ControlSysConfig | None]) -> list[OASignal | Non
                 index_str = "" if sig_cfg.index is None else str(sig_cfg.index)
 
                 if isinstance(sig_cfg, EpicsConfigR):
-                    key = self._cfg.prefix + sig_cfg.read_pvname + index_str
+                    key = self._prefix + sig_cfg.read_pvname + index_str
                     sig_cls = EpicsR
-                    config = dict(read_pvname=self._cfg.prefix + sig_cfg.read_pvname)
+                    config = dict(read_pvname=self._prefix + sig_cfg.read_pvname)
                 elif isinstance(sig_cfg, EpicsConfigW):
-                    key = self._cfg.prefix + sig_cfg.write_pvname + index_str
+                    key = self._prefix + sig_cfg.write_pvname + index_str
                     sig_cls = EpicsW
-                    config = dict(write_pvname=self._cfg.prefix + sig_cfg.write_pvname)
+                    config = dict(write_pvname=self._prefix + sig_cfg.write_pvname)
                 elif isinstance(sig_cfg, EpicsConfigRW):
-                    key = self._cfg.prefix + sig_cfg.read_pvname + sig_cfg.write_pvname + index_str
+                    key = self._prefix + sig_cfg.read_pvname + sig_cfg.write_pvname + index_str
                     sig_cls = EpicsRW
                     config = dict(
-                        read_pvname=self._cfg.prefix + sig_cfg.read_pvname,
-                        write_pvname=self._cfg.prefix + sig_cfg.write_pvname,
+                        read_pvname=self._prefix + sig_cfg.read_pvname,
+                        write_pvname=self._prefix + sig_cfg.write_pvname,
                     )
                 elif isinstance(sig_cfg, TangoConfigAtt):
-                    key = self._cfg.prefix + sig_cfg.attribute + index_str
+                    key = self._prefix + sig_cfg.attribute + index_str
                     sig_cls = TangoAtt
-                    config = dict(attribute=self._cfg.prefix + sig_cfg.attribute)
+                    config = dict(attribute=self._prefix + sig_cfg.attribute)
                 else:
                     raise PyAMLException(f"OphydAsyncControlSystem: Unsupported type {type(sig_cfg)}")
 
                 if key not in self._devices:
                     n_conf = dict(sig_cfg) | config
-                    nr = sig_cls(sig_cfg_cls(**n_conf))
+                    nr = sig_cls(**n_conf)
                     nr.build()
                     self._devices[key] = nr
 
@@ -152,11 +154,11 @@ def name(self) -> str:
         str
             Name of the control system.
         """
-        return self._cfg.name
+        return self._name
 
     def get_aggregator(self) -> DeviceAccessList | None:
         """Return a new empty aggregator for batched device operations."""
         return OAAggregator()
 
     def __repr__(self):
-        return repr(self._cfg).replace("ConfigModel", self.__class__.__name__)
+        return __pyaml_repr__()

pyaml_cs_oa/dynamic_catalog.py

@@ -4,14 +4,16 @@
 
 from pyaml.common.exception import PyAMLException
 from pyaml.control.deviceaccess import DeviceAccess
-from pydantic import BaseModel, ConfigDict
+from pyaml.validation import DynamicValidation, register_schema
+from pydantic import BaseModel
 
 from .catalog import Catalog
 
 PYAMLCLASS = "DynamicCatalog"
 
 
-class ConfigModel(BaseModel):
+@register_schema
+class DynamicCatalog(Catalog, DynamicValidation):
     """
     Default dynamic catalog.
 
@@ -41,19 +43,17 @@ class ConfigModel(BaseModel):
         timeout_ms: 3000
     """
 
-    model_config = ConfigDict(arbitrary_types_allowed=True, extra="forbid")
-    timeout_ms: int = 3000
-    backend: str = ""
+    def __init__(
+        self,
+        timeout_ms: int = 3000,
+        backend: str = "",
+    ):
+        self._timeout_ms = timeout_ms
+        self._backend = backend
 
-
-class DynamicCatalog(Catalog):
-    """Resolve compact backend specifications into configuration models."""
-
-    def __init__(self, cfg: ConfigModel):
-        self._cfg = cfg
         self._dp = {}  # Device proxy cache (Tango Only)
-        if cfg.backend.lower() != "tango" and cfg.backend.lower() != "epics":
-            raise PyAMLException(f"backend must be `epics` or `tango` but got '{cfg.backend}'") from None
+        if self._backend.lower() != "tango" and self._backend.lower() != "epics":
+            raise PyAMLException(f"backend must be `epics` or `tango` but got '{self._backend}'") from None
 
     def resolve(self, key: str) -> BaseModel:
         """Resolve a backend specification string.
@@ -68,10 +68,10 @@ def resolve(self, key: str) -> BaseModel:
         pydantic.BaseModel
             Backend-specific configuration model.
         """
-        if self._cfg.backend.lower() == "epics":
-            return _build_epics_config(key, self._cfg.timeout_ms)
-        elif self._cfg.backend.lower() == "tango":
-            return _build_tango_config(key, self._cfg.timeout_ms)
+        if self._backend.lower() == "epics":
+            return _build_epics_config(key, self._timeout_ms)
+        elif self._backend.lower() == "tango":
+            return _build_tango_config(key, self._timeout_ms)
         else:
             return None
 
@@ -120,18 +120,16 @@ def _parse_pv(token: str) -> tuple[list[str], int | None, str, bool]:
 
 def _build_epics_config(pv_str: str, timeout_ms: int) -> DeviceAccess:
     """Build an EPICS configuration model from a specification string."""
-    from .epicsR import ConfigModel as EpicsRConfig
-    from .epicsRW import ConfigModel as EpicsRWConfig
-    from .epicsW import ConfigModel as EpicsWConfig
+    from .types import EpicsConfigR, EpicsConfigRW, EpicsConfigW
 
     pv_names, index, unit, hasw = _parse_pv(pv_str)
     if len(pv_names) == 1:
         if hasw:
-            return EpicsWConfig(write_pvname=pv_names[0], timeout_ms=timeout_ms, index=index, unit=unit)
+            return EpicsConfigW(write_pvname=pv_names[0], timeout_ms=timeout_ms, index=index, unit=unit)
         else:
-            return EpicsRConfig(read_pvname=pv_names[0], timeout_ms=timeout_ms, index=index, unit=unit)
+            return EpicsConfigR(read_pvname=pv_names[0], timeout_ms=timeout_ms, index=index, unit=unit)
     if len(pv_names) == 2:
-        return EpicsRWConfig(read_pvname=pv_names[0], write_pvname=pv_names[1], timeout_ms=timeout_ms, index=index, unit=unit)
+        return EpicsConfigRW(read_pvname=pv_names[0], write_pvname=pv_names[1], timeout_ms=timeout_ms, index=index, unit=unit)
     raise PyAMLException(f"Too many comma-separated tokens in key '{pv_str}' (max 2)")
 
 
@@ -159,7 +157,7 @@ def _parse_attribute(token: str) -> tuple[list[str], int | None, str]:
 
 def _build_tango_config(att_name: str, timeout_ms: int) -> BaseModel:
     """Build a Tango configuration model from a specification string."""
-    from .tangoAtt import ConfigModel as TangoAtt
+    from .types import TangoConfigAtt
 
     att_name, index, unit = _parse_attribute(att_name)
-    return TangoAtt(attribute=att_name, timeout_ms=timeout_ms, index=index, unit=unit)
+    return TangoConfigAtt(attribute=att_name, timeout_ms=timeout_ms, index=index, unit=unit)

pyaml_cs_oa/epicsR.py

@@ -1,19 +1,19 @@
 """Read-only EPICS signal implementation."""
 
+from pyaml.validation import DynamicValidation, register_schema
+
 from .float_signal import FloatSignalContainer
 from .types import EpicsConfigR
 
 PYAMLCLASS: str = "EpicsR"
 
 
-class ConfigModel(EpicsConfigR):
-    """Configuration model registered for the ``EpicsR`` device class."""
-
-
-class EpicsR(FloatSignalContainer):
+@register_schema
+class EpicsR(FloatSignalContainer, DynamicValidation):
     """PyAML read-only signal backed by an EPICS read signal."""
 
-    def __init__(self, cfg: ConfigModel):
+    def __init__(self, read_pvname: str, timeout_ms: int = 3000, index: int | None = None, unit: str = ""):
+        cfg = EpicsConfigR(read_pvname=read_pvname, timeout_ms=timeout_ms, index=index, unit=unit)
         super().__init__(cfg)
 
     def get_cs(self) -> str:

pyaml_cs_oa/epicsRW.py

@@ -1,19 +1,29 @@
 """Read/write EPICS signal implementation."""
 
+from pyaml.validation import DynamicValidation, register_schema
+
 from .float_signal import FloatSignalContainer
 from .types import EpicsConfigRW
 
 PYAMLCLASS: str = "EpicsRW"
 
 
-class ConfigModel(EpicsConfigRW):
-    """Configuration model registered for the ``EpicsRW`` device class."""
-
-
-class EpicsRW(FloatSignalContainer):
+@register_schema
+class EpicsRW(FloatSignalContainer, DynamicValidation):
     """PyAML read/write signal backed by an EPICS signal."""
 
-    def __init__(self, cfg: ConfigModel):
+    def __init__(
+        self,
+        read_pvname: str,
+        write_pvname: str,
+        timeout_ms: int = 3000,
+        range: list[float] | None = None,
+        index: int | None = None,
+        unit: str = "",
+    ):
+        cfg = EpicsConfigRW(
+            read_pvname=read_pvname, write_pvname=write_pvname, timeout_ms=timeout_ms, range=range, index=index, unit=unit
+        )
         super().__init__(cfg)
 
     def get_cs(self) -> str:

pyaml_cs_oa/epicsW.py

@@ -1,19 +1,32 @@
 """Write-only EPICS signal implementation."""
 
+from pyaml.validation import DynamicValidation, register_schema
+
 from .float_signal import FloatSignalContainer
 from .types import EpicsConfigW
 
 PYAMLCLASS: str = "EpicsW"
 
 
-class ConfigModel(EpicsConfigW):
-    """Configuration model registered for the ``EpicsW`` device class."""
-
-
-class EpicsW(FloatSignalContainer):
+@register_schema
+class EpicsW(FloatSignalContainer, DynamicValidation):
     """PyAML write-only signal backed by an EPICS write signal."""
 
-    def __init__(self, cfg: ConfigModel):
+    def __init__(
+        self,
+        write_pvname: str,
+        timeout_ms: int = 3000,
+        range: list[float] | None = None,
+        index: int | None = None,
+        unit: str = "",
+    ):
+        cfg = EpicsConfigW(
+            write_pvname=write_pvname,
+            timeout_ms=timeout_ms,
+            range=range,
+            index=index,
+            unit=unit,
+        )
         super().__init__(cfg)
 
     def get_cs(self) -> str: