120 lines
3.9 KiB
Python
120 lines
3.9 KiB
Python
"""
|
|
Logging and operation decorators for Pather
|
|
"""
|
|
from typing import TYPE_CHECKING, Any
|
|
from collections.abc import Iterator, Sequence, Callable
|
|
import logging
|
|
from functools import wraps
|
|
import inspect
|
|
import numpy
|
|
from contextlib import contextmanager
|
|
|
|
if TYPE_CHECKING:
|
|
from .pather import Pather
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
def _format_log_args(**kwargs) -> str:
|
|
arg_strs = []
|
|
for k, v in kwargs.items():
|
|
if isinstance(v, str | int | float | bool | None):
|
|
arg_strs.append(f"{k}={v}")
|
|
elif isinstance(v, numpy.ndarray):
|
|
arg_strs.append(f"{k}={v.tolist()}")
|
|
elif isinstance(v, list | tuple) and len(v) <= 10:
|
|
arg_strs.append(f"{k}={v}")
|
|
else:
|
|
arg_strs.append(f"{k}=...")
|
|
return ", ".join(arg_strs)
|
|
|
|
|
|
class PatherLogger:
|
|
"""
|
|
Encapsulates state for Pather diagnostic logging.
|
|
"""
|
|
debug: bool
|
|
indent: int
|
|
depth: int
|
|
|
|
def __init__(self, debug: bool = False) -> None:
|
|
self.debug = debug
|
|
self.indent = 0
|
|
self.depth = 0
|
|
|
|
def _log(self, module_name: str, msg: str) -> None:
|
|
if self.debug and self.depth <= 1:
|
|
log_obj = logging.getLogger(module_name)
|
|
log_obj.info(' ' * self.indent + msg)
|
|
|
|
@contextmanager
|
|
def log_operation(
|
|
self,
|
|
pather: 'Pather',
|
|
op: str,
|
|
portspec: str | Sequence[str] | None = None,
|
|
**kwargs: Any,
|
|
) -> Iterator[None]:
|
|
if not self.debug or self.depth > 0:
|
|
self.depth += 1
|
|
try:
|
|
yield
|
|
finally:
|
|
self.depth -= 1
|
|
return
|
|
|
|
target = f"({portspec})" if portspec else ""
|
|
module_name = pather.__class__.__module__
|
|
self._log(module_name, f"Operation: {op}{target} {_format_log_args(**kwargs)}")
|
|
|
|
before_ports = {name: port.copy() for name, port in pather.ports.items()}
|
|
self.depth += 1
|
|
self.indent += 1
|
|
|
|
try:
|
|
yield
|
|
finally:
|
|
after_ports = pather.ports
|
|
for name in sorted(after_ports.keys()):
|
|
if name not in before_ports or after_ports[name] != before_ports[name]:
|
|
self._log(module_name, f"Port {name}: {pather.ports[name].describe()}")
|
|
for name in sorted(before_ports.keys()):
|
|
if name not in after_ports:
|
|
self._log(module_name, f"Port {name}: removed")
|
|
|
|
self.indent -= 1
|
|
self.depth -= 1
|
|
|
|
|
|
def logged_op(
|
|
portspec_getter: Callable[[dict[str, Any]], str | Sequence[str] | None] | None = None,
|
|
) -> Callable[[Callable[..., Any]], Callable[..., Any]]:
|
|
"""
|
|
Decorator to wrap Pather methods with logging.
|
|
"""
|
|
def decorator(func: Callable[..., Any]) -> Callable[..., Any]:
|
|
sig = inspect.signature(func)
|
|
|
|
@wraps(func)
|
|
def wrapper(self: 'Pather', *args: Any, **kwargs: Any) -> Any:
|
|
logger_obj = getattr(self, '_logger', None)
|
|
if logger_obj is None or not logger_obj.debug:
|
|
return func(self, *args, **kwargs)
|
|
|
|
bound = sig.bind(self, *args, **kwargs)
|
|
bound.apply_defaults()
|
|
all_args = bound.arguments
|
|
# remove 'self' from logged args
|
|
logged_args = {k: v for k, v in all_args.items() if k != 'self'}
|
|
|
|
ps = portspec_getter(all_args) if portspec_getter else None
|
|
|
|
# Remove portspec from logged_args if it's there to avoid duplicate arg to log_operation
|
|
logged_args.pop('portspec', None)
|
|
|
|
with logger_obj.log_operation(self, func.__name__, ps, **logged_args):
|
|
if getattr(self, '_dead', False) and func.__name__ in ('plug', 'place'):
|
|
logger.warning(f"Skipping geometry for {func.__name__}() since device is dead")
|
|
return func(self, *args, **kwargs)
|
|
return wrapper
|
|
return decorator
|