=
271 lines
8.8 KiB
Python
271 lines
8.8 KiB
Python
"""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] = "<unavailable>"
|
|
return summary
|
|
|
|
|
|
def _short_repr(value, limit=240):
|
|
try:
|
|
text = repr(value)
|
|
except Exception:
|
|
text = f"<unrepresentable {value.__class__.__name__}>"
|
|
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
|