From 5d58cac198746aa769281bc44ecb10f8e34fc8d6 Mon Sep 17 00:00:00 2001 From: Willian Galvani Date: Mon, 3 Aug 2026 18:33:12 -0300 Subject: [PATCH 01/15] core: services: recorder_extractor: Repair recordings, let nginx serve them The frontend extracts video from the recording itself now, so the service no longer extracts MP4s and its repair loop is down to giving unindexed recordings an index, which is what random access needs. The static binary that used to do the extraction goes with it. Recordings themselves are read straight from nginx, which already serves the recorder directory under /userdata. Copying those bytes in Python cost the vehicle around eight times the CPU nginx needs for the same read, and made opening a recording ten times slower, since the index is read as a handful of tiny ranges where per request overhead is all there is. Co-authored-by: Cursor --- core/Dockerfile | 1 - core/services/recorder_extractor/main.py | 178 +++++++++++------------ core/tools/install-static-binaries.sh | 1 - core/tools/mcap-extractor/bootstrap.sh | 54 ------- core/tools/nginx/nginx.conf | 6 +- 5 files changed, 88 insertions(+), 152 deletions(-) delete mode 100755 core/tools/mcap-extractor/bootstrap.sh diff --git a/core/Dockerfile b/core/Dockerfile index 5b64031b5c..030fa2bbfb 100644 --- a/core/Dockerfile +++ b/core/Dockerfile @@ -131,7 +131,6 @@ COPY --from=download-binaries \ /usr/bin/mavlink-camera-manager \ /usr/bin/mavlink-server \ /usr/bin/mcap \ - /usr/bin/mcap-foxglove-video-extract \ /usr/bin/zenoh \ /usr/bin/ttyd \ /usr/bin/ diff --git a/core/services/recorder_extractor/main.py b/core/services/recorder_extractor/main.py index 041f514eee..a13a1ee371 100755 --- a/core/services/recorder_extractor/main.py +++ b/core/services/recorder_extractor/main.py @@ -4,11 +4,12 @@ import contextlib import logging import shutil +import struct import tempfile from functools import wraps from io import BytesIO from pathlib import Path -from typing import Any, Callable, List +from typing import Any, Callable, List, Optional from urllib.parse import quote from aiocache import cached @@ -17,7 +18,7 @@ from commonwealth.utils.logs import InterceptHandler, init_logger from commonwealth.utils.sentry_config import init_sentry_async from fastapi import APIRouter, FastAPI, HTTPException, status -from fastapi.responses import FileResponse, HTMLResponse, StreamingResponse +from fastapi.responses import HTMLResponse, StreamingResponse from fastapi_versioning import VersionedFastAPI, versioned_api_route from loguru import logger from pydantic import BaseModel @@ -25,14 +26,25 @@ SERVICE_NAME = "recorder-extractor" RECORDER_DIR = Path("/usr/blueos/userdata/recorder") +# Where nginx serves the same directory from, which recordings are read through +RECORDER_URL = "/userdata/recorder" +SERVICE_URL = "/recorder-extractor/v1.0/recorder" PORT = 9150 +SUPPORTED_SUFFIXES = (".mcap", ".mp4") -# Prevent thumbnails from being generated while MCAP extraction is running +MCAP_MAGIC = b"\x89MCAP0\r\n" +# opcode + u64 length + summary_start + summary_offset_start + summary_crc +MCAP_FOOTER_SIZE = 29 + +# Prevent thumbnails from being generated while a recording is being repaired thumbnail_lock = asyncio.Lock() -# Track MCAP files currently being processed +# Track MCAP files currently being repaired processing_mcap_files: set[str] = set() +# MCAP files whose index was already validated in this session +checked_mcap_files: set[Path] = set() + logging.basicConfig(handlers=[InterceptHandler()], level=logging.DEBUG) init_logger(SERVICE_NAME) logger.info("Starting Recorder Extractor service") @@ -45,7 +57,9 @@ class RecordingFile(BaseModel): modified: float download_url: str stream_url: str - thumbnail_url: str + # MCAP recordings are decoded by the frontend, which also renders their preview + thumbnail_url: Optional[str] + kind: str class ProcessingFile(BaseModel): @@ -73,9 +87,12 @@ def resolve_recording(filename: str) -> Path: if candidate.is_dir(): raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Invalid recording path.") - if candidate.suffix.lower() != ".mp4": - logger.warning(f"Rejected non-mp4 path: {candidate}") - raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Only .mp4 recordings are supported.") + if candidate.suffix.lower() not in SUPPORTED_SUFFIXES: + logger.warning(f"Rejected unsupported path: {candidate}") + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail=f"Only {' and '.join(SUPPORTED_SUFFIXES)} recordings are supported.", + ) if not candidate.exists() or not candidate.is_file(): logger.warning(f"Recording not found: {candidate}") raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Recording not found.") @@ -102,41 +119,42 @@ def parse_duration_ns(discover_output: str) -> int: return duration_ns -# pylint: disable=too-many-locals -async def check_and_recover_mcap(mcap_path: Path) -> None: +def mcap_is_indexed(mcap_path: Path) -> bool: """ - Check if mcap binary is available, run mcap doctor on the file, - and if it fails, run mcap recover to fix the file. + Check whether the recording ends with a footer pointing at a summary section. + + The frontend seeks through recordings using the MCAP index, so a file without one cannot be + streamed. Reading the footer costs a few bytes, unlike scanning the whole file. """ - # Check if mcap binary exists + try: + size = mcap_path.stat().st_size + if size < (len(MCAP_MAGIC) * 2) + MCAP_FOOTER_SIZE: + return False + with mcap_path.open("rb") as recording: + recording.seek(size - len(MCAP_MAGIC) - MCAP_FOOTER_SIZE) + footer = recording.read(MCAP_FOOTER_SIZE + len(MCAP_MAGIC)) + except OSError as exception: + logger.warning(f"Failed to read MCAP footer of {mcap_path}: {exception}") + return False + + if len(footer) != MCAP_FOOTER_SIZE + len(MCAP_MAGIC) or not footer.endswith(MCAP_MAGIC): + return False + summary_start: int = struct.unpack_from(" 0 + + +# pylint: disable=too-many-locals +async def recover_mcap(mcap_path: Path) -> None: + """Rewrite a recording with `mcap recover`, which restores its index and drops truncated data.""" mcap_binary = shutil.which("mcap") if not mcap_binary: - logger.warning("mcap binary not found, skipping doctor/recover check") + logger.warning("mcap binary not found, skipping recover") return - # Ensure path exists and is a file if not mcap_path.exists() or not mcap_path.is_file(): logger.debug(f"MCAP file not found or not a file: {mcap_path}") return - logger.info(f"Running mcap doctor on {mcap_path}") - # Run mcap doctor - doctor_cmd = [mcap_binary, "doctor", str(mcap_path)] - doctor_proc = await asyncio.create_subprocess_exec( - *doctor_cmd, - stdout=asyncio.subprocess.PIPE, - stderr=asyncio.subprocess.PIPE, - text=False, - ) - stdout_bytes, stderr_bytes = await doctor_proc.communicate() - stdout = stdout_bytes.decode("utf-8", "ignore") - stderr = stderr_bytes.decode("utf-8", "ignore") - - if doctor_proc.returncode == 0: - logger.info(f"mcap doctor passed for {mcap_path}: {stdout.strip()}") - return - - logger.warning(f"mcap doctor failed for {mcap_path} (code={doctor_proc.returncode}): {stderr.strip()}") logger.info(f"Attempting to recover {mcap_path}") # Create a temporary file path in the same directory as the mcap file @@ -258,57 +276,37 @@ async def build_thumbnail_bytes(path: Path) -> bytes: return stdout_bytes -async def extract_mcap_recordings() -> None: - """Periodically extract MP4 files from MCAP recordings.""" +async def repair_unindexed_recordings() -> None: + """ + Periodically make sure every recording carries an MCAP index. + + Video is extracted by the frontend straight from the recording, so the only thing the vehicle has + to guarantee is that recordings are seekable. Files left without an index, for example when the + vehicle lost power while recording, are rewritten by `mcap recover`. + """ while True: await asyncio.sleep(10) try: base = ensure_recorder_dir() for mcap_path in base.rglob("*.mcap"): - # If the folder already exists, it's already extracted or deleted by user - output_dir = mcap_path.with_suffix("") - if output_dir.exists(): + if mcap_path in checked_mcap_files: + continue + if mcap_is_indexed(mcap_path): + checked_mcap_files.add(mcap_path) continue - - logger.info(f"Checking if file is in use: {mcap_path}") if await file_is_open_async(mcap_path): - logger.info(f"Skipping MCAP extract, file in use: {mcap_path}") + logger.info(f"Recording is still being written, postponing repair: {mcap_path}") continue - # Check and recover MCAP file if mcap binary is available - await check_and_recover_mcap(mcap_path) - - command = [ - "mcap-foxglove-video-extract", - str(mcap_path), - "all", - "--output", - str(output_dir), - ] - logger.info(f"Extracting MCAP video to {output_dir} with command: {' '.join(command)}") mcap_relative = str(mcap_path.relative_to(base)) processing_mcap_files.add(mcap_relative) try: - async with thumbnail_lock: - process = await asyncio.create_subprocess_exec( - *command, - stdout=asyncio.subprocess.PIPE, - stderr=asyncio.subprocess.PIPE, - text=False, - ) - stdout_bytes, stderr_bytes = await process.communicate() - stdout = stdout_bytes.decode("utf-8", "ignore") - stderr = stderr_bytes.decode("utf-8", "ignore") + await recover_mcap(mcap_path) finally: processing_mcap_files.discard(mcap_relative) - if process.returncode != 0: - logger.error( - f"MCAP extract failed for {mcap_path} (code={process.returncode}): {stderr}", - ) - else: - logger.info(f"MCAP extract completed for {mcap_path}: {stdout.strip()}") + checked_mcap_files.add(mcap_path) except Exception as exception: - logger.exception(f"MCAP extraction loop failed: {exception}") + logger.exception(f"MCAP index check loop failed: {exception}") def to_http_exception(endpoint: Callable[..., Any]) -> Callable[..., Any]: @@ -340,28 +338,31 @@ async def wrapper(*args: Any, **kwargs: Any) -> Any: @recorder_router.get( "/files", response_model=List[RecordingFile], - summary="List available MP4 recordings under /usr/blueos/userdata/recorder.", + summary="List available recordings under /usr/blueos/userdata/recorder.", ) @to_http_exception async def list_recordings() -> List[RecordingFile]: - base_url = "/recorder-extractor/v1.0/recorder/files" files: List[RecordingFile] = [] base_path = ensure_recorder_dir() - mp4_files = sorted(base_path.rglob("*.mp4"), key=lambda item: item.stat().st_mtime, reverse=True) - for path in mp4_files: + recordings = [path for suffix in SUPPORTED_SUFFIXES for path in base_path.rglob(f"*{suffix}")] + for path in sorted(recordings, key=lambda item: item.stat().st_mtime, reverse=True): stat = path.stat() relative_path = path.relative_to(base_path) safe_path = str(relative_path) - encoded_path = quote(safe_path, safe="") + kind = path.suffix.lower().lstrip(".") + # Recordings are read straight from nginx, which serves them with byte ranges and sendfile, + # so playing and saving them costs the vehicle no more than the kernel copying bytes. + recording_url = f"{RECORDER_URL}/{quote(safe_path)}" files.append( RecordingFile( name=path.name, path=safe_path, size_bytes=stat.st_size, modified=stat.st_mtime, - download_url=f"{base_url}/{encoded_path}", - stream_url=f"{base_url}/{encoded_path}", - thumbnail_url=f"{base_url}/{encoded_path}/thumbnail", + download_url=recording_url, + stream_url=recording_url, + thumbnail_url=f"{SERVICE_URL}/files/{quote(safe_path, safe='')}/thumbnail" if kind == "mp4" else None, + kind=kind, ) ) return files @@ -370,11 +371,11 @@ async def list_recordings() -> List[RecordingFile]: @recorder_router.get( "/status", response_model=ProcessingStatus, - summary="Get MCAP extraction processing status.", + summary="Get MCAP repair status.", ) @to_http_exception async def get_processing_status() -> ProcessingStatus: - """Return MCAP files currently being processed.""" + """Return MCAP files currently being repaired.""" # Snapshot the set with list to avoid RuntimeError from concurrent mutation processing = [ProcessingFile(name=Path(path).name, path=path) for path in list(processing_mcap_files)] return ProcessingStatus(processing=processing) @@ -387,6 +388,11 @@ async def get_processing_status() -> ProcessingStatus: @to_http_exception async def get_recording_thumbnail(filename: str) -> StreamingResponse: path = resolve_recording(filename) + if path.suffix.lower() != ".mp4": + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail="Thumbnails are only available for extracted MP4 recordings.", + ) async with thumbnail_lock: thumbnail_bytes = await build_thumbnail_bytes(path) return StreamingResponse(BytesIO(thumbnail_bytes), media_type="image/jpeg") @@ -410,19 +416,9 @@ async def delete_recording(filename: str) -> None: ) from exception -@recorder_router.get( - "/files/{filename:path}", - summary="Download or stream a recording.", -) -@to_http_exception -async def get_recording(filename: str) -> FileResponse: - path = resolve_recording(filename) - return FileResponse(path, media_type="video/mp4", filename=path.name) - - fast_api_app = FastAPI( title="Recorder Extractor API", - description="Serve recorded MP4 files for playback and download.", + description="List recordings, keep them seekable, and preview them. Their bytes are served by nginx.", default_response_class=PrettyJSONResponse, ) fast_api_app.router.route_class = GenericErrorHandlingRoute @@ -449,7 +445,7 @@ async def root() -> HTMLResponse: async def main() -> None: - extractor_task = asyncio.create_task(extract_mcap_recordings()) + extractor_task = asyncio.create_task(repair_unindexed_recordings()) try: await init_sentry_async(SERVICE_NAME) diff --git a/core/tools/install-static-binaries.sh b/core/tools/install-static-binaries.sh index 4593c1f096..f35cdb71b1 100755 --- a/core/tools/install-static-binaries.sh +++ b/core/tools/install-static-binaries.sh @@ -15,7 +15,6 @@ TOOLS=( mavlink_camera_manager mavlink_server mcap - mcap-extractor recorder ttyd zenoh diff --git a/core/tools/mcap-extractor/bootstrap.sh b/core/tools/mcap-extractor/bootstrap.sh deleted file mode 100755 index 51bb9688af..0000000000 --- a/core/tools/mcap-extractor/bootstrap.sh +++ /dev/null @@ -1,54 +0,0 @@ -#!/bin/bash - -set -e - -PROJECT_NAME="mcap-foxglove-video-extract" -REPOSITORY_ORG="bluerobotics" -REPOSITORY_NAME="mcap-foxglove-video-extract" -VERSION="0.1.3" - -echo "Installing project $PROJECT_NAME version $VERSION" - -# Step 1: Prepare the download URL - -ARCH="$(uname -m)" -case "$ARCH" in - x86_64 | amd64) - BUILD_NAME="x86_64-unknown-linux-gnu" - ;; - armv7l | armhf) - BUILD_NAME="armv7-unknown-linux-gnueabihf" - ;; - aarch64 | arm64) - BUILD_NAME="aarch64-unknown-linux-gnu" - ;; - *) - echo "Architecture: $ARCH is unsupported, please create a new issue on https://github.com/${REPOSITORY_ORG}/${REPOSITORY_NAME}/issues" - exit 1 - ;; -esac -ARTIFACT_NAME="${PROJECT_NAME}-${BUILD_NAME}" -REMOTE_URL="https://github.com/${REPOSITORY_ORG}/${REPOSITORY_NAME}/releases/download/${VERSION}/${ARTIFACT_NAME}" -echo "Remote URL is $REMOTE_URL" - -# Step 2: Prepare the installation and tools paths - -if [ -n "$VIRTUAL_ENV" ]; then - BIN_DIR="$VIRTUAL_ENV/bin" -else - BIN_DIR="/usr/bin" -fi -mkdir -p "$BIN_DIR" - -BINARY_PATH="$BIN_DIR/$PROJECT_NAME" -echo "Installing to $BINARY_PATH" - -# Step 3: Download and install - -wget -q "$REMOTE_URL" -O "$BINARY_PATH" -chmod +x "$BINARY_PATH" -strip "$BINARY_PATH" - -echo "Installed binary type: $(file "$BINARY_PATH")" - -echo "Finished installing $PROJECT_NAME" \ No newline at end of file diff --git a/core/tools/nginx/nginx.conf b/core/tools/nginx/nginx.conf index 7b5650682d..9146fee02a 100644 --- a/core/tools/nginx/nginx.conf +++ b/core/tools/nginx/nginx.conf @@ -126,14 +126,10 @@ http { proxy_pass http://127.0.0.1:81/; } + # Recordings themselves are served from /userdata/recorder, this is only their metadata location /recorder-extractor/ { include cors.conf; proxy_pass http://127.0.0.1:9150/; - proxy_http_version 1.1; - proxy_set_header Range $http_range; - proxy_set_header If-Range $http_if_range; - add_header Accept-Ranges bytes; - proxy_buffering off; } location /disk-usage/ { From eac2819056758261010ce182120187dbaccad7c1 Mon Sep 17 00:00:00 2001 From: Willian Galvani Date: Mon, 3 Aug 2026 18:33:29 -0300 Subject: [PATCH 02/15] core: frontend: src: libs: Add in-browser MCAP video streaming Reads the MCAP index over HTTP range requests, decompresses only the chunks covering what is being watched, and muxes the H.264/H.265 frames into fragmented MP4 for Media Source Extensions. MSE is used instead of WebCodecs because WebCodecs requires a secure context and BlueOS is served over HTTP. Keyframes are located from the message indexes, so seeking costs a few kilobytes instead of downloading the surrounding chunks. The chunk index is read in 256 kB windows as playback and seeking reach them, so opening a recording costs about 40 kB rather than the 1.7 MB index a 1.3 GB file carries, and the record walker reports how far it got so a window can resume where the previous one stopped. Seeking has to leave the playhead alone. An append that was in flight when a seek arrived would move the playhead to its own start time, leaving the element waiting forever for media before the seek target that is never read, since reading only goes forward. Reading also holds off while a seek points the stream at its new time, because a stall or time update arriving in that window started reading from the old position again and quietly pulled hundreds of megabytes. The harness exercises playback and a mid-file seek for every video track of a recording, which is how the window sizes and the keyframe heuristic were measured against real recordings. Co-authored-by: Cursor --- core/frontend/.mcap-harness/.gitignore | 1 + core/frontend/.mcap-harness/harness.ts | 181 ++++++ core/frontend/.mcap-harness/sweep.sh | 33 ++ core/frontend/package.json | 1 + core/frontend/src/libs/mcap/bitstream.ts | 104 ++++ core/frontend/src/libs/mcap/codec.ts | 387 +++++++++++++ core/frontend/src/libs/mcap/frame-stream.ts | 128 +++++ core/frontend/src/libs/mcap/keyframe-index.ts | 120 ++++ core/frontend/src/libs/mcap/mp4.ts | 234 ++++++++ core/frontend/src/libs/mcap/player.ts | 538 ++++++++++++++++++ core/frontend/src/libs/mcap/reader.ts | 479 ++++++++++++++++ core/frontend/src/libs/mcap/record-reader.ts | 83 +++ core/frontend/src/libs/mcap/source.ts | 71 +++ core/frontend/src/libs/mcap/video-track.ts | 69 +++ core/frontend/yarn.lock | 5 + 15 files changed, 2434 insertions(+) create mode 100644 core/frontend/.mcap-harness/.gitignore create mode 100644 core/frontend/.mcap-harness/harness.ts create mode 100755 core/frontend/.mcap-harness/sweep.sh create mode 100644 core/frontend/src/libs/mcap/bitstream.ts create mode 100644 core/frontend/src/libs/mcap/codec.ts create mode 100644 core/frontend/src/libs/mcap/frame-stream.ts create mode 100644 core/frontend/src/libs/mcap/keyframe-index.ts create mode 100644 core/frontend/src/libs/mcap/mp4.ts create mode 100644 core/frontend/src/libs/mcap/player.ts create mode 100644 core/frontend/src/libs/mcap/reader.ts create mode 100644 core/frontend/src/libs/mcap/record-reader.ts create mode 100644 core/frontend/src/libs/mcap/source.ts create mode 100644 core/frontend/src/libs/mcap/video-track.ts diff --git a/core/frontend/.mcap-harness/.gitignore b/core/frontend/.mcap-harness/.gitignore new file mode 100644 index 0000000000..a6c7c2852d --- /dev/null +++ b/core/frontend/.mcap-harness/.gitignore @@ -0,0 +1 @@ +*.js diff --git a/core/frontend/.mcap-harness/harness.ts b/core/frontend/.mcap-harness/harness.ts new file mode 100644 index 0000000000..242c3ae6e6 --- /dev/null +++ b/core/frontend/.mcap-harness/harness.ts @@ -0,0 +1,181 @@ +import { + closeSync, openSync, readSync, statSync, writeFileSync, +} from 'fs' + +import { CodecConfig, ParameterSetCache, toMp4Sample } from '../src/libs/mcap/codec' +import VideoFrameStream from '../src/libs/mcap/frame-stream' +import { buildFragment, buildInitSegment, Mp4Sample } from '../src/libs/mcap/mp4' +import { McapIndexedReader } from '../src/libs/mcap/reader' +import { ByteSource } from '../src/libs/mcap/source' +import { listVideoTracks } from '../src/libs/mcap/video-track' + +class FileSource implements ByteSource { + bytesRead = 0 + + private fd: number + + constructor(private path: string) { + this.fd = openSync(path, 'r') + } + + async size(): Promise { + return statSync(this.path).size + } + + async read(offset: number, length: number): Promise { + const buffer = Buffer.allocUnsafe(length) + const read = readSync(this.fd, buffer, 0, length, offset) + this.bytesRead += read + return new Uint8Array(buffer.buffer, buffer.byteOffset, read) + } + + close(): void { + closeSync(this.fd) + } +} + +async function main(): Promise { + const [path, output, secondsArgument, seekArgument] = process.argv.slice(2) + const wantedSeconds = Number(secondsArgument ?? 10) + const seekSeconds = seekArgument === undefined ? null : Number(seekArgument) + + const source = new FileSource(path) + const metadataReader = await McapIndexedReader.open(source, { metadataOnly: true }) + console.log(` metadata-only index cost: ${(source.bytesRead / 1024).toFixed(1)} kB` + + `, duration ${(Number(metadataReader.summary.endTime - metadataReader.summary.startTime) / 1e9).toFixed(2)} s` + + `, video tracks ${listVideoTracks(metadataReader).map((item) => item.name).join(', ') || 'none'}`) + source.bytesRead = 0 + const reader = await McapIndexedReader.open(source) + const { summary } = reader + const durationSeconds = Number(summary.endTime - summary.startTime) / 1e9 + console.log(`file: ${path}`) + console.log(` size: ${(summary.size / 1e6).toFixed(1)} MB, duration: ${durationSeconds.toFixed(2)} s` + + `, chunks: ${summary.chunkIndexes.length}, channels: ${summary.channels.size}`) + console.log(` index cost: ${(source.bytesRead / 1024).toFixed(1)} kB`) + + const tracks = listVideoTracks(reader) + if (tracks.length === 0) { + console.log(' no video tracks') + return + } + for (const track of tracks) { + console.log(` track: ${track.name} (channel ${track.channelId}, ${track.frameCount} frames)`) + } + + const wanted = process.env.TRACK + const track = wanted === undefined + ? tracks.reduce((best, item) => (item.frameCount > best.frameCount ? item : best)) + : tracks.find((item) => item.name === wanted) + if (!track) { + throw new Error(`no track named ${wanted}`) + } + console.log(` playing: ${track.name}`) + const stream = new VideoFrameStream(reader, track) + + const beforeSeek = source.bytesRead + if (seekSeconds === null) { + stream.seekToStart() + } else { + await stream.seekToKeyframe(seekSeconds) + } + console.log(` keyframe lookup cost: ${((source.bytesRead - beforeSeek) / 1024).toFixed(1)} kB`) + + const parts: Uint8Array[] = [] + let config: CodecConfig | null = null + let frames: { logTime: bigint, data: Uint8Array, isKeyframe: boolean }[] = [] + let firstTime: bigint | null = null + let lastTime: bigint | null = null + let sequence = 1 + let keyframes = 0 + let droppedBeforeKeyframe = 0 + let sampleCount = 0 + const parameterSets = new ParameterSetCache() + + const flush = (all: boolean): void => { + const batch = all ? frames : frames.slice(0, -1) + if (batch.length === 0) { + return + } + frames = all ? [] : frames.slice(-1) + const rest = frames + const samples: Mp4Sample[] = batch.map((frame, index) => { + const next = batch[index + 1] ?? rest[0] + const duration = next ? Math.round(Number(next.logTime - frame.logTime) / 1000) : 33_333 + return { + data: frame.data, + duration: Math.min(Math.max(duration, 1000), 10_000_000), + isKeyframe: frame.isKeyframe, + } + }) + sampleCount += samples.length + const base = Math.round(Number(batch[0].logTime - summary.startTime) / 1000) + parts.push(buildFragment(samples, base, sequence)) + sequence += 1 + } + + const startBytes = source.bytesRead + for (;;) { + // eslint-disable-next-line no-await-in-loop + const frame = await stream.next() + if (!frame) { + break + } + const sample = toMp4Sample(frame.data, frame.format, parameterSets) + if (!config) { + if (!sample.isKeyframe) { + droppedBeforeKeyframe += 1 + if (droppedBeforeKeyframe % 30 === 0) { + // eslint-disable-next-line no-await-in-loop + await stream.skipToKeyframeHint() + } + continue + } + config = parameterSets.buildConfig(frame.format) + if (!config) { + throw new Error('keyframe without parameter sets') + } + console.log(` codec: ${config.codec} ${config.width}x${config.height}` + + ` (${config.description.length} byte description), format: ${frame.format}`) + parts.push(buildInitSegment(config)) + firstTime = frame.logTime + } + if (sample.isKeyframe) { + keyframes += 1 + } + lastTime = frame.logTime + frames.push({ logTime: frame.logTime, data: sample.data, isKeyframe: sample.isKeyframe }) + if (frames.length > 1 && Number(frames[frames.length - 1].logTime - frames[0].logTime) / 1e9 >= 0.5) { + flush(false) + } + if (firstTime !== null && Number(frame.logTime - firstTime) / 1e9 >= wantedSeconds) { + break + } + } + flush(true) + + const mediaSeconds = firstTime !== null && lastTime !== null ? Number(lastTime - firstTime) / 1e9 : 0 + const payload = source.bytesRead - startBytes + const startOffset = firstTime === null ? 0 : Number(firstTime - summary.startTime) / 1e9 + console.log(` frames dropped before first keyframe: ${droppedBeforeKeyframe}, keyframes: ${keyframes}`) + console.log(` muxed ${sampleCount} samples covering ${mediaSeconds.toFixed(2)} s` + + ` starting at ${startOffset.toFixed(2)} s`) + console.log(` downloaded ${(payload / 1e6).toFixed(2)} MB for playback` + + ` (${((payload * 8) / 1e6 / Math.max(mediaSeconds, 0.001)).toFixed(1)} Mbps)`) + console.log(` total read: ${(source.bytesRead / 1e6).toFixed(2)} MB of ${(summary.size / 1e6).toFixed(1)} MB`) + + const total = parts.reduce((size, part) => size + part.length, 0) + const file = new Uint8Array(total) + let offset = 0 + for (const part of parts) { + file.set(part, offset) + offset += part.length + } + writeFileSync(output, file) + console.log(` wrote ${output} (${(total / 1e6).toFixed(2)} MB)`) + source.close() +} + +main().catch((error) => { + console.error(error) + process.exit(1) +}) diff --git a/core/frontend/.mcap-harness/sweep.sh b/core/frontend/.mcap-harness/sweep.sh new file mode 100755 index 0000000000..7972a4c4ec --- /dev/null +++ b/core/frontend/.mcap-harness/sweep.sh @@ -0,0 +1,33 @@ +#!/bin/bash +# Exercises every video track of the given recordings: playback from the start plus a mid-file seek. +# Usage: yarn esbuild .mcap-harness/harness.ts --bundle --platform=node --outfile=.mcap-harness/harness.js +# .mcap-harness/sweep.sh ~/Downloads/*.mcap +cd "$(dirname "$0")/.." || exit 1 +for f in "$@"; do + base=$(basename "$f") + probe=$(node .mcap-harness/harness.js "$f" /tmp/probe.mp4 0.01 2>/dev/null) + tracks=$(echo "$probe" | rg -o 'video tracks (.*)$' -r '$1') + duration=$(echo "$probe" | rg -o 'duration ([0-9.]+) s' -r '$1' | head -1) + size=$(du -m "$f" | cut -f1) + echo "### $base (${size} MB, ${duration}s) tracks: $tracks" + if [ "$tracks" = "none" ]; then continue; fi + seek=$(python3 -c "print(round(float('$duration')*0.7, 1))") + echo "$tracks" | tr ',' '\n' | while read -r track; do + track=$(echo "$track" | xargs) + for mode in start seek; do + if [ "$mode" = start ]; then args=(2); else args=(2 "$seek"); fi + out=$(TRACK="$track" timeout 600 node .mcap-harness/harness.js "$f" /tmp/sweep.mp4 "${args[@]}" 2>&1) + if echo "$out" | rg -q '^Error|Error:'; then + printf ' %-32s %-5s FAILED: %s\n' "$track" "$mode" "$(echo "$out" | rg -o 'Error.*' | head -1)" + continue + fi + printf ' %-32s %-5s ok %-26s start=%-8s covered=%-6s payload=%-7s index+lookup=%s kB\n' \ + "$track" "$mode" \ + "$(echo "$out" | rg -o 'codec: (\S+ \S+)' -r '$1')" \ + "$(echo "$out" | rg -o 'starting at ([0-9.]+)' -r '$1')s" \ + "$(echo "$out" | rg -o 'covering ([0-9.]+)' -r '$1')s" \ + "$(echo "$out" | rg -o 'downloaded ([0-9.]+) MB' -r '$1')MB" \ + "$(echo "$out" | rg -o 'keyframe lookup cost: ([0-9.]+)' -r '$1')" + done + done +done diff --git a/core/frontend/package.json b/core/frontend/package.json index 00b12e0440..fac17fa13c 100644 --- a/core/frontend/package.json +++ b/core/frontend/package.json @@ -45,6 +45,7 @@ "date-fns": "^2.23.0", "file-saver": "^2.0.5", "fuse.js": "^6.6.2", + "fzstd": "0.1.1", "gl-matrix": "joaomariolago/gl-matrix#v3.4.x-extended-package", "gsap": "^3.12.3", "http-status-codes": "^2.2.0", diff --git a/core/frontend/src/libs/mcap/bitstream.ts b/core/frontend/src/libs/mcap/bitstream.ts new file mode 100644 index 0000000000..c0f8f49cfd --- /dev/null +++ b/core/frontend/src/libs/mcap/bitstream.ts @@ -0,0 +1,104 @@ +/** Annex B bitstream helpers shared by the H.264 and H.265 parsers. */ + +export interface NalUnit { + offset: number + length: number + /** NAL unit type, already shifted according to the codec's header layout. */ + type: number +} + +/** Splits an Annex B buffer into NAL units, skipping the start codes. */ +export function iterateNalUnits(data: Uint8Array, isH265: boolean): NalUnit[] { + const units: NalUnit[] = [] + let start = -1 + + function pushUnit(end: number): void { + if (start < 0 || end <= start) { + return + } + // Trailing zero bytes belong to the next start code, not to the payload. + let length = end - start + while (length > 0 && data[start + length - 1] === 0) { + length -= 1 + } + if (length > 0) { + const header = data[start] + units.push({ offset: start, length, type: isH265 ? header >> 1 & 0x3f : header & 0x1f }) + } + } + + let index = 0 + while (index + 2 < data.length) { + if (data[index] === 0 && data[index + 1] === 0 && data[index + 2] === 1) { + pushUnit(index) + index += 3 + start = index + } else { + index += 1 + } + } + pushUnit(data.length) + return units +} + +/** Removes emulation prevention bytes so the payload can be read as a raw bit sequence. */ +export function unescapeRbsp(data: Uint8Array): Uint8Array { + const output = new Uint8Array(data.length) + let written = 0 + let zeros = 0 + for (let index = 0; index < data.length; index += 1) { + const byte = data[index] + if (zeros === 2 && byte === 0x03) { + zeros = 0 + continue + } + zeros = byte === 0 ? zeros + 1 : 0 + output[written] = byte + written += 1 + } + return output.subarray(0, written) +} + +export class BitReader { + private position = 0 + + constructor(private data: Uint8Array) {} + + bit(): number { + const byte = this.data[this.position >> 3] ?? 0 + const value = byte >> 7 - (this.position & 7) & 1 + this.position += 1 + return value + } + + bits(count: number): number { + let value = 0 + for (let index = 0; index < count; index += 1) { + value = value * 2 + this.bit() + } + return value + } + + /** Unsigned Exp-Golomb coded value. */ + ue(): number { + let leadingZeros = 0 + while (this.bit() === 0 && leadingZeros < 32) { + leadingZeros += 1 + } + if (leadingZeros === 0) { + return 0 + } + return 2 ** leadingZeros - 1 + this.bits(leadingZeros) + } + + /** Signed Exp-Golomb coded value. */ + se(): number { + const value = this.ue() + const magnitude = Math.ceil(value / 2) + return value % 2 === 0 ? -magnitude : magnitude + } + + skipBits(count: number): void { + this.position += count + } +} diff --git a/core/frontend/src/libs/mcap/codec.ts b/core/frontend/src/libs/mcap/codec.ts new file mode 100644 index 0000000000..af1faf74e6 --- /dev/null +++ b/core/frontend/src/libs/mcap/codec.ts @@ -0,0 +1,387 @@ +/** + * Turns Annex B video frames from `foxglove.CompressedVideo` messages into everything the MP4 muxer + * needs: sample data in AVCC/HVCC form, frame dimensions and the decoder configuration record. + */ +import { BitReader, iterateNalUnits, unescapeRbsp } from './bitstream' + +export type VideoFormat = 'h264' | 'h265' + +export interface CodecConfig { + /** MP4 sample entry box, e.g. `avc1` or `hvc1`. */ + sampleEntry: string + /** Codec string for the MSE mime type, e.g. `avc1.42e01e`. */ + codec: string + width: number + height: number + /** avcC / hvcC payload, embedded in the sample entry and reused as the WebCodecs description. */ + description: Uint8Array +} + +const H264_NAL_SLICE_IDR = 5 +const H264_NAL_SPS = 7 +const H264_NAL_PPS = 8 +const H265_NAL_VPS = 32 +const H265_NAL_SPS = 33 +const H265_NAL_PPS = 34 +const H265_IRAP_RANGE = [16, 23] +const HIGH_PROFILES = [100, 110, 122, 244, 44, 83, 86, 118, 128, 138, 139, 134, 135] + +function toHex(value: number, digits = 2): string { + return value.toString(16).padStart(digits, '0') +} + +function skipScalingList(reader: BitReader, size: number): void { + let lastScale = 8 + let nextScale = 8 + for (let index = 0; index < size; index += 1) { + if (nextScale !== 0) { + nextScale = (lastScale + reader.se() + 256) % 256 + } + lastScale = nextScale === 0 ? lastScale : nextScale + } +} + +function parseH264Sps(nal: Uint8Array): { width: number, height: number } { + const reader = new BitReader(unescapeRbsp(nal.subarray(1))) + const profileIdc = reader.bits(8) + reader.skipBits(16) // constraint flags + level_idc + reader.ue() // seq_parameter_set_id + + let chromaFormatIdc = 1 + if (HIGH_PROFILES.includes(profileIdc)) { + chromaFormatIdc = reader.ue() + if (chromaFormatIdc === 3) { + reader.skipBits(1) // separate_colour_plane_flag + } + reader.ue() // bit_depth_luma_minus8 + reader.ue() // bit_depth_chroma_minus8 + reader.skipBits(1) // qpprime_y_zero_transform_bypass_flag + if (reader.bit() === 1) { + const lists = chromaFormatIdc !== 3 ? 8 : 12 + for (let index = 0; index < lists; index += 1) { + if (reader.bit() === 1) { + skipScalingList(reader, index < 6 ? 16 : 64) + } + } + } + } + + reader.ue() // log2_max_frame_num_minus4 + const picOrderCntType = reader.ue() + if (picOrderCntType === 0) { + reader.ue() // log2_max_pic_order_cnt_lsb_minus4 + } else if (picOrderCntType === 1) { + reader.skipBits(1) // delta_pic_order_always_zero_flag + reader.se() // offset_for_non_ref_pic + reader.se() // offset_for_top_to_bottom_field + const cycleLength = reader.ue() + for (let index = 0; index < cycleLength; index += 1) { + reader.se() // offset_for_ref_frame + } + } + + reader.ue() // max_num_ref_frames + reader.skipBits(1) // gaps_in_frame_num_value_allowed_flag + const widthInMbs = reader.ue() + 1 + const heightInMapUnits = reader.ue() + 1 + const frameMbsOnlyFlag = reader.bit() + if (frameMbsOnlyFlag === 0) { + reader.skipBits(1) // mb_adaptive_frame_field_flag + } + reader.skipBits(1) // direct_8x8_inference_flag + + let cropLeft = 0 + let cropRight = 0 + let cropTop = 0 + let cropBottom = 0 + if (reader.bit() === 1) { + cropLeft = reader.ue() + cropRight = reader.ue() + cropTop = reader.ue() + cropBottom = reader.ue() + } + + const subWidth = chromaFormatIdc === 3 ? 1 : 2 + const subHeight = chromaFormatIdc === 1 ? 2 : 1 + const cropUnitX = chromaFormatIdc === 0 ? 1 : subWidth + const cropUnitY = (chromaFormatIdc === 0 ? 1 : subHeight) * (2 - frameMbsOnlyFlag) + + return { + width: widthInMbs * 16 - cropUnitX * (cropLeft + cropRight), + height: (2 - frameMbsOnlyFlag) * heightInMapUnits * 16 - cropUnitY * (cropTop + cropBottom), + } +} + +function buildAvcC(sps: Uint8Array, pps: Uint8Array): Uint8Array { + const record = new Uint8Array(11 + sps.length + pps.length) + const view = new DataView(record.buffer) + const [, profileIdc, profileCompatibility, levelIdc] = sps + record[0] = 1 + record[1] = profileIdc + record[2] = profileCompatibility + record[3] = levelIdc + record[4] = 0xff // 6 reserved bits + lengthSizeMinusOne = 3 + record[5] = 0xe1 // 3 reserved bits + one SPS + view.setUint16(6, sps.length) + record.set(sps, 8) + record[8 + sps.length] = 1 + view.setUint16(9 + sps.length, pps.length) + record.set(pps, 11 + sps.length) + return record +} + +interface H265SpsInfo { + width: number + height: number + profileSpace: number + tierFlag: number + profileIdc: number + compatibilityFlags: number + constraintBytes: Uint8Array + levelIdc: number + chromaFormatIdc: number + bitDepthLuma: number + bitDepthChroma: number + maxSubLayersMinus1: number + temporalIdNesting: number +} + +function parseH265Sps(nal: Uint8Array): H265SpsInfo { + const payload = unescapeRbsp(nal.subarray(2)) + const reader = new BitReader(payload) + reader.skipBits(4) // sps_video_parameter_set_id + const maxSubLayersMinus1 = reader.bits(3) + const temporalIdNesting = reader.bit() + + // profile_tier_level: the general block is a fixed 12 bytes we can copy straight into hvcC. + const profileSpace = reader.bits(2) + const tierFlag = reader.bit() + const profileIdc = reader.bits(5) + let compatibilityFlags = 0 + for (let index = 0; index < 32; index += 1) { + compatibilityFlags = (compatibilityFlags << 1 | reader.bit()) >>> 0 + } + const constraintBytes = new Uint8Array(6) + for (let index = 0; index < 6; index += 1) { + constraintBytes[index] = reader.bits(8) + } + const levelIdc = reader.bits(8) + + const profilePresent: number[] = [] + const levelPresent: number[] = [] + for (let index = 0; index < maxSubLayersMinus1; index += 1) { + profilePresent.push(reader.bit()) + levelPresent.push(reader.bit()) + } + if (maxSubLayersMinus1 > 0) { + reader.skipBits(2 * (8 - maxSubLayersMinus1)) + } + for (let index = 0; index < maxSubLayersMinus1; index += 1) { + if (profilePresent[index] === 1) { + reader.skipBits(88) + } + if (levelPresent[index] === 1) { + reader.skipBits(8) + } + } + + reader.ue() // sps_seq_parameter_set_id + const chromaFormatIdc = reader.ue() + if (chromaFormatIdc === 3) { + reader.skipBits(1) // separate_colour_plane_flag + } + const widthInSamples = reader.ue() + const heightInSamples = reader.ue() + let cropLeft = 0 + let cropRight = 0 + let cropTop = 0 + let cropBottom = 0 + if (reader.bit() === 1) { + cropLeft = reader.ue() + cropRight = reader.ue() + cropTop = reader.ue() + cropBottom = reader.ue() + } + const bitDepthLuma = reader.ue() + 8 + const bitDepthChroma = reader.ue() + 8 + + const subWidth = chromaFormatIdc === 1 || chromaFormatIdc === 2 ? 2 : 1 + const subHeight = chromaFormatIdc === 1 ? 2 : 1 + + return { + width: widthInSamples - subWidth * (cropLeft + cropRight), + height: heightInSamples - subHeight * (cropTop + cropBottom), + profileSpace, + tierFlag, + profileIdc, + compatibilityFlags, + constraintBytes, + levelIdc, + chromaFormatIdc, + bitDepthLuma, + bitDepthChroma, + maxSubLayersMinus1, + temporalIdNesting, + } +} + +function buildHvcC(info: H265SpsInfo, arrays: { type: number, nals: Uint8Array[] }[]): Uint8Array { + const bytes: number[] = [] + function pushUint16(value: number): void { + bytes.push(value >> 8 & 0xff, value & 0xff) + } + + bytes.push(1) + bytes.push((info.profileSpace & 3) << 6 | (info.tierFlag & 1) << 5 | info.profileIdc & 0x1f) + bytes.push( + info.compatibilityFlags >>> 24 & 0xff, + info.compatibilityFlags >>> 16 & 0xff, + info.compatibilityFlags >>> 8 & 0xff, + info.compatibilityFlags & 0xff, + ) + bytes.push(...info.constraintBytes) + bytes.push(info.levelIdc) + pushUint16(0xf000) // reserved + min_spatial_segmentation_idc + bytes.push(0xfc) // reserved + parallelismType + bytes.push(0xfc | info.chromaFormatIdc & 3) + bytes.push(0xf8 | info.bitDepthLuma - 8 & 7) + bytes.push(0xf8 | info.bitDepthChroma - 8 & 7) + pushUint16(0) // avgFrameRate, unknown + bytes.push( + info.maxSubLayersMinus1 + 1 << 3 | (info.temporalIdNesting & 1) << 2 | 3, + ) + bytes.push(arrays.length) + for (const array of arrays) { + bytes.push(0x80 | array.type & 0x3f) + pushUint16(array.nals.length) + for (const nal of array.nals) { + pushUint16(nal.length) + bytes.push(...nal) + } + } + return new Uint8Array(bytes) +} + +function h265CodecString(info: H265SpsInfo): string { + let reversed = 0 + for (let index = 0; index < 32; index += 1) { + reversed = (reversed << 1 | info.compatibilityFlags >>> index & 1) >>> 0 + } + const space = ['', 'A', 'B', 'C'][info.profileSpace] + const tier = info.tierFlag === 1 ? 'H' : 'L' + const constraints = [...info.constraintBytes] + while (constraints.length > 0 && constraints[constraints.length - 1] === 0) { + constraints.pop() + } + const suffix = constraints.map((byte) => `.${toHex(byte).toUpperCase()}`).join('') + return `hvc1.${space}${info.profileIdc}.${reversed.toString(16)}.${tier}${info.levelIdc}${suffix}` +} + +function isKeyframeNal(type: number, isH265: boolean): boolean { + return isH265 ? type >= H265_IRAP_RANGE[0] && type <= H265_IRAP_RANGE[1] : type === H264_NAL_SLICE_IDR +} + +export function isKeyframe(frame: Uint8Array, format: VideoFormat): boolean { + const isH265 = format === 'h265' + return iterateNalUnits(frame, isH265).some(({ type }) => isKeyframeNal(type, isH265)) +} + +/** + * Keeps the most recent parameter sets seen in a stream. + * + * Foxglove requires keyframes to carry their parameter sets, which is what makes seeking into the + * middle of a recording possible, but some older recordings only send them once at the start of the + * stream. Remembering them lets those recordings play too. + */ +export class ParameterSetCache { + private vps: Uint8Array[] = [] + + private sps: Uint8Array[] = [] + + private pps: Uint8Array[] = [] + + observe(type: number, nal: Uint8Array, format: VideoFormat): void { + if (format === 'h265') { + switch (type) { + case H265_NAL_VPS: this.vps = [nal]; break + case H265_NAL_SPS: this.sps = [nal]; break + case H265_NAL_PPS: this.pps = [nal]; break + default: break + } + return + } + if (type === H264_NAL_SPS) { + this.sps = [nal] + } else if (type === H264_NAL_PPS) { + this.pps = [nal] + } + } + + observeFrame(frame: Uint8Array, format: VideoFormat): void { + for (const unit of iterateNalUnits(frame, format === 'h265')) { + this.observe(unit.type, frame.subarray(unit.offset, unit.offset + unit.length), format) + } + } + + get complete(): boolean { + return this.sps.length > 0 && this.pps.length > 0 + } + + buildConfig(format: VideoFormat): CodecConfig | null { + if (!this.complete) { + return null + } + const [sps] = this.sps + if (format === 'h264') { + const { width, height } = parseH264Sps(sps) + return { + sampleEntry: 'avc1', + codec: `avc1.${toHex(sps[1])}${toHex(sps[2])}${toHex(sps[3])}`, + width, + height, + description: buildAvcC(sps, this.pps[0]), + } + } + + const info = parseH265Sps(sps) + const arrays = [ + { type: H265_NAL_VPS, nals: this.vps }, + { type: H265_NAL_SPS, nals: this.sps }, + { type: H265_NAL_PPS, nals: this.pps }, + ].filter((array) => array.nals.length > 0) + return { + sampleEntry: 'hvc1', + codec: h265CodecString(info), + width: info.width, + height: info.height, + description: buildHvcC(info, arrays), + } + } +} + +/** + * Converts Annex B start codes into the 4 byte length prefixes MP4 samples use, reporting whether + * the frame can be decoded on its own and collecting any parameter sets it carries. + */ +export function toMp4Sample( + frame: Uint8Array, + format: VideoFormat, + parameterSets?: ParameterSetCache, +): { data: Uint8Array, isKeyframe: boolean } { + const isH265 = format === 'h265' + const units = iterateNalUnits(frame, isH265) + const size = units.reduce((total, unit) => total + unit.length + 4, 0) + const data = new Uint8Array(size) + const view = new DataView(data.buffer) + let offset = 0 + let keyframe = false + for (const unit of units) { + const nal = frame.subarray(unit.offset, unit.offset + unit.length) + view.setUint32(offset, unit.length) + data.set(nal, offset + 4) + offset += unit.length + 4 + keyframe = keyframe || isKeyframeNal(unit.type, isH265) + parameterSets?.observe(unit.type, nal, format) + } + return { data, isKeyframe: keyframe } +} diff --git a/core/frontend/src/libs/mcap/frame-stream.ts b/core/frontend/src/libs/mcap/frame-stream.ts new file mode 100644 index 0000000000..9576192d9e --- /dev/null +++ b/core/frontend/src/libs/mcap/frame-stream.ts @@ -0,0 +1,128 @@ +/** + * Sequential frame reader for a single video channel. It downloads one MCAP chunk at a time, so + * memory and bandwidth stay proportional to what is actually being watched. + */ +import { KeyframeLocator } from './keyframe-index' +import { McapIndexedReader } from './reader' +import { VideoFrame, VideoFrameDecoder, VideoTrack } from './video-track' + +/** Chunks worth of message index to inspect while looking for a keyframe around a seek target. */ +const KEYFRAME_SEARCH_CHUNKS = 64 + +export default class VideoFrameStream { + private chunkPositions: number[] = [] + + /** Chunk index length the positions were built from, since the reader loads the index lazily. */ + private knownChunks = -1 + + private cursor = 0 + + private queue: VideoFrame[] = [] + + private decoder: VideoFrameDecoder + + private locator: KeyframeLocator + + constructor(private reader: McapIndexedReader, public readonly track: VideoTrack) { + const channel = reader.summary.channels.get(track.channelId) + if (!channel) { + throw new Error(`Recording has no channel ${track.channelId}.`) + } + this.decoder = VideoFrameDecoder.create(reader, channel) + this.locator = new KeyframeLocator(reader, track.channelId, () => this.positions()) + } + + private positions(): number[] { + const { length } = this.reader.summary.chunkIndexes + if (length !== this.knownChunks) { + this.chunkPositions = this.reader.chunkIndexesForChannel(this.track.channelId) + this.knownChunks = length + } + return this.chunkPositions + } + + get startTime(): bigint { + return this.reader.summary.startTime + } + + get durationSeconds(): number { + return Number(this.reader.summary.endTime - this.startTime) / 1e9 + } + + get bytesRead(): number { + return this.reader.source.bytesRead + } + + toSeconds(logTime: bigint): number { + return Number(logTime - this.startTime) / 1e9 + } + + toLogTime(seconds: number): bigint { + return this.startTime + BigInt(Math.max(0, Math.round(seconds * 1e9))) + } + + private positionForTime(logTime: bigint): number { + const chunkIndex = this.reader.findChunkIndexAtTime(this.track.channelId, logTime) + const position = this.positions().indexOf(chunkIndex) + return position >= 0 ? position : 0 + } + + /** + * Moves the stream to the keyframe that starts playback for the given time, preferring the + * keyframe at or before it so that seeking does not skip forward over content. + */ + async seekToKeyframe(seconds: number, signal?: AbortSignal): Promise { + const logTime = this.toLogTime(seconds) + await this.reader.loadChunkIndexesUntil(logTime, signal) + const position = this.positionForTime(logTime) + const hint = await this.locator.findBefore(logTime, position, KEYFRAME_SEARCH_CHUNKS, signal) + ?? await this.locator.findForward(position, KEYFRAME_SEARCH_CHUNKS, signal) + this.queue = [] + this.cursor = hint?.position ?? position + } + + /** + * Positions the stream at the beginning of the recording. No keyframe lookup here: the first chunk + * has to be downloaded either way, and scanning it for the first real keyframe starts playback + * earlier than any size-based guess could. + */ + seekToStart(): void { + this.queue = [] + this.cursor = 0 + } + + /** + * Skips to the next chunk that looks like it holds a keyframe. Meant for callers that already + * decoded frames and found none, so nothing playable is left behind. Returns false when the guess + * points at the chunk being read, in which case reading on is cheaper than jumping. + */ + async skipToKeyframeHint(signal?: AbortSignal): Promise { + const current = Math.max(0, this.cursor - 1) + const hint = await this.locator.findForward(current, KEYFRAME_SEARCH_CHUNKS, signal) + if (!hint || hint.position <= current) { + return false + } + this.queue = [] + this.cursor = hint.position + return true + } + + async next(signal?: AbortSignal): Promise { + while (this.queue.length === 0) { + const positions = this.positions() + if (this.cursor >= positions.length) { + // eslint-disable-next-line no-await-in-loop + if (!await this.reader.loadMoreChunkIndexes(signal)) { + return null + } + continue + } + const chunkIndex = positions[this.cursor] + this.cursor += 1 + // eslint-disable-next-line no-await-in-loop + const messages = await this.reader.readChunkMessages(chunkIndex, this.track.channelId, signal) + this.queue = messages.map((message) => this.decoder.decode(message)) + } + return this.queue.shift() ?? null + } +} diff --git a/core/frontend/src/libs/mcap/keyframe-index.ts b/core/frontend/src/libs/mcap/keyframe-index.ts new file mode 100644 index 0000000000..878aba990c --- /dev/null +++ b/core/frontend/src/libs/mcap/keyframe-index.ts @@ -0,0 +1,120 @@ +/** + * Locates keyframes using only the MCAP message indexes. + * + * Compressed video frames carry their parameter sets on keyframes, which makes keyframes several + * times larger than delta frames. Message indexes tell us the size of every message for a few + * kilobytes per chunk, so a keyframe can be found without downloading any video payload. The result + * is a hint: the caller still decodes the bitstream and keeps scanning forward if the guess was off. + */ +import { McapIndexedReader, McapMessageEntry } from './reader' + +/** + * How much bigger than the typical frame a message has to be to look like a keyframe. Kept low on + * purpose: a missed keyframe makes a seek land later than asked, while a false one only costs the + * delta frames decoded before the real keyframe shows up. + */ +const KEYFRAME_SIZE_RATIO = 1.8 +const MINIMUM_SAMPLES_FOR_MEDIAN = 8 + +export interface KeyframeHint { + /** Position within the channel's chunk list. */ + position: number + logTime: bigint +} + +export class KeyframeLocator { + private entriesByChunk = new Map() + + private observedSizes: number[] = [] + + private unavailable = false + + constructor( + private reader: McapIndexedReader, + private channelId: number, + private chunkPositions: () => number[], + ) {} + + private async entriesAt(position: number, signal?: AbortSignal): Promise { + const positions = this.chunkPositions() + if (this.unavailable || position < 0 || position >= positions.length) { + return null + } + const chunkIndex = positions[position] + const cached = this.entriesByChunk.get(chunkIndex) + if (cached) { + return cached + } + + const entries = await this.reader.readChunkMessageEntries(chunkIndex, this.channelId, signal) + if (!entries) { + this.unavailable = true + return null + } + this.entriesByChunk.set(chunkIndex, entries) + this.observedSizes.push(...entries.map((entry) => entry.size)) + return entries + } + + private threshold(): number | null { + if (this.observedSizes.length < MINIMUM_SAMPLES_FOR_MEDIAN) { + return null + } + const sorted = [...this.observedSizes].sort((left, right) => left - right) + const median = sorted[Math.floor(sorted.length / 2)] + return median > 0 ? median * KEYFRAME_SIZE_RATIO : null + } + + /** First chunk at or after `position` that looks like it contains a keyframe. */ + async findForward(position: number, maxChunks: number, signal?: AbortSignal): Promise { + const candidates: { position: number, entry: McapMessageEntry }[] = [] + for (let offset = 0; offset < maxChunks; offset += 1) { + const current = position + offset + // eslint-disable-next-line no-await-in-loop + const entries = await this.entriesAt(current, signal) + if (!entries) { + break + } + candidates.push(...entries.map((entry) => ({ position: current, entry }))) + const threshold = this.threshold() + if (threshold === null) { + continue + } + const match = candidates.find(({ entry }) => entry.size >= threshold) + if (match) { + return { position: match.position, logTime: match.entry.logTime } + } + candidates.length = 0 + } + return null + } + + /** Last keyframe at or before `logTime`, searching backwards from the chunk that holds it. */ + async findBefore( + logTime: bigint, + position: number, + maxChunks: number, + signal?: AbortSignal, + ): Promise { + for (let offset = 0; offset < maxChunks; offset += 1) { + const current = position - offset + if (current < 0) { + break + } + // eslint-disable-next-line no-await-in-loop + const entries = await this.entriesAt(current, signal) + if (!entries) { + break + } + const threshold = this.threshold() + if (threshold === null) { + continue + } + const match = [...entries].reverse().find((entry) => entry.size >= threshold && entry.logTime <= logTime) + if (match) { + return { position: current, logTime: match.logTime } + } + } + return null + } +} diff --git a/core/frontend/src/libs/mcap/mp4.ts b/core/frontend/src/libs/mcap/mp4.ts new file mode 100644 index 0000000000..e9ebd11ad9 --- /dev/null +++ b/core/frontend/src/libs/mcap/mp4.ts @@ -0,0 +1,234 @@ +/** + * Minimal fragmented MP4 writer. Produces an initialization segment plus `moof`/`mdat` fragments, + * which is exactly what Media Source Extensions consumes, and is also a valid MP4 file when the + * pieces are concatenated. + */ +import { CodecConfig } from './codec' + +export const MP4_TIMESCALE = 1_000_000 + +const TRACK_ID = 1 +const KEYFRAME_FLAGS = 0x02000000 +const DELTA_FRAME_FLAGS = 0x01010000 + +export interface Mp4Sample { + data: Uint8Array + /** Duration in `MP4_TIMESCALE` units. */ + duration: number + isKeyframe: boolean +} + +function concat(parts: Uint8Array[]): Uint8Array { + const total = parts.reduce((size, part) => size + part.length, 0) + const output = new Uint8Array(total) + let offset = 0 + for (const part of parts) { + output.set(part, offset) + offset += part.length + } + return output +} + +function box(type: string, ...children: Uint8Array[]): Uint8Array { + const payload = concat(children) + const output = new Uint8Array(8 + payload.length) + const view = new DataView(output.buffer) + view.setUint32(0, output.length) + for (let index = 0; index < 4; index += 1) { + output[4 + index] = type.charCodeAt(index) + } + output.set(payload, 8) + return output +} + +class Writer { + private bytes: number[] = [] + + uint8(...values: number[]): Writer { + this.bytes.push(...values.map((value) => value & 0xff)) + return this + } + + uint16(value: number): Writer { + return this.uint8(value >> 8, value) + } + + uint32(value: number): Writer { + return this.uint8(value >> 24, value >> 16, value >> 8, value) + } + + uint64(value: number): Writer { + const high = Math.floor(value / 2 ** 32) + return this.uint32(high).uint32(value >>> 0) + } + + ascii(value: string): Writer { + return this.uint8(...[...value].map((character) => character.charCodeAt(0))) + } + + zeros(count: number): Writer { + return this.uint8(...new Array(count).fill(0)) + } + + raw(data: Uint8Array): Writer { + this.bytes.push(...data) + return this + } + + build(): Uint8Array { + return new Uint8Array(this.bytes) + } +} + +const UNITY_MATRIX = new Writer() + .uint32(0x00010000).uint32(0).uint32(0) + .uint32(0) + .uint32(0x00010000) + .uint32(0) + .uint32(0) + .uint32(0) + .uint32(0x40000000) + .build() + +function sampleEntry(config: CodecConfig): Uint8Array { + const configurationBox = config.sampleEntry === 'hvc1' ? 'hvcC' : 'avcC' + const header = new Writer() + .zeros(6) + .uint16(1) // data_reference_index + .zeros(16) // pre_defined + reserved + .uint16(config.width) + .uint16(config.height) + .uint32(0x00480000) // 72 dpi horizontal + .uint32(0x00480000) // 72 dpi vertical + .uint32(0) + .uint16(1) // frame_count + .zeros(32) // compressor name + .uint16(0x0018) // depth + .uint16(0xffff) // pre_defined = -1 + .build() + return box(config.sampleEntry, header, box(configurationBox, config.description)) +} + +export function buildInitSegment(config: CodecConfig): Uint8Array { + const ftyp = box( + 'ftyp', + new Writer().ascii('isom').uint32(0x200).ascii('isom') + .ascii('iso2') + .ascii('avc1') + .ascii('mp41') + .build(), + ) + + const mvhd = box('mvhd', new Writer() + .uint32(0) // version + flags + .uint32(0).uint32(0) // creation + modification time + .uint32(MP4_TIMESCALE) + .uint32(0) // duration, unknown for fragmented files + .uint32(0x00010000) // rate + .uint16(0x0100) // volume + .zeros(10) + .raw(UNITY_MATRIX) + .zeros(24) // pre_defined + .uint32(TRACK_ID + 1) + .build()) + + const tkhd = box('tkhd', new Writer() + .uint32(0x00000007) // version 0, track enabled + in movie + in preview + .uint32(0).uint32(0) + .uint32(TRACK_ID) + .uint32(0) + .uint32(0) // duration + .zeros(8) + .uint16(0) // layer + .uint16(0) // alternate_group + .uint16(0) // volume + .uint16(0) + .raw(UNITY_MATRIX) + .uint32(config.width * 0x10000) + .uint32(config.height * 0x10000) + .build()) + + const mdhd = box('mdhd', new Writer() + .uint32(0) + .uint32(0).uint32(0) + .uint32(MP4_TIMESCALE) + .uint32(0) + .uint16(0x55c4) // 'und' language + .uint16(0) + .build()) + + const hdlr = box('hdlr', new Writer() + .uint32(0) + .uint32(0) + .ascii('vide') + .zeros(12) + .ascii('VideoHandler') + .zeros(1) + .build()) + + const stbl = box( + 'stbl', + box('stsd', new Writer().uint32(0).uint32(1).build(), sampleEntry(config)), + box('stts', new Writer().uint32(0).uint32(0).build()), + box('stsc', new Writer().uint32(0).uint32(0).build()), + box('stsz', new Writer().uint32(0).uint32(0).uint32(0) + .build()), + box('stco', new Writer().uint32(0).uint32(0).build()), + ) + + const minf = box( + 'minf', + box('vmhd', new Writer().uint32(0x00000001).zeros(8).build()), + box('dinf', box('dref', new Writer().uint32(0).uint32(1).build(), box('url ', new Writer().uint32(1).build()))), + stbl, + ) + + const trak = box('trak', tkhd, box('mdia', mdhd, hdlr, minf)) + const mvex = box('mvex', box('trex', new Writer() + .uint32(0) + .uint32(TRACK_ID) + .uint32(1) // default_sample_description_index + .uint32(0) + .uint32(0) + .uint32(0) + .build())) + + return concat([ftyp, box('moov', mvhd, trak, mvex)]) +} + +/** + * Builds a media fragment. `baseMediaDecodeTime` places the samples on the media timeline, which is + * what lets us append segments out of order after a seek. + */ +export function buildFragment(samples: Mp4Sample[], baseMediaDecodeTime: number, sequence: number): Uint8Array { + function trun(dataOffset: number): Uint8Array { + const writer = new Writer() + .uint32(0x01000701) // version 1, data offset + per sample duration, size and flags + .uint32(samples.length) + .uint32(dataOffset) + for (const sample of samples) { + writer + .uint32(sample.duration) + .uint32(sample.data.length) + .uint32(sample.isKeyframe ? KEYFRAME_FLAGS : DELTA_FRAME_FLAGS) + } + return box('trun', writer.build()) + } + + function moofFor(dataOffset: number): Uint8Array { + return box( + 'moof', + box('mfhd', new Writer().uint32(0).uint32(sequence).build()), + box( + 'traf', + box('tfhd', new Writer().uint32(0x00020000).uint32(TRACK_ID).build()), // default-base-is-moof + box('tfdt', new Writer().uint32(0x01000000).uint64(baseMediaDecodeTime).build()), + trun(dataOffset), + ), + ) + } + + const moof = moofFor(moofFor(0).length + 8) + const mdat = box('mdat', concat(samples.map((sample) => sample.data))) + return concat([moof, mdat]) +} diff --git a/core/frontend/src/libs/mcap/player.ts b/core/frontend/src/libs/mcap/player.ts new file mode 100644 index 0000000000..2cc6bddc39 --- /dev/null +++ b/core/frontend/src/libs/mcap/player.ts @@ -0,0 +1,538 @@ +/** + * Streams video stored in an MCAP recording straight into a `