Coverage for lunar_toolkit/cli/_status.py: 83%
225 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-19 02:58 +0000
1"""What ``lunar serve`` reports in ``/api/status``: indexes, caches, reference data, runs, tools."""
3from __future__ import annotations
5import json
6import os
7import shutil
8import threading
9import time
10from collections.abc import Iterator
11from dataclasses import dataclass
12from pathlib import Path
13from typing import Any
15MAX_SCAN = 20000
16MAX_LISTED = 200
17CACHE_SECONDS = 30.0
18_PATH_KEYS = {"path", "root", "dest", "executable"}
21@dataclass(frozen=True)
22class StatusDirs:
23 """The directories the status page describes."""
25 data_dir: Path
26 cache_dir: Path
27 download_dir: Path
28 reference_dir: Path
29 runs_dir: Path | None
32def _walk(root: Path, limit: int, state: dict[str, bool] | None = None) -> Iterator[os.DirEntry]:
33 """Files under *root* (no symlinks followed), stopping after *limit* entries (``state["truncated"]``)."""
34 stack = [root]
35 n = 0
36 while stack:
37 try:
38 it = os.scandir(stack.pop())
39 except OSError:
40 continue
41 with it:
42 for entry in it:
43 n += 1
44 if n > limit: 44 ↛ 45line 44 didn't jump to line 45 because the condition on line 44 was never true
45 if state is not None:
46 state["truncated"] = True
47 return
48 try:
49 if entry.is_dir(follow_symlinks=False):
50 stack.append(Path(entry.path))
51 elif entry.is_file(follow_symlinks=False): 51 ↛ 42line 51 didn't jump to line 42 because the condition on line 51 was always true
52 yield entry
53 except OSError:
54 continue
57def dir_size(root: Path, limit: int = MAX_SCAN) -> dict[str, Any]:
58 """``size`` (bytes) and ``files`` under *root*, ``truncated`` when more than *limit* entries."""
59 size = files = 0
60 state = {"truncated": False}
61 for entry in _walk(Path(root), limit, state):
62 try:
63 size += entry.stat(follow_symlinks=False).st_size
64 except OSError:
65 continue
66 files += 1
67 return {"size": size, "files": files, "truncated": state["truncated"]}
70def _file(path: Path, name: str | None = None) -> dict[str, Any]:
71 try:
72 st = path.stat()
73 except OSError:
74 return {"name": name or path.name, "path": str(path), "size": None, "mtime": None}
75 return {"name": name or path.name, "path": str(path), "size": st.st_size, "mtime": st.st_mtime}
78def indexes(data_dir: Path) -> list[dict[str, Any]]:
79 """Each dataset's Parquet: existence, size and the build sidecar (rows, release, times)."""
80 from ..constants import DATASETS
82 out = []
83 for ds, entry in DATASETS.items():
84 p = Path(data_dir) / entry["parquet"]
85 meta: dict[str, Any] = {}
86 try:
87 meta = json.loads(p.with_name(p.stem + ".meta.json").read_text())
88 except (OSError, ValueError):
89 pass
90 row = {"dataset": ds, "label": entry["label"], "exists": p.is_file(), **_file(p)}
91 row.update({k: meta.get(k) for k in ("rows", "release", "source", "built_at", "updated_at")})
92 row["volumes"] = len(meta.get("volumes") or {})
93 out.append(row)
94 return out
97def cache_summary(cache_dir: Path) -> dict[str, Any]:
98 """Top-level entries of the cache dir with their sizes."""
99 items = []
100 try:
101 entries = sorted(Path(cache_dir).iterdir())
102 except OSError:
103 entries = []
104 for p in entries[:MAX_LISTED]:
105 if p.is_dir(): 105 ↛ 108line 105 didn't jump to line 108 because the condition on line 105 was always true
106 items.append({"name": p.name + "/", "path": str(p), "dir": True, **dir_size(p)})
107 else:
108 items.append({**_file(p, _cache_name(p.name)), "dir": False, "files": 1})
109 total = sum(i["size"] or 0 for i in items)
110 return {"path": str(cache_dir), "exists": Path(cache_dir).is_dir(), "size": total, "items": items}
113def _cache_name(name: str) -> str:
114 """``226215a6_CUMINDEX.TAB`` -> ``CUMINDEX.TAB (226215a6)`` (PdsDownloader cache names)."""
115 head, sep, rest = name.partition("_")
116 if sep and len(head) == 8 and all(c in "0123456789abcdef" for c in head):
117 return f"{rest} ({head})"
118 return name
121def downloads(download_dir: Path) -> dict[str, Any]:
122 """Image files in the download dir (newest first), with ``.part`` files flagged as partial."""
123 files = []
124 for entry in _walk(Path(download_dir), MAX_SCAN):
125 p = Path(entry.path)
126 f = _file(p, p.relative_to(download_dir).as_posix())
127 f["partial"] = p.suffix == ".part"
128 files.append(f)
129 files.sort(key=lambda f: -(f["mtime"] or 0))
130 return {
131 "path": str(download_dir),
132 "exists": Path(download_dir).is_dir(),
133 "size": sum(f["size"] or 0 for f in files),
134 "count": len(files),
135 "files": files[:MAX_LISTED],
136 }
139def reference(reference_dir: Path) -> dict[str, Any]:
140 """PGDA sites / kinds and extracted Zenodo SDEM members under the reference cache."""
141 from .. import reference as ref
143 site_region = {r.pgda_site: f"{r.id} {r.name}" for r in ref.REGIONS if r.pgda_site}
144 pgda = []
145 root = Path(reference_dir) / "pgda"
146 for site_dir in sorted(p for p in root.glob("*") if p.is_dir()) if root.is_dir() else []:
147 files = []
148 for entry in _walk(site_dir, MAX_SCAN):
149 p = Path(entry.path)
150 kind = ref._classify_pgda(p.relative_to(root).as_posix())
151 files.append({**_file(p, p.relative_to(site_dir).as_posix()), "kind": kind})
152 kinds: dict[str, int] = {}
153 for f in files:
154 if f["kind"]: 154 ↛ 153line 154 didn't jump to line 153 because the condition on line 154 was always true
155 kinds[f["kind"]] = kinds.get(f["kind"], 0) + 1
156 pgda.append(
157 {
158 "site": site_dir.name,
159 "region": site_region.get(site_dir.name),
160 "kinds": kinds,
161 "size": sum(f["size"] or 0 for f in files),
162 "files": sorted(files, key=lambda f: f["name"])[:MAX_LISTED],
163 }
164 )
165 sdem = []
166 sroot = Path(reference_dir) / "sdem"
167 for zdir in sorted(p for p in sroot.glob("*") if p.is_dir()) if sroot.is_dir() else []: 167 ↛ 168line 167 didn't jump to line 168 because the loop on line 167 never started
168 members = [Path(e.path) for e in _walk(zdir, MAX_SCAN) if not e.name.endswith(".part")]
169 sdem.append(
170 {
171 "zip": zdir.name,
172 "path": str(zdir),
173 "members": sorted(p.relative_to(zdir).as_posix() for p in members)[:MAX_LISTED],
174 "count": len(members),
175 "size": sum(p.stat().st_size for p in members if p.exists()),
176 }
177 )
178 return {"path": str(reference_dir), "exists": Path(reference_dir).is_dir(), "pgda": pgda, "sdem": sdem}
181def runs(runs_dir: Path | None) -> dict[str, Any]:
182 """Run dirs with a ``provenance.json``: last run's status, times and steps, and a report flag."""
183 if runs_dir is None: 183 ↛ 184line 183 didn't jump to line 184 because the condition on line 183 was never true
184 return {"configured": False, "runs": []}
185 out: list[dict[str, Any]] = []
186 try:
187 dirs = sorted(p for p in Path(runs_dir).iterdir() if p.is_dir() and not p.name.startswith("."))
188 except OSError:
189 dirs = []
190 for d in dirs[:MAX_LISTED]:
191 prov = d / "provenance.json"
192 if not prov.is_file(): 192 ↛ 193line 192 didn't jump to line 193 because the condition on line 192 was never true
193 continue
194 try:
195 data = json.loads(prov.read_text())
196 except (OSError, ValueError):
197 out.append({"name": d.name, "path": str(d), "status": "unreadable", "steps": []})
198 continue
199 history = data.get("runs") or [{}]
200 last = history[-1]
201 steps = [
202 {"name": s.get("name"), "status": s.get("status"), "seconds": s.get("seconds")}
203 for s in last.get("steps") or []
204 ]
205 out.append(
206 {
207 "name": d.name,
208 "path": str(d),
209 "status": last.get("status"),
210 "started": last.get("started"),
211 "finished": last.get("finished"),
212 "runs": len(history),
213 "steps": steps,
214 "report": (d / "report" / "report.html").is_file(),
215 }
216 )
217 out.sort(key=lambda r: r.get("started") or "", reverse=True)
218 return {"configured": True, "path": str(runs_dir), "exists": Path(runs_dir).is_dir(), "runs": out}
221_ENV_LOCK = threading.Lock()
222_ENV: dict[str, Any] | None = None
225def environment(refresh: bool = False) -> dict[str, Any]:
226 """ASP / ISIS roots, ISISDATA subsets and tool versions (``sfs --version`` etc., cached)."""
227 global _ENV
228 with _ENV_LOCK:
229 if _ENV is None or refresh:
230 _ENV = _environment()
231 return _ENV
234def _environment() -> dict[str, Any]:
235 from ..provenance import isis_paths, kernel_files, tool_versions
236 from ..tools import find_tool
238 paths = isis_paths()
239 isisdata = paths.get("ISISDATA")
240 subsets: dict[str, bool] = {}
241 if isisdata: 241 ↛ 242line 241 didn't jump to line 242 because the condition on line 241 was never true
242 root = Path(isisdata)
243 for key, rel in (
244 ("base kernels", "base/kernels"),
245 ("base dems", "base/dems"),
246 ("lro kernels", "lro/kernels"),
247 ("lro calibration", "lro/calibration"),
248 ):
249 subsets[key] = (root / rel).is_dir()
250 tools = {}
251 for name in (
252 "sfs",
253 "parallel_sfs",
254 "bundle_adjust",
255 "mapproject",
256 "spiceinit",
257 "lronac2isis",
258 "gdalinfo",
259 ):
260 p = find_tool(name)
261 tools[name] = {"found": p is not None, "path": str(p) if p else None}
262 asp_root = os.environ.get("ASP_ROOT")
263 return {
264 "roots": [
265 {"name": "ASP_ROOT", "path": asp_root, "set": bool(asp_root)},
266 {"name": "ISISROOT", "path": paths.get("ISISROOT"), "set": bool(paths.get("ISISROOT"))},
267 {"name": "ISISDATA", "path": isisdata, "set": bool(isisdata)},
268 ],
269 "isisdata": subsets,
270 "kernels": kernel_files(isisdata),
271 "versions": tool_versions(),
272 "tools": tools,
273 "checked": time.time(),
274 }
277def cooldowns() -> list[dict[str, Any]]:
278 """Hosts paused after a throttling reply (persisted and in this process)."""
279 from .. import downloader
281 now = time.time()
282 hosts = dict(downloader._saved_cooldowns())
283 for name, h in list(downloader._HOSTS.items()):
284 if h.cooldown_until > now: 284 ↛ 283line 284 didn't jump to line 283 because the condition on line 284 was always true
285 hosts[name] = max(hosts.get(name, 0.0), h.cooldown_until)
286 return [
287 {
288 "host": h,
289 "until": t,
290 "until_text": time.strftime("%H:%M:%S", time.localtime(t)),
291 "seconds": t - now,
292 }
293 for h, t in sorted(hosts.items())
294 if t > now
295 ]
298def disks(dirs: StatusDirs) -> list[dict[str, Any]]:
299 """Free / total space of the file systems holding the data, cache and runs dirs."""
300 out: list[dict[str, Any]] = []
301 seen: set[int] = set()
302 for label, d in (("data", dirs.data_dir), ("cache", dirs.cache_dir), ("runs", dirs.runs_dir)):
303 if d is None: 303 ↛ 304line 303 didn't jump to line 304 because the condition on line 303 was never true
304 continue
305 p = Path(d)
306 while not p.exists() and p != p.parent: 306 ↛ 307line 306 didn't jump to line 307 because the condition on line 306 was never true
307 p = p.parent
308 try:
309 dev = p.stat().st_dev
310 usage = shutil.disk_usage(p)
311 except OSError:
312 continue
313 if dev in seen:
314 next(o for o in out if o["dev"] == dev)["labels"].append(label)
315 continue
316 seen.add(dev)
317 out.append({"labels": [label], "dev": dev, "path": str(p), "free": usage.free, "total": usage.total})
318 for o in out:
319 del o["dev"]
320 return out
323def redact(obj: Any) -> Any:
324 """*obj* without file system paths (``path`` / ``root`` / ... keys), for read-only clients."""
325 if isinstance(obj, dict):
326 return {k: redact(v) for k, v in obj.items() if k not in _PATH_KEYS and k != "kernels"}
327 if isinstance(obj, list):
328 return [redact(v) for v in obj]
329 return obj
332class StatusCache:
333 """``/api/status`` documents, rebuilt at most every `CACHE_SECONDS` (or on refresh)."""
335 def __init__(self, dirs: StatusDirs) -> None:
336 self.dirs = dirs
337 self._lock = threading.Lock()
338 self._doc: dict[str, Any] | None = None
339 self._at = 0.0
341 def invalidate(self) -> None:
342 with self._lock:
343 self._doc = None
345 def get(self, admin: bool, refresh: bool = False) -> dict[str, Any]:
346 with self._lock:
347 if self._doc is None or refresh or time.time() - self._at > CACHE_SECONDS: 347 ↛ 350line 347 didn't jump to line 350 because the condition on line 347 was always true
348 self._doc = self._build(refresh)
349 self._at = time.time()
350 doc = self._doc
351 live = {**doc, "cooldowns": cooldowns(), "admin": admin}
352 return live if admin else redact(live)
354 def _build(self, refresh: bool) -> dict[str, Any]:
355 from ..constants import APP_VERSION
357 d = self.dirs
358 return {
359 "version": APP_VERSION,
360 "data": {"path": str(d.data_dir), "exists": d.data_dir.is_dir(), "indexes": indexes(d.data_dir)},
361 "cache": cache_summary(d.cache_dir),
362 "downloads": downloads(d.download_dir),
363 "reference": reference(d.reference_dir),
364 "runs": runs(d.runs_dir),
365 "environment": environment(refresh),
366 "disks": disks(d),
367 "built": time.time(),
368 }