747 lines
27 KiB
Python
747 lines
27 KiB
Python
"""
|
|
NodeSlicer Debug Module
|
|
=======================
|
|
Provides tools to capture, serialize, and dump node execution state (inputs, properties, outputs, and errors)
|
|
for immediate review, automated diagnostics, and pair-programming debugging.
|
|
|
|
Usage:
|
|
# 1. Decorate a Node class:
|
|
from ...infra.debug import debug
|
|
|
|
@debug
|
|
@register_class
|
|
class SLICENODE_SeamPlacement(SliceNode):
|
|
...
|
|
|
|
# 2. Decorate evaluate_output or any method/function:
|
|
@debug
|
|
def evaluate_output(self, socket_name: str, **kwargs):
|
|
...
|
|
|
|
# 3. Retrieve or inspect the last execution dump:
|
|
from NodeSlicer.infra.debug import get_last_debug_dump, get_last_debug_text, load_last_inputs
|
|
dump = get_last_debug_dump()
|
|
print(get_last_debug_text())
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import sys
|
|
import json
|
|
import time
|
|
import pickle
|
|
import inspect
|
|
import datetime
|
|
import traceback
|
|
import functools
|
|
from pathlib import Path
|
|
from typing import Any, Callable, TypeVar, cast
|
|
import numpy as np
|
|
|
|
# Global configuration
|
|
_DEFAULT_DUMP_DIR: Path | None = None
|
|
_LAST_DUMP_DICT: dict[str, Any] | None = None
|
|
_LAST_DUMP_TEXT: str | None = None
|
|
_GLOBAL_DEBUG_ENABLED: bool = False
|
|
|
|
|
|
def get_debug_dump_dir() -> Path:
|
|
"""Returns the directory where debug dumps are stored, creating it if necessary."""
|
|
global _DEFAULT_DUMP_DIR
|
|
if _DEFAULT_DUMP_DIR is not None:
|
|
_DEFAULT_DUMP_DIR.mkdir(parents=True, exist_ok=True)
|
|
return _DEFAULT_DUMP_DIR
|
|
|
|
# Try project root
|
|
project_root = Path(__file__).resolve().parent.parent
|
|
dump_dir = project_root / "debug_dumps"
|
|
try:
|
|
dump_dir.mkdir(parents=True, exist_ok=True)
|
|
_DEFAULT_DUMP_DIR = dump_dir
|
|
return _DEFAULT_DUMP_DIR
|
|
except Exception:
|
|
# Fallback to current working directory or user home
|
|
fallback = Path.cwd() / "debug_dumps"
|
|
fallback.mkdir(parents=True, exist_ok=True)
|
|
_DEFAULT_DUMP_DIR = fallback
|
|
return _DEFAULT_DUMP_DIR
|
|
|
|
|
|
def set_debug_dump_dir(path: str | Path) -> None:
|
|
"""Sets a custom directory for debug dumps."""
|
|
global _DEFAULT_DUMP_DIR
|
|
p = Path(path).resolve()
|
|
p.mkdir(parents=True, exist_ok=True)
|
|
_DEFAULT_DUMP_DIR = p
|
|
|
|
|
|
def set_global_debug(enabled: bool = True) -> None:
|
|
"""Enables or disables global debug capture for all nodes."""
|
|
global _GLOBAL_DEBUG_ENABLED
|
|
_GLOBAL_DEBUG_ENABLED = enabled
|
|
|
|
|
|
class debug_session:
|
|
"""Context manager to enable debug capture for a block of execution."""
|
|
def __init__(self, dump_dir: str | Path | None = None):
|
|
self.prev_enabled = _GLOBAL_DEBUG_ENABLED
|
|
self.custom_dir = dump_dir
|
|
self.prev_dir = _DEFAULT_DUMP_DIR
|
|
|
|
def __enter__(self):
|
|
global _GLOBAL_DEBUG_ENABLED
|
|
_GLOBAL_DEBUG_ENABLED = True
|
|
if self.custom_dir is not None:
|
|
set_debug_dump_dir(self.custom_dir)
|
|
return self
|
|
|
|
def __exit__(self, exc_type, exc_val, exc_tb):
|
|
global _GLOBAL_DEBUG_ENABLED, _DEFAULT_DUMP_DIR
|
|
_GLOBAL_DEBUG_ENABLED = self.prev_enabled
|
|
if self.custom_dir is not None:
|
|
_DEFAULT_DUMP_DIR = self.prev_dir
|
|
|
|
|
|
def _serialize_for_json(val: Any, max_depth: int = 4) -> Any:
|
|
"""Recursively converts objects to JSON-serializable structures with rich summaries."""
|
|
if max_depth <= 0:
|
|
return f"<max_depth_reached: {type(val).__name__}>"
|
|
|
|
if val is None or isinstance(val, (bool, int, str)):
|
|
return val
|
|
|
|
if isinstance(val, float):
|
|
if np.isnan(val):
|
|
return "NaN"
|
|
if np.isneginf(val):
|
|
return "-Infinity"
|
|
if np.isposinf(val):
|
|
return "Infinity"
|
|
return round(val, 6)
|
|
|
|
val_cls_name = type(val).__name__
|
|
|
|
# SlicePaths (which is a list subclass)
|
|
if val_cls_name == "SlicePaths" or (hasattr(val, "layer_height") and hasattr(val, "extrusion_rate")):
|
|
num_paths = len(val)
|
|
total_verts = sum(len(p) for p in val) if num_paths > 0 else 0
|
|
min_bounds: list[float] | None = None
|
|
max_bounds: list[float] | None = None
|
|
|
|
if total_verts > 0:
|
|
try:
|
|
all_pts = np.concatenate([p[:, :3] for p in val if len(p) > 0 and p.ndim == 2], axis=0)
|
|
if len(all_pts) > 0:
|
|
min_bounds = [float(v) for v in np.min(all_pts, axis=0)]
|
|
max_bounds = [float(v) for v in np.max(all_pts, axis=0)]
|
|
except Exception:
|
|
pass
|
|
|
|
return {
|
|
"__type__": "SlicePaths",
|
|
"num_paths": num_paths,
|
|
"total_vertices": total_verts,
|
|
"layer_height": getattr(val, "layer_height", None),
|
|
"first_layer_height": getattr(val, "first_layer_height", None),
|
|
"name": getattr(val, "name", ""),
|
|
"transforms": getattr(val, "transforms", []),
|
|
"groups_count": len(getattr(val, "groups", [])),
|
|
"extra_attributes": list(getattr(val, "extra_attributes", {}).keys()),
|
|
"bounding_box_min": min_bounds,
|
|
"bounding_box_max": max_bounds,
|
|
"path_vertex_counts": [len(p) for p in val[:10]] + (["..."] if num_paths > 10 else [])
|
|
}
|
|
|
|
# SlicingObject / SlicingCollection / SlicingGroup
|
|
if val_cls_name in ("SlicingObject", "SlicingCollection", "SlicingGroup") or hasattr(val, "mesh") or hasattr(val, "objects") or hasattr(val, "collections"):
|
|
info = {
|
|
"__type__": val_cls_name,
|
|
"name": getattr(val, "name", ""),
|
|
"transforms": getattr(val, "transforms", []),
|
|
}
|
|
if hasattr(val, "mesh") and isinstance(val.mesh, np.ndarray):
|
|
info["triangles_count"] = len(val.mesh)
|
|
info["mesh_shape"] = list(val.mesh.shape)
|
|
if val.mesh.size > 0:
|
|
info["min_bounds"] = [getattr(val, "min_x", None), getattr(val, "min_y", None), getattr(val, "min_z", None)]
|
|
info["max_bounds"] = [getattr(val, "max_x", None), getattr(val, "max_y", None), getattr(val, "max_z", None)]
|
|
if hasattr(val, "objects") and isinstance(val.objects, list):
|
|
info["objects_count"] = len(val.objects)
|
|
info["object_names"] = [getattr(o, "name", str(idx)) for idx, o in enumerate(val.objects)]
|
|
if hasattr(val, "collections") and isinstance(val.collections, dict):
|
|
info["collections_count"] = len(val.collections)
|
|
info["collection_names"] = list(val.collections.keys())
|
|
return info
|
|
|
|
if isinstance(val, (list, tuple)):
|
|
# If it's a small list of simple numbers/strings
|
|
if len(val) <= 10 and all(isinstance(x, (int, float, bool, str)) for x in val):
|
|
return [_serialize_for_json(x, max_depth - 1) for x in val]
|
|
elif len(val) > 20:
|
|
return {
|
|
"__type__": type(val).__name__,
|
|
"length": len(val),
|
|
"preview": [_serialize_for_json(x, max_depth - 1) for x in val[:5]],
|
|
"summary": f"{type(val).__name__} with {len(val)} items"
|
|
}
|
|
return [_serialize_for_json(x, max_depth - 1) for x in val]
|
|
|
|
if isinstance(val, dict):
|
|
return {str(k): _serialize_for_json(v, max_depth - 1) for k, v in val.items()}
|
|
|
|
# NumPy Array
|
|
if isinstance(val, np.ndarray):
|
|
shape = list(val.shape)
|
|
dtype = str(val.dtype)
|
|
info: dict[str, Any] = {
|
|
"__type__": "numpy.ndarray",
|
|
"shape": shape,
|
|
"dtype": dtype,
|
|
"size": int(val.size)
|
|
}
|
|
if val.size > 0:
|
|
try:
|
|
if np.issubdtype(val.dtype, np.number):
|
|
info["min"] = float(np.min(val))
|
|
info["max"] = float(np.max(val))
|
|
info["mean"] = float(np.mean(val))
|
|
except Exception:
|
|
pass
|
|
|
|
if val.size <= 12:
|
|
info["data"] = val.tolist()
|
|
elif val.ndim == 2 and val.shape[1] in (2, 3) and val.shape[0] <= 10:
|
|
info["data"] = val.tolist()
|
|
else:
|
|
flat = val.ravel()
|
|
info["sample_first"] = flat[:4].tolist()
|
|
info["sample_last"] = flat[-4:].tolist()
|
|
return info
|
|
|
|
# Blender ID Object
|
|
if hasattr(val, "name") and hasattr(val, "type") and hasattr(val, "rna_type"):
|
|
return {
|
|
"__type__": f"bpy.types.{type(val).__name__}",
|
|
"name": getattr(val, "name", ""),
|
|
"type": getattr(val, "type", ""),
|
|
}
|
|
|
|
# Fallback to string representation
|
|
return f"<{val_cls_name}: {str(val)[:200]}>"
|
|
|
|
|
|
def _sanitize_for_pickle(val: Any) -> Any:
|
|
"""Ensures complex objects with Blender C-types can be pickled safely."""
|
|
try:
|
|
pickle.dumps(val)
|
|
return val
|
|
except Exception:
|
|
# If the object fails to pickle directly, convert to a clean dictionary / numpy structure
|
|
return _serialize_for_json(val)
|
|
|
|
|
|
def extract_node_properties(node: Any) -> dict[str, Any]:
|
|
"""Extracts custom user properties and settings defined on a node."""
|
|
props: dict[str, Any] = {}
|
|
ignored = {
|
|
"bl_idname", "bl_label", "bl_description", "bl_icon", "bl_width_default", "bl_width_min", "bl_width_max",
|
|
"bl_height_default", "bl_height_min", "bl_height_max", "inputs", "outputs", "dimensions", "location",
|
|
"width", "height", "select", "hide", "mute", "color", "use_custom_color", "show_options", "show_preview",
|
|
"show_texture", "type", "rna_type", "internal_links", "parent", "label", "_data", "_current_cache",
|
|
"_current_visited", "name"
|
|
}
|
|
|
|
# Check class annotations
|
|
cls = type(node)
|
|
annotations = getattr(cls, "__annotations__", {})
|
|
for key in annotations:
|
|
if key not in ignored and not key.startswith("_"):
|
|
try:
|
|
props[key] = getattr(node, key, None)
|
|
except Exception:
|
|
pass
|
|
|
|
# Check custom properties in dict
|
|
for key, val in getattr(node, "__dict__", {}).items():
|
|
if key not in ignored and not key.startswith("_"):
|
|
try:
|
|
props[key] = getattr(node, key, val)
|
|
except Exception:
|
|
pass
|
|
|
|
return props
|
|
|
|
|
|
def extract_node_inputs(node: Any, evaluated_socket: str = "") -> dict[str, Any]:
|
|
"""Resolves and captures all input sockets defined on the node."""
|
|
inputs_data: dict[str, Any] = {}
|
|
if not hasattr(node, "inputs") or node.inputs is None:
|
|
return inputs_data
|
|
|
|
for sock in node.inputs:
|
|
sock_name = sock.name
|
|
identifier = getattr(sock, "identifier", sock_name)
|
|
is_linked = getattr(sock, "is_linked", False)
|
|
|
|
source_info = None
|
|
if is_linked and hasattr(sock, "links") and sock.links:
|
|
try:
|
|
from_sock = sock.links[0].from_socket
|
|
source_info = {
|
|
"node": from_sock.node.name if from_sock.node else "Unknown",
|
|
"socket": from_sock.name
|
|
}
|
|
except Exception:
|
|
source_info = "linked"
|
|
|
|
# Resolve value
|
|
val = None
|
|
resolved = False
|
|
error_msg = None
|
|
|
|
if hasattr(node, "resolve"):
|
|
try:
|
|
val = node.resolve(sock_name)
|
|
resolved = True
|
|
except Exception as e:
|
|
error_msg = f"Resolution error: {e}"
|
|
|
|
if not resolved and hasattr(sock, "value"):
|
|
try:
|
|
val = sock.value
|
|
resolved = True
|
|
except Exception:
|
|
pass
|
|
|
|
if not resolved and hasattr(sock, "default_value"):
|
|
try:
|
|
val = sock.default_value
|
|
resolved = True
|
|
except Exception:
|
|
pass
|
|
|
|
inputs_data[sock_name] = {
|
|
"identifier": identifier,
|
|
"bl_idname": getattr(sock, "bl_idname", type(sock).__name__),
|
|
"is_linked": is_linked,
|
|
"source": source_info,
|
|
"value": val,
|
|
"error": error_msg
|
|
}
|
|
|
|
return inputs_data
|
|
|
|
|
|
def format_debug_report(dump: dict[str, Any]) -> str:
|
|
"""Formats a debug dump dictionary into a clean, human-readable text report."""
|
|
lines: list[str] = []
|
|
lines.append("=" * 80)
|
|
lines.append(f" NODESLICER DEBUG REPORT: {dump.get('node_name', 'Unknown Node')} ({dump.get('node_type', '')})")
|
|
lines.append("=" * 80)
|
|
lines.append(f"Timestamp: {dump.get('timestamp', '')}")
|
|
lines.append(f"Output Evaluated: {dump.get('socket_evaluated', '')}")
|
|
lines.append(f"Duration: {dump.get('duration_seconds', 0.0) * 1000.0:.2f} ms")
|
|
lines.append(f"Status: {dump.get('status', 'OK')}")
|
|
|
|
if dump.get("exception"):
|
|
lines.append("")
|
|
lines.append("!" * 80)
|
|
lines.append(f" EXCEPTION: {dump.get('exception')}")
|
|
lines.append("!" * 80)
|
|
if dump.get("traceback"):
|
|
lines.append(dump["traceback"])
|
|
|
|
lines.append("")
|
|
lines.append("-" * 40 + " PROPERTIES " + "-" * 40)
|
|
props = dump.get("properties", {})
|
|
if props:
|
|
for k, v in props.items():
|
|
lines.append(f" • {k}: {v}")
|
|
else:
|
|
lines.append(" (None)")
|
|
|
|
lines.append("")
|
|
lines.append("-" * 40 + " INPUT SOCKETS " + "-" * 40)
|
|
inputs = dump.get("inputs", {})
|
|
if inputs:
|
|
for sock_name, sinfo in inputs.items():
|
|
linked_str = f" [Linked from {sinfo.get('source', {}).get('node', '')}.{sinfo.get('source', {}).get('socket', '')}]" if sinfo.get("is_linked") else " [Unlinked]"
|
|
lines.append(f" ▶ Socket '{sock_name}' ({sinfo.get('bl_idname', '')}){linked_str}:")
|
|
if sinfo.get("error"):
|
|
lines.append(f" ERROR: {sinfo['error']}")
|
|
val_summary = sinfo.get("value_summary")
|
|
if isinstance(val_summary, dict):
|
|
for vk, vv in val_summary.items():
|
|
lines.append(f" - {vk}: {vv}")
|
|
else:
|
|
lines.append(f" - Value: {val_summary}")
|
|
else:
|
|
lines.append(" (No input sockets)")
|
|
|
|
lines.append("")
|
|
lines.append("-" * 40 + " OUTPUT SUMMARY " + "-" * 40)
|
|
out_summary = dump.get("output_summary")
|
|
if isinstance(out_summary, dict):
|
|
for ok, ov in out_summary.items():
|
|
lines.append(f" • {ok}: {ov}")
|
|
else:
|
|
lines.append(f" • Result: {out_summary}")
|
|
|
|
lines.append("=" * 80)
|
|
return "\n".join(lines)
|
|
|
|
|
|
def dump_node_execution(
|
|
node: Any,
|
|
socket_name: str,
|
|
output_val: Any = None,
|
|
duration_s: float = 0.0,
|
|
exception: Exception | None = None,
|
|
tb_str: str | None = None,
|
|
save_pickle: bool = True,
|
|
log_console: bool = True
|
|
) -> dict[str, Any]:
|
|
"""
|
|
Captures complete execution details of a node, writes JSON/TXT/PKL dumps,
|
|
and updates the in-memory last execution record.
|
|
"""
|
|
global _LAST_DUMP_DICT, _LAST_DUMP_TEXT
|
|
|
|
now_iso = datetime.datetime.now().isoformat()
|
|
now_tag = datetime.datetime.now().strftime("%Y%m%d_%H%M%S_%f")[:19]
|
|
node_name = getattr(node, "name", type(node).__name__)
|
|
node_type = getattr(node, "bl_idname", type(node).__name__)
|
|
|
|
# Extract properties & inputs
|
|
props = extract_node_properties(node)
|
|
inputs_raw = extract_node_inputs(node, socket_name)
|
|
|
|
# Prepare JSON serializable inputs
|
|
inputs_summary: dict[str, Any] = {}
|
|
raw_inputs_dict: dict[str, Any] = {}
|
|
for sname, sinfo in inputs_raw.items():
|
|
val = sinfo.get("value")
|
|
raw_inputs_dict[sname] = val
|
|
inputs_summary[sname] = {
|
|
"identifier": sinfo.get("identifier"),
|
|
"bl_idname": sinfo.get("bl_idname"),
|
|
"is_linked": sinfo.get("is_linked"),
|
|
"source": sinfo.get("source"),
|
|
"error": sinfo.get("error"),
|
|
"value_summary": _serialize_for_json(val)
|
|
}
|
|
|
|
status = "ERROR" if exception is not None else "OK"
|
|
|
|
dump_dict: dict[str, Any] = {
|
|
"node_name": node_name,
|
|
"node_type": node_type,
|
|
"socket_evaluated": socket_name,
|
|
"timestamp": now_iso,
|
|
"duration_seconds": duration_s,
|
|
"status": status,
|
|
"exception": str(exception) if exception else None,
|
|
"traceback": tb_str if exception else None,
|
|
"properties": _serialize_for_json(props),
|
|
"inputs": inputs_summary,
|
|
"output_summary": _serialize_for_json(output_val) if exception is None else None
|
|
}
|
|
|
|
dump_text = format_debug_report(dump_dict)
|
|
|
|
_LAST_DUMP_DICT = dump_dict
|
|
_LAST_DUMP_TEXT = dump_text
|
|
|
|
# Write files to dump directory
|
|
dump_dir = get_debug_dump_dir()
|
|
clean_node_type = node_type.replace(" ", "_").replace(".", "_")
|
|
|
|
# 1. Write last_execution.json and last_execution.txt
|
|
last_json_file = dump_dir / "last_execution.json"
|
|
last_txt_file = dump_dir / "last_execution.txt"
|
|
last_pkl_file = dump_dir / "last_execution.pkl"
|
|
|
|
# 2. Timestamped files for history
|
|
hist_json_file = dump_dir / f"{now_tag}_{clean_node_type}.json"
|
|
hist_txt_file = dump_dir / f"{now_tag}_{clean_node_type}.txt"
|
|
hist_pkl_file = dump_dir / f"{now_tag}_{clean_node_type}.pkl"
|
|
|
|
try:
|
|
with open(last_json_file, "w", encoding="utf-8") as f:
|
|
json.dump(dump_dict, f, indent=2)
|
|
with open(hist_json_file, "w", encoding="utf-8") as f:
|
|
json.dump(dump_dict, f, indent=2)
|
|
|
|
with open(last_txt_file, "w", encoding="utf-8") as f:
|
|
f.write(dump_text)
|
|
with open(hist_txt_file, "w", encoding="utf-8") as f:
|
|
f.write(dump_text)
|
|
except Exception as e:
|
|
if log_console:
|
|
print(f"[NodeSlicer DEBUG] Warning: failed to write text/json debug dump: {e}")
|
|
|
|
# 3. Pickle exact objects for full reproducibility
|
|
if save_pickle:
|
|
raw_dump = {
|
|
"node_name": node_name,
|
|
"node_type": node_type,
|
|
"socket_evaluated": socket_name,
|
|
"timestamp": now_iso,
|
|
"duration_seconds": duration_s,
|
|
"status": status,
|
|
"exception": str(exception) if exception else None,
|
|
"traceback": tb_str if exception else None,
|
|
"properties": props,
|
|
"inputs": {k: _sanitize_for_pickle(v) for k, v in raw_inputs_dict.items()},
|
|
"output": _sanitize_for_pickle(output_val) if exception is None else None
|
|
}
|
|
try:
|
|
with open(last_pkl_file, "wb") as f:
|
|
pickle.dump(raw_dump, f)
|
|
with open(hist_pkl_file, "wb") as f:
|
|
pickle.dump(raw_dump, f)
|
|
except Exception as e:
|
|
if log_console:
|
|
print(f"[NodeSlicer DEBUG] Warning: failed to write pickle dump: {e}")
|
|
|
|
# Display in Blender Text Editor if available
|
|
try:
|
|
import bpy
|
|
if hasattr(bpy, "data") and hasattr(bpy.data, "texts"):
|
|
tb = bpy.data.texts.get("NodeSlicer_Last_Debug.txt")
|
|
if tb is None:
|
|
tb = bpy.data.texts.new(name="NodeSlicer_Last_Debug.txt")
|
|
tb.from_string(dump_text)
|
|
except Exception:
|
|
pass
|
|
|
|
if log_console:
|
|
if exception:
|
|
print(f"[NodeSlicer DEBUG ERROR] Node '{node_name}' ({node_type}) failed: {exception}")
|
|
print(f"[NodeSlicer DEBUG] Review dump at: {last_txt_file}")
|
|
else:
|
|
print(f"[NodeSlicer DEBUG] Dumped execution of '{node_name}' ({node_type}) -> {last_txt_file}")
|
|
|
|
return dump_dict
|
|
|
|
|
|
F = TypeVar("F", bound=Callable[..., Any])
|
|
|
|
|
|
def debug(
|
|
target: Any = None,
|
|
*,
|
|
save_pickle: bool = True,
|
|
log_console: bool = True
|
|
) -> Any:
|
|
"""
|
|
Decorator for Node classes, evaluate_output methods, or standalone functions to dump inputs and state.
|
|
|
|
Examples:
|
|
@debug
|
|
@register_class
|
|
class SLICENODE_SeamPlacement(SliceNode):
|
|
...
|
|
|
|
@register_class
|
|
class SLICENODE_MyNode(SliceNode):
|
|
@debug
|
|
def evaluate_output(self, socket_name: str, **kwargs):
|
|
...
|
|
|
|
@debug
|
|
def my_calculation(a, b):
|
|
...
|
|
"""
|
|
def decorator(obj: Any) -> Any:
|
|
# Case 1: Decorating a Class (e.g. SliceNode subclass)
|
|
if inspect.isclass(obj):
|
|
orig_eval = getattr(obj, "evaluate_output", None)
|
|
if orig_eval is not None and callable(orig_eval):
|
|
@functools.wraps(orig_eval)
|
|
def wrapped_eval(self, socket_name: str, **kwargs):
|
|
start = time.perf_counter()
|
|
exc_val = None
|
|
tb = None
|
|
result = None
|
|
try:
|
|
result = orig_eval(self, socket_name, **kwargs)
|
|
return result
|
|
except Exception as e:
|
|
exc_val = e
|
|
tb = traceback.format_exc()
|
|
raise
|
|
finally:
|
|
dur = time.perf_counter() - start
|
|
try:
|
|
dump_node_execution(
|
|
self,
|
|
socket_name,
|
|
output_val=result,
|
|
duration_s=dur,
|
|
exception=exc_val,
|
|
tb_str=tb,
|
|
save_pickle=save_pickle,
|
|
log_console=log_console
|
|
)
|
|
except Exception as dump_err:
|
|
print(f"[NodeSlicer DEBUG] Dump error: {dump_err}")
|
|
|
|
obj.evaluate_output = wrapped_eval
|
|
return obj
|
|
|
|
# Case 2: Decorating a method or function
|
|
if callable(obj):
|
|
@functools.wraps(obj)
|
|
def wrapped_func(*args, **kwargs):
|
|
# Check if first arg is a SliceNode instance
|
|
if len(args) > 0 and hasattr(args[0], "inputs") and hasattr(args[0], "resolve"):
|
|
node_inst = args[0]
|
|
socket_name = args[1] if len(args) > 1 and isinstance(args[1], str) else "Output"
|
|
start = time.perf_counter()
|
|
exc_val = None
|
|
tb = None
|
|
result = None
|
|
try:
|
|
result = obj(*args, **kwargs)
|
|
return result
|
|
except Exception as e:
|
|
exc_val = e
|
|
tb = traceback.format_exc()
|
|
raise
|
|
finally:
|
|
dur = time.perf_counter() - start
|
|
try:
|
|
dump_node_execution(
|
|
node_inst,
|
|
socket_name,
|
|
output_val=result,
|
|
duration_s=dur,
|
|
exception=exc_val,
|
|
tb_str=tb,
|
|
save_pickle=save_pickle,
|
|
log_console=log_console
|
|
)
|
|
except Exception as dump_err:
|
|
print(f"[NodeSlicer DEBUG] Dump error: {dump_err}")
|
|
else:
|
|
# Regular standalone function
|
|
func_name = getattr(obj, "__name__", "function")
|
|
start = time.perf_counter()
|
|
exc_val = None
|
|
tb = None
|
|
result = None
|
|
try:
|
|
result = obj(*args, **kwargs)
|
|
return result
|
|
except Exception as e:
|
|
exc_val = e
|
|
tb = traceback.format_exc()
|
|
raise
|
|
finally:
|
|
dur = time.perf_counter() - start
|
|
now_iso = datetime.datetime.now().isoformat()
|
|
status = "ERROR" if exc_val else "OK"
|
|
args_summary = [_serialize_for_json(a) for a in args]
|
|
kwargs_summary = {k: _serialize_for_json(v) for k, v in kwargs.items()}
|
|
dump_dict = {
|
|
"node_name": func_name,
|
|
"node_type": "Function",
|
|
"socket_evaluated": "return",
|
|
"timestamp": now_iso,
|
|
"duration_seconds": dur,
|
|
"status": status,
|
|
"exception": str(exc_val) if exc_val else None,
|
|
"traceback": tb if exc_val else None,
|
|
"properties": {},
|
|
"inputs": {
|
|
f"arg_{i}": {"value_summary": a, "is_linked": False} for i, a in enumerate(args_summary)
|
|
} | {
|
|
f"kwarg_{k}": {"value_summary": v, "is_linked": False} for k, v in kwargs_summary.items()
|
|
},
|
|
"output_summary": _serialize_for_json(result) if exc_val is None else None
|
|
}
|
|
report_text = format_debug_report(dump_dict)
|
|
dump_dir = get_debug_dump_dir()
|
|
try:
|
|
with open(dump_dir / "last_execution.json", "w", encoding="utf-8") as f:
|
|
json.dump(dump_dict, f, indent=2)
|
|
with open(dump_dir / "last_execution.txt", "w", encoding="utf-8") as f:
|
|
f.write(report_text)
|
|
except Exception:
|
|
pass
|
|
return wrapped_func
|
|
|
|
return obj
|
|
|
|
if target is None:
|
|
return decorator
|
|
return decorator(target)
|
|
|
|
|
|
def get_last_debug_dump() -> dict[str, Any] | None:
|
|
"""Returns the parsed dictionary of the last execution dump."""
|
|
global _LAST_DUMP_DICT
|
|
if _LAST_DUMP_DICT is not None:
|
|
return _LAST_DUMP_DICT
|
|
|
|
json_file = get_debug_dump_dir() / "last_execution.json"
|
|
if json_file.exists():
|
|
try:
|
|
with open(json_file, "r", encoding="utf-8") as f:
|
|
_LAST_DUMP_DICT = json.load(f)
|
|
return _LAST_DUMP_DICT
|
|
except Exception:
|
|
pass
|
|
return None
|
|
|
|
|
|
def get_last_debug_text() -> str:
|
|
"""Returns the formatted text summary of the last execution dump."""
|
|
global _LAST_DUMP_TEXT
|
|
if _LAST_DUMP_TEXT is not None:
|
|
return _LAST_DUMP_TEXT
|
|
|
|
txt_file = get_debug_dump_dir() / "last_execution.txt"
|
|
if txt_file.exists():
|
|
try:
|
|
with open(txt_file, "r", encoding="utf-8") as f:
|
|
_LAST_DUMP_TEXT = f.read()
|
|
return _LAST_DUMP_TEXT
|
|
except Exception:
|
|
pass
|
|
|
|
dump = get_last_debug_dump()
|
|
if dump:
|
|
_LAST_DUMP_TEXT = format_debug_report(dump)
|
|
return _LAST_DUMP_TEXT
|
|
|
|
return "(No debug dumps found in debug_dumps/)"
|
|
|
|
|
|
def load_last_inputs() -> dict[str, Any] | None:
|
|
"""Loads the unpickled raw Python objects from the last execution dump."""
|
|
pkl_file = get_debug_dump_dir() / "last_execution.pkl"
|
|
if pkl_file.exists():
|
|
try:
|
|
with open(pkl_file, "rb") as f:
|
|
return pickle.load(f)
|
|
except Exception as e:
|
|
print(f"[NodeSlicer DEBUG] Error loading last execution pickle: {e}")
|
|
return None
|
|
|
|
|
|
def clear_debug_dumps() -> None:
|
|
"""Removes all dump files from the debug_dumps directory."""
|
|
global _LAST_DUMP_DICT, _LAST_DUMP_TEXT
|
|
_LAST_DUMP_DICT = None
|
|
_LAST_DUMP_TEXT = None
|
|
dump_dir = get_debug_dump_dir()
|
|
if dump_dir.exists():
|
|
for item in dump_dir.glob("*"):
|
|
if item.is_file():
|
|
try:
|
|
item.unlink()
|
|
except Exception:
|
|
pass
|