diff --git a/.gitattributes b/.gitattributes index 0857b88e8947db355faab8cf76c7e1562923924e..fb3a8d58506e0b3d8bc3ec8ac1f2e1a740c7b9aa 100644 --- a/.gitattributes +++ b/.gitattributes @@ -41,3 +41,6 @@ collector/**/*.jpg filter=lfs diff=lfs merge=lfs -text collector/**/*.png filter=lfs diff=lfs merge=lfs -text collector/**/*.bmp filter=lfs diff=lfs merge=lfs -text collector/**/*.task filter=lfs diff=lfs merge=lfs -text + +# processor 文档截图走 LFS(HF 拒绝非 LFS 二进制) +processor/**/*.png filter=lfs diff=lfs merge=lfs -text diff --git a/processor/CHANGELOG.md b/processor/CHANGELOG.md new file mode 100644 index 0000000000000000000000000000000000000000..52f61d1787741f0fccb7d9ab46a74a872ef8683d --- /dev/null +++ b/processor/CHANGELOG.md @@ -0,0 +1,11 @@ +# 更新日志 + +## 1.7.0(2026-09-08) + +- 文档:中英文使用文档重构,补充各页面带标注截图与引导图 +- 部署:uvicorn + worker 开机自启(systemd),任务队列不再因 worker 未启动而积压 +- 其他:.gitignore 增加大目录防误传规则 + +## 1.6.9(2026-09-07) + +- 深度预览预热扩展、导出链路优化(本地构建/指纹复用)、下载 gzip 修复 diff --git a/processor/README.md b/processor/README.md index 62be1bc7402b121be9959163e16944be424548b3..08870e84b679c6b2b65b47fa8da66d77933d62e2 100644 --- a/processor/README.md +++ b/processor/README.md @@ -1,13 +1,15 @@ -# Data Acquisition +# processor [English](README.md) | [中文](README.zh-CN.md) -> Documentation: 2026-09-03 · Backend: `1.6.8` · Web frontend: `1.5.6` +> Documentation: 2026-09-02 · Backend: `1.6.8` · Web frontend: `1.5.6` -Data Acquisition is EgoData's service for data capture, processing, review, and export. It is +processor is EgoData's service for data capture, processing, review, and export. It is designed for embodied AI and robot manipulation data. The system consists of a FastAPI backend, an asynchronous Worker, Web review pages, and Workflow Studio. +> Beginners: start with the [illustrated user guide](docs/en/README.md). + ```text Capture upload → Project matching → Workflow processing → AI/Human review → LeRobot/HDF5 export ``` @@ -16,7 +18,7 @@ Capture upload → Project matching → Workflow processing → AI/Human review - [System overview](#1-system-overview) - [Installation and startup](#7-installation-and-startup) -- [Code layout and repository tree](#2-code-layout-and-repository-tree) +- [Code layout](#2-code-layout) - [Project data structure](#3-project-data-structure) - [Video data specification](#4-video-data-specification) - [Workflow modules](#5-workflow-modules) @@ -54,66 +56,30 @@ Web review pages LeRobot v2.1 / v3.0 or HDF5 ``` -## 2. Code layout and repository tree +The local `processor/` directory is the only source of frontend and backend code. The +remote server provides data storage or SFTP access only; its deprecated frontend and backend +code must not be used as the project source. -
-Click to expand: source tree and runtime directories +## 2. Code layout -The current maintained source tree is shown below. Runtime data, caches, virtual environments, -local backups, test exports, and downloaded model weights are intentionally omitted. +
+Click to expand: code layout ```text -Data Acquisition/ -├── app/ -│ ├── api/ # Project, workflow, user, and Worker APIs -│ ├── processing/ -│ │ ├── black_glove/ # Black-glove detection, tracking, and pose backend -│ │ └── modules/ # RGB, RGB-D, stereo, hand, glove, review, and export nodes -│ ├── prompts/ # AI annotation prompts and vocabularies -│ ├── routes/ # Page, session, video, annotation, and export routes -│ ├── ai_annotation.py # Local/API VLM annotation pipeline -│ ├── browser_preview.py # Browser preview metadata and media preparation -│ ├── export_engine.py # Export job orchestration -│ ├── hdf5_export.py # HDF5 dataset writer -│ ├── lerobot_export.py # LeRobot v3.0 exporter and metadata writer -│ ├── lerobot_v21.py # LeRobot v2.1 normalization support -│ ├── project_dataset.py # Canonical project and Episode operations -│ ├── storage.py / remote_storage.py # Local and SFTP-backed storage -│ ├── workflow_*.py # Workflow schema, binding, and dispatch -│ ├── models.py / database.py # Runtime state models and persistence -│ └── main.py # FastAPI application entry point -├── worker/ -│ ├── runner.py # Job claim, heartbeat, execution, and callback loop -│ ├── client.py # Backend Worker API client -│ └── README.md # Worker-specific notes +processor/ +├── app/ # FastAPI, storage, workflows, processing, and export logic +├── worker/ # Asynchronous processing Worker +├── scripts/ # Deployment, startup, checks, and maintenance scripts ├── web/ -│ ├── templates/ # Server-rendered application pages -│ ├── static/ -│ │ ├── js/ # Player, depth preview, overlays, review, and UI JS -│ │ └── workflow-studio/ # Built Workflow Studio assets served by FastAPI -│ └── workflow-studio/ -│ └── src/ -│ ├── api/ # Frontend API clients -│ ├── components/ # Canvas, nodes, palette, drawer, and settings UI -│ ├── store/ # Workflow and UI state stores -│ └── App.tsx # Workflow Studio application root -├── scripts/ # Setup, startup, migration, and maintenance scripts -│ ├── run_backend_linux.sh # Linux FastAPI startup -│ ├── run_worker_linux.sh # Linux Worker startup -│ ├── hot_reload_backend.py # Development backend reload supervisor -│ ├── hot_reload_worker.py # Development Worker reload supervisor -│ ├── migrate_*.py / repack_*.py # Dataset migration and repair tools -│ └── systemd/ # Linux service templates -├── tests/ -│ └── test_depth_codec.py # 12-bit depth codec tests -├── requirements*.txt # Python dependency manifests -├── deploy.py # Cross-platform deployment entry point -├── deploy.sh / deploy.bat # Linux and Windows deployment entry points -├── Dockerfile / .dockerignore # Container build files -├── .env.example # Safe configuration template -├── .gitignore # Runtime data, secrets, models, and cache exclusions -├── README.md # English documentation -└── README.zh-CN.md # Chinese documentation +│ ├── templates/ # Web page templates +│ ├── static/ # CSS, JavaScript, and frontend build artifacts +│ └── workflow-studio/ # React + Vite workflow editor source +├── tests/ # Automated tests +├── requirements*.txt # Python dependency manifests +├── deploy.py # Cross-platform deployment entry point +├── deploy.sh # Linux deployment entry point +├── deploy.bat # Windows deployment entry point +└── .env.example # Configuration template ``` Runtime data is stored under `STORAGE_DIR` by default: @@ -122,9 +88,7 @@ Runtime data is stored under `STORAGE_DIR` by default: data/ ├── sessions/ # Project datasets ├── state/ # System state, queues, run records, and export jobs -├── logs/ # Runtime and archive-sync logs -├── .meta-history/ # Metadata migration rollback snapshots -└── tmp/ # Upload, AI, and Worker temporary files +└── tmp/ # Upload and Worker temporary files ``` Runtime data, model weights, archives, credentials, and virtual environments must not be @@ -132,7 +96,6 @@ committed to Git. Dependency manifests such as `requirements*.txt` and `package. kept in the repository.
- ## 3. Project data structure
@@ -211,7 +174,7 @@ RGB and is never duplicated alongside the RGB stream. New metric-depth videos use: ```text -HEVC (H.265) in MP4 +HEVC (H.265) in MP4 / hvc1 / gray12le / qp=6 / range=full ``` The depth video stores 12-bit logarithmic depth codes, not pseudo-color images: @@ -224,11 +187,6 @@ The corresponding feature in `meta/info.json` is marked with `video.is_depth_map depth encoding parameters. Readers dequantize the codes back to millimeters. The original depth video is preserved during processing and export. -Depth is never exported as a JET/RGB pseudo-color video. Exported depth remains a single-channel -`gray12le` video containing the original 12-bit logarithmic codes. The frontend must use the -`codes_to_heatmap_bgr()` rule for decoded depth codes; it must not quantize decoded codes a second -time as if they were millimeter values. - ### Frontend depth preview Pseudo-color depth is used only for browser preview and is never written back to the dataset: @@ -388,17 +346,6 @@ RGB video in the browser; a second skeleton video is not saved. System-level run under `data/state/runs/`, and temporary export products under `data/state/exports/`; neither is part of the project dataset directory. -Glove tactile data is kept as fixed-size numeric arrays in the Parquet data, for example -`observation.tactile.left` and `observation.tactile.right` with shape `[256]`. The current -LeRobot training export does not generate or retain tactile MP4 files. Tactile visualization is a -frontend preview concern; it is not required as a video feature by the LeRobot export. - -The export module supports LeRobot v2.1 and v3.0. A v3.0 export is directly readable by the -current official LeRobot reader. A v2.1 export follows the official v2.1 layout and must be -converted with the official `convert_dataset_v21_to_v30` tool before using a v3.0 reader. -Batch export creates one dataset directory per selected Episode or project selection; each output -contains only the canonical `data/`, `meta/`, and `videos/` directories. - The upload archive lifecycle is: receive → extract → normalize → write to project → validate → update status. Temporary upload files and extraction directories are cleaned only after extraction, atomic commit, and validation all succeed. On failure, staging content is retained for retry and @@ -411,7 +358,7 @@ diagnosis. ### Linux ```bash -cd "Data Acquisition" +cd "processor" cp .env.example .env # Edit .env and set API_KEY, WORKER_API_KEY, JWT_SECRET, and storage settings. chmod +x deploy.sh @@ -432,14 +379,14 @@ model, and endpoint in the `AI Annotation` settings in Workflow Studio. ### Windows ```bat -cd /d "Data Acquisition" +cd /d "processor" copy .env.example .env && deploy.bat ``` ### Start the backend manually ```bash -cd "Data Acquisition" +cd "processor" source .venv-linux/bin/activate export PYTHONPATH="$PWD" ./scripts/run_backend_linux.sh @@ -454,7 +401,7 @@ uvicorn app.main:app --host 0.0.0.0 --port 8000 ### Start the Worker ```bash -cd "Data Acquisition" +cd "processor" export EGODATA_SERVER_URL=http://127.0.0.1:8000 export EGODATA_WORKER_API_KEY='same Worker API key as the server' ./scripts/run_worker_linux.sh @@ -466,7 +413,7 @@ independent and must not be copied between operating systems. ### Workflow Studio ```bash -cd "Data Acquisition/web/workflow-studio" +cd "processor/web/workflow-studio" npm ci npm run dev ``` @@ -484,7 +431,7 @@ The build output is written to `web/static/workflow-studio/` and served by FastA
Click to expand: environment variables and security configuration -Configuration is stored in `Data Acquisition/.env` or environment variables and must not be +Configuration is stored in `processor/.env` or environment variables and must not be committed: | Setting | Purpose | @@ -505,22 +452,6 @@ credentials in logs, workflow JSON, frontend code, or documentation.
-### GitHub publication checklist - -Before publishing this directory: - -```bash -git status --short --ignored -git check-ignore -v .env data .backups .venv-linux models -git ls-files | rg -i '(^|/)(\.env|.*secret.*|.*token.*|.*credential.*|.*password.*|.*\.pem|.*\.key)$' -``` - -The repository excludes runtime datasets, temporary files, local state, backups, virtual -environments, archives, and secrets through `.gitignore`. Keep `.env` local and commit only -`.env.example`. Do not use `git add -f` for `data/`, `.backups/`, model weights, or generated -exports. If a real credential was ever placed in a tracked file, rotate it before publishing; -removing it from the working tree alone does not remove it from Git history. - ## 9. API
@@ -559,7 +490,7 @@ The running FastAPI OpenAPI document is the authoritative API contract. Backend and Worker checks: ```bash -cd "Data Acquisition" +cd "processor" python -m compileall app worker python deploy.py --check-only PYTEST_DISABLE_PLUGIN_AUTOLOAD=1 python -m pytest -q tests/test_depth_codec.py @@ -568,7 +499,7 @@ PYTEST_DISABLE_PLUGIN_AUTOLOAD=1 python -m pytest -q tests/test_depth_codec.py Frontend check: ```bash -cd "Data Acquisition/web/workflow-studio" +cd "processor/web/workflow-studio" npm run build ``` diff --git a/processor/README.zh-CN.md b/processor/README.zh-CN.md index 8e1a6947513f6c523188a836fcce7ccda014df81..363a79a0cd2046d6a1b0b7d0753c1f2914da90f4 100644 --- a/processor/README.zh-CN.md +++ b/processor/README.zh-CN.md @@ -1,12 +1,14 @@ -# Data Acquisition +# processor [中文](README.zh-CN.md) | [English](README.md) > 文档版本：2026-09-02 · 后端版本：`1.6.8` · WEB 前端版本：`1.5.6` -Data Acquisition 是 EgoData 的数据采集、处理、审核和导出服务，面向具身智能与机器人 +processor 是 EgoData 的数据采集、处理、审核和导出服务，面向具身智能与机器人 操作数据。系统由 FastAPI 后端、异步 Worker、Web 审核页面和 Workflow Studio 组成。 +> 项目文档：[项目文档介绍](docs/README.md)。 + ```text 采集端上传 → 项目匹配 → 工作流处理 → AI/人工审核 → LeRobot/HDF5 导出 ``` @@ -53,13 +55,16 @@ Web 审核页面 LeRobot v2.1 / v3.0 或 HDF5 ``` +本机 `processor/` 目录是前端和后端代码的唯一来源。远端服务器只提供数据目录 +或 SFTP 存储，不作为本项目的前端、后端代码来源；服务器中的旧代码已废弃。 + ## 2. 代码目录
点击展开：代码目录 ```text -Data Acquisition/ +processor/ ├── app/ # FastAPI、存储、工作流、处理和导出逻辑 ├── worker/ # 异步处理 Worker ├── scripts/ # 部署、启动、检查和数据维护脚本 @@ -88,7 +93,6 @@ data/ `package.json` 等依赖清单应保留在仓库中。
- ## 3. 项目数据结构
@@ -165,7 +169,7 @@ RGB 和 Depth 始终作为独立数据流保存。纯 Depth 流不会被当作 R 新的真实深度视频统一使用： ```text -HEVC (H.265) in MP4 +HEVC (H.265) in MP4 / hvc1 / gray12le / qp=6 / range=full ``` 深度视频保存的是 12-bit 对数深度码，不是伪彩色图像： @@ -342,7 +346,7 @@ meta/episodes/chunk-000/episode_000000.parquet ### Linux ```bash -cd "Data Acquisition" +cd "processor" cp .env.example .env # 编辑 .env，设置 API_KEY、WORKER_API_KEY、JWT_SECRET 和存储配置。 chmod +x deploy.sh @@ -363,14 +367,14 @@ python deploy.py --no-services # 前台运行，不安装自启动服务。 ### Windows ```bat -cd /d "Data Acquisition" +cd /d "processor" copy .env.example .env && deploy.bat ``` ### 手动启动后端 ```bash -cd "Data Acquisition" +cd "processor" source .venv-linux/bin/activate export PYTHONPATH="$PWD" ./scripts/run_backend_linux.sh @@ -385,7 +389,7 @@ uvicorn app.main:app --host 0.0.0.0 --port 8000 ### 启动 Worker ```bash -cd "Data Acquisition" +cd "processor" export EGODATA_SERVER_URL=http://127.0.0.1:8000 export EGODATA_WORKER_API_KEY='与服务端一致的 Worker API Key' ./scripts/run_worker_linux.sh @@ -397,7 +401,7 @@ Windows 使用 `scripts/run_worker_windows.ps1`。Linux 和 Windows 的虚拟环 ### Workflow Studio ```bash -cd "Data Acquisition/web/workflow-studio" +cd "processor/web/workflow-studio" npm ci npm run dev ``` @@ -415,7 +419,7 @@ npm run build
点击展开：环境变量和安全配置 -配置写入 `Data Acquisition/.env` 或环境变量，不提交到仓库： +配置写入 `processor/.env` 或环境变量，不提交到仓库： | 配置 | 作用 | | --- | --- | @@ -472,7 +476,7 @@ GET /openapi.json 后端和 Worker 基础检查： ```bash -cd "Data Acquisition" +cd "processor" python -m compileall app worker python deploy.py --check-only PYTEST_DISABLE_PLUGIN_AUTOLOAD=1 python -m pytest -q tests/test_depth_codec.py @@ -481,7 +485,7 @@ PYTEST_DISABLE_PLUGIN_AUTOLOAD=1 python -m pytest -q tests/test_depth_codec.py 前端检查： ```bash -cd "Data Acquisition/web/workflow-studio" +cd "processor/web/workflow-studio" npm run build ``` diff --git a/processor/app/ai_annotation.py b/processor/app/ai_annotation.py index 75982f45e2b60471a0b57e93f588d15dc2c763fb..1a4e16cb15febd3454ce070182d0240b0b28bb1f 100644 --- a/processor/app/ai_annotation.py +++ b/processor/app/ai_annotation.py @@ -2265,6 +2265,33 @@ async def _upload_video_file(client: httpx.AsyncClient, upload_base: str, return None +def _slice_propose_clip(video_path: Path, out_dir: Path) -> Path | None: + """整片规划专用切片:降采样到 ≤640px / 5fps 低码率。 + + 长批次(如 305s)原分辨率整片 base64 内联会打出 50MiB+ 的 JSON, + 租用网关直接把连接掐断(ReadError)。规划阶段只需要看清动作边界, + 5fps/640px 足够;时长不变,start_s/end_s 仍映射源视频时间。 + """ + import subprocess + + out = out_dir / "propose_clip.mp4" + cmd = ["ffmpeg", "-y", "-loglevel", "error", + "-i", str(video_path), + "-vf", "scale='min(640,iw)':-2,fps=5", + "-c:v", "libx264", "-crf", "30", "-preset", "veryfast", + "-pix_fmt", "yuv420p", + "-movflags", "+faststart", "-an", str(out)] + try: + r = subprocess.run(cmd, capture_output=True, text=True, timeout=600) + except (OSError, subprocess.TimeoutExpired) as exc: + print(f"[ai_annotation] propose clip error: {exc}") + return None + if r.returncode != 0 or not out.is_file() or out.stat().st_size < 1024: + print(f"[ai_annotation] propose clip failed: {r.stderr[:300]}") + return None + return out + + def _inline_video_data_url(clip: Path, *, max_bytes: int = API_INLINE_VIDEO_MAX_BYTES, label: str = "API VLM") -> str: @@ -2414,7 +2441,7 @@ async def _vlm_propose_api(video_path: Path, fps: float, total_frames: int, raise VLMAnnotationError("configuration_error", err) whole = {"start_frame_index": 0, "end_frame_index": max(0, total_frames - 1)} - clip = _slice_segment_clip(video_path, fps, whole, None, None, tmp_dir) + clip = _slice_propose_clip(video_path, tmp_dir) if clip is None: raise VLMAnnotationError("video_slice_failed", "API VLM 输入视频切片失败") try: @@ -2425,6 +2452,8 @@ async def _vlm_propose_api(video_path: Path, fps: float, total_frames: int, except ValueError as exc: raise VLMAnnotationError( "api_video_too_large", str(exc)) from exc + # propose 剪辑统一 5fps,fps 提示与剪辑实际一致(时长不变,时间轴对齐) + propose_fps = 5.0 prompts = _prompts() template = prompts.get("plan_prompt_zh" if lang == "zh" else "plan_prompt_en") vocab = prompts.get("vocab_zh" if lang == "zh" else "vocab_en", "") @@ -2437,7 +2466,7 @@ async def _vlm_propose_api(video_path: Path, fps: float, total_frames: int, "\n\n输出要求：最终答案必须放在 assistant 的 content 字段中，" "只输出完整 JSON，不要把答案只放在 reasoning 或思考内容中。") content = [ - _api_video_content(vlm_cfg, video_ref, fps), + _api_video_content(vlm_cfg, video_ref, propose_fps), {"type": "text", "text": prompt}, ] try: diff --git a/processor/app/api/projects.py b/processor/app/api/projects.py index 1b7e372d0ce0563b258a55db43282bfaaebb023f..880b17277dc253cc80cbd856e2b803b2313999b6 100644 --- a/processor/app/api/projects.py +++ b/processor/app/api/projects.py @@ -576,6 +576,15 @@ def get_hierarchy( if s in n["project"]["name"].lower() or any(s in e["name"].lower() or s in e["id"].lower() for e in n["episodes"])] + # 预热深度预览:列表可见批次在用户点开前后台生成(尽力而为) + try: + from app.routes.ingestion import _trigger_depth_prewarm + + _trigger_depth_prewarm( + [e["id"] for node in tree for e in node.get("episodes", [])]) + except Exception: + pass + return {"projects": tree, "total": sum(len(n["episodes"]) for n in tree)} diff --git a/processor/app/browser_preview.py b/processor/app/browser_preview.py index 3b41b1ec61a194085e61ed6a383543c25bb4dd67..ae57de6a8c1c2b77450588efc41f5f6eb9a81887 100644 --- a/processor/app/browser_preview.py +++ b/processor/app/browser_preview.py @@ -13,10 +13,12 @@ the configured storage directory. from __future__ import annotations import hashlib +import json import os import shutil import subprocess import threading +import time from pathlib import Path from app.config import settings @@ -25,6 +27,9 @@ from app.config import settings _LOCKS: dict[str, threading.Lock] = {} _LOCKS_GUARD = threading.Lock() +_FASTSTART_CACHE: dict[str, bool] = {} +_FASTSTART_CACHE_LOCK = threading.Lock() + def _cache_root() -> Path: root = settings.upload_staging_root / "browser-preview" @@ -43,6 +48,296 @@ def _lock_for(key: str) -> threading.Lock: return _LOCKS.setdefault(key, threading.Lock()) +# ── Local source mirror (depth streams) ────────────────────────────── +# Depth videos live on the SSHFS mount, where random reads for seek/decode +# cost seconds. A bit-identical local copy turns every later decode into a +# local-disk read. This is a disposable transport cache of the SOURCE file +# (never a derived/colorized asset); the authoritative stream stays in the +# configured storage directory. + +_MIRROR_MAX_BYTES = 8 * 1024 ** 3 +_MIRROR_INFLIGHT: set[str] = set() +_MIRROR_INFLIGHT_GUARD = threading.Lock() + + +def _mirror_root() -> Path: + root = settings.upload_staging_root / "depth-mirror" + root.mkdir(parents=True, exist_ok=True) + return root + + +def _prune_mirror(keep: Path) -> None: + """Remove oldest mirror files beyond the byte budget (LRU by mtime).""" + try: + files = [p for p in _mirror_root().iterdir() + if p.is_file() and p.suffix == ".mp4"] + except OSError: + return + total = sum(p.stat().st_size for p in files if p != keep) + ( + keep.stat().st_size if keep.is_file() else 0) + if total <= _MIRROR_MAX_BYTES: + return + for path in sorted(files, key=lambda p: p.stat().st_mtime_ns): + if path == keep or total <= _MIRROR_MAX_BYTES: + continue + try: + total -= path.stat().st_size + path.unlink(missing_ok=True) + except OSError: + pass + + +def _copy_mirror(source: Path, destination: Path, key: str) -> None: + """Background copy to a temp file, then atomic rename.""" + temporary = destination.with_suffix(".part") + temporary.unlink(missing_ok=True) + try: + shutil.copy2(source, temporary) + if not temporary.is_file() or temporary.stat().st_size <= 0: + raise RuntimeError("mirror copy produced an empty file") + os.replace(temporary, destination) + _prune_mirror(destination) + except (OSError, RuntimeError): + temporary.unlink(missing_ok=True) + finally: + with _MIRROR_INFLIGHT_GUARD: + _MIRROR_INFLIGHT.discard(key) + + +def _write_turbo_cube(path: Path) -> None: + """Write the static 4096-entry 1D Turbo cube used by the transcode. + + 4096 entries keep the range mapping at 12-bit precision: a 256-entry LUT + evaluated after an 8-bit intermediate caused visible banding (only ~58 + distinct color levels survived the chain instead of ~173). + """ + if path.is_file(): + return + from app.lerobot_v21 import turbo_lut + + lut = turbo_lut(4096) + lines = ["TITLE Turbo", "LUT_1D_SIZE 4096", "", + "DOMAIN_MIN 0 0 0", "DOMAIN_MAX 1 1 1", ""] + for r, g, b in lut: + lines.append(f"{r / 255.0:.6f} {g / 255.0:.6f} {b / 255.0:.6f}") + path.write_text("\n".join(lines) + "\n", encoding="ascii") + + +def _depth_stats_path(source: Path) -> Path: + key = _cache_key(source) + return _cache_root() / f"depth-{key}.stats.json" + + +def read_depth_stats(source: Path) -> dict | None: + """Disk-persisted q01/q99 bounds for ``source`` (same fingerprint key). + + Written once by :func:`_ensure_depth_stats` during preview generation; + survives restarts so neither the stats endpoint nor a re-transcode ever + pays a second full-stream decode for the same source file. + """ + try: + path = _depth_stats_path(source) + if not path.is_file(): + return None + data = json.loads(path.read_text(encoding="utf-8")) + if (isinstance(data, dict) + and isinstance(data.get("q01"), (int, float)) + and isinstance(data.get("q99"), (int, float)) + and float(data["q99"]) > float(data["q01"])): + return data + except (OSError, ValueError): + pass + return None + + +def _histogram_depth_stats(source: Path) -> dict | None: + """Single-decode q01/q99 code bounds via a streaming 4096-bin histogram. + + The stream is read as raw gray12le codes and histogrammed in O(1) + memory; percentiles are exact over every decoded pixel. The previous + ``select=`` sampling pass decoded the whole stream anyway while keeping + only a fraction of the frames, so nothing is lost. + """ + import numpy as np + + ffmpeg = shutil.which("ffmpeg") + if not ffmpeg: + return None + hist = np.zeros(4096, dtype=np.int64) + proc = subprocess.Popen( + [ffmpeg, "-v", "error", "-i", str(source), "-map", "0:v:0", + "-f", "rawvideo", "-pix_fmt", "gray12le", "-"], + stdout=subprocess.PIPE, stderr=subprocess.DEVNULL) + try: + while True: + chunk = proc.stdout.read(8 * 1024 * 1024) + if not chunk: + break + values = np.frombuffer(chunk, dtype=" dict | None: + """Return q01/q99 for ``source``, computed once and persisted to disk. + + ``source`` is the authoritative path used for the cache fingerprint; + ``local`` (the transport mirror) is decoded when it is ready so the pass + never competes with an SSHFS random read. + """ + cached = read_depth_stats(source) + if cached is not None: + return cached + with _lock_for(f"stats:{_cache_key(source)}"): + cached = read_depth_stats(source) + if cached is not None: + return cached + stats = _histogram_depth_stats(local or source) + if stats is None: + return None + try: + _depth_stats_path(source).write_text( + json.dumps(stats), encoding="utf-8") + except OSError: + pass + return stats + + +def depth_preview_exists(source: Path) -> bool: + """Cheap cache-hit check for prewarm scheduling (no transcode).""" + try: + destination = _cache_root() / f"depth-{_cache_key(Path(source))}.mp4" + return destination.is_file() and destination.stat().st_size > 0 + except OSError: + return False + + +def ensure_depth_h264_preview(source: Path, q01: float = 0.0, + q99: float = 4095.0) -> Path: + """Generate (once) an 8-bit Turbo-colored H.264 preview of a depth stream. + + The browser display pipeline already quantizes codes into an 8-bit Turbo + ramp; this bakes the identical mapping (same clamp/scale math, same + colormap) into a disposable local MP4 so the review page plays it through + a native ``