"""Metadata recording helpers for component generation classes.""" import inspect import json import re import sys from datetime import datetime from functools import wraps from pathlib import Path PROJECT_ROOT = Path(__file__).resolve().parents[2] DEFAULT_METADATA_DIR = PROJECT_ROOT / "metafile" _RECORDER_MARK = "_mxpic_generation_metadata_wrapped" _INTERNAL_PREFIX = "_mxpic_generation_" _AUTO_METADATA_ENABLED = True _AUTO_METADATA_DIR = DEFAULT_METADATA_DIR _AUTO_METADATA_INCLUDE_STATE = True _AUTO_METADATA_INCLUDE_PRIVATE = False _CONSTRUCTION_DEPTH = 0 def _json_safe(value, depth=0): """Return a JSON-friendly representation without walking huge objects.""" if value is None or isinstance(value, (bool, int, float, str)): return value if isinstance(value, Path): return str(value) if depth >= 4: return _short_repr(value) if isinstance(value, (list, tuple)): return [_json_safe(item, depth + 1) for item in value] if isinstance(value, set): return [_json_safe(item, depth + 1) for item in sorted(value, key=repr)] if isinstance(value, dict): return { str(_json_safe(key, depth + 1)): _json_safe(item, depth + 1) for key, item in value.items() } return _object_summary(value) def _object_summary(value): summary = { "type": f"{value.__class__.__module__}.{value.__class__.__name__}", "repr": _short_repr(value), } for attr in ("name", "cell_name", "basename", "length", "width"): if hasattr(value, attr): try: summary[attr] = _json_safe(getattr(value, attr), depth=4) except Exception: summary[attr] = "" return summary def _short_repr(value, limit=240): try: text = repr(value) except Exception: text = f"" if len(text) > limit: return text[: limit - 3] + "..." return text def _sanitize_filename_part(value): text = str(value or "") text = re.sub(r"[^A-Za-z0-9_.-]+", "_", text).strip("._") return text or "unnamed" def _capture_init_parameters(init, self, args, kwargs): try: signature = inspect.signature(init) bound = signature.bind(self, *args, **kwargs) bound.apply_defaults() except Exception: return { "args": _json_safe(args), "kwargs": _json_safe(kwargs), } return { name: _json_safe(value) for name, value in bound.arguments.items() if name != "self" } def get_generation_metadata(self, include_state=True, include_private=False): """Return the recorded generation metadata for this component instance.""" metadata = { "class": self.__class__.__name__, "module": self.__class__.__module__, "created_at": getattr(self, "_mxpic_generation_created_at", None), "init_parameters": getattr(self, "_mxpic_generation_parameters", {}), } metadata_file = getattr(self, "_mxpic_generation_metadata_file", None) if metadata_file is not None: metadata["metadata_file"] = metadata_file metadata_error = getattr(self, "_mxpic_generation_metadata_error", None) if metadata_error is not None: metadata["metadata_error"] = metadata_error if include_state: metadata["state"] = _collect_state(self, include_private=include_private) return metadata def _collect_state(instance, include_private=False): state = {} for key, value in getattr(instance, "__dict__", {}).items(): if key.startswith(_INTERNAL_PREFIX): continue if not include_private and key.startswith("_"): continue if callable(value): continue state[key] = _json_safe(value) return state def save_generation_metadata( self, folder=None, filename=None, include_state=True, include_private=False, ): """Write generation metadata to JSON and return the saved file path.""" folder_path = Path(folder) if folder is not None else DEFAULT_METADATA_DIR folder_path.mkdir(parents=True, exist_ok=True) metadata = self.get_generation_metadata( include_state=include_state, include_private=include_private, ) if filename is None: params = metadata.get("init_parameters", {}) name = params.get("name") or getattr(self, "name", None) suffix = datetime.now().strftime("%Y%m%d-%H%M%S-%f") filename = ( f"{self.__class__.__name__}_" f"{_sanitize_filename_part(name)}_" f"{suffix}.json" ) elif not str(filename).lower().endswith(".json"): filename = f"{filename}.json" file_path = folder_path / filename file_path.write_text(json.dumps(metadata, indent=4), encoding="utf-8") self._mxpic_generation_metadata_file = str(file_path) return file_path def set_generation_metadata_auto_save( enabled=True, folder=None, include_state=True, include_private=False, ): """Configure automatic metadata JSON writes after component construction.""" global _AUTO_METADATA_ENABLED global _AUTO_METADATA_DIR global _AUTO_METADATA_INCLUDE_STATE global _AUTO_METADATA_INCLUDE_PRIVATE _AUTO_METADATA_ENABLED = bool(enabled) if folder is not None: _AUTO_METADATA_DIR = Path(folder) _AUTO_METADATA_INCLUDE_STATE = bool(include_state) _AUTO_METADATA_INCLUDE_PRIVATE = bool(include_private) def get_generation_metadata_auto_save_config(): """Return the current automatic metadata write configuration.""" return { "enabled": _AUTO_METADATA_ENABLED, "folder": str(_AUTO_METADATA_DIR), "include_state": _AUTO_METADATA_INCLUDE_STATE, "include_private": _AUTO_METADATA_INCLUDE_PRIVATE, } def install_generation_metadata_recorders(package_prefixes=None): """Attach metadata methods to generation classes already imported.""" if package_prefixes is None: package_prefixes = ( "mxpic_forge_new.components", ) installed = [] for module in list(sys.modules.values()): module_name = getattr(module, "__name__", "") if not module_name.startswith(package_prefixes): continue for _, cls in inspect.getmembers(module, inspect.isclass): if not getattr(cls, "__module__", "").startswith(package_prefixes): continue if _wrap_generation_class(cls): installed.append(cls) return installed def _wrap_generation_class(cls): if cls.__dict__.get(_RECORDER_MARK, False): return False init = cls.__dict__.get("__init__") if init is None: _attach_methods(cls) setattr(cls, _RECORDER_MARK, True) return True @wraps(init) def wrapped_init(self, *args, **kwargs): global _CONSTRUCTION_DEPTH captured = _capture_init_parameters(init, self, args, kwargs) depth = getattr(self, "_mxpic_generation_record_depth", 0) is_outermost_construction = _CONSTRUCTION_DEPTH == 0 _CONSTRUCTION_DEPTH += 1 self._mxpic_generation_record_depth = depth + 1 try: result = init(self, *args, **kwargs) finally: _restore_record_depth(self, depth) _CONSTRUCTION_DEPTH -= 1 if depth == 0: self._mxpic_generation_parameters = captured self._mxpic_generation_created_at = datetime.now().isoformat(timespec="seconds") if is_outermost_construction: _auto_save_generation_metadata(self) return result cls.__init__ = wrapped_init _attach_methods(cls) setattr(cls, _RECORDER_MARK, True) return True def _restore_record_depth(instance, depth): if depth: instance._mxpic_generation_record_depth = depth elif hasattr(instance, "_mxpic_generation_record_depth"): delattr(instance, "_mxpic_generation_record_depth") def _auto_save_generation_metadata(instance): if not _AUTO_METADATA_ENABLED: return try: save_generation_metadata( instance, folder=_AUTO_METADATA_DIR, include_state=_AUTO_METADATA_INCLUDE_STATE, include_private=_AUTO_METADATA_INCLUDE_PRIVATE, ) if hasattr(instance, "_mxpic_generation_metadata_error"): delattr(instance, "_mxpic_generation_metadata_error") except Exception as exc: instance._mxpic_generation_metadata_error = f"{exc.__class__.__name__}: {exc}" def _attach_methods(cls): if not hasattr(cls, "get_generation_metadata"): cls.get_generation_metadata = get_generation_metadata if not hasattr(cls, "save_generation_metadata"): cls.save_generation_metadata = save_generation_metadata if not hasattr(cls, "record_generation_metadata"): cls.record_generation_metadata = save_generation_metadata