Skip to content
Merged
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
3 changes: 2 additions & 1 deletion pyaml/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -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")
Expand Down
10 changes: 10 additions & 0 deletions pyaml/accelerator.py
Original file line number Diff line number Diff line change
Expand Up @@ -96,6 +96,8 @@ class Accelerator(DynamicValidation):
:attr:`live` and :attr:`design` properties.
"""

__pyaml_repr_exclude__ = ("description", "yellow_pages")

def __init__(
self,
facility: str,
Expand Down Expand Up @@ -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.
Expand Down
13 changes: 12 additions & 1 deletion pyaml/arrays/element_array.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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.
Expand Down
16 changes: 8 additions & 8 deletions pyaml/bpm/bpm.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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:
"""
Expand All @@ -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]:
"""
Expand All @@ -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):
"""
Expand Down
241 changes: 196 additions & 45 deletions pyaml/common/element.py
Original file line number Diff line number Diff line change
@@ -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"<unavailable: {error.__class__.__name__}>"


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"<error: {e}>"
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"<error: {e}>"
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"<recursive {obj.__class__.__name__}>"

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):
Expand Down Expand Up @@ -140,6 +289,8 @@ class Element:
Perform post-construction initialization after attachment.
"""

__pyaml_repr_exclude__ = ("description",)

def __init__(
self,
name: str,
Expand Down
5 changes: 4 additions & 1 deletion pyaml/common/holders/generic_array_holder.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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)
Loading
Loading