Behavision/demo/console.py

"""A one-screen live demo of the whole Behavision chain, for showing someone.

Run it on the shop PC (here, this Mac) while the engine and agent are running.
It holds every credential itself and the browser holds none, so the page can be
put on a projector without putting a token on it.

What it shows, and why each part is there:

- the live camera, so the person walking past sees themselves;
- the CHAIN, measured rather than described: the engine recognised a face at
  this instant, the same visit appeared in the cloud API this many seconds
  later. That number is the product's claim, and it is computed here from two
  independent sources rather than asserted;
- the customer, editable - type a name, walk past again, watch the name come
  back through the cloud instead of "Visitor 5";
- the raw JSON a phone and a dashboard receive, side by side, because a
  colleague's real question is "is this actually wired up or is it a mock".

    .venv/bin/python demo/console.py            # http://127.0.0.1:8099
"""
from __future__ import annotations

import base64
import json
import os
import threading
import time
import urllib.error
import urllib.request
from pathlib import Path

ROOT = Path(__file__).resolve().parent.parent
STATE = Path(os.environ.get("BEHAVISION_DATA_DIR", ROOT / ".demo"))
CLOUD = os.environ.get("BEHAVISION_CLOUD", "https://mcp.loyaly.ai")
ENGINE = "http://127.0.0.1:8010"
PORT = int(os.environ.get("DEMO_PORT", "8099"))

# Whoever the demo signs in as. Staff on purpose: it is the weakest role that
# can do everything the shop floor does, so nothing here is only possible
# because we used an owner.
EMAIL = os.environ.get("DEMO_EMAIL", "staff.demo@tenext.in")
PASSWORD = os.environ.get("DEMO_PASSWORD", "admin@123")


def engine_auth() -> str:
    """The engine invents a Basic credential when none is configured, and
    writes it here. Read it rather than keeping a second copy."""
    f = STATE / "data" / "api_credentials.txt"
    if not f.exists():
        return ""
    user = pw = ""
    for line in f.read_text().splitlines():
        # `key=value`, and `key: value` too - the engine writes one and people
        # read the other, and which is which is not worth a support call.
        if "=" in line or ":" in line:
            k, v = line.split("=", 1) if "=" in line else line.split(":", 1)
            if k.strip().lower() == "username":
                user = v.strip()
            elif k.strip().lower() == "password":
                pw = v.strip()
    if not user:
        return ""
    return "Basic " + base64.b64encode(f"{user}:{pw}".encode()).decode()


def fetch(url: str, *, headers=None, body=None, method="GET", timeout=20):
    req = urllib.request.Request(url, method=method,
                                 data=json.dumps(body).encode() if body is not None else None,
                                 headers={k: v for k, v in (headers or {}).items() if v})
    if body is not None:
        req.add_header("content-type", "application/json")
    try:
        with urllib.request.urlopen(req, timeout=timeout) as r:
            raw = r.read()
            return r.status, (json.loads(raw) if raw and r.headers.get("content-type", "").startswith("application/json") else raw)
    except urllib.error.HTTPError as e:
        raw = e.read()
        try:
            return e.code, json.loads(raw or b"{}")
        except Exception:
            return e.code, raw[:400]
    except Exception as e:
        return 0, {"error": str(e)}


class Cloud:
    """The signed-in session, refreshed when it expires."""

    def __init__(self):
        self.token = ""
        self.lock = threading.Lock()

    def sign_in(self) -> bool:
        st, d = fetch(f"{CLOUD}/api/auth/login", method="POST",
                      body={"email": EMAIL, "password": PASSWORD, "device": "Demo console"})
        if st == 200 and isinstance(d, dict):
            self.token = d.get("access_token", "")
            return True
        return False

    def call(self, path, method="GET", body=None, retry=True):
        with self.lock:
            if not self.token and not self.sign_in():
                return 0, {"error": "cannot sign in to the platform"}
            tok = self.token
        st, d = fetch(f"{CLOUD}{path}", method=method, body=body,
                      headers={"authorization": f"Bearer {tok}"})
        if st == 401 and retry:
            with self.lock:
                self.sign_in()
            return self.call(path, method, body, retry=False)
        return st, d


