Coverage for src/signalk_cli/stream/cli.py: 89%
106 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-14 12:47 +0000
1"""Click CLI for the SignalK v1 Streaming (delta) API."""
3import json
4import re
5import sys
6from datetime import UTC, datetime
7from pathlib import Path
8from urllib.parse import urlparse
10import click
11import niquests
13from ..net import api_error, bare_option, host_option, resolve_host, stderr_ctx
14from .output import (
15 FEATHER_EXTENSIONS,
16 delta_matches_source,
17 extract_delta_rows,
18 write_csv_delta,
19 write_csv_header,
20 write_feather_rows,
21 write_json_delta,
22 write_values_delta,
23)
24from .stream_api import (
25 SUBSCRIBE_POLICIES,
26 SUBSCRIPTION_POLICIES,
27 build_subscribe_message,
28 iter_deltas,
29 open_stream,
30)
32_AUTO_OUTPUT = "__auto_output__"
34# ---------------------------------------------------------------------------
35# CLI group
36# ---------------------------------------------------------------------------
39@click.group(context_settings={"help_option_names": ["-h", "--help"]})
40def cli():
41 """SignalK v1 streaming (delta) CLI."""
44# ---------------------------------------------------------------------------
45# deltas
46# ---------------------------------------------------------------------------
49@cli.command()
50@click.argument("paths", nargs=-1, required=False, metavar="PATH...")
51@host_option
52@click.option("--no-cache", is_flag=True, help="Ignore cached host")
53@click.option(
54 "--context",
55 "-c",
56 default="vessels.self",
57 show_default=True,
58 help="SignalK context for the explicit subscription this command "
59 "always sends. Also accepts the SignalK wildcard '*' (or 'vessels.*') "
60 "to subscribe to every vessel at your own --policy/--period — the "
61 "alternative to --subscribe all, which uses the server's default rate.",
62)
63@click.option(
64 "--subscribe",
65 type=click.Choice(SUBSCRIBE_POLICIES, case_sensitive=False),
66 default="none",
67 show_default=True,
68 help="Connection-level auto-subscribe at the server's OWN default "
69 "policy/period — separate from, and in addition to, this command's "
70 "explicit --context subscription above. Defaults to 'none' (the "
71 "SignalK spec's own default is 'self') to avoid double-subscribing "
72 "your own context. 'all' adds other vessels at the server's rate; for "
73 "other vessels at your chosen rate, use --context '*' instead.",
74)
75@click.option(
76 "--policy",
77 type=click.Choice(SUBSCRIPTION_POLICIES, case_sensitive=False),
78 default="ideal",
79 show_default=True,
80 help="Per-path subscribe policy (SignalK Subscription Protocol 'policy' "
81 "field): 'instant' sends every change, throttled by --min-period; "
82 "'ideal' (default) behaves like instant but resends the last value if "
83 "nothing changes within --period; 'fixed' always sends the last known "
84 "value every --period regardless of changes.",
85)
86@click.option(
87 "--period",
88 type=float,
89 default=60.0,
90 show_default=True,
91 metavar="SECONDS",
92 help="Subscribe period in seconds — the resend interval used by the "
93 "'ideal'/'fixed' policies. Converted to milliseconds on the wire.",
94)
95@click.option(
96 "--min-period",
97 "min_period",
98 type=float,
99 default=None,
100 metavar="SECONDS",
101 help="Fastest allowed transmission rate in seconds; only meaningful "
102 "with --policy instant. Converted to milliseconds on the wire.",
103)
104@click.option(
105 "--format",
106 "fmt",
107 default=None,
108 type=click.Choice(
109 ["csv", "json", "raw", "feather", "values"], case_sensitive=False
110 ),
111 help="Output format (default: inferred from --output extension, else csv). "
112 "json is JSON Lines (one row object per line). raw is the exact delta "
113 "message text, one per line. values is the bare value only, one per "
114 "line — no timestamp/context/source/path/kind columns. feather requires "
115 "pip install 'signalk-cli[feather]' and --output (cannot stream to stdout).",
116)
117@click.option("--no-header", is_flag=True, help="Suppress header row (CSV only)")
118@click.option(
119 "--include-meta",
120 is_flag=True,
121 help="Also emit rows for 'meta' entries (units, description, zones, etc.), "
122 "not just 'values'. Adds a 'kind' column (value/meta) to csv/json/feather "
123 "output. Ignored for --format raw, which always includes meta as-is.",
124)
125@click.option(
126 "--source",
127 "source",
128 multiple=True,
129 metavar="PATTERN",
130 help="Only include updates whose $source matches PATTERN. Repeatable "
131 "(OR'd together). PATTERN is a substring match unless it contains a "
132 "glob metacharacter (*/?/[), in which case it's matched as a glob, "
133 "e.g. --source Teltonika or --source '*.GP'. Filtering is client-side, "
134 "applied after receipt — for --format raw (whole message, verbatim) a "
135 "message passes if ANY of its updates match; other formats filter "
136 "per-update.",
137)
138@click.option(
139 "--output",
140 "-o",
141 is_flag=False,
142 flag_value=_AUTO_OUTPUT,
143 default=None,
144 metavar="FILE",
145 help="Write to FILE. Omit for stdout (default). Give without a filename to "
146 "auto-name the file. Required for --format feather.",
147)
148@click.option(
149 "--follow",
150 "-f",
151 is_flag=True,
152 help="Keep streaming until interrupted (Ctrl-C) or --count is reached. "
153 "Without this, print the next message then exit.",
154)
155@click.option(
156 "--count",
157 "-n",
158 type=int,
159 default=None,
160 metavar="N",
161 help="Number of delta messages to output. Default: 1 without --follow, "
162 "unlimited with --follow.",
163)
164@bare_option
165def deltas(
166 paths,
167 host,
168 no_cache,
169 context,
170 subscribe,
171 policy,
172 period,
173 min_period,
174 fmt,
175 no_header,
176 include_meta,
177 source,
178 output,
179 follow,
180 count,
181 bare,
182):
183 """Stream live delta updates from the SignalK v1 Streaming API.
185 Connects via WebSocket and prints delta messages as they arrive, in
186 csv, json (JSON Lines), raw, values (bare value only), or Feather
187 format.
189 Always sends an explicit subscribe message for --context, covering
190 PATH arguments if given, otherwise every path ('*'). PATH arguments
191 are sent verbatim to the server, one per path. They may be literal
192 SignalK paths (e.g. navigation.speedOverGround) or contain the
193 SignalK subscription wildcard '*', matched server-side per the
194 Subscription Protocol: '*' at the end of a path matches any suffix
195 (navigation.*), and '*' as a middle segment matches any single
196 segment there (propulsion.*.oilTemperature). Quote wildcarded paths
197 to stop the shell expanding them. --policy/--period/--min-period
198 control how that subscription behaves; --subscribe only controls
199 whether the connection *additionally* auto-subscribes at the
200 server's own default policy.
202 \b
203 Examples:
204 # Next update for one path, then exit
205 signalk_cli.stream deltas --host 10.36.10.21 navigation.speedOverGround
207 # Tail all navigation updates every 5s until Ctrl-C
208 signalk_cli.stream deltas --host 10.36.10.21 --follow --period 5 'navigation.*'
210 # Tail oil temperature across every engine, sent instantly on change
211 signalk_cli.stream deltas --host 10.36.10.21 --follow --policy instant \\
212 'propulsion.*.oilTemperature'
214 # Next 20 messages across all paths, as JSON Lines
215 signalk_cli.stream deltas --host 10.36.10.21 --format json --count 20
217 # Capture 100 messages to a Feather file (requires signalk-cli[feather])
218 signalk_cli.stream deltas --host 10.36.10.21 --count 100 -o capture.feather
220 # Bare speed values from one sensor, piped straight into another tool
221 signalk_cli.stream deltas --host 10.36.10.21 --follow --format values \\
222 --source Teltonika --bare navigation.speedOverGround
223 """
224 with stderr_ctx(bare):
225 host = resolve_host(host, no_cache)
226 period_ms = int(period * 1000)
227 min_period_ms = int(min_period * 1000) if min_period is not None else None
229 auto_name = output == _AUTO_OUTPUT
231 if fmt is None:
232 if output and not auto_name and output != "-":
233 suffix = Path(output).suffix.lower()
234 if suffix in FEATHER_EXTENSIONS: 234 ↛ 235line 234 didn't jump to line 235 because the condition on line 234 was never true
235 fmt = "feather"
236 elif suffix == ".json":
237 fmt = "json"
238 else:
239 fmt = "csv"
240 else:
241 fmt = "csv"
243 if auto_name: 243 ↛ 244line 243 didn't jump to line 244 because the condition on line 243 was never true
244 server_name = urlparse(host).hostname or re.sub(r"[^\w.-]", "_", host)
245 ts = datetime.now(UTC).strftime("%Y%m%dT%H%M%SZ")
246 ext = (
247 ".feather"
248 if fmt == "feather"
249 else ".json"
250 if fmt in ("json", "raw")
251 else ".txt"
252 if fmt == "values"
253 else ".csv"
254 )
255 output = f"signalk-stream-{server_name}-{ts}{ext}"
257 write_to_stdout = output is None or output == "-"
258 write_to_file = not write_to_stdout
260 if fmt == "feather" and write_to_stdout:
261 raise click.UsageError(
262 "feather cannot be written to stdout (binary format); "
263 "use --output FILE or --output to auto-name"
264 )
266 click.echo(f"Server: {host}", err=True)
267 click.echo(f"Context: {context}", err=True)
268 click.echo(f"Subscribe: {subscribe}", err=True)
269 min_period_note = (
270 f", min_period={min_period}s" if min_period is not None else ""
271 )
272 click.echo(
273 f"Policy: {policy} (period={period}s{min_period_note})", err=True
274 )
275 click.echo(f"Format: {fmt}", err=True)
277 try:
278 ws = open_stream(host, subscribe, timeout=None if follow else 30)
279 except niquests.RequestException as e:
280 click.echo(f"Error connecting to stream: {api_error(e)}", err=True)
281 sys.exit(1)
283 subscribe_message = build_subscribe_message(
284 context,
285 list(paths),
286 period_ms=period_ms,
287 policy=policy,
288 min_period_ms=min_period_ms,
289 )
290 ws.send_payload(json.dumps(subscribe_message))
292 effective_count = count if count is not None else (None if follow else 1)
294 message_count = 0
295 row_total = 0
296 header_written = False
297 feather_rows: list[tuple[str, ...]] = []
298 fh = (
299 open(output, "w", newline="") # noqa: SIM115
300 if write_to_file and fmt != "feather"
301 else None
302 )
303 sink = fh or sys.stdout
304 try:
305 for raw, delta in iter_deltas(ws, effective_count):
306 message_count += 1
307 if fmt == "feather": 307 ↛ 308line 307 didn't jump to line 308 because the condition on line 307 was never true
308 feather_rows.extend(
309 extract_delta_rows(
310 delta, include_meta=include_meta, sources=source
311 )
312 )
313 row_total = len(feather_rows)
314 elif fmt == "raw":
315 if delta_matches_source(delta, source):
316 click.echo(raw, file=sink)
317 elif fmt == "json":
318 row_total += write_json_delta(
319 delta, sink, include_meta=include_meta, sources=source
320 )
321 elif fmt == "values":
322 row_total += write_values_delta(
323 delta, sink, include_meta=include_meta, sources=source
324 )
325 else:
326 if not header_written and not no_header:
327 write_csv_header(sink, include_meta=include_meta)
328 header_written = True
329 row_total += write_csv_delta(
330 delta, sink, include_meta=include_meta, sources=source
331 )
332 except KeyboardInterrupt:
333 pass
334 except niquests.RequestException as e:
335 click.echo(f"Stream connection lost: {api_error(e)}", err=True)
336 finally:
337 ws.close()
338 if fh:
339 fh.close()
341 if fmt == "feather": 341 ↛ 342line 341 didn't jump to line 342 because the condition on line 341 was never true
342 write_feather_rows(feather_rows, output, include_meta=include_meta)
344 if write_to_file:
345 click.echo(f"Wrote {output}", err=True)
347 if fmt == "raw":
348 click.echo(f"{message_count} message(s)", err=True)
349 else:
350 click.echo(f"{message_count} message(s), {row_total} row(s)", err=True)