Coverage for src/signalk_cli/history/history_api.py: 65%
126 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-14 12:47 +0000
1"""SignalK v2 History API client."""
3import fnmatch
4import re
5from datetime import UTC, datetime, timedelta
6from pathlib import Path
8import click
9import niquests
11from ..net import (
12 CACHE_DIR,
13 api_error,
14 discover_host,
15 get_cached_host,
16 normalise_host,
17 save_cached_host,
18)
20__all__ = [
21 "CACHE_DIR",
22 "HISTORY_BASE",
23 "api_error",
24 "apply_time_default",
25 "discover_host",
26 "expand_paths",
27 "fetch_default_provider",
28 "fetch_server_paths",
29 "get_cached_host",
30 "get_cached_provider",
31 "normalise_duration",
32 "normalise_host",
33 "resolve_provider",
34 "save_cached_host",
35 "save_cached_provider",
36]
38HISTORY_BASE = "/signalk/v2/api/history"
40_DURATION_RE = re.compile(
41 r"^P(?:(\d+)Y)?(?:(\d+)M)?(?:(\d+)W)?(?:(\d+)D)?"
42 r"(?:T(?:(\d+)H)?(?:(\d+)M)?(?:(\d+(?:\.\d+)?)S)?)?$"
43)
46def _has_date_parts(duration: str) -> bool:
47 m = _DURATION_RE.match(duration)
48 return bool(m and any(m.group(i) for i in (1, 2, 3, 4)))
51def _duration_to_timedelta(duration: str) -> timedelta:
52 m = _DURATION_RE.match(duration)
53 if not m:
54 raise ValueError(f"Cannot parse duration: {duration!r}")
55 years, months, weeks, days, hours, minutes = (
56 int(m.group(i) or 0) for i in range(1, 7)
57 )
58 secs = float(m.group(7) or 0)
59 return timedelta(
60 days=years * 365 + months * 30 + weeks * 7 + days,
61 hours=hours,
62 minutes=minutes,
63 seconds=secs,
64 )
67def normalise_duration(
68 duration: str | None, from_: str | None, to: str | None
69) -> tuple[str | None, str | None, str | None]:
70 """Convert date-component durations to explicit from/to timestamps.
72 SignalK only accepts PT-prefix (time-only) durations. Durations containing
73 Y/M/W/D are expanded to from/to pairs:
74 from + duration → to = from + duration
75 to + duration → from = to - duration
76 duration alone → from = now - duration, to = now
78 Returns (from_, to, duration_or_None).
79 """
80 if not duration: 80 ↛ 81line 80 didn't jump to line 81 because the condition on line 80 was never true
81 return from_, to, duration
82 try:
83 int(duration)
84 return from_, to, duration # integer seconds, pass through
85 except ValueError:
86 pass
87 if not _has_date_parts(duration): 87 ↛ 90line 87 didn't jump to line 90 because the condition on line 87 was always true
88 return from_, to, duration # PT-only, pass through
90 delta = _duration_to_timedelta(duration)
91 fmt = "%Y-%m-%dT%H:%M:%SZ"
92 now = datetime.now(UTC)
94 if from_ is not None and to is not None:
95 return from_, to, None
96 elif from_ is not None:
97 from_dt = datetime.fromisoformat(from_)
98 return from_, (from_dt + delta).strftime(fmt), None
99 elif to is not None:
100 to_dt = datetime.fromisoformat(to)
101 return (to_dt - delta).strftime(fmt), to, None
102 else:
103 return (now - delta).strftime(fmt), now.strftime(fmt), None
106def apply_time_default(time_params: dict) -> dict:
107 """If neither 'from' nor 'duration' is set, default to the hour ending at 'to' (or now)."""
108 if "from" in time_params or "duration" in time_params:
109 return time_params
110 if "to" in time_params:
111 try:
112 to_dt = datetime.fromisoformat(time_params["to"])
113 except ValueError:
114 to_dt = datetime.now(UTC)
115 else:
116 to_dt = datetime.now(UTC)
117 from_dt = to_dt - timedelta(hours=1)
118 result = {
119 **time_params,
120 "from": from_dt.strftime("%Y-%m-%dT%H:%M:%SZ"),
121 "to": to_dt.strftime("%Y-%m-%dT%H:%M:%SZ"),
122 }
123 click.echo(
124 f"No time range specified — defaulting to from={result['from']} to={result['to']}",
125 err=True,
126 )
127 return result
130# ---------------------------------------------------------------------------
131# Provider cache (on-disk, per host)
132# ---------------------------------------------------------------------------
135def _cache_key(host: str) -> Path:
136 safe = re.sub(r"[^\w.-]", "_", host)
137 return CACHE_DIR / f"{safe}.provider"
140def get_cached_provider(host: str) -> str | None:
141 try:
142 f = _cache_key(host)
143 if f.exists():
144 return f.read_text().strip() or None
145 except OSError:
146 pass
147 return None
150def save_cached_provider(host: str, provider_id: str) -> None:
151 try:
152 CACHE_DIR.mkdir(parents=True, exist_ok=True)
153 _cache_key(host).write_text(provider_id)
154 except OSError:
155 pass
158def fetch_default_provider(base_url: str) -> str:
159 resp = niquests.get(f"{base_url}/_providers/_default", timeout=10)
160 resp.raise_for_status()
161 return resp.json()["id"]
164def resolve_provider(
165 host: str, base_url: str, provider: str | None, no_cache: bool
166) -> str | None:
167 """Return the effective provider id, fetching and caching the default if needed."""
168 if provider:
169 return provider
170 if not no_cache: 170 ↛ 171line 170 didn't jump to line 171 because the condition on line 170 was never true
171 provider = get_cached_provider(host)
172 if not provider: 172 ↛ 181line 172 didn't jump to line 181 because the condition on line 172 was always true
173 try:
174 provider = fetch_default_provider(base_url)
175 if not no_cache: 175 ↛ 176line 175 didn't jump to line 176 because the condition on line 175 was never true
176 save_cached_provider(host, provider)
177 except niquests.HTTPError as e:
178 click.echo(
179 f"Warning: could not fetch default provider: {api_error(e)}", err=True
180 )
181 return provider
184# ---------------------------------------------------------------------------
185# Path resolution
186# ---------------------------------------------------------------------------
189def fetch_server_paths(
190 base_url: str, time_params: dict, provider: str | None
191) -> list[str]:
192 """Fetch all paths that have data for the given time range."""
193 params = {k: v for k, v in time_params.items() if v is not None}
194 if provider:
195 params["provider"] = provider
196 resp = niquests.get(f"{base_url}/paths", params=params, timeout=30)
197 resp.raise_for_status()
198 return resp.json()
201def expand_paths(
202 patterns: list[str],
203 base_url: str,
204 time_params: dict,
205 provider: str | None,
206) -> list[str]:
207 """Expand path patterns to concrete paths.
209 Literal paths pass through unchanged. Patterns containing regex
210 metacharacters are matched against the server's /paths endpoint;
211 invalid regex is retried as a glob pattern.
212 """
213 regex_chars = set(r".*+?[](){}|^$\\")
214 literals: list[str] = []
215 regexps: list[str] = []
217 for p in patterns:
218 if ":" in p:
219 literals.append(p) # inline spec — pass through unchanged
220 elif any(c in p for c in regex_chars):
221 regexps.append(p)
222 else:
223 literals.append(p)
225 if not regexps:
226 return literals
228 click.echo("Resolving patterns against server paths...", err=True)
229 available = fetch_server_paths(base_url, apply_time_default(time_params), provider)
231 _GLOB_ONLY_RE = re.compile(
232 r"^[^.+(){}|^$\\]+$"
233 ) # only glob chars, no regex-specific
235 def _compile(pattern: str) -> re.Pattern:
236 if _GLOB_ONLY_RE.match(pattern):
237 return re.compile(fnmatch.translate(pattern))
238 try:
239 return re.compile(pattern)
240 except re.PatternError:
241 click.echo(
242 f"Note: '{pattern}' is not valid regex, treating as glob", err=True
243 )
244 return re.compile(fnmatch.translate(pattern))
246 matched: set[str] = set()
247 for pattern, rx in [(p, _compile(p)) for p in regexps]:
248 hits = {path for path in available if rx.search(path)}
249 if not hits:
250 click.echo(f"Warning: '{pattern}' matched no paths", err=True)
251 matched |= hits
253 return literals + sorted(matched)