Coverage for src / tracekit / exporters / json_export.py: 38%
87 statements
« prev ^ index » next coverage.py v7.13.1, created at 2026-01-11 23:04 +0000
1"""JSON export functionality.
3This module provides measurement results export to JSON format.
6Example:
7 >>> from tracekit.exporters.json_export import export_json
8 >>> export_json(measurements, "results.json")
10References:
11 RFC 8259 (JSON format)
12"""
14from __future__ import annotations
16import json
17import math
18from dataclasses import asdict, is_dataclass
19from datetime import datetime
20from pathlib import Path
21from typing import Any
23import numpy as np
25from tracekit.core.types import DigitalTrace, TraceMetadata, WaveformTrace
28class TraceKitJSONEncoder(json.JSONEncoder):
29 """JSON encoder with numpy, datetime, and TraceKit object support."""
31 def default(self, obj: Any) -> Any:
32 if isinstance(obj, WaveformTrace):
33 return {
34 "_type": "WaveformTrace",
35 "data": obj.data.tolist(),
36 "metadata": self.default(obj.metadata),
37 }
38 if isinstance(obj, DigitalTrace):
39 return {
40 "_type": "DigitalTrace",
41 "data": obj.data.tolist(),
42 "metadata": self.default(obj.metadata),
43 "edges": obj.edges,
44 }
45 if isinstance(obj, TraceMetadata): 45 ↛ 59line 45 didn't jump to line 59 because the condition on line 45 was always true
46 return {
47 "_type": "TraceMetadata",
48 "sample_rate": obj.sample_rate,
49 "time_base": obj.time_base,
50 "vertical_scale": obj.vertical_scale,
51 "vertical_offset": obj.vertical_offset,
52 "acquisition_time": obj.acquisition_time.isoformat()
53 if obj.acquisition_time
54 else None,
55 "trigger_info": obj.trigger_info,
56 "source_file": obj.source_file,
57 "channel_name": obj.channel_name,
58 }
59 if isinstance(obj, np.ndarray):
60 return obj.tolist()
61 if isinstance(obj, np.integer | np.floating):
62 val = float(obj)
63 # Handle Infinity and NaN - convert to null for JSON compliance (RFC 8259)
64 if math.isinf(val) or math.isnan(val):
65 return None
66 return val
67 if isinstance(obj, np.bool_):
68 return bool(obj)
69 if isinstance(obj, float):
70 # Also handle Python float inf/nan
71 if math.isinf(obj) or math.isnan(obj):
72 return None
73 return obj
74 if isinstance(obj, datetime):
75 return obj.isoformat()
76 if isinstance(obj, complex):
77 # Handle complex with inf/nan components
78 if (
79 math.isinf(obj.real)
80 or math.isnan(obj.real)
81 or math.isinf(obj.imag)
82 or math.isnan(obj.imag)
83 ):
84 return None
85 return {"real": obj.real, "imag": obj.imag}
86 if isinstance(obj, bytes):
87 return obj.hex()
88 if is_dataclass(obj):
89 # Convert dataclasses to dict, then recursively encode
90 return asdict(obj) # type: ignore[arg-type]
91 return super().default(obj)
94def export_json(
95 data: WaveformTrace | DigitalTrace | dict[str, Any] | list[Any],
96 path: str | Path,
97 *,
98 pretty: bool = True,
99 include_metadata: bool = True,
100 compress: bool = False,
101) -> None:
102 """Export data to JSON format.
104 Args:
105 data: Data to export. Can be:
106 - WaveformTrace or DigitalTrace (full trace with metadata)
107 - Dictionary of measurements or data
108 - List of data
109 path: Output file path.
110 pretty: Use pretty printing with indentation.
111 include_metadata: Include export metadata.
112 compress: Compress output (save as .json.gz).
114 Example:
115 >>> results = measure(trace)
116 >>> export_json(results, "measurements.json")
117 >>> export_json(trace, "waveform.json", pretty=True)
118 >>> export_json(trace, "waveform.json.gz", compress=True)
120 References:
121 EXP-003
122 """
123 path = Path(path)
125 output: dict[str, Any] = {}
127 if include_metadata: 127 ↛ 134line 127 didn't jump to line 134 because the condition on line 127 was always true
128 output["_metadata"] = {
129 "format": "tracekit_json",
130 "version": "1.0",
131 "exported_at": datetime.now().isoformat(),
132 }
134 output["data"] = data
136 # Sanitize to handle inf/nan in nested dictionaries (Python float inf/nan)
137 # are handled directly by json encoder before calling default()
138 from tracekit.reporting.output import _sanitize_for_serialization
140 output = _sanitize_for_serialization(output)
142 # Serialize to JSON string
143 if pretty:
144 json_str = json.dumps(output, cls=TraceKitJSONEncoder, indent=2)
145 else:
146 json_str = json.dumps(output, cls=TraceKitJSONEncoder)
148 # Write to file (with optional compression)
149 if compress:
150 import gzip
152 # Ensure .gz extension
153 if not str(path).endswith(".gz"): 153 ↛ 154line 153 didn't jump to line 154 because the condition on line 153 was never true
154 path = path.with_suffix(path.suffix + ".gz")
156 with gzip.open(path, "wt", encoding="utf-8") as f:
157 f.write(json_str)
158 else:
159 with open(path, "w", encoding="utf-8") as f:
160 f.write(json_str)
163def export_measurements(
164 measurements: dict[str, Any],
165 path: str | Path,
166 *,
167 trace_info: dict[str, Any] | None = None,
168 pretty: bool = True,
169) -> None:
170 """Export measurement results to JSON.
172 Specialized function for measurement export with trace info.
174 Args:
175 measurements: Dictionary of measurements.
176 path: Output file path.
177 trace_info: Optional trace metadata.
178 pretty: Use pretty printing.
180 Example:
181 >>> measurements = measure(trace)
182 >>> trace_info = {
183 ... "source_file": "scope_capture.wfm",
184 ... "sample_rate": 1e9,
185 ... "duration": 0.001
186 ... }
187 >>> export_measurements(measurements, "results.json", trace_info=trace_info)
188 """
189 path = Path(path)
191 output = {
192 "_metadata": {
193 "format": "tracekit_measurements",
194 "version": "1.0",
195 "exported_at": datetime.now().isoformat(),
196 },
197 "measurements": measurements,
198 }
200 if trace_info:
201 output["trace_info"] = trace_info
203 # Sanitize to ensure inf/nan handling
204 from tracekit.reporting.output import _sanitize_for_serialization
206 output = _sanitize_for_serialization(output)
208 with open(path, "w") as f:
209 if pretty:
210 json.dump(output, f, cls=TraceKitJSONEncoder, indent=2)
211 else:
212 json.dump(output, f, cls=TraceKitJSONEncoder)
215def export_protocol_decode(
216 packets: list[dict[str, Any]],
217 path: str | Path,
218 *,
219 protocol: str = "unknown",
220 trace_info: dict[str, Any] | None = None,
221 pretty: bool = True,
222) -> None:
223 """Export protocol decode results to JSON.
225 Args:
226 packets: List of decoded packets.
227 path: Output file path.
228 protocol: Protocol name.
229 trace_info: Optional trace metadata.
230 pretty: Use pretty printing.
232 Example:
233 >>> packets = [{"timestamp": 0.001, "data": "0x48"}]
234 >>> export_protocol_decode(packets, "uart_decode.json", protocol="uart")
235 """
236 path = Path(path)
238 output = {
239 "_metadata": {
240 "format": "tracekit_protocol",
241 "version": "1.0",
242 "exported_at": datetime.now().isoformat(),
243 "protocol": protocol,
244 },
245 "packets": packets,
246 "summary": {
247 "total_packets": len(packets),
248 },
249 }
251 if trace_info:
252 output["trace_info"] = trace_info
254 # Sanitize to ensure inf/nan handling
255 from tracekit.reporting.output import _sanitize_for_serialization
257 output = _sanitize_for_serialization(output)
259 with open(path, "w") as f:
260 if pretty:
261 json.dump(output, f, cls=TraceKitJSONEncoder, indent=2)
262 else:
263 json.dump(output, f, cls=TraceKitJSONEncoder)
266def load_json(path: str | Path) -> dict[str, Any]:
267 """Load JSON data file.
269 Args:
270 path: Input file path.
272 Returns:
273 Loaded data dictionary.
275 Example:
276 >>> data = load_json("results.json")
277 >>> measurements = data.get("measurements", data.get("data", {}))
278 """
279 path = Path(path)
281 with open(path) as f:
282 return json.load(f) # type: ignore[no-any-return]
285__all__ = [
286 "TraceKitJSONEncoder",
287 "export_json",
288 "export_measurements",
289 "export_protocol_decode",
290 "load_json",
291]