Coverage for src / tracekit / session / history.py: 83%
105 statements
« prev ^ index » next coverage.py v7.13.1, created at 2026-01-11 23:04 +0000
1"""Operation history tracking.
3This module provides operation history tracking for analysis sessions.
6Example:
7 >>> history = OperationHistory()
8 >>> history.record('load', {'file': 'capture.wfm'})
9 >>> history.record('measure_rise_time', {'result': 1.5e-9})
10 >>> print(history.to_script())
11"""
13from __future__ import annotations
15from dataclasses import dataclass, field
16from datetime import datetime
17from typing import Any
20@dataclass
21class HistoryEntry:
22 """Single history entry recording an operation.
24 Attributes:
25 operation: Operation name (function/method called)
26 parameters: Input parameters
27 result: Operation result (summary)
28 timestamp: When operation was performed
29 duration_ms: Operation duration in milliseconds
30 success: Whether operation succeeded
31 error_message: Error message if failed
32 metadata: Additional metadata
33 """
35 operation: str
36 parameters: dict[str, Any] = field(default_factory=dict)
37 result: Any = None
38 timestamp: datetime = field(default_factory=datetime.now)
39 duration_ms: float = 0.0
40 success: bool = True
41 error_message: str | None = None
42 metadata: dict[str, Any] = field(default_factory=dict)
44 def to_dict(self) -> dict[str, Any]:
45 """Convert to dictionary for serialization."""
46 return {
47 "operation": self.operation,
48 "parameters": self.parameters,
49 "result": self._serialize_result(self.result),
50 "timestamp": self.timestamp.isoformat(),
51 "duration_ms": self.duration_ms,
52 "success": self.success,
53 "error_message": self.error_message,
54 "metadata": self.metadata,
55 }
57 @staticmethod
58 def _serialize_result(result: Any) -> Any:
59 """Serialize result for JSON storage."""
60 if result is None:
61 return None
62 if isinstance(result, str | int | float | bool): 62 ↛ 64line 62 didn't jump to line 64 because the condition on line 62 was always true
63 return result
64 if isinstance(result, dict):
65 return {k: HistoryEntry._serialize_result(v) for k, v in result.items()}
66 if isinstance(result, list | tuple):
67 return [HistoryEntry._serialize_result(v) for v in result]
68 # For complex objects, store string representation
69 return str(result)
71 @classmethod
72 def from_dict(cls, data: dict[str, Any]) -> HistoryEntry:
73 """Create from dictionary."""
74 data = data.copy()
75 if "timestamp" in data and isinstance(data["timestamp"], str): 75 ↛ 77line 75 didn't jump to line 77 because the condition on line 75 was always true
76 data["timestamp"] = datetime.fromisoformat(data["timestamp"])
77 return cls(**data)
79 def to_code(self) -> str:
80 """Generate Python code to replay this operation.
82 Returns:
83 Python code string.
84 """
85 # Format parameters
86 params = []
87 for k, v in self.parameters.items():
88 if isinstance(v, str):
89 params.append(f'{k}="{v}"')
90 else:
91 params.append(f"{k}={v!r}")
93 param_str = ", ".join(params)
94 return f"tk.{self.operation}({param_str})"
97@dataclass
98class OperationHistory:
99 """History of analysis operations.
101 Supports recording, replaying, and exporting operation history.
103 Attributes:
104 entries: List of history entries
105 max_entries: Maximum entries to keep (0 = unlimited)
106 auto_record: Whether to automatically record operations
107 """
109 entries: list[HistoryEntry] = field(default_factory=list)
110 max_entries: int = 0
111 auto_record: bool = True
112 _current_session_start: datetime = field(default_factory=datetime.now)
114 def record(
115 self,
116 operation: str,
117 parameters: dict[str, Any] | None = None,
118 result: Any = None,
119 duration_ms: float = 0.0,
120 success: bool = True,
121 error_message: str | None = None,
122 **metadata: Any,
123 ) -> HistoryEntry:
124 """Record an operation.
126 Args:
127 operation: Operation name.
128 parameters: Input parameters.
129 result: Operation result.
130 duration_ms: Duration in milliseconds.
131 success: Whether operation succeeded.
132 error_message: Error message if failed.
133 **metadata: Additional metadata.
135 Returns:
136 Created history entry.
137 """
138 entry = HistoryEntry(
139 operation=operation,
140 parameters=parameters or {},
141 result=result,
142 duration_ms=duration_ms,
143 success=success,
144 error_message=error_message,
145 metadata=metadata,
146 )
148 self.entries.append(entry)
150 # Trim if exceeded max entries
151 if self.max_entries > 0 and len(self.entries) > self.max_entries:
152 self.entries = self.entries[-self.max_entries :]
154 return entry
156 def undo(self) -> HistoryEntry | None:
157 """Remove and return the last entry.
159 Returns:
160 Removed entry, or None if empty.
161 """
162 if self.entries: 162 ↛ 164line 162 didn't jump to line 164 because the condition on line 162 was always true
163 return self.entries.pop()
164 return None
166 def clear(self) -> int:
167 """Clear all history.
169 Returns:
170 Number of entries cleared.
171 """
172 count = len(self.entries)
173 self.entries.clear()
174 return count
176 def find(
177 self,
178 operation: str | None = None,
179 success_only: bool = False,
180 since: datetime | None = None,
181 ) -> list[HistoryEntry]:
182 """Find entries matching criteria.
184 Args:
185 operation: Filter by operation name.
186 success_only: Only return successful operations.
187 since: Only return entries after this time.
189 Returns:
190 Matching entries.
191 """
192 results = []
193 for entry in self.entries:
194 if operation and entry.operation != operation:
195 continue
196 if success_only and not entry.success:
197 continue
198 if since and entry.timestamp < since: 198 ↛ 199line 198 didn't jump to line 199 because the condition on line 198 was never true
199 continue
200 results.append(entry)
201 return results
203 def to_script(
204 self,
205 include_imports: bool = True,
206 include_comments: bool = True,
207 ) -> str:
208 """Export history as Python script.
210 Args:
211 include_imports: Include import statements.
212 include_comments: Include timestamp comments.
214 Returns:
215 Python script string.
217 Example:
218 >>> script = history.to_script()
219 >>> print(script)
220 # Generated by TraceKit
221 import tracekit as tk
222 tk.load("capture.wfm")
223 result = tk.measure_rise_time()
224 """
225 lines = []
227 if include_imports: 227 ↛ 241line 227 didn't jump to line 241 because the condition on line 227 was always true
228 lines.extend(
229 [
230 "#!/usr/bin/env python3",
231 '"""TraceKit analysis script.',
232 "",
233 f"Generated: {datetime.now().isoformat()}",
234 '"""',
235 "",
236 "import tracekit as tk",
237 "",
238 ]
239 )
241 for entry in self.entries:
242 if not entry.success: 242 ↛ 243line 242 didn't jump to line 243 because the condition on line 242 was never true
243 continue
245 if include_comments: 245 ↛ 248line 245 didn't jump to line 248 because the condition on line 245 was always true
246 lines.append(f"# {entry.timestamp.strftime('%H:%M:%S')} - {entry.operation}")
248 lines.append(entry.to_code())
249 lines.append("")
251 return "\n".join(lines)
253 def summary(self) -> dict[str, Any]:
254 """Get history summary statistics.
256 Returns:
257 Dictionary with summary statistics.
258 """
259 if not self.entries: 259 ↛ 260line 259 didn't jump to line 260 because the condition on line 259 was never true
260 return {
261 "total_operations": 0,
262 "successful": 0,
263 "failed": 0,
264 "total_duration_ms": 0,
265 "unique_operations": 0,
266 }
268 successful = sum(1 for e in self.entries if e.success)
269 failed = len(self.entries) - successful
270 total_duration = sum(e.duration_ms for e in self.entries)
271 unique_ops = len({e.operation for e in self.entries})
273 # Operation frequency
274 op_counts: dict[str, int] = {}
275 for entry in self.entries:
276 op_counts[entry.operation] = op_counts.get(entry.operation, 0) + 1
278 return {
279 "total_operations": len(self.entries),
280 "successful": successful,
281 "failed": failed,
282 "total_duration_ms": total_duration,
283 "unique_operations": unique_ops,
284 "operation_counts": op_counts,
285 "session_start": self._current_session_start.isoformat(),
286 "last_operation": self.entries[-1].timestamp.isoformat() if self.entries else None,
287 }
289 def to_dict(self) -> dict[str, Any]:
290 """Convert to dictionary for serialization."""
291 return {
292 "entries": [e.to_dict() for e in self.entries],
293 "max_entries": self.max_entries,
294 "session_start": self._current_session_start.isoformat(),
295 }
297 @classmethod
298 def from_dict(cls, data: dict[str, Any]) -> OperationHistory:
299 """Create from dictionary."""
300 entries = [HistoryEntry.from_dict(e) for e in data.get("entries", [])]
301 history = cls(
302 entries=entries,
303 max_entries=data.get("max_entries", 0),
304 )
305 if "session_start" in data: 305 ↛ 307line 305 didn't jump to line 307 because the condition on line 305 was always true
306 history._current_session_start = datetime.fromisoformat(data["session_start"])
307 return history
310__all__ = [
311 "HistoryEntry",
312 "OperationHistory",
313]