diff --git a/.github/workflows/python-regressions.yml b/.github/workflows/python-regressions.yml new file mode 100644 index 00000000..d2582275 --- /dev/null +++ b/.github/workflows/python-regressions.yml @@ -0,0 +1,37 @@ +name: Python regressions + +on: [push, pull_request] + +permissions: + contents: read + +jobs: + tests: + runs-on: ${{ matrix.os }} + strategy: + fail-fast: false + matrix: + os: [ubuntu-latest, windows-latest] + python: ['3.10', '3.12'] + steps: + - uses: actions/checkout@v4 + - uses: actions/setup-python@v5 + with: + python-version: ${{ matrix.python }} + cache: pip + cache-dependency-path: tests/requirements.txt + - name: Install regression test dependencies + run: python -m pip install -r tests/requirements.txt + - name: Run regression tests + env: + PYTEST_DISABLE_PLUGIN_AUTOLOAD: '1' + run: python -m pytest tests -q --junitxml=regressions.xml --basetemp=test-output + - name: Upload test evidence + if: always() + uses: actions/upload-artifact@v4 + with: + name: regressions-${{ matrix.os }}-python-${{ matrix.python }} + path: | + regressions.xml + test-output/**/*.mp4 + test-output/**/*.png diff --git a/docs/source/components/ue_detail.rst b/docs/source/components/ue_detail.rst index efb0fe56..ac50538d 100644 --- a/docs/source/components/ue_detail.rst +++ b/docs/source/components/ue_detail.rst @@ -76,6 +76,79 @@ How to get images **Related files:** ``communicator.py``, ``unrealcv.py``. +Panoramic video export +~~~~~~~~~~~~~~~~~~~~~~ + +``simworld.utils.panorama`` exports six synchronized square camera views as +a 2:1 equirectangular projection (ERP) MP4. It covers 360 degrees horizontally +and 180 degrees vertically. A single perspective recording, even at a 2:1 +resolution, does not contain the views needed to produce a full panorama. + +Each input frame is a dictionary of six BGR ``uint8`` images with equal +dimensions. All six views must use the same camera location, the same +simulation instant, a square resolution and a 90-degree horizontal FOV. +The face convention uses Unreal's +X forward, +Y right and +Z up axes: + +.. list-table:: Camera rotations (pitch, yaw, roll), in degrees + :header-rows: 1 + + * - Face + - Rotation + * - ``front`` + - ``(0, 0, 0)`` + * - ``right`` + - ``(0, 90, 0)`` + * - ``back`` + - ``(0, 180, 0)`` + * - ``left`` + - ``(0, -90, 0)`` + * - ``up`` + - ``(90, 0, 0)`` + * - ``down`` + - ``(-90, 0, 0)`` + +These rotations are also available as ``CUBEMAP_ROTATIONS``. Capture all six +faces before advancing the simulation. With sequential camera reads, use +synchronous mode and verify that each captured image reflects the requested +camera pose in your UE build. Asynchronous captures can introduce moving-object +seams; projection cannot recover views missing from the input recordings. + +For example, export existing recordings stored as +``recording/front/000000.png``, ``recording/right/000000.png``, etc.: + +.. code-block:: python + + from pathlib import Path + + import cv2 + + from simworld.utils.panorama import CUBEMAP_ROTATIONS, save_panorama_video + + recording = Path('recording') + + def cubemap_frames(): + for front_path in sorted((recording / 'front').glob('*.png')): + yield { + face: cv2.imread(str(recording / face / front_path.name)) + for face in CUBEMAP_ROTATIONS + } + + save_panorama_video( + cubemap_frames(), 'panorama.mp4', resolution=(1440, 720), fps=25, + ) + +Frames are projected and written incrementally, so the whole recording does +not need to fit in memory. The output height must be even to avoid video-codec +cropping. ``fps`` controls playback speed, independently of capture throughput. +For a single image, use ``CubemapProjector(face_size, resolution).project(faces)``. +See :mod:`simworld.utils.panorama` for the API. + +The MP4 contains ERP pixels using OpenCV's ``mp4v`` codec. It does not insert +spherical-video metadata; select equirectangular/360 mode in your player or +add the metadata required by your publishing platform. Export tests use +synthetic cubemaps and a real MP4 encode/decode round trip; UE capture and +player-specific metadata are separate integration steps. + Synchronous and Asynchronous mode --------------------------------- diff --git a/docs/source/resources/simworld.utils.rst b/docs/source/resources/simworld.utils.rst index 5c6fca8c..72b1c2f8 100644 --- a/docs/source/resources/simworld.utils.rst +++ b/docs/source/resources/simworld.utils.rst @@ -108,6 +108,14 @@ simworld.utils.video\_recorder module :undoc-members: :show-inheritance: +simworld.utils.panorama module +------------------------------ + +.. automodule:: simworld.utils.panorama + :members: + :undoc-members: + :show-inheritance: + Module contents --------------- diff --git a/simworld/utils/panorama.py b/simworld/utils/panorama.py new file mode 100644 index 00000000..2ccca951 --- /dev/null +++ b/simworld/utils/panorama.py @@ -0,0 +1,149 @@ +"""Project six perspective camera views into 2:1 equirectangular videos.""" + +import math +import os +from numbers import Integral + +import cv2 +import numpy as np + +# Unreal coordinates: +X forward, +Y right, +Z up. Each face is square, +# has a 90-degree horizontal field of view, and uses (pitch, yaw, roll). +CUBEMAP_ROTATIONS = { + 'front': (0, 0, 0), + 'right': (0, 90, 0), + 'back': (0, 180, 0), + 'left': (0, -90, 0), + 'up': (90, 0, 0), + 'down': (-90, 0, 0), +} + +# Forward, image-right and image-up vectors for the rotations above. +_FACE_AXES = { + 'front': ((1, 0, 0), (0, 1, 0), (0, 0, 1)), + 'right': ((0, 1, 0), (-1, 0, 0), (0, 0, 1)), + 'back': ((-1, 0, 0), (0, -1, 0), (0, 0, 1)), + 'left': ((0, -1, 0), (1, 0, 0), (0, 0, 1)), + 'up': ((0, 0, 1), (0, 1, 0), (-1, 0, 0)), + 'down': ((0, 0, -1), (0, 1, 0), (1, 0, 0)), +} + + +class CubemapProjector: + """Reuse projection maps to convert BGR cubemaps to ERP frames. + + Longitude zero (+X/front) is at the image center; +Y/right is at three + quarters of its width. The back face wraps across both image edges. + +Z/up is at the top. All six inputs must share one optical center and + simulation instant. This class performs projection, not camera capture. + """ + + def __init__(self, face_size, resolution=(1440, 720)): + """Precompute maps for square faces and a (width, height) ERP output.""" + if isinstance(face_size, bool) or not isinstance(face_size, Integral) or face_size < 1: + raise ValueError('face_size must be a positive integer') + if len(resolution) != 2 or any( + isinstance(size, bool) or not isinstance(size, Integral) or size < 1 + for size in resolution + ): + raise ValueError('resolution must contain two positive integers') + width, height = resolution + if width != 2 * height: + raise ValueError('ERP resolution must have a 2:1 width-to-height ratio') + if max(face_size, width, height) >= 32767: + raise ValueError('Image dimensions must be below 32767 for OpenCV remap') + self.face_size = int(face_size) + self.resolution = (int(width), int(height)) + + longitude = ((np.arange(width, dtype=np.float32) + 0.5) / width - 0.5) * (2 * np.pi) + latitude = (0.5 - (np.arange(height, dtype=np.float32) + 0.5) / height) * np.pi + lon, lat = np.meshgrid(longitude, latitude) + rays = np.stack((np.cos(lat) * np.cos(lon), np.cos(lat) * np.sin(lon), np.sin(lat)), axis=-1) + forwards = np.array([axes[0] for axes in _FACE_AXES.values()], dtype=np.float32) + face_indices = np.argmax(rays @ forwards.T, axis=-1) + self._maps = [] + for index, (name, (forward, right, up)) in enumerate(_FACE_AXES.items()): + mask = face_indices == index + # Only selected rays face this camera; avoid division by zero + # for the unused pixels in the full-sized OpenCV remap arrays. + distance = np.where(mask, rays @ np.array(forward, dtype=np.float32), 1.0) + u = (rays @ np.array(right, dtype=np.float32)) / distance + v = -(rays @ np.array(up, dtype=np.float32)) / distance + map_x = ((u + 1) * face_size / 2 - 0.5).astype(np.float32) + map_y = ((v + 1) * face_size / 2 - 0.5).astype(np.float32) + self._maps.append((name, mask, map_x, map_y)) + + def project(self, faces): + """Return an ERP BGR uint8 frame from six named BGR uint8 arrays. + + Args: + faces: Mapping with front, right, back, left, up and down keys. + Each value has shape (face_size, face_size, 3), with BGR + channels as returned by SimWorld's camera observations. + + Raises: + ValueError: A face is missing or has an incompatible shape/dtype. + """ + if set(faces) != set(_FACE_AXES): + raise ValueError('faces must contain exactly front, right, back, left, up and down') + expected_shape = (self.face_size, self.face_size, 3) + for name, face in faces.items(): + if not isinstance(face, np.ndarray) or face.shape != expected_shape or face.dtype != np.uint8: + raise ValueError(f'{name} must be a uint8 BGR array with shape {expected_shape}') + width, height = self.resolution + frame = np.empty((height, width, 3), dtype=np.uint8) + for name, mask, map_x, map_y in self._maps: + sampled = cv2.remap( + faces[name], map_x, map_y, cv2.INTER_LINEAR, + borderMode=cv2.BORDER_REPLICATE, + ) + frame[mask] = sampled[mask] + return frame + + +def save_panorama_video(cubemap_frames, video_path, resolution=(1440, 720), fps=25.0): + """Stream cubemap recordings into an MP4 with 2:1 ERP pixel projection. + + Args: + cubemap_frames: Iterable of six-face mappings accepted by + CubemapProjector.project. A generator can capture frames on demand. + + video_path: Output MP4 path; its parent directory must exist. + + resolution: Output (width, height), with width == 2 * height and an + even height so the video codec does not crop odd dimensions. + + fps: Positive, finite playback frame rate, independent of capture speed. + + Returns: + Output path as a string. No spherical-player metadata is inserted. + + Raises: + ValueError: Frames are empty or invalid, or encoding parameters are invalid. + RuntimeError: OpenCV cannot open the output video writer. + """ + if not math.isfinite(fps) or fps <= 0: + raise ValueError('fps must be positive and finite') + frames = iter(cubemap_frames) + try: + first = next(frames) + except StopIteration as error: + raise ValueError('At least one cubemap frame is required') from error + front = first.get('front') + if not isinstance(front, np.ndarray) or front.ndim != 3: + raise ValueError('front must be a square uint8 BGR array') + projector = CubemapProjector(front.shape[0], resolution) + if projector.resolution[1] % 2: + raise ValueError('Video height must be even to avoid codec cropping') + image = projector.project(first) + video_path = os.fspath(video_path) + writer = cv2.VideoWriter(video_path, cv2.VideoWriter_fourcc(*'mp4v'), fps, projector.resolution) + try: + if not writer.isOpened(): + raise RuntimeError(f'Could not open video writer: {video_path}') + writer.write(image) + for faces in frames: + writer.write(projector.project(faces)) + finally: + writer.release() + return video_path diff --git a/tests/requirements.txt b/tests/requirements.txt new file mode 100644 index 00000000..90499606 --- /dev/null +++ b/tests/requirements.txt @@ -0,0 +1,13 @@ +# Dependencies needed by the regression tests and simworld's package imports. +# Asset retrieval models are not exercised by this suite. +pytest +numpy +pandas +pyqtgraph +PyQt5 +unrealcv +opencv-python +Pillow +openai +PyYAML +ipython diff --git a/tests/test_panorama.py b/tests/test_panorama.py new file mode 100644 index 00000000..124949a2 --- /dev/null +++ b/tests/test_panorama.py @@ -0,0 +1,216 @@ +"""Check panorama geometry and actual MP4 encoding without an Unreal server.""" + +from unittest.mock import Mock + +import cv2 +import numpy as np +import pytest + +from simworld.utils.panorama import CubemapProjector, save_panorama_video + + +def solid_faces(size=32, color=None): + """Create distinct face colors or a single constant-color sphere.""" + colors = { + 'front': (0, 0, 255), 'right': (0, 255, 0), 'back': (255, 0, 0), + 'left': (0, 255, 255), 'up': (255, 0, 255), 'down': (255, 255, 0), + } + return {name: np.full((size, size, 3), rgb if color is None else color, dtype=np.uint8) + for name, rgb in colors.items()} + + +def direction_faces(size=64): + """Render an analytic direction-colored sphere with yaw/pitch matrices.""" + rotations = {'front': (0, 0), 'right': (0, 90), 'back': (0, 180), + 'left': (0, -90), 'up': (90, 0), 'down': (-90, 0)} + coordinates = (np.arange(size) + 0.5) / size * 2 - 1 + right, down = np.meshgrid(coordinates, coordinates) + local = np.stack((np.ones_like(right), right, -down), axis=-1) + local /= np.linalg.norm(local, axis=-1, keepdims=True) + faces = {} + for name, (pitch, yaw) in rotations.items(): + pitch, yaw = np.radians([pitch, yaw]) + cp, sp, cy, sy = np.cos(pitch), np.sin(pitch), np.cos(yaw), np.sin(yaw) + rotation = np.array([[cp * cy, -sy, -sp * cy], + [cp * sy, cy, -sp * sy], [sp, 0, cp]]) + world = local @ rotation.T + faces[name] = np.rint((world + 1) * 127.5).astype(np.uint8) + return faces + + +def test_cardinal_directions_and_poles(): + """The panorama contains all six faces in the documented orientation.""" + faces = solid_faces() + image = CubemapProjector(32, (256, 128)).project(faces) + assert image.shape == (128, 256, 3) + assert image.dtype == np.uint8 + for name, (row, col) in { + 'front': (64, 128), 'right': (64, 192), 'back': (64, 0), + 'left': (64, 64), 'up': (0, 128), 'down': (127, 128), + }.items(): + np.testing.assert_array_equal(image[row, col], faces[name][0, 0]) + np.testing.assert_array_equal(image[64, -1], faces['back'][0, 0]) + + +def test_projection_matches_independent_analytic_sphere(): + """Per-pixel world directions catch mirrored faces and incorrect pole rolls.""" + image = CubemapProjector(64, (256, 128)).project(direction_faces()) + longitude = (np.arange(256) + 0.5) * (2 * np.pi / 256) - np.pi + latitude = np.pi / 2 - (np.arange(128) + 0.5) * (np.pi / 128) + expected = np.empty((128, 256, 3)) + expected[:, :, 0] = np.outer(np.cos(latitude), np.cos(longitude)) + expected[:, :, 1] = np.outer(np.cos(latitude), np.sin(longitude)) + expected[:, :, 2] = np.sin(latitude)[:, None] + error = np.abs(image.astype(float) - (expected + 1) * 127.5) + assert error.max() < 3 + assert error.mean() < 0.5 + + +@pytest.mark.parametrize('resolution', [(128, 128), (0, 0), (128.0, 64), (128, -64), (128,), (65536, 32768)]) +def test_rejects_invalid_resolution(resolution): + """Reject dimensions that cannot represent a complete ERP image.""" + with pytest.raises(ValueError): + CubemapProjector(32, resolution) + + +@pytest.mark.parametrize('size', [0, -1, 2.5, True, 32767]) +def test_rejects_invalid_face_size(size): + """Require positive square dimensions supported by OpenCV.""" + with pytest.raises(ValueError): + CubemapProjector(size, (128, 64)) + + +@pytest.mark.parametrize('problem', ['missing', 'extra', 'size', 'dtype', 'channels', 'none']) +def test_rejects_incompatible_faces(problem): + """Malformed frames must not silently produce incomplete panoramas.""" + faces = solid_faces() + if problem == 'missing': + del faces['up'] + elif problem == 'extra': + faces['extra'] = faces['front'] + elif problem == 'size': + faces['up'] = np.zeros((31, 32, 3), dtype=np.uint8) + elif problem == 'dtype': + faces['up'] = faces['up'].astype(np.float32) + elif problem == 'channels': + faces['up'] = faces['up'][:, :, 0] + else: + faces['up'] = None + with pytest.raises(ValueError): + CubemapProjector(32, (128, 64)).project(faces) + + +def test_mp4_round_trip_preserves_dimensions_fps_frames_and_colors(tmp_path): + """Encode a real MP4 and inspect every decoded frame with OpenCV.""" + colors = [(0, 0, 255), (0, 255, 0), (255, 0, 0)] + path = tmp_path / 'panorama.mp4' + result = save_panorama_video((solid_faces(color=color) for color in colors), path, (128, 64), fps=12) + assert result == str(path) + capture = cv2.VideoCapture(str(path)) + try: + assert capture.isOpened() + assert capture.get(cv2.CAP_PROP_FRAME_WIDTH) == 128 + assert capture.get(cv2.CAP_PROP_FRAME_HEIGHT) == 64 + assert capture.get(cv2.CAP_PROP_FPS) == pytest.approx(12) + assert capture.get(cv2.CAP_PROP_FRAME_COUNT) == 3 + for color in colors: + ok, frame = capture.read() + assert ok + assert frame.shape == (64, 128, 3) + np.testing.assert_allclose(frame.mean(axis=(0, 1)), color, atol=6) + assert not capture.read()[0] + finally: + capture.release() + + +def test_direction_sphere_video_round_trip(tmp_path): + """Save a visible ERP fixture and verify its geometry survives encoding.""" + faces = direction_faces(128) + expected = CubemapProjector(128, (512, 256)).project(faces) + assert cv2.imwrite(str(tmp_path / 'synthetic-erp-reference.png'), expected) + path = tmp_path / 'synthetic-erp.mp4' + save_panorama_video((faces for _ in range(24)), path, (512, 256), fps=12) + capture = cv2.VideoCapture(str(path)) + try: + assert capture.isOpened() + for _ in range(24): + ok, frame = capture.read() + assert ok + assert frame.shape == expected.shape + assert np.abs(frame.astype(float) - expected).mean() < 3 + assert not capture.read()[0] + finally: + capture.release() + + +def test_export_consumes_frames_incrementally_and_releases_writer(tmp_path, monkeypatch): + """Write each frame before asking a potentially unbounded source for more.""" + writer = Mock() + monkeypatch.setattr(cv2, 'VideoWriter', Mock(return_value=writer)) + + def frames(): + for index in range(3): + assert writer.write.call_count == index + yield solid_faces() + + save_panorama_video(frames(), tmp_path / 'panorama.mp4', (128, 64)) + assert writer.write.call_count == 3 + writer.release.assert_called_once() + + +def test_invalid_later_frame_releases_writer(tmp_path, monkeypatch): + """Flush already written frames when a capture or projection fails.""" + writer = Mock() + monkeypatch.setattr(cv2, 'VideoWriter', Mock(return_value=writer)) + with pytest.raises(ValueError): + save_panorama_video([solid_faces(), {}], tmp_path / 'panorama.mp4', (128, 64)) + writer.write.assert_called_once() + writer.release.assert_called_once() + + +def test_source_failure_releases_writer(tmp_path, monkeypatch): + """Do not leak the encoder when the source raises during capture.""" + writer = Mock() + monkeypatch.setattr(cv2, 'VideoWriter', Mock(return_value=writer)) + + def frames(): + yield solid_faces() + raise RuntimeError('capture failed') + + with pytest.raises(RuntimeError, match='capture failed'): + save_panorama_video(frames(), tmp_path / 'panorama.mp4', (128, 64)) + writer.release.assert_called_once() + + +def test_writer_open_failure_is_reported(tmp_path, monkeypatch): + """A missing codec or unwritable path must not be reported as success.""" + writer = Mock() + writer.isOpened.return_value = False + monkeypatch.setattr(cv2, 'VideoWriter', Mock(return_value=writer)) + with pytest.raises(RuntimeError, match='Could not open video writer'): + save_panorama_video([solid_faces()], tmp_path / 'panorama.mp4', (128, 64)) + writer.write.assert_not_called() + writer.release.assert_called_once() + + +@pytest.mark.parametrize('fps', [0, -1, float('nan'), float('inf')]) +def test_rejects_invalid_fps(tmp_path, fps): + """Frame rates must be finite and positive.""" + with pytest.raises(ValueError, match='fps'): + save_panorama_video([solid_faces()], tmp_path / 'panorama.mp4', fps=fps) + + +def test_empty_source_does_not_create_output(tmp_path): + """Report an empty recording before creating an unusable file.""" + path = tmp_path / 'panorama.mp4' + with pytest.raises(ValueError, match='At least one'): + save_panorama_video(iter(()), path) + assert not path.exists() + + +def test_odd_video_height_is_rejected_before_encoding(tmp_path): + """Reject video dimensions that would silently crop the ERP aspect ratio.""" + path = tmp_path / 'panorama.mp4' + with pytest.raises(ValueError, match='height must be even'): + save_panorama_video([solid_faces()], path, (126, 63)) + assert not path.exists()