From ae48621f1cc13054c2246210d8f92d30e27fe354 Mon Sep 17 00:00:00 2001 From: Hougant Chen Date: Wed, 28 Oct 2026 19:56:23 -0400 Subject: [PATCH] Add a Vulkan desktop mirror to OVXR teleoperation Restack the reviewed optional mirror on the rebased camera-consumer layer, preserving frame ownership and teardown behavior. Signed-off-by: Hougant Chen --- README.md | 4 +- docs/source/getting_started/ovxr.rst | 28 +- pyproject.toml | 5 + scripts/check_camera_quad_wheel.py | 11 +- src/compat/pyproject.toml | 1 + src/core/python/CMakeLists.txt | 1 + src/core/python/pyproject.toml.in | 1 + .../python/requirements-desktop-mirror.txt | 3 + .../ovxr_session_manager/__init__.py | 3 + .../ovxr_session_manager/desktop_mirror.py | 131 ++++ .../ovxr_session_manager/gpu_output.py | 124 +++- .../ovxr_session_manager/owned_mirror.py | 655 ++++++++++++++++++ .../core/ovxr_session_manager/conftest.py | 65 ++ .../test_desktop_mirror.py | 122 ++++ .../ovxr_session_manager/test_gpu_mirror.py | 414 +++++++++++ .../ovxr_session_manager/test_ovxr_session.py | 407 ++++++++++- .../ovxr_session_manager/test_owned_mirror.py | 568 +++++++++++++++ .../test_passive_mirror.py | 172 +++++ 18 files changed, 2707 insertions(+), 8 deletions(-) create mode 100644 src/core/python/requirements-desktop-mirror.txt create mode 100644 src/python/isaaccapture/ovxr_session_manager/desktop_mirror.py create mode 100644 src/python/isaaccapture/ovxr_session_manager/owned_mirror.py create mode 100644 tests/python/core/ovxr_session_manager/conftest.py create mode 100644 tests/python/core/ovxr_session_manager/test_desktop_mirror.py create mode 100644 tests/python/core/ovxr_session_manager/test_gpu_mirror.py create mode 100644 tests/python/core/ovxr_session_manager/test_owned_mirror.py create mode 100644 tests/python/core/ovxr_session_manager/test_passive_mirror.py diff --git a/README.md b/README.md index ad3a0f51bf..fad32540ec 100644 --- a/README.md +++ b/README.md @@ -91,10 +91,10 @@ Third-party materials may carry their own licenses. ### Separately retrieved optional materials The optional `isaaccapture[ovxr]` extra retrieves OVXR and its OVRTX/OVStage -dependencies. +dependencies. The separate `isaaccapture[desktop-mirror]` extra retrieves OVUI. These packages are downloaded separately, rather than bundled in the Isaac Capture wheel. Their licenses apply independently of Isaac Capture's Apache-2.0 license. -OVXR, OVRTX and OVStage are NVIDIA proprietary software. +OVXR, OVRTX and OVStage are NVIDIA proprietary software. OVUI is Apache-2.0. See [package sources and license details](docs/source/getting_started/ovxr.rst#optional-packages-and-licenses). Review the applicable package licenses and terms before installing or using these optional dependencies. diff --git a/docs/source/getting_started/ovxr.rst b/docs/source/getting_started/ovxr.rst index ccec8c4e7f..bf961aeece 100644 --- a/docs/source/getting_started/ovxr.rst +++ b/docs/source/getting_started/ovxr.rst @@ -37,6 +37,9 @@ from the Isaac Capture wheel. These packages retain their own licenses: `OVRTX `__ and `OVStage `__ are NVIDIA proprietary software. Review the license included with each distribution before use. +* `OVUI `__ is Apache-2.0 and is installed by + the independent ``desktop-mirror`` extra. Review its package metadata for + supported Python versions and platforms. CloudXR native libraries are bundled in Isaac Capture wheels and remain subject to their separate NVIDIA license and CloudXR EULA acceptance. @@ -72,6 +75,25 @@ Use ``CUDAToolkit_ROOT`` to select a CUDA Toolkit installation and ``VULKAN_SDK` to help locate a Vulkan SDK. The GPU driver and Vulkan runtime loader are required on the host. +Optional desktop mirror +----------------------- + +The desktop mirror requires **Python 3.12 or newer**. Isaac Capture's core +continues to support Python 3.11; on that version the extra does not install OVUI +and the desktop mirror is unavailable. Use a Python 3.12+ environment and select +both extras with ``isaaccapture[ovxr,desktop-mirror]`` in the installation +requirement or the application's optional dependencies to include the OVUI +mirror. The mirror requires CUDA 13 Python bindings, a compatible NVIDIA driver +as described above, and a desktop display associated with a CUDA device. +The native camera helper is required only when camera panels are also enabled. + +Initialize the renderer before starting the mirror. OVUI uses its default Vulkan +backend; no ``OMNIUI_BACKEND`` setting is required. Capture does not change the +process-wide backend. GPU images stay on their producing CUDA device, and OVUI +copies them into its Vulkan image. The mirror requires the active OVUI backend +to be Vulkan. Remove conflicting ``OMNIUI_BACKEND`` settings before OVUI starts. +This selection does not change headset rendering. + Use from another application ---------------------------- @@ -83,9 +105,11 @@ resolved its render-product paths. Capture does not depend on an Isaac Lab task or scene configuration. ``camera_paths`` is required and must contain both stereo cameras. Set ``up_axis="Y"`` for a Y-up Stage; the default is ``"Z"``. -``CameraPanelConsumer`` implements the ``OvxrFrameConsumer`` lifecycle. Configure panels with a tuple of +``CameraPanelConsumer`` and ``OwnedDesktopMirror`` implement the same +``OvxrFrameConsumer`` lifecycle. Configure panels with a tuple of ``CameraPanelConfig`` values, each pairing a render-product path with its placement -and image bound. +and image bound. Configure the mirror's camera choices with ``MirrorProduct`` +values. Either consumer can be enabled independently. Consumers receive ``FrameOutputs`` on the XR owner thread. Use ``latest_color`` to select an output, then ``copy_host_rgba`` or ``map_gpu`` to consume it without diff --git a/pyproject.toml b/pyproject.toml index b7f75e9bd0..b6de4f50bd 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -54,6 +54,11 @@ ovxr = [ "ovxr==0.1.0.5842", "cuda-bindings>=13,<14", ] +desktop-mirror = [ + # OVUI requires 3.12; core Capture continues to support 3.11. + "ovui==0.3.0; python_version >= '3.12'", + "cuda-bindings>=13,<14", +] retargeters = [ "dex-retargeting>=0.4.6,<0.6.0", "scipy>=1.15.0", diff --git a/scripts/check_camera_quad_wheel.py b/scripts/check_camera_quad_wheel.py index 00ac948ef4..671d759de0 100644 --- a/scripts/check_camera_quad_wheel.py +++ b/scripts/check_camera_quad_wheel.py @@ -1,7 +1,7 @@ # SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. # SPDX-License-Identifier: Apache-2.0 -"""Check optional camera-helper delivery in an actual project wheel. +"""Check camera-helper and desktop-mirror delivery in an actual project wheel. Run against wheels built with BUILD_OVXR_CAMERA_QUADS ON and OFF, including ON -> OFF -> ON in the same build directory to detect stale packaged data. @@ -26,6 +26,15 @@ def main(): prefix + "libisaaccapture_camera_panels.so", } with ZipFile(args.wheel) as wheel: + mirror_modules = { + "isaaccapture/ovxr_session_manager/" + name + ".py" + for name in ("owned_mirror", "desktop_mirror", "gpu_output") + } + missing = mirror_modules.difference(wheel.namelist()) + if missing: + raise RuntimeError( + f"Wheel is missing desktop mirror modules: {sorted(missing)}" + ) members = { name for name in wheel.namelist() diff --git a/src/compat/pyproject.toml b/src/compat/pyproject.toml index 9bdb3894f0..bf89de013e 100644 --- a/src/compat/pyproject.toml +++ b/src/compat/pyproject.toml @@ -23,6 +23,7 @@ Homepage = "https://github.com/NVIDIA/IsaacCapture" dev = ["isaaccapture[dev]>=1.6.dev0,<1.9"] ui = ["isaaccapture[ui]>=1.6.dev0,<1.9"] ovxr = ["isaaccapture[ovxr]>=1.6.dev0,<1.9"] +desktop-mirror = ["isaaccapture[desktop-mirror]>=1.6.dev0,<1.9"] retargeters = ["isaaccapture[retargeters]>=1.6.dev0,<1.9"] retargeters-lite = ["isaaccapture[retargeters-lite]>=1.6.dev0,<1.9"] grounding = ["isaaccapture[grounding]>=1.6.dev0,<1.9"] diff --git a/src/core/python/CMakeLists.txt b/src/core/python/CMakeLists.txt index c55fd4a7d0..6a826f498c 100644 --- a/src/core/python/CMakeLists.txt +++ b/src/core/python/CMakeLists.txt @@ -84,6 +84,7 @@ add_custom_target(python_package ALL COMMAND ${CMAKE_COMMAND} -E copy "${CMAKE_CURRENT_SOURCE_DIR}/requirements.txt" "${CMAKE_BINARY_DIR}/python_package/$/" COMMAND ${CMAKE_COMMAND} -E copy "${CMAKE_CURRENT_SOURCE_DIR}/requirements-dev.txt" "${CMAKE_BINARY_DIR}/python_package/$/" COMMAND ${CMAKE_COMMAND} -E copy "${CMAKE_CURRENT_SOURCE_DIR}/requirements-ui.txt" "${CMAKE_BINARY_DIR}/python_package/$/" + COMMAND ${CMAKE_COMMAND} -E copy "${CMAKE_CURRENT_SOURCE_DIR}/requirements-desktop-mirror.txt" "${CMAKE_BINARY_DIR}/python_package/$/" COMMAND ${CMAKE_COMMAND} -E copy "${CMAKE_CURRENT_SOURCE_DIR}/requirements-ovxr.txt" "${CMAKE_BINARY_DIR}/python_package/$/" COMMAND ${CMAKE_COMMAND} -E copy "${CMAKE_CURRENT_SOURCE_DIR}/requirements-cloudxr.txt" "${CMAKE_BINARY_DIR}/python_package/$/" COMMAND ${CMAKE_COMMAND} -E copy "${CMAKE_CURRENT_SOURCE_DIR}/requirements-retargeters.txt" "${CMAKE_BINARY_DIR}/python_package/$/" diff --git a/src/core/python/pyproject.toml.in b/src/core/python/pyproject.toml.in index 33c1855baf..31ca29b176 100644 --- a/src/core/python/pyproject.toml.in +++ b/src/core/python/pyproject.toml.in @@ -31,6 +31,7 @@ dynamic = ["dependencies", "optional-dependencies"] dependencies = {file = ["requirements.txt"]} optional-dependencies.dev = {file = ["requirements-dev.txt"]} optional-dependencies.ui = {file = ["requirements-ui.txt"]} +optional-dependencies.desktop-mirror = {file = ["requirements-desktop-mirror.txt"]} optional-dependencies.ovxr = {file = ["requirements-ovxr.txt"]} optional-dependencies.retargeters = {file = ["requirements-retargeters.txt"]} optional-dependencies.retargeters-lite = {file = ["requirements-retargeters-lite.txt"]} diff --git a/src/core/python/requirements-desktop-mirror.txt b/src/core/python/requirements-desktop-mirror.txt new file mode 100644 index 0000000000..7588c37a06 --- /dev/null +++ b/src/core/python/requirements-desktop-mirror.txt @@ -0,0 +1,3 @@ +# Optional standalone desktop mirror (Python 3.12+) +ovui==0.3.0; python_version >= "3.12" +cuda-bindings>=13,<14 diff --git a/src/python/isaaccapture/ovxr_session_manager/__init__.py b/src/python/isaaccapture/ovxr_session_manager/__init__.py index d85ab4d2ad..075a5ee491 100644 --- a/src/python/isaaccapture/ovxr_session_manager/__init__.py +++ b/src/python/isaaccapture/ovxr_session_manager/__init__.py @@ -13,6 +13,7 @@ OvxrSession, StageOrdinalLedger, ) +from .desktop_mirror import DesktopMirrorConfig, DesktopMirrorStatus __all__ = [ "OvxrSessionBinding", @@ -21,6 +22,8 @@ "OvxrFrameConsumer", "OvxrSession", "StageOrdinalLedger", + "DesktopMirrorConfig", + "DesktopMirrorStatus", ] from .camera_panels import CameraPanelConfig, CameraPanelConsumer diff --git a/src/python/isaaccapture/ovxr_session_manager/desktop_mirror.py b/src/python/isaaccapture/ovxr_session_manager/desktop_mirror.py new file mode 100644 index 0000000000..03d153075a --- /dev/null +++ b/src/python/isaaccapture/ovxr_session_manager/desktop_mirror.py @@ -0,0 +1,131 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Shared configuration, cadence and copy completion for the owned desktop mirror.""" + +from __future__ import annotations + +import math +from collections import deque +from dataclasses import dataclass + + +@dataclass(frozen=True) +class DesktopMirrorConfig: + """Configuration for the optional, application-side OVUI mirror.""" + + max_fps: float = 30.0 + window_width: int = 960 + title: str = "Isaac Teleop — render products" + initial_status_text: str = "" + + def __post_init__(self) -> None: + if not math.isfinite(self.max_fps) or self.max_fps <= 0.0: + raise ValueError("desktop mirror max_fps must be finite and positive") + if self.window_width <= 0: + raise ValueError("desktop mirror window_width must be positive") + if not self.title: + raise ValueError("desktop mirror title must be non-empty") + + +@dataclass(frozen=True) +class DesktopMirrorStatus: + """Thread-safe snapshot of mirror lifecycle and latest-frame behavior.""" + + thread_id: int | None = None + window_opened: bool = False + closed: bool = False + error: str | None = None + presented_frames: int = 0 + last_sequence: int = 0 + main_thread_fps: float = 0.0 + status_text: str = "" + render_thread_fps: float = 0.0 + + +class _CadenceSampler: + """Measure recent host and render cadence from monotonic samples.""" + + def __init__(self, window_s: float = 1.0) -> None: + self._window_ns = int(window_s * 1_000_000_000) + self._host_samples: deque[int] = deque() + self._render_samples: deque[int] = deque() + + def record_host_frame(self, now_ns: int) -> None: + self._host_samples.append(now_ns) + while ( + len(self._host_samples) > 2 + and self._host_samples[1] <= now_ns - self._window_ns + ): + self._host_samples.popleft() + + def record_render_frame(self, now_ns: int) -> None: + self._render_samples.append(now_ns) + while ( + len(self._render_samples) > 2 + and self._render_samples[1] <= now_ns - self._window_ns + ): + self._render_samples.popleft() + + @staticmethod + def _event_rate(samples: deque[int] | list[int]) -> float: + if len(samples) < 2: + return 0.0 + elapsed_s = (samples[-1] - samples[0]) / 1_000_000_000 + return 0.0 if elapsed_s <= 0.0 else (len(samples) - 1) / elapsed_s + + def rates(self, now_ns: int | None = None) -> tuple[float, float]: + def recent_rate(samples: deque[int]) -> float: + if now_ns is not None and ( + not samples or now_ns - samples[-1] > self._window_ns + ): + return 0.0 + return self._event_rate(samples) + + return recent_rate(self._host_samples), recent_rate(self._render_samples) + + +class _InputCopyCompletion: + """Fence OVUI input uploads, not its unrelated output-streaming API. + + Confirm the source is reusable even after an upload exception. Only the UI + worker waits; OVXR skips captures when its optional slots are leased. + """ + + def __init__(self, device: int) -> None: + from cuda.bindings import runtime + + self._cuda = runtime + self._device = device + display_device = self._check(runtime.cudaGetDevice()) + try: + self._check(runtime.cudaSetDevice(device)) + self._event = self._check( + runtime.cudaEventCreateWithFlags(runtime.cudaEventDisableTiming) + ) + finally: + self._check(runtime.cudaSetDevice(display_device)) + + @staticmethod + def _check(result): + if int(result[0]): + raise RuntimeError(f"OVUI input-copy completion failed: CUDA {result[0]}") + return result[1] if len(result) > 1 else None + + def wait(self) -> None: + # Vulkan may select its import device while performing its blocking copy. + display_device = self._check(self._cuda.cudaGetDevice()) + try: + self._check(self._cuda.cudaSetDevice(self._device)) + self._check( + self._cuda.cudaEventRecord(self._event, self._cuda.cudaStreamLegacy) + ) + self._check(self._cuda.cudaEventSynchronize(self._event)) + finally: + # Subsequent UI frames must retain the provider's import device. + self._check(self._cuda.cudaSetDevice(display_device)) + + def close(self) -> None: + if self._event is not None: + self._check(self._cuda.cudaEventDestroy(self._event)) + self._event = None diff --git a/src/python/isaaccapture/ovxr_session_manager/gpu_output.py b/src/python/isaaccapture/ovxr_session_manager/gpu_output.py index 3e9478fc29..0aeef03539 100644 --- a/src/python/isaaccapture/ovxr_session_manager/gpu_output.py +++ b/src/python/isaaccapture/ovxr_session_manager/gpu_output.py @@ -126,6 +126,128 @@ class MappedGpuImage(MappedOutput): is_array: bool = False +@dataclass +class OwnedGpuImage: + """One reusable CUDA allocation, leased until the UI's source read completes.""" + + width: int + height: int + pointer: int + pitch: int + device: int + owner: int + closed: bool = True + + def close(self) -> None: + if threading.get_ident() != self.owner: + raise RuntimeError("GPU image must be retired on its allocation owner") + self.closed = True + + +class OwnedCudaImagePool: + """At most two mirror copies; no borrowed OVRTX map escapes a frame.""" + + def __init__(self) -> None: + self._images: list[OwnedGpuImage] = [] + self._source: MappedGpuImage | None = None + self._ready = False + + def enable(self) -> None: + """UI owner, after OVUI has initialized its Vulkan image provider.""" + self._ready = True + + def __call__( + self, renderer: int, handle: int, max_bytes: int + ) -> OwnedGpuImage | None: + if not self._ready: + return None # GUI startup is an expected optional-output omission. + self.finish_frame() + from cuda.bindings import runtime as cuda + + try: + source = map_cuda_array_rgba(renderer, handle, max_bytes) + except OutputMapCleanupError as error: + self._source = error.image + raise + self._source = source + destination = source.device + try: + image = next((item for item in self._images if item.closed), None) + if image is not None and (image.width, image.height, image.device) != ( + source.width, + source.height, + destination, + ): + _check_cuda(cuda.cudaSetDevice(image.device)) + _check_cuda(cuda.cudaFree(image.pointer)) + self._images.remove(image) + image = None + _check_cuda(cuda.cudaSetDevice(destination)) + if image is None: + if len(self._images) >= 2: + raise RuntimeError("both owned mirror images are in use") + result = cuda.cudaMallocPitch(source.width * 4, source.height) + _check_cuda(result) + image = OwnedGpuImage( + source.width, + source.height, + int(result[1]), + int(result[2]), + destination, + threading.get_ident(), + ) + self._images.append(image) + # Check reused allocations too: CUDA may pad a new allocation past + # the limit, and a rejected allocation remains owned until cleanup. + if image.pitch * image.height > max_bytes: + raise RuntimeError("owned mirror allocation exceeds the byte bound") + # The source map and the destination have separate lifetimes. Prove + # the read complete before unmapping even when enqueue reports error. + source.busy = True + _check_cuda( + cuda.cudaMemcpy2DFromArrayAsync( + image.pointer, + image.pitch, + source.pointer, + 0, + 0, + source.width * 4, + source.height, + cuda.cudaMemcpyKind.cudaMemcpyDeviceToDevice, + 0, + ) + ) + finally: + _check_cuda(cuda.cudaSetDevice(destination)) + if source.busy: + _check_cuda(cuda.cudaStreamSynchronize(0)) + source.busy = False + source.close() + self._source = None + image.closed = False + return image + + def finish_frame(self) -> None: + """Fail closed before submit/cancel if a source read/unmap was not proved.""" + if self._source is not None: + raise RuntimeError( + "mirror still holds a borrowed frame output; retain its owner" + ) + + def close(self) -> None: + self.finish_frame() + if not self._images: + return + from cuda.bindings import runtime as cuda + + for image in list(self._images): + if not image.closed: + raise RuntimeError("owned mirror image still has a consumer") + _check_cuda(cuda.cudaSetDevice(image.device)) + _check_cuda(cuda.cudaFree(image.pointer)) + self._images.remove(image) + + def map_cuda_array_rgba(renderer: int, handle: int, max_bytes: int) -> MappedGpuImage: """Retain the native image array, avoiding OVRTX array-to-linear conversion.""" from cuda.bindings import runtime as cuda @@ -164,7 +286,7 @@ def map_cuda_array_rgba(renderer: int, handle: int, max_bytes: int) -> MappedGpu or len(shape) != 3 or shape[2] != 4 ): - raise ValueError("mirror GPU output must be an opaque RGBA CUDA array") + raise ValueError("camera GPU output must be an opaque RGBA CUDA array") info = cuda.cudaArrayGetInfo(int(dl.data)) _check_cuda(info) channel, extent = info[1:3] diff --git a/src/python/isaaccapture/ovxr_session_manager/owned_mirror.py b/src/python/isaaccapture/ovxr_session_manager/owned_mirror.py new file mode 100644 index 0000000000..3f58b8b6dd --- /dev/null +++ b/src/python/isaaccapture/ovxr_session_manager/owned_mirror.py @@ -0,0 +1,655 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Desktop mirror over completed, application-owned eye and camera pixels.""" + +from __future__ import annotations + +import asyncio +import logging +import threading +import time +from dataclasses import dataclass, replace +from typing import TYPE_CHECKING, Callable + +import numpy as np + +from .camera_mailbox import FrameOutputs +from .gpu_output import ( + MappedGpuImage, + OutputMapCleanupError, + OwnedCudaImagePool, + OwnedGpuImage, +) +from .desktop_mirror import ( + DesktopMirrorConfig, + DesktopMirrorStatus, + _CadenceSampler, + _InputCopyCompletion, +) +from .ovxr_session import OvxrFrameConsumer + +_MAX_MIRROR_IMAGE_BYTES = 64 * 1024 * 1024 +_LOGGER = logging.getLogger(__name__) + +if TYPE_CHECKING: + from .ovxr_session import OvxrSessionBinding, OvxrSessionConfig + + +class OwnedDesktopMirror(OvxrFrameConsumer): + """Own the mirror's product selection, image copies, and UI worker.""" + + def __init__( + self, + config: DesktopMirrorConfig, + *, + camera_products: tuple[MirrorProduct, ...] = (), + ) -> None: + self._config = config + self._camera_products = tuple(camera_products) + self._mailbox = OwnedMirrorMailbox( + config.max_fps, gpu_mapper=OwnedCudaImagePool() + ) + self._worker: OwnedDesktopMirrorWorker | None = None + self._start_attempted = False + self._started = False + self._last_status = DesktopMirrorStatus(status_text=config.initial_status_text) + + @property + def render_product_paths(self) -> tuple[str, ...]: + return tuple(product.path for product in self._camera_products) + + def on_session_started( + self, + binding: OvxrSessionBinding, + config: OvxrSessionConfig, + request_shutdown: Callable[[], None], + ) -> None: + eye_paths = config.render_products or ( + "/OVXR/Render/Products/Left", + "/OVXR/Render/Products/Right", + ) + products = ( + MirrorProduct(eye_paths[0], "Left eye"), + MirrorProduct(eye_paths[1], "Right eye"), + *self._camera_products, + ) + self._mailbox.set_products(products) + self._worker = OwnedDesktopMirrorWorker( + self._mailbox.snapshot, + self._config, + request_shutdown, + products=self._mailbox, + ) + + def on_outputs(self, frame: FrameOutputs) -> None: + # The UI owns its GPU copies independently of render-thread consumers. + self._mailbox.on_outputs(frame) + + def on_frame_submitted(self) -> None: + if self._worker is None: + return + # Delay desktop graphics initialization until the first primary image + # has cleared the client's signaling/first-image admission window. + try: + if not self._start_attempted: + self._start_attempted = True + self._worker.start() + self._started = True + if self._started: + self._worker.record_render_frame() + except Exception: + _LOGGER.exception("OVUI desktop mirror update failed") + + def finish_frame(self) -> None: + self._mailbox.finish_frame() + + def record_host_frame(self) -> None: + if self._worker is not None: + self._worker.record_host_frame() + + def set_status_text(self, text: str) -> None: + if self._worker is not None: + self._worker.set_status_text(text) + + @property + def status(self) -> DesktopMirrorStatus: + return self._worker.status if self._worker is not None else self._last_status + + def close(self) -> None: + """Stop the UI, then retire image storage on the renderer owner thread.""" + if self._worker is not None: + try: + self._worker.close() + finally: + self._last_status = self._worker.status + _LOGGER.info("OVUI desktop mirror stopped: %s", self._last_status) + self._worker = None + # A failed UI join must retain its image storage and borrowed providers. + self._mailbox.close() + + +@dataclass(frozen=True) +class MirrorProduct: + """An application-selected production identity, never a per-frame arrival.""" + + path: str + label: str + generation: int = 0 + + +@dataclass(frozen=True) +class OwnedMirrorFrame: + sequence: int + width: int + height: int + rgba: bytes | None + predicted_display_time_ns: int + product: MirrorProduct | None = None + selection_revision: int = 0 + gpu: MappedGpuImage | OwnedGpuImage | None = None + + +class OwnedMirrorMailbox: + """One selected completed output; copies and slow consumers never hold its lock.""" + + def __init__(self, max_fps: float, *, gpu_mapper=None) -> None: + self._period_ns = int(1_000_000_000 / max_fps) + self._next_ns = 0 + self._lock = threading.Lock() + self._products: tuple[MirrorProduct, ...] = () + self._selected: MirrorProduct | None = None + self._revision = 0 + self._sequence = 0 + self._latest: OwnedMirrorFrame | None = None + self._source_key = None + self._gpu_mapper = gpu_mapper + self._mapped: list[MappedGpuImage | OwnedGpuImage] = [] + self._in_use: OwnedMirrorFrame | None = None + self._gpu_error: str | None = None + self._error: str | None = None + + def _invalidate(self) -> None: + self._revision += 1 + self._latest = None + self._source_key = None + self._next_ns = 0 + self._error = None + + def set_products(self, products: tuple[MirrorProduct, ...]) -> None: + """Called only when application production selection/resources change.""" + products = tuple(products) + if len({p.path for p in products}) != len(products) or any( + not p.path.startswith("/") or not p.label for p in products + ): + raise ValueError("mirror products need unique absolute paths and labels") + with self._lock: + if products == self._products: + return + self._products = products + if self._selected not in products: + self._selected = next( + ( + p + for p in products + if self._selected and p.path == self._selected.path + ), + products[0] if products else None, + ) + self._invalidate() + + def select(self, product: MirrorProduct) -> None: + with self._lock: + if product not in self._products: + return # A dropdown event from an already retired production set. + if product != self._selected: + self._selected = product + self._invalidate() + + def reset(self) -> None: + """Invalidate in-flight copies when the application recreates resources.""" + with self._lock: + self._invalidate() + + def view(self): + with self._lock: + return ( + self._products, + self._selected, + self._revision, + self._latest, + self._error, + ) + + def snapshot(self) -> OwnedMirrorFrame | None: + with self._lock: + return self._latest + + def acquire(self, after_sequence: int) -> OwnedMirrorFrame | None: + """Pin only a new image; a GUI reference alone cannot retain a map.""" + with self._lock: + image = self._latest + if image is None or image.sequence <= after_sequence: + return None + if self._in_use is not None: + raise RuntimeError("mirror already has an active consumer") + self._in_use = image + return image + + def release(self, image: OwnedMirrorFrame) -> None: + # Called only after stock OVUI's final source read has completed. + with self._lock: + if self._in_use is not image: + raise RuntimeError("mirror consumer identity changed") + self._in_use = None + + def _drain_gpu(self, *, replace_latest: bool = False) -> None: + # Only the render thread closes maps. No GPU call holds the UI lock. + with self._lock: + if replace_latest and self._latest is not self._in_use: + self._latest = None + keep = [ + f.gpu + for f in (self._latest, self._in_use) + if f is not None and f.gpu is not None + ] + retired = [m for m in self._mapped if not any(m is k for k in keep)] + for mapped in retired: + try: + mapped.close() + except Exception as error: + self._gpu_error = str(error) + raise + self._mapped = [m for m in self._mapped if m is not mapped] + + def close(self) -> None: + """Renderer owner, after joining the consumer and before context destruction.""" + with self._lock: + if self._in_use is not None: + raise RuntimeError("cannot retire an unfinished mirror GPU consumer") + self._invalidate() + self._drain_gpu() + if hasattr(self._gpu_mapper, "close"): + self._gpu_mapper.close() + + def finish_frame(self) -> None: + """Prove the optional copy released all borrowed frame resources.""" + if hasattr(self._gpu_mapper, "finish_frame"): + self._gpu_mapper.finish_frame() + + def report_error(self, error: Exception) -> None: + """Expose an optional consumer failure without changing image ownership.""" + with self._lock: + self._error = str(error) + + def enable_gpu_capture(self) -> None: + """Start optional captures after the Vulkan provider is ready.""" + if hasattr(self._gpu_mapper, "enable"): + self._gpu_mapper.enable() + + def on_outputs(self, frame: FrameOutputs) -> None: + """Copy the selected output, sharing CPU snapshots within this frame.""" + now = time.perf_counter_ns() + with self._lock: + product, revision = self._selected, self._revision + previous_key = self._source_key + gpu = None + try: + if self._gpu_error is not None: + raise RuntimeError(self._gpu_error) + self._drain_gpu() + if product is None or now < self._next_ns: + return + result = frame.latest_color(product.path) + if result is None: + return # Cadence gap preserves both entry and last image. + handle, sensor_time = result + key = (handle, sensor_time) + if key == previous_key: + return + if self._gpu_mapper is not None: + self._drain_gpu(replace_latest=True) + if len(self._mapped) >= 2: + return # Bounded latest/in-use storage; never wait for GUI. + gpu = frame.map_gpu(handle, _MAX_MIRROR_IMAGE_BYTES, self._gpu_mapper) + if gpu is None: + return # The Vulkan provider is not ready yet. + self._mapped.append(gpu) + width, height, rgba = gpu.width, gpu.height, None + else: + width, height, rgba = frame.copy_host_rgba( + handle, _MAX_MIRROR_IMAGE_BYTES + ) + if ( + width <= 0 + or height <= 0 + or ( + gpu is None + and ( + len(rgba) != width * height * 4 + or len(rgba) > _MAX_MIRROR_IMAGE_BYTES + ) + ) + ): + raise ValueError("mirror output is not bounded contiguous RGBA8") + rgba = bytes(rgba) if rgba is not None else None + with self._lock: + if (product, revision) != (self._selected, self._revision): + return # Selection/reset changed while native copy ran. + self._sequence += 1 + self._latest = OwnedMirrorFrame( + self._sequence, + width, + height, + rgba, + int(sensor_time * 1e9), + product, + revision, + gpu, + ) + self._source_key = key + self._next_ns = now + self._period_ns + self._error = None + except Exception as error: + if isinstance(error, OutputMapCleanupError): + self._mapped.append(error.image) + self._gpu_error = str(error) + with self._lock: + if (product, revision) == (self._selected, self._revision): + self._latest = None + self._source_key = None + self._error = str(error) + + +class OwnedDesktopMirrorWorker: + """OVUI consumes only immutable snapshots supplied by the frame owner.""" + + def __init__( + self, + acquire: Callable[[], OwnedMirrorFrame | None], + config: DesktopMirrorConfig, + on_user_close: Callable[[], None] | None = None, + *, + products: OwnedMirrorMailbox | None = None, + ) -> None: + self._products = products + self._acquire = acquire + self._config = config + self._on_user_close = on_user_close + self._stop = threading.Event() + self._thread: threading.Thread | None = None + self._lock = threading.Lock() + self._status = DesktopMirrorStatus(status_text=config.initial_status_text) + self._cadence = _CadenceSampler() + + def _update(self, **changes: object) -> None: + with self._lock: + self._status = replace(self._status, **changes) + + def start(self) -> None: + if self._thread is not None: + raise RuntimeError("desktop mirror is already started") + self._thread = threading.Thread( + target=self._run, name="isaacteleop-ovui-owned-mirror" + ) + self._thread.start() + + def _run(self) -> None: + self._update(thread_id=threading.get_native_id()) + ui = None + window = None + failure = None + completions = {} + try: + import omni.ui as ui + + ui.init( + self._config.title, + width=self._config.window_width, + height=self._config.window_width, + max_fps=self._config.max_fps, + ) + if ui.standalone.get_frame_pacing_status().get("backend") != "vulkan": + raise RuntimeError( + "Desktop mirror requires OVUI's Vulkan backend; remove any " + "OMNIUI_BACKEND override before initializing OVUI" + ) + if self._products is not None: + if not ui.has_gpu_byte_image(): + raise RuntimeError( + "OVUI backend does not support CUDA image uploads" + ) + self._products.enable_gpu_capture() + provider = ui.ByteImageProvider() + window = ui.Window( + self._config.title, + width=self._config.window_width, + height=self._config.window_width, + fill_app_window=True, + flags=ui.WINDOW_FLAGS_NO_TITLE_BAR, + raster_policy=ui.RasterPolicy.NEVER, + ) + with window.frame: + with ui.VStack(): + selector_frame = ui.Frame(height=30) if self._products else None + status_label = ui.Label( + self.status.status_text, + height=48, + word_wrap=True, + style={"color": 0xFFFFFFFF, "font_size": 18}, + ) + with ui.ZStack(): + ui.Rectangle(style={"background_color": 0xFF181818}) + image_widget = ui.ImageWithProvider( + provider, + fill_policy=ui.IwpFillPolicy.IWP_PRESERVE_ASPECT_FIT, + ) + waiting_label = ui.Label("Waiting for a completed image") + self._update(window_opened=True) + + def rebuild_selector(products, selected): + def choose(model, _item): + index = model.get_item_value_model().get_value_as_int() + if 0 <= index < len(products): + previous_revision = self._products.view()[2] + self._products.select(products[index]) + if self._products.view()[2] != previous_revision: + image_widget.visible = False + waiting_label.visible = True + waiting_label.text = f"Waiting for {products[index].label}" + + def build(): + if products: + combo = ui.ComboBox( + products.index(selected), *(p.label for p in products) + ) + combo.model.add_item_changed_fn(choose) + else: + ui.Label("No selected render products") + + selector_frame.set_build_fn(build) + selector_frame.rebuild() + + async def drive() -> None: + nonlocal failure + last = 0 + next_status_ns = 0 + shown_products = None + shown_selected = None + shown_revision = None + try: + while not self._stop.is_set(): + # Status belongs to the application, not to a selected + # image. Update while waiting or paused, on the UI thread. + now_ns = time.perf_counter_ns() + if now_ns >= next_status_ns: + status_label.text = self.status.status_text + next_status_ns = now_ns + 250_000_000 + if self._products is not None: + products, selected, revision, _latest, error = ( + self._products.view() + ) + if products != shown_products or selected != shown_selected: + rebuild_selector(products, selected) + shown_products = products + shown_selected = selected + # The provider owns its uploaded texture. A producer + # replacement gap must not hide it; identity or errors do. + if revision != shown_revision or error is not None: + image_widget.visible = False + waiting_label.visible = True + waiting_label.text = error or ( + f"Waiting for {selected.label}" + if selected + else "No selected render products" + ) + shown_revision = revision + image = self._products.acquire(last) + else: + image = self._acquire() + if image is not None and image.sequence > last: + safe_to_release = image.gpu is None + try: + if image.gpu is not None: + gpu = image.gpu + if gpu.closed: + raise RuntimeError( + "mirror GPU map already retired" + ) + if gpu.device not in completions: + completions[gpu.device] = _InputCopyCompletion( + gpu.device + ) + completion = completions[gpu.device] + try: + provider.set_bytes_data_from_gpu( + gpu.pointer, + [image.width, image.height], + ui.TextureFormat.RGBA8_UNORM, + gpu.pitch, + ) + finally: + completion.wait() + safe_to_release = True + else: + if ( + image.width <= 0 + or image.height <= 0 + or len(image.rgba) + != image.width * image.height * 4 + ): + raise ValueError( + "mirror snapshot is not tightly packed RGBA8" + ) + pixels = np.frombuffer(image.rgba, dtype=np.uint8) + provider.set_data_array( + pixels, [image.width, image.height] + ) + finally: + if self._products is not None and safe_to_release: + self._products.release(image) + last = image.sequence + current = self._products.view() if self._products else None + valid = current is None or ( + image.product == current[1] + and image.selection_revision == current[2] + and current[4] is None + ) + image_widget.visible = valid + waiting_label.visible = not valid + with self._lock: + status = self._status + if valid: + self._update( + presented_frames=status.presented_frames + 1, + last_sequence=last, + ) + await asyncio.sleep(1.0 / self._config.max_fps) + except Exception as error: + # Preserve upload errors so they are not reported as a + # user closing the mirror window. + failure = error + self._update(error=str(error)) + self._stop.set() + + # Own the loop so window closure cannot strand the update task. + async def run(): + tasks = ( + asyncio.create_task(drive()), + asyncio.create_task(ui.run_async()), + ) + try: + done, _ = await asyncio.wait( + tasks, return_when=asyncio.FIRST_COMPLETED + ) + await asyncio.gather(*done) + finally: + for task in tasks: + task.cancel() + await asyncio.gather(*tasks, return_exceptions=True) + + asyncio.run(run()) + except BaseException as error: + failure = error + self._update(error=str(error)) + finally: + + def record_cleanup_error(error): + nonlocal failure + # Interruption must not be hidden by an earlier optional UI error. + if failure is None or not isinstance(error, Exception): + failure = error + self._update(error=str(error)) + + # Event destruction must leave the UI's import device current. + for completion in completions.values(): + try: + completion.close() + except BaseException as error: + record_cleanup_error(error) + if window is not None: + try: + window.destroy() + except BaseException as error: + record_cleanup_error(error) + if ui is not None: + try: + ui.shutdown() + except BaseException as error: + record_cleanup_error(error) + self._update(closed=True) + if failure is None and not self._stop.is_set() and self._on_user_close: + try: + self._on_user_close() + except Exception as error: + record_cleanup_error(error) + + # Cleanup must run for every exit, but cancellation/process exit is not + # an optional mirror failure. Propagate it after retirement. + if failure is not None and not isinstance(failure, Exception): + raise failure + + def close(self, timeout_s: float = 10.0) -> None: + if self._thread is None: + return + self._stop.set() + self._thread.join(timeout_s) + if self._thread.is_alive(): + raise TimeoutError("owned desktop mirror worker did not stop") + + def record_host_frame(self) -> None: + with self._lock: + self._cadence.record_host_frame(time.perf_counter_ns()) + + def record_render_frame(self) -> None: + with self._lock: + self._cadence.record_render_frame(time.perf_counter_ns()) + + def set_status_text(self, text: str) -> None: + self._update(status_text=str(text)) + + @property + def status(self) -> DesktopMirrorStatus: + with self._lock: + host, render = self._cadence.rates(time.perf_counter_ns()) + return replace(self._status, main_thread_fps=host, render_thread_fps=render) diff --git a/tests/python/core/ovxr_session_manager/conftest.py b/tests/python/core/ovxr_session_manager/conftest.py new file mode 100644 index 0000000000..52df025494 --- /dev/null +++ b/tests/python/core/ovxr_session_manager/conftest.py @@ -0,0 +1,65 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Isolated provider doubles for CPU-only mirror lifecycle tests.""" + +import asyncio +import sys +from types import ModuleType, SimpleNamespace +from unittest.mock import Mock + +import pytest + + +@pytest.fixture +def mock_omni_ui(monkeypatch): + def install(ui): + async def run_async(): + await asyncio.Future() + + if ui is not None and not hasattr(ui, "run_async"): + ui.run_async = run_async + if ui is not None and not hasattr(ui, "has_gpu_byte_image"): + ui.has_gpu_byte_image = lambda: True + if ui is not None and not hasattr(ui, "standalone"): + ui.standalone = SimpleNamespace( + get_frame_pacing_status=lambda: {"backend": "vulkan"} + ) + omni = ModuleType("omni") + omni.__path__ = [] + omni.ui = ui + monkeypatch.setitem(sys.modules, "omni", omni) + monkeypatch.setitem(sys.modules, "omni.ui", ui) + + return install + + +@pytest.fixture +def mock_cuda(monkeypatch): + cuda = ModuleType("cuda") + cuda.__path__ = [] + bindings = ModuleType("cuda.bindings") + bindings.__path__ = [] + runtime = ModuleType("cuda.bindings.runtime") + runtime.cudaStreamLegacy = 1 + runtime.cudaEventDisableTiming = 2 + runtime.cudaMemcpyKind = SimpleNamespace(cudaMemcpyDeviceToDevice=3) + for name in ( + "cudaSetDevice", + "cudaGetDevice", + "cudaMallocPitch", + "cudaFree", + "cudaMemcpy2DAsync", + "cudaMemcpy2DFromArrayAsync", + "cudaStreamSynchronize", + "cudaEventCreateWithFlags", + "cudaEventRecord", + "cudaEventSynchronize", + "cudaEventDestroy", + ): + setattr(runtime, name, Mock(side_effect=AssertionError(f"Unexpected {name}"))) + cuda.bindings = bindings + bindings.runtime = runtime + for module in (cuda, bindings, runtime): + monkeypatch.setitem(sys.modules, module.__name__, module) + return runtime diff --git a/tests/python/core/ovxr_session_manager/test_desktop_mirror.py b/tests/python/core/ovxr_session_manager/test_desktop_mirror.py new file mode 100644 index 0000000000..1db35510e8 --- /dev/null +++ b/tests/python/core/ovxr_session_manager/test_desktop_mirror.py @@ -0,0 +1,122 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Shared mirror configuration and the production input-copy fence.""" + +from __future__ import annotations + +import subprocess +import sys +import textwrap +from pathlib import Path + +import pytest + +import isaaccapture.ovxr_session_manager.desktop_mirror as desktop_mirror + + +def test_config_rejects_invalid_window_and_cadence(): + assert desktop_mirror.DesktopMirrorConfig().max_fps == 30.0 + for fps in (0.0, float("nan"), float("inf")): + with pytest.raises(ValueError, match="max_fps"): + desktop_mirror.DesktopMirrorConfig(max_fps=fps) + with pytest.raises(ValueError, match="window_width"): + desktop_mirror.DesktopMirrorConfig(window_width=0) + with pytest.raises(ValueError, match="title"): + desktop_mirror.DesktopMirrorConfig(title="") + + +def test_session_import_and_mirror_config_do_not_request_optional_ovui(): + script = textwrap.dedent( + """ + import importlib.abc + import sys + from pathlib import Path + + # CTest supplies the built package via PYTHONPATH, which isolated mode ignores. + # Select that exact package explicitly without inheriting parent import state. + sys.path.insert(0, sys.argv[1]) + requests = [] + + class UnavailableOVUI(importlib.abc.MetaPathFinder): + def find_spec(self, fullname, path=None, target=None): + if fullname == "omni.ui" or fullname.startswith("omni.ui."): + requests.append(fullname) + raise ModuleNotFoundError(fullname) + + sys.meta_path.insert(0, UnavailableOVUI()) + import isaaccapture + from isaaccapture.ovxr_session_manager import DesktopMirrorConfig + + assert Path(isaaccapture.__file__).resolve().parent.parent == Path(sys.argv[1]) + + DesktopMirrorConfig() + assert not requests, requests + assert "omni.ui" not in sys.modules + """ + ) + result = subprocess.run( + [ + sys.executable, + "-I", + "-c", + script, + str(Path(desktop_mirror.__file__).resolve().parents[2]), + ], + capture_output=True, + text=True, + # Keep child diagnostics in the assertion output on failure. + check=False, + timeout=30, + ) + assert result.returncode == 0, result.stdout + result.stderr + + +@pytest.mark.parametrize("fail_wait", [False, True]) +def test_input_copy_completion_uses_legacy_stream_and_retires_event_once( + monkeypatch, fail_wait, mock_cuda +): + cuda = mock_cuda + + calls = [] + monkeypatch.setattr(cuda, "cudaGetDevice", lambda: (0, 0)) + monkeypatch.setattr( + cuda, "cudaSetDevice", lambda device: (calls.append(("device", device)) or 0,) + ) + monkeypatch.setattr( + cuda, + "cudaEventCreateWithFlags", + lambda flags: (calls.append(("create", flags)) or 0, 7), + ) + monkeypatch.setattr( + cuda, + "cudaEventRecord", + lambda event, stream: (calls.append(("record", event, stream)) or 0,), + ) + monkeypatch.setattr( + cuda, + "cudaEventSynchronize", + lambda event: (calls.append(("wait", event)) or int(fail_wait),), + ) + monkeypatch.setattr( + cuda, "cudaEventDestroy", lambda event: (calls.append(("destroy", event)) or 0,) + ) + + completion = desktop_mirror._InputCopyCompletion(3) + if fail_wait: + with pytest.raises(RuntimeError, match="input-copy completion"): + completion.wait() + else: + completion.wait() + completion.close() + completion.close() + assert calls == [ + ("device", 3), + ("create", cuda.cudaEventDisableTiming), + ("device", 0), + ("device", 3), + ("record", 7, cuda.cudaStreamLegacy), + ("wait", 7), + ("device", 0), + ("destroy", 7), + ] diff --git a/tests/python/core/ovxr_session_manager/test_gpu_mirror.py b/tests/python/core/ovxr_session_manager/test_gpu_mirror.py new file mode 100644 index 0000000000..5408c587e9 --- /dev/null +++ b/tests/python/core/ovxr_session_manager/test_gpu_mirror.py @@ -0,0 +1,414 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +import sys +import threading +from types import SimpleNamespace + +import pytest + +from isaaccapture.ovxr_session_manager.camera_mailbox import FrameOutputs + +from isaaccapture.ovxr_session_manager import owned_mirror as mirror +from isaaccapture.ovxr_session_manager.desktop_mirror import DesktopMirrorConfig +from test_passive_mirror import LEFT, RIGHT, outputs + + +class FakeMap: + width = height = 1 + pitch = 4 + device = 0 + + def __init__(self, handle): + self.pointer = handle + self.closed = False + self.owner = threading.get_ident() + + def close(self): + assert threading.get_ident() == self.owner + assert not self.closed + self.closed = True + + +def mailbox(): + mapped = [] + + def map_image(_renderer, handle, _limit): + m = FakeMap(handle) + mapped.append(m) + return m + + box = mirror.OwnedMirrorMailbox(30, gpu_mapper=map_image) + box.set_products((LEFT, RIGHT)) + return box, mapped + + +def publish(box, handle): + box._next_ns = 0 + box.on_outputs(FrameOutputs(1, outputs(Left=handle, Right=handle + 100))) + + +def test_slow_gpu_consumer_retains_map_and_drops_only_unused_pending_maps(): + box, maps = mailbox() + publish(box, 1) + first = box.acquire(0) + publish(box, 2) + publish(box, 3) + assert len(box._mapped) == 2 + assert not maps[0].closed and maps[1].closed and not maps[2].closed + assert box.snapshot().gpu.pointer == 3 and box.snapshot().rgba is None + box.select(RIGHT) + assert box.snapshot() is None and not maps[0].closed + publish(box, 4) + assert maps[2].closed and box.snapshot().gpu.pointer == 104 + assert box.snapshot().product == RIGHT + with pytest.raises(RuntimeError, match="unfinished"): + box.close() + box.release(first) + box.close() + assert all(m.closed for m in maps) and not box._mapped + + +def test_gpu_selection_change_during_map_does_not_relabel_or_leak(): + box, maps = mailbox() + mapper = box._gpu_mapper + + def switch_during_map(*args): + value = mapper(*args) + box.select(RIGHT) + return value + + box._gpu_mapper = switch_during_map + publish(box, 1) + assert box.snapshot() is None + box.close() + assert all(m.closed for m in maps) + + +def test_gui_device_bootstrap_omits_copy_without_a_spurious_failure(): + from isaaccapture.ovxr_session_manager.gpu_output import OwnedCudaImagePool + + pool = OwnedCudaImagePool() + box = mirror.OwnedMirrorMailbox(30, gpu_mapper=pool) + box.set_products((LEFT, RIGHT)) + publish(box, 1) + assert box.snapshot() is None and box.view()[4] is None + assert not pool._images and pool._source is None + box.close() + + +@pytest.mark.parametrize( + "upload_error,wait_error,close_error", + [ + (False, False, False), + (True, False, False), + (False, True, False), + (False, False, True), + ], +) +def test_gpu_upload_release_is_after_completion_and_failed_wait_keeps_map( + monkeypatch, upload_error, wait_error, close_error, mock_omni_ui +): + box, maps = mailbox() + publish(box, 1) + events = [] + maps[0].device = 1 + current_device = [0] + + def set_device(device): + current_device[0] = device + return (0,) + + def shutdown(): + assert current_device[0] == 0 + events.append("shutdown") + + worker = mirror.OwnedDesktopMirrorWorker( + box.snapshot, DesktopMirrorConfig(), products=box + ) + + class Context: + def __init__(self, *_args, **_kwargs): + pass + + def __enter__(self): + return self + + def __exit__(self, *_args): + return False + + def destroy(self): + assert current_device[0] == 0 + events.append("destroy") + + def set_build_fn(self, _fn): + pass + + def rebuild(self): + pass + + Context.frame = Context() + + class Provider: + def set_bytes_data_from_gpu(self, pointer, sizes, _format, pitch): + assert pointer == 1 and sizes == [1, 1] and pitch == 4 + assert box._in_use is not None and not maps[0].closed + events.append("upload") + worker._stop.set() + if upload_error: + raise RuntimeError("upload failed") + + def set_data_array(self, *_args): + pytest.fail("GPU image made a CPU round trip") + + class Completion: + _cuda = SimpleNamespace(cudaSetDevice=set_device) + + def __init__(self, _device): + pass + + def _check(self, _result): + pass + + def wait(self): + assert box._in_use is not None and not maps[0].closed + events.append("complete") + if wait_error: + raise RuntimeError("completion failed") + + def close(self): + if close_error: + raise RuntimeError("event cleanup failed") + + ui = SimpleNamespace( + init=lambda *_a, **_k: None, + shutdown=shutdown, + ByteImageProvider=Provider, + Window=Context, + Frame=Context, + ZStack=Context, + VStack=Context, + Label=lambda text, **_k: SimpleNamespace(text=text), + Rectangle=lambda **_k: None, + ImageWithProvider=lambda *_a, **_k: SimpleNamespace(visible=False), + WINDOW_FLAGS_NO_TITLE_BAR=1, + RasterPolicy=SimpleNamespace(NEVER=1), + IwpFillPolicy=SimpleNamespace(IWP_PRESERVE_ASPECT_FIT=1), + TextureFormat=SimpleNamespace(RGBA8_UNORM=1), + ) + mock_omni_ui(ui) + monkeypatch.setattr(mirror, "_InputCopyCompletion", Completion) + worker._run() + assert ( + events == ["upload", "complete", "destroy", "shutdown"] and worker.status.closed + ) + if wait_error: + assert box._in_use is not None and worker.status.error == "completion failed" + with pytest.raises(RuntimeError, match="unfinished"): + box.close() + assert not maps[0].closed + else: + assert box._in_use is None + assert worker.status.error == ( + "upload failed" + if upload_error + else "event cleanup failed" + if close_error + else None + ) + box.close() + assert maps[0].closed + + +def test_retirement_failure_is_contained_and_stops_mapping(): + box, maps = mailbox() + publish(box, 1) + + def fail(): + raise RuntimeError("unmap failed") + + maps[0].close = fail + publish(box, 2) + assert box.snapshot() is None and "unmap failed" in box.view()[4] + publish(box, 3) + assert len(maps) == 1 and box._mapped == maps + with pytest.raises(RuntimeError, match="unmap failed"): + box.close() + + +def test_failed_map_cleanup_is_retained_and_prevents_additional_maps(): + box, maps = mailbox() + + def fail(*_args): + image = FakeMap(1) + maps.append(image) + raise mirror.OutputMapCleanupError(image, "invalid tensor", "unmap failed") + + box._gpu_mapper = fail + publish(box, 1) + publish(box, 2) + assert len(maps) == 1 and box._mapped == maps + assert box.snapshot() is None and "cleanup failed" in box.view()[4] + box.close() + assert maps[0].closed + + +@pytest.mark.parametrize("stream,event", [(0, 7), (5, 0), (1, 7), (0, 0)]) +def test_producer_event_and_stream_are_both_synchronized(stream, event): + from isaaccapture.ovxr_session_manager.gpu_output import _wait_ready + + calls = [] + cuda = SimpleNamespace( + cudaStreamLegacy=999, + cudaEventSynchronize=lambda value: (calls.append(("event", value)) or 0,), + cudaStreamSynchronize=lambda value: (calls.append(("stream", value)) or 0,), + ) + _wait_ready(cuda, SimpleNamespace(stream=stream, wait_event=event)) + expected = ([("event", event)] if event else []) + ( + [("stream", 999 if stream == 1 else stream)] if stream else [] + ) + assert calls == expected + + +@pytest.mark.parametrize("copy_failure", [False, True]) +@pytest.mark.parametrize("source_device", [0, 1]) +def test_owned_mirror_copy_releases_frame_maps_and_reuses_two_allocations( + monkeypatch, copy_failure, source_device, mock_cuda +): + cuda = mock_cuda + from isaaccapture.ovxr_session_manager import gpu_output as gpu + + events = [] + allocations = [] + sources = [] + + def mapped(*args): + source = SimpleNamespace( + width=2, + height=3, + device=source_device, + pointer=42, + busy=False, + closed=False, + ) + + def close(): + assert not source.busy + source.closed = True + events.append("unmap") + + source.close = close + sources.append(source) + return source + + def allocate(*args): + allocations.append(len(allocations) + 100) + return 0, allocations[-1], 8 + + monkeypatch.setattr(gpu, "map_cuda_array_rgba", mapped) + monkeypatch.setattr(cuda, "cudaSetDevice", lambda device: (0,)) + monkeypatch.setattr(cuda, "cudaMallocPitch", allocate) + monkeypatch.setattr(cuda, "cudaFree", lambda pointer: (events.append("free") or 0,)) + monkeypatch.setattr( + cuda, + "cudaMemcpy2DFromArrayAsync", + lambda *args: (events.append("copy") or int(copy_failure),), + ) + + monkeypatch.setattr( + cuda, "cudaStreamSynchronize", lambda *args: (events.append("complete") or 0,) + ) + pool = gpu.OwnedCudaImagePool() + pool.enable() + if copy_failure: + with pytest.raises(RuntimeError, match="CUDA"): + pool(1, 2, 100) + assert sources[-1].closed + pool.finish_frame() + else: + first = pool(1, 2, 100) + assert first.device == source_device + second = pool(1, 3, 100) + assert first.pointer != second.pointer and all( + source.closed for source in sources + ) + assert events[:3] == ["copy", "complete", "unmap"] + with pytest.raises(RuntimeError, match="both owned"): + pool(1, 4, 100) + first.close() + replacement = pool(1, 5, 100) + assert replacement.pointer == first.pointer and len(allocations) == 2 + replacement.close() + second.close() + pool.close() + assert events.count("free") == len(allocations) + + +def test_oversized_pitched_allocation_stays_rejected_on_reuse(monkeypatch, mock_cuda): + cuda = mock_cuda + from isaaccapture.ovxr_session_manager import gpu_output as gpu + + events = [] + source = SimpleNamespace(width=1, height=2, device=0, pointer=42, busy=False) + source.close = lambda: events.append("unmap") + monkeypatch.setattr(gpu, "map_cuda_array_rgba", lambda *args: source) + monkeypatch.setattr(cuda, "cudaSetDevice", lambda *args: (0,)) + monkeypatch.setattr(cuda, "cudaMallocPitch", lambda *args: (0, 100, 256)) + monkeypatch.setattr(cuda, "cudaFree", lambda *args: (events.append("free") or 0,)) + monkeypatch.setattr( + cuda, "cudaMemcpy2DFromArrayAsync", lambda *args: pytest.fail("oversized copy") + ) + pool = gpu.OwnedCudaImagePool() + pool.enable() + for handle in (1, 2): + with pytest.raises(RuntimeError, match="byte bound"): + pool(1, handle, 100) + pool.finish_frame() + pool.close() + assert events == ["unmap", "unmap", "free"] + + +def test_unproved_owned_copy_keeps_borrowed_source_and_blocks_consumption( + monkeypatch, mock_cuda +): + cuda = mock_cuda + from isaaccapture.ovxr_session_manager import gpu_output as gpu + + source = SimpleNamespace(width=1, height=1, device=0, pointer=42, busy=False) + source.close = lambda: pytest.fail("unproved source read was unmapped") + monkeypatch.setattr(gpu, "map_cuda_array_rgba", lambda *args: source) + monkeypatch.setattr(cuda, "cudaSetDevice", lambda *args: (0,)) + monkeypatch.setattr(cuda, "cudaMallocPitch", lambda *args: (0, 100, 4)) + monkeypatch.setattr(cuda, "cudaMemcpy2DFromArrayAsync", lambda *args: (0,)) + monkeypatch.setattr(cuda, "cudaStreamSynchronize", lambda *args: (1,)) + pool = gpu.OwnedCudaImagePool() + pool.enable() + with pytest.raises(RuntimeError, match="CUDA"): + pool(1, 2, 100) + assert source.busy + with pytest.raises(RuntimeError, match="borrowed frame"): + pool.finish_frame() + with pytest.raises(RuntimeError, match="borrowed frame"): + pool.close() + + +def test_unconfigured_pool_needs_no_optional_providers(monkeypatch): + from isaaccapture.ovxr_session_manager.gpu_output import OwnedCudaImagePool + + for name in ("cuda", "cuda.bindings", "cuda.bindings.runtime", "ovxr"): + monkeypatch.setitem(sys.modules, name, None) + pool = OwnedCudaImagePool() + assert pool(1, 2, 100) is None + pool.close() + + +def test_empty_pool_still_retains_unproved_borrowed_sources(monkeypatch): + from isaaccapture.ovxr_session_manager.gpu_output import OwnedCudaImagePool + + monkeypatch.setitem(sys.modules, "cuda", None) + pool = OwnedCudaImagePool() + borrowed = object() + pool._source = borrowed + with pytest.raises(RuntimeError, match="retain"): + pool.close() + assert pool._source is borrowed diff --git a/tests/python/core/ovxr_session_manager/test_ovxr_session.py b/tests/python/core/ovxr_session_manager/test_ovxr_session.py index f8daca3c0a..f538ef67f2 100644 --- a/tests/python/core/ovxr_session_manager/test_ovxr_session.py +++ b/tests/python/core/ovxr_session_manager/test_ovxr_session.py @@ -19,6 +19,11 @@ CameraPanelConfig, CameraPanelConsumer, ) +from isaaccapture.ovxr_session_manager.desktop_mirror import ( + DesktopMirrorConfig, + DesktopMirrorStatus, +) +import isaaccapture.ovxr_session_manager.owned_mirror as owned_mirror import isaaccapture.ovxr_session_manager.ovxr_session as ovxr_session from isaaccapture.ovxr_session_manager import frame_pump, runtime_owner @@ -252,9 +257,13 @@ def abort(self): self.events.append("abort") -def _session_with_consumers(host, config, *, frame_consumers): +def _session_with_consumers( + host, config, *, frame_consumers, runtime_launcher_factory=None +): host.frame_consumers = frame_consumers - return ovxr_session.OvxrSession(host, config) + return ovxr_session.OvxrSession( + host, config, runtime_launcher_factory=runtime_launcher_factory + ) @pytest.fixture(autouse=True) @@ -1200,6 +1209,178 @@ def close(self): assert "destroy" in [name for name, _ in _FakeContext.operations] +def test_optional_mirror_enables_readback_and_closes_before_context(monkeypatch): + host = _FakeHost() + mirror_events = [] + acquired = [] + + class FakeMirror: + status = DesktopMirrorStatus() + + def __init__(self, acquire_frame): + self.acquire_frame = acquire_frame + + def start(self): + acquired.append(self.acquire_frame()) + mirror_events.append(("start", threading.get_native_id())) + + def record_host_frame(self): + mirror_events.append(("host",)) + + def set_status_text(self, text): + mirror_events.append(("status", text)) + + def record_render_frame(self): + mirror_events.append(("render", threading.get_native_id())) + + def close(self): + assert "destroy" not in [name for name, _thread in _FakeContext.operations] + mirror_events.append(("close", threading.get_native_id())) + + monkeypatch.setattr( + owned_mirror, + "OwnedDesktopMirrorWorker", + lambda acquire_frame, config, _on_user_close, *, products=None: ( + mirror_events.append(("create", threading.get_native_id(), config)) + or FakeMirror(acquire_frame) + ), + ) + config = _config() + mirror_config = DesktopMirrorConfig() + consumer = owned_mirror.OwnedDesktopMirror(mirror_config) + owner = _session_with_consumers(host, config, frame_consumers=(consumer,)) + + owner.start() + context = owner._pump._context + assert not getattr(context.desc, "enable_frame_readback", False) + assert context.desc.render_products == list(config.render_products) + assert _FakeContext.selected_products == () + assert [p.path for p in consumer._mailbox.view()[0]] == [ + "/OVXR/Render/Products/Left", + "/OVXR/Render/Products/Right", + ] + assert consumer.status == DesktopMirrorStatus() + consumer.record_host_frame() + consumer.set_status_text("Application status") + assert owner.request_render("step-1") > 0 + _wait_for_operation("submit") + owner.close() + + owner_thread = owner.render_worker_native_thread_id + assert mirror_events[0:2] == [ + ("create", owner_thread, mirror_config), + ("host",), + ] + assert ("start", owner_thread) in mirror_events + assert mirror_events.index(("start", owner_thread)) < mirror_events.index( + ("render", owner_thread) + ) + assert ("status", "Application status") in mirror_events + assert ("render", owner_thread) in mirror_events + assert mirror_events[-1] == ("close", owner_thread) + assert acquired == [None] + assert consumer.status == DesktopMirrorStatus() + + +def test_status_mirror_starts_after_cloudxr_context_creation(monkeypatch): + host = _FakeHost() + events = [] + + class FakeMirror: + status = DesktopMirrorStatus() + + def start(self): + assert any(name == "create" for name, _thread in _FakeContext.operations) + events.append("mirror-start") + + def close(self): + events.append("mirror-close") + + class FakeLauncher: + def __init__(self): + events.append("cloudxr-start") + + def interrupt_runtime(self): + events.append("cloudxr-interrupt") + + def stop(self): + events.append("cloudxr-stop") + + monkeypatch.setattr( + owned_mirror, + "OwnedDesktopMirrorWorker", + lambda _acquire_frame, _config, _on_user_close, *, products=None: FakeMirror(), + ) + + def launcher_factory(_path, *, accept_eula=False): + return FakeLauncher() + + config = _config() + mirror_config = DesktopMirrorConfig() + consumer = owned_mirror.OwnedDesktopMirror(mirror_config) + config.cloudxr_env_file = "/cloudxr.env" + config.auto_launch_cloudxr = True + owner = _session_with_consumers( + host, + config, + runtime_launcher_factory=launcher_factory, + frame_consumers=(consumer,), + ) + + owner.start() + assert owner.request_render("step-1") > 0 + _wait_for_operation("submit") + deadline = time.monotonic() + 1.0 + while "mirror-start" not in events and time.monotonic() < deadline: + time.sleep(0.001) + owner.close() + + assert events.index("cloudxr-start") < events.index("mirror-start") + assert "cloudxr-interrupt" not in events + assert events.index("mirror-close") < events.index("cloudxr-stop") + + +def test_stuck_mirror_prevents_borrowed_context_teardown(monkeypatch): + host = _FakeHost() + + class FakeMirror: + status = DesktopMirrorStatus() + + def start(self): + pass + + def close(self): + raise TimeoutError("mirror stuck") + + mirror = FakeMirror() + monkeypatch.setattr( + owned_mirror, + "OwnedDesktopMirrorWorker", + lambda _context, _config, _on_user_close, *, products=None: mirror, + ) + config = _config() + mirror_config = DesktopMirrorConfig() + consumer = owned_mirror.OwnedDesktopMirror(mirror_config) + owner = _session_with_consumers(host, config, frame_consumers=(consumer,)) + + owner.start() + context = owner._pump._context + with pytest.raises(TimeoutError, match="mirror stuck"): + owner.close() + + assert owner._pump._context is context + assert consumer._worker is mirror + assert not owner.retirement_complete + with pytest.raises(TimeoutError, match="mirror stuck"): + owner.close() + assert "destroy" not in [name for name, _thread in _FakeContext.operations] + assert not any( + event[0] in {"host-close", "environment-close"} + for event in host.events + if isinstance(event, tuple) + ) + + def test_quad_detach_failure_reaches_close_and_retains_borrowed_owners(monkeypatch): host = _FakeHost() panel = CameraPanelConsumer((CameraPanelConfig("/Render/Head"),)) @@ -1238,6 +1419,154 @@ def fail_detach(): ) +def test_user_closing_mirror_requests_application_shutdown_before_owner_teardown( + monkeypatch, +): + host = _FakeHost() + mirror_events = [] + + class FakeMirror: + status = DesktopMirrorStatus(closed=True) + + def __init__(self, on_user_close): + self.on_user_close = on_user_close + + def start(self): + mirror_events.append(("start", threading.get_native_id())) + + def close(self): + mirror_events.append(("close", threading.get_native_id())) + + mirror = None + + def create_mirror(_acquire_frame, _config, on_user_close, *, products=None): + nonlocal mirror + mirror = FakeMirror(on_user_close) + return mirror + + monkeypatch.setattr(owned_mirror, "OwnedDesktopMirrorWorker", create_mirror) + config = _config() + mirror_config = DesktopMirrorConfig() + consumer = owned_mirror.OwnedDesktopMirror(mirror_config) + owner = _session_with_consumers(host, config, frame_consumers=(consumer,)) + owner.start() + + assert mirror is not None + assert owner.request_render("step-1") > 0 + _wait_for_operation("submit") + deadline = time.monotonic() + 1.0 + while not mirror_events and time.monotonic() < deadline: + time.sleep(0.001) + assert mirror_events and mirror_events[0][0] == "start" + mirror.on_user_close() + assert owner.shutdown_requested + assert owner._phase is not ovxr_session._Phase.CLOSED + assert not host.publication_closed + + owner.close() + assert host.publication_closed + assert [name for name, _thread in _FakeContext.operations].count("destroy") == 1 + assert host.events[-2:] == [ + ("host-close", host.attach_thread), + ("environment-close", host.attach_thread), + ] + assert [name for name, _thread in mirror_events] == ["start", "close"] + + +@pytest.mark.parametrize("close_from_mirror", [False, True]) +def test_shutdown_request_exit_wakes_frame_wait_without_runtime_interrupt( + monkeypatch, caplog, close_from_mirror +): + host = _FakeHost() + frame_wait_entered = threading.Event() + runtime_stopped = threading.Event() + launcher_events = [] + + class BlockingContext(_FakeContext): + def begin_frame(self): + frame_wait_entered.set() + runtime_stopped.wait(timeout=1.0) + return super().begin_frame() + + class FakeMirror: + status = DesktopMirrorStatus(closed=True) + + def __init__(self, on_user_close): + self.on_user_close = on_user_close + + def start(self): + pass + + def close(self): + pass + + def record_render_frame(self): + pass + + class FakeLauncher: + def interrupt_runtime(self): + launcher_events.append(("interrupt", threading.get_native_id())) + runtime_stopped.set() + + def stop(self): + launcher_events.append(("stop", threading.get_native_id())) + + mirror = None + + def create_mirror(_acquire_frame, _config, on_user_close, *, products=None): + nonlocal mirror + mirror = FakeMirror(on_user_close) + return mirror + + fake_ovxr = _fake_ovxr() + fake_ovxr.Context = BlockingContext + monkeypatch.setattr(frame_pump, "_import_ovxr", lambda: fake_ovxr) + monkeypatch.setattr(owned_mirror, "OwnedDesktopMirrorWorker", create_mirror) + + def launcher_factory(_path, *, accept_eula=False): + return FakeLauncher() + + config = _config() + mirror_config = DesktopMirrorConfig() + consumer = owned_mirror.OwnedDesktopMirror(mirror_config) + config.cloudxr_env_file = "/cloudxr.env" + config.auto_launch_cloudxr = True + owner = _session_with_consumers( + host, + config, + runtime_launcher_factory=launcher_factory, + frame_consumers=(consumer,), + ) + owner.start() + _FakeContext.request_exit_callback = runtime_stopped.set + + assert mirror is not None + assert owner.request_render("step-1") == 10 + assert frame_wait_entered.wait(timeout=1.0) + try: + if close_from_mirror: + mirror.on_user_close() + assert owner.shutdown_requested + assert owner._phase is not ovxr_session._Phase.CLOSED + owner.close() + else: + owner.close() + deadline = time.monotonic() + 0.25 + while ( + owner._phase is not ovxr_session._Phase.CLOSED + and time.monotonic() < deadline + ): + time.sleep(0.001) + assert owner._phase is ovxr_session._Phase.CLOSED + assert owner.shutdown_requested + assert [name for name, _thread in launcher_events].count("interrupt") == 0 + assert launcher_events == [("stop", owner.render_worker_native_thread_id)] + assert "interrupting the owned OpenXR runtime" not in caplog.text + finally: + runtime_stopped.set() + owner.close() + + def test_shutdown_timeout_preserves_runtime_until_frame_worker_retires(monkeypatch): host = _FakeHost() frame_wait_entered = threading.Event() @@ -1518,6 +1847,57 @@ def launcher_factory(_path, *, accept_eula=False): ] +def test_user_closing_mirror_on_first_submit_retires_created_ovxr( + monkeypatch, +): + host = _FakeHost() + + class FakeMirror: + status = DesktopMirrorStatus(closed=True) + + def __init__(self, on_user_close): + self.on_user_close = on_user_close + + def start(self): + self.on_user_close() + + def record_render_frame(self): + pass + + def close(self): + pass + + monkeypatch.setattr( + owned_mirror, + "OwnedDesktopMirrorWorker", + lambda _acquire, _config, on_user_close, *, products=None: FakeMirror( + on_user_close + ), + ) + config = _config() + mirror_config = DesktopMirrorConfig() + consumer = owned_mirror.OwnedDesktopMirror(mirror_config) + owner = _session_with_consumers(host, config, frame_consumers=(consumer,)) + + assert owner.start() is True + assert owner.request_render("step-1") > 0 + _wait_for_operation("submit") + deadline = time.monotonic() + 1.0 + while not owner.shutdown_requested and time.monotonic() < deadline: + time.sleep(0.001) + assert owner.shutdown_requested + owner.close() + operation_names = [name for name, _thread in _FakeContext.operations] + assert operation_names.count("create") == 1 + assert operation_names.count("destroy") == 1 + assert operation_names.count("submit") >= 1 + assert operation_names.index("destroy") > operation_names.index("submit") + assert host.events[-2:] == [ + ("host-close", host.attach_thread), + ("environment-close", host.attach_thread), + ] + + @pytest.mark.parametrize("failure_type", [RuntimeError, _WorkerAbort]) def test_render_failure_waits_for_borrowers_and_preserves_first_error( monkeypatch, failure_type @@ -1743,6 +2123,29 @@ def test_config_rejects_unknown_session_mode(): _config(session_mode="automatic") +def test_success_without_xr_submission_does_not_count_render_frame(monkeypatch): + class NoSubmission(_FakeContext): + def submit_frame(self, plan, **kwargs): + super().submit_frame(plan, **kwargs) + return 0, SimpleNamespace(submitted=False) + + api = _fake_ovxr() + api.Context = NoSubmission + monkeypatch.setattr(frame_pump, "_import_ovxr", lambda: api) + config = _config() + mirror_config = DesktopMirrorConfig() + consumer = owned_mirror.OwnedDesktopMirror(mirror_config) + owner = _session_with_consumers(_FakeHost(), config, frame_consumers=(consumer,)) + owner.start() + worker = consumer + with owner: + assert owner.request_render("no-submit") > 0 + _wait_for_operation("submit") + assert not worker._start_attempted + assert worker.status.render_thread_fps == 0 + assert worker.status.presented_frames == 0 + + def test_three_products_keep_selection_order_and_one_native_owner(monkeypatch): events = [] paths = ("/Render/Head", "/Render/Left", "/Render/Right") diff --git a/tests/python/core/ovxr_session_manager/test_owned_mirror.py b/tests/python/core/ovxr_session_manager/test_owned_mirror.py new file mode 100644 index 0000000000..0e444f47d3 --- /dev/null +++ b/tests/python/core/ovxr_session_manager/test_owned_mirror.py @@ -0,0 +1,568 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +import asyncio +import os +import threading +import time +from types import SimpleNamespace +from unittest.mock import Mock + +import numpy as np +import pytest + +from isaaccapture.ovxr_session_manager import camera_mailbox +from isaaccapture.ovxr_session_manager.camera_mailbox import FrameOutputs + +from isaaccapture.ovxr_session_manager.desktop_mirror import ( + DesktopMirrorConfig, + DesktopMirrorStatus, +) +import isaaccapture.ovxr_session_manager.owned_mirror as mirror + + +def _prepared_mirror(*, render_products=(), auxiliary_products=()): + owner = mirror.OwnedDesktopMirror( + DesktopMirrorConfig(), + camera_products=tuple( + mirror.MirrorProduct(path, label or path.rsplit("/", 1)[-1]) + for path, label in auxiliary_products + ), + ) + binding = SimpleNamespace() + config = SimpleNamespace(render_products=render_products) + owner.on_session_started(binding, config, lambda: None) + return owner + + +@pytest.mark.parametrize( + "eye_paths", + [(), ("/Render/CustomLeft", "/Render/CustomRight")], +) +def test_mirror_menu_includes_eyes_and_camera_labels(monkeypatch, eye_paths): + captured = [] + + def create_worker(_acquire, _config, _on_close, *, products): + captured.extend(products.view()[0]) + return SimpleNamespace(status=DesktopMirrorStatus(), close=lambda: None) + + monkeypatch.setattr(mirror, "OwnedDesktopMirrorWorker", create_worker) + owner = _prepared_mirror( + render_products=eye_paths, + auxiliary_products=(("/Camera/Head", "Head camera"), ("/Camera/Wrist", None)), + ) + try: + expected_eyes = eye_paths or ( + "/OVXR/Render/Products/Left", + "/OVXR/Render/Products/Right", + ) + assert [(product.path, product.label) for product in captured] == [ + (expected_eyes[0], "Left eye"), + (expected_eyes[1], "Right eye"), + ("/Camera/Head", "Head camera"), + ("/Camera/Wrist", "Wrist"), + ] + finally: + owner.close() + + +@pytest.mark.parametrize("failure_site", ["ui", "images"]) +def test_mirror_close_preserves_failed_retirement(monkeypatch, failure_site): + events = [] + + def close_at(site): + events.append(site) + if site == failure_site: + raise RuntimeError(f"{site} retirement failed") + + worker = SimpleNamespace( + status=DesktopMirrorStatus(closed=True), close=lambda: close_at("ui") + ) + monkeypatch.setattr(mirror, "OwnedDesktopMirrorWorker", lambda *a, **kw: worker) + owner = _prepared_mirror(render_products=(), auxiliary_products=()) + monkeypatch.setattr(owner._mailbox, "close", lambda: close_at("images")) + with pytest.raises(RuntimeError, match=f"{failure_site} retirement failed"): + owner.close() + assert events == (["ui"] if failure_site == "ui" else ["ui", "images"]) + assert owner.status == worker.status + + +def test_mirror_start_failure_is_optional_and_attempted_once(monkeypatch): + worker = Mock(status=DesktopMirrorStatus()) + worker.start.side_effect = RuntimeError("UI thread unavailable") + monkeypatch.setattr(mirror, "OwnedDesktopMirrorWorker", lambda *a, **kw: worker) + owner = _prepared_mirror(render_products=(), auxiliary_products=()) + try: + owner.on_frame_submitted() + owner.on_frame_submitted() + worker.start.assert_called_once_with() + worker.record_render_frame.assert_not_called() + finally: + owner.close() + + +def test_optional_consumer_error_is_visible_without_invalidating_owned_image(): + mailbox = mirror.OwnedMirrorMailbox(30) + image = mirror.OwnedMirrorFrame(1, 1, 1, b"\0" * 4, 1) + mailbox._latest = image + consumer = mailbox.acquire(0) + mailbox.report_error(RuntimeError("optional upload failed")) + assert mailbox.view()[-1] == "optional upload failed" + assert mailbox.snapshot() is image and consumer is image + mailbox.release(consumer) + mailbox.close() + + +@pytest.mark.parametrize( + "configured_backend", + [None, "other", "vulkan"], +) +def test_mirror_preserves_application_backend(monkeypatch, configured_backend): + # Register restoration even when the original environment has no selection. + monkeypatch.setenv("OMNIUI_BACKEND", configured_backend or "") + if configured_backend is None: + monkeypatch.delenv("OMNIUI_BACKEND") + owner = mirror.OwnedDesktopMirror(DesktopMirrorConfig()) + assert os.environ.get("OMNIUI_BACKEND") == configured_backend + assert owner._worker is None + owner.close() + + +def test_missing_optional_ui_records_failure_and_closes_worker( + monkeypatch, mock_omni_ui +): + monkeypatch.delenv("OMNIUI_BACKEND", raising=False) + mock_omni_ui(None) + callbacks = [] + worker = mirror.OwnedDesktopMirrorWorker( + lambda: None, DesktopMirrorConfig(), lambda: callbacks.append("user-close") + ) + worker._run() + assert "OMNIUI_BACKEND" not in os.environ + assert worker.status.closed and "omni.ui" in worker.status.error + assert callbacks == [] + + +@pytest.mark.parametrize( + "error_type", [KeyboardInterrupt, SystemExit, asyncio.CancelledError] +) +def test_interrupted_mirror_worker_retires_before_reporting_thread_failure( + monkeypatch, mock_omni_ui, error_type +): + failure = error_type("mirror startup interrupted") + events = [] + + def init(*_args, **_kwargs): + raise failure + + mock_omni_ui( + SimpleNamespace(init=init, shutdown=lambda: events.append("ui-retired")) + ) + + def report(args): + assert args.exc_value is failure + assert events == ["ui-retired"] + events.append("thread-failure") + + monkeypatch.setattr(threading, "excepthook", report) + worker = mirror.OwnedDesktopMirrorWorker( + lambda: None, DesktopMirrorConfig(), lambda: events.append("user-close") + ) + worker.start() + worker.close() + assert worker.status.closed + assert events == ["ui-retired", "thread-failure"] + + +@pytest.mark.parametrize( + "width,height,rgba,expected_error,error_type", + [ + (1, 1, b"\0" * 4, "owned upload failed", RuntimeError), + (1, 1, b"\0" * 3, "mirror snapshot is not tightly packed RGBA8", RuntimeError), + (1, 1, b"\0" * 8, "mirror snapshot is not tightly packed RGBA8", RuntimeError), + (0, 1, b"", "mirror snapshot is not tightly packed RGBA8", RuntimeError), + (1, -1, b"", "mirror snapshot is not tightly packed RGBA8", RuntimeError), + (1, 1, b"\0" * 4, "owned upload interrupted", KeyboardInterrupt), + (1, 1, b"\0" * 4, "owned upload interrupted", SystemExit), + ], +) +def test_mirror_async_upload_failure_is_not_reported_as_user_close( + monkeypatch, width, height, rgba, expected_error, error_type, mock_omni_ui +): + retired = [] + + class Context: + frame = None + + def __enter__(self): + return self + + def __exit__(self, *_args): + return False + + def destroy(self): + retired.append("window") + + Context.frame = Context() + + class RaisingProvider: + def set_data_array(self, pixels, sizes): + assert isinstance(pixels, np.ndarray) + assert pixels.dtype == np.uint8 and pixels.flags.c_contiguous + assert pixels.ndim == 1 and pixels.size == width * height * 4 + assert pixels.base is rgba and not pixels.flags.writeable + assert sizes == [width, height] + raise error_type(expected_error) + + ui = SimpleNamespace( + init=lambda *_args, **_kwargs: None, + shutdown=lambda: retired.append("ui"), + ByteImageProvider=RaisingProvider, + Window=lambda *_args, **_kwargs: Context(), + ZStack=Context, + VStack=Context, + Label=lambda text, **_kwargs: SimpleNamespace(text=text), + Rectangle=lambda **_kwargs: None, + ImageWithProvider=lambda *_args, **_kwargs: SimpleNamespace(visible=True), + WINDOW_FLAGS_NO_TITLE_BAR=1, + RasterPolicy=SimpleNamespace(NEVER=1), + IwpFillPolicy=SimpleNamespace(IWP_PRESERVE_ASPECT_FIT=1), + TextureFormat=SimpleNamespace(RGBA8_UNORM=1), + ) + mock_omni_ui(ui) + callbacks = [] + image = mirror.OwnedMirrorFrame(1, width, height, rgba, 1) + worker = mirror.OwnedDesktopMirrorWorker( + lambda: image, DesktopMirrorConfig(), lambda: callbacks.append("user-close") + ) + if issubclass(error_type, Exception): + worker._run() + else: + with pytest.raises(error_type, match=expected_error): + worker._run() + assert worker.status.closed + assert worker.status.error == expected_error + assert worker.status.presented_frames == 0 + assert callbacks == [] + assert retired == ["window", "ui"] + + +@pytest.mark.parametrize( + "invalidate_during_upload,report_after_present,replacement_gap", + [ + (False, False, None), + (True, False, None), + (False, True, None), + (False, False, "upload"), + (False, False, "after"), + ], +) +def test_inprocess_slow_upload_keeps_latest_snapshot_and_closes( + monkeypatch, + invalidate_during_upload, + report_after_present, + replacement_gap, + mock_omni_ui, +): + entered, release, consumed_latest = (threading.Event() for _ in range(3)) + uploaded = [] + image_widget = SimpleNamespace(visible=False) + labels = [] + + class Label: + def __init__(self, text, **_kwargs): + self.owner = threading.get_native_id() + self.text = text + labels.append(self) + + @property + def text(self): + return self._text + + @text.setter + def text(self, value): + assert threading.get_native_id() == self.owner + self._text = value + + class Context: + def __init__(self, **_kwargs): + pass + + def set_build_fn(self, fn): + pass + + def rebuild(self): + pass + + def __enter__(self): + return self + + def __exit__(self, *_args): + return False + + def destroy(self): + pass + + Context.frame = Context() + + class Provider: + def set_data_array(self, pixels, sizes): + assert isinstance(pixels, np.ndarray) + assert pixels.dtype == np.uint8 and pixels.flags.c_contiguous + assert pixels.ndim == 1 and pixels.size == 4 and sizes == [1, 1] + assert isinstance(pixels.base, bytes) and not pixels.flags.writeable + # Keep the supplied buffer alive while the producer replaces its + # latest snapshot. No mailbox lock may cover this slow consumer. + if not uploaded: + entered.set() + assert release.wait(2) + uploaded.append(pixels.tobytes()[0]) + if uploaded[-1] == 3 and replacement_gap == "upload": + publish(4) + mailbox._drain_gpu(replace_latest=True) + assert mailbox.snapshot() is None + if uploaded[-1] == 3 or invalidate_during_upload: + consumed_latest.set() + + ui = SimpleNamespace( + init=lambda *_args, **_kwargs: None, + shutdown=lambda: None, + ByteImageProvider=Provider, + Window=lambda *_args, **_kwargs: Context(), + Frame=Context, + ZStack=Context, + VStack=Context, + Label=Label, + Rectangle=lambda **_kwargs: None, + ImageWithProvider=lambda *_args, **_kwargs: image_widget, + WINDOW_FLAGS_NO_TITLE_BAR=1, + RasterPolicy=SimpleNamespace(NEVER=1), + IwpFillPolicy=SimpleNamespace(IWP_PRESERVE_ASPECT_FIT=1), + TextureFormat=SimpleNamespace(RGBA8_UNORM=1), + ) + mock_omni_ui(ui) + from test_camera_mailbox import _outputs, _frame + + monkeypatch.setattr( + camera_mailbox, "_copy_host_rgba", lambda _r, h, _m: (1, 1, bytes([h]) * 4) + ) + mailbox = mirror.OwnedMirrorMailbox(30) + mailbox.set_products((mirror.MirrorProduct("/Camera", "Camera"),)) + + def publish(sequence): + mailbox._next_ns = 0 # exercise ownership, not the independent rate cap + mailbox.on_outputs(FrameOutputs(1, _outputs(_frame(sequence, sequence)))) + + publish(1) + callbacks = [] + worker = mirror.OwnedDesktopMirrorWorker( + mailbox.snapshot, + DesktopMirrorConfig(), + lambda: callbacks.append("user-close"), + products=mailbox + if invalidate_during_upload or report_after_present or replacement_gap + else None, + ) + worker.set_status_text("before") + worker.start() + try: + assert entered.wait(2) + worker.set_status_text("after: Physics 0/120 Hz (0.00× RT)") + if invalidate_during_upload: + mailbox._next_ns = 0 + monkeypatch.setattr( + camera_mailbox, "_copy_host_rgba", lambda *_: (1, 1, b"bad") + ) + mailbox.on_outputs(FrameOutputs(1, _outputs(_frame(2, 2)))) + assert mailbox.snapshot() is None + else: + publish(2) + publish(3) + assert mailbox.snapshot().sequence == 3 + release.set() + assert consumed_latest.wait(2) + deadline = time.monotonic() + 2 + while labels[0].text != "after: Physics 0/120 Hz (0.00× RT)": + assert time.monotonic() < deadline + time.sleep(0.005) + if replacement_gap: + deadline = time.monotonic() + 2 + while worker.status.last_sequence != 3: + assert time.monotonic() < deadline + time.sleep(0.005) + if replacement_gap == "after": + mailbox._drain_gpu(replace_latest=True) + assert mailbox.snapshot() is None + time.sleep(0.1) # Observe several UI polls during the producer gap. + assert image_widget.visible + if report_after_present: + assert image_widget.visible and worker.status.last_sequence == 3 + mailbox.report_error(RuntimeError("optional readback failed")) + deadline = time.monotonic() + 2 + while labels[-1].text != "optional readback failed": + assert time.monotonic() < deadline + time.sleep(0.005) + assert not image_widget.visible + assert mailbox.snapshot().sequence == 3 + + finally: + release.set() + worker.close(2) + assert uploaded == ([1] if invalidate_during_upload else [1, 3]) + assert worker.status.closed and worker.status.error is None + assert worker.status.last_sequence == (0 if invalidate_during_upload else 3) + if invalidate_during_upload: + assert not image_widget.visible + assert worker.status.presented_frames == 0 + assert callbacks == [] + + +@pytest.mark.parametrize( + "run_error,callback_raises,cleanup_error_type", + [ + (False, False, None), + (False, True, None), + (True, False, None), + (True, False, KeyboardInterrupt), + (False, False, SystemExit), + ], +) +def test_ui_exit_finishes_async_tasks_before_cleanup( + monkeypatch, run_error, callback_raises, cleanup_error_type, mock_omni_ui +): + retired = [] + + class Context: + def __enter__(self): + return self + + def __exit__(self, *_args): + return False + + def destroy(self): + assert loops + assert not any(asyncio.all_tasks(loop) for loop in loops) + retired.append("window") + if cleanup_error_type is not None: + raise cleanup_error_type("cleanup interrupted") + + Context.frame = Context() + loops = [] + + async def run_async(): + loops.append(asyncio.get_running_loop()) + await asyncio.sleep(0) + if run_error: + raise RuntimeError("native mirror failed") + + ui = SimpleNamespace( + init=lambda *_args, **_kwargs: None, + shutdown=lambda: retired.append("ui"), + run_async=run_async, + ByteImageProvider=lambda: None, + Window=lambda *_args, **_kwargs: Context(), + ZStack=Context, + VStack=Context, + Label=lambda text, **_kwargs: SimpleNamespace(text=text), + Rectangle=lambda **_kwargs: None, + ImageWithProvider=lambda *_args, **_kwargs: SimpleNamespace(visible=True), + WINDOW_FLAGS_NO_TITLE_BAR=1, + RasterPolicy=SimpleNamespace(NEVER=1), + IwpFillPolicy=SimpleNamespace(IWP_PRESERVE_ASPECT_FIT=1), + TextureFormat=SimpleNamespace(RGBA8_UNORM=1), + ) + mock_omni_ui(ui) + callbacks = [] + + def on_close(): + callbacks.append("user-close") + if callback_raises: + raise RuntimeError("application close failed") + + worker = mirror.OwnedDesktopMirrorWorker( + lambda: None, DesktopMirrorConfig(), on_close + ) + if cleanup_error_type is None: + worker._run() + else: + with pytest.raises(cleanup_error_type, match="cleanup interrupted"): + worker._run() + assert worker.status.closed + assert callbacks == ([] if run_error or cleanup_error_type else ["user-close"]) + assert worker.status.error == ( + "cleanup interrupted" + if cleanup_error_type + else "native mirror failed" + if run_error + else "application close failed" + if callback_raises + else None + ) + assert retired == ["window", "ui"] + + +def test_rates_measure_host_and_submitted_render_events_not_gui_uploads(monkeypatch): + now = [0] + monkeypatch.setattr(mirror.time, "perf_counter_ns", lambda: now[0]) + worker = mirror.OwnedDesktopMirrorWorker(lambda: None, DesktopMirrorConfig()) + for n in range(21): + now[0] = n * 50_000_000 + worker.record_host_frame() + if n % 2 == 0: + worker.record_render_frame() + worker.set_status_text("Physics 120 Hz / 120 Hz 1.00x") + assert worker.status.main_thread_fps == pytest.approx(20) + assert worker.status.render_thread_fps == pytest.approx(10) + assert worker.status.presented_frames == 0 + assert worker.status.status_text == "Physics 120 Hz / 120 Hz 1.00x" + now[0] += 1_000_000_001 + assert worker.status.main_thread_fps == worker.status.render_thread_fps == 0 + assert worker.status.status_text == "Physics 120 Hz / 120 Hz 1.00x" + worker.close() + + +@pytest.mark.parametrize("level, expected_records", [("WARNING", 0), ("INFO", 1)]) +def test_mirror_retirement_respects_named_logger_level( + monkeypatch, caplog, capsys, level, expected_records +): + worker = SimpleNamespace( + status=DesktopMirrorStatus(closed=True), close=lambda: None + ) + monkeypatch.setattr( + mirror, "OwnedDesktopMirrorWorker", lambda *args, **kwargs: worker + ) + with caplog.at_level(level, logger=mirror.__name__): + owner = _prepared_mirror(render_products=(), auxiliary_products=()) + owner.close() + records = [ + record + for record in caplog.records + if record.name == mirror.__name__ + and record.getMessage().startswith("OVUI desktop mirror stopped:") + ] + assert len(records) == expected_records + if level == "WARNING": + assert "OVUI desktop mirror stopped:" not in capsys.readouterr().out + + +def test_mirror_rejects_active_non_vulkan_backend_without_changing_environment( + monkeypatch, mock_omni_ui +): + events = [] + monkeypatch.delenv("OMNIUI_BACKEND", raising=False) + mock_omni_ui( + SimpleNamespace( + init=lambda *args, **kwargs: None, + shutdown=lambda: events.append("retired"), + standalone=SimpleNamespace( + get_frame_pacing_status=lambda: {"backend": "other"} + ), + ) + ) + worker = mirror.OwnedDesktopMirrorWorker( + lambda: None, DesktopMirrorConfig(), lambda: events.append("user-close") + ) + worker._run() + assert "requires OVUI's Vulkan backend" in worker.status.error + assert events == ["retired"] + assert "OMNIUI_BACKEND" not in os.environ diff --git a/tests/python/core/ovxr_session_manager/test_passive_mirror.py b/tests/python/core/ovxr_session_manager/test_passive_mirror.py new file mode 100644 index 0000000000..011ee2885b --- /dev/null +++ b/tests/python/core/ovxr_session_manager/test_passive_mirror.py @@ -0,0 +1,172 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +import threading +from types import SimpleNamespace + +import pytest + +from isaaccapture.ovxr_session_manager import camera_mailbox +from isaaccapture.ovxr_session_manager.camera_mailbox import FrameOutputs + +from isaaccapture.ovxr_session_manager.owned_mirror import ( + MirrorProduct, + OwnedMirrorMailbox, +) +from test_camera_mailbox import _frame, _string + +LEFT = MirrorProduct("/Left", "Left eye") +RIGHT = MirrorProduct("/Right", "Right eye") +CAMERA = MirrorProduct("/Camera", "Camera") + + +def outputs(**handles): + products = [ + SimpleNamespace( + render_product_path=_string("/" + path), + output_frame_count=int(handle is not None), + output_frames=[] if handle is None else [_frame(float(handle), handle)], + ) + for path, handle in handles.items() + ] + return SimpleNamespace( + contents=SimpleNamespace(output_count=len(products), outputs=products) + ) + + +def mailbox(monkeypatch): + calls = [] + + def copy(_renderer, handle, _limit): + calls.append(handle) + return 1, 1, bytes([handle]) * 4 + + monkeypatch.setattr(camera_mailbox, "_copy_host_rgba", copy) + m = OwnedMirrorMailbox(30) + m.set_products((LEFT, RIGHT, CAMERA)) + return m, calls + + +def test_selection_only_copies_selected_output_without_changing_production_set( + monkeypatch, +): + m, calls = mailbox(monkeypatch) + source = outputs(Left=1, Right=2, Camera=3) + m.on_outputs(FrameOutputs(1, source)) + assert calls == [1] and m.snapshot().product == LEFT + m.select(RIGHT) + assert m.snapshot() is None + m.on_outputs(FrameOutputs(1, source)) + assert calls == [1, 2] and m.snapshot().rgba == b"\2" * 4 + assert m.view()[0] == (LEFT, RIGHT, CAMERA) + assert [p.render_product_path.length for p in source.contents.outputs] == [5, 6, 7] + + +@pytest.mark.parametrize("mirror_first", [False, True]) +def test_camera_cpu_copy_is_shared_in_either_consumer_order(monkeypatch, mirror_first): + from isaaccapture.ovxr_session_manager.camera_mailbox import ( + CompletedCameraMailbox, + ) + + m, calls = mailbox(monkeypatch) + m.select(CAMERA) + copies = [] + monkeypatch.setattr( + camera_mailbox, + "_copy_host_rgba", + lambda *_args: copies.append(1) or (1, 1, b"RGBA"), + ) + source = outputs(Left=1, Right=2, Camera=3) + frame = FrameOutputs(1, source) + camera = CompletedCameraMailbox(CAMERA.path, 4) + operations = [ + lambda: camera.on_outputs(frame), + lambda: m.on_outputs(frame), + ] + for operation in operations[:: (-1 if mirror_first else 1)]: + operation() + held = m.snapshot() + assert held.rgba is camera.snapshot().rgba and copies == [1] and calls == [] + bounded_camera = CompletedCameraMailbox(CAMERA.path, 3) + bounded_camera.on_outputs(frame) + assert bounded_camera.snapshot() is None + assert "bounded mailbox size" in bounded_camera.error() + assert m.snapshot() is held and copies == [1] + m._next_ns = 0 + m.on_outputs(FrameOutputs(1, outputs(Left=5, Right=6, Camera=None))) + assert m.snapshot() is held and m.view()[0] == (LEFT, RIGHT, CAMERA) + + +@pytest.mark.parametrize("change", ["selection", "remove", "generation", "reset"]) +def test_inflight_copy_is_discarded_after_identity_change(change, monkeypatch): + entered, release = threading.Event(), threading.Event() + + def copy(*_): + entered.set() + assert release.wait(2) + return 1, 1, b"old!" + + monkeypatch.setattr(camera_mailbox, "_copy_host_rgba", copy) + m = OwnedMirrorMailbox(30) + m.set_products((LEFT, RIGHT)) + worker = threading.Thread( + target=m.on_outputs, args=(FrameOutputs(1, outputs(Left=1, Right=2)),) + ) + worker.start() + try: + assert entered.wait(2) + if change == "selection": + m.select(RIGHT) + elif change == "remove": + m.set_products((RIGHT,)) + elif change == "generation": + m.set_products((MirrorProduct(LEFT.path, LEFT.label, 1), RIGHT)) + else: + m.reset() + finally: + release.set() + worker.join(2) + assert not worker.is_alive() and m.snapshot() is None + + +def test_add_remove_is_not_arrival_driven_and_stale_dropdown_event_is_ignored( + monkeypatch, +): + m, _calls = mailbox(monkeypatch) + m.on_outputs(FrameOutputs(1, outputs(Left=1))) + held = m.snapshot() + m.set_products((LEFT, RIGHT)) + assert m.snapshot() is held + m.select(CAMERA) + assert m.view()[1] == LEFT + m.set_products((RIGHT,)) + assert m.view()[1] == RIGHT and m.snapshot() is None + m.set_products(()) + assert m.view()[1] is None + + +def test_invalid_selected_image_clears_stale_pixels_and_error_is_contained(monkeypatch): + m, _calls = mailbox(monkeypatch) + m.on_outputs(FrameOutputs(1, outputs(Left=1))) + m._next_ns = 0 + monkeypatch.setattr(camera_mailbox, "_copy_host_rgba", lambda *_: (1, 1, b"bad")) + m.on_outputs(FrameOutputs(1, outputs(Left=2))) + assert m.snapshot() is None and "RGBA8" in m.view()[4] + + +def test_mirror_cadence_does_not_copy_another_output_or_reset_on_gap(monkeypatch): + m, calls = mailbox(monkeypatch) + m.on_outputs(FrameOutputs(1, outputs(Left=1, Right=2))) + m.on_outputs(FrameOutputs(1, outputs(Left=3, Right=4))) + assert calls == [1] + m._next_ns = 0 + m.on_outputs(FrameOutputs(1, outputs(Left=None, Right=4))) + assert calls == [1] and m.snapshot().rgba == b"\1" * 4 + + +def test_product_paths_must_be_unique_and_absolute(monkeypatch): + m, _ = mailbox(monkeypatch) + with pytest.raises(ValueError): + m.set_products((LEFT, LEFT)) + with pytest.raises(ValueError): + m.set_products((MirrorProduct("relative", "bad"),))