Coverage for src/signalk_cli/history/output.py: 79%
236 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-14 12:47 +0000
1"""CSV and Feather output writers for SignalK data."""
3import csv
4import json
5import re
6import sys
7from typing import IO, cast
9_POSITION_RE = re.compile(r"navigation.*\.position")
11FEATHER_EXTENSIONS = {".feather", ".arrow", ".fea"}
14class _MultiWriter:
15 """Fans writes out to multiple underlying streams."""
17 def __init__(self, *fhs):
18 self._fhs = fhs
20 def write(self, s):
21 for fh in self._fhs:
22 fh.write(s)
24 def flush(self):
25 for fh in self._fhs:
26 fh.flush()
29# ---------------------------------------------------------------------------
30# Narrow mode (single value column)
31# ---------------------------------------------------------------------------
34def extract_rows(result: dict) -> tuple[list, list, list, set[str]]:
35 """Flatten an API result into parallel (timestamps, paths, values, unique_paths) lists."""
36 value_columns = result.get("values", [])
37 data_rows = result.get("data", [])
38 timestamps, paths, values = [], [], []
39 unique_paths: set[str] = set()
41 for row in data_rows:
42 if not row: 42 ↛ 43line 42 didn't jump to line 43 because the condition on line 42 was never true
43 continue
44 timestamp = row[0]
45 for i, col in enumerate(value_columns):
46 path_name = col.get("path", f"col_{i}")
47 value = row[i + 1] if i + 1 < len(row) else None
48 if value is None:
49 continue
50 if isinstance(value, (dict, list)):
51 value = json.dumps(value)
52 elif not isinstance(value, str): 52 ↛ 54line 52 didn't jump to line 54 because the condition on line 52 was always true
53 value = str(value)
54 timestamps.append(timestamp)
55 paths.append(path_name)
56 values.append(value)
57 unique_paths.add(path_name)
59 return timestamps, paths, values, unique_paths
62def write_csv(result: dict, sink, no_header: bool) -> tuple[int, set[str]]:
63 """Write result as CSV rows (timestamp, path, value). Returns (row_count, unique_paths)."""
64 timestamps, paths, values, unique_paths = extract_rows(result)
65 writer = csv.writer(sink)
66 if not no_header:
67 writer.writerow(["timestamp", "path", "value"])
68 for ts, path, val in zip(timestamps, paths, values):
69 writer.writerow([ts, path, val])
70 return len(timestamps), unique_paths
73def write_feather(result: dict, output: str) -> tuple[int, set[str]]:
74 """Write result as Feather (timestamp, path, value). Returns (row_count, unique_paths)."""
75 try:
76 import pyarrow as pa
77 from pyarrow import feather
78 except ImportError:
79 raise ImportError(
80 "pyarrow is required for Feather output: pip install 'signalk-cli[feather]'"
81 ) from None
82 timestamps, paths, values, unique_paths = extract_rows(result)
83 table = pa.table(
84 {
85 "timestamp": pa.array(timestamps, type=pa.string()),
86 "path": pa.array(paths, type=pa.string()),
87 "value": pa.array(values, type=pa.string()),
88 }
89 )
90 feather.write_feather(table, output)
91 return len(timestamps), unique_paths
94# ---------------------------------------------------------------------------
95# Wide mode (min_value / max_value / avg_value columns)
96# ---------------------------------------------------------------------------
99def _cell(row: list, col_idx: int) -> str:
100 """Get a string cell value from a data row by 0-based column index (row[0] is timestamp)."""
101 i = col_idx + 1
102 if i >= len(row): 102 ↛ 103line 102 didn't jump to line 103 because the condition on line 102 was never true
103 return ""
104 v = row[i]
105 if v is None:
106 return ""
107 if isinstance(v, (dict, list)): 107 ↛ 108line 107 didn't jump to line 108 because the condition on line 107 was never true
108 return json.dumps(v)
109 return str(v)
112def _peek_first_value(data_rows: list, col_idx: int):
113 """First non-None value at col_idx+1 across data rows."""
114 for row in data_rows: 114 ↛ 119line 114 didn't jump to line 119 because the loop on line 114 didn't complete
115 if row: 115 ↛ 114line 115 didn't jump to line 114 because the condition on line 115 was always true
116 i = col_idx + 1
117 if i < len(row) and row[i] is not None:
118 return row[i]
119 return None
122def _array_col_names(path: str, length: int) -> list[str]:
123 """Column names for an array-valued path."""
124 if length == 2 and _POSITION_RE.fullmatch(path):
125 return ["longitude", "latitude"]
126 return [f"value_{i}" for i in range(length)]
129def extract_rows_wide(result: dict) -> tuple[list, list, dict[str, list], set[str]]:
130 """Extract rows as (timestamps, paths, value_cols, unique_paths).
132 value_cols is an ordered dict of column_name -> list of string values.
133 Scalar paths produce min_value / avg_value / max_value columns.
134 Array paths produce value_0 / value_1 / ... (or latitude / longitude for
135 navigation.*.position paths of length 2). Both may appear in one result;
136 non-applicable cells are empty strings.
137 """
138 value_columns = result.get("values", [])
139 data_rows = result.get("data", [])
141 # Build {path: {method: col_idx}} preserving path order
142 path_method_idx: dict[str, dict[str, int]] = {}
143 for i, col in enumerate(value_columns):
144 path = col.get("path", f"col_{i}")
145 method = col.get("method", "")
146 path_method_idx.setdefault(path, {})[method] = i
148 ordered_paths = list(path_method_idx)
150 # Determine per-path output columns by peeking at the first non-null value
151 path_col_names: dict[str, list[str]] = {}
152 path_is_array: dict[str, bool] = {}
153 for path, methods in path_method_idx.items():
154 sample = None
155 for col_idx in methods.values(): 155 ↛ 159line 155 didn't jump to line 159 because the loop on line 155 didn't complete
156 sample = _peek_first_value(data_rows, col_idx)
157 if sample is not None: 157 ↛ 155line 157 didn't jump to line 155 because the condition on line 157 was always true
158 break
159 if isinstance(sample, list):
160 path_is_array[path] = True
161 path_col_names[path] = _array_col_names(path, len(sample))
162 else:
163 path_is_array[path] = False
164 path_col_names[path] = ["min_value", "avg_value", "max_value"]
166 # Collect all unique column names in first-seen order
167 all_col_names: list[str] = []
168 seen_cols: set[str] = set()
169 for path in ordered_paths:
170 for col in path_col_names[path]:
171 if col not in seen_cols:
172 all_col_names.append(col)
173 seen_cols.add(col)
175 timestamps: list = []
176 paths_out: list = []
177 value_cols: dict[str, list] = {col: [] for col in all_col_names}
178 unique_paths: set[str] = set()
180 for row in data_rows:
181 if not row: 181 ↛ 182line 181 didn't jump to line 182 because the condition on line 181 was never true
182 continue
183 ts = row[0]
184 for path in ordered_paths:
185 methods = path_method_idx[path]
186 is_array = path_is_array[path]
187 cols = path_col_names[path]
189 if is_array:
190 arr_val = None
191 for col_idx in methods.values(): 191 ↛ 196line 191 didn't jump to line 196 because the loop on line 191 didn't complete
192 i = col_idx + 1
193 if i < len(row) and row[i] is not None: 193 ↛ 191line 193 didn't jump to line 191 because the condition on line 193 was always true
194 arr_val = row[i]
195 break
196 if arr_val is None: 196 ↛ 197line 196 didn't jump to line 197 because the condition on line 196 was never true
197 continue
198 timestamps.append(ts)
199 paths_out.append(path)
200 unique_paths.add(path)
201 row_vals: dict[str, str] = {c: "" for c in all_col_names}
202 for j, col_name in enumerate(cols):
203 if j < len(arr_val): 203 ↛ 202line 203 didn't jump to line 202 because the condition on line 203 was always true
204 row_vals[col_name] = str(arr_val[j])
205 for col_name in all_col_names:
206 value_cols[col_name].append(row_vals[col_name])
207 else:
208 mn = _cell(row, methods["min"]) if "min" in methods else ""
209 av = _cell(row, methods["average"]) if "average" in methods else ""
210 mx = _cell(row, methods["max"]) if "max" in methods else ""
211 if not mn and not av and not mx:
212 continue
213 timestamps.append(ts)
214 paths_out.append(path)
215 unique_paths.add(path)
216 row_vals = {c: "" for c in all_col_names}
217 row_vals["min_value"] = mn
218 row_vals["avg_value"] = av
219 row_vals["max_value"] = mx
220 for col_name in all_col_names:
221 value_cols[col_name].append(row_vals[col_name])
223 return timestamps, paths_out, value_cols, unique_paths
226def write_csv_wide(result: dict, sink, no_header: bool) -> tuple[int, set[str]]:
227 """Write result as CSV with dynamic value columns (scalar: min/avg/max; array: named elements)."""
228 timestamps, paths, value_cols, unique_paths = extract_rows_wide(result)
229 col_names = list(value_cols.keys())
230 writer = csv.writer(sink)
231 if not no_header: 231 ↛ 233line 231 didn't jump to line 233 because the condition on line 231 was always true
232 writer.writerow(["timestamp", "path"] + col_names)
233 for i, (ts, path) in enumerate(zip(timestamps, paths)):
234 writer.writerow([ts, path] + [value_cols[col][i] for col in col_names])
235 return len(timestamps), unique_paths
238def write_feather_wide(result: dict, output: str) -> tuple[int, set[str]]:
239 """Write result as Feather with dynamic value columns."""
240 try:
241 import pyarrow as pa
242 from pyarrow import feather
243 except ImportError:
244 raise ImportError(
245 "pyarrow is required for Feather output: pip install 'signalk-cli[feather]'"
246 ) from None
247 timestamps, paths, value_cols, unique_paths = extract_rows_wide(result)
248 table = pa.table(
249 {
250 "timestamp": pa.array(timestamps, type=pa.string()),
251 "path": pa.array(paths, type=pa.string()),
252 **{
253 col: pa.array(vals, type=pa.string())
254 for col, vals in value_cols.items()
255 },
256 }
257 )
258 feather.write_feather(table, output)
259 return len(timestamps), unique_paths
262def write_json(result: dict, sink, indent: int | None = None) -> tuple[int, set[str]]:
263 """Write result as JSON array of row objects (narrow mode)."""
264 timestamps, paths, values, unique_paths = extract_rows(result)
265 rows = [
266 {"timestamp": ts, "path": p, "value": v}
267 for ts, p, v in zip(timestamps, paths, values)
268 ]
269 sink.write(json.dumps(rows, indent=indent))
270 return len(rows), unique_paths
273def write_json_wide(
274 result: dict, sink, indent: int | None = None
275) -> tuple[int, set[str]]:
276 """Write result as JSON array of row objects (wide mode)."""
277 timestamps, paths, value_cols, unique_paths = extract_rows_wide(result)
278 col_names = list(value_cols.keys())
279 rows = [
280 {"timestamp": ts, "path": p, **{col: value_cols[col][i] for col in col_names}}
281 for i, (ts, p) in enumerate(zip(timestamps, paths))
282 ]
283 sink.write(json.dumps(rows, indent=indent))
284 return len(rows), unique_paths
287# ---------------------------------------------------------------------------
288# Cardinality
289# ---------------------------------------------------------------------------
291CARDINALITY_COLUMNS = [
292 "path",
293 "distinct_values",
294 "distinct_values_2_decimal_places",
295 "nulls",
296 "zeroes",
297 "min",
298 "max",
299 "average",
300]
303def compute_cardinality(result: dict) -> list[dict]:
304 """Compute per-path value statistics from a narrow-mode API result.
306 Each returned dict has keys matching CARDINALITY_COLUMNS. min, max,
307 average, and distinct_values_2_decimal_places are empty strings for
308 non-scalar (array/dict) paths.
309 """
310 value_columns = result.get("values", [])
311 data_rows = result.get("data", [])
313 ordered_paths: list[str] = []
314 path_col_idxs: dict[str, list[int]] = {}
315 for i, col in enumerate(value_columns):
316 path = col.get("path", f"col_{i}")
317 if path not in path_col_idxs: 317 ↛ 320line 317 didn't jump to line 320 because the condition on line 317 was always true
318 ordered_paths.append(path)
319 path_col_idxs[path] = []
320 path_col_idxs[path].append(i)
322 path_vals: dict[str, list] = {p: [] for p in ordered_paths}
323 path_nulls: dict[str, int] = {p: 0 for p in ordered_paths}
324 path_zeroes: dict[str, int] = {p: 0 for p in ordered_paths}
326 for row in data_rows:
327 if not row: 327 ↛ 328line 327 didn't jump to line 328 because the condition on line 327 was never true
328 continue
329 for path, col_idxs in path_col_idxs.items():
330 for col_idx in col_idxs:
331 val = row[col_idx + 1] if col_idx + 1 < len(row) else None
332 if val is None:
333 path_nulls[path] += 1
334 else:
335 if (
336 isinstance(val, (int, float))
337 and not isinstance(val, bool)
338 and val == 0
339 ):
340 path_zeroes[path] += 1
341 path_vals[path].append(val)
343 rows = []
344 for path in ordered_paths:
345 vals = path_vals[path]
346 nulls = path_nulls[path]
347 is_scalar = bool(vals) and all(
348 isinstance(v, (int, float)) and not isinstance(v, bool) for v in vals
349 )
351 distinct = len(
352 {
353 json.dumps(v, sort_keys=True) if isinstance(v, (dict, list)) else str(v)
354 for v in vals
355 }
356 )
358 if not vals:
359 mn = mx = avg = ""
360 d2dp = "0"
361 elif is_scalar:
362 mn = str(min(vals))
363 mx = str(max(vals))
364 avg = str(sum(vals) / len(vals))
365 d2dp = str(len({round(float(v), 2) for v in vals}))
366 else:
367 mn = mx = avg = ""
368 if vals and isinstance(vals[0], list): 368 ↛ 384line 368 didn't jump to line 384 because the condition on line 368 was always true
369 d2dp = str(
370 len(
371 {
372 tuple(
373 round(x, 2)
374 if isinstance(x, (int, float))
375 and not isinstance(x, bool)
376 else x
377 for x in v
378 )
379 for v in vals
380 }
381 )
382 )
383 else:
384 d2dp = ""
386 rows.append(
387 {
388 "path": path,
389 "distinct_values": str(distinct),
390 "min": mn,
391 "max": mx,
392 "average": avg,
393 "distinct_values_2_decimal_places": d2dp,
394 "nulls": str(nulls),
395 "zeroes": str(path_zeroes[path]),
396 }
397 )
399 return rows
402# ---------------------------------------------------------------------------
403# Sink helper
404# ---------------------------------------------------------------------------
407def csv_sink(
408 output: str, write_to_file: bool, write_to_stdout: bool
409) -> tuple[IO[str] | None, _MultiWriter | IO[str]]:
410 """Return (file_handle_or_None, sink) for CSV writing. Caller must close file_handle."""
411 file_fh: IO[str] | None = (
412 open(output, "w", newline="") if write_to_file else None # noqa: SIM115
413 )
414 sink: _MultiWriter | IO[str]
415 if write_to_file and write_to_stdout:
416 sink = _MultiWriter(cast(IO[str], file_fh), sys.stdout)
417 elif write_to_file:
418 sink = cast(IO[str], file_fh)
419 else:
420 sink = sys.stdout
421 return file_fh, sink