"""Unprivileged model broker; public authority stays in Node/Core.""" import pwd import re import threading import uuid from concurrent.futures import ThreadPoolExecutor, as_completed from pathlib import Path from .errors import failure from .http import Server, request from .operations import UNKNOWN, Operations ROOT = Path("/run/mission-core-insta360") INSTANCES = Path("/run/mission-core-x4-instances") IDENTIFIER = re.compile(r"instax4_[0-9a-f]{32}") STATE = Path("/var/lib/mission-core-insta360-broker") def worker_socket(identifier): if not isinstance(identifier, str) or not IDENTIFIER.fullmatch(identifier): raise ValueError("Invalid camera identity") return INSTANCES / identifier / "driver.sock" class Broker: def __init__(self): self.instance = "x4broker_" + uuid.uuid4().hex self.media_lock = threading.Lock() self.media = None self.media_operations = {} self.capture_locks = {} def capture_lock(self, identifier): with self.media_lock: if identifier not in self.capture_locks and len(self.capture_locks) >= 500: raise RuntimeError("Camera inventory exceeds the protocol limit") return self.capture_locks.setdefault(identifier, threading.Lock()) def capture_operation(self, identifier, value): path = worker_socket(identifier) lock = self.capture_lock(identifier) with self.media_lock: if self.media is None: from .media import Engine self.media = Engine() # Only one camera's START/STOP is serialized here; other devices keep # independent locks and SDK operation journals. with lock: current = request(path, "/snapshot", timeout=2) session = current["session_id"] if current.get("id") != identifier or value.get("session") != { "device_id": identifier, "session_id": session, }: raise ValueError("Camera session changed") key = (identifier, session) starting = value["action_id"] == "preview.start" starting_idle = starting and current.get("status", {}).get("preview") != 1 if starting: if not current.get("online"): raise RuntimeError("Camera state is unavailable") self.media.capture(key, path, True) try: result = request(path, "/operation", value, timeout=55) except (OSError, RuntimeError, ValueError): if starting_idle: self.media.capture(key, path, False) raise if (starting_idle and result.get("state") != "complete") or ( not starting and result.get("state") == "complete" ): # Releasing a decoder never sends STOP to the camera or SD. self.media.capture(key, path, False) return result def verify_operation(self, identifier, value): path = worker_socket(identifier) # Serialize with preview transitions, preserving one receipt namespace # whether verification uses the live decoder or an idle camera check. with self.capture_lock(identifier): current = request(path, "/snapshot", timeout=2) if current.get("id") != identifier: raise ValueError("Camera identity changed") operation = Operations( identifier, current["session_id"], STATE / identifier / "verification", VerificationCalls(self, path, value), ) return operation.execute(value) def media_operation(self, identifier, value): path = worker_socket(identifier) current = request(path, "/snapshot", timeout=2) if current.get("id") != identifier: raise ValueError("Camera identity changed") session = current["session_id"] with self.media_lock: if self.media is None: from .media import Engine self.media = Engine() previous = self.media_operations.get(identifier) if previous is None or previous.session_id != session: previous = Operations( identifier, session, STATE / identifier, MediaCalls(self.media, (identifier, session), path), ) self.media_operations[identifier] = previous return previous.execute(value) def snapshots(self): paths = sorted(INSTANCES.glob("instax4_*/driver.sock")) if len(paths) > 500: raise RuntimeError("Camera inventory exceeds the protocol limit") def read(path): if not IDENTIFIER.fullmatch(path.parent.name): return None try: value = request(path, "/snapshot", timeout=0.75) return value if value.get("id") == path.parent.name else None except (OSError, RuntimeError, ValueError): return None pool = ThreadPoolExecutor(max_workers=8) pending = [pool.submit(read, path) for path in paths] values = [] try: for future in as_completed(pending, timeout=2): value = future.result() if value is not None: values.append(value) except TimeoutError: pass finally: pool.shutdown(wait=False, cancel_futures=True) return sorted(values, key=lambda value: value["id"]) def item(self, value, node): status = value["status"] preview = status.get("preview") return { "id": value["id"], "name": "Insta360 X4", "model": "Insta360 X4", "kind": "insta360.x4", "initializable": True, "configured": False, "prepared": value["prepared"], "verified": value["verified"], "online": value["online"], "preparation_safe": value["preparation_safe"], "usb": "USB", "firmware": status.get("firmware"), "layers": [], "camera_status": status, "snapshot": { "revision": value["revision"], "observed_at": value["observed_at"], "context": { "session_id": value["session_id"], "opened_at": value["opened_at"], "device": { "device_id": value["id"], "stability": "stable", "basis": "hardware-identifier", "model": { "plugin_id": "missioncore.insta360", "plugin_version": "0.1.3", "model_id": "insta360.x4", }, }, "execution": { "node_id": node, "agent_instance_id": self.instance, "platform": "linux", }, }, "acquisition": "streaming" if preview == 1 else "idle" if preview == 0 else "failed", "connectivity": "connected" if value["online"] else "offline", "enrollment": "enrolled" if value["prepared"] else "empty", "message": value["message"], }, } def dispatch(self, method, route, value, headers): if method == "GET" and route == "/prepare-safe": snapshots = self.snapshots() expected = list(INSTANCES.glob("instax4_*")) return { "safe": len(snapshots) == len(expected) and all(item["preparation_safe"] for item in snapshots) } node = headers.get("X-Node-Id", "") if not re.fullmatch(r"node_[0-9a-f]{64}", node): raise ValueError("Node identity is required") if method == "GET" and route == "/inventory": return {"items": [self.item(item, node) for item in self.snapshots()]} if method == "POST" and route == "/operation": identifier = value["session"]["device_id"] if value.get("action_id") in ("preview.start", "preview.stop"): return self.capture_operation(identifier, value) if value.get("action_id") in ("offer", "close-peer"): return self.media_operation(identifier, value) if value.get("action_id") == "verify": return self.verify_operation(identifier, value) return request(worker_socket(identifier), "/operation", value, timeout=55) raise ValueError("Unsupported model route") class MediaCalls: def __init__(self, engine, key, path): self.engine, self.key, self.path = engine, key, path def call(self, action, params): if action == "offer": current = request(self.path, "/snapshot", timeout=2) if ( current.get("session_id") != self.key[1] or not current.get("online") or current.get("status", {}).get("preview") != 1 ): raise RuntimeError("Camera preview is not active") return self.engine.call(self.key, self.path, action, params) class VerificationCalls: def __init__(self, broker, path, command): self.broker, self.path, self.command = broker, path, command def call(self, action, params): current = request(self.path, "/snapshot", timeout=2) session = self.command["session"] if ( current.get("id") != session["device_id"] or current.get("session_id") != session["session_id"] or not current.get("online") ): return dict(UNKNOWN) status = current.get("status", {}) if status.get("recording") != 0: return failure("verification_recording_active") if status.get("preview") == 1: engine = self.broker.media if engine is None: return failure("preview_restart_required") return engine.verify((session["device_id"], session["session_id"])) if status.get("preview") == 0: # Preserve worker ownership of the bounded temporary idle preview. return request(self.path, "/operation", self.command, timeout=55) return dict(UNKNOWN) def main(): path = ROOT / "driver.sock" if path.exists(): path.unlink() allowed = {0, pwd.getpwnam("mission-core-node").pw_uid} with Server(path, Broker().dispatch, allowed) as server: path.chmod(0o660) server.serve_forever(poll_interval=0.5)