NODEDC_MISSION_CORE/plugins/insta360-x4/runtime/worker.py

"""One SDK session per OS-isolated USB instance. No public network listener."""

import os
import pwd
import threading
import time
import uuid
from datetime import UTC, datetime
from pathlib import Path

from .frames import Feed, encode
from .http import Binary, Server
from .identity import read_binding
from .lifecycle import acquisition
from .native import NativeCamera
from .operations import Operations
from .verification import Control


class Worker:
    def __init__(self, binding, runtime, state):
        self.binding = binding
        self.session = "x4_" + uuid.uuid4().hex
        self.opened_at = datetime.now(UTC).isoformat()
        self.status_lock = threading.Lock()
        self.status = {}
        self.observed = 0.0
        self.observed_at = self.opened_at
        self.revision = 0
        self.snapshot_signature = None
        self.failure = None
        self.ready = False
        self.done = threading.Event()
        self.camera = None
        self.control = None
        self.feed = None
        self.operations = None
        self.runtime, self.state = runtime, state

    def connect(self):
        # Bounded startup even if a vendor call never returns. No restart loop:
        # a new attempt requires explicit preparation or physical reconnection.
        timer = threading.Timer(40, lambda: os._exit(70))
        timer.daemon = True
        timer.start()
        try:
            logs = self.state / "sdk-logs"
            logs.mkdir(mode=0o700, exist_ok=True)
            # The prepare process starts services before releasing its lock.
            # Wait for that bounded transaction without starting the SDK early.
            with acquisition(wait=True):
                self.camera = NativeCamera(
                    self.binding, self.runtime / "lib/libmissioncore_x4.so", logs
                )
            self.feed = Feed(self.camera)
            self.feed.thread.start()
            self.control = Control(self.camera, self.feed)
            self.operations = Operations(
                self.binding.device_id, self.session, self.state / "operations", self.control
            )
            self.refresh()
            self.ready = True
        except (RuntimeError, OSError, ValueError):
            self.failure = "Не удалось открыть X4. Проверьте питание и режим USB на камере."
        finally:
            timer.cancel()

    def refresh(self):
        if read_binding(self.binding.port) != self.binding:
            raise RuntimeError("Camera transport changed")
        response = self.camera.call("details", {})
        if response["state"] != "complete":
            raise RuntimeError("Camera status is unconfirmed")
        with self.status_lock:
            self.status = response["result"]
            self.observed = time.monotonic()
            self.observed_at = datetime.now(UTC).isoformat()
            self.revision += 1

    def snapshot(self):
        with self.status_lock:
            state = dict(self.status)
            fresh = time.monotonic() - self.observed < 12
            connected = self.ready and fresh and state.get("connected") is True
            verified = bool(self.control and self.control.verified)
            signature = (self.ready, connected, fresh, verified, self.failure)
            if signature != self.snapshot_signature:
                self.revision += 1
                self.snapshot_signature = signature
                self.observed_at = datetime.now(UTC).isoformat()
            revision, observed_at = self.revision, self.observed_at
        return {
            "id": self.binding.device_id,
            "session_id": self.session,
            "opened_at": self.opened_at,
            "prepared": self.ready,
            "online": connected,
            "verified": verified,
            "revision": revision,
            "observed_at": observed_at,
            "preparation_safe": connected
            and state.get("preview") == 0
            and state.get("recording") == 0,
            "status": state if fresh else {},
            "message": self.failure,
        }

    def dispatch(self, method, route, value, headers):
        if method == "GET" and route == "/snapshot":
            return self.snapshot()
        if method == "POST" and route == "/video":
            if (
                not self.ready
                or value.get("session_id") != self.session
                or set(value) != {"session_id", "cursor"}
            ):
                raise ValueError("Camera video session changed")
            return Binary(encode(self.feed.read(value["cursor"])))
        if method == "POST" and route == "/operation":
            if not self.ready:
                raise RuntimeError("Camera is unavailable")
            with acquisition():
                result = self.operations.execute(value)
            # Poll asynchronously; a lost state read must not change a durable
            # acknowledged command result into a different receipt.
            if value.get("action_id") not in ("details", "settings.read", "files.list"):
                with self.status_lock:
                    self.observed = 0
            return result
        raise ValueError("Unsupported worker route")

    def poll(self):
        self.connect()
        while self.ready and not self.done.wait(3):
            try:
                self.refresh()
            except (RuntimeError, OSError, ValueError):
                with self.status_lock:
                    self.observed = 0


def main(port, runtime):
    binding = read_binding(port)
    state = Path(os.environ["STATE_DIRECTORY"])
    socket = Path(os.environ["RUNTIME_DIRECTORY"]) / "driver.sock"
    worker = Worker(binding, runtime, state)
    thread = threading.Thread(target=worker.poll, daemon=True)
    thread.start()
    if socket.exists():
        socket.unlink()
    allowed = {0, pwd.getpwnam("mission-core-insta360").pw_uid}
    with Server(socket, worker.dispatch, allowed) as server:
        socket.chmod(0o660)
        server.serve_forever(poll_interval=0.5)