Coverage for src / tracekit / core / types.py: 100%
171 statements
« prev ^ index » next coverage.py v7.13.1, created at 2026-01-11 23:04 +0000
1"""Core data types for TraceKit signal analysis framework.
3This module implements the fundamental data structures for oscilloscope
4and logic analyzer data analysis.
6Requirements addressed:
7- CORE-001: TraceMetadata Data Class
8- CORE-002: WaveformTrace Data Class
9- CORE-003: DigitalTrace Data Class
10- CORE-004: ProtocolPacket Data Class
11- CORE-005: CalibrationInfo Data Class (regulatory compliance)
12"""
14from __future__ import annotations
16from dataclasses import dataclass, field
17from typing import TYPE_CHECKING, Any
19import numpy as np
21if TYPE_CHECKING:
22 from datetime import datetime
24 from numpy.typing import NDArray
27@dataclass
28class CalibrationInfo:
29 """Calibration and instrument provenance information.
31 Stores traceability metadata for measurements performed on oscilloscope
32 or logic analyzer data. Essential for regulatory compliance and quality
33 assurance in DOD/aerospace/medical applications.
35 Attributes:
36 instrument: Instrument make and model (e.g., "Tektronix DPO7254C").
37 serial_number: Instrument serial number for traceability (optional).
38 calibration_date: Date of last calibration (optional).
39 calibration_due_date: Date when next calibration is due (optional).
40 firmware_version: Instrument firmware version (optional).
41 calibration_lab: Calibration lab name or accreditation (optional).
42 calibration_cert_number: Calibration certificate number (optional).
43 probe_attenuation: Probe attenuation factor (e.g., 10.0 for 10x probe) (optional).
44 coupling: Input coupling ("DC", "AC", "GND") (optional).
45 bandwidth_limit: Bandwidth limit in Hz, None if disabled (optional).
46 vertical_resolution: ADC resolution in bits (optional).
48 Example:
49 >>> from datetime import datetime
50 >>> cal_info = CalibrationInfo(
51 ... instrument="Tektronix DPO7254C",
52 ... serial_number="C012345",
53 ... calibration_date=datetime(2024, 12, 15),
54 ... probe_attenuation=10.0,
55 ... vertical_resolution=8
56 ... )
57 >>> print(f"Instrument: {cal_info.instrument}")
58 Instrument: Tektronix DPO7254C
60 References:
61 ISO/IEC 17025: General Requirements for Testing/Calibration Laboratories
62 NIST Handbook 150: Laboratory Accreditation Program Requirements
63 21 CFR Part 11: Electronic Records (FDA)
64 """
66 instrument: str
67 serial_number: str | None = None
68 calibration_date: datetime | None = None
69 calibration_due_date: datetime | None = None
70 firmware_version: str | None = None
71 calibration_lab: str | None = None
72 calibration_cert_number: str | None = None
73 probe_attenuation: float | None = None
74 coupling: str | None = None
75 bandwidth_limit: float | None = None
76 vertical_resolution: int | None = None
78 def __post_init__(self) -> None:
79 """Validate calibration info after initialization."""
80 if self.probe_attenuation is not None and self.probe_attenuation <= 0:
81 raise ValueError(f"probe_attenuation must be positive, got {self.probe_attenuation}")
82 if self.bandwidth_limit is not None and self.bandwidth_limit <= 0:
83 raise ValueError(f"bandwidth_limit must be positive, got {self.bandwidth_limit}")
84 if self.vertical_resolution is not None and self.vertical_resolution <= 0:
85 raise ValueError(
86 f"vertical_resolution must be positive, got {self.vertical_resolution}"
87 )
89 @property
90 def is_calibration_current(self) -> bool | None:
91 """Check if calibration is current.
93 Returns:
94 True if calibration is current, False if expired, None if dates not set.
95 """
96 if self.calibration_date is None or self.calibration_due_date is None:
97 return None
98 from datetime import datetime
100 return datetime.now() < self.calibration_due_date
102 @property
103 def traceability_summary(self) -> str:
104 """Generate a traceability summary string.
106 Returns:
107 Human-readable summary of calibration traceability.
108 """
109 parts = [f"Instrument: {self.instrument}"]
110 if self.serial_number:
111 parts.append(f"S/N: {self.serial_number}")
112 if self.calibration_date:
113 parts.append(f"Cal Date: {self.calibration_date.strftime('%Y-%m-%d')}")
114 if self.calibration_due_date:
115 parts.append(f"Due: {self.calibration_due_date.strftime('%Y-%m-%d')}")
116 if self.calibration_cert_number:
117 parts.append(f"Cert: {self.calibration_cert_number}")
118 return ", ".join(parts)
121@dataclass
122class TraceMetadata:
123 """Metadata describing a captured trace.
125 Contains sample rate, scaling information, acquisition details,
126 and provenance information for a captured waveform or digital trace.
128 Attributes:
129 sample_rate: Sample rate in Hz (required).
130 vertical_scale: Vertical scale in volts/division (optional).
131 vertical_offset: Vertical offset in volts (optional).
132 acquisition_time: Time of acquisition (optional).
133 trigger_info: Trigger configuration dictionary (optional).
134 source_file: Path to source file (optional).
135 channel_name: Name of the channel (optional).
136 calibration_info: Calibration and instrument traceability information (optional).
138 Example:
139 >>> metadata = TraceMetadata(sample_rate=1e9) # 1 GSa/s
140 >>> print(f"Time base: {metadata.time_base} s/sample")
141 Time base: 1e-09 s/sample
143 Example with calibration info:
144 >>> from datetime import datetime
145 >>> cal = CalibrationInfo(
146 ... instrument="Tektronix DPO7254C",
147 ... calibration_date=datetime(2024, 12, 15)
148 ... )
149 >>> metadata = TraceMetadata(sample_rate=1e9, calibration_info=cal)
150 >>> print(metadata.calibration_info.traceability_summary)
151 Instrument: Tektronix DPO7254C, Cal Date: 2024-12-15
153 References:
154 IEEE 181-2011: Standard for Transitional Waveform Definitions
155 ISO/IEC 17025: General Requirements for Testing/Calibration Laboratories
156 """
158 sample_rate: float
159 vertical_scale: float | None = None
160 vertical_offset: float | None = None
161 acquisition_time: datetime | None = None
162 trigger_info: dict[str, Any] | None = None
163 source_file: str | None = None
164 channel_name: str | None = None
165 calibration_info: CalibrationInfo | None = None
167 def __post_init__(self) -> None:
168 """Validate metadata after initialization."""
169 if self.sample_rate <= 0:
170 raise ValueError(f"sample_rate must be positive, got {self.sample_rate}")
172 @property
173 def time_base(self) -> float:
174 """Time between samples in seconds (derived from sample_rate).
176 Returns:
177 Time per sample in seconds (1 / sample_rate).
178 """
179 return 1.0 / self.sample_rate
182@dataclass
183class WaveformTrace:
184 """Analog waveform data with metadata.
186 Stores sampled analog voltage data as a numpy array along with
187 associated metadata for timing and scaling.
189 Attributes:
190 data: Waveform samples as numpy float array.
191 metadata: Associated trace metadata.
193 Example:
194 >>> import numpy as np
195 >>> data = np.sin(2 * np.pi * 1e6 * np.linspace(0, 1e-3, 1000))
196 >>> trace = WaveformTrace(data=data, metadata=TraceMetadata(sample_rate=1e6))
197 >>> print(f"Duration: {trace.time_vector[-1]:.6f} seconds")
198 Duration: 0.000999 seconds
200 References:
201 IEEE 1241-2010: Standard for Terminology and Test Methods for ADCs
202 """
204 data: NDArray[np.floating[Any]]
205 metadata: TraceMetadata
207 def __post_init__(self) -> None:
208 """Validate waveform data after initialization."""
209 if not isinstance(self.data, np.ndarray):
210 raise TypeError(f"data must be a numpy array, got {type(self.data).__name__}")
211 if not np.issubdtype(self.data.dtype, np.floating):
212 # Convert to float64 if not already floating point
213 self.data = self.data.astype(np.float64)
215 @property
216 def time_vector(self) -> NDArray[np.float64]:
217 """Time axis in seconds.
219 Computes a time vector starting from 0, with intervals
220 determined by the sample rate.
222 Returns:
223 Array of time values in seconds, same length as data.
224 """
225 n_samples = len(self.data)
226 return np.arange(n_samples, dtype=np.float64) * self.metadata.time_base
228 @property
229 def duration(self) -> float:
230 """Total duration of the trace in seconds.
232 Returns:
233 Duration from first to last sample in seconds.
234 """
235 if len(self.data) == 0:
236 return 0.0
237 return (len(self.data) - 1) * self.metadata.time_base
239 def __len__(self) -> int:
240 """Return number of samples in the trace."""
241 return len(self.data)
244@dataclass
245class DigitalTrace:
246 """Digital/logic signal data with metadata.
248 Stores sampled digital signal data as a boolean numpy array,
249 with optional edge timestamp information.
251 Attributes:
252 data: Digital samples as numpy boolean array.
253 metadata: Associated trace metadata.
254 edges: Optional list of (timestamp, is_rising) tuples.
256 Example:
257 >>> import numpy as np
258 >>> data = np.array([False, False, True, True, False], dtype=bool)
259 >>> trace = DigitalTrace(data=data, metadata=TraceMetadata(sample_rate=1e6))
260 >>> print(f"High samples: {np.sum(trace.data)}")
261 High samples: 2
263 References:
264 IEEE 1076.6-2004: Standard for VHDL Register Transfer Level Synthesis
265 """
267 data: NDArray[np.bool_]
268 metadata: TraceMetadata
269 edges: list[tuple[float, bool]] | None = None
271 def __post_init__(self) -> None:
272 """Validate digital data after initialization."""
273 if not isinstance(self.data, np.ndarray):
274 raise TypeError(f"data must be a numpy array, got {type(self.data).__name__}")
275 if self.data.dtype != np.bool_:
276 # Convert to boolean if not already
277 self.data = self.data.astype(np.bool_)
279 @property
280 def time_vector(self) -> NDArray[np.float64]:
281 """Time axis in seconds.
283 Returns:
284 Array of time values in seconds, same length as data.
285 """
286 n_samples = len(self.data)
287 return np.arange(n_samples, dtype=np.float64) * self.metadata.time_base
289 @property
290 def duration(self) -> float:
291 """Total duration of the trace in seconds.
293 Returns:
294 Duration from first to last sample in seconds.
295 """
296 if len(self.data) == 0:
297 return 0.0
298 return (len(self.data) - 1) * self.metadata.time_base
300 @property
301 def rising_edges(self) -> list[float]:
302 """Timestamps of rising edges.
304 Returns:
305 List of timestamps where signal transitions from low to high.
306 """
307 if self.edges is None:
308 return []
309 return [ts for ts, is_rising in self.edges if is_rising]
311 @property
312 def falling_edges(self) -> list[float]:
313 """Timestamps of falling edges.
315 Returns:
316 List of timestamps where signal transitions from high to low.
317 """
318 if self.edges is None:
319 return []
320 return [ts for ts, is_rising in self.edges if not is_rising]
322 def __len__(self) -> int:
323 """Return number of samples in the trace."""
324 return len(self.data)
327@dataclass
328class IQTrace:
329 """I/Q (In-phase/Quadrature) waveform data with metadata.
331 Stores complex-valued signal data as separate I and Q components,
332 commonly used for RF and software-defined radio applications.
334 Attributes:
335 i_data: In-phase component samples as numpy float array.
336 q_data: Quadrature component samples as numpy float array.
337 metadata: Associated trace metadata.
339 Example:
340 >>> import numpy as np
341 >>> t = np.linspace(0, 1e-3, 1000)
342 >>> i_data = np.cos(2 * np.pi * 1e6 * t)
343 >>> q_data = np.sin(2 * np.pi * 1e6 * t)
344 >>> trace = IQTrace(i_data=i_data, q_data=q_data, metadata=TraceMetadata(sample_rate=1e6))
345 >>> print(f"Complex samples: {len(trace)}")
346 Complex samples: 1000
348 References:
349 IEEE Std 181-2011: Transitional Waveform Definitions
350 """
352 i_data: NDArray[np.floating[Any]]
353 q_data: NDArray[np.floating[Any]]
354 metadata: TraceMetadata
356 def __post_init__(self) -> None:
357 """Validate I/Q data after initialization."""
358 if not isinstance(self.i_data, np.ndarray):
359 raise TypeError(f"i_data must be a numpy array, got {type(self.i_data).__name__}")
360 if not isinstance(self.q_data, np.ndarray):
361 raise TypeError(f"q_data must be a numpy array, got {type(self.q_data).__name__}")
362 if len(self.i_data) != len(self.q_data):
363 raise ValueError(
364 f"I and Q data must have same length, got {len(self.i_data)} and {len(self.q_data)}"
365 )
366 # Convert to float64 if not already floating point
367 if not np.issubdtype(self.i_data.dtype, np.floating):
368 self.i_data = self.i_data.astype(np.float64)
369 if not np.issubdtype(self.q_data.dtype, np.floating):
370 self.q_data = self.q_data.astype(np.float64)
372 @property
373 def complex_data(self) -> NDArray[np.complex128]:
374 """Return I/Q data as complex array.
376 Returns:
377 Complex array where real=I, imag=Q.
378 """
379 return self.i_data + 1j * self.q_data
381 @property
382 def magnitude(self) -> NDArray[np.float64]:
383 """Magnitude (amplitude) of the complex signal.
385 Returns:
386 Array of magnitude values sqrt(I² + Q²).
387 """
388 return np.sqrt(self.i_data**2 + self.q_data**2)
390 @property
391 def phase(self) -> NDArray[np.float64]:
392 """Phase angle of the complex signal in radians.
394 Returns:
395 Array of phase values atan2(Q, I).
396 """
397 return np.arctan2(self.q_data, self.i_data)
399 @property
400 def time_vector(self) -> NDArray[np.float64]:
401 """Time axis in seconds.
403 Returns:
404 Array of time values in seconds, same length as data.
405 """
406 n_samples = len(self.i_data)
407 return np.arange(n_samples, dtype=np.float64) * self.metadata.time_base
409 @property
410 def duration(self) -> float:
411 """Total duration of the trace in seconds.
413 Returns:
414 Duration from first to last sample in seconds.
415 """
416 if len(self.i_data) == 0:
417 return 0.0
418 return (len(self.i_data) - 1) * self.metadata.time_base
420 def __len__(self) -> int:
421 """Return number of samples in the trace."""
422 return len(self.i_data)
425@dataclass
426class ProtocolPacket:
427 """Decoded protocol packet data.
429 Represents a decoded packet from a serial protocol (UART, SPI, I2C, etc.)
430 with timing, data content, annotations, and error information.
432 Attributes:
433 timestamp: Start time of the packet in seconds.
434 protocol: Name of the protocol (e.g., "UART", "SPI", "I2C").
435 data: Decoded data bytes.
436 annotations: Multi-level annotations dictionary (optional).
437 errors: List of detected errors (optional).
438 end_timestamp: End time of the packet in seconds (optional).
440 Example:
441 >>> packet = ProtocolPacket(
442 ... timestamp=1.23e-3,
443 ... protocol="UART",
444 ... data=b"Hello"
445 ... )
446 >>> print(f"Received at {packet.timestamp}s: {packet.data.decode()}")
447 Received at 0.00123s: Hello
449 References:
450 sigrok Protocol Decoder API
451 """
453 timestamp: float
454 protocol: str
455 data: bytes
456 annotations: dict[str, Any] = field(default_factory=dict)
457 errors: list[str] = field(default_factory=list)
458 end_timestamp: float | None = None
460 def __post_init__(self) -> None:
461 """Validate packet data after initialization."""
462 if self.timestamp < 0:
463 raise ValueError(f"timestamp must be non-negative, got {self.timestamp}")
464 if not isinstance(self.data, bytes):
465 raise TypeError(f"data must be bytes, got {type(self.data).__name__}")
467 @property
468 def duration(self) -> float | None:
469 """Duration of the packet in seconds.
471 Returns:
472 Duration if end_timestamp is set, None otherwise.
473 """
474 if self.end_timestamp is None:
475 return None
476 return self.end_timestamp - self.timestamp
478 @property
479 def has_errors(self) -> bool:
480 """Check if packet has any errors.
482 Returns:
483 True if errors list is non-empty.
484 """
485 return len(self.errors) > 0
487 def __len__(self) -> int:
488 """Return number of bytes in the packet."""
489 return len(self.data)
492# Type aliases for convenience
493Trace = WaveformTrace | DigitalTrace | IQTrace
494"""Union type for any trace type."""
496__all__ = [
497 "CalibrationInfo",
498 "DigitalTrace",
499 "IQTrace",
500 "ProtocolPacket",
501 "Trace",
502 "TraceMetadata",
503 "WaveformTrace",
504]