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

1"""SignalK server discovery (mDNS) and the on-disk host cache.""" 

2 

3import time 

4from pathlib import Path 

5 

6try: 

7 from zeroconf import ServiceBrowser, ServiceStateChange, Zeroconf 

8except ImportError: # slim/Pyodide installs ship without zeroconf 

9 Zeroconf = None # type: ignore[assignment,misc] 

10 

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

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

13_HOST_CACHE_FILE = CACHE_DIR / "host.cache" 

14 

15 

16def get_cached_host() -> str | None: 

17 try: 

18 if _HOST_CACHE_FILE.exists(): 

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

20 except OSError: 

21 pass 

22 return None 

23 

24 

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

26 try: 

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

28 _HOST_CACHE_FILE.write_text(host) 

29 except OSError: 

30 pass 

31 

32 

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

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

35 if Zeroconf is None: 

36 return None 

37 found: list[str] = [] 

38 

39 def _on_change( 

40 zeroconf: Zeroconf, 

41 service_type: str, 

42 name: str, 

43 state_change: ServiceStateChange, 

44 ) -> None: 

45 if state_change is not ServiceStateChange.Added: 

46 return 

47 info = zeroconf.get_service_info(service_type, name) 

48 if info is None: 

49 return 

50 addrs = info.parsed_addresses() 

51 if not addrs: 

52 return 

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

54 found.append(host) 

55 

56 zc = Zeroconf() 

57 try: 

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

59 deadline = time.monotonic() + timeout 

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

61 time.sleep(0.1) 

62 finally: 

63 zc.close() 

64 

65 return found[0] if found else None 

66 

67 

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

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

70 if "://" not in host: 

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

72 return host