diff --git a/pyaml/common/element.py b/pyaml/common/element.py index 53053d04..613d63b3 100644 --- a/pyaml/common/element.py +++ b/pyaml/common/element.py @@ -172,7 +172,7 @@ def check_peer(self): to a simulator or to a control system """ if self._peer is None: - raise PyAMLException(f"{str(self)} is not attachedto a control system or the a simulator") + raise PyAMLException(f"{str(self.name)} is not attachedto a control system or the a simulator") @property def peer(self) -> "ElementHolder": diff --git a/pyaml/control/controlsystem.py b/pyaml/control/controlsystem.py index b938f662..fd96c4d6 100644 --- a/pyaml/control/controlsystem.py +++ b/pyaml/control/controlsystem.py @@ -197,7 +197,7 @@ def fill_device(self, elements: list[Element]): elif isinstance(e, BetatronTuneMonitor): # Built in tune monitor - tuneDevs = self.get_devices_access([e._cfg.tune_h, e._cfg.tune_v]) + tuneDevs = self.get_devices_access([e.tune_h, e.tune_v]) betatron_tune = RBetatronTuneArray(e, tuneDevs) e = e.attach(self, betatron_tune) self.add_betatron_tune_monitor(e) diff --git a/pyaml/diagnostics/tune_monitor.py b/pyaml/diagnostics/tune_monitor.py index 3ea38c3d..55db430b 100644 --- a/pyaml/diagnostics/tune_monitor.py +++ b/pyaml/diagnostics/tune_monitor.py @@ -1,65 +1,66 @@ +import copy +from typing import Self + +from numpy.typing import NDArray + from ..common.abstract import ReadFloatArray -from ..common.element import Element, ElementConfigModel -from ..control.deviceaccess import DeviceAccess +from ..common.element import Element, __pyaml_repr__ +from ..validation import DynamicValidation, register_schema from .atune_monitor import ABetatronTuneMonitor -try: - from typing import Self # Python 3.11+ -except ImportError: - from typing_extensions import Self # Python 3.10 and earlier -import numpy as np -from pydantic import ConfigDict - PYAMLCLASS = "BetatronTuneMonitor" -class ConfigModel(ElementConfigModel): +@register_schema +class BetatronTuneMonitor(Element, DynamicValidation, ABetatronTuneMonitor): """ - Configuration model for BetatronTuneMonitor + Betatron tune monitor. + + Represents a betatron tune monitor in the accelerator lattice and + provides access to the measured horizontal and vertical betatron + tunes. Optionally, the monitor can be associated with an RF plant + used to convert tune measurements to frequency values. Parameters ---------- - tune_h : str, optional - Horizontal betatron tune device - tune_v : str, optional - Vertical betatron tune device - rf_plant_name : str, optional - PyAML element name of the RF plant + name : str + Name of the betatron tune monitor. + description : str | None, optional + Description of the monitor. + tune_h : str | None, optional + Device catalog key for the horizontal betatron tune measurement. + tune_v : str | None, optional + Device catalog key for the vertical betatron tune measurement. + rf_plant_name : str | None, optional + Name of the associated RF plant element used by the monitor. """ - model_config = ConfigDict(arbitrary_types_allowed=True, extra="forbid") - - tune_h: str | None - tune_v: str | None - rf_plant_name: str | None = None - - -class BetatronTuneMonitor(Element, ABetatronTuneMonitor): - """ - Class providing access to a betatron tune monitor - of a physical or simulated lattice. - The monitor provides horizontal and vertical betatron tune measurements. - """ - - def __init__(self, cfg: ConfigModel): - """ - Construct a BetatronTuneMonitor - - Parameters - ---------- - cfg : ConfigModel - Configuration for the BetatronTuneMonitor, including - device access for horizontal and vertical tunes. - """ - - super().__init__(cfg.name) - self._cfg = cfg + def __init__( + self, + name: str, + description: str | None = None, + tune_h: str | None = None, + tune_v: str | None = None, + rf_plant_name: str | None = None, + ): + super().__init__(name, None, description) + self._tune_h = tune_h + self._tune_v = tune_v + self._rf_plant_name = rf_plant_name self.__tune = None self._h = None def set_harmonic(self, h: int): self._h = float(h) + @property + def tune_h(self) -> str | None: + return self._tune_h + + @property + def tune_v(self) -> str | None: + return self._tune_v + @property def tune(self) -> ReadFloatArray: """ @@ -88,9 +89,9 @@ class TuneFreq(ReadFloatArray): def __init__(self, parent: BetatronTuneMonitor): self.parent = parent - def get(self) -> np.array: + def get(self) -> NDArray: h = self.parent._h - rf_name = self.parent._cfg.rf_plant_name + rf_name = self.parent._rf_plant_name if h is not None and rf_name is not None: tune = self.parent.tune.get() rf = self.parent.peer.get_rf_plant(rf_name) @@ -119,7 +120,10 @@ def attach(self, peer, betatron_tune: ReadFloatArray) -> Self: Self A new attached instance of TuneMonitor """ - obj = self.__class__(self._cfg) + obj = copy.copy(self) obj.__tune = betatron_tune obj._peer = peer return obj + + def __repr__(self): + return __pyaml_repr__(self, exclude=["tune", "frequency"]) diff --git a/pyaml/validation/validation_models.py b/pyaml/validation/validation_models.py index afecfb36..0683d86d 100644 --- a/pyaml/validation/validation_models.py +++ b/pyaml/validation/validation_models.py @@ -2,6 +2,7 @@ import inspect import logging +from abc import ABCMeta from typing import Any from pydantic import BaseModel, ConfigDict, create_model @@ -23,7 +24,7 @@ class ValidationModel(PyAMLBaseModel): model_config = ConfigDict(arbitrary_types_allowed=True, extra="forbid") -class ValidationMeta(type): +class ValidationMeta(ABCMeta): """ Metaclass that validates constructor arguments before object creation. diff --git a/pyaml/validation/validator.py b/pyaml/validation/validator.py index 03cd4cec..b86a69f2 100644 --- a/pyaml/validation/validator.py +++ b/pyaml/validation/validator.py @@ -1,18 +1,51 @@ """Module for schema validation.""" +from collections.abc import Mapping import logging import warnings from typing import Any -from pydantic import ValidationError +from pydantic import ValidationError, BaseModel from .configuration_models import ConfigurationSchema, ModuleConfigurationSchema from .errors import extract_location_metadata, raise_validation_error from .registry import SchemaRegistry + logger = logging.getLogger(__name__) +def dump_nested(value: Any) -> Any: + """ + Recursively convert nested Pydantic models to plain Python objects. + + Traverses mappings, lists, and tuples, replacing any + :class:`pydantic.BaseModel` instances with the result of + :meth:`BaseModel.model_dump`. All other values are returned unchanged. + + Parameters + ---------- + value : Any + The object to convert. + + Returns + ------- + Any + The converted object with all nested Pydantic models represented as + dictionaries. + """ + + if isinstance(value, BaseModel): + return value.model_dump() + if isinstance(value, Mapping): + return {k: dump_nested(v) for k, v in value.items()} + if isinstance(value, list): + return [dump_nested(v) for v in value] + if isinstance(value, tuple): + return tuple(dump_nested(v) for v in value) + return value + + class SchemaValidator: """Recursive validator for configuration dictionaries. @@ -62,6 +95,7 @@ def validate( # raise TypeError("Top-level configuration did not validate to a ConfigurationSchema.") return validated + @classmethod def validate_to_dict( @@ -73,9 +107,9 @@ def validate_to_dict( """ validated = cls.validate(data) - if isinstance(validated, dict): - return validated - return validated.model_dump() + return dump_nested(validated) + + @classmethod def _recursive_validate(cls, obj: Any) -> Any: