Coverage for src / tracekit / onboarding / wizard.py: 87%
224 statements
« prev ^ index » next coverage.py v7.13.1, created at 2026-01-11 23:04 +0000
1"""Interactive analysis wizard for guided signal analysis.
3This module provides step-by-step guided analysis wizards that help
4non-expert users analyze their signals with intelligent recommendations.
6 - Step-by-step guidance
7 - Intelligent defaults
8 - Context-aware recommendations
9 - Result interpretation
11Example:
12 >>> from tracekit.onboarding import run_wizard
13 >>> run_wizard(trace)
14 Analysis Wizard
15 Step 1: What type of signal is this?
16 ...
17"""
19from __future__ import annotations
21from dataclasses import dataclass, field
22from enum import Enum
23from typing import TYPE_CHECKING, Any
25if TYPE_CHECKING:
26 from collections.abc import Callable
29class WizardAction(Enum):
30 """Actions the wizard can perform."""
32 MEASURE = "measure"
33 CHARACTERIZE = "characterize"
34 DECODE = "decode"
35 FILTER = "filter"
36 SPECTRAL = "spectral"
37 COMPARE = "compare"
40@dataclass
41class WizardStep:
42 """A step in the analysis wizard.
44 Attributes:
45 title: Step title
46 question: Question to ask user
47 options: Available options
48 action: Action to perform based on choice
49 help_text: Additional help for this step
50 """
52 title: str
53 question: str
54 options: list[str]
55 action: Callable[[int], None] | None = None
56 help_text: str = ""
57 skip_condition: Callable[[Any], bool] | None = None
60@dataclass
61class WizardResult:
62 """Result from wizard analysis.
64 Attributes:
65 steps_completed: Number of steps completed
66 measurements: Collected measurements
67 recommendations: Analysis recommendations
68 summary: Human-readable summary
69 """
71 steps_completed: int = 0
72 measurements: dict[str, Any] = field(default_factory=dict)
73 recommendations: list[str] = field(default_factory=list)
74 summary: str = ""
77class AnalysisWizard:
78 """Interactive analysis wizard.
80 Guides users through signal analysis with intelligent
81 recommendations and plain English explanations.
82 """
84 def __init__(self, trace: Any) -> None:
85 """Initialize wizard with a trace.
87 Args:
88 trace: WaveformTrace or DigitalTrace to analyze
89 """
90 self.trace = trace
91 self.result = WizardResult()
92 self.steps: list[WizardStep] = self._build_steps()
93 self.current_step = 0
95 def _build_steps(self) -> list[WizardStep]:
96 """Build the wizard steps based on trace characteristics."""
97 steps = [
98 WizardStep(
99 title="Signal Type Detection",
100 question="What type of analysis do you want to perform?",
101 options=[
102 "Auto-detect (let TraceKit figure it out)",
103 "Digital signal analysis",
104 "Analog/waveform analysis",
105 "Protocol decoding",
106 "Power analysis",
107 ],
108 action=self._handle_signal_type,
109 help_text="Not sure? Choose 'Auto-detect' and we'll analyze your signal.",
110 ),
111 WizardStep(
112 title="Basic Measurements",
113 question="Would you like to run basic measurements?",
114 options=[
115 "Yes, run all standard measurements",
116 "Yes, but only timing measurements",
117 "Yes, but only amplitude measurements",
118 "No, skip this step",
119 ],
120 action=self._handle_measurements,
121 help_text="Basic measurements give you an overview of your signal.",
122 ),
123 WizardStep(
124 title="Spectral Analysis",
125 question="Would you like to analyze the frequency content?",
126 options=[
127 "Yes, compute FFT spectrum",
128 "Yes, compute power spectral density",
129 "Yes, both FFT and PSD",
130 "No, skip spectral analysis",
131 ],
132 action=self._handle_spectral,
133 help_text="Spectral analysis shows what frequencies are in your signal.",
134 ),
135 WizardStep(
136 title="Signal Quality",
137 question="Would you like to assess signal quality?",
138 options=[
139 "Yes, measure THD and SNR",
140 "Yes, check for anomalies",
141 "Yes, full quality assessment",
142 "No, skip quality check",
143 ],
144 action=self._handle_quality,
145 help_text="Quality metrics help identify issues with your signal.",
146 ),
147 ]
148 return steps
150 def run(self, interactive: bool = True) -> WizardResult:
151 """Run the analysis wizard.
153 Args:
154 interactive: If True, prompt for user input
156 Returns:
157 WizardResult with all collected data
158 """
159 print("\n" + "=" * 60)
160 print("TraceKit Analysis Wizard")
161 print("=" * 60)
162 print("Let's analyze your signal step by step.\n")
164 # Show trace summary
165 self._show_trace_summary()
167 for i, step in enumerate(self.steps):
168 # Check skip condition
169 if step.skip_condition and step.skip_condition(self.result): 169 ↛ 170line 169 didn't jump to line 170 because the condition on line 169 was never true
170 continue
172 print(f"\n{'=' * 60}")
173 print(f"Step {i + 1}/{len(self.steps)}: {step.title}")
174 print("=" * 60)
176 if step.help_text: 176 ↛ 179line 176 didn't jump to line 179 because the condition on line 176 was always true
177 print(f"Tip: {step.help_text}\n")
179 print(step.question)
180 for j, option in enumerate(step.options, 1):
181 print(f" {j}. {option}")
183 if interactive: 183 ↛ 184line 183 didn't jump to line 184 because the condition on line 183 was never true
184 choice = self._get_user_choice(len(step.options))
185 else:
186 choice = 1 # Auto-select first option
188 if step.action: 188 ↛ 191line 188 didn't jump to line 191 because the condition on line 188 was always true
189 step.action(choice)
191 self.result.steps_completed = i + 1
193 # Generate summary
194 self._generate_summary()
196 print("\n" + "=" * 60)
197 print("Analysis Complete!")
198 print("=" * 60)
199 print(self.result.summary)
201 if self.result.recommendations:
202 print("\nRecommendations:")
203 for rec in self.result.recommendations:
204 print(f" - {rec}")
206 return self.result
208 def _show_trace_summary(self) -> None:
209 """Show a summary of the loaded trace."""
210 trace = self.trace
211 print("Loaded trace summary:")
213 if hasattr(trace, "data"): 213 ↛ 216line 213 didn't jump to line 216 because the condition on line 213 was always true
214 print(f" Samples: {len(trace.data):,}")
216 if hasattr(trace, "metadata"):
217 meta = trace.metadata
218 if hasattr(meta, "sample_rate") and meta.sample_rate: 218 ↛ 227line 218 didn't jump to line 227 because the condition on line 218 was always true
219 rate = meta.sample_rate
220 if rate >= 1e9:
221 print(f" Sample rate: {rate / 1e9:.3f} GSa/s")
222 elif rate >= 1e6:
223 print(f" Sample rate: {rate / 1e6:.3f} MSa/s")
224 else:
225 print(f" Sample rate: {rate / 1e3:.3f} kSa/s")
227 if hasattr(meta, "channel_name") and meta.channel_name: 227 ↛ 230line 227 didn't jump to line 230 because the condition on line 227 was always true
228 print(f" Channel: {meta.channel_name}")
230 if hasattr(trace, "data"): 230 ↛ exitline 230 didn't return from function '_show_trace_summary' because the condition on line 230 was always true
231 import numpy as np
233 data = trace.data
234 print(f" Value range: {np.min(data):.4g} to {np.max(data):.4g}")
236 def _get_user_choice(self, max_options: int) -> int:
237 """Get user's choice with validation."""
238 while True:
239 try:
240 choice_str = input(f"\nEnter choice (1-{max_options}): ")
241 choice = int(choice_str)
242 if 1 <= choice <= max_options:
243 return choice
244 print(f"Please enter a number between 1 and {max_options}")
245 except ValueError:
246 print("Please enter a valid number")
248 def _handle_signal_type(self, choice: int) -> None:
249 """Handle signal type selection."""
250 if choice == 1: # Auto-detect
251 print("\nAuto-detecting signal type...")
252 try:
253 from tracekit.discovery import characterize_signal
255 result = characterize_signal(self.trace)
256 self.result.measurements["signal_type"] = result.signal_type
257 self.result.measurements["signal_confidence"] = result.confidence
258 print(f"Detected: {result.signal_type} (confidence: {result.confidence:.0%})")
260 if result.confidence < 0.8:
261 self.result.recommendations.append(
262 f"Signal type detection has low confidence. "
263 f"Consider alternatives: {[a.signal_type for a in result.alternatives[:2]]}" # type: ignore[attr-defined]
264 )
265 except Exception as e:
266 print(f"Auto-detection failed: {e}")
267 self.result.measurements["signal_type"] = "unknown"
269 elif choice == 2: # Digital
270 self.result.measurements["signal_type"] = "digital"
271 print("\nDigital analysis mode selected.")
273 elif choice == 3: # Analog
274 self.result.measurements["signal_type"] = "analog"
275 print("\nAnalog/waveform analysis mode selected.")
277 elif choice == 4: # Protocol
278 self.result.measurements["signal_type"] = "protocol"
279 print("\nProtocol decoding mode selected.")
280 self.result.recommendations.append(
281 "For protocol decoding, try: decode_uart(), decode_spi(), decode_i2c()"
282 )
284 elif choice == 5: # Power 284 ↛ exitline 284 didn't return from function '_handle_signal_type' because the condition on line 284 was always true
285 self.result.measurements["signal_type"] = "power"
286 print("\nPower analysis mode selected.")
288 def _handle_measurements(self, choice: int) -> None:
289 """Handle measurement selection."""
290 import tracekit as tk
292 if choice == 4: # Skip
293 print("\nSkipping measurements.")
294 return
296 print("\nRunning measurements...")
298 try:
299 if choice == 1: # All 299 ↛ 301line 299 didn't jump to line 301 because the condition on line 299 was always true
300 results = tk.measure(self.trace)
301 elif choice == 2: # Timing only
302 results = {
303 "rise_time": tk.rise_time(self.trace),
304 "fall_time": tk.fall_time(self.trace),
305 "frequency": tk.frequency(self.trace),
306 "period": tk.period(self.trace),
307 "duty_cycle": tk.duty_cycle(self.trace),
308 }
309 elif choice == 3: # Amplitude only
310 results = {
311 "amplitude": tk.amplitude(self.trace),
312 "rms": tk.rms(self.trace),
313 "mean": tk.mean(self.trace),
314 "overshoot": tk.overshoot(self.trace),
315 "undershoot": tk.undershoot(self.trace),
316 }
318 self.result.measurements.update(results)
320 print("\nMeasurement results:")
321 for name, value in results.items():
322 if isinstance(value, float): 322 ↛ 323line 322 didn't jump to line 323 because the condition on line 322 was never true
323 print(f" {name}: {value:.4g}")
324 else:
325 print(f" {name}: {value}")
327 except Exception as e:
328 print(f"Measurement error: {e}")
330 def _handle_spectral(self, choice: int) -> None:
331 """Handle spectral analysis selection."""
332 import numpy as np
334 import tracekit as tk
336 if choice == 4: # Skip
337 print("\nSkipping spectral analysis.")
338 return
340 print("\nRunning spectral analysis...")
342 try:
343 if choice in (1, 3): # FFT 343 ↛ 350line 343 didn't jump to line 350 because the condition on line 343 was always true
344 freq, mag = tk.fft(self.trace) # type: ignore[misc]
345 peak_idx = np.argmax(mag)
346 self.result.measurements["fft_peak_freq"] = freq[peak_idx]
347 self.result.measurements["fft_peak_mag"] = mag[peak_idx]
348 print(f" FFT peak: {freq[peak_idx] / 1e6:.3f} MHz at {mag[peak_idx]:.1f} dB")
350 if choice in (2, 3): # PSD 350 ↛ 351line 350 didn't jump to line 351 because the condition on line 350 was never true
351 freq, _psd_vals = tk.psd(self.trace)
352 self.result.measurements["psd_computed"] = True
353 print(f" PSD computed over {len(freq)} frequency bins")
355 except Exception as e:
356 print(f"Spectral analysis error: {e}")
358 def _handle_quality(self, choice: int) -> None:
359 """Handle quality assessment selection."""
360 import tracekit as tk
362 if choice == 4: # Skip
363 print("\nSkipping quality assessment.")
364 return
366 print("\nAssessing signal quality...")
368 try:
369 if choice in (1, 3): # THD and SNR 369 ↛ 387line 369 didn't jump to line 387 because the condition on line 369 was always true
370 thd_val = tk.thd(self.trace)
371 snr_val = tk.snr(self.trace)
372 self.result.measurements["thd"] = thd_val
373 self.result.measurements["snr"] = snr_val
374 print(f" THD: {thd_val:.1f} dB")
375 print(f" SNR: {snr_val:.1f} dB")
377 # Recommendations based on quality
378 if thd_val > -40:
379 self.result.recommendations.append(
380 f"THD is {thd_val:.1f} dB - consider filtering to reduce distortion"
381 )
382 if snr_val < 40:
383 self.result.recommendations.append(
384 f"SNR is {snr_val:.1f} dB - signal is noisy, try averaging or filtering"
385 )
387 if choice in (2, 3): # Anomalies 387 ↛ 388line 387 didn't jump to line 388 because the condition on line 387 was never true
388 from tracekit.discovery import find_anomalies
390 anomalies = find_anomalies(self.trace)
391 self.result.measurements["anomaly_count"] = len(anomalies)
392 print(f" Found {len(anomalies)} anomalies")
394 if anomalies:
395 self.result.recommendations.append(
396 f"Found {len(anomalies)} anomalies - review the anomaly list for issues"
397 )
399 except Exception as e:
400 print(f"Quality assessment error: {e}")
402 def _generate_summary(self) -> None:
403 """Generate a human-readable summary of the analysis."""
404 lines = ["Analysis Summary:"]
406 if "signal_type" in self.result.measurements:
407 lines.append(f" Signal type: {self.result.measurements['signal_type']}")
409 if "frequency" in self.result.measurements:
410 freq = self.result.measurements["frequency"]
411 # Handle dict format from measure()
412 if isinstance(freq, dict):
413 freq = freq.get("value", 0)
415 if freq >= 1e6:
416 lines.append(f" Frequency: {freq / 1e6:.3f} MHz")
417 elif freq >= 1e3: 417 ↛ 418line 417 didn't jump to line 418 because the condition on line 417 was never true
418 lines.append(f" Frequency: {freq / 1e3:.3f} kHz")
419 else:
420 lines.append(f" Frequency: {freq:.1f} Hz")
422 if "rise_time" in self.result.measurements:
423 rt = self.result.measurements["rise_time"]
424 # Handle dict format from measure()
425 if isinstance(rt, dict):
426 rt = rt.get("value", 0)
428 lines.append(f" Rise time: {rt * 1e9:.2f} ns")
430 if "thd" in self.result.measurements:
431 thd = self.result.measurements["thd"]
432 # Handle dict format
433 if isinstance(thd, dict): 433 ↛ 434line 433 didn't jump to line 434 because the condition on line 433 was never true
434 thd = thd.get("value", 0)
435 lines.append(f" THD: {thd:.1f} dB")
437 if "snr" in self.result.measurements:
438 snr = self.result.measurements["snr"]
439 # Handle dict format
440 if isinstance(snr, dict): 440 ↛ 441line 440 didn't jump to line 441 because the condition on line 440 was never true
441 snr = snr.get("value", 0)
442 lines.append(f" SNR: {snr:.1f} dB")
444 self.result.summary = "\n".join(lines)
447def run_wizard(trace: Any, interactive: bool = True) -> WizardResult:
448 """Run the analysis wizard on a trace.
450 This is the main entry point for guided analysis.
452 Args:
453 trace: WaveformTrace or DigitalTrace to analyze
454 interactive: If True, prompt for user input
456 Returns:
457 WizardResult with measurements, recommendations, and summary
459 Example:
460 >>> import tracekit as tk
461 >>> from tracekit.onboarding import run_wizard
462 >>> trace = tk.load("signal.csv")
463 >>> result = run_wizard(trace)
464 """
465 wizard = AnalysisWizard(trace)
466 return wizard.run(interactive=interactive)