diff --git a/.github/workflows/deploy-pypi.yaml b/.github/workflows/deploy-pypi.yaml index 4cf8a66..b1c116a 100644 --- a/.github/workflows/deploy-pypi.yaml +++ b/.github/workflows/deploy-pypi.yaml @@ -8,7 +8,7 @@ on: jobs: deploy: - + runs-on: ubuntu-latest #environment: release @@ -27,7 +27,7 @@ jobs: cache-dependency-path: '**/pyproject.toml' - name: Install dependencies run: | - python -m pip install --upgrade pip + python -m pip install --upgrade pip pip install hatch - name: Build package run: hatch build diff --git a/.github/workflows/tests.yml b/.github/workflows/tests.yml index 5215324..8205baa 100644 --- a/.github/workflows/tests.yml +++ b/.github/workflows/tests.yml @@ -36,7 +36,7 @@ jobs: pip install accelerator-toolbox pip install matplotlib pip install h5py - pip install accelerator-middle-layer + pip install "git+https://github.com/python-accelerator-middle-layer/pyaml.git" pip install flake8 pytest if [ -f requirements.txt ]; then pip install -r requirements.txt; fi - name: Lint with flake8 diff --git a/.gitignore b/.gitignore index bc0e58b..5a92b6b 100644 --- a/.gitignore +++ b/.gitignore @@ -55,7 +55,7 @@ coverage.xml .pytest_cache/ cover/ -# DS +# DS **/.DS_Store # Translations diff --git a/README.md b/README.md index a65fd1a..1f098f7 100644 --- a/README.md +++ b/README.md @@ -53,14 +53,12 @@ Python code: ```python from tango.pyaml.attribute import Attribute -from tango.pyaml.tango_attribute import ConfigModel import yaml with open("attribute.yaml") as f: cfg_dict = yaml.safe_load(f) -cfg = ConfigModel(**cfg_dict) -attr = Attribute(cfg) +attr = Attribute(**cfg_dict) attr.set(10.0) value = attr.get() @@ -99,4 +97,3 @@ This project is licensed under the MIT License. ## Links - ๐Ÿงบ [Repository](https://github.com/python-accelerator-middle-layer/tango-pyaml) - diff --git a/pyproject.toml b/pyproject.toml index 39cb6a6..48d1d9a 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -56,6 +56,9 @@ dev = [ "pre-commit", ] +[project.entry-points."pyaml.schemas"] +tango_pyaml = "tango.pyaml" + [project.urls] Homepage = "https://github.com/python-accelerator-middle-layer/tango-pyaml" Documentation = "https://python-accelerator-middle-layer.github.io/tango-pyaml/" diff --git a/tango/pyaml/attribute.py b/tango/pyaml/attribute.py index 88aedbc..36d90fe 100644 --- a/tango/pyaml/attribute.py +++ b/tango/pyaml/attribute.py @@ -1,22 +1,25 @@ import copy import logging -from typing import Optional, Tuple from pydantic import BaseModel +import pyaml +import tango +from pyaml.common.element import __pyaml_repr__ from pyaml.control.deviceaccess import DeviceAccess -from pyaml.control.readback_value import Value, Quality +from pyaml.control.readback_value import Quality, Value +from pyaml.validation import DynamicValidation, register_schema -from .initializable_element import InitializableElement from .device_factory import DeviceFactory -from .tango_pyaml_utils import * +from .initializable_element import InitializableElement +from .tango_pyaml_utils import tango_to_PyAMLException, to_float_or_none PYAMLCLASS: str = "Attribute" logger = logging.getLogger(__name__) -class ConfigModel(BaseModel): +class AttributeConfig(BaseModel): """ Configuration model for Tango attributes. @@ -36,18 +39,29 @@ class ConfigModel(BaseModel): attribute: str unit: str = "" - range: Optional[Tuple[Optional[float], Optional[float]]] = None - index: Optional[int] = None + range: tuple[float | None, float | None] | None = None + index: int | None = None -class Attribute(DeviceAccess, InitializableElement): +@register_schema +class Attribute(DeviceAccess, InitializableElement, DynamicValidation): """ Tango attribute that can be written to. Parameters ---------- - cfg : ConfigModel - Configuration object containing attribute path and units. + attribute : str + Full path of the Tango attribute (e.g., 'my/ps/device/current'). + unit : str, optional + The unit of the attribute. + range : tuple(min, max), optional + Range of valid values. Use null for -โˆž or +โˆž. + index : int, optional + Zero-based index into a SPECTRUM attribute. When set, the instance + behaves as a read-only scalar view of one vector element; writes are + always rejected and a SPECTRUM data_format is enforced on init. + writable : bool, optional + If the attribute should be writable. Default is True. Raises ------ @@ -55,10 +69,21 @@ class Attribute(DeviceAccess, InitializableElement): If the Tango attribute is not writable. """ - def __init__(self, cfg: ConfigModel, writable=True): + def __init__( + self, + attribute: str, + unit: str = "", + range: tuple[float | None, float | None] | None = None, + index: int | None = None, + writable=True, + ): super().__init__() - self._cfg = cfg - self._index = cfg.index + + self._attribute = attribute + self._unit = unit + self._range = range + self._index = index + # Indexed access never writes individual array elements. self._writable = writable and self._index is None self._attribute_dev: tango.DeviceProxy = None @@ -69,9 +94,7 @@ def __init__(self, cfg: ConfigModel, writable=True): def initialize(self): super().initialize() try: - self._attribute_dev_name, self._attr_name = self._cfg.attribute.rsplit( - "/", 1 - ) + self._attribute_dev_name, self._attr_name = self._attribute.rsplit("/", 1) self._attribute_dev = DeviceFactory().get_device(self._attribute_dev_name) except tango.DevFailed as df: raise tango_to_PyAMLException(df) @@ -80,22 +103,23 @@ def initialize(self): self._attribute_dev.get_attribute_config(self._attr_name, wait=True) ) - if self._index is not None: - if self._attr_config.data_format != tango.AttrDataFormat.SPECTRUM: - raise pyaml.PyAMLException( - f"Tango attribute '{self._cfg.attribute}' is not a SPECTRUM; " - "indexed access requires a vector attribute." - ) - - if self._writable: - if self._attr_config.writable not in [ - tango.AttrWriteType.READ_WRITE, - tango.AttrWriteType.WRITE, - tango.AttrWriteType.READ_WITH_WRITE, - ]: - raise pyaml.PyAMLException( - f"Tango attribute {self._cfg.attribute} is not writable." - ) + if ( + self._index is not None + and self._attr_config.data_format != tango.AttrDataFormat.SPECTRUM + ): + raise pyaml.PyAMLException( + f"Tango attribute '{self._attribute}' is not a SPECTRUM; " + "indexed access requires a vector attribute." + ) + + if self._writable and self._attr_config.writable not in [ + tango.AttrWriteType.READ_WRITE, + tango.AttrWriteType.WRITE, + tango.AttrWriteType.READ_WITH_WRITE, + ]: + raise pyaml.PyAMLException( + f"Tango attribute {self._attribute} is not writable." + ) def is_writable(self): return self._writable @@ -116,12 +140,12 @@ def set(self, value: float): """ if self._index is not None: raise pyaml.PyAMLException( - f"Indexed attribute '{self._cfg.attribute}[{self._index}]' " + f"Indexed attribute '{self._attribute}[{self._index}]' " "does not support individual element writes." ) self._ensure_initialized() logger.log( - logging.DEBUG, f"Setting asynchronously {self._cfg.attribute} to {value}" + logging.DEBUG, f"Setting asynchronously {self._attribute} to {value}" ) try: self._attribute_dev.write_attribute_asynch(self._attr_name, value) @@ -144,11 +168,11 @@ def set_and_wait(self, value: float): """ if self._index is not None: raise pyaml.PyAMLException( - f"Indexed attribute '{self._cfg.attribute}[{self._index}]' " + f"Indexed attribute '{self._attribute}[{self._index}]' " "does not support individual element writes." ) self._ensure_initialized() - logger.log(logging.DEBUG, f"Setting {self._cfg.attribute} to {value}") + logger.log(logging.DEBUG, f"Setting {self._attribute} to {value}") try: self._attribute_dev.write_attribute(self._attr_name, value) except tango.DevFailed as df: @@ -169,13 +193,17 @@ def readback(self) -> Value: If the Tango read fails. """ self._ensure_initialized() - logger.log(logging.DEBUG, f"Reading {self._cfg.attribute}") + logger.log(logging.DEBUG, f"Reading {self._attribute}") try: attr_value = self._attribute_dev.read_attribute(self._attr_name) quality = Quality[ attr_value.quality.name.rsplit("_", 1)[1] ] # AttrQuality.ATTR_VALID gives Quality.VALID - raw = attr_value.value[self._index] if self._index is not None else attr_value.value + raw = ( + attr_value.value[self._index] + if self._index is not None + else attr_value.value + ) value = Value(raw, quality, attr_value.time.todatetime()) except tango.DevFailed as df: raise tango_to_PyAMLException(df) @@ -190,7 +218,7 @@ def unit(self) -> str: str The unit string. """ - return self._cfg.unit + return self._unit def name(self) -> str: """ @@ -203,8 +231,8 @@ def name(self) -> str: notation when indexed (e.g., 'my/ps/device/current[2]'). """ if self._index is not None: - return f"{self._cfg.attribute}[{self._index}]" - return self._cfg.attribute + return f"{self._attribute}[{self._index}]" + return self._attribute def get_tango_attribute(self) -> str: """ @@ -215,7 +243,7 @@ def get_tango_attribute(self) -> str: str Tango attribute path stored in the configuration. """ - return self._cfg.attribute + return self._attribute def clone_with_tango_attribute(self, attribute: str) -> "Attribute": """ @@ -227,8 +255,7 @@ def clone_with_tango_attribute(self, attribute: str) -> "Attribute": Tango attribute path to store in the cloned instance. """ new_obj = copy.copy(self) - new_obj._cfg = copy.copy(self._cfg) - new_obj._cfg.attribute = attribute + new_obj._attribute = attribute return new_obj def measure_name(self) -> str: @@ -241,7 +268,7 @@ def measure_name(self) -> str: The attribute name (e.g., 'current'), with index notation when indexed (e.g., 'current[2]'). """ - short = self._cfg.attribute.rsplit("/", 1)[1] + short = self._attribute.rsplit("/", 1)[1] if self._index is not None: return f"{short}[{self._index}]" return short @@ -274,13 +301,9 @@ def get(self) -> float: def get_range(self) -> list[float]: attr_range: list[float] = [None, None] - if self._cfg.range is not None: - attr_range[0] = ( - self._cfg.range[0] if self._cfg.range[0] is not None else None - ) - attr_range[1] = ( - self._cfg.range[1] if self._cfg.range[1] is not None else None - ) + if self._range is not None: + attr_range[0] = self._range[0] if self._range[0] is not None else None + attr_range[1] = self._range[1] if self._range[1] is not None else None else: self._ensure_initialized() min_value = self._attr_config.min_value @@ -295,9 +318,9 @@ def check_device_availability(self) -> bool: try: self._ensure_initialized() self._attribute_dev.ping() - except tango.DevFailed | pyaml.PyAMLException: + except (tango.DevFailed, pyaml.PyAMLException): available = False return available def __repr__(self): - return repr(self._cfg).replace("ConfigModel", self.__class__.__name__) + return __pyaml_repr__(self) diff --git a/tango/pyaml/attribute_list.py b/tango/pyaml/attribute_list.py index 44156e5..6e39a62 100644 --- a/tango/pyaml/attribute_list.py +++ b/tango/pyaml/attribute_list.py @@ -1,11 +1,14 @@ import logging -import pyaml from numpy import array from pydantic import BaseModel -from pyaml.control.deviceaccess import DeviceAccess -from pyaml.control.readback_value import Value, Quality + +import pyaml import tango +from pyaml.common.element import __pyaml_repr__ +from pyaml.control.deviceaccess import DeviceAccess +from pyaml.control.readback_value import Quality, Value +from pyaml.validation import DynamicValidation, register_schema from .initializable_element import InitializableElement from .tango_pyaml_utils import to_float_or_none @@ -15,7 +18,7 @@ logger = logging.getLogger(__name__) -class ConfigModel(BaseModel): +class AttributeListConfig(BaseModel): """ Configuration model for a list of Tango attributes. @@ -34,25 +37,34 @@ class ConfigModel(BaseModel): unit: str = "" -class AttributeList(DeviceAccess, InitializableElement): +@register_schema +class AttributeList(DeviceAccess, InitializableElement, DynamicValidation): """ Handle a list of Tango attributes using Tango Groups. Parameters ---------- - cfg : ConfigModel - Configuration object with attribute list, name and unit. + attributes : list of str + List of Tango attribute paths. + name : str, optional + Group name. + unit : str, optional + Unit of the attributes. """ - def __init__(self, cfg: ConfigModel): + def __init__(self, attributes: list[str], name: str = "", unit: str = ""): super().__init__() - self._cfg = cfg + + self._attributes = attributes + self._name = name + self._unit = unit + self._tango_groups: dict[str, tango.Group] = {} self._attr_dev: dict[str, list[str]] = {} - for attribute in self._cfg.attributes: + for attribute in self._attributes: attribute_dev_name, attr_name = attribute.rsplit("/", 1) - if attr_name not in self._attr_dev.keys(): + if attr_name not in self._attr_dev: self._attr_dev[attr_name] = [] if attribute_dev_name not in self._attr_dev[attr_name]: self._attr_dev[attr_name].append(attribute_dev_name) @@ -60,7 +72,7 @@ def __init__(self, cfg: ConfigModel): def initialize(self): super().initialize() for attr_name, dev_list in self._attr_dev.items(): - self._tango_groups[attr_name] = tango.Group(self._cfg.name) + self._tango_groups[attr_name] = tango.Group(self._name) [self._tango_groups[attr_name].add(dev) for dev in dev_list] def name(self) -> str: @@ -72,7 +84,7 @@ def name(self) -> str: str Group name. """ - return self._cfg.name + return self._name def measure_name(self) -> str: """ @@ -83,7 +95,7 @@ def measure_name(self) -> str: str Group name. """ - return self._cfg.name + return self._name def get_tango_attributes(self) -> list[str]: """ @@ -94,7 +106,7 @@ def get_tango_attributes(self) -> list[str]: list[str] Tango attribute paths in configured order. """ - return self._cfg.attributes + return self._attributes def set(self, value: float): """ @@ -152,7 +164,7 @@ def get(self) -> array: result[val.dev_name + "/" + val.obj_name] = attr_value.w_value else: result[val.dev_name + "/" + val.obj_name] = None - return array([result[attribute] for attribute in self._cfg.attributes]) + return array([result[attribute] for attribute in self._attributes]) def readback(self) -> array: """ @@ -183,7 +195,7 @@ def readback(self) -> array: result[val.dev_name + "/" + val.obj_name] = value else: result[val.dev_name + "/" + val.obj_name] = None - list_res = [result[attribute] for attribute in self._cfg.attributes] + list_res = [result[attribute] for attribute in self._attributes] return array(list_res) def unit(self) -> str: @@ -195,17 +207,13 @@ def unit(self) -> str: str Unit string. """ - return self._cfg.unit + return self._unit def get_range(self) -> list[float]: attr_range: list[float] = [None, None] - if self._cfg.range is not None: - attr_range[0] = ( - self._cfg.range[0] if self._cfg.range[0] is not None else None - ) - attr_range[1] = ( - self._cfg.range[1] if self._cfg.range[1] is not None else None - ) + if self._range is not None: + attr_range[0] = self._range[0] if self._range[0] is not None else None + attr_range[1] = self._range[1] if self._range[1] is not None else None else: self._ensure_initialized() devices: list[tango.DeviceProxy] = [] @@ -226,9 +234,9 @@ def check_device_availability(self) -> bool: try: self._ensure_initialized() [group.ping() for group in self._tango_groups.values()] - except tango.DevFailed | pyaml.PyAMLException: + except (tango.DevFailed, pyaml.PyAMLException): available = False return available def __repr__(self): - return repr(self._cfg).replace("ConfigModel", self.__class__.__name__) + return __pyaml_repr__(self) diff --git a/tango/pyaml/attribute_list_read_only.py b/tango/pyaml/attribute_list_read_only.py index 32828b5..fec7d89 100644 --- a/tango/pyaml/attribute_list_read_only.py +++ b/tango/pyaml/attribute_list_read_only.py @@ -1,29 +1,39 @@ import logging import pyaml -from .attribute_list import AttributeList, ConfigModel as AttributeListConfigModel +from pyaml.validation import DynamicValidation, register_schema + +from .attribute_list import AttributeList, AttributeListConfig PYAMLCLASS: str = "AttributeListReadOnly" logger = logging.getLogger(__name__) -class ConfigModel(AttributeListConfigModel): - """Configuration model for a read-only Tango attribute list.""" +class AttributeListReadOnlyConfig(AttributeListConfig): ... -class AttributeListReadOnly(AttributeList): +@register_schema +class AttributeListReadOnly(AttributeList, DynamicValidation): """ Handle a list of Tango attributes using Tango Groups. Parameters ---------- - cfg : ConfigModel - Configuration object with attribute list, name and unit. + attributes : list of str + List of Tango attribute paths. + name : str, optional + Group name. + unit : str, optional + Unit of the attributes. """ - def __init__(self, cfg: ConfigModel): - super().__init__(cfg) + def __init__(self, attributes: list[str], name: str = "", unit: str = ""): + super().__init__(attributes, name, unit) + + self._attributes = attributes + self._name = name + self._unit = unit def set(self, value: float): """ diff --git a/tango/pyaml/attribute_read_only.py b/tango/pyaml/attribute_read_only.py index 614fe94..a8e6f65 100644 --- a/tango/pyaml/attribute_read_only.py +++ b/tango/pyaml/attribute_read_only.py @@ -1,29 +1,53 @@ import logging -from .attribute import Attribute, ConfigModel as AttributeConfigModel -from .tango_pyaml_utils import * +import pyaml +from pyaml.validation import DynamicValidation, register_schema + +from .attribute import Attribute, AttributeConfig PYAMLCLASS: str = "AttributeReadOnly" logger = logging.getLogger(__name__) -class ConfigModel(AttributeConfigModel): +class AttributeReadOnlyConfig(AttributeConfig): """Configuration model for a read-only Tango attribute.""" -class AttributeReadOnly(Attribute): +@register_schema +class AttributeReadOnly(Attribute, DynamicValidation): """ Read-only Tango attribute. Parameters ---------- - cfg : ConfigModel - Configuration model containing attribute path and unit. + attribute : str + Full path of the Tango attribute (e.g., 'my/ps/device/current'). + unit : str, optional + The unit of the attribute. + range : tuple(min, max), optional + Range of valid values. Use null for -โˆž or +โˆž. + index : int, optional + Zero-based index into a SPECTRUM attribute. When set, the instance + behaves as a read-only scalar view of one vector element; writes are + always rejected and a SPECTRUM data_format is enforced on init. """ - def __init__(self, cfg: ConfigModel): - super().__init__(cfg, False) + def __init__( + self, + attribute: str, + unit: str = "", + range: tuple[float | None, float | None] | None = None, + index: int | None = None, + ): + super().__init__( + attribute=attribute, unit=unit, range=range, index=index, writable=False + ) + + self._attribute = attribute + self._unit = unit + self._range = range + self._index = index def set(self, value: float): """ @@ -35,7 +59,7 @@ def set(self, value: float): Always raised because the attribute is read-only. """ raise pyaml.PyAMLException( - f"Tango attribute {self._cfg.attribute} is not writable." + f"Tango attribute {self._attribute} is not writable." ) def set_and_wait(self, value: float): @@ -48,7 +72,7 @@ def set_and_wait(self, value: float): Always raised because the attribute is read-only. """ raise pyaml.PyAMLException( - f"Tango attribute {self._cfg.attribute} is not writable." + f"Tango attribute {self._attribute} is not writable." ) def get(self) -> float: diff --git a/tango/pyaml/catalog.py b/tango/pyaml/catalog.py index 91d2e5c..4805952 100644 --- a/tango/pyaml/catalog.py +++ b/tango/pyaml/catalog.py @@ -21,4 +21,3 @@ def resolve(self, key: str) -> BaseModel: """ Return a configuration model for a DeviceAccess """ - pass diff --git a/tango/pyaml/controlsystem.py b/tango/pyaml/controlsystem.py index eb77043..729a8f7 100644 --- a/tango/pyaml/controlsystem.py +++ b/tango/pyaml/controlsystem.py @@ -1,21 +1,18 @@ import logging -from pydantic import BaseModel, ConfigDict +from pydantic import BaseModel from pyaml import PyAMLException +from pyaml.common.element import __pyaml_repr__ from pyaml.control.controlsystem import ControlSystem from pyaml.control.deviceaccess import DeviceAccess +from pyaml.validation import DynamicValidation, register_schema + from . import __version__ -from .attribute import Attribute, ConfigModel as AttributeConfigModel -from .attribute_list import AttributeList, ConfigModel as AttributeListConfigModel -from .attribute_list_read_only import ( - AttributeListReadOnly, - ConfigModel as AttributeListReadOnlyConfigModel, -) -from .attribute_read_only import ( - AttributeReadOnly, - ConfigModel as AttributeReadOnlyConfigModel, -) +from .attribute import Attribute, AttributeConfig +from .attribute_list import AttributeList, AttributeListConfig +from .attribute_list_read_only import AttributeListReadOnly, AttributeListReadOnlyConfig +from .attribute_read_only import AttributeReadOnly, AttributeReadOnlyConfig from .catalog import Catalog from .multi_attribute import MultiAttribute @@ -24,11 +21,12 @@ logger = logging.getLogger(__name__) -class ConfigModel(BaseModel): +@register_schema +class TangoControlSystem(ControlSystem, DynamicValidation): """ - Configuration model for a Tango Control System. + Tango-specific implementation of a Control System. - Attributes + Parameters ---------- name : str Name of the control system. @@ -46,43 +44,36 @@ class ConfigModel(BaseModel): Device timeout in milli seconds. """ - model_config = ConfigDict(arbitrary_types_allowed=True, extra="forbid") - - name: str - tango_host: str | None = None - catalog: Catalog | None = None - debug_level: str | int | None = None - lazy_devices: bool = True - timeout_ms: int = 3000 - - -class TangoControlSystem(ControlSystem): - """ - Tango-specific implementation of a Control System. - - Parameters - ---------- - cfg : ConfigModel - Configuration parameters including name, host and debug level. - """ - - def __init__(self, cfg: ConfigModel): + def __init__( + self, + name: str, + tango_host: str | None = None, + catalog: Catalog | None = None, + debug_level: str | int | None = None, + lazy_devices: bool = True, + timeout_ms: int = 3000, + ): super().__init__() - self._cfg = cfg + self._name = name + self._tango_host = tango_host + self._catalog = catalog + self._debug_level = debug_level + self._lazy_devices = lazy_devices + self._timeout_ms = timeout_ms self.__devices = {} # Dict containing all attached DeviceAccess - if self._cfg.debug_level: - if isinstance(self._cfg.debug_level, int): - log_level = self._cfg.debug_level + if self._debug_level: + if isinstance(self._debug_level, int): + log_level = self._debug_level else: - log_level = getattr(logging, self._cfg.debug_level, logging.WARNING) + log_level = getattr(logging, self._debug_level, logging.WARNING) logger.parent.setLevel(log_level) logger.setLevel(log_level) logger.log( logging.WARNING, - f"PyAML Tango control system binding ({__version__}) initialized with name '{self._cfg.name}'" - f" and TANGO_HOST={self._cfg.tango_host}", + f"PyAML Tango control system binding ({__version__}) initialized with name '{self._name}'" + f" and TANGO_HOST={self._tango_host}", ) def attach_array(self, devs: list[DeviceAccess]) -> list[DeviceAccess]: @@ -129,8 +120,8 @@ def get_device_access(self, ref: str | BaseModel | None) -> DeviceAccess | None: if isinstance(ref, DeviceAccess): raise PyAMLException( - "TangoControlSystem.get_device_access() expects a catalog key, Tango " - "ConfigModel, or None. Use attach() for already constructed " + "TangoControlSystem.get_device_access() expects a catalog key " + "or None. Use attach() for already constructed " "DeviceAccess objects." ) @@ -155,17 +146,19 @@ def get_device_access(self, ref: str | BaseModel | None) -> DeviceAccess | None: device = resolve(ref, self) return self._attach([device])[0] - if isinstance(ref, AttributeReadOnlyConfigModel): - return self._attach([AttributeReadOnly(ref)])[0] + if isinstance(ref, AttributeReadOnlyConfig): + return self._attach([AttributeReadOnly(**ref.model_dump())])[0] - if isinstance(ref, AttributeConfigModel): - return self._attach([Attribute(ref)])[0] + if isinstance(ref, AttributeConfig): + return self._attach([Attribute(**ref.model_dump())])[0] - if isinstance(ref, AttributeListReadOnlyConfigModel): - return AttributeListReadOnly(self._attach_attribute_list_config(ref)) + if isinstance(ref, AttributeListReadOnlyConfig): + cfg = self._attach_attribute_list_config(ref) + return AttributeListReadOnly(**cfg.model_dump()) - if isinstance(ref, AttributeListConfigModel): - return AttributeList(self._attach_attribute_list_config(ref)) + if isinstance(ref, AttributeListConfig): + cfg = self._attach_attribute_list_config(ref) + return AttributeList(**cfg.model_dump()) if isinstance(ref, BaseModel): raise PyAMLException( @@ -175,12 +168,12 @@ def get_device_access(self, ref: str | BaseModel | None) -> DeviceAccess | None: raise PyAMLException( f"TangoControlSystem.get_device_access() cannot resolve references of type " - f"{type(ref).__name__}; expected str, Tango ConfigModel, or None." + f"{type(ref).__name__}; expected str or None." ) def _attach_attribute_list_config( - self, cfg: AttributeListConfigModel - ) -> AttributeListConfigModel: + self, cfg: AttributeListConfig + ) -> AttributeListConfig: tango_host = self.get_tango_host() if not tango_host: return cfg @@ -202,7 +195,7 @@ def name(self) -> str: str Name of the control system. """ - return self._cfg.name + return self._name def get_tango_host(self) -> str | None: """ @@ -213,7 +206,7 @@ def get_tango_host(self) -> str | None: str | None Tango host URL, or ``None`` when unconfigured. """ - return self._cfg.tango_host + return self._tango_host def get_aggregator(self) -> MultiAttribute | None: """Returns a new empty DeviceAccessList. If None is returned serialized readings/writtings are performed""" @@ -250,7 +243,7 @@ def get_catalog(self) -> Catalog | None: Catalog The catalog """ - return self._cfg.catalog + return self._catalog def __repr__(self): - return repr(self._cfg).replace("ConfigModel", self.__class__.__name__) + return __pyaml_repr__(self) diff --git a/tango/pyaml/device_factory.py b/tango/pyaml/device_factory.py index abfa5ef..167fb3f 100644 --- a/tango/pyaml/device_factory.py +++ b/tango/pyaml/device_factory.py @@ -1,5 +1,6 @@ -from threading import Lock from collections import defaultdict +from threading import Lock + import tango diff --git a/tango/pyaml/initializable_element.py b/tango/pyaml/initializable_element.py index 5c2947b..4a9b43e 100644 --- a/tango/pyaml/initializable_element.py +++ b/tango/pyaml/initializable_element.py @@ -19,4 +19,3 @@ def is_initialized(self) -> bool: def _ensure_initialized(self): if not self.is_initialized(): self.initialize() - pass diff --git a/tango/pyaml/multi_attribute.py b/tango/pyaml/multi_attribute.py index fdd27db..c0a4b02 100644 --- a/tango/pyaml/multi_attribute.py +++ b/tango/pyaml/multi_attribute.py @@ -1,15 +1,16 @@ import logging -from typing import Tuple, Optional import numpy as np -import pyaml from numpy import typing as npt -from pyaml.control.deviceaccess import DeviceAccess from pydantic import BaseModel +import pyaml +from pyaml.common.element import __pyaml_repr__ +from pyaml.control.deviceaccess import DeviceAccess from pyaml.control.deviceaccesslist import DeviceAccessList +from pyaml.validation import DynamicValidation, register_schema -from .attribute import Attribute, ConfigModel as AttrConfig +from .attribute import Attribute, AttributeConfig from .device_factory import DeviceFactory PYAMLCLASS: str = "MultiAttribute" @@ -17,7 +18,7 @@ logger = logging.getLogger(__name__) -class ConfigModel(BaseModel): +class MultiAttributeConfig(BaseModel): """ Configuration model for a list of Tango attributes. @@ -36,21 +37,32 @@ class ConfigModel(BaseModel): attributes: list[str] = [] name: str = "" unit: str = "" - range: Optional[Tuple[Optional[float], Optional[float]]] = None - - -class MultiAttribute(DeviceAccessList): - def __init__(self, cfg: ConfigModel = None): + range: tuple[float | None, float | None] | None = None + + +@register_schema +class MultiAttribute(DeviceAccessList, DynamicValidation): + def __init__( + self, + attributes: list[str] | None = None, + name: str = "", + unit: str = "", + range: tuple[float | None, float | None] | None = None, + ): super().__init__() + + self._attributes = attributes + self._name = name + self._unit = unit + self._range = range self._items: list[Attribute] = [] - self._cfg = cfg - if self._cfg: - for attribute in self._cfg.attributes: - attr_config = AttrConfig( - attribute=attribute, unit=self._cfg.unit, range=self._cfg.range - ) - attr = Attribute(attr_config) - self._items.append(attr) + + for attribute in self._attributes: + attr_config = AttributeConfig( + attribute=attribute, unit=self._unit, range=self._range + ) + attr = Attribute(**attr_config.model_dump()) + self._items.append(attr) def len(self) -> int: return len(self._items) @@ -60,7 +72,7 @@ def get_device_at(self, index: int) -> DeviceAccess: def add_devices(self, devices: DeviceAccess | list[DeviceAccess]): if isinstance(devices, list): - if any([not isinstance(device, Attribute) for device in devices]): + if any(not isinstance(device, Attribute) for device in devices): raise pyaml.PyAMLException( "All devices must be instances of Attribute (tango.pyaml.attribute)." ) @@ -153,10 +165,7 @@ def check_device_availability(self) -> bool: return available def unit(self) -> str: - if self._cfg: - return self._cfg.unit - else: - return "" + return self._unit def __repr__(self): - return repr(self._cfg).replace("ConfigModel", self.__class__.__name__) + return __pyaml_repr__(self) diff --git a/tango/pyaml/static_catalog.py b/tango/pyaml/static_catalog.py index 2161227..70d8b19 100644 --- a/tango/pyaml/static_catalog.py +++ b/tango/pyaml/static_catalog.py @@ -1,7 +1,6 @@ -from pydantic import ConfigDict, BaseModel - from pyaml import PyAMLException from pyaml.control.deviceaccess import DeviceAccess +from pyaml.validation import DynamicValidation, register_schema from .catalog import Catalog from .static_catalog_entry import StaticCatalogEntry @@ -9,25 +8,8 @@ PYAMLCLASS = "StaticCatalog" -class ConfigModel(BaseModel): - """ - Configuration model for a static catalog. - - Attributes - ---------- - name : str - Catalog identifier. - entries : list[StaticCatalogEntry] - Explicit list of key-to-device mappings. Must contain at least one - entry, and keys must be unique within the catalog. - """ - - model_config = ConfigDict(arbitrary_types_allowed=True, extra="forbid") - - entries: list[StaticCatalogEntry] - - -class StaticCatalog(Catalog): +@register_schema +class StaticCatalog(Catalog, DynamicValidation): """ Catalog backed by a fixed list of key-to-device mappings. @@ -37,8 +19,11 @@ class StaticCatalog(Catalog): Parameters ---------- - cfg : ConfigModel - Configuration containing the catalog name and its entries. + name : str + Catalog identifier. + entries : list[StaticCatalogEntry] + Explicit list of key-to-device mappings. Must contain at least one + entry, and keys must be unique within the catalog. Raises ------ @@ -46,15 +31,16 @@ class StaticCatalog(Catalog): If ``cfg.entries`` is empty or contains duplicate keys. """ - def __init__(self, cfg: ConfigModel): + def __init__(self, entries: list[StaticCatalogEntry]): super().__init__() - self._cfg = cfg - if len(cfg.entries) == 0: + + self._entries = entries + if len(self._entries) == 0: raise PyAMLException( "StaticCatalog.entries must contain at least one entry" ) self._refs: dict[str, DeviceAccess] = {} - for entry in cfg.entries: + for entry in self._entries: key = entry.get_key() if key in self._refs: raise PyAMLException( @@ -87,6 +73,4 @@ def resolve(self, key: str, control_system: object | None = None) -> DeviceAcces try: return self._refs[key] except KeyError as exc: - raise PyAMLException( - f"Catalog cannot resolve key '{key}'" - ) from exc + raise PyAMLException(f"Catalog cannot resolve key '{key}'") from exc diff --git a/tango/pyaml/static_catalog_entry.py b/tango/pyaml/static_catalog_entry.py index 600fcb1..a62509b 100644 --- a/tango/pyaml/static_catalog_entry.py +++ b/tango/pyaml/static_catalog_entry.py @@ -1,15 +1,15 @@ -from pydantic import BaseModel, ConfigDict - from pyaml.control.deviceaccess import DeviceAccess +from pyaml.validation import DynamicValidation, register_schema PYAMLCLASS = "StaticCatalogEntry" -class ConfigModel(BaseModel): +@register_schema +class StaticCatalogEntry(DynamicValidation): """ - Configuration model for a static catalog entry. + A single key-to-device mapping in a static catalog. - Attributes + Parameters ---------- key : str Catalog key used to look up the device. @@ -17,29 +17,14 @@ class ConfigModel(BaseModel): Device access object returned when the key is resolved. """ - model_config = ConfigDict(arbitrary_types_allowed=True, extra="forbid") - - key: str - device: DeviceAccess - - -class StaticCatalogEntry: - """ - A single key-to-device mapping in a static catalog. - - Parameters - ---------- - cfg : ConfigModel - Configuration containing the key and device. - """ - - def __init__(self, cfg: ConfigModel): - self._cfg = cfg + def __init__(self, key: str, device: DeviceAccess): + self.key = key + self.device = device def get_key(self) -> str: """Return the catalog key for this entry.""" - return self._cfg.key + return self.key def get_device(self) -> DeviceAccess: """Return the device access object associated with this entry.""" - return self._cfg.device + return self.device diff --git a/tango/pyaml/tango_catalog.py b/tango/pyaml/tango_catalog.py index 157c37d..5e4d9df 100644 --- a/tango/pyaml/tango_catalog.py +++ b/tango/pyaml/tango_catalog.py @@ -1,10 +1,11 @@ -import tango -import pyaml +from typing import ClassVar -from pydantic import ConfigDict, BaseModel +import pyaml +import tango from pyaml.control.deviceaccess import DeviceAccess +from pyaml.validation import DynamicValidation, register_schema -from .attribute import Attribute, ConfigModel as AttributeConfigModel +from .attribute import Attribute, AttributeConfig from .attribute_read_only import AttributeReadOnly from .catalog import Catalog from .tango_pyaml_utils import tango_to_PyAMLException, to_float_or_none @@ -12,41 +13,29 @@ PYAMLCLASS = "TangoCatalog" -class ConfigModel(BaseModel): - """ - Configuration model for a Tango catalog. - - Attributes - ---------- - name : str - Catalog identifier. - disconnected : bool - If true, resolve Tango attribute names without querying Tango. - """ - - model_config = ConfigDict(arbitrary_types_allowed=True, extra="forbid") - - disconnected: bool = False - - -class TangoCatalog(Catalog): +@register_schema +class TangoCatalog(Catalog, DynamicValidation): """ Catalog resolving keys that are direct Tango attribute references. Keys can be plain Tango attribute paths (``domain/family/member/attribute``) or indexed references into a SPECTRUM attribute (``domain/family/member/attribute@index``). + + disconnected : bool + If true, resolve Tango attribute names without querying Tango. """ - _WRITABLE_TYPES = { + _WRITABLE_TYPES: ClassVar[set[tango.AttrWriteType]] = { tango.AttrWriteType.READ_WRITE, tango.AttrWriteType.WRITE, tango.AttrWriteType.READ_WITH_WRITE, } - def __init__(self, cfg: ConfigModel): + def __init__(self, disconnected: bool = False): super().__init__() - self._cfg = cfg + + self._disconnected = disconnected # Resolved DeviceAccess objects are bound to one control-system context # because metadata lookup depends on that control system's Tango host. self._refs: dict[tuple[int, str], DeviceAccess] = {} @@ -114,7 +103,7 @@ def resolve(self, key: str, control_system: object | None = None) -> DeviceAcces return self._refs[cache_key] def is_disconnected(self) -> bool: - return self._cfg.disconnected + return self._disconnected def get_data_format( self, key: str, control_system: object | None = None @@ -149,7 +138,9 @@ def _validate_control_system(self, control_system: object | None, key: str) -> N ) if not isinstance(control_system, TangoControlSystem): - raise pyaml.PyAMLException("Tango catalog can only resolve through TangoControlSystem") + raise pyaml.PyAMLException( + "Tango catalog can only resolve through TangoControlSystem" + ) def _parse_key(self, key: str) -> tuple[str, int | None]: """ @@ -166,14 +157,18 @@ def _parse_key(self, key: str) -> tuple[str, int | None]: not a valid integer. """ if not isinstance(key, str): - raise pyaml.PyAMLException(f"Tango catalog expects string keys, got {type(key).__name__}") + raise pyaml.PyAMLException( + f"Tango catalog expects string keys, got {type(key).__name__}" + ) if "@" in key: attr_path, idx_str = key.rsplit("@", 1) try: index = int(idx_str) except ValueError as exc: - raise pyaml.PyAMLException(f"Tango catalog invalid index '{idx_str}' in key '{key}'.") from exc + raise pyaml.PyAMLException( + f"Tango catalog invalid index '{idx_str}' in key '{key}'." + ) from exc else: attr_path = key index = None @@ -194,16 +189,14 @@ def _build_disconnected_attribute( # In disconnected mode, keep all metadata local. In particular, setting # range avoids Attribute.get_range() from lazily querying Tango later. self._data_formats[cache_key] = tango.AttrDataFormat.FMT_UNKNOWN - return Attribute(AttributeConfigModel(attribute=key, range=(None, None))) + return Attribute(attribute=key, range=(None, None)) def _build_disconnected_indexed( self, cache_key: tuple[int, str], attr_path: str, index: int ) -> DeviceAccess: # Cannot verify SPECTRUM in disconnected mode; store FMT_UNKNOWN. self._data_formats[cache_key] = tango.AttrDataFormat.FMT_UNKNOWN - return Attribute( - AttributeConfigModel(attribute=attr_path, index=index, range=(None, None)) - ) + return Attribute(attribute=attr_path, index=index, range=(None, None)) def _build_connected_attribute( self, cache_key: tuple[int, str], control_system: object, key: str @@ -215,17 +208,19 @@ def _build_connected_attribute( attr_config = tango.AttributeProxy(tango_attr_name).get_config() except tango.DevFailed as df: pyaml_exception = tango_to_PyAMLException(df) - raise pyaml.PyAMLException(f"Tango catalog cannot resolve '{key}': {pyaml_exception}") from df + raise pyaml.PyAMLException( + f"Tango catalog cannot resolve '{key}': {pyaml_exception}" + ) from df unit, attr_range, data_format, writable = self._read_config_metadata( attr_config, key ) self._data_formats[cache_key] = data_format - cfg = AttributeConfigModel(attribute=key, unit=unit, range=attr_range) + cfg = AttributeConfig(attribute=key, unit=unit, range=attr_range) if writable in self._WRITABLE_TYPES: - return Attribute(cfg) - return AttributeReadOnly(cfg) + return Attribute(**cfg.model_dump()) + return AttributeReadOnly(**cfg.model_dump()) def _build_connected_indexed( self, @@ -262,13 +257,13 @@ def _build_connected_indexed( ) self._data_formats[cache_key] = tango.AttrDataFormat.SPECTRUM - cfg = AttributeConfigModel( + cfg = AttributeConfig( attribute=attr_path, index=index, unit=unit, range=attr_range ) if writable in self._WRITABLE_TYPES: - return Attribute(cfg) - return AttributeReadOnly(cfg) + return Attribute(**cfg.model_dump()) + return AttributeReadOnly(**cfg.model_dump()) def _read_config_metadata( self, attr_config, key: str diff --git a/tango/pyaml/tango_pyaml_utils.py b/tango/pyaml/tango_pyaml_utils.py index fb48686..be20027 100644 --- a/tango/pyaml/tango_pyaml_utils.py +++ b/tango/pyaml/tango_pyaml_utils.py @@ -1,5 +1,5 @@ -import tango import pyaml +import tango def to_float_or_none(s): diff --git a/tests/conftest.py b/tests/conftest.py index e12bd02..309cdfc 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -1,11 +1,11 @@ import pytest import yaml -from tango.pyaml.attribute_list import ConfigModel as GrpCM -from tango.pyaml.attribute import ConfigModel as AttrCM -from tango.pyaml.multi_attribute import ConfigModel as MultiAttrCM -from tango.pyaml.controlsystem import ConfigModel as CsCM, TangoControlSystem +from tango.pyaml.attribute import AttributeConfig as AttrCM +from tango.pyaml.attribute_list import AttributeListConfig as GrpCM +from tango.pyaml.controlsystem import TangoControlSystem from tango.pyaml.device_factory import DeviceFactory +from tango.pyaml.multi_attribute import MultiAttributeConfig as MultiAttrCM @pytest.fixture(autouse=True) @@ -98,7 +98,7 @@ def config_tango_cs(): lazy_devices: false """ cfg_dict = yaml.safe_load(conf) - return CsCM(**cfg_dict) + return cfg_dict @pytest.fixture @@ -109,7 +109,7 @@ def config_tango_cs_lazy_default(): debug_level: INFO """ cfg_dict = yaml.safe_load(conf) - return CsCM(**cfg_dict) + return cfg_dict @pytest.fixture @@ -120,4 +120,4 @@ def config_tango_cs_false(): debug_level: nope """ cfg_dict = yaml.safe_load(conf) - return CsCM(**cfg_dict) + return cfg_dict diff --git a/tests/mocked_device_proxy.py b/tests/mocked_device_proxy.py index f6d7f7f..b16fbd6 100644 --- a/tests/mocked_device_proxy.py +++ b/tests/mocked_device_proxy.py @@ -1,7 +1,9 @@ -import tango -import numpy as np from unittest.mock import MagicMock +import numpy as np + +import tango + class MockedAttributeInfoEx: def __init__( @@ -84,7 +86,7 @@ def command_inout_reply(self, idx, timeout=None): return val def read_attribute(self, attr_name: str): - if attr_name not in self.values.keys(): + if attr_name not in self.values: return MockedDeviceAttribute(attr_name, None) return self.values[attr_name] diff --git a/tests/mocked_group.py b/tests/mocked_group.py index e26d2d2..70460d8 100644 --- a/tests/mocked_group.py +++ b/tests/mocked_group.py @@ -1,4 +1,4 @@ -from .mocked_device_proxy import * +from .mocked_device_proxy import MagicMock, MockedDeviceAttribute, MockedDeviceProxy class MockedGroupReply: @@ -53,7 +53,7 @@ def command_inout(self, command_name, *args, **kwargs): try: idx = dev.command_inout_asynch(command_name) replies_id[name] = idx - except Exception as e: + except Exception as e: # noqa: BLE001 replies.append(MockedGroupCmdReply(name, command_name, None, e)) for name, idx in replies_id.items(): dev = self.devices[name] @@ -68,7 +68,7 @@ def read_attribute(self, attr_name) -> list[MockedGroupAttrReply]: try: idx = dev.read_attribute_asynch(attr_name) replies_id[name] = idx - except Exception as e: + except Exception as e: # noqa: BLE001 replies.append(MockedGroupAttrReply(name, attr_name, None, e)) for name, idx in replies_id.items(): dev = self.devices[name] @@ -94,7 +94,7 @@ def write_attribute(self, attr_name, value): try: dev.write_attribute(attr_name, value) replies.append(MockedGroupReply(name, attr_name)) - except Exception as e: + except Exception as e: # noqa: BLE001 replies.append(MockedGroupReply(name, attr_name, e)) return replies diff --git a/tests/test_attribute.py b/tests/test_attribute.py index a983a34..f387327 100644 --- a/tests/test_attribute.py +++ b/tests/test_attribute.py @@ -1,14 +1,19 @@ +from unittest.mock import patch + import pyaml.control.readback_value import pytest +from tango.pyaml.attribute import Attribute +from tango.pyaml.attribute_list import AttributeList from tango.pyaml.attribute_read_only import AttributeReadOnly from .mocked_control_system_initialized import MockedControlSystemInitialized -from .mocked_device_proxy import * +from .mocked_device_proxy import ( + MockedAttributeInfoEx, + MockedDeviceProxy, + tango, +) from .mocked_group import MockedGroup -from unittest.mock import patch -from tango.pyaml.attribute import Attribute -from tango.pyaml.attribute_list import AttributeList class MockedReadExceptDeviceProxy(MockedDeviceProxy): @@ -33,7 +38,7 @@ def test_attribute_get_set(self, config): new=MockedControlSystemInitialized, ), ): - attr = Attribute(config) + attr = Attribute(**config.model_dump()) attr.set_and_wait(42.0) assert attr.get() == 42.0 assert attr.readback() == 42.0 @@ -53,7 +58,7 @@ def test_attribute_except(self, config): new=MockedControlSystemInitialized, ), ): - attr = Attribute(config) + attr = Attribute(**config.model_dump()) with pytest.raises(pyaml.PyAMLException) as exc: attr.readback() assert exc is not None @@ -70,13 +75,13 @@ def test_attribute_read_only(self, config): expected_message = ( "Tango attribute sys/tg_test/1/float_scalar is not writable." ) - attr1 = Attribute(config) + attr1 = Attribute(**config.model_dump()) with pytest.raises(pyaml.PyAMLException) as exc: attr1.get() assert exc.value.message == expected_message # Read-only attributes cannot be sets. - attr = AttributeReadOnly(config) + attr = AttributeReadOnly(**config.model_dump()) with pytest.raises(pyaml.PyAMLException) as exc2: attr.set(10) assert exc2.value.message == expected_message @@ -89,7 +94,7 @@ def test_group_read_write(self, config_group): new=MockedControlSystemInitialized, ), ): - attr_list = AttributeList(config_group) + attr_list = AttributeList(**config_group.model_dump()) attr_list.set_and_wait(10) vals = attr_list.readback() for val in vals: @@ -103,6 +108,6 @@ def test_unique_device(self, config): new=MockedControlSystemInitialized, ), ): - attr1 = Attribute(config) - attr2 = Attribute(config) + attr1 = Attribute(**config.model_dump()) + attr2 = Attribute(**config.model_dump()) assert attr1._attribute_dev is attr2._attribute_dev diff --git a/tests/test_attribute_indexed.py b/tests/test_attribute_indexed.py index 3d36aa9..672b154 100644 --- a/tests/test_attribute_indexed.py +++ b/tests/test_attribute_indexed.py @@ -1,14 +1,18 @@ -import numpy as np -import pytest -import tango from unittest.mock import patch +import numpy as np import pyaml +import pytest -from tango.pyaml.attribute import Attribute, ConfigModel +import tango +from tango.pyaml.attribute import Attribute, AttributeConfig from tango.pyaml.attribute_read_only import AttributeReadOnly -from .mocked_device_proxy import MockedAttributeInfoEx, MockedDeviceProxy, MockedDeviceAttribute +from .mocked_device_proxy import ( + MockedAttributeInfoEx, + MockedDeviceAttribute, + MockedDeviceProxy, +) SPECTRUM_ARRAY = np.array([10.0, 20.0, 30.0]) @@ -57,67 +61,71 @@ def attribute_query(self, name): def test_attribute_indexed_get_returns_w_value_at_index(): - cfg = ConfigModel(attribute="domain/family/member/position", index=1, unit="mm") + cfg = AttributeConfig(attribute="domain/family/member/position", index=1, unit="mm") with patch("tango.DeviceProxy", new=MockedSpectrumDeviceProxy): - attr = Attribute(cfg) + attr = Attribute(**cfg.model_dump()) assert attr.get() == SPECTRUM_ARRAY[1] def test_attribute_indexed_readback_returns_value_at_index(): - cfg = ConfigModel(attribute="domain/family/member/position", index=0, unit="mm") + cfg = AttributeConfig(attribute="domain/family/member/position", index=0, unit="mm") with patch("tango.DeviceProxy", new=MockedSpectrumDeviceProxy): - attr = Attribute(cfg) + attr = Attribute(**cfg.model_dump()) rb = attr.readback() assert rb.value == SPECTRUM_ARRAY[0] def test_attribute_indexed_set_raises(): - cfg = ConfigModel(attribute="domain/family/member/position", index=0) + cfg = AttributeConfig(attribute="domain/family/member/position", index=0) with patch("tango.DeviceProxy", new=MockedSpectrumDeviceProxy): - attr = Attribute(cfg) - with pytest.raises(pyaml.PyAMLException, match="does not support individual element writes"): + attr = Attribute(**cfg.model_dump()) + with pytest.raises( + pyaml.PyAMLException, match="does not support individual element writes" + ): attr.set(99.0) def test_attribute_indexed_set_and_wait_raises(): - cfg = ConfigModel(attribute="domain/family/member/position", index=0) + cfg = AttributeConfig(attribute="domain/family/member/position", index=0) with patch("tango.DeviceProxy", new=MockedSpectrumDeviceProxy): - attr = Attribute(cfg) - with pytest.raises(pyaml.PyAMLException, match="does not support individual element writes"): + attr = Attribute(**cfg.model_dump()) + with pytest.raises( + pyaml.PyAMLException, match="does not support individual element writes" + ): attr.set_and_wait(99.0) def test_attribute_indexed_name_includes_index(): - cfg = ConfigModel(attribute="domain/family/member/position", index=2) - attr = Attribute(cfg) + cfg = AttributeConfig(attribute="domain/family/member/position", index=2) + attr = Attribute(**cfg.model_dump()) assert attr.name() == "domain/family/member/position[2]" def test_attribute_indexed_measure_name_includes_index(): - cfg = ConfigModel(attribute="domain/family/member/position", index=2) - attr = Attribute(cfg) + cfg = AttributeConfig(attribute="domain/family/member/position", index=2) + attr = Attribute(**cfg.model_dump()) assert attr.measure_name() == "position[2]" def test_attribute_indexed_unit(): - cfg = ConfigModel(attribute="domain/family/member/position", index=0, unit="mm") - attr = Attribute(cfg) + cfg = AttributeConfig(attribute="domain/family/member/position", index=0, unit="mm") + attr = Attribute(**cfg.model_dump()) assert attr.unit() == "mm" def test_attribute_indexed_raises_when_not_spectrum(): - cfg = ConfigModel(attribute="domain/family/member/current", index=0) + cfg = AttributeConfig(attribute="domain/family/member/current", index=0) with patch("tango.DeviceProxy", new=MockedScalarDeviceProxy): - attr = Attribute(cfg) + attr = Attribute(**cfg.model_dump()) with pytest.raises(pyaml.PyAMLException, match="not a SPECTRUM"): attr.get() def test_attribute_indexed_range_from_config(): - cfg = ConfigModel( + cfg = AttributeConfig( attribute="domain/family/member/position", index=0, unit="mm", range=(-5.0, 5.0) ) - attr = Attribute(cfg) + attr = Attribute(**cfg.model_dump()) assert attr.get_range() == [-5.0, 5.0] @@ -125,29 +133,29 @@ def test_attribute_indexed_range_from_config(): def test_attribute_indexed_read_only_get_returns_measured_value(): - cfg = ConfigModel(attribute="domain/family/member/position", index=2, unit="mm") + cfg = AttributeConfig(attribute="domain/family/member/position", index=2, unit="mm") with patch("tango.DeviceProxy", new=MockedSpectrumRODeviceProxy): - attr = AttributeReadOnly(cfg) + attr = AttributeReadOnly(**cfg.model_dump()) assert attr.get() == SPECTRUM_ARRAY[2] def test_attribute_indexed_read_only_readback_returns_value_at_index(): - cfg = ConfigModel(attribute="domain/family/member/position", index=1, unit="mm") + cfg = AttributeConfig(attribute="domain/family/member/position", index=1, unit="mm") with patch("tango.DeviceProxy", new=MockedSpectrumRODeviceProxy): - attr = AttributeReadOnly(cfg) + attr = AttributeReadOnly(**cfg.model_dump()) assert attr.readback().value == SPECTRUM_ARRAY[1] def test_attribute_indexed_read_only_set_raises(): - cfg = ConfigModel(attribute="domain/family/member/position", index=0) + cfg = AttributeConfig(attribute="domain/family/member/position", index=0) with patch("tango.DeviceProxy", new=MockedSpectrumRODeviceProxy): - attr = AttributeReadOnly(cfg) + attr = AttributeReadOnly(**cfg.model_dump()) with pytest.raises(pyaml.PyAMLException): attr.set(1.0) def test_attribute_indexed_read_only_get_equals_readback(): - cfg = ConfigModel(attribute="domain/family/member/position", index=0, unit="mm") + cfg = AttributeConfig(attribute="domain/family/member/position", index=0, unit="mm") with patch("tango.DeviceProxy", new=MockedSpectrumRODeviceProxy): - attr = AttributeReadOnly(cfg) + attr = AttributeReadOnly(**cfg.model_dump()) assert attr.get() == attr.readback().value diff --git a/tests/test_attribute_range.py b/tests/test_attribute_range.py index a4d9119..abee187 100644 --- a/tests/test_attribute_range.py +++ b/tests/test_attribute_range.py @@ -1,8 +1,13 @@ -from .mocked_device_proxy import * - from unittest.mock import patch + from tango.pyaml.attribute import Attribute + from .mocked_control_system_initialized import MockedControlSystemInitialized +from .mocked_device_proxy import ( + MockedAttributeInfoEx, + MockedDeviceProxy, + tango, +) class MockedMinMaxAttrDeviceProxy(MockedDeviceProxy): @@ -29,7 +34,7 @@ def test_attribute_range_by_conf(config_range): new=MockedControlSystemInitialized, ), ): - attr = Attribute(config_range) + attr = Attribute(**config_range.model_dump()) attr_range = attr.get_range() assert attr_range is not None @@ -45,12 +50,12 @@ def test_attribute_range_by_conf_with_null(config_range_with_null): new=MockedControlSystemInitialized, ), ): - attr = Attribute(config_range_with_null) + attr = Attribute(**config_range_with_null.model_dump()) attr_range = attr.get_range() assert attr_range is not None assert len(attr_range) == 2 - assert attr_range[0] == 0 and attr_range[1] == None + assert attr_range[0] == 0 and attr_range[1] is None def test_attribute_range_by_device(config): @@ -61,7 +66,7 @@ def test_attribute_range_by_device(config): new=MockedControlSystemInitialized, ), ): - attr = Attribute(config) + attr = Attribute(**config.model_dump()) attr_range = attr.get_range() assert attr_range is not None @@ -77,9 +82,9 @@ def test_attribute_range_by_device_min_only(config): new=MockedControlSystemInitialized, ), ): - attr = Attribute(config) + attr = Attribute(**config.model_dump()) attr_range = attr.get_range() assert attr_range is not None assert len(attr_range) == 2 - assert attr_range[0] == -10 and attr_range[1] == None + assert attr_range[0] == -10 and attr_range[1] is None diff --git a/tests/test_controlsystem.py b/tests/test_controlsystem.py index c2d2d29..abf8815 100644 --- a/tests/test_controlsystem.py +++ b/tests/test_controlsystem.py @@ -1,37 +1,32 @@ import logging +from unittest.mock import patch import pyaml import pytest -from tango.pyaml.static_catalog import ConfigModel as StaticCatalogConfigModel -from tango.pyaml.static_catalog import StaticCatalog -from tango.pyaml.static_catalog_entry import ( - ConfigModel as StaticCatalogEntryConfigModel, + +from tango.pyaml import __version__ +from tango.pyaml.attribute import Attribute, AttributeConfig +from tango.pyaml.attribute_list import AttributeList, AttributeListConfig +from tango.pyaml.attribute_list_read_only import ( + AttributeListReadOnly, + AttributeListReadOnlyConfig, ) +from tango.pyaml.attribute_read_only import AttributeReadOnly, AttributeReadOnlyConfig +from tango.pyaml.controlsystem import TangoControlSystem +from tango.pyaml.static_catalog import StaticCatalog from tango.pyaml.static_catalog_entry import StaticCatalogEntry -from tango.pyaml.controlsystem import ConfigModel, TangoControlSystem from .mocked_device_proxy import MockedDeviceProxy -from unittest.mock import patch -from tango.pyaml.attribute_list import AttributeList -from tango.pyaml.attribute_list import ConfigModel as AttributeListConfigModel -from tango.pyaml.attribute_list_read_only import AttributeListReadOnly -from tango.pyaml.attribute_list_read_only import ( - ConfigModel as AttributeListReadOnlyConfigModel, -) -from tango.pyaml.attribute import Attribute, ConfigModel as AttributeConfigModel -from tango.pyaml.attribute_read_only import AttributeReadOnly -from tango.pyaml.attribute_read_only import ConfigModel as AttributeReadOnlyConfigModel -from tango.pyaml import __version__ def test_init_cs(caplog, config_tango_cs): # Capture logs with caplog.at_level(logging.INFO): - TangoControlSystem(config_tango_cs) + TangoControlSystem(**config_tango_cs) expected_message = ( - f"PyAML Tango control system binding ({__version__}) initialized with name '{config_tango_cs.name}'" - f" and TANGO_HOST={config_tango_cs.tango_host}" + f"PyAML Tango control system binding ({__version__}) initialized with name '{config_tango_cs['name']}'" + f" and TANGO_HOST={config_tango_cs['tango_host']}" ) # Check that the INFO init message was actually logged with correct values @@ -40,7 +35,7 @@ def test_init_cs(caplog, config_tango_cs): def test_laziness_init_cs_attribute(config_tango_cs_lazy_default, config): with patch("tango.DeviceProxy", side_effect=MockedDeviceProxy) as mock_ctor: - attr = Attribute(config) + attr = Attribute(**config.model_dump()) mock_ctor.assert_not_called() attr.set_and_wait(42.0) mock_ctor.assert_called_once() @@ -50,27 +45,19 @@ def test_laziness_init_cs_attribute(config_tango_cs_lazy_default, config): def test_catalog_can_be_configured_and_resolved(): - device = AttributeReadOnly( - AttributeConfigModel(attribute="sys/tg_test/1/float_scalar", unit="A") - ) + device = AttributeReadOnly(attribute="sys/tg_test/1/float_scalar", unit="A") catalog = StaticCatalog( - StaticCatalogConfigModel( - entries=[ - StaticCatalogEntry( - StaticCatalogEntryConfigModel( - key="BPM_C01-01/x", - device=device, - ) - ) - ], - ) + entries=[ + StaticCatalogEntry( + key="BPM_C01-01/x", + device=device, + ) + ], ) cs = TangoControlSystem( - ConfigModel( - name="test_tango_cs", - tango_host="tangodb:10000", - catalog=catalog, - ) + name="test_tango_cs", + tango_host="tangodb:10000", + catalog=catalog, ) resolved = cs.get_device_access("BPM_C01-01/x") @@ -81,12 +68,10 @@ def test_catalog_can_be_configured_and_resolved(): def test_get_device_builds_attribute_from_config_model(): - cs = TangoControlSystem( - ConfigModel(name="test_tango_cs", tango_host="tangodb:10000") - ) + cs = TangoControlSystem(name="test_tango_cs", tango_host="tangodb:10000") resolved = cs.get_device_access( - AttributeConfigModel(attribute="sys/tg_test/1/float_scalar", unit="A") + AttributeConfig(attribute="sys/tg_test/1/float_scalar", unit="A") ) assert isinstance(resolved, Attribute) @@ -95,12 +80,10 @@ def test_get_device_builds_attribute_from_config_model(): def test_get_device_builds_read_only_attribute_from_config_model(): - cs = TangoControlSystem( - ConfigModel(name="test_tango_cs", tango_host="tangodb:10000") - ) + cs = TangoControlSystem(name="test_tango_cs", tango_host="tangodb:10000") resolved = cs.get_device_access( - AttributeReadOnlyConfigModel(attribute="sys/tg_test/1/float_scalar", unit="A") + AttributeReadOnlyConfig(attribute="sys/tg_test/1/float_scalar", unit="A") ) assert isinstance(resolved, AttributeReadOnly) @@ -109,12 +92,9 @@ def test_get_device_builds_read_only_attribute_from_config_model(): def test_get_device_builds_attribute_list_from_config_model(): - cs = TangoControlSystem( - ConfigModel(name="test_tango_cs", tango_host="tangodb:10000") - ) - + cs = TangoControlSystem(name="test_tango_cs", tango_host="tangodb:10000") resolved = cs.get_device_access( - AttributeListConfigModel( + AttributeListConfig( name="group", attributes=[ "sys/tg_test/1/float_scalar", @@ -135,12 +115,10 @@ def test_get_device_builds_attribute_list_from_config_model(): def test_get_device_builds_read_only_attribute_list_from_config_model(): - cs = TangoControlSystem( - ConfigModel(name="test_tango_cs", tango_host="tangodb:10000") - ) + cs = TangoControlSystem(name="test_tango_cs", tango_host="tangodb:10000") resolved = cs.get_device_access( - AttributeListReadOnlyConfigModel( + AttributeListReadOnlyConfig( name="group", attributes=[ "sys/tg_test/1/float_scalar", @@ -160,52 +138,44 @@ def test_get_device_builds_read_only_attribute_list_from_config_model(): def test_get_device_none_returns_none(): - cs = TangoControlSystem(ConfigModel(name="test_tango_cs")) + cs = TangoControlSystem(name="test_tango_cs") assert cs.get_device_access(None) is None def test_get_device_rejects_preconstructed_device_access(config): - cs = TangoControlSystem(ConfigModel(name="test_tango_cs")) + cs = TangoControlSystem(name="test_tango_cs") with pytest.raises(pyaml.PyAMLException, match="Use attach\\(\\)"): - cs.get_device_access(Attribute(config)) + cs.get_device_access(Attribute(**config.model_dump())) def test_get_device_requires_catalog_for_string_key(): - cs = TangoControlSystem(ConfigModel(name="test_tango_cs")) + cs = TangoControlSystem(name="test_tango_cs") with pytest.raises(pyaml.PyAMLException, match="has no catalog configured"): cs.get_device_access("BPM_C01-01/x") def test_get_device_reports_unknown_catalog_key(): - device = Attribute(AttributeConfigModel(attribute="sys/tg_test/1/float_scalar")) + device = Attribute(attribute="sys/tg_test/1/float_scalar") catalog = StaticCatalog( - StaticCatalogConfigModel( - entries=[ - StaticCatalogEntry( - StaticCatalogEntryConfigModel(key="BPM_C01-01/x", device=device) - ) - ], - ) + entries=[StaticCatalogEntry(key="BPM_C01-01/x", device=device)], ) - cs = TangoControlSystem(ConfigModel(name="test_tango_cs", catalog=catalog)) + cs = TangoControlSystem(name="test_tango_cs", catalog=catalog) with pytest.raises(pyaml.PyAMLException, match="cannot resolve key 'BPM_C01-02/x'"): cs.get_device_access("BPM_C01-02/x") def test_get_device_rejects_unknown_reference_type(): - cs = TangoControlSystem(ConfigModel(name="test_tango_cs")) + cs = TangoControlSystem(name="test_tango_cs") with pytest.raises(pyaml.PyAMLException, match="type int"): cs.get_device_access(42) def test_tango_control_system_exposes_tango_host(): - cs = TangoControlSystem( - ConfigModel(name="test_tango_cs", tango_host="tangodb:10000") - ) + cs = TangoControlSystem(name="test_tango_cs", tango_host="tangodb:10000") assert cs.get_tango_host() == "tangodb:10000" diff --git a/tests/test_multi_attribute.py b/tests/test_multi_attribute.py index 4185f03..408db86 100644 --- a/tests/test_multi_attribute.py +++ b/tests/test_multi_attribute.py @@ -1,9 +1,10 @@ import random +from unittest.mock import patch + +from tango.pyaml.multi_attribute import MultiAttribute from .mocked_control_system_initialized import MockedControlSystemInitialized from .mocked_device_proxy import MockedDeviceProxy -from unittest.mock import patch -from tango.pyaml.multi_attribute import MultiAttribute class TestMultiAttributes: @@ -15,7 +16,7 @@ def test_multi_read_write(self, config_multi): new=MockedControlSystemInitialized, ), ): - attr_list = MultiAttribute(config_multi) + attr_list = MultiAttribute(**config_multi.model_dump()) rand = random.Random() values = [rand.random() for _ in range(4)] attr_list.set(values) @@ -32,7 +33,7 @@ def test_multiattribute_range(self, config_multi_range): new=MockedControlSystemInitialized, ), ): - ma = MultiAttribute(config_multi_range) + ma = MultiAttribute(**config_multi_range.model_dump()) attr_range = ma.get_range() assert attr_range is not None assert len(attr_range) == 8 # (4*2) diff --git a/tests/test_static_catalog.py b/tests/test_static_catalog.py index 0d1ba19..7a38887 100644 --- a/tests/test_static_catalog.py +++ b/tests/test_static_catalog.py @@ -1,32 +1,29 @@ +import pyaml import pytest -import pyaml -from tango.pyaml.attribute import Attribute, ConfigModel as AttributeConfigModel +from tango.pyaml.attribute import Attribute from tango.pyaml.attribute_read_only import AttributeReadOnly -from tango.pyaml.controlsystem import ConfigModel as TangoControlSystemConfigModel from tango.pyaml.controlsystem import TangoControlSystem -from tango.pyaml.static_catalog import ConfigModel as StaticCatalogConfigModel from tango.pyaml.static_catalog import StaticCatalog -from tango.pyaml.static_catalog_entry import ConfigModel as EntryConfigModel from tango.pyaml.static_catalog_entry import StaticCatalogEntry def make_attribute( path: str = "domain/family/member/attr", unit: str = "mm" ) -> Attribute: - return Attribute(AttributeConfigModel(attribute=path, unit=unit)) + return Attribute(attribute=path, unit=unit) def make_entry(key: str, device=None) -> StaticCatalogEntry: if device is None: device = make_attribute() - return StaticCatalogEntry(EntryConfigModel(key=key, device=device)) + return StaticCatalogEntry(key=key, device=device) def make_catalog(name: str = "static", entries=None) -> StaticCatalog: if entries is None: entries = [make_entry("default/key")] - return StaticCatalog(StaticCatalogConfigModel(entries=entries)) + return StaticCatalog(entries=entries) # --- StaticCatalogEntry --- @@ -48,13 +45,13 @@ def test_static_catalog_entry_returns_device(): def test_static_catalog_rejects_empty_entries(): with pytest.raises(pyaml.PyAMLException, match="must contain at least one entry"): - StaticCatalog(StaticCatalogConfigModel(entries=[])) + StaticCatalog(entries=[]) def test_static_catalog_rejects_duplicate_keys(): entries = [make_entry("BPM/x"), make_entry("BPM/x")] with pytest.raises(pyaml.PyAMLException, match="duplicate key 'BPM/x'"): - StaticCatalog(StaticCatalogConfigModel(entries=entries)) + StaticCatalog(entries=entries) # --- StaticCatalog.resolve --- @@ -93,8 +90,8 @@ def test_static_catalog_raises_on_unknown_key(): def test_static_catalog_is_shared_across_control_systems(): device = make_attribute() catalog = make_catalog(entries=[make_entry("BPM/x", device=device)]) - live = TangoControlSystem(TangoControlSystemConfigModel(name="live", catalog=catalog)) - ops = TangoControlSystem(TangoControlSystemConfigModel(name="ops", catalog=catalog)) + live = TangoControlSystem(name="live", catalog=catalog) + ops = TangoControlSystem(name="ops", catalog=catalog) assert live.get_catalog() is catalog assert ops.get_catalog() is catalog @@ -108,9 +105,7 @@ def test_static_catalog_is_shared_across_control_systems(): def test_static_catalog_works_with_attribute_read_only(): - device = AttributeReadOnly( - AttributeConfigModel(attribute="sr/bpm/c01-01/pos", unit="mm") - ) + device = AttributeReadOnly(attribute="sr/bpm/c01-01/pos", unit="mm") catalog = make_catalog(entries=[make_entry("BPM/x", device=device)]) resolved = catalog.resolve("BPM/x") @@ -122,7 +117,7 @@ def test_static_catalog_works_with_attribute_read_only(): def test_static_catalog_can_be_used_through_tango_control_system(): device = make_attribute("sr/bpm/c01-01/x", unit="mm") catalog = make_catalog(entries=[make_entry("BPM/x", device=device)]) - control_system = TangoControlSystem(TangoControlSystemConfigModel(name="live", catalog=catalog)) + control_system = TangoControlSystem(name="live", catalog=catalog) resolved = control_system.get_device_access("BPM/x") diff --git a/tests/test_tango_catalog.py b/tests/test_tango_catalog.py index bce87cf..6f7af50 100644 --- a/tests/test_tango_catalog.py +++ b/tests/test_tango_catalog.py @@ -2,24 +2,24 @@ import pyaml import pytest -import tango from pyaml.control.controlsystem import ControlSystemAdapter -from .mocked_device_proxy import MockedAttributeInfoEx, MockedAttributeProxy +import tango from tango.pyaml.attribute import Attribute from tango.pyaml.attribute_read_only import AttributeReadOnly -from tango.pyaml.controlsystem import ConfigModel as TangoControlSystemConfigModel from tango.pyaml.controlsystem import TangoControlSystem -from tango.pyaml.tango_catalog import ConfigModel, TangoCatalog +from tango.pyaml.tango_catalog import TangoCatalog + +from .mocked_device_proxy import MockedAttributeInfoEx, MockedAttributeProxy def build_control_system(catalog: TangoCatalog, name="live"): - control_system = TangoControlSystem(TangoControlSystemConfigModel(name=name, catalog=catalog)) + control_system = TangoControlSystem(name=name, catalog=catalog) return control_system def test_tango_catalog_disconnected_resolves_without_querying_tango(): - catalog = TangoCatalog(ConfigModel(disconnected=True)) + catalog = TangoCatalog(disconnected=True) control_system = build_control_system(catalog) with patch("tango.AttributeProxy") as attr_proxy: @@ -41,7 +41,7 @@ def test_tango_catalog_connected_resolves_writable_attribute(): max_value="12.0", data_format=tango.AttrDataFormat.SPECTRUM, ) - catalog = TangoCatalog(ConfigModel()) + catalog = TangoCatalog() control_system = build_control_system(catalog) with patch( @@ -65,7 +65,7 @@ def test_tango_catalog_connected_resolves_read_only_attribute(): attr_config = MockedAttributeInfoEx( name="position", writable=tango.AttrWriteType.READ, unit="mm" ) - catalog = TangoCatalog(ConfigModel()) + catalog = TangoCatalog() control_system = build_control_system(catalog) with patch( @@ -79,7 +79,7 @@ def test_tango_catalog_connected_resolves_read_only_attribute(): def test_tango_catalog_caches_resolved_devices(): - catalog = TangoCatalog(ConfigModel()) + catalog = TangoCatalog() control_system = build_control_system(catalog) with patch( @@ -94,7 +94,7 @@ def test_tango_catalog_caches_resolved_devices(): def test_tango_catalog_cache_is_bound_to_control_system_resolver(): - catalog = TangoCatalog(ConfigModel()) + catalog = TangoCatalog() live = build_control_system(catalog, name="live") ops = build_control_system(catalog, name="ops") @@ -113,11 +113,9 @@ def test_tango_catalog_cache_is_bound_to_control_system_resolver(): def test_tango_catalog_connected_metadata_uses_control_system_tango_host(): key = "domain/family/member/current" - catalog = TangoCatalog(ConfigModel()) - live = TangoControlSystem( - TangoControlSystemConfigModel(name="live", tango_host="live-db:10000", catalog=catalog)) - ops = TangoControlSystem( - TangoControlSystemConfigModel(name="ops", tango_host="ops-db:10000", catalog=catalog)) + catalog = TangoCatalog() + live = TangoControlSystem(name="live", tango_host="live-db:10000", catalog=catalog) + ops = TangoControlSystem(name="ops", tango_host="ops-db:10000", catalog=catalog) attr_configs = { "//live-db:10000/domain/family/member/current": MockedAttributeInfoEx( @@ -150,8 +148,8 @@ def attribute_proxy(attr_full_name): def test_tango_catalog_can_be_used_through_tango_control_system(): - catalog = TangoCatalog(ConfigModel(disconnected=True)) - control_system = TangoControlSystem(TangoControlSystemConfigModel(name="live", catalog=catalog)) + catalog = TangoCatalog(disconnected=True) + control_system = TangoControlSystem(name="live", catalog=catalog) device = control_system.get_device_access("domain/family/member/attribute") @@ -160,7 +158,7 @@ def test_tango_catalog_can_be_used_through_tango_control_system(): def test_tango_catalog_rejects_non_tango_control_system(): - catalog = TangoCatalog(ConfigModel()) + catalog = TangoCatalog() with pytest.raises( pyaml.PyAMLException, match="can only resolve through TangoControlSystem" @@ -172,7 +170,7 @@ def test_tango_catalog_rejects_external_tango_control_system_class(): class TangoControlSystem: pass - catalog = TangoCatalog(ConfigModel()) + catalog = TangoCatalog() with pytest.raises( pyaml.PyAMLException, match="can only resolve through TangoControlSystem" @@ -181,7 +179,7 @@ class TangoControlSystem: def test_tango_catalog_requires_control_system_attachment(): - catalog = TangoCatalog(ConfigModel()) + catalog = TangoCatalog() with pytest.raises( pyaml.PyAMLException, match="needs a TangoControlSystem context" @@ -190,7 +188,7 @@ def test_tango_catalog_requires_control_system_attachment(): def test_tango_catalog_rejects_invalid_tango_reference(): - catalog = TangoCatalog(ConfigModel()) + catalog = TangoCatalog() control_system = build_control_system(catalog) with pytest.raises( @@ -200,7 +198,7 @@ def test_tango_catalog_rejects_invalid_tango_reference(): def test_tango_catalog_rejects_invalid_index(): - catalog = TangoCatalog(ConfigModel()) + catalog = TangoCatalog() control_system = build_control_system(catalog) with pytest.raises(pyaml.PyAMLException, match="invalid index"): @@ -208,7 +206,7 @@ def test_tango_catalog_rejects_invalid_index(): def test_tango_catalog_disconnected_resolves_indexed_attribute(): - catalog = TangoCatalog(ConfigModel(disconnected=True)) + catalog = TangoCatalog(disconnected=True) control_system = build_control_system(catalog) with patch("tango.AttributeProxy") as attr_proxy: @@ -228,7 +226,7 @@ def test_tango_catalog_connected_resolves_indexed_writable_spectrum(): unit="mm", data_format=tango.AttrDataFormat.SPECTRUM, ) - catalog = TangoCatalog(ConfigModel()) + catalog = TangoCatalog() control_system = build_control_system(catalog) with patch( @@ -254,7 +252,7 @@ def test_tango_catalog_connected_resolves_indexed_read_only_spectrum(): unit="mm", data_format=tango.AttrDataFormat.SPECTRUM, ) - catalog = TangoCatalog(ConfigModel()) + catalog = TangoCatalog() control_system = build_control_system(catalog) with patch( @@ -273,15 +271,19 @@ def test_tango_catalog_connected_rejects_indexed_scalar_attribute(): writable=tango.AttrWriteType.READ_WRITE, data_format=tango.AttrDataFormat.SCALAR, ) - catalog = TangoCatalog(ConfigModel()) + catalog = TangoCatalog() control_system = build_control_system(catalog) - with patch( - "tango.AttributeProxy", - return_value=MockedAttributeProxy("domain/family/member/current", attr_config), + with ( + patch( + "tango.AttributeProxy", + return_value=MockedAttributeProxy( + "domain/family/member/current", attr_config + ), + ), + pytest.raises(pyaml.PyAMLException, match="not a SPECTRUM"), ): - with pytest.raises(pyaml.PyAMLException, match="not a SPECTRUM"): - catalog.resolve("domain/family/member/current@0", control_system) + catalog.resolve("domain/family/member/current@0", control_system) def test_tango_catalog_indexed_caches_resolved_devices(): @@ -289,7 +291,7 @@ def test_tango_catalog_indexed_caches_resolved_devices(): name="position", data_format=tango.AttrDataFormat.SPECTRUM, ) - catalog = TangoCatalog(ConfigModel()) + catalog = TangoCatalog() control_system = build_control_system(catalog) with patch( @@ -304,16 +306,17 @@ def test_tango_catalog_indexed_caches_resolved_devices(): def test_tango_catalog_wraps_tango_errors(): - catalog = TangoCatalog(ConfigModel()) + catalog = TangoCatalog() control_system = build_control_system(catalog) - with patch("tango.AttributeProxy", side_effect=tango.DevFailed()): - with pytest.raises( + with ( + patch("tango.AttributeProxy", side_effect=tango.DevFailed()), + pytest.raises( pyaml.PyAMLException, - match="Tango catalog" - " cannot resolve 'domain/family/member/attribute'", - ): - catalog.resolve("domain/family/member/attribute", control_system) + match="Tango catalog cannot resolve 'domain/family/member/attribute'", + ), + ): + catalog.resolve("domain/family/member/attribute", control_system) def test_tango_catalog_rejects_incomplete_tango_config(): @@ -323,17 +326,19 @@ class IncompleteAttributeConfig: max_value = "1" data_format = tango.AttrDataFormat.SCALAR - catalog = TangoCatalog(ConfigModel()) + catalog = TangoCatalog() control_system = build_control_system(catalog) - with patch( - "tango.AttributeProxy", - return_value=MockedAttributeProxy( - "domain/family/member/attribute", IncompleteAttributeConfig() + with ( + patch( + "tango.AttributeProxy", + return_value=MockedAttributeProxy( + "domain/family/member/attribute", IncompleteAttributeConfig() + ), ), - ): - with pytest.raises( + pytest.raises( pyaml.PyAMLException, match="incomplete Tango attribute config, missing 'writable'", - ): - catalog.resolve("domain/family/member/attribute", control_system) + ), + ): + catalog.resolve("domain/family/member/attribute", control_system)