cloud = Cloud()

# The chain, as the watcher builds it. One dict per recognition, newest first.
events: list[dict] = []
events_lock = threading.Lock()


def watch():
    """Poll the engine's own event log and the cloud feed, and join them.

    They are joined on the identity and the second, not on a shared id,
    because the engine numbers identities locally and the server numbers them
    per tenant - the two are deliberately different (see CLAUDE.md). What
    matters for the demo is the LATENCY between one seeing a person and the
    other, and that only needs the same person and the same moment.
    """
    seen_local: set[str] = set()
    while True:
        try:
            auth = engine_auth()
            st, d = fetch(f"{ENGINE}/api/events?limit=25", headers={"authorization": auth})
            # The engine returns a bare list; a dict with "events" is accepted
            # too so this survives either shape.
            evs = d if isinstance(d, list) else (d or {}).get("events", []) if isinstance(d, dict) else []
            if st == 200:
                for e in evs:
                    if e.get("type") not in ("person.new", "person.seen"):
                        continue
                    key = f"{e.get('ts')}|{e.get('camera_id')}|{(e.get('data') or {}).get('identity_id')}"
                    if key in seen_local:
                        continue
                    seen_local.add(key)
                    data = e.get("data") or {}
                    with events_lock:
                        events.insert(0, {
                            "key": key,
                            "at": time.time(),
                            "kind": "new" if e["type"] == "person.new" else "seen",
                            "engine": {
                                "label": data.get("label"),
                                "identity_id": data.get("identity_id"),
                                "similarity": data.get("similarity"),
                                "quality": data.get("quality"),
                                "gender": data.get("gender"),
                                "age": data.get("age"),
                                "camera": e.get("camera_id"),
                                "ts": e.get("ts"),
                            },
                            "cloud": None,
                            "latency": None,
                        })
                        del events[40:]
        except Exception:
            pass

        # the other half: has the cloud got it yet?
        try:
            with events_lock:
                pending = [e for e in events if e["cloud"] is None][:6]
            if pending:
                st, d = cloud.call("/api/visits?limit=12")
                arrivals = (d or {}).get("arrivals", []) if isinstance(d, dict) else []
                for e in pending:
                    for a in arrivals:
                        # same camera, and the cloud's visit is not older than
                        # the engine's sighting
                        if a.get("camera_id") != e["engine"]["camera"]:
                            continue
                        if a.get("visit_id") in [x["cloud"].get("visit_id") for x in events if x["cloud"]]:
                            continue
                        with events_lock:
                            e["cloud"] = {
                                "visit_id": a.get("visit_id"),
                                "visitor_id": a.get("visitor_id"),
                                "label": a.get("label"),
                                "ref": a.get("customer_ref") or a.get("ref"),
                                "is_new": a.get("is_new_visitor"),
                                "similarity": a.get("similarity"),
                                "site": a.get("site"),
                                "occurred_at": a.get("occurred_at"),
                                "image": a.get("image"),
                            }
                            e["latency"] = round(time.time() - e["at"], 1)
                        break
        except Exception:
            pass
        time.sleep(1.0)


def snapshot() -> dict:
    """Everything the page draws, in one reply."""
    auth = engine_auth()
    _, health = fetch(f"{ENGINE}/api/health", headers={"authorization": auth})
    _, stats = fetch(f"{ENGINE}/api/stats", headers={"authorization": auth})
    st_v, visits = cloud.call("/api/visits?limit=3")
    st_f, foot = cloud.call("/api/reports/footfall?from=%s&to=%s"
                            % (time.strftime("%Y-%m-%d", time.localtime(time.time() - 7 * 86400)),
                               time.strftime("%Y-%m-%d")))
    with events_lock:
        chain = json.loads(json.dumps(events[:8]))
    cams = (stats or {}).get("cameras", []) if isinstance(stats, dict) else []
    return {
        "engine": {
            "up": isinstance(health, dict) and bool(health.get("status")),
            "model": (health or {}).get("recognition_model") if isinstance(health, dict) else None,
            "cameras": [{"id": c.get("camera_id"), "connected": c.get("connected"),
                         "frames": c.get("frames_processed") or c.get("frames_total"),
                         "faces": c.get("faces_seen"),
                         "tracks": c.get("active_tracks")} for c in cams],
        },
        "cloud_ok": st_v == 200,
        "chain": chain,
        "mobile": {"request": "GET /api/visits?limit=3", "status": st_v, "body": visits},
        "dashboard": {"request": "GET /api/reports/footfall?from=…&to=…", "status": st_f, "body": foot},
    }


