Spaces:
Running
Running
| from __future__ import annotations | |
| from dataclasses import dataclass, field | |
| from typing import Any | |
| from velai.dataflow.graph import DataGraph | |
| from velai.nodes.base_node import BaseNode | |
| class GraphUndoStack: | |
| """Stores snapshots of deleted graph fragments for Ctrl+Z restore.""" | |
| max_entries: int = 50 | |
| _stack: list[dict[str, Any]] = field(default_factory=list) | |
| def record_node_deletion(self, graph: DataGraph, node_ids: set[str]) -> None: | |
| if not node_ids: | |
| return | |
| nodes_data: list[dict[str, Any]] = [] | |
| for node_id in node_ids: | |
| node = graph.nodes.get(node_id) | |
| if node is None: | |
| continue | |
| values: dict[str, Any] = {} | |
| if isinstance(node, BaseNode): | |
| try: | |
| values = node.get_state() | |
| except Exception: | |
| values = {} | |
| nodes_data.append( | |
| { | |
| "id": node.node_id, | |
| "kind": node.node_type.kind.value, | |
| "x": node.x, | |
| "y": node.y, | |
| "values": values, | |
| "width": node.width, | |
| "height": node.height, | |
| } | |
| ) | |
| if not nodes_data: | |
| return | |
| edges_data: list[dict[str, Any]] = [] | |
| for conn in graph.connections: | |
| src_id = conn.start_node.node_id | |
| tgt_id = conn.end_node.node_id | |
| if src_id not in node_ids and tgt_id not in node_ids: | |
| continue | |
| edges_data.append( | |
| { | |
| "source": src_id, | |
| "sourceHandle": conn.start_port.name, | |
| "target": tgt_id, | |
| "targetHandle": conn.end_port.name, | |
| } | |
| ) | |
| self._stack.append({"nodes": nodes_data, "edges": edges_data}) | |
| if len(self._stack) > self.max_entries: | |
| self._stack.pop(0) | |
| def pop_deletion(self) -> dict[str, Any] | None: | |
| if not self._stack: | |
| return None | |
| return self._stack.pop() | |
| def clear(self) -> None: | |
| self._stack.clear() | |