136 lines
4.6 KiB
Python
136 lines
4.6 KiB
Python
#!/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()
|