diff --git a/firmware/esp32-csi-node/main/Kconfig.projbuild b/firmware/esp32-csi-node/main/Kconfig.projbuild index ee1dff164a..98a1c6dea0 100644 --- a/firmware/esp32-csi-node/main/Kconfig.projbuild +++ b/firmware/esp32-csi-node/main/Kconfig.projbuild @@ -487,10 +487,18 @@ menu "Onboard LED (ADR-183)" config LED_MOTION_FULLSCALE_MILLI int "Motion value (x1000) that saturates the colormap to yellow" depends on LED_GAMMA_VIZ - default 250 + default 10000 range 1 100000 help edge motion_energy that maps to the top (yellow) of the viridis - colormap, in milli-units (250 = 0.25). Lower = more sensitive + colormap, in milli-units (10000 = 10.0). Lower = more sensitive (reaches yellow with less motion). + + 10.0 matches the saturation point adaptive_controller.c's + collect_observation() uses to normalize this same raw + motion_energy/presence_score value into motion_score/presence_score + (RuView#1286) — yellow now means the same "fully saturated motion" + as motion_score == 1.0 elsewhere in the system, instead of an + unrelated 0.25 threshold that saturated to yellow almost + immediately given real-world motion_energy commonly runs 4-14. endmenu diff --git a/firmware/esp32-csi-node/main/adaptive_controller.c b/firmware/esp32-csi-node/main/adaptive_controller.c index 99272ee906..9164a112d1 100644 --- a/firmware/esp32-csi-node/main/adaptive_controller.c +++ b/firmware/esp32-csi-node/main/adaptive_controller.c @@ -116,18 +116,24 @@ static void collect_observation(adapt_observation_t *out) } } - /* Edge-derived state. The ADR-039 vitals packet exposes presence_score - * and motion_energy directly; we treat motion_energy as a proxy for - * motion_score by clamping to [0,1]. anomaly_score and node_coherence - * are not yet emitted by edge_processing — placeholder until Layer 4 - * extraction lands. */ + /* Edge-derived state. The ADR-039 vitals packet's presence_score and + * motion_energy are the same raw, unbounded phase-variance quantity + * (edge_processing.c sets s_presence_score = s_motion_energy directly), + * so both must go through the same scale-then-clamp normalization + * send_feature_vector() already uses (divide by 10, clamp to [0,1]). + * A bare [0,1] clamp on either field saturates at 1.0 almost + * unconditionally, since raw values routinely exceed 1.0 — this is + * what caused presence/motion to read "detected" regardless of actual + * occupancy. anomaly_score and node_coherence are not yet emitted by + * edge_processing — placeholder until Layer 4 extraction lands. */ edge_vitals_pkt_t vitals; if (edge_get_vitals(&vitals)) { - out->presence_score = vitals.presence_score; + float p = vitals.presence_score; + if (p < 0.0f) p = 0.0f; + out->presence_score = (p > 10.0f) ? 1.0f : (p / 10.0f); float m = vitals.motion_energy; if (m < 0.0f) m = 0.0f; - if (m > 1.0f) m = 1.0f; - out->motion_score = m; + out->motion_score = (m > 10.0f) ? 1.0f : (m / 10.0f); } out->anomaly_score = 0.0f; out->node_coherence = 1.0f; diff --git a/scripts/c6-presence-watcher.py b/scripts/c6-presence-watcher.py index f48a380711..b300f8bbce 100644 --- a/scripts/c6-presence-watcher.py +++ b/scripts/c6-presence-watcher.py @@ -317,53 +317,63 @@ def write_feature(gated: dict, motion: bool, occupancy: bool, # ADR-118 PrivacyGate: classify + redact before the # HAP boundary. Returns None for non-eligible classes. gated = apply_privacy_gate(pkt, privacy_class) - if gated is not None and gated["presence_valid"]: - n_valid += 1 - presence_sum += gated["presence"] - motion_sum += gated["motion"] - last_packet_ts = now - # MotionDetected — short-window (each packet) + if gated is not None: prev_motion = motion - if not motion and gated["presence"] >= PRESENCE_ON_THRESHOLD: - motion = set_motion(args.toggle, True, motion) - elif motion and gated["presence"] <= PRESENCE_OFF_THRESHOLD: - motion = set_motion(args.toggle, False, motion) - - # OccupancyDetected — rolling-window avg (§2.1.d - # "Unexpected Occupancy" is a future iter; for now - # we expose Occupancy as sustained presence). - occ_window.append((now, gated["presence"])) - cutoff = now - args.occupancy_window - while occ_window and occ_window[0][0] < cutoff: - occ_window.popleft() - if occ_window: - occ_avg = (sum(p for _, p in occ_window) - / len(occ_window)) - if not occupancy and occ_avg >= OCC_ON_THRESH: - occupancy = True + if gated["presence_valid"]: + n_valid += 1 + presence_sum += gated["presence"] + motion_sum += gated["motion"] + last_packet_ts = now + # MotionDetected — short-window (each packet) + if not motion and gated["presence"] >= PRESENCE_ON_THRESHOLD: + motion = set_motion(args.toggle, True, motion) + elif motion and gated["presence"] <= PRESENCE_OFF_THRESHOLD: + motion = set_motion(args.toggle, False, motion) + + # OccupancyDetected — rolling-window avg (§2.1.d + # "Unexpected Occupancy" is a future iter; for now + # we expose Occupancy as sustained presence). + occ_window.append((now, gated["presence"])) + cutoff = now - args.occupancy_window + while occ_window and occ_window[0][0] < cutoff: + occ_window.popleft() + if occ_window: + occ_avg = (sum(p for _, p in occ_window) + / len(occ_window)) + if not occupancy and occ_avg >= OCC_ON_THRESH: + occupancy = True + print(f"[{time.strftime('%H:%M:%S')}] " + f"Unknown Presence — Occupancy ON " + f"(rolling_avg={occ_avg:.2f})", + flush=True) + elif occupancy and occ_avg <= OCC_OFF_THRESH: + occupancy = False + print(f"[{time.strftime('%H:%M:%S')}] " + f"Occupancy OFF " + f"(rolling_avg={occ_avg:.2f})", + flush=True) + + # Anomaly — only when class allows (Restricted + # gate drops anomaly_score entirely; the dict + # missing the key is the type-level enforcement). + if ("anomaly" in gated + and gated["anomaly"] >= args.anomaly_threshold): + last_anomaly_ts = now + n_anomaly_fires += 1 print(f"[{time.strftime('%H:%M:%S')}] " - f"Unknown Presence — Occupancy ON " - f"(rolling_avg={occ_avg:.2f})", + f"Unrecognized Activity Pattern " + f"(anomaly={gated['anomaly']:.2f})", flush=True) - elif occupancy and occ_avg <= OCC_OFF_THRESH: - occupancy = False - print(f"[{time.strftime('%H:%M:%S')}] " - f"Occupancy OFF " - f"(rolling_avg={occ_avg:.2f})", - flush=True) - - # Anomaly — only when class allows (Restricted - # gate drops anomaly_score entirely; the dict - # missing the key is the type-level enforcement). - if ("anomaly" in gated - and gated["anomaly"] >= args.anomaly_threshold): - last_anomaly_ts = now - n_anomaly_fires += 1 - print(f"[{time.strftime('%H:%M:%S')}] " - f"Unrecognized Activity Pattern " - f"(anomaly={gated['anomaly']:.2f})", - flush=True) + # Write on every privacy-eligible packet, not just + # presence_valid ones — otherwise the feature file's + # `ts` only advances during "confident" readings, and + # goes stale (per ruview-sensing-server.py's 10s + # staleness gate) during perfectly normal stretches + # where the room is calm/empty and presence_score + # legitimately sits below the firmware's 0.5 quality + # threshold. That previously read as "watcher/device + # dead" to consumers when it was actually just quiet. if (motion != prev_motion or not state_path.endswith(".disabled")): write_state(motion, occupancy, last_anomaly_ts) diff --git a/scripts/ruview-sensing-server.py b/scripts/ruview-sensing-server.py index 0ebc0ef1a2..db96d882fe 100644 --- a/scripts/ruview-sensing-server.py +++ b/scripts/ruview-sensing-server.py @@ -19,6 +19,20 @@ GET /api/v1/edge/registry — node enumeration GET /api/v1/vitals//latest — EdgeVitalsMessage GET /api/v1/bfld//last_scan — BfldScanResponse + GET /api/v1/sensing/history/?hours=24&bucket_min=10 + — bucketed presence history + + exact state-change events, + for the ui/live-status.html + 24h chart. A background + thread samples the feature + file every HISTORY_SAMPLE_S + and appends to a per-node + JSONL file so history + survives server restarts. + There is no backfill: the + window only has real data + from whenever this sampler + was first started. POST /api/v1/bfld//subscribe?duration_s=N — { subscription_id } The source-of-truth file is `/tmp/ruview-last-feature.json` written @@ -35,7 +49,9 @@ import os import re import sys +import threading import time +from collections import deque from http.server import BaseHTTPRequestHandler, HTTPServer from urllib.parse import urlparse, parse_qs @@ -44,6 +60,134 @@ STALENESS_S = 10.0 DEFAULT_PORT = int(os.environ.get("PORT", "3000")) +# --- Presence history (for the 24h chart) ----------------------------------- +# The feature file only ever holds the latest sample, so a background thread +# samples it every HISTORY_SAMPLE_S and keeps a rolling HISTORY_WINDOW_S of +# (timestamp, presence) pairs in memory, persisted to a per-node JSONL file +# so history survives a server restart. No backfill — a freshly started +# sampler has no history before its own start time. +HISTORY_DIR = os.environ.get("RUVIEW_HISTORY_DIR", "/tmp") +HISTORY_SAMPLE_S = 30.0 +HISTORY_WINDOW_S = 24 * 3600.0 +_history_lock = threading.Lock() +_history: dict[str, deque] = {} + + +def _history_file(node_id: str) -> str: + safe = re.sub(r"[^A-Za-z0-9_-]", "_", node_id) + return os.path.join(HISTORY_DIR, f"ruview-history-{safe}.jsonl") + + +def _load_history_from_disk(node_id: str) -> deque: + dq: deque = deque() + cutoff = time.time() - HISTORY_WINDOW_S + try: + with open(_history_file(node_id), "r") as fh: + for line in fh: + try: + rec = json.loads(line) + except json.JSONDecodeError: + continue + ts = rec.get("ts") + if ts is not None and ts >= cutoff: + dq.append((ts, bool(rec.get("presence")))) + except OSError: + pass + return dq + + +def _history_get(node_id: str) -> deque: + with _history_lock: + if node_id not in _history: + _history[node_id] = _load_history_from_disk(node_id) + return _history[node_id] + + +def _append_history_disk(node_id: str, ts: float, presence: bool) -> None: + try: + with open(_history_file(node_id), "a") as fh: + fh.write(json.dumps({"ts": ts, "presence": presence}) + "\n") + except OSError as e: + print(f"[sensing-server] history write error: {e}", flush=True) + + +def _prune_history_disk(node_id: str) -> None: + """Rewrite the history file from the in-memory window, bounding growth.""" + with _history_lock: + records = list(_history.get(node_id, ())) + path = _history_file(node_id) + tmp = path + ".tmp" + try: + with open(tmp, "w") as fh: + for ts, presence in records: + fh.write(json.dumps({"ts": ts, "presence": presence}) + "\n") + os.replace(tmp, path) + except OSError as e: + print(f"[sensing-server] history prune error: {e}", flush=True) + + +def _history_sampler_loop() -> None: + prune_countdown = {} + while True: + time.sleep(HISTORY_SAMPLE_S) + f = _load_feature() + if f is None: + continue + node_id = f["node_id"] + presence = bool(f.get("presence", False)) + now = time.time() + dq = _history_get(node_id) + with _history_lock: + dq.append((now, presence)) + cutoff = now - HISTORY_WINDOW_S + while dq and dq[0][0] < cutoff: + dq.popleft() + _append_history_disk(node_id, now, presence) + prune_countdown[node_id] = prune_countdown.get(node_id, 0) + 1 + if prune_countdown[node_id] >= 200: # ~100 min at 30s cadence + _prune_history_disk(node_id) + prune_countdown[node_id] = 0 + + +def history_for(node_id: str, hours: float, bucket_min: float) -> dict: + now = time.time() + bucket_s = max(1.0, bucket_min * 60.0) + n_buckets = max(1, int((hours * 3600.0) // bucket_s)) + start = now - n_buckets * bucket_s + + dq = _history_get(node_id) + with _history_lock: + samples = sorted((ts, p) for ts, p in dq if ts >= start) + + changes = [] + prev = None + for ts, presence in samples: + if prev is not None and presence != prev: + changes.append({"ts": ts, "presence": presence}) + prev = presence + + buckets = [] + for i in range(n_buckets): + b_start = start + i * bucket_s + b_end = b_start + bucket_s + in_bucket = [p for ts, p in samples if b_start <= ts < b_end] + buckets.append({ + "ts_start": b_start, + "ts_end": b_end, + "has_data": bool(in_bucket), + "presence_frac": (sum(in_bucket) / len(in_bucket) + if in_bucket else None), + }) + + return { + "node_id": node_id, + "window_hours": hours, + "bucket_minutes": bucket_min, + "sample_interval_s": HISTORY_SAMPLE_S, + "buckets": buckets, + "changes": changes, + } + def _load_feature() -> dict | None: try: @@ -100,6 +244,7 @@ def bfld_scan_for(node_id: str) -> dict | None: _PATH_BFLD_SCAN = re.compile(r"^/api/v1/bfld/([^/]+)/last_scan$") _PATH_BFLD_SUBSCRIBE = re.compile(r"^/api/v1/bfld/([^/]+)/subscribe$") _PATH_SEMANTIC = re.compile(r"^/api/v1/semantic-events/([^/]+)/latest$") +_PATH_HISTORY = re.compile(r"^/api/v1/sensing/history/([^/]+)$") def semantic_events_for(node_id: str) -> dict | None: @@ -164,6 +309,7 @@ def _json(self, code: int, body: dict) -> None: self.send_response(code) self.send_header("Content-Type", "application/json") self.send_header("Content-Length", str(len(payload))) + self.send_header("Access-Control-Allow-Origin", "*") self.end_headers() self.wfile.write(payload) @@ -243,6 +389,15 @@ def do_GET(self) -> None: self._json(200, r) return + m = _PATH_HISTORY.match(path) + if m: + node_id = m.group(1) + qs = parse_qs(parsed.query) + hours = float(qs.get("hours", ["24"])[0]) + bucket_min = float(qs.get("bucket_min", ["10"])[0]) + self._json(200, history_for(node_id, hours, bucket_min)) + return + self._json(404, {"error": "not found", "path": path}) def do_POST(self) -> None: @@ -266,9 +421,12 @@ def do_POST(self) -> None: def main() -> int: port = DEFAULT_PORT server = HTTPServer(("0.0.0.0", port), Handler) + threading.Thread(target=_history_sampler_loop, daemon=True).start() print(f"[sensing-server] listening on 0.0.0.0:{port}", flush=True) print(f"[sensing-server] feature source: {FEATURE_FILE}", flush=True) print(f"[sensing-server] staleness limit: {STALENESS_S} s", flush=True) + print(f"[sensing-server] history sampler: every {HISTORY_SAMPLE_S}s, " + f"{HISTORY_WINDOW_S/3600:.0f}h window, dir={HISTORY_DIR}", flush=True) try: server.serve_forever() except KeyboardInterrupt: diff --git a/ui/live-status.html b/ui/live-status.html new file mode 100644 index 0000000000..fb07132aa1 --- /dev/null +++ b/ui/live-status.html @@ -0,0 +1,434 @@ + + + + +RuView — Live Status + + + +
+

RuView — Live Sensing Status

+ connecting… +
+ +
+ +
+ + + + +