MCPcopy Create free account
hub / github.com/Tong89/smartNode / take

Method take

backend/snapshot.py:64–107  ·  view source on GitHub ↗

对仿真引擎当前状态拍快照。 Parameters ---------- engine: ``SimulationEngine`` 实例。 label: 可选的快照标签,便于人类识别(例如场景名称或事件描述)。 Returns ------- dict 快照字典,包含 current_time、请求列表、资源占用与统计数据。

(engine: Any, label: str = "")

Source from the content-addressed store, hash-verified

62
63 @staticmethod
64 def take(engine: Any, label: str = "") -> Dict[str, Any]:
65 """对仿真引擎当前状态拍快照。
66
67 Parameters
68 ----------
69 engine:
70 ``SimulationEngine`` 实例。
71 label:
72 可选的快照标签,便于人类识别(例如场景名称或事件描述)。
73
74 Returns
75 -------
76 dict
77 快照字典,包含 current_time、请求列表、资源占用与统计数据。
78 """
79 from backend.core import TransmissionRequest # 避免循环导入
80
81 with engine.lock:
82 # 序列化进行中的请求
83 active_requests = [req.to_dict() for req in engine.transmission_requests]
84 # 序列化历史请求
85 history_requests = [req.to_dict() for req in engine.request_history]
86
87 # 深拷贝资源占用状态(嵌套容器)
88 resource_usage = copy.deepcopy(dict(engine.resource_usage))
89 resource_time_pool = copy.deepcopy(dict(engine._resources.time_pool))
90
91 # 深拷贝统计数据
92 stats = copy.deepcopy(engine.stats)
93
94 snapshot = {
95 "version": SNAPSHOT_VERSION,
96 "saved_at": datetime.now(tz=timezone.utc).isoformat(),
97 "label": label or "",
98 "current_time": engine.current_time,
99 "id_counter": TransmissionRequest._id_counter,
100 "transmission_requests": active_requests,
101 "request_history": history_requests,
102 "resource_usage": resource_usage,
103 "resource_time_pool": resource_time_pool,
104 "stats": stats,
105 }
106
107 return snapshot
108
109 # ── 从快照恢复 ──────────────────────────────────────────────────────────────
110

Calls 1

to_dictMethod · 0.45