Coverage for src/signalk_cli/stream/stream_api.py: 100%

1"""SignalK v1 Streaming (delta) API client. 

2 

3Spec: https://signalk.org/specification/1.8.2/doc/streaming_api.html 

4Subscribe path wildcards: https://signalk.org/specification/1.8.2/doc/subscription_protocol.html 

5""" 

6 

7import json 

8from collections.abc import Iterator 

9from typing import Any 

10from urllib.parse import urlparse, urlunparse 

11 

12import niquests 

13 

14STREAM_PATH = "/signalk/v1/stream" 

15 

16SUBSCRIBE_POLICIES = ["none", "self", "all"] 

17SUBSCRIPTION_POLICIES = ["instant", "ideal", "fixed"] 

18 

19 

20def to_ws_url(host: str) -> str: 

21 """Convert an http(s) host base URL to the ws(s) streaming endpoint URL.""" 

22 parsed = urlparse(host) 

23 scheme = "wss" if parsed.scheme == "https" else "ws" 

24 return urlunparse((scheme, parsed.netloc, STREAM_PATH, "", "", "")) 

25 

26 

27def open_stream(host: str, subscribe: str, timeout: float | None = 30): 

28 """Open a WebSocket connection to the SignalK streaming endpoint. 

29 

30 `timeout` sets the socket's read timeout for the life of the connection — 

31 every subsequent `next_payload()` read reuses it, not just the initial 

32 handshake. Pass `None` when the caller intends to block indefinitely 

33 between messages (e.g. `--follow`), since deltas can legitimately go 

34 quiet for longer than any fixed timeout depending on subscribe policy. 

35 

36 Returns the underlying HTTP extension object, used to send/receive frames 

37 via `send_payload`/`next_payload`. 

38 """ 

39 url = to_ws_url(host) 

40 resp = niquests.get(url, params={"subscribe": subscribe}, timeout=timeout) 

41 resp.raise_for_status() 

42 return resp.extension 

43 

44 

45def build_subscribe_message( 

46 context: str, 

47 paths: list[str], 

48 *, 

49 period_ms: int | None = None, 

50 policy: str | None = None, 

51 min_period_ms: int | None = None, 

52) -> dict: 

53 """Build a client subscribe message for the given context and paths. 

54 

55 An empty path list subscribes to all paths within the context (equivalent 

56 to a single "*" path). Paths are passed through unchanged — wildcarding is 

57 handled server-side per the SignalK Subscription Protocol: "*" at the end 

58 of a path matches any suffix (e.g. "navigation.*"), and "*" as a middle 

59 segment matches any single segment there (e.g. "propulsion.*.oilTemperature"). 

60 

61 period_ms/policy/min_period_ms map directly to the per-path "period", 

62 "policy", and "minPeriod" fields of the Subscription Protocol and are 

63 applied identically to every path when given. `policy` 

64 ("instant"/"ideal"/"fixed") defaults to "ideal" server-side if omitted; 

65 `min_period_ms` only affects the "instant" policy. The protocol also 

66 defines a per-path "format" ("delta"/"full") field, but it's omitted 

67 here: signalk-server rejects "full" outright and always sends delta 

68 messages regardless, so exposing the choice would be misleading. 

69 """ 

70 entry_extra: dict[str, Any] = {} 

71 if period_ms is not None: 

72 entry_extra["period"] = period_ms 

73 if policy is not None: 

74 entry_extra["policy"] = policy 

75 if min_period_ms is not None: 

76 entry_extra["minPeriod"] = min_period_ms 

77 

78 path_list = paths if paths else ["*"] 

79 return { 

80 "context": context, 

81 "subscribe": [{"path": p, **entry_extra} for p in path_list], 

82 } 

83 

84 

85def iter_deltas(ws, count: int | None = None) -> Iterator[tuple[str, dict[str, Any]]]: 

86 """Yield (raw_text, parsed) pairs for delta messages from an open WebSocket connection. 

87 

88 Control messages without an "updates" key (e.g. the initial Hello message) 

89 are skipped. Stops after `count` deltas if given, or when the server 

90 closes the connection. 

91 """ 

92 yielded = 0 

93 while count is None or yielded < count: 

94 payload = ws.next_payload() 

95 if payload is None: 

96 return 

97 if isinstance(payload, bytes): 

98 payload = payload.decode("utf-8", errors="replace") 

99 try: 

100 message = json.loads(payload) 

101 except (TypeError, json.JSONDecodeError): 

102 continue 

103 if "updates" not in message: 

104 continue 

105 yield payload, message 

106 yielded += 1