From b59275ba3862b0d05de21615b168f7b3039284f9 Mon Sep 17 00:00:00 2001 From: Gareth Morgan Date: Tue, 3 Nov 2026 06:29:59 -0500 Subject: [PATCH 1/6] capture_server: add a device query and MCAL file selection Add a read-only /query route on the existing listener. One QueryDevices request gets one Devices reply on its own connection, without taking the control connection's ownership or touching a session, so it works before and during capture. The Manus entry lists the .mcal files in the folder set with --manus-calibration-dir. Begin can select those files by basename per glove (left_calibration / right_calibration). The server checks them against the same folder policy the listing uses, before any hardware starts, and passes the resolved paths to the Manus plugin's own calibration arguments. The folder is a plain folder of .mcal files, independent of how they were produced. Co-Authored-By: Claude Sonnet 5.5 Signed-off-by: Gareth Morgan --- examples/egocap/capture_server/README.md | 15 +- .../acquisition/isaac_capture.py | 15 + .../capture_server/capture_server/client.py | 86 ++- .../capture_server/capture_server/launch.py | 42 +- .../capture_server/mcal_catalog.py | 102 +++ .../capture_server/capture_server/protocol.py | 227 ++++++- .../capture_server/capture_server/server.py | 168 ++++- .../tests/capture_server/test_device_query.py | 584 ++++++++++++++++++ .../tests/capture_server/test_protocol.py | 2 +- 9 files changed, 1224 insertions(+), 17 deletions(-) create mode 100644 examples/egocap/capture_server/capture_server/mcal_catalog.py create mode 100644 examples/egocap/tests/capture_server/test_device_query.py diff --git a/examples/egocap/capture_server/README.md b/examples/egocap/capture_server/README.md index bfaebbdde3..f981f76196 100644 --- a/examples/egocap/capture_server/README.md +++ b/examples/egocap/capture_server/README.md @@ -352,6 +352,7 @@ The client validates the wire SessionInit notification and converts it to Captur | `enabled` | `bool` | Whether the station launches it; true for a plugin selected through start. | | `required` | `bool` | Whether failure to discover the plugin fails startup; false for start selections. It does not require connected gloves. | | `datasets` | `tuple[ManusDataset, ...]`: `HUMAN=1`. | Requested live human-skeleton dataset. start selects HUMAN. | +| `left_calibration`, `right_calibration` | `str` or omitted: an `.mcal` basename. | MCAL file for that glove, chosen from the server's MCAL folder (see Device query). Omitted when none was selected. | #### CaptureClock @@ -686,19 +687,29 @@ One JSON object is one text message. One complete header and payload is one bina | Server → client | 5 | SessionStatus | | Server → client | 6 | SessionStopped | | Server → client | 10 | RequestRejected | +| Client → server, on `/query` | 11 | QueryDevices | +| Server → client, on `/query` | 12 | Devices | | Server → client | BinaryKind 1 | CaptureFrame | | Server → client | BinaryKind 2 | CaptureImage | -There are ten message families. SessionInit describes the resolved session configuration and complete decoding layout before hardware initialization. SessionStarting reports progress and observed availability. SessionStatus carries IC's changing provider/device state without treating absence as terminal or confusing status with pose validity. No application recording command or file path is accepted. +There are ten session message families, plus the device query described below. SessionInit describes the resolved session configuration and complete decoding layout before hardware initialization. SessionStarting reports progress and observed availability. SessionStatus carries IC's changing provider/device state without treating absence as terminal or confusing status with pose validity. No application recording command or file path is accepted. Every JSON message has these envelope fields. An envelope is the small common part of a message that allows the receiver to route it before reading message-specific content. | Field | Allowed contents | What it does | |---|---|---| -| `type` | `MessageType`: `BEGIN=1`, `STOP=2`, `SESSION_STARTING=3`, `SESSION_BEGUN=4`, `SESSION_STATUS=5`, `SESSION_STOPPED=6`, `SESSION_INIT=9`, `REQUEST_REJECTED=10`. Binary families use the header's BinaryKind instead. | Selects the parser and action. Each message accepts only its own type value. The human-readable names in this document are explanations, not wire values. | +| `type` | `MessageType`: `BEGIN=1`, `STOP=2`, `SESSION_STARTING=3`, `SESSION_BEGUN=4`, `SESSION_STATUS=5`, `SESSION_STOPPED=6`, `SESSION_INIT=9`, `REQUEST_REJECTED=10`, `QUERY_DEVICES=11`, `DEVICES=12`. Binary families use the header's BinaryKind instead. | Selects the parser and action. Each message accepts only its own type value. The human-readable names in this document are explanations, not wire values. | | `version` | Not an enum: exactly integer 1; not boolean true. | Identifies the protocol contract used to interpret all remaining fields. A receiver must reject another version rather than guess its meaning. | | `session_id` | Nonempty string, at most 128 characters; a new UUID string is recommended for each new session. | Associates configuration, status, samples and terminal replies with one session. A reply for another ID must not reset the active session. This is a client-chosen correlation ID, not a hardware identifier. | +#### Device query + +`QueryDevices` is a read-only request on the server's `/query` path: its own short-lived connection, one request and one reply, no ownership and no session, so it works before capture and while a capture is running. Any other message on `/query`, including every capture command, closes the connection with 1002. `await client.query_devices()` returns the `DeviceEntry` tuple and raises `DeviceQueryError` when no answer is obtained; a failure is never reported as an empty list. + +Request: `{"type": 11, "version": 1, "request_id": "<1..128 chars>"}`. Reply: `{"type": 12, "version": 1, "request_id": ..., "devices": [...]}` with one entry per device kind. The Manus entry is `{"device": 1, "mcal_status": ..., "mcal_files": [...], "error_message": ...}`. `mcal_status` is `READY=1` (the list may be empty), `DISABLED=2` (the server has no MCAL folder) or `UNAVAILABLE=3` (the folder could not be read; `error_message` says why). A listing over the 2000-file limit is `UNAVAILABLE`, never truncated. + +The MCAL folder is set with `--manus-calibration-dir DIR`. It is a plain folder: any nonempty, readable `.mcal` regular file directly in it is offered, whether placed by hand or written by a calibration tool. Symbolic links are never offered; copy files in. Select files by name with `ManusPluginConfig(left_calibration=..., right_calibration=...)` in `start(plugins=...)`. Begin is refused as `INVALID_CONFIG` when a name is not in the folder at that moment. A name only selects a file; whether the glove accepted it is the Manus plugin's per-hand status. + #### Begin: configure sources, plugins and cameras ```json diff --git a/examples/egocap/capture_server/capture_server/acquisition/isaac_capture.py b/examples/egocap/capture_server/capture_server/acquisition/isaac_capture.py index 96be446dbe..bbc379528e 100644 --- a/examples/egocap/capture_server/capture_server/acquisition/isaac_capture.py +++ b/examples/egocap/capture_server/capture_server/acquisition/isaac_capture.py @@ -153,6 +153,11 @@ class TrackingSelection: # silently. Meaningless when manus_enabled is False (no plugin is # constructed at all in that case). manus_required: bool = False + # Resolved host paths of MCAL files for the Manus plugin (the server turns the client's basenames into these + # before any hardware starts). Passed to the plugin as its own calibration arguments, so whether the glove + # accepted a file is still reported by the plugin's per-hand status, not assumed here. + manus_left_calibration: str | None = None + manus_right_calibration: str | None = None class IsaacCaptureAdapter: @@ -229,6 +234,7 @@ def _build_config(self) -> Any: plugin_name=MANUS_PLUGIN_NAME, plugin_root_id="manus", search_paths=[], + plugin_args=self._manus_args(), required=self._selection.manus_required, ) ] @@ -237,6 +243,15 @@ def _build_config(self) -> Any: ) return self._sdk.TeleopSessionConfig(app_name="capture_server", pipeline=pipeline, plugins=plugins) + def _manus_args(self) -> list[str]: + """The Manus plugin's calibration arguments for the files this session selected.""" + args = [] + if self._selection.manus_left_calibration is not None: + args.append(f"--left-calibration={self._selection.manus_left_calibration}") + if self._selection.manus_right_calibration is not None: + args.append(f"--right-calibration={self._selection.manus_right_calibration}") + return args + def _enter_sync(self) -> Any: # CloudXRLauncher.launch_context is the shared launcher/context # boundary the approved API subset requires the adapter to diff --git a/examples/egocap/capture_server/capture_server/client.py b/examples/egocap/capture_server/capture_server/client.py index 9776eb38ce..397f3d8e29 100644 --- a/examples/egocap/capture_server/capture_server/client.py +++ b/examples/egocap/capture_server/capture_server/client.py @@ -50,7 +50,8 @@ from dataclasses import dataclass from dataclasses import field as dataclass_field from enum import IntEnum -from typing import Any, TypeVar +from typing import Any, ClassVar, TypeVar +from urllib.parse import urlsplit import numpy as np import websockets @@ -68,6 +69,8 @@ __all__ = [ "CaptureClient", + "DeviceQueryError", + "ManusPluginConfig", "CaptureFrame", "CaptureStarted", "CaptureStatus", @@ -86,6 +89,8 @@ _DEFAULT_RATE_HZ = 60.0 _ACCEPT_TIMEOUT_S = 10.0 _STOP_TIMEOUT_S = 10.0 +# A device query is one short exchange; the server bounds its own scan well inside this. +_QUERY_TIMEOUT_S = 20.0 _CLOSE_TIMEOUT_S = 2.0 # The documented transport ceiling is a complete JPEG (2 MiB) plus the # 32-byte binary header; websockets' own default max_size (~1 MiB) would @@ -290,6 +295,35 @@ class CaptureStopped: streams: tuple[p.StreamCounters, ...] | None +@dataclass(frozen=True) +class ManusPluginConfig: + """The Manus plugin and, optionally, MCAL files to load for each glove. + + The names are bare ``.mcal`` basenames of files in the server's managed MCAL folder, as listed by + :meth:`CaptureClient.query_devices`; they are never paths on this machine. A name only selects a file: + whether the glove accepted it is reported by the plugin's own per-hand status. + """ + + left_calibration: str | None = None + right_calibration: str | None = None + plugin: ClassVar[p.PluginType] = p.PluginType.MANUS + + def __post_init__(self) -> None: + """Reject names the protocol cannot carry, and one file chosen for both hands.""" + for name in (self.left_calibration, self.right_calibration): + if name is not None and (not isinstance(name, str) or not p._MCAL_NAME_RE.fullmatch(name)): + raise ValueError(f"calibration file must be an .mcal basename, got {name!r}") + if self.left_calibration is not None and self.left_calibration == self.right_calibration: + raise ValueError("left_calibration and right_calibration must name different files") + + +class DeviceQueryError(Exception): + """The device list could not be retrieved (transport, timeout, protocol, or a server without the query). + + Never reported as an empty list: an answer with no MCAL files is a READY entry with none. + """ + + def _is_awaitable(value: Any) -> bool: return inspect.isawaitable(value) @@ -370,7 +404,7 @@ async def start( *, sources: Sequence[p.SourceType], cameras: Sequence[p.CameraSelection] = (), - plugins: Sequence[p.PluginType] = (), + plugins: Sequence[p.PluginType | ManusPluginConfig] = (), on_started: Callback[CaptureStarted] | None = None, on_status: Callback[CaptureStatus] | None = None, on_capture_frame: Callback[CaptureFrame] | None = None, @@ -733,6 +767,41 @@ async def _finalize(self, attempt: _Attempt, stopped: CaptureStopped) -> None: finally: self._attempt = None + async def query_devices(self) -> tuple[p.DeviceEntry, ...]: + """List the devices the server offers, without taking the control connection. + + Uses its own short-lived connection to the server's read-only ``/query`` route, so it works before + capture and while capturing (including from a callback), and needs no glove or session. The Manus + entry lists the MCAL files a :class:`ManusPluginConfig` can select. Concurrent calls are independent. + Raises :class:`DeviceQueryError` when no answer is obtained, including when the server is too old to + know the query. + """ + # The query route is a sibling of the control path on the same listener. + url = urlsplit(self._url)._replace(path="/query", query="", fragment="").geturl() + request = p.QueryDevicesRequest(request_id=f"devices-{uuid.uuid4()}") + try: + async with asyncio.timeout(_QUERY_TIMEOUT_S): + async with websockets.connect(url, compression=None, max_size=_MAX_MESSAGE_BYTES) as ws: + await ws.send(_to_json_text(request.to_json())) + raw = await ws.recv() + except ConnectionClosed as exc: + code = exc.rcvd.code if exc.rcvd is not None else None + raise DeviceQueryError( + f"the server closed the device query (close code {code}); it may not support device queries" + ) from None + except (OSError, TimeoutError, websockets.exceptions.WebSocketException) as exc: + raise DeviceQueryError(f"the device query failed: {exc or type(exc).__name__}") from None + if isinstance(raw, bytes): + raise DeviceQueryError("the server answered the device query with a binary message") + try: + reply = p.parse_server_message(_from_json_text(raw)) + except p.ProtocolError as exc: + raise DeviceQueryError(f"the server's device reply was invalid: {exc}") from None + # A reply for another request, or of another kind, is never accepted as this query's answer. + if not isinstance(reply, p.DevicesReply) or reply.request_id != request.request_id: + raise DeviceQueryError("the server's reply does not answer this device query") + return reply.devices + # -- message/result construction --------------------------------------- def _build_begin_request( @@ -740,11 +809,20 @@ def _build_begin_request( *, sources: Sequence[p.SourceType], cameras: Sequence[p.CameraSelection], - plugins: Sequence[p.PluginType], + plugins: Sequence[p.PluginType | ManusPluginConfig], ) -> p.BeginRequest: try: tracking = p.TrackingConfig(sources=tuple(sources), rate_hz=_DEFAULT_RATE_HZ) - plugin_requests = tuple(p.PluginRequest(plugin=plugin) for plugin in plugins) + plugin_requests = tuple( + p.PluginRequest( + plugin=plugin.plugin, + left_calibration=plugin.left_calibration, + right_calibration=plugin.right_calibration, + ) + if isinstance(plugin, ManusPluginConfig) + else p.PluginRequest(plugin=plugin) + for plugin in plugins + ) camera_requests = tuple(p.CameraRequest(name=c["name"], type=c["type"]) for c in cameras) return p.BeginRequest( tracking=tracking, diff --git a/examples/egocap/capture_server/capture_server/launch.py b/examples/egocap/capture_server/capture_server/launch.py index 626ebe06be..0b98574354 100644 --- a/examples/egocap/capture_server/capture_server/launch.py +++ b/examples/egocap/capture_server/capture_server/launch.py @@ -26,9 +26,11 @@ from dataclasses import dataclass from functools import partial +from pathlib import Path from types import ModuleType from typing import TYPE_CHECKING, Any, Callable +from . import mcal_catalog from . import protocol as p from .acquisition import Acquisition, DepthAIAdapter, IsaacCaptureAdapter, TrackingSelection from .server import ServerConfig, StationCamera @@ -58,8 +60,25 @@ def _group_selected(begin: p.BeginRequest, source_type: p.SourceType) -> bool: return source_type in begin.tracking.sources +def _calibration_path(name: str | None, directory: Path | None) -> str | None: + """The host path of a selected MCAL file; the name itself where no folder exists to resolve it. + + The server has already checked the name against the folder before this runs. Without a folder (the mock + and replay setups, which never reach real hardware) the name is kept as an inert marker. + """ + if name is None: + return None + if directory is None: + return name + return str(mcal_catalog.resolve(directory, name)) + + def _tracking_adapter( - begin: p.BeginRequest, *, sdk: ModuleType, session_factory: Callable[[Any], Any] + begin: p.BeginRequest, + *, + sdk: ModuleType, + session_factory: Callable[[Any], Any], + calibration_dir: Path | None = None, ) -> IsaacCaptureAdapter | None: """The IsaacCapture adapter for the requested tracking outputs, or None when none are requested. @@ -80,6 +99,12 @@ def _tracking_adapter( hand_right=_group_selected(begin, p.SourceType.HANDS), manus_enabled=manus_plugin_request is not None, manus_required=manus_plugin_request.required if manus_plugin_request is not None else False, + manus_left_calibration=_calibration_path( + manus_plugin_request.left_calibration if manus_plugin_request is not None else None, calibration_dir + ), + manus_right_calibration=_calibration_path( + manus_plugin_request.right_calibration if manus_plugin_request is not None else None, calibration_dir + ), ) return IsaacCaptureAdapter( sdk=sdk, session_factory=session_factory, selection=selection, rate_hz=begin.tracking.rate_hz @@ -183,7 +208,13 @@ def source_factory(begin: p.BeginRequest, cameras: tuple[p.CameraStreamConfig, . from .acquisition.isaac_sdk import load_isaac_capture sdk = load_isaac_capture() if begin.tracking.sources else None - tracking = _tracking_adapter(begin, sdk=sdk, session_factory=sdk.TeleopSession) if sdk else None + tracking = ( + _tracking_adapter( + begin, sdk=sdk, session_factory=sdk.TeleopSession, calibration_dir=config.manus_calibration_dir + ) + if sdk + else None + ) return Acquisition(tracking=tracking, cameras=_real_camera_adapters(config, cameras)) source_factory.mock = False # type: ignore[attr-defined] @@ -280,6 +311,7 @@ def build_gmsl_source_factory( stations: dict[str, StationCamera], *, slot_mapper: SlotMapper | None = None, + calibration_dir: Path | None = None, ) -> SourceFactory: """The live-station source factory: real IsaacCapture tracking (from ``$ISAAC_ROOT``, only when a session requests tracking) plus GMSL cameras read from the receiver's frame feed. @@ -297,7 +329,11 @@ def source_factory(begin: p.BeginRequest, requested: tuple[p.CameraStreamConfig, from .acquisition.isaac_sdk import load_isaac_capture sdk = load_isaac_capture() if begin.tracking.sources else None - tracking = _tracking_adapter(begin, sdk=sdk, session_factory=sdk.TeleopSession) if sdk else None + tracking = ( + _tracking_adapter(begin, sdk=sdk, session_factory=sdk.TeleopSession, calibration_dir=calibration_dir) + if sdk + else None + ) if not requested: return Acquisition(tracking=tracking, cameras=()) for camera in requested: diff --git a/examples/egocap/capture_server/capture_server/mcal_catalog.py b/examples/egocap/capture_server/capture_server/mcal_catalog.py new file mode 100644 index 0000000000..af5b7c2cea --- /dev/null +++ b/examples/egocap/capture_server/capture_server/mcal_catalog.py @@ -0,0 +1,102 @@ +# SPDX-FileCopyrightText: 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# +# SPDX-License-Identifier: Apache-2.0 + +"""The MCAL files in the server's managed directory, and the one policy deciding which are selectable. + +The device query and the begin-time check of a selected file both go through :func:`scan`, so a file that is +listed is a file that can be selected, and the reverse. The directory is just a folder of ``.mcal`` files: +anything placed there, by hand or by a calibration tool, is offered. Scanning reads the directory, so callers +run it off the event loop. + +A file is selectable when it is a nonempty, readable ``.mcal`` regular file directly in the directory. Symbolic +links are never offered; copy files into the directory instead. Whether a glove accepts a file is the Manus +plugin's decision, not this module's. +""" + +from __future__ import annotations + +import os +from dataclasses import dataclass +from pathlib import Path + +from . import protocol as p + +__all__ = ["Catalog", "resolve", "scan", "selection_refusal"] + + +@dataclass(frozen=True) +class Catalog: + """One scan's result: its status, the selectable files sorted by name, and the reason when UNAVAILABLE.""" + + status: p.ManusCalibrationCatalogStatus + files: tuple[str, ...] = () + error_message: str | None = None + + +def scan(directory: Path | None) -> Catalog: + """Read the managed directory once and return everything selectable, sorted by filename.""" + # No directory configured: nothing can be listed or selected. + if directory is None: + return Catalog(p.ManusCalibrationCatalogStatus.DISABLED) + try: + # The directory is created by whatever first writes a file into it, so absence is an empty catalog. + if not directory.exists(): + return Catalog(p.ManusCalibrationCatalogStatus.READY) + root = directory.resolve(strict=True) + if not root.is_dir(): + return _unavailable("the calibration directory is not a directory") + files = _selectable(root) + except PermissionError: + return _unavailable("the calibration directory cannot be read (permission denied)") + except OSError as exc: + return _unavailable(f"the calibration directory cannot be read ({exc.strerror or 'I/O error'})") + # A listing too large for one reply is an error, never a truncated list. + if len(files) > p.MAX_MCAL_FILES: + return _unavailable(f"the calibration directory has more than {p.MAX_MCAL_FILES} selectable files") + return Catalog(p.ManusCalibrationCatalogStatus.READY, tuple(sorted(files))) + + +def _unavailable(message: str) -> Catalog: + """An UNAVAILABLE catalog carrying the operator-facing reason.""" + return Catalog(p.ManusCalibrationCatalogStatus.UNAVAILABLE, (), message) + + +def _selectable(root: Path) -> list[str]: + """Files directly in ``root`` that pass the selection policy.""" + files = [] + with os.scandir(root) as listing: + for entry in listing: + # Names the protocol cannot carry are never offered. + if not p._MCAL_NAME_RE.fullmatch(entry.name): + continue + try: + # Links are refused even when their target stays inside the directory. + if entry.is_symlink() or not entry.is_file(follow_symlinks=False): + continue + if entry.stat(follow_symlinks=False).st_size == 0 or not os.access(entry.path, os.R_OK): + continue + except OSError: + # An entry that vanished mid-scan is simply not offered. + continue + files.append(entry.name) + return files + + +def selection_refusal(catalog: Catalog, names: list[str]) -> str | None: + """Why a requested file name cannot be used against this catalog, or None when all can.""" + if catalog.status is p.ManusCalibrationCatalogStatus.DISABLED: + return "calibration files can only be selected when the server has a --manus-calibration-dir" + if catalog.status is p.ManusCalibrationCatalogStatus.UNAVAILABLE: + return f"calibration files cannot be selected: {catalog.error_message}" + available = set(catalog.files) + for name in names: + if name not in available: + return f"calibration file {name!r} is not an available file in the calibration directory" + return None + + +def resolve(directory: Path, name: str) -> Path: + """The host path of one selected file; the server has already checked it against a scan.""" + # The resolved real path is handed to the tracking plugin, so nothing outside the directory can reach it. + return directory.resolve() / name diff --git a/examples/egocap/capture_server/capture_server/protocol.py b/examples/egocap/capture_server/capture_server/protocol.py index 04ef35bd86..010febde5d 100644 --- a/examples/egocap/capture_server/capture_server/protocol.py +++ b/examples/egocap/capture_server/capture_server/protocol.py @@ -28,6 +28,7 @@ from __future__ import annotations +import re import struct from dataclasses import dataclass from dataclasses import field as dataclass_field @@ -70,6 +71,9 @@ class MessageType(IntEnum): CAPTURE_IMAGE = 8 SESSION_INIT = 9 REQUEST_REJECTED = 10 + # Station queries on the read-only /query route: one request, one reply, no session and no ownership. + QUERY_DEVICES = 11 + DEVICES = 12 class SourceType(IntEnum): @@ -86,6 +90,18 @@ class ManusDataset(IntEnum): HUMAN = 1 +class HandSide(IntEnum): + LEFT = 1 + RIGHT = 2 + + +class ManusCalibrationCatalogStatus(IntEnum): + # Whether the Manus MCAL folder listing is usable. + READY = 1 + DISABLED = 2 + UNAVAILABLE = 3 + + class CaptureMode(IntEnum): MOCK = 0 LIVE = 1 @@ -925,6 +941,27 @@ def _require_session_id(data: dict) -> str: return session_id +# A bare filename: no separators, no leading dot, bounded length. +_MCAL_NAME_RE = re.compile(r"[A-Za-z0-9][A-Za-z0-9_.-]{0,127}\.mcal") +_MAX_TOKEN_LEN = 128 + + +def _require_mcal_name(value: Any, context: str) -> str | None: + """An optional ``.mcal`` basename, or ``ProtocolError`` when present but not one.""" + if value is None: + return None + if not isinstance(value, str) or not _MCAL_NAME_RE.fullmatch(value): + raise ProtocolError(f"{context} must be an .mcal basename or null") + return value + + +def _require_token(value: Any, context: str) -> str: + """A nonempty, bounded string identifier.""" + if not isinstance(value, str) or not value or len(value) > _MAX_TOKEN_LEN: + raise ProtocolError(f"{context} must be a nonempty string of at most {_MAX_TOKEN_LEN} characters") + return value + + def require_envelope(data: dict, expected_type: MessageType) -> str: """Envelope-level validation shared by every message kind that needs it: message type, protocol version, and the canonical session_id rule @@ -985,10 +1022,18 @@ class PluginRequest: enabled: bool = dataclass_field(default=True, kw_only=True) required: bool = dataclass_field(default=False, kw_only=True) datasets: tuple[ManusDataset, ...] = dataclass_field(default=(ManusDataset.HUMAN,), kw_only=True) + # MCAL basenames in the server's managed directory for the Manus plugin; resolved by the server, never paths. + left_calibration: str | None = dataclass_field(default=None, kw_only=True) + right_calibration: str | None = dataclass_field(default=None, kw_only=True) def __post_init__(self) -> None: if type(self.plugin) is not PluginType: raise ValueError(f"PluginRequest.plugin must be a PluginType, got {self.plugin!r}") + for name in (self.left_calibration, self.right_calibration): + if name is not None and (type(name) is not str or not _MCAL_NAME_RE.fullmatch(name)): + raise ValueError(f"PluginRequest calibration must be an .mcal basename, got {name!r}") + if self.left_calibration is not None and self.left_calibration == self.right_calibration: + raise ValueError("left_calibration and right_calibration must name different files") if type(self.enabled) is not bool: raise ValueError("PluginRequest.enabled must be a boolean") if type(self.required) is not bool: @@ -1125,18 +1170,30 @@ class PluginConfiguration: enabled: bool required: bool datasets: tuple[ManusDataset, ...] + left_calibration: str | None = None + right_calibration: str | None = None def to_json(self) -> dict[str, Any]: - return { + data: dict[str, Any] = { "plugin": int(self.plugin), "enabled": self.enabled, "required": self.required, "datasets": [int(d) for d in self.datasets], } + # Omitted, not null, so a selection without files keeps its exact earlier wire form. + if self.left_calibration is not None: + data["left_calibration"] = self.left_calibration + if self.right_calibration is not None: + data["right_calibration"] = self.right_calibration + return data @classmethod def from_json(cls, data: Any) -> PluginConfiguration: - _require_keys(data, {"plugin", "enabled", "required", "datasets"}, "plugin configuration") + _require_keys( + data, + {"plugin", "enabled", "required", "datasets", "left_calibration", "right_calibration"}, + "plugin configuration", + ) datasets_raw = data.get("datasets") if not isinstance(datasets_raw, list) or not datasets_raw: raise ProtocolError("plugin configuration datasets must be a nonempty list") @@ -1147,6 +1204,8 @@ def from_json(cls, data: Any) -> PluginConfiguration: datasets=tuple( _require_int_enum(d, ManusDataset, "plugin configuration datasets[]") for d in datasets_raw ), + left_calibration=_require_mcal_name(data.get("left_calibration"), "plugin configuration left"), + right_calibration=_require_mcal_name(data.get("right_calibration"), "plugin configuration right"), ) @@ -1192,7 +1251,14 @@ def effective_config(begin: BeginRequest) -> CaptureConfiguration: headset=HeadsetConfiguration(device_index=begin.tracking.headset.device_index), ), plugins=tuple( - PluginConfiguration(plugin=p.plugin, enabled=p.enabled, required=p.required, datasets=p.datasets) + PluginConfiguration( + plugin=p.plugin, + enabled=p.enabled, + required=p.required, + datasets=p.datasets, + left_calibration=p.left_calibration, + right_calibration=p.right_calibration, + ) for p in begin.plugins ), cameras=tuple(CameraSelection(name=c.name, type=c.type) for c in begin.cameras), @@ -1292,7 +1358,11 @@ def _parse_plugin_requests(data: Any) -> tuple[PluginRequest, ...]: def _parse_plugin_request(data: Any) -> PluginRequest: - _require_keys(data, {"plugin", "enabled", "required", "datasets"}, "a plugin request") + _require_keys( + data, + {"plugin", "enabled", "required", "datasets", "left_calibration", "right_calibration"}, + "a plugin request", + ) plugin = _require_int_enum(data.get("plugin"), PluginType, "plugin.plugin") enabled = _require_strict_bool(data.get("enabled", True), "plugin.enabled") required = _require_strict_bool(data.get("required", False), "plugin.required") @@ -1304,7 +1374,18 @@ def _parse_plugin_request(data: Any) -> PluginRequest: if len(datasets) != len(set(datasets)): raise ProtocolError(f"plugin.datasets has duplicate entries: {[int(d) for d in datasets]}") - return PluginRequest(plugin=plugin, enabled=enabled, required=required, datasets=datasets) + left = _require_mcal_name(data.get("left_calibration"), "plugin.left_calibration") + right = _require_mcal_name(data.get("right_calibration"), "plugin.right_calibration") + if left is not None and left == right: + raise ProtocolError("plugin.left_calibration and plugin.right_calibration must name different files") + return PluginRequest( + plugin=plugin, + enabled=enabled, + required=required, + datasets=datasets, + left_calibration=left, + right_calibration=right, + ) def _parse_camera_requests(data: Any) -> tuple[CameraRequest, ...]: @@ -1848,8 +1929,141 @@ def from_json(cls, data: Any) -> RequestRejected: ) +# --------------------------------------------------------------------------- +# Station queries. A query is one request and one reply on the read-only /query route; it carries a +# ``request_id`` (never a session) and neither takes the control connection's ownership nor touches a session. +# --------------------------------------------------------------------------- + +# A larger MCAL listing is an error, never a silently truncated list. +MAX_MCAL_FILES = 2000 +# The serialized reply must fit well inside the transport's message ceiling whatever the file count. +MAX_QUERY_REPLY_BYTES = 1024 * 1024 + + +@dataclass(frozen=True) +class QueryDevicesRequest: + """Ask which devices this server offers and what each can be configured with.""" + + request_id: str + + def to_json(self) -> dict[str, Any]: + """The request as a wire object.""" + return {"type": int(MessageType.QUERY_DEVICES), "version": PROTOCOL_VERSION, "request_id": self.request_id} + + +def parse_query_devices(data: dict) -> QueryDevicesRequest: + """Strict parse of a device query; the reply is addressed by its ``request_id``.""" + _require_keys(data, {"type", "version", "request_id"}, "QueryDevices") + _require_message_type(data, MessageType.QUERY_DEVICES) + _require_version(data) + return QueryDevicesRequest(request_id=_require_token(data.get("request_id"), "request_id")) + + +@dataclass(frozen=True) +class DeviceEntry: + """One device and its configuration choices. + + For the Manus device, ``mcal_files`` are the basenames in the server's managed MCAL folder that a Begin + can select. ``mcal_status`` says whether that listing is usable: READY (possibly empty), DISABLED (the + server has no MCAL folder) or UNAVAILABLE (the folder could not be read, with ``error_message``). + """ + + device: PluginType + mcal_status: ManusCalibrationCatalogStatus + mcal_files: tuple[str, ...] = () + error_message: str | None = None + + def __post_init__(self) -> None: + unavailable = self.mcal_status is ManusCalibrationCatalogStatus.UNAVAILABLE + if unavailable != (self.error_message is not None): + raise ValueError("error_message is present exactly when the MCAL listing is UNAVAILABLE") + if self.error_message is not None and not self.error_message: + raise ValueError("error_message must be nonempty or null") + if self.mcal_status is not ManusCalibrationCatalogStatus.READY and self.mcal_files: + raise ValueError("mcal_files are present only for a READY listing") + if len(self.mcal_files) > MAX_MCAL_FILES: + raise ValueError(f"at most {MAX_MCAL_FILES} MCAL files fit in one reply") + if not all(_MCAL_NAME_RE.fullmatch(name) for name in self.mcal_files): + raise ValueError("mcal_files must be .mcal basenames") + if len(set(self.mcal_files)) != len(self.mcal_files): + raise ValueError("an MCAL file appears twice in the listing") + + def to_json(self) -> dict[str, Any]: + """The entry as a wire object.""" + return { + "device": int(self.device), + "mcal_status": int(self.mcal_status), + "mcal_files": list(self.mcal_files), + "error_message": self.error_message, + } + + @classmethod + def from_json(cls, data: Any) -> DeviceEntry: + """Strict parse of one device entry.""" + _require_keys(data, {"device", "mcal_status", "mcal_files", "error_message"}, "device entry") + files, error_message = data.get("mcal_files"), data.get("error_message") + if not isinstance(files, list) or not all(isinstance(f, str) for f in files): + raise ProtocolError("device entry mcal_files must be a list of strings") + if error_message is not None and not isinstance(error_message, str): + raise ProtocolError("device entry error_message must be a string or null") + try: + return cls( + device=_require_int_enum(data.get("device"), PluginType, "device entry device"), + mcal_status=_require_int_enum( + data.get("mcal_status"), ManusCalibrationCatalogStatus, "device entry mcal_status" + ), + mcal_files=tuple(files), + error_message=error_message, + ) + except ValueError as exc: + raise ProtocolError(str(exc)) from None + + +@dataclass(frozen=True) +class DevicesReply: + """The reply to a device query: one entry per device kind the server knows.""" + + request_id: str + devices: tuple[DeviceEntry, ...] + type: MessageType = dataclass_field(default=MessageType.DEVICES, init=False) + version: int = dataclass_field(default=PROTOCOL_VERSION, init=False) + + def __post_init__(self) -> None: + kinds = [d.device for d in self.devices] + if len(kinds) != len(set(kinds)): + raise ValueError("a device appears twice in the reply") + + def to_json(self) -> dict[str, Any]: + """The reply as a wire object.""" + return { + "type": int(self.type), + "version": self.version, + "request_id": self.request_id, + "devices": [d.to_json() for d in self.devices], + } + + @classmethod + def from_json(cls, data: Any) -> DevicesReply: + """Strict parse of a device reply.""" + _require_keys(data, {"type", "version", "request_id", "devices"}, "Devices") + _require_message_type(data, MessageType.DEVICES) + _require_version(data) + devices = data.get("devices") + if not isinstance(devices, list): + raise ProtocolError("Devices.devices must be a list") + try: + return cls( + request_id=_require_token(data.get("request_id"), "request_id"), + devices=tuple(DeviceEntry.from_json(d) for d in devices), + ) + except ValueError as exc: + raise ProtocolError(str(exc)) from None + + # A server-to-client JSON message, dispatched by its envelope's "type". -ServerJsonMessage = SessionInit | SessionStarting | SessionBegun | SessionStatus | SessionStopped | RequestRejected +ServerJsonMessage = ( + SessionInit | SessionStarting | SessionBegun | SessionStatus | SessionStopped | RequestRejected | DevicesReply +) _JSON_PARSERS_BY_TYPE: dict[MessageType, Any] = { MessageType.SESSION_INIT: SessionInit.from_json, @@ -1858,6 +2072,7 @@ def from_json(cls, data: Any) -> RequestRejected: MessageType.SESSION_STATUS: SessionStatus.from_json, MessageType.SESSION_STOPPED: SessionStopped.from_json, MessageType.REQUEST_REJECTED: RequestRejected.from_json, + MessageType.DEVICES: DevicesReply.from_json, } diff --git a/examples/egocap/capture_server/capture_server/server.py b/examples/egocap/capture_server/capture_server/server.py index 6a46e8c895..5865e77b7c 100644 --- a/examples/egocap/capture_server/capture_server/server.py +++ b/examples/egocap/capture_server/capture_server/server.py @@ -44,13 +44,17 @@ import signal import sys import time +import traceback +from concurrent.futures import ThreadPoolExecutor from dataclasses import dataclass +from pathlib import Path from typing import Any, Awaitable, Callable, TypeVar import websockets from websockets.exceptions import ConnectionClosed from websockets.server import ServerConnection +from . import mcal_catalog from . import protocol as p from .acquisition import Acquisition, CameraSourceError, SourceExited @@ -73,6 +77,11 @@ _CLOSE_TIMEOUT_S = 2.0 _SOURCE_RELEASE_TIMEOUT_S = 5.0 _COMPLETED_TOMBSTONE_CAPACITY = 16 +# A station query is one short exchange on its own connection; both the wait for the request and the directory +# scan are bounded so a stalled peer or a huge directory cannot hold a connection open. +_QUERY_DEADLINE_S = 10.0 +# Directory scans run off the event loop, at most this many at once. +_MCAL_SCANS = 2 class _StopDuringStartup(Exception): @@ -160,6 +169,10 @@ class ServerConfig: # prints the mode's default (the mock's SYNTHETIC-data warning, nothing for live); replay sets it so # recorded data is never announced as synthetic. data_note: str | None = None + # The folder of MCAL files this host offers: clients list it through the device query and select files from + # it by name at Begin. Anything placed in it (by hand or by a calibration tool) is offered; None disables + # listing and selection. + manus_calibration_dir: Path | None = None def __post_init__(self) -> None: """Reject one physical camera assigned to more than one logical name.""" @@ -1123,6 +1136,10 @@ def __init__( self._mode = p.CaptureMode.MOCK if self.config.mock else p.CaptureMode.LIVE self._source_factory = source_factory self._stop_event = asyncio.Event() + # Each scan holds a permit from its start until its thread really finishes, whatever happens to the + # request that asked for it, so timed-out queries can never accumulate live scans. + self._mcal_slots = asyncio.Semaphore(_MCAL_SCANS) + self._mcal_pool = ThreadPoolExecutor(max_workers=_MCAL_SCANS, thread_name_prefix="mcal-scan") self._ws_server: websockets.WebSocketServer | None = None self._owner: ServerConnection | None = None # The ONE connection-owned Outbox, constructed once per owner @@ -1233,6 +1250,8 @@ async def close(self) -> None: timeout=_SOURCE_RELEASE_TIMEOUT_S + _SEND_DEADLINE_S + 1.0, ) finally: + # Queued scans are dropped; a scan already running finishes on its own and its result is discarded. + self._mcal_pool.shutdown(wait=False, cancel_futures=True) if self._ws_server is not None: self._ws_server.close() await asyncio.wait_for(self._ws_server.wait_closed(), timeout=_CLOSE_TIMEOUT_S) @@ -1245,6 +1264,11 @@ async def _handle_connection(self, ws: ServerConnection) -> None: # which a brand new connection could otherwise slip in. await ws.close(code=1012, reason="server is shutting down") return + # Station queries are dispatched before owner admission: they never take or disturb the owner. + request_path = getattr(getattr(ws, "request", None), "path", "") + if request_path.split("?", 1)[0] == "/query": + await self._handle_query(ws) + return if self._owner is not None: await ws.close(code=1013, reason="owner connection already active") return @@ -1499,6 +1523,27 @@ async def _handle_begin(self, ws: ServerConnection, data: dict[str, Any]) -> Non ) return + # Only a Begin that would allocate a new session scans the folder, so a duplicate of an accepted Begin is + # replayed above without depending on a file that may have changed since. + refusal = await self._calibration_selection_refusal(begin) + if refusal is not None: + self._reject(begin.session_id, p.MessageType.BEGIN, p.RejectionReason.INVALID_CONFIG, refusal) + return + # The scan awaited: shutdown, the writer or admission may have changed since the checks above. + if self._stop_event.is_set() or self._outbox is None or self._outbox.closed: + self._reject( + begin.session_id, + p.MessageType.BEGIN, + p.RejectionReason.BUSY, + "server is shutting down" if self._stop_event.is_set() else "this connection is no longer usable", + ) + return + if self._active_session is not None: + self._reject( + begin.session_id, p.MessageType.BEGIN, p.RejectionReason.BUSY, "another session is already active" + ) + return + # required is a fixed station acquisition policy, independent of any # mock fault -- whether a camera's failure proves fatal is decided # here once per camera, and the source (mock or live) separately @@ -1528,6 +1573,111 @@ async def _handle_begin(self, ws: ServerConnection, data: dict[str, Any]) -> Non self._active_session = session asyncio.ensure_future(self._run_session(session)) + async def _scan_mcal(self) -> mcal_catalog.Catalog: + """The current MCAL folder listing, read on a bounded pool and never on the event loop. + + The permit is released by the scan's own completion, not by the caller, so a request that times out or + is cancelled cannot leave more than ``_MCAL_SCANS`` scans running. The caller waits at most + ``_QUERY_DEADLINE_S`` for a free slot and for the result, and gets UNAVAILABLE otherwise. + """ + directory = self.config.manus_calibration_dir + try: + async with asyncio.timeout(_QUERY_DEADLINE_S): + await self._mcal_slots.acquire() + try: + future = asyncio.get_running_loop().run_in_executor( + self._mcal_pool, mcal_catalog.scan, directory + ) + except BaseException: + self._mcal_slots.release() + raise + future.add_done_callback(self._mcal_scan_finished) + # Shielded: cancelling this wait must not cancel the job whose permit is still held. + return await asyncio.shield(future) + except TimeoutError: + return mcal_catalog.Catalog( + p.ManusCalibrationCatalogStatus.UNAVAILABLE, (), "the calibration directory scan timed out" + ) + except Exception: + print("capture-server: calibration directory scan failed", file=sys.stderr) + traceback.print_exc() + return mcal_catalog.Catalog( + p.ManusCalibrationCatalogStatus.UNAVAILABLE, (), "the calibration directory scan failed" + ) + + def _mcal_scan_finished(self, future: asyncio.Future[Any]) -> None: + """Free the scan's slot when its thread ends, and consume an abandoned job's exception.""" + self._mcal_slots.release() + if not future.cancelled(): + future.exception() + + async def _calibration_selection_refusal(self, begin: p.BeginRequest) -> str | None: + """Why the selected MCAL files cannot be used, checked before any hardware starts; None when they can. + + Uses the same policy as the device query, so a listed file is a selectable file, and a file that + disappeared or became invalid since it was listed is refused here rather than substituted. + """ + names = [ + name + for plugin in begin.plugins + if plugin.plugin == p.PluginType.MANUS and plugin.enabled + for name in (plugin.left_calibration, plugin.right_calibration) + if name is not None + ] + if not names: + return None + return mcal_catalog.selection_refusal(await self._scan_mcal(), names) + + async def _handle_query(self, ws: ServerConnection) -> None: + """One read-only device query on its own connection: no ownership, no session, one reply. + + Only a ``QueryDevices`` request is answered; anything else, including every capture command, closes the + connection. Whatever happens here never touches a running session. + """ + try: + raw = await asyncio.wait_for(ws.recv(), timeout=_QUERY_DEADLINE_S) + if isinstance(raw, bytes): + await ws.close(code=1002, reason="query must be a JSON text message") + return + try: + data = json.loads(raw) + except ValueError: + await ws.close(code=1002, reason="malformed JSON") + return + if not isinstance(data, dict) or data.get("type") != int(p.MessageType.QUERY_DEVICES): + await ws.close(code=1002, reason="unsupported query type") + return + try: + request = p.parse_query_devices(data) + except p.ProtocolError: + await ws.close(code=1002, reason="invalid QueryDevices request") + return + reply = p.DevicesReply(request.request_id, await self._device_entries()) + payload = json.dumps(reply.to_json()) + if len(payload.encode()) > p.MAX_QUERY_REPLY_BYTES: + # A valid error entry, never a truncated list or a message the client cannot receive. + reply = p.DevicesReply( + request.request_id, + ( + p.DeviceEntry( + p.PluginType.MANUS, + p.ManusCalibrationCatalogStatus.UNAVAILABLE, + error_message="the MCAL listing is too large to send in one reply", + ), + ), + ) + payload = json.dumps(reply.to_json()) + await asyncio.wait_for(ws.send(payload), timeout=_QUERY_DEADLINE_S) + await ws.close(code=1000) + except (asyncio.TimeoutError, ConnectionClosed): + # The peer stalled or left; there is no session to clean up and nothing else to report. + pass + + async def _device_entries(self) -> tuple[p.DeviceEntry, ...]: + """The devices this server offers; today the Manus plugin with its MCAL folder listing.""" + catalog = await self._scan_mcal() + return (p.DeviceEntry(p.PluginType.MANUS, catalog.status, catalog.files, catalog.error_message),) + async def _run_session(self, session: _Session) -> None: try: await session.run() @@ -1645,6 +1795,12 @@ def _build_arg_parser() -> argparse.ArgumentParser: metavar="NAME=DEVICE_ID", help="real only: station camera, logical NAME -> stable physical DEVICE_ID (OAK-D MXID); repeatable", ) + parser.add_argument( + "--manus-calibration-dir", + default=None, + metavar="DIR", + help="folder of MCAL files this host offers: listed by the device query and selectable by name at Begin", + ) parser.add_argument("--camera-width", type=int, default=1280, help="real only") parser.add_argument("--camera-height", type=int, default=720, help="real only") parser.add_argument("--camera-fps", type=float, default=30.0, help="real only") @@ -1655,6 +1811,11 @@ def _build_arg_parser() -> argparse.ArgumentParser: return parser +def _manus_calibration_dir(args: argparse.Namespace) -> Path | None: + """The ``--manus-calibration-dir`` folder, or None when the flag is absent.""" + return Path(args.manus_calibration_dir) if args.manus_calibration_dir else None + + def _station_cameras(args: argparse.Namespace) -> dict[str, StationCamera] | None: """The ``--camera NAME=DEVICE_ID`` entries sharing the global size/fps/quality; None when none given.""" if not args.camera: @@ -1711,6 +1872,7 @@ def _replay_setup(args: argparse.Namespace) -> tuple[ServerConfig, Any]: host=args.host, port=args.port, cameras=replay_station_cameras(episode), + manus_calibration_dir=_manus_calibration_dir(args), data_note=( f" -- REPLAY of recorded episode {episode.path} (streams: {', '.join(streams)}; loops every " f"{episode.period:.1f}s). Recorded data, not live hardware; reported to clients as MOCK mode." @@ -1761,6 +1923,7 @@ def _gmsl_setup(args: argparse.Namespace) -> tuple[ServerConfig, Any]: host=args.host, port=args.port, cameras=stations, + manus_calibration_dir=_manus_calibration_dir(args), data_note=( " -- GMSL cameras from the frame feed at " f"{args.gmsl_ipc}: " @@ -1770,7 +1933,9 @@ def _gmsl_setup(args: argparse.Namespace) -> tuple[ServerConfig, Any]: ) ), ) - return config, build_gmsl_source_factory(args.gmsl_ipc, cameras, stations) + return config, build_gmsl_source_factory( + args.gmsl_ipc, cameras, stations, calibration_dir=config.manus_calibration_dir + ) async def _run_cli(args: argparse.Namespace) -> int: @@ -1788,6 +1953,7 @@ async def _run_cli(args: argparse.Namespace) -> int: port=args.port, connect_delay=args.connect_delay, cameras=_station_cameras(args), + manus_calibration_dir=_manus_calibration_dir(args), ) # The one place the implementation is chosen, before any session is accepted. source_factory = build_source_factory( diff --git a/examples/egocap/tests/capture_server/test_device_query.py b/examples/egocap/tests/capture_server/test_device_query.py new file mode 100644 index 0000000000..e1265cc5b4 --- /dev/null +++ b/examples/egocap/tests/capture_server/test_device_query.py @@ -0,0 +1,584 @@ +# SPDX-FileCopyrightText: 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# +# SPDX-License-Identifier: Apache-2.0 + +"""The device query: protocol, folder scan, the read-only ``/query`` route, Begin selection and the client.""" + +from __future__ import annotations + +import asyncio +import json +from pathlib import Path + +import pytest +import websockets + +from capture_server import mcal_catalog +from capture_server import protocol as p +from capture_server.acquisition.isaac_capture import IsaacCaptureAdapter, TrackingSelection +from capture_server.client import CaptureClient, DeviceQueryError, ManusPluginConfig, RequestRejectedError +from capture_server.launch import _calibration_path +from capture_server.server import ServerConfig, _build_arg_parser +from tests.support.capture_server import start_capture_server, stop_capture_server + +STATUS = p.ManusCalibrationCatalogStatus + + +def _mcal(directory: Path, name: str, data: bytes = b"calibration") -> Path: + """Write one MCAL file into ``directory``, creating it.""" + directory.mkdir(parents=True, exist_ok=True) + path = directory / name + path.write_bytes(data) + return path + + +# -- protocol ----------------------------------------------------------------------------------------------- + + +def test_query_request_round_trips(): + request = p.QueryDevicesRequest(request_id="r-1") + assert p.parse_query_devices(request.to_json()) == request + + +@pytest.mark.parametrize( + "mutate", + [ + lambda d: d.update(type=1), + lambda d: d.update(version=99), + lambda d: d.update(request_id=""), + lambda d: d.update(request_id=7), + lambda d: d.update(extra=1), + lambda d: d.pop("request_id"), + ], +) +def test_query_request_rejects_malformed_envelopes(mutate): + data = p.QueryDevicesRequest(request_id="r-1").to_json() + mutate(data) + with pytest.raises(p.ProtocolError): + p.parse_query_devices(data) + + +def test_devices_reply_round_trips_through_the_server_message_parser(): + entry = p.DeviceEntry(p.PluginType.MANUS, STATUS.READY, ("a.mcal", "b.mcal")) + reply = p.DevicesReply("r-1", (entry,)) + assert p.parse_server_message(json.loads(json.dumps(reply.to_json()))) == reply + + +@pytest.mark.parametrize( + "kwargs", + [ + dict(mcal_status=STATUS.UNAVAILABLE), + dict(mcal_status=STATUS.READY, error_message="boom"), + dict(mcal_status=STATUS.UNAVAILABLE, error_message=""), + dict(mcal_status=STATUS.DISABLED, mcal_files=("a.mcal",)), + dict(mcal_status=STATUS.READY, mcal_files=("a.mcal", "a.mcal")), + dict(mcal_status=STATUS.READY, mcal_files=("../a.mcal",)), + ], +) +def test_device_entry_rejects_inconsistent_listings(kwargs): + with pytest.raises(ValueError): + p.DeviceEntry(p.PluginType.MANUS, **kwargs) + + +def test_device_entry_rejects_too_many_files(monkeypatch): + monkeypatch.setattr(p, "MAX_MCAL_FILES", 1) + with pytest.raises(ValueError, match="at most"): + p.DeviceEntry(p.PluginType.MANUS, STATUS.READY, ("a.mcal", "b.mcal")) + + +def test_devices_reply_rejects_a_repeated_device(): + entry = p.DeviceEntry(p.PluginType.MANUS, STATUS.DISABLED) + with pytest.raises(ValueError, match="twice"): + p.DevicesReply("r-1", (entry, entry)) + + +@pytest.mark.parametrize( + "mutate", + [ + lambda d: d.update(devices="x"), + lambda d: d["devices"][0].update(mcal_files="x"), + lambda d: d["devices"][0].update(mcal_files=[1]), + lambda d: d["devices"][0].update(error_message=3), + lambda d: d["devices"][0].update(mcal_status=9), + lambda d: d["devices"][0].update(device=9), + lambda d: d["devices"][0].update(mcal_files=["nope.txt"]), + lambda d: d["devices"][0].update(error_message="x"), + lambda d: d["devices"][0].update(extra=1), + ], +) +def test_devices_reply_rejects_malformed_replies(mutate): + data = p.DevicesReply("r-1", (p.DeviceEntry(p.PluginType.MANUS, STATUS.READY, ("a.mcal",)),)).to_json() + mutate(data) + with pytest.raises(p.ProtocolError): + p.parse_server_message(data) + + +def test_plugin_request_carries_calibration_names_through_the_wire(): + begin = p.BeginRequest( + tracking=p.TrackingConfig(sources=(p.SourceType.HEAD,)), + plugins=( + p.PluginRequest(plugin=p.PluginType.MANUS, left_calibration="l.mcal", right_calibration="r.mcal"), + ), + session_id="s-1", + ) + config = p.effective_config(begin) + assert config.plugins[0].left_calibration == "l.mcal" + wire = json.loads(json.dumps(config.to_json())) + assert p.CaptureConfiguration.from_json(wire) == config + assert p._parse_plugin_request(wire["plugins"][0]).right_calibration == "r.mcal" + + +def test_plugin_request_without_files_keeps_its_earlier_wire_form(): + config = p.effective_config( + p.BeginRequest( + tracking=p.TrackingConfig(sources=(p.SourceType.HEAD,)), + plugins=(p.PluginRequest(plugin=p.PluginType.MANUS),), + ) + ) + assert set(config.to_json()["plugins"][0]) == {"plugin", "enabled", "required", "datasets"} + + +@pytest.mark.parametrize("name", ["../x.mcal", "dir/x.mcal", ".hidden.mcal", "x.txt", "", "x" * 200 + ".mcal"]) +def test_plugin_request_rejects_names_that_are_not_basenames(name): + with pytest.raises(ValueError): + p.PluginRequest(plugin=p.PluginType.MANUS, left_calibration=name) + with pytest.raises(p.ProtocolError): + p._parse_plugin_request({"plugin": 1, "left_calibration": name}) + + +def test_plugin_request_rejects_one_file_for_both_hands(): + with pytest.raises(ValueError, match="different files"): + p.PluginRequest(plugin=p.PluginType.MANUS, left_calibration="x.mcal", right_calibration="x.mcal") + with pytest.raises(p.ProtocolError, match="different files"): + p._parse_plugin_request({"plugin": 1, "left_calibration": "x.mcal", "right_calibration": "x.mcal"}) + + +# -- folder scan -------------------------------------------------------------------------------------------- + + +def test_scan_without_a_directory_is_disabled(): + catalog = mcal_catalog.scan(None) + assert (catalog.status, catalog.files) == (STATUS.DISABLED, ()) + + +def test_scan_of_a_missing_directory_is_an_empty_ready_listing(tmp_path): + catalog = mcal_catalog.scan(tmp_path / "absent") + assert (catalog.status, catalog.files) == (STATUS.READY, ()) + + +def test_scan_offers_only_nonempty_readable_regular_mcal_files_sorted(tmp_path): + _mcal(tmp_path, "b.mcal") + _mcal(tmp_path, "a.mcal") + _mcal(tmp_path, "empty.mcal", b"") + _mcal(tmp_path, "notes.txt") + (tmp_path / "subdir.mcal").mkdir() + (tmp_path / "link.mcal").symlink_to(tmp_path / "a.mcal") + unreadable = _mcal(tmp_path, "locked.mcal") + unreadable.chmod(0) + try: + catalog = mcal_catalog.scan(tmp_path) + finally: + unreadable.chmod(0o600) + assert catalog.status is STATUS.READY + # The unreadable file is only hidden when the tests do not run as root, which ignores file modes. + expected = ( + ["a.mcal", "b.mcal"] + if unreadable.stat().st_uid != 0 and not _is_root() + else ["a.mcal", "b.mcal", "locked.mcal"] + ) + assert list(catalog.files) == expected + + +def _is_root() -> bool: + import os + + return os.geteuid() == 0 + + +def test_scan_of_a_file_instead_of_a_directory_is_unavailable(tmp_path): + path = tmp_path / "file" + path.write_text("x") + catalog = mcal_catalog.scan(path) + assert catalog.status is STATUS.UNAVAILABLE and "not a directory" in catalog.error_message + + +def test_scan_reports_an_unreadable_directory(tmp_path, monkeypatch): + def deny(self, *args, **kwargs): + raise PermissionError + + monkeypatch.setattr(Path, "resolve", deny) + catalog = mcal_catalog.scan(tmp_path) + assert catalog.status is STATUS.UNAVAILABLE and "permission denied" in catalog.error_message + + +def test_scan_reports_other_io_errors(tmp_path, monkeypatch): + def fail(self, *args, **kwargs): + raise OSError(5, "boom") + + monkeypatch.setattr(Path, "resolve", fail) + catalog = mcal_catalog.scan(tmp_path) + assert catalog.status is STATUS.UNAVAILABLE and "boom" in catalog.error_message + + +def test_scan_over_the_limit_is_an_error_not_a_truncated_list(tmp_path, monkeypatch): + monkeypatch.setattr(p, "MAX_MCAL_FILES", 2) + for name in ("a.mcal", "b.mcal", "c.mcal"): + _mcal(tmp_path, name) + catalog = mcal_catalog.scan(tmp_path) + assert (catalog.status, catalog.files) == (STATUS.UNAVAILABLE, ()) + + +def test_scan_skips_an_entry_that_vanishes_mid_scan(tmp_path, monkeypatch): + _mcal(tmp_path, "a.mcal") + real_stat = Path.stat + + def flaky(self, *args, **kwargs): + return real_stat(self, *args, **kwargs) + + import os + + real_scandir = os.scandir + + class _Gone: + name = "gone.mcal" + path = str(tmp_path / "gone.mcal") + + def is_symlink(self): + raise FileNotFoundError + + class _Listing: + def __enter__(self): + return iter([*real_scandir(tmp_path), _Gone()]) + + def __exit__(self, *exc): + return False + + monkeypatch.setattr(mcal_catalog.os, "scandir", lambda root: _Listing()) + monkeypatch.setattr(Path, "stat", flaky) + assert list(mcal_catalog.scan(tmp_path).files) == ["a.mcal"] + + +def test_selection_refusal_explains_each_case(tmp_path): + _mcal(tmp_path, "a.mcal") + ready = mcal_catalog.scan(tmp_path) + assert mcal_catalog.selection_refusal(ready, ["a.mcal"]) is None + assert "not an available file" in mcal_catalog.selection_refusal(ready, ["a.mcal", "b.mcal"]) + assert "--manus-calibration-dir" in mcal_catalog.selection_refusal(mcal_catalog.scan(None), ["a.mcal"]) + broken = mcal_catalog.Catalog(STATUS.UNAVAILABLE, (), "disk on fire") + assert "disk on fire" in mcal_catalog.selection_refusal(broken, ["a.mcal"]) + + +def test_resolve_joins_the_name_onto_the_real_directory(tmp_path): + assert mcal_catalog.resolve(tmp_path, "a.mcal") == tmp_path.resolve() / "a.mcal" + + +# -- launch and adapter ------------------------------------------------------------------------------------- + + +def test_calibration_path_resolves_inside_the_folder_or_keeps_an_inert_name(tmp_path): + assert _calibration_path(None, tmp_path) is None + assert _calibration_path("a.mcal", None) == "a.mcal" + assert _calibration_path("a.mcal", tmp_path) == str(tmp_path.resolve() / "a.mcal") + + +def test_adapter_passes_selected_files_to_the_manus_plugin(): + def args(**kwargs): + adapter = IsaacCaptureAdapter.__new__(IsaacCaptureAdapter) + adapter._selection = TrackingSelection( + head=False, controller_left=False, controller_right=False, hand_left=False, hand_right=False, **kwargs + ) + return adapter._manus_args() + + assert args(manus_enabled=True) == [] + assert args(manus_enabled=True, manus_left_calibration="/d/l.mcal", manus_right_calibration="/d/r.mcal") == [ + "--left-calibration=/d/l.mcal", + "--right-calibration=/d/r.mcal", + ] + + +# -- CLI ---------------------------------------------------------------------------------------------------- + + +def test_cli_flag_defaults_to_no_folder_and_round_trips(): + parser = _build_arg_parser() + assert parser.parse_args(["--mock"]).manus_calibration_dir is None + assert parser.parse_args(["--mock", "--manus-calibration-dir", "/mcal"]).manus_calibration_dir == "/mcal" + + +# -- the /query route and Begin selection, against a running server ------------------------------------------ + + +async def _ask(url: str, request: dict | str | bytes, *, request_id: str = "q-1"): + """Send one message to /query; returns (reply or None, close code).""" + ws = await websockets.connect(f"{url}/query") + try: + await ws.send(request if isinstance(request, (str, bytes)) else json.dumps(request)) + reply = None + try: + reply = p.parse_server_message(json.loads(await asyncio.wait_for(ws.recv(), 5))) + except websockets.ConnectionClosed: + pass + await asyncio.wait_for(ws.wait_closed(), 5) + return reply, ws.close_code + finally: + await ws.close() + + +def _query(request_id: str = "q-1") -> dict: + return p.QueryDevicesRequest(request_id=request_id).to_json() + + +@pytest.fixture +async def station(unused_tcp_port, tmp_path): + """A running mock server with a managed MCAL folder that already holds two files.""" + folder = tmp_path / "mcal" + _mcal(folder, "left.mcal") + _mcal(folder, "right.mcal") + server = await start_capture_server(unused_tcp_port, config=ServerConfig(manus_calibration_dir=folder)) + try: + yield f"ws://127.0.0.1:{unused_tcp_port}", folder, server + finally: + await stop_capture_server(server) + + +async def test_query_lists_the_manus_device_with_its_mcal_files(station): + url, folder, _ = station + _mcal(folder, "later.mcal") + reply, code = await _ask(url, _query("echo-me")) + assert code == 1000 + assert reply.request_id == "echo-me" + (entry,) = reply.devices + assert (entry.device, entry.mcal_status) == (p.PluginType.MANUS, STATUS.READY) + assert entry.mcal_files == ("later.mcal", "left.mcal", "right.mcal") + + +async def test_query_without_a_folder_reports_disabled(unused_tcp_port): + server = await start_capture_server(unused_tcp_port) + try: + reply, _ = await _ask(f"ws://127.0.0.1:{unused_tcp_port}", _query()) + finally: + await stop_capture_server(server) + assert (reply.devices[0].mcal_status, reply.devices[0].mcal_files) == (STATUS.DISABLED, ()) + + +async def test_query_reports_an_unusable_folder_as_unavailable(station): + url, folder, _ = station + for child in folder.iterdir(): + child.unlink() + folder.rmdir() + folder.write_text("not a directory") + reply, _ = await _ask(url, _query()) + assert reply.devices[0].mcal_status is STATUS.UNAVAILABLE and reply.devices[0].error_message + + +@pytest.mark.parametrize( + "message", + [ + b"binary", + "not json", + "[]", + json.dumps({"type": int(p.MessageType.BEGIN), "version": 1, "session_id": "s"}), + json.dumps({"type": int(p.MessageType.STOP), "version": 1, "session_id": "s"}), + json.dumps({"type": int(p.MessageType.QUERY_DEVICES), "version": 1}), + ], +) +async def test_query_route_closes_on_anything_but_a_valid_device_query(station, message): + url, _, _ = station + reply, code = await _ask(url, message) + assert reply is None and code == 1002 + + +async def test_query_route_closes_when_the_peer_sends_nothing(station, monkeypatch): + from capture_server import server as server_module + + monkeypatch.setattr(server_module, "_QUERY_DEADLINE_S", 0.2) + url, _, _ = station + ws = await websockets.connect(f"{url}/query") + await asyncio.wait_for(ws.wait_closed(), 5) + + +async def test_query_during_an_active_capture_does_not_disturb_it(station): + url, _, server = station + client = CaptureClient(url) + await client.start(sources=(p.SourceType.HEAD,)) + try: + reply, code = await _ask(url, _query()) + assert code == 1000 and reply.devices[0].mcal_status is STATUS.READY + assert server.server._active_session is not None and not server.server._active_session.done.is_set() + finally: + await client.stop() + + +async def test_oversized_reply_becomes_an_unavailable_entry(station, monkeypatch): + url, _, _ = station + monkeypatch.setattr(p, "MAX_QUERY_REPLY_BYTES", 10) + reply, code = await _ask(url, _query()) + assert code == 1000 + assert reply.devices[0].mcal_status is STATUS.UNAVAILABLE and "too large" in reply.devices[0].error_message + + +async def test_a_scan_that_times_out_or_fails_is_unavailable(station, monkeypatch): + url, _, _ = station + monkeypatch.setattr(mcal_catalog, "scan", lambda directory: (_ for _ in ()).throw(RuntimeError("boom"))) + reply, _ = await _ask(url, _query()) + assert reply.devices[0].mcal_status is STATUS.UNAVAILABLE and "failed" in reply.devices[0].error_message + + +async def test_scan_timeout_is_reported_without_leaking_the_slot(station, monkeypatch): + from capture_server import server as server_module + + url, _, server = station + release = asyncio.Event() + import threading + + gate = threading.Event() + monkeypatch.setattr( + mcal_catalog, "scan", lambda directory: (gate.wait(5), mcal_catalog.Catalog(STATUS.READY))[1] + ) + monkeypatch.setattr(server_module, "_QUERY_DEADLINE_S", 0.2) + reply, _ = await _ask(url, _query()) + assert "timed out" in reply.devices[0].error_message + gate.set() + release.set() + for _ in range(100): + if server.server._mcal_slots._value == server_module._MCAL_SCANS: + break + await asyncio.sleep(0.02) + assert server.server._mcal_slots._value == server_module._MCAL_SCANS + + +async def test_begin_accepts_files_that_are_in_the_folder_and_echoes_the_names(station): + url, _, _ = station + client = CaptureClient(url) + started = asyncio.Event() + seen = [] + await client.start( + sources=(p.SourceType.HEAD,), + plugins=(ManusPluginConfig(left_calibration="left.mcal", right_calibration="right.mcal"),), + on_started=lambda s: (seen.append(s), started.set()), + ) + try: + await asyncio.wait_for(started.wait(), 5) + plugin = seen[0].effective_config.plugins[0] + assert (plugin.left_calibration, plugin.right_calibration) == ("left.mcal", "right.mcal") + finally: + await client.stop() + + +async def test_begin_refuses_a_file_that_is_not_in_the_folder(station): + url, _, _ = station + with pytest.raises(RequestRejectedError) as excinfo: + await CaptureClient(url).start( + sources=(p.SourceType.HEAD,), plugins=(ManusPluginConfig(left_calibration="missing.mcal"),) + ) + assert excinfo.value.reason == p.RejectionReason.INVALID_CONFIG + assert "not an available file" in excinfo.value.error_message + + +async def test_begin_refuses_a_file_that_became_a_link_or_empty_since_listing(station): + url, folder, _ = station + (folder / "left.mcal").write_bytes(b"") + with pytest.raises(RequestRejectedError): + await CaptureClient(url).start( + sources=(p.SourceType.HEAD,), plugins=(ManusPluginConfig(left_calibration="left.mcal"),) + ) + + +async def test_begin_with_files_but_no_folder_is_refused(unused_tcp_port): + server = await start_capture_server(unused_tcp_port) + try: + with pytest.raises(RequestRejectedError) as excinfo: + await CaptureClient(f"ws://127.0.0.1:{unused_tcp_port}").start( + sources=(p.SourceType.HEAD,), plugins=(ManusPluginConfig(left_calibration="x.mcal"),) + ) + finally: + await stop_capture_server(server) + assert "--manus-calibration-dir" in excinfo.value.error_message + + +# -- client ------------------------------------------------------------------------------------------------- + + +async def test_client_query_devices_returns_the_listing(station): + url, _, _ = station + (entry,) = await CaptureClient(url).query_devices() + assert entry.mcal_files == ("left.mcal", "right.mcal") + + +async def test_client_query_devices_works_while_the_same_client_is_capturing(station): + url, _, _ = station + client = CaptureClient(url) + await client.start(sources=(p.SourceType.HEAD,)) + try: + assert (await client.query_devices())[0].mcal_status is STATUS.READY + finally: + await client.stop() + + +def test_manus_plugin_config_validates_names(): + with pytest.raises(ValueError, match="basename"): + ManusPluginConfig(left_calibration="../x.mcal") + with pytest.raises(ValueError, match="different files"): + ManusPluginConfig(left_calibration="x.mcal", right_calibration="x.mcal") + assert ManusPluginConfig().plugin is p.PluginType.MANUS + + +async def _fake_server(handler, port): + return await websockets.serve(handler, "127.0.0.1", port) + + +async def test_client_reports_a_server_that_does_not_know_the_query(unused_tcp_port): + async def handler(ws): + await ws.recv() + await ws.close(code=1013, reason="owner connection already active") + + server = await _fake_server(handler, unused_tcp_port) + try: + with pytest.raises(DeviceQueryError, match="close code 1013"): + await CaptureClient(f"ws://127.0.0.1:{unused_tcp_port}").query_devices() + finally: + server.close() + await server.wait_closed() + + +@pytest.mark.parametrize( + ("reply", "message"), + [ + (b"binary", "binary"), + ("not json", "invalid"), + (json.dumps({"type": 1}), "invalid"), + (None, "does not answer"), + ], +) +async def test_client_never_accepts_a_wrong_reply_as_an_answer(unused_tcp_port, reply, message): + async def handler(ws): + await ws.recv() + # A well-formed reply for some other request id must not be accepted as this query's answer. + await ws.send(reply if reply is not None else json.dumps(p.DevicesReply("someone-else", ()).to_json())) + await ws.close() + + server = await _fake_server(handler, unused_tcp_port) + try: + with pytest.raises(DeviceQueryError, match=message): + await CaptureClient(f"ws://127.0.0.1:{unused_tcp_port}").query_devices() + finally: + server.close() + await server.wait_closed() + + +async def test_client_reports_an_unreachable_server_and_a_timeout(unused_tcp_port, monkeypatch): + from capture_server import client as client_module + + with pytest.raises(DeviceQueryError, match="failed"): + await CaptureClient(f"ws://127.0.0.1:{unused_tcp_port}").query_devices() + + async def silent(ws): + await asyncio.sleep(5) + + monkeypatch.setattr(client_module, "_QUERY_TIMEOUT_S", 0.2) + server = await _fake_server(silent, unused_tcp_port) + try: + with pytest.raises(DeviceQueryError): + await CaptureClient(f"ws://127.0.0.1:{unused_tcp_port}").query_devices() + finally: + server.close() diff --git a/examples/egocap/tests/capture_server/test_protocol.py b/examples/egocap/tests/capture_server/test_protocol.py index b721e2cd35..37bdc561c9 100644 --- a/examples/egocap/tests/capture_server/test_protocol.py +++ b/examples/egocap/tests/capture_server/test_protocol.py @@ -942,7 +942,7 @@ def test_request_rejected_rejects_empty_error_message(): def test_message_type_wire_values(): - assert [int(m) for m in p.MessageType] == [1, 2, 3, 4, 5, 6, 7, 8, 9, 10] + assert [int(m) for m in p.MessageType] == [1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12] def test_status_reason_wire_values_are_0_through_23(): From 8d0c061d703a905e0df0e2e45403d002f4c59e17 Mon Sep 17 00:00:00 2001 From: Gareth Morgan Date: Thu, 5 Nov 2026 13:30:13 -0800 Subject: [PATCH 2/6] examples/egocap: address code-quality findings in MCAL scan Narrow the permit-release handler to Exception and document the intentional pass in the query test helper. Co-Authored-By: Claude Sonnet 5.5 Signed-off-by: Gareth Morgan --- examples/egocap/capture_server/capture_server/server.py | 3 ++- examples/egocap/tests/capture_server/test_device_query.py | 2 +- 2 files changed, 3 insertions(+), 2 deletions(-) diff --git a/examples/egocap/capture_server/capture_server/server.py b/examples/egocap/capture_server/capture_server/server.py index 5865e77b7c..c14f5ecf7c 100644 --- a/examples/egocap/capture_server/capture_server/server.py +++ b/examples/egocap/capture_server/capture_server/server.py @@ -1588,7 +1588,8 @@ async def _scan_mcal(self) -> mcal_catalog.Catalog: future = asyncio.get_running_loop().run_in_executor( self._mcal_pool, mcal_catalog.scan, directory ) - except BaseException: + except Exception: + # run_in_executor is synchronous and has no await to cancel at; only a submit failure lands here. self._mcal_slots.release() raise future.add_done_callback(self._mcal_scan_finished) diff --git a/examples/egocap/tests/capture_server/test_device_query.py b/examples/egocap/tests/capture_server/test_device_query.py index e1265cc5b4..4d39bbc2c3 100644 --- a/examples/egocap/tests/capture_server/test_device_query.py +++ b/examples/egocap/tests/capture_server/test_device_query.py @@ -317,7 +317,7 @@ async def _ask(url: str, request: dict | str | bytes, *, request_id: str = "q-1" try: reply = p.parse_server_message(json.loads(await asyncio.wait_for(ws.recv(), 5))) except websockets.ConnectionClosed: - pass + pass # a rejected request is closed without a reply; the close code is what the caller checks await asyncio.wait_for(ws.wait_closed(), 5) return reply, ws.close_code finally: From 0f0a6c487d68dd97207c4f1f5ff149504e4f9887 Mon Sep 17 00:00:00 2001 From: Gareth Morgan Date: Thu, 5 Nov 2026 13:59:37 -0800 Subject: [PATCH 3/6] examples/egocap: shorten comment to satisfy ruff E501 Co-Authored-By: Claude Sonnet 5.5 Signed-off-by: Gareth Morgan --- examples/egocap/capture_server/capture_server/server.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/examples/egocap/capture_server/capture_server/server.py b/examples/egocap/capture_server/capture_server/server.py index c14f5ecf7c..bb99d3f994 100644 --- a/examples/egocap/capture_server/capture_server/server.py +++ b/examples/egocap/capture_server/capture_server/server.py @@ -1589,7 +1589,7 @@ async def _scan_mcal(self) -> mcal_catalog.Catalog: self._mcal_pool, mcal_catalog.scan, directory ) except Exception: - # run_in_executor is synchronous and has no await to cancel at; only a submit failure lands here. + # run_in_executor never awaits, so only a submit failure lands here. self._mcal_slots.release() raise future.add_done_callback(self._mcal_scan_finished) From c09cf33260347564c704130819ad2f722c67fa9c Mon Sep 17 00:00:00 2001 From: Gareth Morgan Date: Thu, 5 Nov 2026 14:43:36 -0800 Subject: [PATCH 4/6] examples/egocap: address review findings in the device query Log the scan failure through the isaaccapture logger tree, document the new protocol symbols and their validation, drop the unused HandSide enum, and reconcile the README envelope and public API sections. Co-Authored-By: Claude Sonnet 5.5 Signed-off-by: Gareth Morgan --- examples/egocap/capture_server/README.md | 12 ++++++++---- .../capture_server/capture_server/protocol.py | 13 +++++++------ .../egocap/capture_server/capture_server/server.py | 7 ++++--- 3 files changed, 19 insertions(+), 13 deletions(-) diff --git a/examples/egocap/capture_server/README.md b/examples/egocap/capture_server/README.md index f981f76196..d6dd225124 100644 --- a/examples/egocap/capture_server/README.md +++ b/examples/egocap/capture_server/README.md @@ -121,7 +121,7 @@ async def start( *, sources: Sequence[SourceType], cameras: Sequence[CameraSelection] = (), - plugins: Sequence[PluginType] = (), + plugins: Sequence[PluginType | ManusPluginConfig] = (), on_started: Callback[CaptureStarted] | None = None, on_status: Callback[CaptureStatus] | None = None, on_capture_frame: Callback[CaptureFrame] | None = None, @@ -134,7 +134,11 @@ async def start( async def stop(self) -> None: ... ``` -`Callback[T]` means `Callable[[T], None | Awaitable[None]]`: a synchronous or asynchronous function taking one value of type T. All start arguments are keyword-only. The client exposes two operations, start and stop. +```python +async def query_devices(self) -> tuple[protocol.DeviceEntry, ...]: ... +``` + +`Callback[T]` means `Callable[[T], None | Awaitable[None]]`: a synchronous or asynchronous function taking one value of type T. All start arguments are keyword-only. The client exposes three operations: start, stop and the read-only device query. `ManusPluginConfig` and `DeviceQueryError` are imported from `capture_server.client`; `DeviceEntry` and `ManusCalibrationCatalogStatus` are in `capture_server.protocol`. #### Constructor and start arguments @@ -143,7 +147,7 @@ async def stop(self) -> None: ... | Constructor `url` | Nonempty `str`, a `ws://` or `wss://` station URL. | Identifies the remote capture server. Construction creates an idle client without connecting. | | `sources` | Sequence of unique `SourceType`: `HEAD=1`, `CONTROLLERS=2`, `HANDS=3`. Required; an empty sequence is allowed. | Chooses tracking outputs. CONTROLLERS selects both controller sides; HANDS selects both hand sides. An empty sequence supports camera-only capture. | | `cameras` | Sequence of `CameraSelection`, default empty; at most three, with unique names. | Selects station-configured cameras by logical name and type. | -| `plugins` | Sequence of unique `PluginType`: `MANUS=1`; default empty. | Enables these tracking plugins. Manus uses its HUMAN dataset and is optional: missing plugin discovery does not itself require session failure. This selection does not require connected gloves or filter hands by vendor. | +| `plugins` | Sequence of unique plugins: a `PluginType` (`MANUS=1`) or a `ManusPluginConfig` that also names MCAL files; default empty. | Enables these tracking plugins. Manus uses its HUMAN dataset and is optional: missing plugin discovery does not itself require session failure. This selection does not require connected gloves or filter hands by vendor. | | Five `on_...` arguments | `Callback[T] \| None`, default None; argument types appear below. | Registers notifications. None discards that notification; it does not disable acquisition. | An empty `sources` sequence together with any enabled plugin is rejected as `INVALID_CONFIG`: an enabled plugin needs IC actually entered for discovery to run, but the real IC graph-construction API rejects an empty output mapping outright, so that combination has no valid native graph to build. Plugin requests are otherwise independent of output subscriptions — a session can enable Manus without selecting `HANDS`, and vice versa; only the specific zero-sources-with-an-enabled-plugin combination is rejected, not any particular source choice. Camera-only capture (empty `sources`, no enabled plugin) remains supported and never enters IC. @@ -694,7 +698,7 @@ One JSON object is one text message. One complete header and payload is one bina There are ten session message families, plus the device query described below. SessionInit describes the resolved session configuration and complete decoding layout before hardware initialization. SessionStarting reports progress and observed availability. SessionStatus carries IC's changing provider/device state without treating absence as terminal or confusing status with pose validity. No application recording command or file path is accepted. -Every JSON message has these envelope fields. An envelope is the small common part of a message that allows the receiver to route it before reading message-specific content. +Every JSON session message has these envelope fields. The two `/query` messages instead carry `type`, `version` and a client-chosen `request_id`, and no `session_id`; a query message that carries one is rejected. An envelope is the small common part of a message that allows the receiver to route it before reading message-specific content. | Field | Allowed contents | What it does | |---|---|---| diff --git a/examples/egocap/capture_server/capture_server/protocol.py b/examples/egocap/capture_server/capture_server/protocol.py index 010febde5d..8e6adfd2de 100644 --- a/examples/egocap/capture_server/capture_server/protocol.py +++ b/examples/egocap/capture_server/capture_server/protocol.py @@ -90,13 +90,9 @@ class ManusDataset(IntEnum): HUMAN = 1 -class HandSide(IntEnum): - LEFT = 1 - RIGHT = 2 - - class ManusCalibrationCatalogStatus(IntEnum): - # Whether the Manus MCAL folder listing is usable. + """Whether the Manus MCAL folder listing is usable.""" + READY = 1 DISABLED = 2 UNAVAILABLE = 3 @@ -1974,11 +1970,14 @@ class DeviceEntry: error_message: str | None = None def __post_init__(self) -> None: + """Enforce that the status, error message and file list agree with each other.""" + # The error message exists exactly when the listing is unavailable, and is never empty. unavailable = self.mcal_status is ManusCalibrationCatalogStatus.UNAVAILABLE if unavailable != (self.error_message is not None): raise ValueError("error_message is present exactly when the MCAL listing is UNAVAILABLE") if self.error_message is not None and not self.error_message: raise ValueError("error_message must be nonempty or null") + # Only a READY listing carries files, and they must fit one reply as unique basenames. if self.mcal_status is not ManusCalibrationCatalogStatus.READY and self.mcal_files: raise ValueError("mcal_files are present only for a READY listing") if len(self.mcal_files) > MAX_MCAL_FILES: @@ -2029,6 +2028,8 @@ class DevicesReply: version: int = dataclass_field(default=PROTOCOL_VERSION, init=False) def __post_init__(self) -> None: + """Reject a reply that lists the same device kind twice.""" + # One entry per device kind keeps a client's lookup unambiguous. kinds = [d.device for d in self.devices] if len(kinds) != len(set(kinds)): raise ValueError("a device appears twice in the reply") diff --git a/examples/egocap/capture_server/capture_server/server.py b/examples/egocap/capture_server/capture_server/server.py index bb99d3f994..93a9de8967 100644 --- a/examples/egocap/capture_server/capture_server/server.py +++ b/examples/egocap/capture_server/capture_server/server.py @@ -41,10 +41,10 @@ import collections import contextlib import json +import logging import signal import sys import time -import traceback from concurrent.futures import ThreadPoolExecutor from dataclasses import dataclass from pathlib import Path @@ -58,6 +58,8 @@ from . import protocol as p from .acquisition import Acquisition, CameraSourceError, SourceExited +_LOG = logging.getLogger("isaaccapture.capture_server.server") + __all__ = ["CaptureServer", "ServerConfig", "main"] # Inbound (client -> server) messages are only ever small Begin/Stop JSON, @@ -1600,8 +1602,7 @@ async def _scan_mcal(self) -> mcal_catalog.Catalog: p.ManusCalibrationCatalogStatus.UNAVAILABLE, (), "the calibration directory scan timed out" ) except Exception: - print("capture-server: calibration directory scan failed", file=sys.stderr) - traceback.print_exc() + _LOG.exception("calibration directory scan failed") return mcal_catalog.Catalog( p.ManusCalibrationCatalogStatus.UNAVAILABLE, (), "the calibration directory scan failed" ) From dfe06f337a1ede7deb49ec07f9d7193a2db1f500 Mon Sep 17 00:00:00 2001 From: Gareth Morgan Date: Thu, 5 Nov 2026 14:44:33 -0800 Subject: [PATCH 5/6] examples/egocap: settle query handlers on shutdown and resolve the MCAL folder once close() cancels in-flight /query handlers so a slow scan cannot hold wait_closed() past its deadline. ServerConfig resolves the MCAL folder at construction, so Begin no longer touches the filesystem to build a path. Co-Authored-By: Claude Sonnet 5.5 Signed-off-by: Gareth Morgan --- .../capture_server/mcal_catalog.py | 4 +- .../capture_server/capture_server/server.py | 19 +++++++++- .../tests/capture_server/test_device_query.py | 38 ++++++++++++++++++- 3 files changed, 55 insertions(+), 6 deletions(-) diff --git a/examples/egocap/capture_server/capture_server/mcal_catalog.py b/examples/egocap/capture_server/capture_server/mcal_catalog.py index af5b7c2cea..d03eba16fd 100644 --- a/examples/egocap/capture_server/capture_server/mcal_catalog.py +++ b/examples/egocap/capture_server/capture_server/mcal_catalog.py @@ -98,5 +98,5 @@ def selection_refusal(catalog: Catalog, names: list[str]) -> str | None: def resolve(directory: Path, name: str) -> Path: """The host path of one selected file; the server has already checked it against a scan.""" - # The resolved real path is handed to the tracking plugin, so nothing outside the directory can reach it. - return directory.resolve() / name + # ``directory`` is the real path ServerConfig resolved at construction, so this does no filesystem access. + return directory / name diff --git a/examples/egocap/capture_server/capture_server/server.py b/examples/egocap/capture_server/capture_server/server.py index 93a9de8967..c0ad83cf5f 100644 --- a/examples/egocap/capture_server/capture_server/server.py +++ b/examples/egocap/capture_server/capture_server/server.py @@ -177,7 +177,10 @@ class ServerConfig: manus_calibration_dir: Path | None = None def __post_init__(self) -> None: - """Reject one physical camera assigned to more than one logical name.""" + """Resolve the MCAL folder once, then reject one physical camera assigned to more than one logical name.""" + # Resolved here, at construction, so no later request re-resolves it on the event loop. + if self.manus_calibration_dir is not None: + self.manus_calibration_dir = self.manus_calibration_dir.resolve() if self.cameras is not None: ids = [c.device_id for c in self.cameras.values()] duplicates = sorted({i for i in ids if ids.count(i) > 1}) @@ -1142,6 +1145,8 @@ def __init__( # request that asked for it, so timed-out queries can never accumulate live scans. self._mcal_slots = asyncio.Semaphore(_MCAL_SCANS) self._mcal_pool = ThreadPoolExecutor(max_workers=_MCAL_SCANS, thread_name_prefix="mcal-scan") + # In-flight /query handlers, so close() can cancel one that is waiting on a slow scan. + self._query_tasks: set[asyncio.Task[Any]] = set() self._ws_server: websockets.WebSocketServer | None = None self._owner: ServerConnection | None = None # The ONE connection-owned Outbox, constructed once per owner @@ -1254,6 +1259,9 @@ async def close(self) -> None: finally: # Queued scans are dropped; a scan already running finishes on its own and its result is discarded. self._mcal_pool.shutdown(wait=False, cancel_futures=True) + # A query waiting on a running scan would otherwise hold wait_closed() past its deadline. + for task in list(self._query_tasks): + task.cancel() if self._ws_server is not None: self._ws_server.close() await asyncio.wait_for(self._ws_server.wait_closed(), timeout=_CLOSE_TIMEOUT_S) @@ -1269,7 +1277,14 @@ async def _handle_connection(self, ws: ServerConnection) -> None: # Station queries are dispatched before owner admission: they never take or disturb the owner. request_path = getattr(getattr(ws, "request", None), "path", "") if request_path.split("?", 1)[0] == "/query": - await self._handle_query(ws) + task = asyncio.current_task() + if task is not None: + self._query_tasks.add(task) + try: + await self._handle_query(ws) + finally: + if task is not None: + self._query_tasks.discard(task) return if self._owner is not None: await ws.close(code=1013, reason="owner connection already active") diff --git a/examples/egocap/tests/capture_server/test_device_query.py b/examples/egocap/tests/capture_server/test_device_query.py index 4d39bbc2c3..834f44bff8 100644 --- a/examples/egocap/tests/capture_server/test_device_query.py +++ b/examples/egocap/tests/capture_server/test_device_query.py @@ -268,8 +268,17 @@ def test_selection_refusal_explains_each_case(tmp_path): assert "disk on fire" in mcal_catalog.selection_refusal(broken, ["a.mcal"]) -def test_resolve_joins_the_name_onto_the_real_directory(tmp_path): - assert mcal_catalog.resolve(tmp_path, "a.mcal") == tmp_path.resolve() / "a.mcal" +def test_resolve_joins_the_name_onto_the_directory_without_touching_the_filesystem(tmp_path): + assert mcal_catalog.resolve(tmp_path / "gone", "a.mcal") == tmp_path / "gone" / "a.mcal" + + +def test_server_config_resolves_the_calibration_folder_once(tmp_path): + real = tmp_path / "real" + real.mkdir() + link = tmp_path / "link" + link.symlink_to(real) + assert ServerConfig(manus_calibration_dir=link).manus_calibration_dir == real.resolve() + assert ServerConfig().manus_calibration_dir is None # -- launch and adapter ------------------------------------------------------------------------------------- @@ -447,6 +456,31 @@ async def test_scan_timeout_is_reported_without_leaking_the_slot(station, monkey assert server.server._mcal_slots._value == server_module._MCAL_SCANS +async def test_close_returns_within_its_deadline_while_a_query_waits_on_a_scan(station, monkeypatch): + import threading + + url, _, server = station + started = threading.Event() + gate = threading.Event() + + def blocked_scan(directory): + started.set() + gate.wait(10) + return mcal_catalog.Catalog(STATUS.READY) + + monkeypatch.setattr(mcal_catalog, "scan", blocked_scan) + ws = await websockets.connect(f"{url}/query") + try: + await ws.send(json.dumps(_query())) + await asyncio.wait_for(asyncio.to_thread(started.wait, 5), 6) + server.server.request_stop() + # close() must settle the waiting query handler rather than raise TimeoutError from wait_closed(). + await asyncio.wait_for(server.server.close(), 4) + finally: + gate.set() + await ws.close() + + async def test_begin_accepts_files_that_are_in_the_folder_and_echoes_the_names(station): url, _, _ = station client = CaptureClient(url) From 8d4f05a8e9f9f0842c17b67fdee261a52fcbb697 Mon Sep 17 00:00:00 2001 From: Gareth Morgan Date: Thu, 5 Nov 2026 15:29:59 -0800 Subject: [PATCH 6/6] examples/egocap: make shutdown settle a Begin's MCAL scan and address device-query review findings The MCAL scan wait now stops when shutdown begins, so neither a query nor a Begin can hold close() past its deadline, and a Begin cut short that way is BUSY rather than a bad selection. query_devices() parses the reply directly instead of through the control-connection table, falls back to the exception type for an empty timeout message, and real hardware refuses a selected file with no calibration directory. Document the offered-name rule with a public is_mcal_name(), and tidy the device-query tests. Co-Authored-By: Claude Sonnet 5.5 Signed-off-by: Gareth Morgan --- examples/egocap/capture_server/README.md | 2 +- .../capture_server/capture_server/client.py | 13 +-- .../capture_server/capture_server/launch.py | 30 +++++-- .../capture_server/mcal_catalog.py | 2 +- .../capture_server/capture_server/protocol.py | 13 ++- .../capture_server/capture_server/server.py | 64 +++++++++----- .../tests/capture_server/test_device_query.py | 83 +++++++++++++++---- 7 files changed, 153 insertions(+), 54 deletions(-) diff --git a/examples/egocap/capture_server/README.md b/examples/egocap/capture_server/README.md index d6dd225124..9bc82a5f98 100644 --- a/examples/egocap/capture_server/README.md +++ b/examples/egocap/capture_server/README.md @@ -712,7 +712,7 @@ Every JSON session message has these envelope fields. The two `/query` messages Request: `{"type": 11, "version": 1, "request_id": "<1..128 chars>"}`. Reply: `{"type": 12, "version": 1, "request_id": ..., "devices": [...]}` with one entry per device kind. The Manus entry is `{"device": 1, "mcal_status": ..., "mcal_files": [...], "error_message": ...}`. `mcal_status` is `READY=1` (the list may be empty), `DISABLED=2` (the server has no MCAL folder) or `UNAVAILABLE=3` (the folder could not be read; `error_message` says why). A listing over the 2000-file limit is `UNAVAILABLE`, never truncated. -The MCAL folder is set with `--manus-calibration-dir DIR`. It is a plain folder: any nonempty, readable `.mcal` regular file directly in it is offered, whether placed by hand or written by a calibration tool. Symbolic links are never offered; copy files in. Select files by name with `ManusPluginConfig(left_calibration=..., right_calibration=...)` in `start(plugins=...)`. Begin is refused as `INVALID_CONFIG` when a name is not in the folder at that moment. A name only selects a file; whether the glove accepted it is the Manus plugin's per-hand status. +The MCAL folder is set with `--manus-calibration-dir DIR`. It is a plain folder: any nonempty, readable `.mcal` regular file directly in it is offered, whether placed by hand or written by a calibration tool. Symbolic links are never offered; copy files in. A name is offered only if it uses ASCII letters, digits, `_`, `.` and `-`, starts with a letter or digit, ends in lower-case `.mcal` and is at most 133 characters; any other file (for example `left glove (2).mcal` or `A.MCAL`) is silently skipped, so rename it. Select files by name with `ManusPluginConfig(left_calibration=..., right_calibration=...)` in `start(plugins=...)`. Begin is refused as `INVALID_CONFIG` when a name is not in the folder at that moment. A name only selects a file; whether the glove accepted it is the Manus plugin's per-hand status. If the file disappears or becomes unreadable after Begin is accepted, the plugin cannot load it and both hands are lost for that session. #### Begin: configure sources, plugins and cameras diff --git a/examples/egocap/capture_server/capture_server/client.py b/examples/egocap/capture_server/capture_server/client.py index 397f3d8e29..b88906c29f 100644 --- a/examples/egocap/capture_server/capture_server/client.py +++ b/examples/egocap/capture_server/capture_server/client.py @@ -311,7 +311,7 @@ class ManusPluginConfig: def __post_init__(self) -> None: """Reject names the protocol cannot carry, and one file chosen for both hands.""" for name in (self.left_calibration, self.right_calibration): - if name is not None and (not isinstance(name, str) or not p._MCAL_NAME_RE.fullmatch(name)): + if name is not None and not p.is_mcal_name(name): raise ValueError(f"calibration file must be an .mcal basename, got {name!r}") if self.left_calibration is not None and self.left_calibration == self.right_calibration: raise ValueError("left_calibration and right_calibration must name different files") @@ -776,7 +776,8 @@ async def query_devices(self) -> tuple[p.DeviceEntry, ...]: Raises :class:`DeviceQueryError` when no answer is obtained, including when the server is too old to know the query. """ - # The query route is a sibling of the control path on the same listener. + # The query route is at the root path of the same listener; a station URL with a path prefix is not + # supported. url = urlsplit(self._url)._replace(path="/query", query="", fragment="").geturl() request = p.QueryDevicesRequest(request_id=f"devices-{uuid.uuid4()}") try: @@ -790,15 +791,15 @@ async def query_devices(self) -> tuple[p.DeviceEntry, ...]: f"the server closed the device query (close code {code}); it may not support device queries" ) from None except (OSError, TimeoutError, websockets.exceptions.WebSocketException) as exc: - raise DeviceQueryError(f"the device query failed: {exc or type(exc).__name__}") from None + raise DeviceQueryError(f"the device query failed: {str(exc) or type(exc).__name__}") from None if isinstance(raw, bytes): raise DeviceQueryError("the server answered the device query with a binary message") try: - reply = p.parse_server_message(_from_json_text(raw)) + reply = p.DevicesReply.from_json(_from_json_text(raw)) except p.ProtocolError as exc: raise DeviceQueryError(f"the server's device reply was invalid: {exc}") from None - # A reply for another request, or of another kind, is never accepted as this query's answer. - if not isinstance(reply, p.DevicesReply) or reply.request_id != request.request_id: + # A reply for another request is never accepted as this query's answer. + if reply.request_id != request.request_id: raise DeviceQueryError("the server's reply does not answer this device query") return reply.devices diff --git a/examples/egocap/capture_server/capture_server/launch.py b/examples/egocap/capture_server/capture_server/launch.py index 0b98574354..234ba35311 100644 --- a/examples/egocap/capture_server/capture_server/launch.py +++ b/examples/egocap/capture_server/capture_server/launch.py @@ -60,15 +60,18 @@ def _group_selected(begin: p.BeginRequest, source_type: p.SourceType) -> bool: return source_type in begin.tracking.sources -def _calibration_path(name: str | None, directory: Path | None) -> str | None: +def _calibration_path(name: str | None, directory: Path | None, *, real_hardware: bool = False) -> str | None: """The host path of a selected MCAL file; the name itself where no folder exists to resolve it. The server has already checked the name against the folder before this runs. Without a folder (the mock - and replay setups, which never reach real hardware) the name is kept as an inert marker. + and replay setups, which never reach real hardware) the name is kept as an inert marker; real hardware + refuses it, because the plugin would throw on a relative path and both hands would be lost. """ if name is None: return None if directory is None: + if real_hardware: + raise ValueError(f"calibration file {name!r} was selected but no calibration directory is configured") return name return str(mcal_catalog.resolve(directory, name)) @@ -79,6 +82,7 @@ def _tracking_adapter( sdk: ModuleType, session_factory: Callable[[Any], Any], calibration_dir: Path | None = None, + real_hardware: bool = False, ) -> IsaacCaptureAdapter | None: """The IsaacCapture adapter for the requested tracking outputs, or None when none are requested. @@ -100,10 +104,14 @@ def _tracking_adapter( manus_enabled=manus_plugin_request is not None, manus_required=manus_plugin_request.required if manus_plugin_request is not None else False, manus_left_calibration=_calibration_path( - manus_plugin_request.left_calibration if manus_plugin_request is not None else None, calibration_dir + manus_plugin_request.left_calibration if manus_plugin_request is not None else None, + calibration_dir, + real_hardware=real_hardware, ), manus_right_calibration=_calibration_path( - manus_plugin_request.right_calibration if manus_plugin_request is not None else None, calibration_dir + manus_plugin_request.right_calibration if manus_plugin_request is not None else None, + calibration_dir, + real_hardware=real_hardware, ), ) return IsaacCaptureAdapter( @@ -210,7 +218,11 @@ def source_factory(begin: p.BeginRequest, cameras: tuple[p.CameraStreamConfig, . sdk = load_isaac_capture() if begin.tracking.sources else None tracking = ( _tracking_adapter( - begin, sdk=sdk, session_factory=sdk.TeleopSession, calibration_dir=config.manus_calibration_dir + begin, + sdk=sdk, + session_factory=sdk.TeleopSession, + calibration_dir=config.manus_calibration_dir, + real_hardware=True, ) if sdk else None @@ -330,7 +342,13 @@ def source_factory(begin: p.BeginRequest, requested: tuple[p.CameraStreamConfig, sdk = load_isaac_capture() if begin.tracking.sources else None tracking = ( - _tracking_adapter(begin, sdk=sdk, session_factory=sdk.TeleopSession, calibration_dir=calibration_dir) + _tracking_adapter( + begin, + sdk=sdk, + session_factory=sdk.TeleopSession, + calibration_dir=calibration_dir, + real_hardware=True, + ) if sdk else None ) diff --git a/examples/egocap/capture_server/capture_server/mcal_catalog.py b/examples/egocap/capture_server/capture_server/mcal_catalog.py index d03eba16fd..210eb71761 100644 --- a/examples/egocap/capture_server/capture_server/mcal_catalog.py +++ b/examples/egocap/capture_server/capture_server/mcal_catalog.py @@ -68,7 +68,7 @@ def _selectable(root: Path) -> list[str]: with os.scandir(root) as listing: for entry in listing: # Names the protocol cannot carry are never offered. - if not p._MCAL_NAME_RE.fullmatch(entry.name): + if not p.is_mcal_name(entry.name): continue try: # Links are refused even when their target stays inside the directory. diff --git a/examples/egocap/capture_server/capture_server/protocol.py b/examples/egocap/capture_server/capture_server/protocol.py index 8e6adfd2de..7338793fcd 100644 --- a/examples/egocap/capture_server/capture_server/protocol.py +++ b/examples/egocap/capture_server/capture_server/protocol.py @@ -942,11 +942,17 @@ def _require_session_id(data: dict) -> str: _MAX_TOKEN_LEN = 128 +def is_mcal_name(value: object) -> bool: + """True for a name the protocol can carry: ASCII letters, digits, ``_``, ``.`` and ``-``, starting with a + letter or digit, lower-case ``.mcal`` extension, at most 133 characters.""" + return type(value) is str and _MCAL_NAME_RE.fullmatch(value) is not None + + def _require_mcal_name(value: Any, context: str) -> str | None: """An optional ``.mcal`` basename, or ``ProtocolError`` when present but not one.""" if value is None: return None - if not isinstance(value, str) or not _MCAL_NAME_RE.fullmatch(value): + if not is_mcal_name(value): raise ProtocolError(f"{context} must be an .mcal basename or null") return value @@ -1026,7 +1032,7 @@ def __post_init__(self) -> None: if type(self.plugin) is not PluginType: raise ValueError(f"PluginRequest.plugin must be a PluginType, got {self.plugin!r}") for name in (self.left_calibration, self.right_calibration): - if name is not None and (type(name) is not str or not _MCAL_NAME_RE.fullmatch(name)): + if name is not None and not is_mcal_name(name): raise ValueError(f"PluginRequest calibration must be an .mcal basename, got {name!r}") if self.left_calibration is not None and self.left_calibration == self.right_calibration: raise ValueError("left_calibration and right_calibration must name different files") @@ -1982,7 +1988,7 @@ def __post_init__(self) -> None: raise ValueError("mcal_files are present only for a READY listing") if len(self.mcal_files) > MAX_MCAL_FILES: raise ValueError(f"at most {MAX_MCAL_FILES} MCAL files fit in one reply") - if not all(_MCAL_NAME_RE.fullmatch(name) for name in self.mcal_files): + if not all(is_mcal_name(name) for name in self.mcal_files): raise ValueError("mcal_files must be .mcal basenames") if len(set(self.mcal_files)) != len(self.mcal_files): raise ValueError("an MCAL file appears twice in the listing") @@ -2073,7 +2079,6 @@ def from_json(cls, data: Any) -> DevicesReply: MessageType.SESSION_STATUS: SessionStatus.from_json, MessageType.SESSION_STOPPED: SessionStopped.from_json, MessageType.REQUEST_REJECTED: RequestRejected.from_json, - MessageType.DEVICES: DevicesReply.from_json, } diff --git a/examples/egocap/capture_server/capture_server/server.py b/examples/egocap/capture_server/capture_server/server.py index c0ad83cf5f..ad7c91f03f 100644 --- a/examples/egocap/capture_server/capture_server/server.py +++ b/examples/egocap/capture_server/capture_server/server.py @@ -1145,8 +1145,6 @@ def __init__( # request that asked for it, so timed-out queries can never accumulate live scans. self._mcal_slots = asyncio.Semaphore(_MCAL_SCANS) self._mcal_pool = ThreadPoolExecutor(max_workers=_MCAL_SCANS, thread_name_prefix="mcal-scan") - # In-flight /query handlers, so close() can cancel one that is waiting on a slow scan. - self._query_tasks: set[asyncio.Task[Any]] = set() self._ws_server: websockets.WebSocketServer | None = None self._owner: ServerConnection | None = None # The ONE connection-owned Outbox, constructed once per owner @@ -1259,9 +1257,6 @@ async def close(self) -> None: finally: # Queued scans are dropped; a scan already running finishes on its own and its result is discarded. self._mcal_pool.shutdown(wait=False, cancel_futures=True) - # A query waiting on a running scan would otherwise hold wait_closed() past its deadline. - for task in list(self._query_tasks): - task.cancel() if self._ws_server is not None: self._ws_server.close() await asyncio.wait_for(self._ws_server.wait_closed(), timeout=_CLOSE_TIMEOUT_S) @@ -1277,14 +1272,7 @@ async def _handle_connection(self, ws: ServerConnection) -> None: # Station queries are dispatched before owner admission: they never take or disturb the owner. request_path = getattr(getattr(ws, "request", None), "path", "") if request_path.split("?", 1)[0] == "/query": - task = asyncio.current_task() - if task is not None: - self._query_tasks.add(task) - try: - await self._handle_query(ws) - finally: - if task is not None: - self._query_tasks.discard(task) + await self._handle_query(ws) return if self._owner is not None: await ws.close(code=1013, reason="owner connection already active") @@ -1543,10 +1531,8 @@ async def _handle_begin(self, ws: ServerConnection, data: dict[str, Any]) -> Non # Only a Begin that would allocate a new session scans the folder, so a duplicate of an accepted Begin is # replayed above without depending on a file that may have changed since. refusal = await self._calibration_selection_refusal(begin) - if refusal is not None: - self._reject(begin.session_id, p.MessageType.BEGIN, p.RejectionReason.INVALID_CONFIG, refusal) - return - # The scan awaited: shutdown, the writer or admission may have changed since the checks above. + # The scan awaited: shutdown, the writer or admission may have changed since the checks above. Checked + # before the refusal, so a scan cut short by shutdown reads as BUSY rather than as a bad selection. if self._stop_event.is_set() or self._outbox is None or self._outbox.closed: self._reject( begin.session_id, @@ -1555,6 +1541,9 @@ async def _handle_begin(self, ws: ServerConnection, data: dict[str, Any]) -> Non "server is shutting down" if self._stop_event.is_set() else "this connection is no longer usable", ) return + if refusal is not None: + self._reject(begin.session_id, p.MessageType.BEGIN, p.RejectionReason.INVALID_CONFIG, refusal) + return if self._active_session is not None: self._reject( begin.session_id, p.MessageType.BEGIN, p.RejectionReason.BUSY, "another session is already active" @@ -1595,12 +1584,17 @@ async def _scan_mcal(self) -> mcal_catalog.Catalog: The permit is released by the scan's own completion, not by the caller, so a request that times out or is cancelled cannot leave more than ``_MCAL_SCANS`` scans running. The caller waits at most - ``_QUERY_DEADLINE_S`` for a free slot and for the result, and gets UNAVAILABLE otherwise. + ``_QUERY_DEADLINE_S`` for a free slot and for the result, and gets UNAVAILABLE otherwise; it stops + waiting as soon as shutdown begins, so neither a query nor a Begin can hold ``close()`` open. """ directory = self.config.manus_calibration_dir + shutting_down = mcal_catalog.Catalog( + p.ManusCalibrationCatalogStatus.UNAVAILABLE, (), "the server is shutting down" + ) try: async with asyncio.timeout(_QUERY_DEADLINE_S): - await self._mcal_slots.acquire() + if not await self._acquire_scan_slot(): + return shutting_down try: future = asyncio.get_running_loop().run_in_executor( self._mcal_pool, mcal_catalog.scan, directory @@ -1611,7 +1605,18 @@ async def _scan_mcal(self) -> mcal_catalog.Catalog: raise future.add_done_callback(self._mcal_scan_finished) # Shielded: cancelling this wait must not cancel the job whose permit is still held. - return await asyncio.shield(future) + scan = asyncio.shield(future) + stop = asyncio.ensure_future(self._stop_event.wait()) + try: + await asyncio.wait({scan, stop}, return_when=asyncio.FIRST_COMPLETED) + finally: + stop.cancel() + # Cancels only this wait, never the scan thread; a timeout or shutdown leaves nothing dangling. + if not scan.done(): + scan.cancel() + if scan.cancelled(): + return shutting_down + return scan.result() except TimeoutError: return mcal_catalog.Catalog( p.ManusCalibrationCatalogStatus.UNAVAILABLE, (), "the calibration directory scan timed out" @@ -1622,6 +1627,25 @@ async def _scan_mcal(self) -> mcal_catalog.Catalog: p.ManusCalibrationCatalogStatus.UNAVAILABLE, (), "the calibration directory scan failed" ) + async def _acquire_scan_slot(self) -> bool: + """Take a scan permit, or give up (False) the moment shutdown begins; never leaves a permit held.""" + if self._stop_event.is_set(): + return False + acquire = asyncio.ensure_future(self._mcal_slots.acquire()) + stop = asyncio.ensure_future(self._stop_event.wait()) + try: + await asyncio.wait({acquire, stop}, return_when=asyncio.FIRST_COMPLETED) + finally: + stop.cancel() + if not acquire.done(): + acquire.cancel() + if acquire.done() and not acquire.cancelled(): + if not self._stop_event.is_set(): + return True + # Shutdown arrived as the permit was granted: hand it straight back. + self._mcal_slots.release() + return False + def _mcal_scan_finished(self, future: asyncio.Future[Any]) -> None: """Free the scan's slot when its thread ends, and consume an abandoned job's exception.""" self._mcal_slots.release() diff --git a/examples/egocap/tests/capture_server/test_device_query.py b/examples/egocap/tests/capture_server/test_device_query.py index 834f44bff8..077564fece 100644 --- a/examples/egocap/tests/capture_server/test_device_query.py +++ b/examples/egocap/tests/capture_server/test_device_query.py @@ -58,10 +58,10 @@ def test_query_request_rejects_malformed_envelopes(mutate): p.parse_query_devices(data) -def test_devices_reply_round_trips_through_the_server_message_parser(): +def test_devices_reply_round_trips_through_its_own_parser(): entry = p.DeviceEntry(p.PluginType.MANUS, STATUS.READY, ("a.mcal", "b.mcal")) reply = p.DevicesReply("r-1", (entry,)) - assert p.parse_server_message(json.loads(json.dumps(reply.to_json()))) == reply + assert p.DevicesReply.from_json(json.loads(json.dumps(reply.to_json()))) == reply @pytest.mark.parametrize( @@ -110,7 +110,7 @@ def test_devices_reply_rejects_malformed_replies(mutate): data = p.DevicesReply("r-1", (p.DeviceEntry(p.PluginType.MANUS, STATUS.READY, ("a.mcal",)),)).to_json() mutate(data) with pytest.raises(p.ProtocolError): - p.parse_server_message(data) + p.DevicesReply.from_json(data) def test_plugin_request_carries_calibration_names_through_the_wire(): @@ -230,11 +230,6 @@ def test_scan_over_the_limit_is_an_error_not_a_truncated_list(tmp_path, monkeypa def test_scan_skips_an_entry_that_vanishes_mid_scan(tmp_path, monkeypatch): _mcal(tmp_path, "a.mcal") - real_stat = Path.stat - - def flaky(self, *args, **kwargs): - return real_stat(self, *args, **kwargs) - import os real_scandir = os.scandir @@ -254,7 +249,6 @@ def __exit__(self, *exc): return False monkeypatch.setattr(mcal_catalog.os, "scandir", lambda root: _Listing()) - monkeypatch.setattr(Path, "stat", flaky) assert list(mcal_catalog.scan(tmp_path).files) == ["a.mcal"] @@ -284,10 +278,41 @@ def test_server_config_resolves_the_calibration_folder_once(tmp_path): # -- launch and adapter ------------------------------------------------------------------------------------- -def test_calibration_path_resolves_inside_the_folder_or_keeps_an_inert_name(tmp_path): +def test_calibration_path_joins_onto_the_folder_or_keeps_an_inert_name(tmp_path): assert _calibration_path(None, tmp_path) is None assert _calibration_path("a.mcal", None) == "a.mcal" - assert _calibration_path("a.mcal", tmp_path) == str(tmp_path.resolve() / "a.mcal") + assert _calibration_path("a.mcal", tmp_path) == str(tmp_path / "a.mcal") + + +def test_real_hardware_refuses_a_selected_file_without_a_folder(): + with pytest.raises(ValueError, match="no calibration directory"): + _calibration_path("a.mcal", None, real_hardware=True) + assert _calibration_path(None, None, real_hardware=True) is None + + +@pytest.mark.parametrize( + ("name", "ok"), + [ + ("a.mcal", True), + ("left-1_v2.mcal", True), + ("a" * 128 + ".mcal", True), + ("left glove (2).mcal", False), + ("_a.mcal", False), + ("-a.mcal", False), + ("A.MCAL", False), + ("é.mcal", False), + ("a" * 129 + ".mcal", False), + (None, False), + ], +) +def test_is_mcal_name_states_the_offered_name_rule(name, ok): + assert p.is_mcal_name(name) is ok + + +def test_a_devices_message_is_not_a_control_connection_message(): + reply = p.DevicesReply("r-1", (p.DeviceEntry(p.PluginType.MANUS, STATUS.READY),)) + with pytest.raises(p.ProtocolError): + p.parse_server_message(reply.to_json()) def test_adapter_passes_selected_files_to_the_manus_plugin(): @@ -324,7 +349,7 @@ async def _ask(url: str, request: dict | str | bytes, *, request_id: str = "q-1" await ws.send(request if isinstance(request, (str, bytes)) else json.dumps(request)) reply = None try: - reply = p.parse_server_message(json.loads(await asyncio.wait_for(ws.recv(), 5))) + reply = p.DevicesReply.from_json(json.loads(await asyncio.wait_for(ws.recv(), 5))) except websockets.ConnectionClosed: pass # a rejected request is closed without a reply; the close code is what the caller checks await asyncio.wait_for(ws.wait_closed(), 5) @@ -426,7 +451,7 @@ async def test_oversized_reply_becomes_an_unavailable_entry(station, monkeypatch assert reply.devices[0].mcal_status is STATUS.UNAVAILABLE and "too large" in reply.devices[0].error_message -async def test_a_scan_that_times_out_or_fails_is_unavailable(station, monkeypatch): +async def test_a_scan_that_fails_is_unavailable(station, monkeypatch): url, _, _ = station monkeypatch.setattr(mcal_catalog, "scan", lambda directory: (_ for _ in ()).throw(RuntimeError("boom"))) reply, _ = await _ask(url, _query()) @@ -437,7 +462,6 @@ async def test_scan_timeout_is_reported_without_leaking_the_slot(station, monkey from capture_server import server as server_module url, _, server = station - release = asyncio.Event() import threading gate = threading.Event() @@ -448,7 +472,6 @@ async def test_scan_timeout_is_reported_without_leaking_the_slot(station, monkey reply, _ = await _ask(url, _query()) assert "timed out" in reply.devices[0].error_message gate.set() - release.set() for _ in range(100): if server.server._mcal_slots._value == server_module._MCAL_SCANS: break @@ -481,6 +504,34 @@ def blocked_scan(directory): await ws.close() +async def test_close_returns_within_its_deadline_while_a_begin_waits_on_a_scan(station, monkeypatch): + import threading + + url, _, server = station + started = threading.Event() + gate = threading.Event() + + def blocked_scan(directory): + started.set() + gate.wait(10) + return mcal_catalog.Catalog(STATUS.READY, ("left.mcal",)) + + monkeypatch.setattr(mcal_catalog, "scan", blocked_scan) + client = CaptureClient(url) + begin = asyncio.ensure_future( + client.start(sources=(p.SourceType.HEAD,), plugins=(ManusPluginConfig(left_calibration="left.mcal"),)) + ) + try: + await asyncio.wait_for(asyncio.to_thread(started.wait, 5), 6) + server.server.request_stop() + # close() must settle the Begin waiting on the scan rather than raise TimeoutError from wait_closed(). + await asyncio.wait_for(server.server.close(), 4) + finally: + gate.set() + begin.cancel() + await asyncio.gather(begin, return_exceptions=True) + + async def test_begin_accepts_files_that_are_in_the_folder_and_echoes_the_names(station): url, _, _ = station client = CaptureClient(url) @@ -509,7 +560,7 @@ async def test_begin_refuses_a_file_that_is_not_in_the_folder(station): assert "not an available file" in excinfo.value.error_message -async def test_begin_refuses_a_file_that_became_a_link_or_empty_since_listing(station): +async def test_begin_refuses_a_file_that_became_empty_since_listing(station): url, folder, _ = station (folder / "left.mcal").write_bytes(b"") with pytest.raises(RequestRejectedError):