diff --git a/pyaml/__init__.py b/pyaml/__init__.py index 65050bc0..0bb1c8b0 100644 --- a/pyaml/__init__.py +++ b/pyaml/__init__.py @@ -16,9 +16,10 @@ import logging.config import os +from pyaml.common.element import ReprOptions, set_repr_options from pyaml.common.exception import PyAMLConfigException, PyAMLException -__all__ = ["PyAMLException", "PyAMLConfigException"] +__all__ = ["PyAMLException", "PyAMLConfigException", "ReprOptions", "set_repr_options"] config_file = os.getenv("PYAML_LOG_CONFIG", "pyaml_logging.conf") diff --git a/pyaml/accelerator.py b/pyaml/accelerator.py index f6682599..0bc27301 100644 --- a/pyaml/accelerator.py +++ b/pyaml/accelerator.py @@ -96,6 +96,8 @@ class Accelerator(DynamicValidation): :attr:`live` and :attr:`design` properties. """ + __pyaml_repr_exclude__ = ("description", "yellow_pages") + def __init__( self, facility: str, @@ -318,6 +320,14 @@ def modes(self) -> dict[str, "ElementHolder"]: modes.update(self._controls) return modes + def _pyaml_repr_fields(self) -> dict[str, object]: + return { + "facility": self.facility, + "machine": self.machine, + "simulators": list(self._simulators), + "controls": list(self._controls), + } + def __repr__(self): """ Implement the __repr__ string. diff --git a/pyaml/arrays/element_array.py b/pyaml/arrays/element_array.py index 56985b43..3268d488 100644 --- a/pyaml/arrays/element_array.py +++ b/pyaml/arrays/element_array.py @@ -11,7 +11,7 @@ import numpy as np from ..bpm.bpm import BPM -from ..common.element import Element +from ..common.element import Element, __pyaml_repr__ from ..common.exception import PyAMLException from ..magnet.cfm_magnet import CombinedFunctionMagnet from ..magnet.magnet import Magnet @@ -100,6 +100,17 @@ def names(self) -> list[str]: """ return [e.get_name() for e in self] + def _pyaml_repr_fields(self) -> dict[str, object]: + return { + "name": self.get_name(), + "size": len(self), + "peer": self.get_peer(), + "elements": self.names(), + } + + def __repr__(self): + return __pyaml_repr__(self) + def __create_array(self, array_name: str, element_type: type, elements: list): """ Implement the __create_array protocol operation. diff --git a/pyaml/bpm/bpm.py b/pyaml/bpm/bpm.py index a50e4975..dd7015c2 100644 --- a/pyaml/bpm/bpm.py +++ b/pyaml/bpm/bpm.py @@ -78,11 +78,11 @@ def __init__( Initialize a beam-position monitor configuration. """ super().__init__(name, lattice_names, description) - self._x_pos = x_pos - self._y_pos = y_pos - self._x_offset = x_offset - self._y_offset = y_offset - self._tilt_name = tilt + self.x_pos = x_pos + self.y_pos = y_pos + self.x_offset = x_offset + self.y_offset = y_offset + self.tilt_name = tilt self._positions = None self._offset = None self._tilt = None @@ -191,7 +191,7 @@ def get_pos_devices(self) -> list[str | None]: list of str or None Horizontal and vertical position device keys. """ - return [self._x_pos, self._y_pos] + return [self.x_pos, self.y_pos] def get_tilt_device(self) -> str | None: """ @@ -202,7 +202,7 @@ def get_tilt_device(self) -> str | None: str or None Tilt device key. """ - return self._tilt_name + return self.tilt_name def get_offset_devices(self) -> list[str | None]: """ @@ -213,7 +213,7 @@ def get_offset_devices(self) -> list[str | None]: list of str or None Horizontal and vertical offset device keys. """ - return [self._x_offset, self._y_offset] + return [self.x_offset, self.y_offset] def __repr__(self): """ diff --git a/pyaml/common/element.py b/pyaml/common/element.py index d8614b6d..e18da8d6 100644 --- a/pyaml/common/element.py +++ b/pyaml/common/element.py @@ -1,68 +1,217 @@ """Base classes for configured accelerator and lattice elements.""" +from collections.abc import Mapping, Sequence +from dataclasses import dataclass from typing import TYPE_CHECKING, Any from pydantic import BaseModel, ConfigDict +from . import abstract from .exception import PyAMLException if TYPE_CHECKING: from .holders.element_holder import ElementHolder -def __pyaml_repr__(obj, exclude: list[str] | None = None): +@dataclass(frozen=True) +class ReprOptions: + """Limits applied to PyAML object representations.""" + + max_items: int = 3 + max_depth: int = 2 + max_length: int = 800 + + +_repr_options = ReprOptions() + + +def set_repr_options( + max_items: int | None = None, + max_depth: int | None = None, + max_length: int | None = None, +) -> ReprOptions: """ - Build a representation from configuration fields and public properties. + Configure the limits used by PyAML object representations. - Parameters - ---------- - obj : object - Object to represent. - exclude : list[str] | None - Attribute or property names to exclude from the output. + Passing no arguments returns the current options. Every supplied value must + be a positive integer. """ + global _repr_options + + values = { + "max_items": _repr_options.max_items if max_items is None else max_items, + "max_depth": _repr_options.max_depth if max_depth is None else max_depth, + "max_length": _repr_options.max_length if max_length is None else max_length, + } + for name, value in values.items(): + if not isinstance(value, int) or value < 1: + raise ValueError(f"{name} must be a positive integer") + + _repr_options = ReprOptions(**values) + return _repr_options + - if exclude is None: - exclude = [] - - cls_name = obj.__class__.__name__ - - # Keep the old behavior when _cfg exists - cfg = getattr(obj, "_cfg", None) - if cfg is not None: - if isinstance(obj, Element): - return repr(cfg).replace( - "ConfigModel(", - f"{cls_name}(peer={obj.attached_to()!r}, ", - 1, - ) - return repr(cfg).replace("ConfigModel", cls_name, 1) - - # Generic fallback when there is no _cfg - attrs = {} - - # Instance attributes - for k, v in obj.__dict__.items(): - # Exclude private attributes and excluded - if not k.startswith("_") and k not in exclude: - attrs[k] = v - - # Properties - for name, attr in vars(type(obj)).items(): - if isinstance(attr, property) and name not in exclude: +def _unavailable(error: Exception) -> str: + return f"" + + +def _class_exclusions(obj) -> set[str]: + exclusions: set[str] = set() + for cls in type(obj).__mro__: + exclusions.update(getattr(cls, "__pyaml_repr_exclude__", ())) + return exclusions + + +def _properties(obj) -> dict[str, Any]: + values: dict[str, Any] = {} + for cls in reversed(type(obj).__mro__): + for name, descriptor in vars(cls).items(): + if name.startswith("_") or not isinstance(descriptor, property): + continue try: - attrs[name] = getattr(obj, name) - except Exception as e: - attrs[name] = f"" + values[name] = getattr(obj, name) + except Exception as error: + values[name] = _unavailable(error) + return values + - if isinstance(obj, Element) and "name" not in attrs and "name" not in exclude: +def _fields(obj) -> dict[str, Any]: + custom_fields = getattr(obj, "_pyaml_repr_fields", None) + if custom_fields is not None: try: - attrs["name"] = obj.get_name() - except Exception as e: - attrs["name"] = f"" + return custom_fields() + except Exception as error: + return {"value": _unavailable(error)} + + values = _properties(obj) + for name, value in vars(obj).items(): + if not name.startswith("_"): + values.setdefault(name, value) + return values + + +def _identity(obj) -> str: + name = getattr(obj, "name", None) + try: + name = name() if callable(name) else name + except Exception: + name = None + if isinstance(name, str): + return f"{obj.__class__.__name__}:{name}" + return obj.__class__.__name__ + + +def _short_object_repr(obj) -> str: + name = getattr(obj, "name", None) + try: + name = name() if callable(name) else name + except Exception: + name = None + if isinstance(name, str): + return f"{obj.__class__.__name__}(name={_format_value(name, 0, set())})" + return obj.__class__.__name__ + + +def _selected_items(values: Sequence | set) -> tuple[list[Any], int]: + items = list(values) + omitted = len(items) - _repr_options.max_items + if omitted <= 0: + return items, 0 + + head_count = (_repr_options.max_items + 1) // 2 + tail_count = _repr_options.max_items - head_count + selected = items[:head_count] + if tail_count: + selected.extend(items[-tail_count:]) + return selected, omitted + + +def _format_sequence(values: Sequence | set, depth: int, active: set[int]) -> str: + selected, omitted = _selected_items(values) + parts = [_format_value(value, depth, active) for value in selected] + if omitted: + insert_at = (_repr_options.max_items + 1) // 2 + parts.insert(insert_at, f"... +{omitted} more ...") + + if isinstance(values, tuple): + if len(parts) == 1 and not omitted: + return f"({parts[0]},)" + return f"({', '.join(parts)})" + if isinstance(values, set): + return "{" + ", ".join(parts) + "}" + return "[" + ", ".join(parts) + "]" + + +def _format_mapping(values: Mapping, depth: int, active: set[int]) -> str: + selected, omitted = _selected_items(list(values.items())) + parts = [f"{_format_value(key, depth, active)}: {_format_value(value, depth, active)}" for key, value in selected] + if omitted: + parts.insert((_repr_options.max_items + 1) // 2, f"... +{omitted} more ...") + return "{" + ", ".join(parts) + "}" + + +def _format_object(obj, depth: int, active: set[int], extra_exclusions: set[str] | None = None) -> str: + if depth >= _repr_options.max_depth: + return _short_object_repr(obj) + if id(obj) in active: + return f"" + + active.add(id(obj)) + try: + values = _fields(obj) + exclusions = _class_exclusions(obj) + if extra_exclusions: + exclusions.update(extra_exclusions) + parts = [] + for name, value in values.items(): + if name.startswith("_") or name in exclusions or callable(value): + continue + if isinstance(value, (abstract.ReadFloatScalar, abstract.ReadFloatArray, abstract.ReadWriteFloatArray)): + continue + formatted = _identity(value) if name == "peer" and value is not None else _format_value(value, depth + 1, active) + parts.append(f"{name}={formatted}") + return f"{obj.__class__.__name__}({', '.join(parts)})" if parts else obj.__class__.__name__ + except Exception as error: + return f"{obj.__class__.__name__}({_unavailable(error)})" + finally: + active.remove(id(obj)) + + +def _format_value(value, depth: int, active: set[int]) -> str: + if isinstance(value, str): + limit = max(1, _repr_options.max_length // 4) + suffix = "..." if len(value) > limit else "" + return repr(value[:limit] + suffix) + if value is None or isinstance(value, (bool, int, float, complex)): + return repr(value) + if isinstance(value, Mapping): + return _format_mapping(value, depth, active) + if isinstance(value, (list, tuple, set, frozenset)): + return _format_sequence(value, depth, active) + if type(value).__module__.startswith("numpy"): + shape = getattr(value, "shape", None) + dtype = getattr(value, "dtype", None) + return f"{value.__class__.__name__}(shape={shape!r}, dtype={dtype!r})" + if type(value).__module__.startswith("pyaml"): + return _format_object(value, depth, active) + try: + result = repr(value) + except Exception as error: + return _unavailable(error) + limit = max(1, _repr_options.max_length // 2) + return result if len(result) <= limit else result[:limit] + "..." - parts = ", ".join(f"{k}={v!r}" for k, v in attrs.items()) - return f"{cls_name}({parts})" if parts else cls_name + +def __pyaml_repr__(obj, exclude: list[str] | None = None): + """ + Return an informative, bounded representation of a PyAML object. + + Public attributes and read-only properties are included unless they are + excluded. Device accessors are omitted so rendering never reads a control + system value. + """ + result = _format_object(obj, 0, set(), set(exclude or ())) + return result[: _repr_options.max_length] + ("..." if len(result) > _repr_options.max_length else "") class ElementConfigModel(BaseModel): @@ -140,6 +289,8 @@ class Element: Perform post-construction initialization after attachment. """ + __pyaml_repr_exclude__ = ("description",) + def __init__( self, name: str, diff --git a/pyaml/common/holders/generic_array_holder.py b/pyaml/common/holders/generic_array_holder.py index 56524b23..0ce62441 100644 --- a/pyaml/common/holders/generic_array_holder.py +++ b/pyaml/common/holders/generic_array_holder.py @@ -3,7 +3,7 @@ from collections.abc import Callable from typing import TYPE_CHECKING, Generic, TypeVar -from ..element import Element +from ..element import Element, __pyaml_repr__ if TYPE_CHECKING: from .element_holder import ElementHolder @@ -116,3 +116,6 @@ def __getitem__(self, key): Element or sub-array selected by ``key``. """ return self.get().__getitem__(key) + + def __repr__(self): + return __pyaml_repr__(self) diff --git a/pyaml/common/holders/generic_element_holder.py b/pyaml/common/holders/generic_element_holder.py index 4554f267..a23dec8b 100644 --- a/pyaml/common/holders/generic_element_holder.py +++ b/pyaml/common/holders/generic_element_holder.py @@ -2,7 +2,7 @@ from typing import TYPE_CHECKING, Generic, TypeVar -from ..element import Element +from ..element import Element, __pyaml_repr__ if TYPE_CHECKING: from .element_holder import ElementHolder @@ -88,3 +88,6 @@ def add(self, m: T): This method updates the holder in place. """ self._peer._add(self._store, m) + + def __repr__(self): + return __pyaml_repr__(self) diff --git a/pyaml/common/holders/rf_holder.py b/pyaml/common/holders/rf_holder.py index 506b7e95..257c396a 100644 --- a/pyaml/common/holders/rf_holder.py +++ b/pyaml/common/holders/rf_holder.py @@ -5,6 +5,7 @@ from ...rf.rf_plant import RFPlant from ...rf.rf_transmitter import RFTransmitter from ..abstract import ReadWriteFloatScalar +from ..element import __pyaml_repr__ if TYPE_CHECKING: from .element_holder import ElementHolder @@ -65,6 +66,9 @@ def add(self, rf: RFTransmitter): """ self._peer._add(self._peer._RFTRANSMITTER, rf) + def __repr__(self): + return __pyaml_repr__(self) + class RFHolder: """ @@ -145,3 +149,6 @@ def add(self, rf: RFPlant): The plant is registered in the parent holder in place. """ self._peer._add(self._peer._RFPLANT, rf) + + def __repr__(self): + return __pyaml_repr__(self) diff --git a/pyaml/lattice/simulator.py b/pyaml/lattice/simulator.py index 5b37e062..81092825 100644 --- a/pyaml/lattice/simulator.py +++ b/pyaml/lattice/simulator.py @@ -115,6 +115,8 @@ class Simulator(ElementHolder, DynamicValidation): Resolve a PyAML element to matching PyAT lattice elements. """ + __pyaml_repr_exclude__ = ("description", "ring") + def __init__( self, name: str, @@ -429,6 +431,14 @@ def get_at_elems(self, element: Element) -> list[at.Element]: else: return [elts[idx] for idx in indices] + def _pyaml_repr_fields(self) -> dict[str, object]: + return { + "name": self.name(), + "lattice": self.lattice, + "mat_key": self.mat_key, + "n_elements": len(self.ring), + } + def __repr__(self): """ Implement the ``__repr__`` string. diff --git a/pyaml/magnet/identity_model.py b/pyaml/magnet/identity_model.py index 077de6db..9d5f94f6 100644 --- a/pyaml/magnet/identity_model.py +++ b/pyaml/magnet/identity_model.py @@ -75,20 +75,20 @@ def __init__( """ Initialize the IdentityMagnetModel. """ - self._physics = physics - self._powerconverter = powerconverter - self._unit = unit + self.physics = physics + self.powerconverter = powerconverter + self.unit = unit - if self._physics is None and self._powerconverter is None: + if self.physics is None and self.powerconverter is None: raise PyAMLException("Invalid IdentityMagnetModel configuration,physics or powerconverter device required") - if self._physics is not None and self._powerconverter is not None: + if self.physics is not None and self.powerconverter is not None: raise PyAMLException( "Invalid IdentityMagnetModel configuration,physics or powerconverter device required but not both" ) - if self._physics: - self.__device = self._physics + if self.physics: + self.__device = self.physics else: - self.__device = self._powerconverter + self.__device = self.powerconverter def compute_hardware_values(self, strengths: np.array) -> np.array: """ @@ -124,11 +124,11 @@ def compute_strengths(self, currents: np.array) -> np.array: def get_strength_units(self) -> list[str]: """Return the units of magnet strengths.""" - return [self._unit] + return [self.unit] def get_hardware_units(self) -> list[str]: """Return the units of hardware values.""" - return [self._unit] + return [self.unit] def get_device_names(self) -> list[str | None]: """Return the associated device names.""" @@ -147,11 +147,11 @@ def set_magnet_rigidity(self, brho: np.double): def has_physics(self) -> bool: """Return whether the model provides physics strengths.""" - return self._physics is not None + return self.physics is not None def has_hardware(self) -> bool: """Return whether the model provides hardware values.""" - return self._powerconverter is not None + return self.powerconverter is not None def __repr__(self): """ diff --git a/pyaml/magnet/magnet.py b/pyaml/magnet/magnet.py index daa2db7d..b01491d7 100644 --- a/pyaml/magnet/magnet.py +++ b/pyaml/magnet/magnet.py @@ -13,7 +13,7 @@ from .. import PyAMLException from ..common import abstract -from ..common.element import Element +from ..common.element import Element, __pyaml_repr__ from .model import MagnetModel if TYPE_CHECKING: @@ -158,14 +158,13 @@ def get_model_name(self) -> str: """ return self.__modelName + @property + def model_name(self) -> str: + """Name used to identify this magnet in its model.""" + return self.__modelName + def __repr__(self): """ Implement the ``__repr__`` string. """ - return "%s(peer='%s', name='%s', model_name='%s', magnet_model=%s)" % ( - self.__class__.__name__, - self.attached_to(), - self.get_name(), - self.__modelName, - repr(self.__model), - ) + return __pyaml_repr__(self, exclude=["strength", "hardware"]) diff --git a/tests/test_accelerator_load.py b/tests/test_accelerator_load.py index 326f0a33..f30a68e3 100644 --- a/tests/test_accelerator_load.py +++ b/tests/test_accelerator_load.py @@ -1,7 +1,7 @@ import pytest from pydantic import BaseModel, ConfigDict -from pyaml import PyAMLConfigException +from pyaml import PyAMLConfigException, set_repr_options from pyaml.accelerator import Accelerator, ElementHolder from pyaml.common.element import Element, ElementConfigModel, __pyaml_repr__ from pyaml.control.controlsystem import ControlSystemAdapter @@ -17,6 +17,45 @@ def test_peer(): assert isinstance(tm.peer, ElementHolder) +def test_repr_is_informative_and_bounded(): + sr = Accelerator.load("tests/config/EBSOrbit.yaml") + bpm = sr.design.bpm.get("BPM_C04-04") + bpms = sr.design.bpms.get("BPM") + + assert repr(bpm) == ( + "BPM(name='BPM_C04-04', lattice_names='BPM_C04-04', " + + "peer=Simulator:design, x_pos='srdiag/bpm/c04-04/SA_HPosition', " + + "y_pos='srdiag/bpm/c04-04/SA_VPosition', " + + "x_offset='srdiag/bpm/c04-04/HOffset', " + + "y_offset='srdiag/bpm/c04-04/VOffset', " + + "tilt_name=None)" + ) + + assert repr(sr.design) == ( + f"Simulator(name='design', lattice={sr.design.lattice!r}, mat_key=None, n_elements={len(sr.design.ring)})" + ) + assert repr(sr) == "Accelerator(facility='ESRF', machine='sr', simulators=['design'], controls=['live'])" + assert len(repr(bpms)) < 250 + assert f"size={len(bpms)}" in repr(bpms) + assert f"... +{len(bpms) - 3} more ..." in repr(bpms) + + +def test_repr_options_limit_sequences(): + original = set_repr_options() + try: + set_repr_options(max_items=1) + sr = Accelerator.load("tests/config/EBSOrbit.yaml") + + bpms = sr.design.bpms.get("BPM") + assert f"... +{len(bpms) - 1} more ..." in repr(bpms) + finally: + set_repr_options( + max_items=original.max_items, + max_depth=original.max_depth, + max_length=original.max_length, + ) + + def test_accelerator_load_rejects_non_accelerator_root(tmp_path): config_file = tmp_path / "quadrupole.yaml" config_file.write_text("type: pyaml.magnet.quadrupole\nname: QF1A-C01\n", encoding="utf-8")