Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion pyaml/common/element.py
Original file line number Diff line number Diff line change
Expand Up @@ -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":
Expand Down
2 changes: 1 addition & 1 deletion pyaml/control/controlsystem.py
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand Down
98 changes: 51 additions & 47 deletions pyaml/diagnostics/tune_monitor.py
Original file line number Diff line number Diff line change
@@ -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:
"""
Expand Down Expand Up @@ -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)
Expand Down Expand Up @@ -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"])
3 changes: 2 additions & 1 deletion pyaml/validation/validation_models.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,7 @@

import inspect
import logging
from abc import ABCMeta
from typing import Any

from pydantic import BaseModel, ConfigDict, create_model
Expand All @@ -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.

Expand Down
42 changes: 38 additions & 4 deletions pyaml/validation/validator.py
Original file line number Diff line number Diff line change
@@ -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.

Expand Down Expand Up @@ -62,6 +95,7 @@ def validate(
# raise TypeError("Top-level configuration did not validate to a ConfigurationSchema.")

return validated


@classmethod
def validate_to_dict(
Expand All @@ -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:
Expand Down