Coverage for src/signalk_cli/stream/output.py: 100%
81 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-14 12:47 +0000
1"""Row extraction and CSV/JSON/Feather writers for SignalK delta messages."""
3import csv
4import fnmatch
5import json
6from typing import IO
8FEATHER_EXTENSIONS = {".feather", ".arrow", ".fea"}
11def _normalize_value(value: object) -> str:
12 if isinstance(value, (dict, list)):
13 return json.dumps(value)
14 if value is None:
15 return ""
16 return str(value)
19def _update_source(update: dict) -> str:
20 return update.get("$source") or json.dumps(update.get("source", {}))
23def source_matches(source: str, patterns: tuple[str, ...]) -> bool:
24 """Match a `$source` string against `--source` filter patterns (OR'd).
26 No patterns means no filtering (always matches). A pattern containing
27 glob metacharacters (`*`/`?`/`[`) is matched as-is via `fnmatch`;
28 otherwise it's treated as a substring match, e.g. "Teltonika" matches
29 the source "Teltonika.GP".
30 """
31 if not patterns:
32 return True
33 return any(
34 fnmatch.fnmatch(source, p if any(c in p for c in "*?[") else f"*{p}*")
35 for p in patterns
36 )
39def delta_matches_source(delta: dict, patterns: tuple[str, ...]) -> bool:
40 """True if any update in the delta has a `$source` matching `patterns`.
42 Used for `--format raw`, which echoes the whole message verbatim and so
43 can only filter at message granularity, not per-update.
44 """
45 if not patterns:
46 return True
47 return any(
48 source_matches(_update_source(update), patterns)
49 for update in delta.get("updates", [])
50 )
53def extract_delta_rows(
54 delta: dict, *, include_meta: bool = False, sources: tuple[str, ...] = ()
55) -> list[tuple[str, ...]]:
56 """Flatten a single delta message into rows.
58 Without `include_meta`, rows are (timestamp, context, source, path,
59 value) from each update's "values" entries. With `include_meta`, a
60 "kind" column ("value"/"meta") is inserted before "value", and each
61 update's "meta" entries are included too — per the Streaming API spec,
62 "meta" entries have the same path/value shape but "value" is a metadata
63 object (units, description, zones, etc.), not a telemetry reading.
65 `sources`, if given, drops entire updates whose `$source` doesn't match
66 any pattern (see `source_matches`) — filtering is per-update, since
67 that's the granularity at which SignalK attaches a source.
68 """
69 context = delta.get("context", "")
70 rows: list[tuple[str, ...]] = []
71 for update in delta.get("updates", []):
72 source = _update_source(update)
73 if not source_matches(source, sources):
74 continue
75 timestamp = update.get("timestamp", "")
76 for entry in update.get("values", []):
77 path = entry.get("path", "")
78 value = _normalize_value(entry.get("value"))
79 if include_meta:
80 rows.append((timestamp, context, source, path, "value", value))
81 else:
82 rows.append((timestamp, context, source, path, value))
83 if include_meta:
84 for entry in update.get("meta", []):
85 path = entry.get("path", "")
86 value = _normalize_value(entry.get("value"))
87 rows.append((timestamp, context, source, path, "meta", value))
88 return rows
91CSV_COLUMNS = ["timestamp", "context", "source", "path", "value"]
92CSV_COLUMNS_WITH_KIND = ["timestamp", "context", "source", "path", "kind", "value"]
95def _columns(include_meta: bool) -> list[str]:
96 return CSV_COLUMNS_WITH_KIND if include_meta else CSV_COLUMNS
99def write_csv_header(sink: IO[str], *, include_meta: bool = False) -> None:
100 csv.writer(sink).writerow(_columns(include_meta))
101 sink.flush()
104def write_csv_delta(
105 delta: dict,
106 sink: IO[str],
107 *,
108 include_meta: bool = False,
109 sources: tuple[str, ...] = (),
110) -> int:
111 """Write one delta's rows as CSV lines. Returns the number of rows written."""
112 rows = extract_delta_rows(delta, include_meta=include_meta, sources=sources)
113 writer = csv.writer(sink)
114 for row in rows:
115 writer.writerow(row)
116 sink.flush()
117 return len(rows)
120def write_json_delta(
121 delta: dict,
122 sink: IO[str],
123 *,
124 include_meta: bool = False,
125 sources: tuple[str, ...] = (),
126) -> int:
127 """Write one delta's rows as JSON Lines (one row object per line). Returns row count."""
128 rows = extract_delta_rows(delta, include_meta=include_meta, sources=sources)
129 columns = _columns(include_meta)
130 for row in rows:
131 sink.write(json.dumps(dict(zip(columns, row))))
132 sink.write("\n")
133 sink.flush()
134 return len(rows)
137def write_values_delta(
138 delta: dict,
139 sink: IO[str],
140 *,
141 include_meta: bool = False,
142 sources: tuple[str, ...] = (),
143) -> int:
144 """Write one delta's bare values, one per line — no other columns.
146 For `--format values`: useful for piping a single path's readings
147 straight into another tool/script. Returns the number of values written.
148 """
149 rows = extract_delta_rows(delta, include_meta=include_meta, sources=sources)
150 for row in rows:
151 sink.write(row[-1])
152 sink.write("\n")
153 sink.flush()
154 return len(rows)
157def write_feather_rows(
158 rows: list[tuple[str, ...]], output: str, *, include_meta: bool = False
159) -> int:
160 """Write accumulated delta rows as Feather. Returns the number of rows written.
162 Unlike CSV/JSON, Feather cannot be appended to incrementally — callers must
163 buffer rows across messages and call this once at the end of the session.
164 """
165 try:
166 import pyarrow as pa
167 from pyarrow import feather
168 except ImportError:
169 raise ImportError(
170 "pyarrow is required for Feather output: pip install 'signalk-cli[feather]'"
171 ) from None
172 columns = _columns(include_meta)
173 row_columns = list(zip(*rows)) if rows else [()] * len(columns)
174 table = pa.table(
175 {
176 name: pa.array(values, type=pa.string())
177 for name, values in zip(columns, row_columns)
178 }
179 )
180 feather.write_feather(table, output)
181 return len(rows)