Coverage for src / tracekit / guidance / wizard.py: 91%
181 statements
« prev ^ index » next coverage.py v7.13.1, created at 2026-01-11 23:04 +0000
1"""Interactive analysis wizard for guided workflows.
3This module provides step-by-step guided workflows that walk non-experts
4through signal analysis, asking simple questions and adapting based on
5responses.
8Example:
9 >>> from tracekit.guidance import AnalysisWizard
10 >>> wizard = AnalysisWizard(trace)
11 >>> result = wizard.run()
13References:
14 TraceKit Auto-Discovery Specification
15 Phase 34 Task-247
16"""
18from __future__ import annotations
20from dataclasses import dataclass, field
21from datetime import datetime
22from typing import TYPE_CHECKING, Any
24if TYPE_CHECKING:
25 from collections.abc import Callable
27 from tracekit.core.types import WaveformTrace
30@dataclass
31class WizardStep:
32 """Single step in the analysis wizard.
34 Attributes:
35 number: Step number (1-based).
36 id: Step identifier.
37 question: Question text in plain English.
38 options: List of answer options.
39 default: Default/recommended answer.
40 skip_if_confident: Skip if auto-detection confidence >= threshold.
41 user_response: User's answer.
42 confidence_before: Confidence before step.
43 confidence_after: Confidence after step.
44 preview: Preview of result.
45 """
47 number: int
48 id: str
49 question: str
50 options: list[str] = field(default_factory=list)
51 default: str | None = None
52 skip_if_confident: bool = False
53 user_response: str | None = None
54 confidence_before: float = 0.0
55 confidence_after: float = 0.0
56 preview: Any = None
59@dataclass
60class WizardResult:
61 """Result from wizard analysis.
63 Attributes:
64 summary: Summary of analysis.
65 signal_type: Detected signal type.
66 parameters: Signal parameters.
67 quality: Quality assessment.
68 decode: Decoded data (if applicable).
69 recommendations: Next step recommendations.
70 confidence: Overall confidence.
71 """
73 summary: str
74 signal_type: str | None = None
75 parameters: dict | None = None # type: ignore[type-arg]
76 quality: Any = None
77 decode: Any = None
78 recommendations: list = field(default_factory=list) # type: ignore[type-arg]
79 confidence: float = 0.0
82class AnalysisWizard:
83 """Interactive analysis wizard for guided workflows.
85 Provides step-by-step guided analysis with smart defaults,
86 auto-skip based on confidence, progress tracking, and live previews.
88 Attributes:
89 trace: Waveform being analyzed.
90 max_questions: Maximum questions to ask (default 5).
91 auto_detect_threshold: Confidence threshold for auto-skip (default 0.8).
92 enable_preview: Enable live result preview.
93 allow_backtrack: Allow back/forward navigation.
94 interactive: Enable interactive mode.
95 step_history: History of completed steps.
96 steps_completed: Number of steps completed.
97 questions_asked: Number of questions asked.
98 questions_skipped: Number of questions skipped.
99 session_duration_seconds: Total session duration.
100 _start_time: Session start time.
101 _current_state: Current analysis state.
102 """
104 def __init__(
105 self,
106 trace: WaveformTrace,
107 *,
108 max_questions: int = 5,
109 auto_detect_threshold: float = 0.8,
110 enable_preview: bool = True,
111 allow_backtrack: bool = True,
112 interactive: bool = True,
113 ) -> None:
114 """Initialize analysis wizard.
116 Args:
117 trace: Waveform to analyze.
118 max_questions: Maximum questions (default 5, range 3-7).
119 auto_detect_threshold: Skip question if confidence >= this.
120 enable_preview: Enable live result preview after each step.
121 allow_backtrack: Allow back/forward navigation.
122 interactive: Enable interactive mode (vs programmatic).
124 References:
125 DISC-006: Interactive Analysis Wizard
126 """
127 self.trace = trace
128 self.max_questions = max(3, min(7, max_questions))
129 self.auto_detect_threshold = auto_detect_threshold
130 self.enable_preview = enable_preview
131 self.allow_backtrack = allow_backtrack
132 self.interactive = interactive
134 self.step_history: list[WizardStep] = []
135 self.steps_completed = 0
136 self.questions_asked = 0
137 self.questions_skipped = 0
138 self.session_duration_seconds = 0.0
140 self._start_time = datetime.now()
141 self._current_state: dict[str, Any] = {}
142 self._predefined_answers: dict[str, str] = {}
144 def add_custom_step(
145 self,
146 step_id: str,
147 *,
148 question: str,
149 options: list[str],
150 default: str | None = None,
151 skip_if_confident: bool = True,
152 ) -> None:
153 """Add a custom step to the wizard.
155 Args:
156 step_id: Unique step identifier.
157 question: Question text in plain English.
158 options: List of answer options.
159 default: Default/recommended answer.
160 skip_if_confident: Skip if auto-detection confident.
161 """
162 # Store for later use during run()
163 if not hasattr(self, "_custom_steps"):
164 self._custom_steps = []
166 self._custom_steps.append(
167 {
168 "id": step_id,
169 "question": question,
170 "options": options,
171 "default": default,
172 "skip_if_confident": skip_if_confident,
173 }
174 )
176 def set_answers(self, answers: dict[str, str]) -> None:
177 """Set predefined answers for programmatic mode.
179 Args:
180 answers: Dictionary mapping step IDs to answers.
181 """
182 self._predefined_answers = answers
184 def run(
185 self,
186 *,
187 preview_callback: Callable[[Any], None] | None = None,
188 ) -> WizardResult:
189 """Run the analysis wizard.
191 Guides user through analysis steps, auto-skipping where confident,
192 showing progress, and providing live previews.
194 Args:
195 preview_callback: Optional callback for step previews.
197 Returns:
198 WizardResult with analysis summary and findings.
200 Example:
201 >>> wizard = AnalysisWizard(trace)
202 >>> result = wizard.run()
203 >>> print(result.summary)
205 References:
206 DISC-006: Interactive Analysis Wizard
207 """
208 from tracekit.discovery import (
209 assess_data_quality,
210 characterize_signal,
211 decode_protocol,
212 find_anomalies,
213 )
214 from tracekit.guidance import suggest_next_steps
216 # Step 1: Auto-characterization
217 step1 = WizardStep(
218 number=1,
219 id="characterization",
220 question="What type of signal are you analyzing?",
221 options=[
222 "Serial data (UART, SPI, I2C)",
223 "PWM / Motor control",
224 "Analog sensor output",
225 "Not sure - auto-detect",
226 ],
227 default="Not sure - auto-detect",
228 skip_if_confident=False,
229 )
231 # Always do auto-characterization first
232 char_result = characterize_signal(self.trace)
233 step1.confidence_before = 0.0
234 step1.confidence_after = char_result.confidence
235 self._current_state["characterization"] = char_result
237 if self.interactive and char_result.confidence < self.auto_detect_threshold:
238 # Ask user to confirm
239 step1.user_response = self._predefined_answers.get("signal_type", step1.default)
240 self.questions_asked += 1
241 else:
242 # Auto-detected with high confidence
243 signal_type = getattr(char_result, "signal_type", "Unknown")
244 step1.user_response = f"Auto-detected: {signal_type}"
245 self.questions_skipped += 1
247 self.step_history.append(step1)
248 self.steps_completed += 1
250 # Preview callback
251 if preview_callback and self.enable_preview:
252 preview_callback(char_result)
254 # Step 2: Quality assessment
255 step2 = WizardStep(
256 number=2,
257 id="quality",
258 question="Check data quality?",
259 options=["Yes", "No"],
260 default="Yes",
261 skip_if_confident=True,
262 )
264 quality = assess_data_quality(self.trace)
265 step2.confidence_before = char_result.confidence
266 step2.confidence_after = quality.confidence
267 self._current_state["quality"] = quality
269 if self.interactive and self.questions_asked < self.max_questions:
270 step2.user_response = self._predefined_answers.get("check_quality", "Yes")
271 if step2.user_response == "Yes": 271 ↛ 277line 271 didn't jump to line 277 because the condition on line 271 was always true
272 self.questions_asked += 1
273 else:
274 step2.user_response = "Skipped (auto-assessed)"
275 self.questions_skipped += 1
277 self.step_history.append(step2)
278 self.steps_completed += 1
280 if preview_callback and self.enable_preview:
281 preview_callback(quality)
283 # Step 3: Protocol decode (if applicable)
284 decode_result = None
285 if hasattr(char_result, "signal_type") and char_result.confidence >= 0.7:
286 signal_type = char_result.signal_type.lower()
288 if any(proto in signal_type for proto in ["uart", "spi", "i2c", "can"]): 288 ↛ 321line 288 didn't jump to line 321 because the condition on line 288 was always true
289 step3 = WizardStep(
290 number=3,
291 id="decode",
292 question=f"Auto-detected {char_result.signal_type}. Decode data?",
293 options=["Yes", "No"],
294 default="Yes",
295 skip_if_confident=True,
296 )
298 if self.interactive and self.questions_asked < self.max_questions: 298 ↛ 299line 298 didn't jump to line 299 because the condition on line 298 was never true
299 step3.user_response = self._predefined_answers.get("decode_data", "Yes")
300 if step3.user_response == "Yes":
301 decode_result = decode_protocol(self.trace)
302 self._current_state["decode"] = decode_result
303 self.questions_asked += 1
304 else:
305 # Auto-decode
306 decode_result = decode_protocol(self.trace)
307 self._current_state["decode"] = decode_result
308 step3.user_response = "Auto-decoded"
309 self.questions_skipped += 1
311 step3.confidence_before = char_result.confidence
312 step3.confidence_after = decode_result.overall_confidence if decode_result else 0.0
314 self.step_history.append(step3)
315 self.steps_completed += 1
317 if preview_callback and self.enable_preview and decode_result:
318 preview_callback(decode_result)
320 # Step 4: Anomaly detection (if quality issues)
321 anomalies = None
322 if quality.status in ["WARNING", "FAIL"]:
323 step4 = WizardStep(
324 number=self.steps_completed + 1,
325 id="anomalies",
326 question="Quality concerns detected. Check for anomalies?",
327 options=["Yes", "No"],
328 default="Yes",
329 )
331 if self.interactive and self.questions_asked < self.max_questions: 331 ↛ 332line 331 didn't jump to line 332 because the condition on line 331 was never true
332 step4.user_response = self._predefined_answers.get("check_anomalies", "Yes")
333 if step4.user_response == "Yes":
334 anomalies = find_anomalies(self.trace)
335 self._current_state["anomalies"] = anomalies
336 self.questions_asked += 1
337 else:
338 # Auto-check
339 anomalies = find_anomalies(self.trace)
340 self._current_state["anomalies"] = anomalies
341 step4.user_response = "Auto-checked"
342 self.questions_skipped += 1
344 self.step_history.append(step4)
345 self.steps_completed += 1
347 if preview_callback and self.enable_preview and anomalies: 347 ↛ 348line 347 didn't jump to line 348 because the condition on line 347 was never true
348 preview_callback(anomalies)
350 # Get recommendations for next steps
351 recommendations = suggest_next_steps(
352 self.trace,
353 current_state=self._current_state,
354 )
356 # Build summary
357 summary_parts = []
359 if hasattr(char_result, "signal_type") and char_result.signal_type:
360 summary_parts.append(f"Signal type: {char_result.signal_type}")
361 if hasattr(char_result, "parameters"): 361 ↛ 364line 361 didn't jump to line 364 because the condition on line 361 was always true
362 summary_parts.append(f"Parameters: {_format_params(char_result.parameters)}")
364 if quality.status == "PASS":
365 summary_parts.append("Quality: Good")
366 elif quality.status == "WARNING":
367 summary_parts.append("Quality: Fair (some concerns)")
368 else:
369 summary_parts.append("Quality: Poor (issues detected)")
371 if decode_result:
372 byte_count = len(decode_result.data) if hasattr(decode_result, "data") else 0
373 summary_parts.append(f"Decoded: {byte_count} bytes")
375 if anomalies and len(anomalies) > 0:
376 critical = sum(1 for a in anomalies if a.severity == "CRITICAL")
377 if critical > 0: 377 ↛ 380line 377 didn't jump to line 380 because the condition on line 377 was always true
378 summary_parts.append(f"Anomalies: {critical} critical issues")
380 summary = "\n".join(summary_parts)
382 # Session duration
383 self.session_duration_seconds = (datetime.now() - self._start_time).total_seconds()
385 return WizardResult(
386 summary=summary,
387 signal_type=char_result.signal_type if hasattr(char_result, "signal_type") else None,
388 parameters=char_result.parameters if hasattr(char_result, "parameters") else None,
389 quality=quality,
390 decode=decode_result,
391 recommendations=recommendations,
392 confidence=char_result.confidence,
393 )
395 @classmethod
396 def from_session(cls, session_file: str) -> AnalysisWizard:
397 """Load wizard from saved session file.
399 Args:
400 session_file: Path to session JSON file.
402 Returns:
403 AnalysisWizard instance configured from session.
405 Raises:
406 FileNotFoundError: If session file doesn't exist.
407 ValueError: If session file is invalid.
408 """
409 import json
410 from pathlib import Path
412 path = Path(session_file)
413 if not path.exists():
414 msg = f"Session file not found: {session_file}"
415 raise FileNotFoundError(msg)
417 with path.open() as f:
418 session_data = json.load(f)
420 # Extract trace path and load
421 from tracekit import load
423 trace_path = session_data.get("trace_path")
424 if not trace_path:
425 msg = "Session file missing trace_path"
426 raise ValueError(msg)
428 trace = load(trace_path)
430 # Create wizard with saved settings
431 wizard = cls(
432 trace, # type: ignore[arg-type]
433 max_questions=session_data.get("max_questions", 5),
434 auto_detect_threshold=session_data.get("auto_detect_threshold", 0.8),
435 enable_preview=session_data.get("enable_preview", True),
436 allow_backtrack=session_data.get("allow_backtrack", True),
437 interactive=session_data.get("interactive", True),
438 )
440 # Set predefined answers
441 if "answers" in session_data:
442 wizard.set_answers(session_data["answers"])
444 return wizard
446 def save_session(self, output_path: str) -> None:
447 """Save wizard session to JSON file.
449 Args:
450 output_path: Path for output JSON file.
451 """
452 import json
453 from pathlib import Path
455 session_data = {
456 "trace_path": str(getattr(self.trace, "path", "")),
457 "max_questions": self.max_questions,
458 "auto_detect_threshold": self.auto_detect_threshold,
459 "enable_preview": self.enable_preview,
460 "allow_backtrack": self.allow_backtrack,
461 "interactive": self.interactive,
462 "steps_completed": self.steps_completed,
463 "questions_asked": self.questions_asked,
464 "questions_skipped": self.questions_skipped,
465 "session_duration_seconds": self.session_duration_seconds,
466 "answers": self._predefined_answers,
467 "step_history": [
468 {
469 "number": step.number,
470 "id": step.id,
471 "question": step.question,
472 "user_response": step.user_response,
473 "confidence_before": step.confidence_before,
474 "confidence_after": step.confidence_after,
475 }
476 for step in self.step_history
477 ],
478 }
480 path = Path(output_path)
481 path.parent.mkdir(parents=True, exist_ok=True)
483 with path.open("w") as f:
484 json.dump(session_data, f, indent=2)
487def _format_params(params: dict) -> str: # type: ignore[type-arg]
488 """Format parameters dictionary for display.
490 Args:
491 params: Parameters dictionary.
493 Returns:
494 Formatted string.
495 """
496 if not params:
497 return ""
499 parts = []
500 for key, value in params.items():
501 if isinstance(value, int | float):
502 if key.endswith("_hz") or key.endswith("_freq"):
503 parts.append(f"{key}={value / 1e3:.1f}kHz")
504 elif key.endswith("_baud") or key.endswith("baud_rate"):
505 parts.append(f"{key}={value:.0f}")
506 else:
507 parts.append(f"{key}={value}")
508 else:
509 parts.append(f"{key}={value}")
511 return ", ".join(parts[:3]) # Limit to 3 params
514__all__ = [
515 "AnalysisWizard",
516 "WizardResult",
517 "WizardStep",
518]