"""
Base utilities and decorators for the scene edition API.
This module provides foundational components used across all editor classes,
including logging decorators, value limit checking, and referential validation.
"""
from __future__ import annotations
import functools
import inspect
import logging
from pathlib import Path
from typing import Any, Concatenate, ParamSpec, TypeVar, TypedDict, get_args, Literal
from collections.abc import Callable
from mimic_scene_edition.core.limits import DEFAULT_VALUE_LIMITS
from mimic_data_api.actor import Actor
from mimic_data_api.scene import Scene
from mimic_data_api.trajectory_numpy import WorldTrajectory
logger = logging.getLogger(__name__)
P = ParamSpec("P")
C = TypeVar("C")
R = TypeVar("R")
Referential = Literal["local", "world"]
"""Type alias for referential frame specification.
- ``"local"``: Operations are performed in the object's local coordinate frame.
- ``"world"``: Operations are performed in the world coordinate frame.
"""
Axis = Literal["x", "y", "z"]
"""Type alias for axis specification.
- ``"x"``: Forward axis in vehicle convention.
- ``"y"``: Left axis in vehicle convention.
- ``"z"``: Up axis in vehicle convention.
"""
VALID_REFERENTIALS: tuple[str, ...] = get_args(Referential)
"""Tuple of valid referential values for validation."""
[docs]
class EditionDict(TypedDict):
"""Dictionary structure for recording edit operations.
Attributes
----------
method : str
The name of the method that was called.
params : dict[str, Any]
The parameters passed to the method.
description : str
A human-readable description of the edit operation.
Examples
--------
>>> edition = EditionDict(
... method="shift",
... params={"forward": 5.0, "left": 2.0},
... description="Shift: forward=5.0, left=2.0"
... )
"""
method: str
params: dict[str, Any]
description: str
def _bind_args(func, args, kwargs) -> dict[str, Any]:
"""Bind function arguments to parameter names.
Parameters
----------
func : Callable
The function whose signature to use.
args : tuple
Positional arguments passed to the function.
kwargs : dict
Keyword arguments passed to the function.
Returns
-------
dict[str, Any]
Dictionary mapping parameter names to their values,
excluding 'self'.
"""
sig = inspect.signature(func)
ba = sig.bind_partial(None, *args, **kwargs)
ba.apply_defaults()
return {k: v for k, v in ba.arguments.items() if k != "self"}
def _fmt_value(v: Any) -> str:
"""Format a value for display in edition logs.
Provides human-readable representations of common types
used in the scene edition API.
Parameters
----------
v : Any
The value to format.
Returns
-------
str
A formatted string representation of the value.
Examples
--------
>>> _fmt_value(Path("/home/user/file.txt"))
"'file.txt'"
>>> _fmt_value(42)
'42'
>>> _fmt_value("hello")
"'hello'"
"""
if isinstance(v, Path):
return f"'{v.name}'"
if isinstance(v, (int, float, bool)) or v is None:
return repr(v)
if isinstance(v, str):
return f"'{v}'"
if isinstance(v, Actor):
return f"Actor(name={getattr(v, 'name', None)!r})"
if isinstance(v, Scene):
return f"Scene(name={getattr(v, 'name', None)!r})"
if hasattr(v, "object2world") and hasattr(v, "timestamps"):
try:
n = len(getattr(v, "timestamps"))
T = getattr(v, "object2world")
shape = getattr(T, "shape", None)
return f"WorldTrajectory(len={n}, T_shape={shape})"
except Exception:
return "WorldTrajectory(...)"
return f"{type(v).__name__}(...)"
# Add specific methods known to modify state but not return self
_MUTATING_NON_FLUENT = {
"insert_actor",
"duplicate_actor",
"remove_actor",
"remove_all_actors",
"insert_actor_on_different_actor_trajectory",
"insert_actor_on_ego_trajectory",
"insert_and_stop_actor_on_ego_trajectory",
}
[docs]
def log_method(func: Callable[Concatenate[C, P], R]) -> Callable[Concatenate[C, P], R]:
"""Decorator that logs method calls to the editions list.
This decorator automatically records method invocations to the
``_editions`` list on the object, enabling tracking of all
modifications made to scene elements.
The decorator only logs if:
- The method returns ``self`` (fluent interface pattern), or
- The method name is in ``_MUTATING_NON_FLUENT``.
Parameters
----------
func : Callable
The method to wrap.
Returns
-------
Callable
The wrapped method with logging behavior.
Notes
-----
The object must have a ``_editions`` attribute that is a list
of :class:`EditionDict` objects.
Examples
--------
>>> class MyEditor:
... def __init__(self):
... self._editions = []
...
... @log_method
... def shift(self, forward: float = 0.0) -> "MyEditor":
... # ... implementation ...
... return self
"""
@functools.wraps(func)
def wrapper(self: C, *args: P.args, **kwargs: P.kwargs) -> R:
editions_list = getattr(self, "_editions", None)
pre_count = len(editions_list) if isinstance(editions_list, list) else 0
try:
result = func(self, *args, **kwargs)
except Exception:
logger.exception("Error during method call %s", func.__name__)
raise
should_log = False
post_count = len(editions_list) if isinstance(editions_list, list) else 0
if post_count > pre_count:
should_log = False
elif result is self or func.__name__ in _MUTATING_NON_FLUENT:
should_log = True
if should_log and isinstance(editions_list, list):
try:
bound = _bind_args(func, args, kwargs)
pretty = ", ".join(f"{k}={_fmt_value(v)}" for k, v in bound.items())
action = func.__name__.replace("_", " ").title()
desc = f"{action}" + (f": {pretty}" if pretty else "")
editions_list.append(
{"method": func.__name__, "params": bound, "description": desc}
)
except Exception:
logger.debug(
"Failed to append edition for %s", func.__name__, exc_info=True
)
return result
return wrapper
[docs]
def check_value_limits(
func: Callable[Concatenate[C, P], R],
) -> Callable[Concatenate[C, P], R]:
"""Decorator that validates parameter values against defined limits.
When ``limited_mode`` is enabled on the object, this decorator
checks that numeric parameters fall within the bounds specified
in ``value_limits``.
Parameters
----------
func : Callable
The method to wrap.
Returns
-------
Callable
The wrapped method with value limit checking.
Raises
------
ValueError
If a parameter value is outside its defined limits.
Notes
-----
The object must have:
- ``limited_mode``: bool attribute enabling/disabling checks
- ``value_limits``: dict mapping parameter names to (min, max) tuples
Examples
--------
>>> class MyEditor:
... def __init__(self):
... self.limited_mode = True
... self.value_limits = {"forward": (-10.0, 10.0)}
...
... @check_value_limits
... def shift(self, forward: float = 0.0) -> "MyEditor":
... # ... implementation ...
... return self
...
>>> editor = MyEditor()
>>> editor.shift(forward=15.0) # Raises ValueError
"""
sig = inspect.signature(func)
@functools.wraps(func)
def w(self: C, *args: P.args, **kwargs: P.kwargs) -> R:
if not getattr(
self, "limited_mode", False
): # (always) TODO add limited mode to classes that use this callable
return func(self, *args, **kwargs)
try:
b = sig.bind_partial(self, *args, **kwargs)
b.apply_defaults()
params = {k: v for k, v in b.arguments.items() if k != "self"}
except TypeError:
return func(self, *args, **kwargs)
resolver = getattr(self, "_resolve_value_limits", None)
if resolver is not None:
limits = resolver(params)
else:
limits = getattr(self, "value_limits", DEFAULT_VALUE_LIMITS)
for name, (mn, mx) in limits.items():
if name in params and isinstance(params[name], (int, float)):
val = float(params[name])
if not (mn <= val <= mx):
raise ValueError(
f"Parameter '{name}'={val} outside [{mn},{mx}] in {func.__name__}"
)
return func(self, *args, **kwargs)
return w
[docs]
def validate_referential(param: str = "referential") -> Callable[Concatenate[C, P], R]:
"""Decorator factory that validates a referential parameter.
Creates a decorator that ensures the specified parameter
has a valid referential value (``"local"`` or ``"world"``).
Parameters
----------
param : str, optional
Name of the parameter to validate. Default is ``"referential"``.
Returns
-------
Callable
A decorator that validates the referential parameter.
Raises
------
ValueError
If the referential value is not in :data:`VALID_REFERENTIALS`.
Examples
--------
>>> @validate_referential("referential")
... def transform(self, *, referential: str = "local") -> "MyEditor":
... # ... implementation ...
... return self
"""
def deco(func: Callable[Concatenate[C, P], R]) -> Callable[Concatenate[C, P], R]:
sig = inspect.signature(func)
@functools.wraps(func)
def w(self: C, *args: P.args, **kwargs: P.kwargs) -> R:
ba = sig.bind_partial(self, *args, **kwargs)
ba.apply_defaults()
ref = ba.arguments.get(param)
if ref not in VALID_REFERENTIALS:
raise ValueError(
f"Invalid referential '{ref}'. Expected one of {VALID_REFERENTIALS}."
)
return func(self, *args, **kwargs)
return w
return deco