def main():
    from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
    from urllib.parse import urlparse, parse_qs

    page = (Path(__file__).parent / "console.html").read_bytes()

    class H(BaseHTTPRequestHandler):
        def log_message(self, *a):  # quiet
            pass

        def _send(self, code, body, ctype="application/json"):
            self.send_response(code)
            self.send_header("content-type", ctype)
            self.send_header("content-length", str(len(body)))
            self.end_headers()
            try:
                self.wfile.write(body)
            except (BrokenPipeError, ConnectionResetError):
                pass

        def do_GET(self):
            u = urlparse(self.path)
            if u.path == "/":
                return self._send(200, page, "text/html; charset=utf-8")
            if u.path == "/api/snapshot":
                return self._send(200, json.dumps(snapshot()).encode())
            if u.path == "/api/customer":
                vid = parse_qs(u.query).get("id", [""])[0]
                if not vid:
                    return self._send(400, b'{"error":"no id"}')
                st, d = cloud.call(f"/api/visitors/{vid}/history?limit=8")
                return self._send(200, json.dumps({"status": st, "history": d}).encode())
            if u.path == "/camera.jpg":
                # A polled still, not the MJPEG stream. The engine re-serves its
                # latest frame until the pipeline produces a new one, so polling
                # shows the same picture - and a multipart stream through a
                # proxy is one more thing to fail in front of an audience.
                cam = parse_qs(u.query).get("id", [""])[0]
                st, body = fetch(f"{ENGINE}/api/cameras/{cam}/frame.jpg",
                                 headers={"authorization": engine_auth()}, timeout=15)
                if st != 200 or not isinstance(body, (bytes, bytearray)):
                    return self._send(502, b'{"error":"no frame"}')
                self.send_response(200)
                self.send_header("content-type", "image/jpeg")
                self.send_header("cache-control", "no-store")
                self.send_header("content-length", str(len(body)))
                self.end_headers()
                try:
                    self.wfile.write(body)
                except (BrokenPipeError, ConnectionResetError):
                    pass
                return
            self._send(404, b'{"error":"no"}')

        def do_POST(self):
            u = urlparse(self.path)
            n = int(self.headers.get("content-length", 0))
            body = json.loads(self.rfile.read(n) or b"{}")
            if u.path == "/api/profile":
                vid = body.pop("id", "")
                st, d = cloud.call(f"/api/visitors/{vid}/profile", method="PUT", body=body)
                return self._send(200, json.dumps({"status": st, "body": d}).encode())
            self._send(404, b'{"error":"no"}')

        def _proxy_stream(self, url):
            """The engine's MJPEG, relayed so the browser needs no credential.

            The engine's API is Basic-authenticated with a credential it
            generated locally; putting that in a page would hand the whole
            biometric API to anyone who opened it.
            """
            try:
                req = urllib.request.Request(url, headers={"authorization": engine_auth()})
                up = urllib.request.urlopen(req, timeout=20)
            except Exception:
                return self._send(502, b'{"error":"camera not available"}')
            self.send_response(200)
            self.send_header("content-type", up.headers.get("content-type", "multipart/x-mixed-replace"))
            self.end_headers()
            try:
                while True:
                    chunk = up.read(8192)
                    if not chunk:
                        break
                    self.wfile.write(chunk)
                    self.wfile.flush()
            except Exception:
                pass
            finally:
                up.close()

    threading.Thread(target=watch, daemon=True).start()
    print(f"\n  Demo console  →  http://127.0.0.1:{PORT}\n")
    print(f"  engine {ENGINE} · cloud {CLOUD} · signed in as {EMAIL}\n")
    ThreadingHTTPServer(("127.0.0.1", PORT), H).serve_forever()


if __name__ == "__main__":
    main()