Coverage for src / tracekit / core / results.py: 98%
89 statements
« prev ^ index » next coverage.py v7.13.1, created at 2026-01-11 23:04 +0000
1"""Analysis result classes with intermediate data access.
3This module provides rich result objects that store intermediate computation
4results (FFT coefficients, filter states, etc.) for multi-step analysis
5without recomputation.
6"""
8from __future__ import annotations
10from dataclasses import dataclass, field
11from typing import TYPE_CHECKING, Any
13import numpy as np
15if TYPE_CHECKING:
16 from numpy.typing import NDArray
18 from .types import WaveformTrace
21@dataclass
22class AnalysisResult:
23 """Container for analysis results with intermediate data.
25 Stores the final result along with intermediate computation artifacts
26 like FFT coefficients, filter states, wavelet coefficients, etc.
27 Enables multi-step analysis without recomputation.
29 Attributes:
30 value: The final computed value (measurement, trace, etc.).
31 intermediates: Dictionary of intermediate computation results.
32 metadata: Additional metadata about the computation.
34 Example:
35 >>> result = AnalysisResult(
36 ... value=42.5,
37 ... intermediates={'fft_coeffs': coeffs, 'frequencies': freqs}
38 ... )
39 >>> fft_data = result.get_intermediate('fft_coeffs')
41 References:
42 API-005: Intermediate Result Access
43 """
45 value: Any
46 intermediates: dict[str, Any] = field(default_factory=dict)
47 metadata: dict[str, Any] = field(default_factory=dict)
49 def get_intermediate(self, key: str) -> Any:
50 """Get intermediate result by key.
52 Args:
53 key: Name of the intermediate result.
55 Returns:
56 The intermediate data.
58 Raises:
59 KeyError: If key not found in intermediates.
61 Example:
62 >>> spectrum = result.get_intermediate('fft_spectrum')
63 """
64 if key not in self.intermediates:
65 available = list(self.intermediates.keys())
66 raise KeyError(f"Intermediate '{key}' not found. Available: {available}")
67 return self.intermediates[key]
69 def has_intermediate(self, key: str) -> bool:
70 """Check if intermediate result exists.
72 Args:
73 key: Name of the intermediate result.
75 Returns:
76 True if key exists in intermediates.
78 Example:
79 >>> if result.has_intermediate('fft_coeffs'):
80 ... coeffs = result.get_intermediate('fft_coeffs')
81 """
82 return key in self.intermediates
84 def list_intermediates(self) -> list[str]:
85 """List all available intermediate result keys.
87 Returns:
88 List of intermediate result names.
90 Example:
91 >>> print(result.list_intermediates())
92 ['fft_spectrum', 'fft_frequencies', 'fft_power', 'fft_phase']
93 """
94 return list(self.intermediates.keys())
97@dataclass
98class FFTResult(AnalysisResult):
99 """Result object for FFT analysis with intermediate data.
101 Provides convenient access to FFT spectrum, frequencies, power,
102 and phase information.
104 Attributes:
105 spectrum: Complex FFT coefficients.
106 frequencies: Frequency bins in Hz.
107 power: Power spectrum (magnitude squared).
108 phase: Phase spectrum in radians.
109 trace: Original or transformed trace (optional).
111 Example:
112 >>> fft_result = tk.fft(trace, nfft=8192)
113 >>> spectrum = fft_result.spectrum
114 >>> frequencies = fft_result.frequencies
115 >>> power = fft_result.power
116 >>> phase = fft_result.phase
117 >>> peak_freq = frequencies[power.argmax()]
119 References:
120 API-005: Intermediate Result Access
121 """
123 spectrum: NDArray[np.complex128] = field(default_factory=lambda: np.array([]))
124 frequencies: NDArray[np.float64] = field(default_factory=lambda: np.array([]))
125 power: NDArray[np.float64] = field(default_factory=lambda: np.array([]))
126 phase: NDArray[np.float64] = field(default_factory=lambda: np.array([]))
127 trace: WaveformTrace | None = None
129 def __post_init__(self) -> None:
130 """Initialize intermediate results dictionary."""
131 # Store as intermediates for generic access
132 self.intermediates.update(
133 {
134 "spectrum": self.spectrum,
135 "frequencies": self.frequencies,
136 "power": self.power,
137 "phase": self.phase,
138 }
139 )
140 if self.trace is not None:
141 self.intermediates["trace"] = self.trace
143 # Set value to spectrum by default
144 if self.value is None:
145 self.value = self.spectrum
147 @property
148 def peak_frequency(self) -> float:
149 """Frequency of maximum power.
151 Returns:
152 Frequency in Hz where power spectrum peaks.
154 Example:
155 >>> print(f"Peak at {fft_result.peak_frequency:.2e} Hz")
156 """
157 if len(self.power) == 0:
158 return 0.0
159 return float(self.frequencies[self.power.argmax()])
161 @property
162 def magnitude(self) -> NDArray[np.float64]:
163 """Magnitude spectrum (absolute value of FFT).
165 Returns:
166 Magnitude of complex spectrum.
168 Example:
169 >>> mag = fft_result.magnitude
170 """
171 return np.abs(self.spectrum)
174@dataclass
175class FilterResult(AnalysisResult):
176 """Result object for filter operations with intermediate data.
178 Provides access to filtered trace along with filter characteristics
179 like transfer function and impulse response.
181 Attributes:
182 trace: Filtered WaveformTrace.
183 transfer_function: Filter transfer function H(f) (optional).
184 impulse_response: Filter impulse response h[n] (optional).
185 frequency_response: Tuple of (frequencies, response) (optional).
186 filter_coefficients: Filter coefficients (sos or ba format) (optional).
188 Example:
189 >>> filter_result = tk.low_pass(trace, cutoff=1e6, return_details=True)
190 >>> filtered_trace = filter_result.trace
191 >>> transfer_func = filter_result.transfer_function
192 >>> impulse_resp = filter_result.impulse_response
194 References:
195 API-005: Intermediate Result Access
196 API-009: Filter Introspection API
197 """
199 trace: WaveformTrace | None = None
200 transfer_function: NDArray[np.complex128] | None = None
201 impulse_response: NDArray[np.float64] | None = None
202 frequency_response: tuple[NDArray[np.float64], NDArray[np.complex128]] | None = None
203 filter_coefficients: Any | None = None
205 def __post_init__(self) -> None:
206 """Initialize intermediate results dictionary."""
207 if self.trace is not None:
208 self.intermediates["trace"] = self.trace
209 if self.transfer_function is not None:
210 self.intermediates["transfer_function"] = self.transfer_function
211 if self.impulse_response is not None:
212 self.intermediates["impulse_response"] = self.impulse_response
213 if self.frequency_response is not None:
214 self.intermediates["frequency_response"] = self.frequency_response
215 if self.filter_coefficients is not None:
216 self.intermediates["filter_coefficients"] = self.filter_coefficients
218 # Set value to trace by default
219 if self.value is None: 219 ↛ exitline 219 didn't return from function '__post_init__' because the condition on line 219 was always true
220 self.value = self.trace
223@dataclass
224class WaveletResult(AnalysisResult):
225 """Result object for wavelet transform with intermediate data.
227 Provides access to wavelet coefficients, scales, and frequencies.
229 Attributes:
230 coeffs: Wavelet coefficients.
231 scales: Wavelet scales.
232 frequencies: Corresponding frequencies in Hz.
233 trace: Original trace (optional).
235 Example:
236 >>> wavelet_result = tk.wavelet_transform(trace)
237 >>> coeffs = wavelet_result.coeffs
238 >>> scales = wavelet_result.scales
239 >>> frequencies = wavelet_result.frequencies
241 References:
242 API-005: Intermediate Result Access
243 """
245 coeffs: NDArray[np.complex128] | None = None
246 scales: NDArray[np.float64] | None = None
247 frequencies: NDArray[np.float64] | None = None
248 trace: WaveformTrace | None = None
250 def __post_init__(self) -> None:
251 """Initialize intermediate results dictionary."""
252 if self.coeffs is not None:
253 self.intermediates["coeffs"] = self.coeffs
254 if self.scales is not None:
255 self.intermediates["scales"] = self.scales
256 if self.frequencies is not None:
257 self.intermediates["frequencies"] = self.frequencies
258 if self.trace is not None:
259 self.intermediates["trace"] = self.trace
261 # Set value to coeffs by default
262 if self.value is None: 262 ↛ exitline 262 didn't return from function '__post_init__' because the condition on line 262 was always true
263 self.value = self.coeffs
266@dataclass
267class MeasurementResult(AnalysisResult):
268 """Result object for measurements with metadata.
270 Stores a measurement value along with units, method, and parameters
271 used for computation.
273 Attributes:
274 value: Measured value.
275 units: Units of measurement (e.g., 'V', 'Hz', 's').
276 method: Method or algorithm used.
277 parameters: Dictionary of parameters used.
278 confidence: Confidence interval or uncertainty (optional).
280 Example:
281 >>> result = MeasurementResult(
282 ... value=3.3,
283 ... units='V',
284 ... method='peak_to_peak',
285 ... parameters={'window': (0, 1e-3)}
286 ... )
288 References:
289 API-005: Intermediate Result Access
290 API-011: Measurement Provenance Tracking
291 """
293 units: str | None = None
294 method: str | None = None
295 parameters: dict[str, Any] = field(default_factory=dict)
296 confidence: tuple[float, float] | None = None
298 def __post_init__(self) -> None:
299 """Initialize metadata dictionary."""
300 self.metadata.update(
301 {
302 "units": self.units,
303 "method": self.method,
304 "parameters": self.parameters,
305 "confidence": self.confidence,
306 }
307 )
309 def __str__(self) -> str:
310 """String representation of measurement."""
311 if self.units:
312 return f"{self.value} {self.units}"
313 return str(self.value)
315 def __repr__(self) -> str:
316 """Detailed representation of measurement."""
317 parts = [f"value={self.value}"]
318 if self.units:
319 parts.append(f"units='{self.units}'")
320 if self.method:
321 parts.append(f"method='{self.method}'")
322 return f"MeasurementResult({', '.join(parts)})"
325__all__ = [
326 "AnalysisResult",
327 "FFTResult",
328 "FilterResult",
329 "MeasurementResult",
330 "WaveletResult",
331]