| """The coop phone's own health: is it alive, charged, connected, and can it see.""" | |
| from __future__ import annotations | |
| from datetime import datetime | |
| from google.cloud import firestore | |
| from . import config, store | |
| APP_VERSION = "0.3" | |
| def record_beat(coop_id: str, sensors: dict) -> None: | |
| now = store.now() | |
| update = {"device.last_beat": now} | |
| for key in ("battery", "charging", "network", "sharpness", "brightness", "width", "height", | |
| "version", "camera", "mic", "temperature_c", "humidity_pct", "ammonia_ppm"): | |
| if sensors.get(key) is not None: | |
| update[f"device.{key}"] = sensors[key] | |
| store.coop_ref(coop_id).update(update) | |
| def record_frame(coop_id: str, ok: bool, error: str = "") -> None: | |
| update: dict = {"device.last_frame": store.now(), "device.frames": firestore.Increment(1)} | |
| if ok: | |
| update["device.last_vision_ok"] = store.now() | |
| else: | |
| update["device.vision_errors"] = firestore.Increment(1) | |
| update["device.last_error"] = error[:160] | |
| store.coop_ref(coop_id).update(update) | |
| def _ago_s(ts: datetime | None) -> float | None: | |
| return (store.now() - ts).total_seconds() if ts else None | |
| def health(coop: dict) -> dict: | |
| d = coop.get("device") or {} | |
| beat = _ago_s(d.get("last_beat")) | |
| frame = _ago_s(d.get("last_frame")) | |
| checks = [] | |
| def check(name: str, level: str, text: str) -> None: | |
| checks.append({"name": name, "level": level, "text": text}) | |
| if beat is None: | |
| check("link", "down", "No coop phone has reported yet.") | |
| elif beat > config.NODE_SILENT_AFTER_S: | |
| check("link", "down", f"Silent for {int(beat // 60)} minutes.") | |
| else: | |
| net = d.get("network") | |
| check("link", "ok", f"Online{f' on {net}' if net else ''}, last heard {int(beat)} s ago.") | |
| batt, charging = d.get("battery"), d.get("charging") | |
| if batt is not None: | |
| pct = round(batt * 100) | |
| if charging: | |
| check("power", "ok", f"{pct}%, charging.") | |
| elif pct < 20: | |
| check("power", "down", f"{pct}% and not charging.") | |
| elif pct < 50: | |
| check("power", "warn", f"{pct}% and not charging.") | |
| else: | |
| check("power", "ok", f"{pct}%, on battery.") | |
| bright, sharp = d.get("brightness"), d.get("sharpness") | |
| if d.get("camera") == "off": | |
| check("camera", "down", "Camera is off.") | |
| elif bright is not None and bright < 18: | |
| check("camera", "warn", "Too dark to see. Add a light or move the phone.") | |
| elif sharp is not None and sharp < 4: | |
| check("camera", "warn", "View looks blurred or blocked. Wipe the lens.") | |
| elif frame is not None and frame > 600 and beat is not None and beat < 120: | |
| check("camera", "warn", f"No picture for {int(frame // 60)} minutes.") | |
| elif bright is not None: | |
| check("camera", "ok", "Clear view.") | |
| t, rh, nh3 = d.get("temperature_c"), d.get("humidity_pct"), d.get("ammonia_ppm") | |
| if t is not None or rh is not None or nh3 is not None: | |
| parts = [f"{t:.1f} °C" if t is not None else None, f"{rh:.0f}% humidity" if rh is not None else None, | |
| f"{nh3:.0f} ppm ammonia" if nh3 is not None else None] | |
| level = "warn" if (nh3 is not None and nh3 >= 25) else "ok" | |
| check("air", level, ", ".join(p for p in parts if p) + (". Ammonia above 25 ppm; ventilate." if level == "warn" else ".")) | |
| errs = d.get("vision_errors", 0) | |
| ok_at = _ago_s(d.get("last_vision_ok")) | |
| if frame is not None and (ok_at is None or ok_at > 900) and errs: | |
| check("vision", "warn", f"Pictures arrive but reading them failed {errs} times.") | |
| elif ok_at is not None: | |
| check("vision", "ok", f"Last reading {int(ok_at // 60)} min ago.") | |
| order = {"down": 2, "warn": 1, "ok": 0} | |
| level = max((c["level"] for c in checks), key=lambda l: order[l], default="down") | |
| summary = { | |
| "down": next((c["text"] for c in checks if c["level"] == "down"), "Offline."), | |
| "warn": next((c["text"] for c in checks if c["level"] == "warn"), ""), | |
| "ok": "Watching", | |
| }[level] | |
| return { | |
| "level": level, "summary": summary, "checks": checks, | |
| "battery": round(batt * 100) if batt is not None else None, "charging": charging, | |
| "network": d.get("network"), "frames": d.get("frames", 0), | |
| "resolution": f"{d['width']}×{d['height']}" if d.get("width") else None, | |
| "version": d.get("version"), | |
| } | |
| def spoken(h: dict) -> str: | |
| return "Coop phone: " + " ".join(f"{c['name']}: {c['text']}" for c in h["checks"]) | |