feat: publish Alta Local Camera Discovery

app.py

@@ -0,0 +1,135 @@
+#!/usr/bin/env python3
+import argparse
+import json
+import re
+import threading
+import time
+from http import HTTPStatus
+from http.server import SimpleHTTPRequestHandler, ThreadingHTTPServer
+from pathlib import Path
+
+from zeroconf import ServiceBrowser, ServiceInfo, ServiceListener, Zeroconf
+
+ROOT = Path(__file__).resolve().parent
+SERVICE_TYPE = "_vaioncam._tcp.local."
+KEY_PATTERN = re.compile(r"^[A-Z0-9]{4}(?:-[A-Z0-9]{4}){2}$")
+
+
+def camera_from_service(name: str, hostname: str, addresses: list[str], port: int):
+    suffix = "." + SERVICE_TYPE
+    if not name.lower().endswith(suffix.lower()):
+        return None
+    alta_key = name[: -len(suffix)].upper()
+    if not KEY_PATTERN.fullmatch(alta_key):
+        return None
+    ipv4 = next((address for address in addresses if ":" not in address), None)
+    hostname = (hostname or "").rstrip(".")
+    return {
+        "alta_key": alta_key,
+        "hostname": hostname,
+        "web_ui_url": f"https://{hostname}" if hostname else "",
+        "ip_address": ipv4 or (addresses[0] if addresses else ""),
+        "port": port,
+        "service": SERVICE_TYPE.rstrip("."),
+    }
+
+
+def normalize_results(rows):
+    unique = {}
+    for row in rows:
+        unique[(row["alta_key"], row.get("ip_address", ""))] = row
+    return sorted(unique.values(), key=lambda row: (row["alta_key"], row.get("ip_address", "")))
+
+
+class CameraListener(ServiceListener):
+    def __init__(self):
+        self.rows = []
+        self.lock = threading.Lock()
+
+    def _capture(self, zeroconf: Zeroconf, service_type: str, name: str):
+        info: ServiceInfo | None = zeroconf.get_service_info(service_type, name, timeout=2500)
+        if not info:
+            return
+        row = camera_from_service(name, info.server or "", info.parsed_addresses(), info.port)
+        if row:
+            with self.lock:
+                self.rows.append(row)
+
+    def add_service(self, zeroconf, service_type, name):
+        self._capture(zeroconf, service_type, name)
+
+    def update_service(self, zeroconf, service_type, name):
+        self._capture(zeroconf, service_type, name)
+
+    def remove_service(self, zeroconf, service_type, name):
+        pass
+
+
+def scan_cameras(timeout: float = 5.0):
+    listener = CameraListener()
+    zeroconf = Zeroconf()
+    browser = ServiceBrowser(zeroconf, SERVICE_TYPE, listener)
+    try:
+        time.sleep(timeout)
+        with listener.lock:
+            cameras = normalize_results(listener.rows)
+    finally:
+        browser.cancel()
+        zeroconf.close()
+    return cameras
+
+
+class Handler(SimpleHTTPRequestHandler):
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, directory=str(ROOT / "static"), **kwargs)
+
+    def log_message(self, fmt, *args):
+        print(f"{self.client_address[0]} - {fmt % args}")
+
+    def _json(self, payload, status=HTTPStatus.OK):
+        body = json.dumps(payload).encode()
+        self.send_response(status)
+        self.send_header("Content-Type", "application/json; charset=utf-8")
+        self.send_header("Content-Length", str(len(body)))
+        self.send_header("Cache-Control", "no-store")
+        self.send_header("X-Content-Type-Options", "nosniff")
+        self.send_header("X-Frame-Options", "DENY")
+        self.end_headers()
+        self.wfile.write(body)
+
+    def do_GET(self):
+        if self.path == "/api/health":
+            return self._json({"ok": True, "service": "Alta Camera Finder"})
+        if self.path == "/api/scan":
+            started = time.monotonic()
+            try:
+                cameras = scan_cameras()
+                return self._json({
+                    "ok": True,
+                    "count": len(cameras),
+                    "duration_seconds": round(time.monotonic() - started, 1),
+                    "cameras": cameras,
+                })
+            except Exception as exc:
+                return self._json({"ok": False, "error": str(exc)}, HTTPStatus.INTERNAL_SERVER_ERROR)
+        if self.path == "/":
+            self.path = "/index.html"
+        return super().do_GET()
+
+
+def main():
+    parser = argparse.ArgumentParser(description="Discover Alta cloud-native cameras using mDNS")
+    parser.add_argument("--host", default="0.0.0.0")
+    parser.add_argument("--port", type=int, default=8787)
+    parser.add_argument("--scan", action="store_true", help="scan once and print JSON")
+    args = parser.parse_args()
+    if args.scan:
+        print(json.dumps(scan_cameras(), indent=2))
+        return
+    server = ThreadingHTTPServer((args.host, args.port), Handler)
+    print(f"Alta Camera Finder listening on http://{args.host}:{args.port}", flush=True)
+    server.serve_forever()
+
+
+if __name__ == "__main__":
+    main()