Coverage for src/signalk_cli/net.py: 38%

1"""Shared host discovery/caching and CLI option helpers for SignalK API clients.""" 

2 

3import contextlib 

4import io 

5import time 

6from pathlib import Path 

7 

8import click 

9import niquests 

10from zeroconf import ServiceBrowser, ServiceStateChange, Zeroconf 

11 

12CACHE_DIR = Path.home() / ".cache" / "signalk-cli" 

13_SIGNALK_TYPE = "_signalk-ws._tcp.local." 

14_HOST_CACHE_FILE = CACHE_DIR / "host.cache" 

15 

16 

17def get_cached_host() -> str | None: 

18 try: 

19 if _HOST_CACHE_FILE.exists(): 

20 return _HOST_CACHE_FILE.read_text().strip() or None 

21 except OSError: 

22 pass 

23 return None 

24 

25 

26def save_cached_host(host: str) -> None: 

27 try: 

28 CACHE_DIR.mkdir(parents=True, exist_ok=True) 

29 _HOST_CACHE_FILE.write_text(host) 

30 except OSError: 

31 pass 

32 

33 

34def discover_host(timeout: float = 5.0) -> str | None: 

35 """Browse mDNS for a SignalK server and return its base URL, or None.""" 

36 found: list[str] = [] 

37 

38 def _on_change( 

39 zeroconf: Zeroconf, 

40 service_type: str, 

41 name: str, 

42 state_change: ServiceStateChange, 

43 ) -> None: 

44 if state_change is not ServiceStateChange.Added: 

45 return 

46 info = zeroconf.get_service_info(service_type, name) 

47 if info is None: 

48 return 

49 addrs = info.parsed_addresses() 

50 if not addrs: 

51 return 

52 host = f"http://{addrs[0]}:{info.port}" 

53 found.append(host) 

54 

55 zc = Zeroconf() 

56 try: 

57 ServiceBrowser(zc, _SIGNALK_TYPE, handlers=[_on_change]) 

58 deadline = time.monotonic() + timeout 

59 while not found and time.monotonic() < deadline: 

60 time.sleep(0.1) 

61 finally: 

62 zc.close() 

63 

64 return found[0] if found else None 

65 

66 

67def normalise_host(host: str) -> str: 

68 """Prepend http:// if the host has no scheme.""" 

69 if "://" not in host: 

70 return f"http://{host}" 

71 return host 

72 

73 

74def api_error(exc: niquests.RequestException) -> str: 

75 """Return the most informative message from an API error response.""" 

76 resp = getattr(exc, "response", None) 

77 if resp is not None: 

78 with contextlib.suppress(Exception): 

79 body = resp.json() 

80 return body.get("error") or body.get("message") or str(exc) 

81 return str(exc) 

82 

83 

84# --------------------------------------------------------------------------- 

85# Shared Click option decorators 

86# --------------------------------------------------------------------------- 

87 

88 

89def host_option(f): 

90 return click.option( 

91 "--host", 

92 default=None, 

93 envvar="SIGNALK_HOST", 

94 help="SignalK server base URL. http:// added if scheme omitted. " 

95 "Discovered via mDNS if omitted.", 

96 )(f) 

97 

98 

99def bare_option(f): 

100 return click.option( 

101 "--bare", 

102 is_flag=True, 

103 help="Suppress all informational messages, outputting data only.", 

104 )(f) 

105 

106 

107def stderr_ctx(bare: bool) -> contextlib.AbstractContextManager: 

108 return ( 

109 contextlib.redirect_stderr(io.StringIO()) if bare else contextlib.nullcontext() 

110 ) 

111 

112 

113def resolve_host(host: str | None, no_cache: bool = False) -> str: 

114 """Return a normalised host URL, discovering via mDNS if none provided.""" 

115 if host: 115 ↛ 117line 115 didn't jump to line 117 because the condition on line 115 was always true

116 return normalise_host(host) 

117 if not no_cache: 

118 cached = get_cached_host() 

119 if cached: 

120 click.echo(f"Using cached host: {cached}", err=True) 

121 return cached 

122 click.echo("No host specified — searching for SignalK via mDNS...", err=True) 

123 discovered = discover_host() 

124 if not discovered: 

125 raise click.UsageError( 

126 "No SignalK server found via mDNS. Use --host or set SIGNALK_HOST." 

127 ) 

128 click.echo(f"Discovered: {discovered}", err=True) 

129 if not no_cache: 

130 save_cached_host(discovered) 

131 return discovered