"""Project-level LeRobot 2.1 dataset storage. The active dataset root is ``sessions/``. Individual uploads are episodes in the same ``data/meta/videos`` tree; they are not nested under a second batch directory. Workflow outputs are merged into the corresponding episode parquet files; preview overlays are rendered in the browser and are never stored as a second video. There is deliberately no ``processed/``, ``meta/processing/`` or ``state/runs/`` directory inside a project. This module is shared by the migration command and the upload path so both produce the same layout. """ from __future__ import annotations import copy import json import os import re import shutil import threading import time from datetime import datetime, timezone from pathlib import Path from typing import Any, Iterable from app.lerobot_v21 import ( DEPTH_MAX_MM, DEPTH_MIN_MM, DEPTH_QMAX, DEPTH_QP, DEPTH_VIDEO_ENCODING, _source_feature, encode_depth_png_sequence, iter_video_streams, normalize_metadata_sources, normalize_extracted_dataset, source_key_from_video, ) _EPISODE_RE = re.compile(r"episode_(\d+)", re.IGNORECASE) _CHUNK_RE = re.compile(r"chunk[-_](\d+)", re.IGNORECASE) _CANONICAL_ROOTS = {"data", "meta", "videos"} PROJECT_CHUNK_SIZE = 1000 EPISODES_METADATA_CHUNK_SIZE = 1000 def _episode_parquet_files(project_root: Path) -> list[Path]: """Return one canonical Parquet index file per episode.""" root = Path(project_root) / "meta" / "episodes" if not root.is_dir(): return [] return sorted( path for path in root.glob("chunk-*/episode_*.parquet") if path.is_file() ) def has_episode_index(project_root: Path) -> bool: """Whether a project has the canonical Parquet episode index.""" root = Path(project_root) return bool(_episode_parquet_files(root)) def write_project_episode_index(project_root: Path, rows: Iterable[dict[str, Any]]) -> None: """Write the canonical sharded Parquet episode index. Each Parquet file contains exactly one episode row. Episode files are grouped into chunks of 1000 and use the same episode number as data/video. """ import pyarrow as pa import pyarrow.parquet as pq project_root = Path(project_root) normalized = [] for row in rows: item = dict(row) # Arrow cannot serialize an empty struct. Missing per-episode stats # are valid; the project-level stats.json writer will compute a # fallback from the frame parquet when needed. if item.get("stats") == {}: item.pop("stats", None) normalized.append(item) normalized.sort(key=lambda row: int(row.get("episode_index", 10**9))) episodes_root = project_root / "meta" / "episodes" episodes_root.mkdir(parents=True, exist_ok=True) for path in sorted(episodes_root.rglob("*.parquet")): path.unlink(missing_ok=True) for path in sorted(episodes_root.rglob("*"), reverse=True): if path.is_dir() and not any(path.iterdir()): path.rmdir() for row in normalized: episode_index = int(row.get("episode_index", 0)) chunk_index = episode_index // EPISODES_METADATA_CHUNK_SIZE target = (episodes_root / f"chunk-{chunk_index:03d}" / f"episode_{episode_index:06d}.parquet") target.parent.mkdir(parents=True, exist_ok=True) pq.write_table(pa.Table.from_pylist([row]), target) # The legacy flat JSONL index is intentionally not generated anymore. (project_root / "meta" / "episodes.jsonl").unlink(missing_ok=True) # ★ 必须失效缓存 —— 本函数是**索引的唯一写入点**，而 project_episode_rows() # 读的是带 60s TTL 的模块级缓存。不失效的话，同一次调用里"写完再读"会拿到 # 删除前的旧行。 # # 实测后果（2026-09-11 引入缓存时漏掉这步，删除从此全坏）： # delete_project_episode: 读一次(预热缓存) → 删文件 → 写索引 → 校验 # 校验又调 project_episode_rows → 旧缓存 → 报 "missing data" → RuntimeError # →「清空回收站」500。盘上的索引其实是对的，纯粹是缓存没跟上。 # （delete_episode 末尾也会 invalidate_session_cache()，但那要等异常之后。） invalidate_project_cache(project_root) def is_project_dataset(root: Path) -> bool: root = Path(root) return ((root / "data").is_dir() and has_episode_index(root) and (root / "videos").is_dir() and (root / "meta" / "info.json").is_file() and (root / "meta" / "stats.json").is_file() and (root / "meta" / "tasks.json").is_file()) def _read_json(path: Path, default: Any = None) -> Any: try: return json.loads(path.read_text(encoding="utf-8")) except (OSError, ValueError, TypeError): return default def _write_json(path: Path, value: Any) -> None: path.parent.mkdir(parents=True, exist_ok=True) path.write_text(json.dumps(value, ensure_ascii=False, indent=2), encoding="utf-8") def _read_jsonl(path: Path) -> list[dict[str, Any]]: rows: list[dict[str, Any]] = [] if not path.is_file(): return rows for line in path.read_text(encoding="utf-8").splitlines(): try: value = json.loads(line) except (ValueError, TypeError): continue if isinstance(value, dict): rows.append(value) return rows def _read_tasks_json(path: Path) -> list[dict[str, Any]]: """Read the compact internal task index stored as one JSON document.""" value = _read_json(path, None) if isinstance(value, dict): value = value.get("tasks", value) if not isinstance(value, list): return [] rows: list[dict[str, Any]] = [] for index, item in enumerate(value): if isinstance(item, str): rows.append({"task_index": index, "task": item}) continue if not isinstance(item, dict): continue text = str(item.get("task") or item.get("description") or "").strip() if not text: continue try: task_index = int(item.get("task_index", item.get("task_id", index))) except (TypeError, ValueError): task_index = index rows.append({"task_index": task_index, "task": text}) return rows def _write_tasks_json(project_root: Path, rows: Iterable[dict[str, Any]]) -> None: normalized = [] for index, row in enumerate(rows): if not isinstance(row, dict): continue text = str(row.get("task") or row.get("description") or "").strip() if not text: continue try: task_index = int(row.get("task_index", row.get("task_id", index))) except (TypeError, ValueError): task_index = index normalized.append({"task_index": task_index, "task": text}) normalized.sort(key=lambda row: int(row["task_index"])) path = Path(project_root) / "meta" / "tasks.json" _write_json(path, normalized) (Path(project_root) / "meta" / "tasks.jsonl").unlink(missing_ok=True) def _summary_count(summary: dict[str, Any]) -> int: value = summary.get("count", 1) if isinstance(value, list): value = value[0] if value else 1 try: return max(1, int(value)) except (TypeError, ValueError): return 1 def _as_numeric_array(value: Any): import numpy as np return np.asarray(value, dtype=np.float64) def _aggregate_episode_stats(rows: Iterable[dict[str, Any]]) -> dict[str, dict[str, Any]]: """Aggregate per-episode summaries into LeRobot-style global stats.""" import numpy as np accumulated: dict[str, dict[str, Any]] = {} for row in rows: summaries = row.get("stats") if isinstance(row, dict) else None if not isinstance(summaries, dict): continue for name, summary in summaries.items(): if not isinstance(summary, dict): continue try: count = _summary_count(summary) minimum = _as_numeric_array(summary["min"]) maximum = _as_numeric_array(summary["max"]) mean = _as_numeric_array(summary["mean"]) std = _as_numeric_array(summary.get("std", 0.0)) if not (minimum.shape == maximum.shape == mean.shape == std.shape): continue except (KeyError, TypeError, ValueError): continue item = accumulated.get(name) if item is None: accumulated[name] = { "min": minimum.copy(), "max": maximum.copy(), "sum": mean * count, "m2": (std ** 2) * count, "count": count, } continue if item["sum"].shape != mean.shape: continue old_count = int(item["count"]) total = old_count + count delta = mean - item["sum"] / old_count item["m2"] = (item["m2"] + (std ** 2) * count + delta ** 2 * old_count * count / total) item["sum"] += mean * count item["min"] = np.minimum(item["min"], minimum) item["max"] = np.maximum(item["max"], maximum) item["count"] = total result: dict[str, dict[str, Any]] = {} for name, item in accumulated.items(): count = int(item["count"]) mean = item["sum"] / count variance = np.maximum(item["m2"] / count, 0.0) values = { "min": item["min"], "max": item["max"], "mean": mean, "std": np.sqrt(variance), "count": [count], } normalized_values = {} for key, value in values.items(): if hasattr(value, "ndim"): normalized_values[key] = ( float(value) if value.ndim == 0 else value.tolist() ) else: normalized_values[key] = value result[name] = normalized_values return result def _compute_data_stats(project_root: Path) -> dict[str, dict[str, Any]]: """Compute fallback global statistics when old per-episode stats are absent.""" import numpy as np import pandas as pd columns: dict[str, list[Any]] = {} for path in sorted((Path(project_root) / "data").rglob("*.parquet")): try: frame = pd.read_parquet(path) except Exception: continue for name in frame.columns: try: array = np.asarray(frame[name].tolist(), dtype=np.float64) except (TypeError, ValueError): continue if array.ndim == 0 or array.ndim > 2: continue columns.setdefault(str(name), []).append(array.reshape(len(frame), -1)) result: dict[str, dict[str, Any]] = {} for name, arrays in columns.items(): try: array = np.concatenate(arrays, axis=0) finite = np.where(np.isfinite(array), array, np.nan) if not np.isfinite(finite).any(): continue stats = { "min": np.nanmin(finite, axis=0), "max": np.nanmax(finite, axis=0), "mean": np.nanmean(finite, axis=0), "std": np.nanstd(finite, axis=0), "count": [int(np.isfinite(finite).sum())], } normalized_stats = {} for key, value in stats.items(): if hasattr(value, "size"): normalized_stats[key] = ( float(value.reshape(-1)[0]) if value.size == 1 else value.tolist() ) else: normalized_stats[key] = value result[name] = normalized_stats except (TypeError, ValueError): continue return result def write_project_stats(project_root: Path, rows: Iterable[dict[str, Any]]) -> None: rows = [dict(row) for row in rows] stats = _aggregate_episode_stats(rows) if not stats: stats = _compute_data_stats(project_root) _write_json(Path(project_root) / "meta" / "stats.json", stats) def _write_jsonl(path: Path, rows: Iterable[dict[str, Any]]) -> None: path.parent.mkdir(parents=True, exist_ok=True) path.write_text( "".join(json.dumps(row, ensure_ascii=False) + "\n" for row in rows), encoding="utf-8", ) def _episode_stats_frame_count(row: dict[str, Any]) -> int | None: """Return the frame count advertised by one v2.1 stats row.""" stats = row.get("stats") if not isinstance(stats, dict): return None frame_stats = stats.get("frame_index") if not isinstance(frame_stats, dict): return None count = frame_stats.get("count") if isinstance(count, list) and count: try: return int(count[0]) except (TypeError, ValueError): return None return None def _repair_episode_stats(stats_rows: list[dict[str, Any]], episode_rows: list[dict[str, Any]]) -> list[dict[str, Any]]: """Align v2.1 per-episode statistics with the canonical episode indexes. Older migrations occasionally preserved the old ``episode_index`` in a stats row, or omitted a row altogether. The frame count is a stable second key, so use it to recover a unique match and create an empty stats row when the source did not provide statistics for an episode. """ candidates = [dict(row) for row in stats_rows if isinstance(row, dict)] used: set[int] = set() repaired: list[dict[str, Any]] = [] ordered_episodes = sorted( episode_rows, key=lambda row: int(row.get("episode_index", 10**9)), ) for episode in ordered_episodes: target_index = int(episode["episode_index"]) expected_length = int(episode.get("length") or 0) chosen_index: int | None = None # Prefer an already-correct row when its frame count agrees. for index, candidate in enumerate(candidates): if index in used or _episode_index(candidate.get("episode_index")) != target_index: continue count = _episode_stats_frame_count(candidate) if count is None or count == expected_length: chosen_index = index break # If the index is stale, recover by a unique frame-count match. if chosen_index is None and expected_length: matches = [ index for index, candidate in enumerate(candidates) if index not in used and _episode_stats_frame_count(candidate) == expected_length ] if len(matches) == 1: chosen_index = matches[0] if chosen_index is None: repaired.append({"episode_index": target_index, "stats": {}}) continue used.add(chosen_index) candidate = candidates[chosen_index] candidate["episode_index"] = target_index stats = candidate.get("stats") if isinstance(stats, dict) and isinstance(stats.get("episode_index"), dict): index_stats = stats["episode_index"] for key in ("min", "max", "mean"): if key in index_stats: index_stats[key] = float(target_index) repaired.append(candidate) return repaired def _link_or_copy(source: Path, target: Path) -> None: target.parent.mkdir(parents=True, exist_ok=True) if target.exists(): if target.is_file() and source.is_file() and target.stat().st_size == source.stat().st_size: return target.unlink() try: os.link(source, target) except OSError: shutil.copy2(source, target) def clone_tree(source: Path, target: Path) -> None: """Clone a tree with hard links when possible. Hard links keep the migration recoverable without temporarily doubling a large depth-PNG dataset. Filesystems without hard-link support fall back to normal copies. """ source = Path(source) target = Path(target) if not source.is_dir(): return for path in source.rglob("*"): rel = path.relative_to(source) dest = target / rel if path.is_dir(): dest.mkdir(parents=True, exist_ok=True) elif path.is_file(): _link_or_copy(path, dest) def _episode_index(value: Any) -> int | None: try: number = int(value) return number if number >= 0 else None except (TypeError, ValueError): return None def episode_index_from_name(name: str) -> int | None: match = re.search(r"_(\d{6})$", str(name or "")) return int(match.group(1)) if match else None # ── Project index cache ─────────────────────────────── # Every _canonical_episode_data_files() call reads the complete episode # metadata parquet set (52 episodes = 52 remote reads) plus two project-wide # directory walks — measured at 1.4s warm, and the review page issues ~11 # video requests per episode open, so the same index was rebuilt a dozen # times per click. Cache the parsed rows and the file index against a cheap # fingerprint of the episode index files. _PROJECT_CACHE_LOCK = threading.RLock() _PROJECT_CACHE: dict[str, dict[str, Any]] = {} # Computing the fingerprint still stats every episode index file, so a short # TTL skips even that inside one page load. Structural changes are announced # explicitly through invalidate_project_cache(). _PROJECT_CACHE_TTL = 60.0 _PROJECT_SCAN_LOCK = threading.Lock() def _project_source_revision(project_root: Path) -> tuple: """Fingerprint the episode index files (name, size, mtime). The index files are written by whoever writes the data (upload, workflow merge), so a change there means the project changed. ``data/`` and ``videos/`` are deliberately not walked: any addition or removal there is accompanied by an index write. """ parts: list[tuple] = [] try: for path in _episode_parquet_files(project_root): stat = path.stat() parts.append((path.name, stat.st_size, stat.st_mtime_ns)) except OSError: return () return tuple(parts) def _read_project_rows(project_root: Path) -> list[dict[str, Any]]: """Uncached: one parquet read per episode, sorted by episode index.""" rows: list[dict[str, Any]] = [] for path in _episode_parquet_files(project_root): try: import pyarrow.parquet as pq values = pq.read_table(path).to_pylist() except (ImportError, OSError, ValueError): values = [] rows.extend(value for value in values if isinstance(value, dict)) rows.sort(key=lambda row: int(row.get("episode_index", 10**9))) return rows def _project_cache_entry(project_root: Path) -> dict[str, Any]: """Return the cached ``{rows, index}`` for a project, rebuilding if stale.""" root = Path(project_root) key = str(root) now = time.monotonic() with _PROJECT_CACHE_LOCK: entry = _PROJECT_CACHE.get(key) if entry is not None and now - entry["checked_at"] < _PROJECT_CACHE_TTL: return entry # The fingerprint touches the remote mount, so keep it outside the lock. revision = _project_source_revision(root) with _PROJECT_CACHE_LOCK: entry = _PROJECT_CACHE.get(key) if entry is not None and entry["revision"] == revision: entry["checked_at"] = now return entry # Single-flight: a page load fires its requests in parallel, and they all # miss at once. One rebuild serves the whole burst. with _PROJECT_SCAN_LOCK: now = time.monotonic() with _PROJECT_CACHE_LOCK: entry = _PROJECT_CACHE.get(key) if entry is not None and ( entry["revision"] == revision or now - entry["checked_at"] < _PROJECT_CACHE_TTL): return entry rows = _read_project_rows(root) index = _build_project_file_index(root, rows) entry = { "revision": revision, "checked_at": time.monotonic(), "rows": rows, "index": index, } with _PROJECT_CACHE_LOCK: _PROJECT_CACHE[key] = entry return entry def invalidate_project_cache(project_root: Path | None = None) -> None: """Drop cached project indexes after a structural change. Called from ``localstore.invalidate_session_cache`` so an upload, delete or rename becomes visible on the next read instead of after the TTL. """ with _PROJECT_CACHE_LOCK: if project_root is None: _PROJECT_CACHE.clear() else: _PROJECT_CACHE.pop(str(Path(project_root)), None) def project_episode_rows(project_root: Path) -> list[dict[str, Any]]: # Callers may assume they own the rows, so never hand out the cache itself. return copy.deepcopy(_project_cache_entry(project_root)["rows"]) def _match_episode_row(rows: list[dict[str, Any]], episode_id: str) -> dict[str, Any] | None: """Match an episode row by canonical id first, then trailing index. Same semantics as :func:`episode_row` but over a caller-supplied row list, so a second parquet read is avoided when rows are already loaded. """ wanted = str(episode_id) for row in rows: if str(row.get("episode_id") or row.get("source_batch") or "") == wanted: return row index = episode_index_from_name(wanted) if index is not None: for row in rows: if _episode_index(row.get("episode_index")) == index: return row return None def episode_row(project_root: Path, episode_id: str) -> dict[str, Any] | None: return _match_episode_row(project_episode_rows(project_root), episode_id) def _episode_id_for_row(row: dict[str, Any], project_name: str) -> str: return str( row.get("episode_id") or row.get("source_batch") or f"{project_name}_{int(row.get('episode_index', 0)):06d}" ) def _chunk_index(value: Any) -> int: try: number = int(value) return max(0, number) except (TypeError, ValueError): return 0 def episode_chunk_for_index(episode_index: int, chunks_size: int = PROJECT_CHUNK_SIZE) -> int: """Map the 0-based episode number to a LeRobot v2.1 chunk. Episodes 000000--000999 are stored in ``chunk-000``; 001000 starts ``chunk-001``. The result is based on the stable dataset index, never on upload order or on the number of legacy directories found. """ size = max(1, int(chunks_size or PROJECT_CHUNK_SIZE)) return max(0, int(episode_index) // size) def _episode_number_from_path(path: Path) -> int | None: match = _EPISODE_RE.search(path.stem) return int(match.group(1)) if match else None def _chunk_number_from_path(path: Path) -> int: for part in path.parts: match = _CHUNK_RE.fullmatch(part) if match: return int(match.group(1)) return 0 def _build_project_file_index( project_root: Path, rows: list[dict[str, Any]] | None = None ) -> dict[int, dict[str, Any]]: """Uncached: one pass over ``data/``, ``videos/`` and the episode rows.""" root = Path(project_root) index: dict[int, dict[str, Any]] = {} data_root = root / "data" if data_root.is_dir(): for path in data_root.rglob("*.parquet"): if not path.is_file(): continue number = _episode_number_from_path(path) if number is None: continue index.setdefault(number, {"data": []})["data"].append(path) for source, path in iter_video_streams(root / "videos"): number = _episode_number_from_path(path) if number is None: continue index.setdefault(number, {"data": []}).setdefault( "videos", []).append((source, path)) if rows is None: rows = _read_project_rows(root) for row in rows: number = _episode_index(row.get("episode_index")) if number is None: continue # ``episode_files`` resolves a duplicate episode number to the first # row in index order; ``setdefault`` keeps that same winner. index.setdefault(number, {"data": []}).setdefault("row", row) return index def project_episode_file_index(project_root: Path) -> dict[int, dict[str, Any]]: """Index every episode's source files with a single pass over the project. :func:`episode_files` re-walks ``data/``, ``videos/`` and the whole episode index for each episode it is asked about, so scanning a project costs ``(1 + episodes) x index files`` parquet reads on the (possibly remote) storage mount. This entry point keeps that contract while serving repeat callers from the revision-checked project cache. Keys are episode numbers; values carry the same ``data`` / ``videos`` / ``row`` payload :func:`episode_files` reports, and are ordered exactly as a per-episode walk would have produced them. """ return copy.deepcopy(_project_cache_entry(Path(project_root))["index"]) def episode_files_from_index(project_root: Path, episode_index: int, index: dict[int, dict[str, Any]]) -> dict[str, Any]: """Look one episode up in a prebuilt :func:`project_episode_file_index`. Returns the same shape as :func:`episode_files`; the caller owns the index lifetime and is responsible for rebuilding it when the project changes. """ root = Path(project_root) entry = index.get(episode_index) or {} row = entry.get("row") or {} relative_meta: list[Path] = [Path("meta/info.json"), Path("meta/tasks.json")] return { "episode_index": episode_index, "episode_id": _episode_id_for_row(row, root.name), "row": row, "data": list(entry.get("data") or []), "videos": list(entry.get("videos") or []), "meta": [path for path in relative_meta if (root / path).is_file()], } def episode_files(project_root: Path, episode_index: int) -> dict[str, Any]: """Return only one episode's source files from a project dataset. Single-episode callers keep this entry point; it builds a one-shot index, so it now walks the project once instead of three times. """ return episode_files_from_index( project_root, episode_index, project_episode_file_index(project_root)) def _processing_token(value: Any, fallback: str = "result") -> str: text = re.sub(r"[^A-Za-z0-9_.-]+", "_", str(value or "")).strip("._-") return text or fallback def _processing_artifacts(outputs: dict[str, Any]) -> list[tuple[str, str, dict[str, Any]]]: """Flatten the worker output manifest into ``(node, handle, ref)`` rows.""" if not isinstance(outputs, dict): return [] values = outputs.get("artifacts", outputs) if not isinstance(values, dict): return [] result: list[tuple[str, str, dict[str, Any]]] = [] for node_id, handles in values.items(): if not isinstance(handles, dict): continue for handle, ref in handles.items(): if isinstance(ref, dict): result.append((str(node_id), str(handle), dict(ref))) return result def _processing_json(value: Any, default: Any) -> Any: if isinstance(value, str): try: return json.loads(value) except (TypeError, ValueError): return default return value if value is not None else default def _processing_ref_path(extracted_root: Path, ref: dict[str, Any]) -> Path | None: """Resolve a worker reference without allowing it to escape the result zip.""" raw = str(ref.get("path") or "").strip() if not raw: return None root = Path(extracted_root).resolve() candidate = (root / raw).resolve() try: candidate.relative_to(root) except ValueError: return None return candidate if candidate.is_file() else None def _atomic_copy(source: Path, target: Path) -> None: target.parent.mkdir(parents=True, exist_ok=True) temporary = target.with_name(f".{target.name}.tmp-{os.getpid()}") try: shutil.copy2(source, temporary) os.replace(temporary, target) finally: temporary.unlink(missing_ok=True) def _atomic_write_parquet(frame: Any, target: Path) -> None: target.parent.mkdir(parents=True, exist_ok=True) temporary = target.with_name(f".{target.name}.tmp-{os.getpid()}") try: frame.to_parquet(temporary, index=False) os.replace(temporary, target) finally: temporary.unlink(missing_ok=True) def _merge_processing_frame(data_frame: Any, result_frame: Any, column_names: list[str]) -> Any: """Merge a processing parquet onto one canonical episode by frame index.""" import pandas as pd if "frame_index" not in data_frame.columns: raise ValueError("canonical episode has no frame_index column") if "frame_index" not in result_frame.columns: raise ValueError("processing result has no frame_index column") result = data_frame.copy() result_index = result["frame_index"] source = result_frame.drop_duplicates("frame_index").set_index("frame_index") lookup = source.reindex(result_index.tolist()) for old_name, new_name in column_names: if old_name not in source.columns: continue # Keep the canonical row order and permit sparse detectors: missing # result frames become null rather than shifting another frame. values = lookup[old_name].reset_index(drop=True) result[new_name] = values return result def _publish_export_artifact(project_root: Path, run_id: str, ref_path: Path, ref: dict[str, Any]) -> Path: """Keep export products outside the three-directory dataset contract.""" export_root = (Path(project_root).parent.parent / "state" / "exports" / _processing_token(run_id, "run")) metadata = ref.get("metadata") or {} relative_root = str(metadata.get("root") or "").strip() if relative_root and relative_root != ".": candidate = ref_path.parent # The module stores ``root`` relative to its node output directory. # Walk up until the declared dataset root is visible. for parent in [ref_path.parent, *ref_path.parents]: if (parent / "meta" / "info.json").is_file(): candidate = parent break elif (ref_path.parent / "data").is_dir() and (ref_path.parent / "meta").is_dir(): candidate = ref_path.parent else: candidate = ref_path if candidate.is_dir(): destination = export_root / "dataset" if destination.exists(): shutil.rmtree(destination) clone_tree(candidate, destination) return destination destination = export_root / ref_path.name _atomic_copy(candidate, destination) return destination def publish_processing_result( project_root: Path, episode_id: str, run_id: str, extracted_root: Path, outputs: dict[str, Any], node_states: dict[str, Any] | None = None, completed_at: str | None = None, ) -> dict[str, Any]: """Merge one completed worker result into the canonical episode. Processing parquet columns are written to ``data/.../episode_*.parquet``; skeleton/preview videos are deliberately not persisted; the episode-level run summary is written to its row in ``meta/episodes/.../episode_*.parquet``. No project-local processing directory or processing manifest is created. """ import pandas as pd root = Path(project_root) row = episode_row(root, str(episode_id)) if row is None: raise RuntimeError(f"Cannot publish processing result: episode not found: {episode_id}") episode_index = int(row.get("episode_index", 0)) chunk_index = episode_chunk_for_index(episode_index) data_path = (root / "data" / f"chunk-{chunk_index:03d}" / f"episode_{episode_index:06d}.parquet") meta_path = (root / "meta" / "episodes" / f"chunk-{chunk_index:03d}" / f"episode_{episode_index:06d}.parquet") if not data_path.is_file() or not meta_path.is_file(): raise RuntimeError(f"Canonical episode files are incomplete: {episode_id}") data_frame = pd.read_parquet(data_path) old_columns = _processing_json(row.get("processing_columns"), []) if not isinstance(old_columns, list): old_columns = [] data_frame = data_frame.drop( columns=[str(name) for name in old_columns if str(name) in data_frame.columns], errors="ignore", ) records = _processing_artifacts(outputs) tabular_records = [item for item in records if str(item[2].get("kind") or "") in {"hand_keypoints", "hand_3d", "glove_sensor", "umi_slam_action"}] # A reprocess is authoritative for its hand-3D sources. Remove any # source-qualified hand-3D columns left by an earlier partial publish # (including legacy ``*_1`` collision columns) before merging the fresh # left/right results. Other processing kinds remain untouched. if any(str(ref.get("kind") or "") == "hand_3d" for _node, _handle, ref in records): stale_hand3d = [str(name) for name in data_frame.columns if str(name).startswith("processing.hand_3d.")] if stale_hand3d: data_frame = data_frame.drop(columns=stale_hand3d, errors="ignore") # Derived action replaces the collector's placeholder column instead of # being renamed to ``processing..*.action``. The UMI collector # writes an all-zero ``action`` that the LeRobot exporter correctly drops # as a constant/zero-information column, so without this drop the fresh # action would land under a name no exporter reads. Only drop it when a # umi_slam_action result is actually being published for this episode. if any(str(ref.get("kind") or "") == "umi_slam_action" for _node, _handle, ref in records): # 本模块产出的列全部按原名覆盖(重跑幂等):action 之外还有 # observation.state,它同样不该因为二次合并被改名成 processing.*。 stale_action = [str(name) for name in data_frame.columns if str(name) in {"action", "observation.state"} or str(name).startswith("processing.umi_slam_action.")] if stale_action: data_frame = data_frame.drop(columns=stale_action, errors="ignore") assigned_columns: set[str] = set(data_frame.columns) merged_columns: list[str] = [] published = {"artifacts": {}} merged_hand3d_sources: set[str] = set() merged_action_files: set[str] = set() for node_id, handle, ref in records: source_path = _processing_ref_path(Path(extracted_root), ref) if source_path is None: # Input passthrough references point into the worker input cache and # are not part of result_zip. They remain in the run state but # are not treated as newly published processing files. continue kind = str(ref.get("kind") or "") metadata = dict(ref.get("metadata") or {}) published_ref = dict(ref) source = _processing_token(ref.get("source_key") or metadata.get("source_key") or handle, kind) # A review/review-export node forwards its upstream action ref, so the # same derived action arrives twice (once from the producing node, # once forwarded). Merging both would add a duplicate # ``processing..*.action`` column holding identical values. # Deduplicate on the artifact file, not the node id: two genuinely # separate umi_slam_action nodes produce different files and must both # merge. if kind == "umi_slam_action": action_file = str(source_path.resolve()) if action_file in merged_action_files: published_ref["path"] = str(data_path.relative_to(root)).replace("\\", "/") published_ref["metadata"] = { **metadata, "merged_into": published_ref["path"], "deduplicated": True, "merged_columns": [], } published["artifacts"].setdefault(node_id, {})[handle] = published_ref continue merged_action_files.add(action_file) # Review/export nodes pass the node-36 hand-3D refs through again. # They are the same source data, not a second measurement. Merge each # source once and keep the passthrough artifact only as an audit link. if kind == "hand_3d" and source in merged_hand3d_sources: published_ref["path"] = str(data_path.relative_to(root)).replace("\\", "/") published_ref["metadata"] = { **metadata, "merged_into": published_ref["path"], "deduplicated": True, "merged_columns": [], } published["artifacts"].setdefault(node_id, {})[handle] = published_ref continue if kind == "hand_3d": merged_hand3d_sources.add(source) if kind == "hand_3d" and not ( bool(metadata.get("metric_3d_available")) or str(metadata.get("unit") or "") in {"camera_meters", "meters"} or str(metadata.get("mode") or "").lower() in {"depth_hand_3d", "d435_depth_lifted", "depth_lifted"}): # RGB_TO_3D/other RGB-only estimators may create a temporary # spatial preview artifact. It is not real depth and must not be # persisted in the LeRobot source data. published_ref["path"] = None published_ref["metadata"] = {**metadata, "preview_only": True, "storage": "browser_overlay"} published["artifacts"].setdefault(node_id, {})[handle] = published_ref continue if kind in {"hand_keypoints", "hand_3d", "glove_sensor", "umi_slam_action"}: result_frame = pd.read_parquet(source_path) raw_columns = [str(name) for name in result_frame.columns if name not in {"frame_index", "episode_index"}] # A propagated canonical reference can already contain several # hand-3D source namespaces. When that happens, publish only the # namespace belonging to this artifact ref and strip it back to # the raw fields before applying the canonical name below. source_columns: list[tuple[str, str]] = [] if kind == "hand_3d": prefix = f"processing.hand_3d.{source}." namespaced = [name for name in raw_columns if name.startswith(prefix)] raw_names = {name[len(prefix):] for name in namespaced} source_columns = [ (name, name[len(prefix):]) for name in namespaced if not (name[len(prefix):].endswith("_1") and name[len(prefix):-2] in raw_names) ] if not source_columns: source_columns = [(name, name) for name in raw_columns] column_names: list[tuple[str, str]] = [] for source_name, raw_name in source_columns: # Hand-3D is produced once per RGB view. Always retain the # source namespace, even for the first result, otherwise the # first stereo view stays as plain ``hand_0_*`` columns and # the second view is namespaced only because of a collision. # That makes it impossible for the review UI to choose the # view with more valid landmarks deterministically. public_name = (f"processing.{kind}.{source}.{raw_name}" if kind == "hand_3d" else raw_name) if public_name in assigned_columns: public_name = f"processing.{kind}.{source}.{raw_name}" while public_name in assigned_columns: public_name = f"{public_name}_1" assigned_columns.add(public_name) column_names.append((source_name, public_name)) merged_columns.append(public_name) data_frame = _merge_processing_frame(data_frame, result_frame, column_names) published_ref["path"] = str(data_path.relative_to(root)).replace("\\", "/") published_ref["metadata"] = { **metadata, "merged_into": published_ref["path"], "merged_columns": [name for _raw, name in column_names], } elif kind == "video" and (metadata.get("skeleton") or "skeleton" in handle.lower() or metadata.get("render_video")): # Skeleton videos are preview-only. The browser overlays the # merged keypoints on the original RGB stream, so persisting a # second rendered video would duplicate large source files. published_ref["path"] = None published_ref["metadata"] = {**metadata, "preview_only": True, "storage": "browser_overlay"} elif kind in {"dataset", "hdf5", "export"}: target = _publish_export_artifact(root, run_id, source_path, ref) published_ref["path"] = str(target) published_ref["metadata"] = {**metadata, "published_path": str(target)} else: # Unknown artifacts are not copied into the dataset tree. Keeping # their manifest entry is useful for diagnostics and preserves # forward compatibility with new modules. published_ref["path"] = str(ref.get("path") or "") published["artifacts"].setdefault(node_id, {})[handle] = published_ref _atomic_write_parquet(data_frame, data_path) row = dict(row) row.update({ "processing_run_id": str(run_id), "processing_status": "completed", "processing_completed_at": completed_at or datetime.now(timezone.utc).isoformat(), "processing_node_states": json.dumps(node_states or {}, ensure_ascii=False, separators=(",", ":")), "processing_outputs": json.dumps({ "artifacts": published.get("artifacts") or {}, }, ensure_ascii=False, separators=(",", ":")), "processing_columns": json.dumps(merged_columns, ensure_ascii=False, separators=(",", ":")), }) rows = [row if int(item.get("episode_index", -1)) == episode_index else item for item in project_episode_rows(root)] write_project_episode_index(root, rows) return published def _task_rows(root: Path, fallback: str) -> list[dict[str, Any]]: rows = _read_tasks_json(root / "meta" / "tasks.json") if not rows: rows = _read_jsonl(root / "meta" / "tasks.jsonl") if rows: return rows return [{"task_index": 0, "task": fallback or "default_recording"}] def _task_text(row: dict[str, Any], fallback: str) -> str: tasks = row.get("tasks") if isinstance(tasks, list) and tasks: return str(tasks[0]) return str(row.get("task") or fallback or "default_recording") def _merge_unique_dicts(existing: list[dict[str, Any]], additions: Any, keys: tuple[str, ...]) -> list[dict[str, Any]]: result = [dict(item) for item in existing if isinstance(item, dict)] values = additions if isinstance(additions, list) else [] seen = {tuple(str(item.get(key) or "") for key in keys) for item in result} for item in values: if not isinstance(item, dict): continue marker = tuple(str(item.get(key) or "") for key in keys) if marker in seen: continue seen.add(marker) result.append(dict(item)) return result def _project_data_contract(project_root: Path) -> tuple[set[str], set[str], set[str]]: """Return the data columns, real video sources and semantic device slots. ``meta/info.json`` is an aggregate and can outlive a camera stream after a re-upload. The canonical files are authoritative for the feature list. This helper deliberately treats unreadable Parquet as empty instead of preserving stale declarations that make the UI request missing media. """ root = Path(project_root) data_columns: set[str] = set() data_root = root / "data" if data_root.is_dir(): try: import pyarrow.parquet as pq for path in data_root.rglob("*.parquet"): try: data_columns.update(str(name) for name in pq.read_schema(path).names) except (OSError, ValueError): continue except ImportError: pass video_sources = { str(source) for source, _path in iter_video_streams(root / "videos") if str(source).strip() } semantic_slots = set(video_sources) if "observation.slam_pose" in data_columns: semantic_slots.add("gripper_pose") if "observation.gripper_left_force" in data_columns: semantic_slots.add("gripper_force_left") if "observation.gripper_right_force" in data_columns: semantic_slots.add("gripper_force_right") if "observation.gripper_left_force_matrix" in data_columns: semantic_slots.add("gripper_force_matrix_left") if "observation.gripper_right_force_matrix" in data_columns: semantic_slots.add("gripper_force_matrix_right") return data_columns, video_sources, semantic_slots def _feature_is_present(key: str, feature: Any, data_columns: set[str], video_sources: set[str]) -> bool: if not isinstance(feature, dict): return key in data_columns if feature.get("dtype") == "video" or key.startswith("observation.images."): return key.removeprefix("observation.images.") in video_sources return key in data_columns def _merge_info(project_root: Path, source_info: dict[str, Any], rows: list[dict[str, Any]], tasks: list[dict[str, Any]]) -> dict[str, Any]: target = _read_json(project_root / "meta" / "info.json", {}) if not isinstance(target, dict): target = {} target = normalize_metadata_sources(target) source_info = normalize_metadata_sources(source_info) if not target: target = dict(source_info) for key in ("project_id", "project_name"): if source_info.get(key) and not target.get(key): target[key] = source_info[key] features = dict(target.get("features") or {}) features.update(dict(source_info.get("features") or {})) data_columns, video_sources, semantic_slots = _project_data_contract(project_root) features = { key: value for key, value in features.items() if _feature_is_present(key, value, data_columns, video_sources) } target.update({ "format": "lerobot_v2.1", "codebase_version": "v2.1", "total_episodes": len(rows), "total_frames": sum(int(row.get("length") or 0) for row in rows), "total_tasks": len(tasks), "total_chunks": len({episode_chunk_for_index( int(row.get("episode_index"))) for row in rows if _episode_index(row.get("episode_index")) is not None}), "chunks_size": PROJECT_CHUNK_SIZE, "data_path": "data/chunk-{episode_chunk:03d}/episode_{episode_index:06d}.parquet", "video_path": "videos/{video_key}/chunk-{episode_chunk:03d}/episode_{episode_index:06d}.{ext}", "features": features, "extensions": { **(target.get("extensions") or {}), "episodes_file": "meta/episodes/chunk-{episode_chunk:03d}/episode_{episode_index:06d}.parquet", "stats_file": "meta/stats.json", "tasks_file": "meta/tasks.json", "project_storage": "sessions//{data,meta,videos}", "processing_storage": "data+meta/episodes+videos", }, }) target["extensions"].pop("calibration_root", None) target["extensions"].pop("episodes_stats_file", None) target["extensions"].pop("episodes_compat_file", None) if source_info.get("fps") and not target.get("fps"): target["fps"] = source_info["fps"] target["devices"] = _merge_unique_dicts( target.get("devices") or [], source_info.get("devices"), ("key",), ) filtered_devices: list[dict[str, Any]] = [] for device in target["devices"]: if not isinstance(device, dict): continue slots = [str(slot) for slot in (device.get("slots") or []) if str(slot) in semantic_slots] if not slots: continue item = dict(device) item["slots"] = slots filtered_devices.append(item) target["devices"] = filtered_devices names = dict(target.get("device_names") or {}) names.update(dict(source_info.get("device_names") or {})) target["device_names"] = { key: value for key, value in names.items() if key in semantic_slots } cameras = dict(target.get("cameras") or {}) cameras.update(dict(source_info.get("cameras") or {})) target["cameras"] = { key: value for key, value in cameras.items() if key in video_sources } sensors = list(dict.fromkeys( [str(value) for value in (target.get("sensors") or [])] + [str(value) for value in (source_info.get("sensors") or [])] )) target["sensors"] = [ sensor for sensor in sensors if any(sensor.casefold() in column.casefold() for column in data_columns) ] # Make the common UMI/SLAM contract machine-readable without guessing the # meaning of the legacy three-value ``gripper_state`` vector. semantic_fields = { "observation.slam_pose": { "names": ["x", "y", "z", "qx", "qy", "qz", "qw"], "units": ["m", "m", "m", "unitless", "unitless", "unitless", "unitless"], "quaternion_order": "xyzw", }, "observation.gripper_left_force": { "names": ["fx", "fy", "fz"], "units": ["sensor_unit", "sensor_unit", "sensor_unit"], }, "observation.gripper_right_force": { "names": ["fx", "fy", "fz"], "units": ["sensor_unit", "sensor_unit", "sensor_unit"], }, "observation.gripper_left_force_matrix": { "names": ["fx", "fy", "fz"], "units": ["mN", "mN", "mN"], "encoding": "row_diff_quantized", }, "observation.gripper_right_force_matrix": { "names": ["fx", "fy", "fz"], "units": ["mN", "mN", "mN"], "encoding": "row_diff_quantized", }, } for key, metadata in semantic_fields.items(): if key in target["features"]: target["features"][key] = { **target["features"][key], **metadata, } return target def _read_episode_sidecars(episode_root: Path, episode_id: str) -> dict[str, Any]: """Inline calibration and collector provenance into the episode row.""" def drop_empty_maps(value: Any) -> Any: if isinstance(value, dict): result = {} for key, child in value.items(): cleaned = drop_empty_maps(child) if cleaned is not None: result[key] = cleaned return result or None if isinstance(value, list): return [cleaned for child in value if (cleaned := drop_empty_maps(child)) is not None] return value result: dict[str, Any] = {} for kind in ("calibration", "collector"): base = Path(episode_root) / "meta" / kind candidate = base / str(episode_id) if not candidate.is_dir(): candidate = base if not candidate.is_dir(): continue values: dict[str, Any] = {} for path in sorted(candidate.rglob("*.json")): value = _read_json(path, None) if value is not None: if isinstance(value, dict): value = normalize_metadata_sources(value) value = drop_empty_maps(value) if value is not None: values[str(path.relative_to(candidate))] = value if values: result[kind] = values return result def _copy_data_with_episode_index(source: Path, target: Path, episode_index: int) -> None: """Copy a parquet and normalize its internal episode index if present.""" try: import pandas as pd frame_table = pd.read_parquet(source) if "episode_index" not in frame_table.columns: _link_or_copy(source, target) return values = frame_table["episode_index"].dropna().unique().tolist() if len(values) == 1 and int(values[0]) == int(episode_index): _link_or_copy(source, target) return target.parent.mkdir(parents=True, exist_ok=True) frame_table["episode_index"] = int(episode_index) frame_table.to_parquet(target, index=False) except Exception as exc: raise RuntimeError( f"Failed to normalize episode_index in {source.name}: {exc}" ) from exc def merge_normalized_episode(project_root: Path, episode_root: Path, episode_id: str, chunk_index: int, replace: bool = False, target_episode_index: int | None = None) -> dict[str, Any]: """Merge one normalized episode into a project-level dataset root.""" project_root = Path(project_root) episode_root = Path(episode_root) project_root.mkdir(parents=True, exist_ok=True) (project_root / "meta").mkdir(parents=True, exist_ok=True) source_info = _read_json(episode_root / "meta" / "info.json", {}) source_rows = project_episode_rows(episode_root) source_row = source_rows[0] if source_rows else {} source_episode_index = _episode_index(source_row.get("episode_index")) if source_episode_index is None: source_episode_index = episode_index_from_name(episode_id) episode_index = (int(target_episode_index) if target_episode_index is not None else source_episode_index) if episode_index is None: raise RuntimeError(f"Cannot determine episode_index for {episode_id}") # Re-apply the canonical rule here even when a migration caller supplied # a temporary chunk number based on directory order. chunk_index = episode_chunk_for_index(episode_index) existing_rows = project_episode_rows(project_root) old = next((row for row in existing_rows if _episode_index(row.get("episode_index")) == episode_index or str(row.get("episode_id") or "") == str(episode_id)), None) if old is not None and not replace: raise RuntimeError(f"Episode already exists: {episode_id}") if old is not None: old_index = existing_rows.index(old) existing_rows.pop(old_index) old_chunk = _chunk_index(old.get("data/chunk_index")) old_data = project_root / "data" / f"chunk-{old_chunk:03d}" / f"episode_{episode_index:06d}.parquet" old_data.unlink(missing_ok=True) for path in (project_root / "videos").rglob(f"episode_{episode_index:06d}.*"): path.unlink(missing_ok=True) # Sidecar metadata is embedded into the episode row below, so a # replacement cannot leave an obsolete per-episode metadata folder. source_data = [path for path in (episode_root / "data").rglob("*.parquet") if path.is_file()] if (episode_root / "data").is_dir() else [] if not source_data: raise RuntimeError(f"Normalized episode has no data parquet: {episode_id}") source_data.sort() data_target = project_root / "data" / f"chunk-{chunk_index:03d}" / f"episode_{episode_index:06d}.parquet" _copy_data_with_episode_index(source_data[0], data_target, episode_index) source_videos = iter_video_streams(episode_root / "videos") for source, video_path in source_videos: destination = (project_root / "videos" / f"observation.images.{source}" / f"chunk-{chunk_index:03d}" / f"episode_{episode_index:06d}{video_path.suffix.lower()}") _link_or_copy(video_path, destination) source_task_rows = _task_rows(episode_root, str(source_info.get("task_name") or episode_id)) target_tasks = _task_rows(project_root, str(source_info.get("task_name") or episode_id)) task_by_text = {_task_text(row, ""): int(row.get("task_index", index)) for index, row in enumerate(target_tasks)} next_task = max(task_by_text.values(), default=-1) + 1 task_text = _task_text(source_row, str(source_info.get("task_name") or episode_id)) if task_text not in task_by_text: task_by_text[task_text] = next_task next_task += 1 target_tasks = [{"task_index": index, "task": text} for text, index in sorted(task_by_text.items(), key=lambda item: item[1])] row = dict(source_row) if not isinstance(row.get("stats"), dict): source_stats = _read_jsonl(episode_root / "meta" / "episodes_stats.jsonl") source_stats_row = next( (item for item in source_stats if _episode_index(item.get("episode_index")) == source_episode_index), None, ) if isinstance(source_stats_row, dict): row["stats"] = source_stats_row.get("stats") or {} row.update(_read_episode_sidecars(episode_root, str(episode_id))) row.update({ "episode_index": episode_index, "episode_id": str(episode_id), "source_batch": str(episode_id), "data/chunk_index": int(chunk_index), "data/file_index": int(episode_index), "task_index": int(task_by_text[task_text]), "tasks": [task_text], }) for key in list(row): if key.endswith("/chunk_index"): row[key] = int(chunk_index) elif key.endswith("/file_index"): row[key] = int(episode_index) previous_frames = sum(int(item.get("length") or 0) for item in existing_rows) length = int(row.get("length") or 0) row["dataset_from_index"] = previous_frames row["dataset_to_index"] = previous_frames + length existing_rows.append(row) existing_rows.sort(key=lambda item: int(item.get("episode_index", 10**9))) write_project_episode_index(project_root, existing_rows) write_project_stats(project_root, existing_rows) _write_tasks_json(project_root, target_tasks) info = _merge_info(project_root, source_info, existing_rows, target_tasks) _write_json(project_root / "meta" / "info.json", info) return { "episode_id": str(episode_id), "episode_index": episode_index, "chunk_index": int(chunk_index), "frame_count": length, "video_sources": sorted(source for source, _path in source_videos), "path": str(project_root), } # 人工/脚本留下的备份与临时文件 —— 不是删除或合并操作产生的，不该算到校验头上。 # # ★ 为什么必须放过它们：``delete_episode`` 在**删完文件之后**才跑本校验，校验失败就 # 抛 RuntimeError → 500，而「清空回收站」是循环处理的，一个项目失败整轮中止。 # 实测踩过：两个 D435 项目的 meta/ 里各有一个 ``info.json.bak-20260908-144550`` # （9-08 留下的），于是**每次清空都 500、什么都删不掉**，而前端没有错误分支， # 用户看到的是"点了没反应"。一个备份文件不该有这种杀伤力。 _IGNORED_ENTRY_SUFFIXES = (".bak", ".orig", ".tmp", ".save", "~") def _is_ignorable_entry(name: str) -> bool: """备份/临时文件，或隐藏文件 —— 校验一律不看。""" if name.startswith("."): return True return name.endswith(_IGNORED_ENTRY_SUFFIXES) or ".bak-" in name def verify_project_dataset(project_root: Path) -> dict[str, Any]: root = Path(project_root) rows = project_episode_rows(root) errors: list[str] = [] unexpected = sorted( path.name for path in root.iterdir() if path.name not in _CANONICAL_ROOTS and not _is_ignorable_entry(path.name) ) if root.is_dir() else [] errors.extend(f"unexpected project root entry: {name}" for name in unexpected) expected_meta = {"info.json", "stats.json", "tasks.json", "episodes"} meta_root = root / "meta" if meta_root.is_dir(): unexpected_meta = sorted( path.name for path in meta_root.iterdir() if path.name not in expected_meta and not _is_ignorable_entry(path.name) ) errors.extend(f"unexpected meta entry: {name}" for name in unexpected_meta) for row in rows: index = _episode_index(row.get("episode_index")) if index is None: errors.append("episode row without episode_index") continue chunk = _chunk_index(row.get("data/chunk_index")) expected_chunk = episode_chunk_for_index(index) if chunk != expected_chunk: errors.append( f"wrong chunk for episode {index:06d}: {chunk:03d} != {expected_chunk:03d}" ) data_path = root / "data" / f"chunk-{chunk:03d}" / f"episode_{index:06d}.parquet" if not data_path.is_file(): errors.append(f"missing data: {data_path.relative_to(root)}") if not any(_episode_number_from_path(path) == index for path in (root / "videos").rglob("*") if path.is_file()): errors.append(f"missing video for episode {index:06d}") return { "passed": not errors and bool(rows), "episodes": len(rows), "data_files": sum(1 for path in (root / "data").rglob("*.parquet") if path.is_file()) if (root / "data").is_dir() else 0, "video_files": sum(1 for path in (root / "videos").rglob("*") if path.is_file()) if (root / "videos").is_dir() else 0, "errors": errors, } def _prune_empty_artifact_dirs(project_root: Path) -> None: """Remove emptied chunk/camera directories under data/ and videos/. Never removes ``data``/``videos`` themselves; the project must remain a valid skeleton for future uploads. """ root = Path(project_root) for base in ("data", "videos"): base_dir = root / base if not base_dir.is_dir(): continue for path in sorted(base_dir.rglob("*"), reverse=True): if path.is_dir() and not any(path.iterdir()): path.rmdir() def _reindex_dataset_ranges(rows: list[dict[str, Any]]) -> None: """Re-derive contiguous dataset_from_index/dataset_to_index. Official LeRobot v2.1 keeps a dense global frame order. Episode indexes stay stable (files are named after them); only the dataset ranges are compacted after a deletion. """ rows.sort(key=lambda row: int(row.get("episode_index", 10**9))) cursor = 0 for row in rows: length = int(row.get("length") or 0) row["dataset_from_index"] = cursor cursor += length row["dataset_to_index"] = cursor def _reconcile_task_rows(rows: list[dict[str, Any]], existing_tasks: list[dict[str, Any]]) -> list[dict[str, Any]]: """Rebuild the tasks.json index from surviving rows. Keeps the old task_index↔text mapping for texts that still appear, so unrelated tasks never shift; drops tasks no surviving episode references (an empty project yields an empty tasks.json), and fixes each row's ``task_index`` in place before the index rewrite. """ surviving_texts = {_task_text(row, "default_recording") for row in rows} text_index: dict[str, int] = { str(task["task"]): int(task["task_index"]) for task in existing_tasks if isinstance(task, dict) and task.get("task") and str(task["task"]) in surviving_texts } next_index = max(text_index.values(), default=-1) + 1 for row in rows: text = _task_text(row, "default_recording") if text not in text_index: text_index[text] = next_index next_index += 1 row["task_index"] = text_index[text] return sorted( ({"task_index": index, "task": text} for text, index in text_index.items()), key=lambda item: int(item["task_index"]), ) def delete_project_episode(project_root: Path | str, episode_id: str) -> dict[str, Any]: """Permanently remove one episode and every artifact from a project dataset. Removes its data parquet and all per-source videos, drops its row from the canonical meta/episodes index, recomputes contiguous dataset frame ranges, rebuilds meta/stats.json, meta/tasks.json and meta/info.json from the surviving rows, and prunes emptied chunk directories. Episode indexes stay stable (gaps are valid); only dataset ranges are compacted. Idempotent: any artifact already gone is skipped. Raises LookupError when the id matches neither a row nor an index-shaped name. """ root = Path(project_root) info0 = _read_json(root / "meta" / "info.json", {}) if not isinstance(info0, dict): info0 = {} tasks0 = _read_tasks_json(root / "meta" / "tasks.json") rows = project_episode_rows(root) row0 = _match_episode_row(rows, episode_id) index = _episode_index(row0.get("episode_index")) if row0 is not None else None if index is None: index = episode_index_from_name(episode_id) if index is None: raise LookupError(f"Episode not found in project dataset: {episode_id}") removed_data: list[str] = [] if (root / "data").is_dir(): for path in sorted((root / "data").rglob("*.parquet")): if path.is_file() and _episode_number_from_path(path) == index: path.unlink(missing_ok=True) removed_data.append(str(path)) removed_videos: list[str] = [] if (root / "videos").is_dir(): for path in sorted((root / "videos").rglob(f"episode_{index:06d}.*")): if path.is_file(): path.unlink(missing_ok=True) removed_videos.append(str(path)) _prune_empty_artifact_dirs(root) if row0 is not None: rows.remove(row0) _reindex_dataset_ranges(rows) tasks = _reconcile_task_rows(rows, tasks0) # Same rebuild tail as merge/repack: index → stats → tasks → info. write_project_episode_index(root, rows) write_project_stats(root, rows) _write_tasks_json(root, tasks) info = _merge_info(root, info0, rows, tasks) _write_json(root / "meta" / "info.json", info) verification = verify_project_dataset(root) # With zero surviving rows verify reports passed=False by design # (bool(rows)); an empty project is still a successful deletion. if rows and not verification["passed"]: raise RuntimeError( f"Episode delete verification failed: {verification['errors']}") return { "episode_id": str(episode_id), "episode_index": index, "row_removed": row0 is not None, "removed_data": removed_data, "removed_videos": removed_videos, "remaining": len(rows), "verification": verification, } def repack_project_chunks(project_root: Path, *, keep_backup: bool = True) -> dict[str, Any]: """Rebuild canonical ``data`` and ``videos`` chunk directories. This is used for projects migrated before the 1000-episode rule was enforced. Only the canonical payload and the three supported metadata files are rebuilt. The old active tree is kept as a recoverable backup until the caller explicitly removes it. """ project_root = Path(project_root) if not is_project_dataset(project_root): return {"changed": False, "project": project_root.name, "verification": verify_project_dataset(project_root)} rows = project_episode_rows(project_root) if not rows: return {"changed": False, "project": project_root.name, "verification": verify_project_dataset(project_root)} storage_root = project_root.parent.parent stamp = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ") temp_root = project_root.parent / f".{project_root.name}.repack-{stamp}" backup_root = (storage_root / ".backups" / "session-migrations" / f"{project_root.name}-repack-{stamp}") shutil.rmtree(temp_root, ignore_errors=True) temp_root.mkdir(parents=True, exist_ok=False) mappings: list[dict[str, Any]] = [] try: for name in _CANONICAL_ROOTS: clone_tree(project_root / name, temp_root / name) # Remove canonical payload files before rebuilding them. Metadata is # reduced to the active contract below; no collector/calibration/depth # sidecar directory survives in the new project root. for path in sorted((temp_root / "data").rglob("*.parquet"), reverse=True): path.unlink(missing_ok=True) for path in sorted((temp_root / "videos").rglob("*"), reverse=True): if path.is_file() and path.suffix.lower() in {".mp4", ".webm", ".mkv", ".avi", ".mov"}: path.unlink(missing_ok=True) for root in (temp_root / "data", temp_root / "videos"): for path in sorted(root.rglob("*"), reverse=True): if path.is_dir() and not any(path.iterdir()): path.rmdir() rebuilt_rows: list[dict[str, Any]] = [] actual_sources: set[str] = set() for new_index, row in enumerate(rows): old_index = _episode_index(row.get("episode_index")) if old_index is None: raise RuntimeError("Cannot repack an episode without episode_index") old_chunk = _chunk_index(row.get("data/chunk_index")) old_data = (project_root / "data" / f"chunk-{old_chunk:03d}" / f"episode_{old_index:06d}.parquet") if not old_data.is_file(): raise RuntimeError(f"Missing data parquet: {old_data}") new_chunk = episode_chunk_for_index(new_index) new_data = (temp_root / "data" / f"chunk-{new_chunk:03d}" / f"episode_{new_index:06d}.parquet") _copy_data_with_episode_index(old_data, new_data, new_index) for video in (project_root / "videos").rglob("*"): if not video.is_file() or video.suffix.lower() not in {".mp4", ".webm", ".mkv", ".avi", ".mov"}: continue if _episode_number_from_path(video) != old_index: continue source = source_key_from_video(video, project_root / "videos") actual_sources.add(source) target = (temp_root / "videos" / f"observation.images.{source}" / f"chunk-{new_chunk:03d}" / f"episode_{new_index:06d}{video.suffix.lower()}") _link_or_copy(video, target) updated = dict(row) updated["episode_index"] = new_index updated["data/chunk_index"] = new_chunk for key in list(updated): if key.endswith("/chunk_index"): updated[key] = new_chunk elif key.endswith("/file_index"): updated[key] = new_index rebuilt_rows.append(updated) mappings.append({"old_episode_index": old_index, "episode_index": new_index, "old_chunk": old_chunk, "chunk_index": new_chunk}) rebuilt_rows.sort(key=lambda item: int(item.get("episode_index", 10**9))) info = _read_json(temp_root / "meta" / "info.json", {}) if not isinstance(info, dict): info = {} declared_sources = sorted( str(key)[len("observation.images."):] for key in (info.get("features") or {}) if str(key).startswith("observation.images.") ) source_aliases: dict[str, str] = {} if len(declared_sources) == len(actual_sources): source_aliases = dict(zip(declared_sources, sorted(actual_sources))) elif len(actual_sources) == 1: # Older migrations may have left both a stale declared key and # the real on-disk key in info.json. The canonical video folder # is authoritative for the normalized dataset. actual = next(iter(actual_sources)) source_aliases = { source: actual for source in declared_sources if source != actual } if source_aliases: def rename_keys(value: Any, *, feature_keys: bool = False) -> Any: if not isinstance(value, dict): return value result: dict[Any, Any] = {} for key, item in value.items(): raw_key = str(key) prefix = "observation.images." if feature_keys and raw_key.startswith(prefix): source = raw_key[len(prefix):] new_key = prefix + source_aliases.get(source, source) else: new_key = source_aliases.get(raw_key, raw_key) # If both the stale key and the real key are present, # keep the real-key value and drop the duplicate. if key in source_aliases and new_key in value: continue result[new_key] = item return result info["features"] = rename_keys(info.get("features") or {}, feature_keys=True) info["cameras"] = rename_keys(info.get("cameras") or {}) info["device_names"] = rename_keys(info.get("device_names") or {}) info["video_extensions"] = rename_keys(info.get("video_extensions") or {}) for device in info.get("devices") or []: if isinstance(device, dict): device["slots"] = list(dict.fromkeys( source_aliases.get(str(slot), str(slot)) for slot in device.get("slots") or [] )) device["resolution"] = rename_keys(device.get("resolution") or {}) device["fps"] = rename_keys(device.get("fps") or {}) for row in rebuilt_rows: for key in list(row): for old, new in source_aliases.items(): prefix = f"videos/observation.images.{old}/" if key.startswith(prefix): row[key.replace(prefix, f"videos/observation.images.{new}/", 1)] = row.pop(key) break for path in (temp_root / "meta" / "collector").glob("*/metadata.json"): metadata = _read_json(path, {}) if not metadata: continue metadata["cameras"] = rename_keys(metadata.get("cameras") or {}) # The collector may number its metadata from 1 while the # canonical LeRobot files use a 0-based episode index. collector_id = path.parent.name row_for_metadata = next( (item for item in rebuilt_rows if _episode_id_for_row(item, project_root.name) == collector_id), None, ) if row_for_metadata is not None: metadata["episode_index"] = int(row_for_metadata["episode_index"]) metadata["chunk_index"] = int( row_for_metadata.get("data/chunk_index") or 0 ) metadata["file_index"] = int(row_for_metadata["data/file_index"]) for device in metadata.get("devices") or []: if isinstance(device, dict): device["slots"] = [source_aliases.get(str(slot), str(slot)) for slot in device.get("slots") or []] device["resolution"] = rename_keys(device.get("resolution") or {}) device["fps"] = rename_keys(device.get("fps") or {}) _write_json(path, metadata) # A v2.1 reader must never advertise a stream that has no corresponding # file. This handles older migrations where ``features`` retained a # stale ``head_right_rgb_4`` key after the actual folder became # ``head_right_rgb``. actual_sources = sorted(actual_sources) feature_map = info.get("features") or {} if isinstance(feature_map, dict): stream_features: dict[str, Any] = {} for key, value in feature_map.items(): raw_key = str(key) if not raw_key.startswith("observation.images."): stream_features[raw_key] = value continue source = raw_key[len("observation.images."):] source = source_aliases.get(source, source) if source in actual_sources: stream_features[f"observation.images.{source}"] = value info["features"] = stream_features for field in ("cameras", "device_names", "video_extensions"): value = info.get(field) if isinstance(value, dict): info[field] = { source_aliases.get(str(key), str(key)): item for key, item in value.items() if source_aliases.get(str(key), str(key)) in actual_sources } for device in info.get("devices") or []: if isinstance(device, dict): device["slots"] = [ source_aliases.get(str(slot), str(slot)) for slot in device.get("slots") or [] if source_aliases.get(str(slot), str(slot)) in actual_sources ] write_project_episode_index(temp_root, rebuilt_rows) write_project_stats(temp_root, rebuilt_rows) # Replace the cloned info before merging so stale source aliases from # an earlier migration cannot be reintroduced by _merge_info(). _write_json(temp_root / "meta" / "info.json", info) tasks = _task_rows(temp_root, project_root.name) info = _merge_info(temp_root, info, rebuilt_rows, tasks) info["total_chunks"] = len({episode_chunk_for_index( int(row["episode_index"])) for row in rebuilt_rows}) _write_json(temp_root / "meta" / "info.json", info) _write_tasks_json(temp_root, tasks) meta_root = temp_root / "meta" for item in list(meta_root.iterdir()): if item.name in {"info.json", "stats.json", "tasks.json", "episodes"}: continue if item.is_dir(): shutil.rmtree(item) else: item.unlink(missing_ok=True) verification = verify_project_dataset(temp_root) if not verification["passed"]: raise RuntimeError(f"Chunk repack verification failed: {verification['errors']}") backup_root.parent.mkdir(parents=True, exist_ok=True) project_root.rename(backup_root) try: temp_root.rename(project_root) except Exception: if backup_root.exists(): backup_root.rename(project_root) raise if not keep_backup: shutil.rmtree(backup_root, ignore_errors=True) return {"changed": True, "project": project_root.name, "backup": str(backup_root) if keep_backup else None, "mappings": mappings, "verification": verify_project_dataset(project_root)} except Exception: shutil.rmtree(temp_root, ignore_errors=True) raise def migrate_project_depth_pngs(project_root: Path, *, keep_backup: bool = True) -> dict[str, Any]: """Convert an already-canonical project's PNG depth sidecars to HEVC. Older project migrations kept ``meta/depth///*.png`` for compatibility. The current contract stores logarithmic depth codes beside RGB streams as CQP HEVC ``gray12le`` video. Conversion is completed and verified for every sequence before the PNG tree is moved to a recoverable backup. """ project_root = Path(project_root) depth_root = project_root / "meta" / "depth" if not depth_root.is_dir(): return {"changed": False, "project": project_root.name, "reason": "no meta/depth directory"} if not is_project_dataset(project_root): raise RuntimeError(f"Not a canonical project dataset: {project_root}") rows = project_episode_rows(project_root) rows_by_id = { _episode_id_for_row(row, project_root.name): row for row in rows } rows_by_index = { int(row["episode_index"]): row for row in rows if _episode_index(row.get("episode_index")) is not None } sequences: list[tuple[str, Path, list[Path], dict[str, Any]]] = [] def add_sequence(episode_id: str | None, source_dir: Path) -> None: frames = sorted( path for path in source_dir.iterdir() if path.is_file() and path.suffix.lower() == ".png" and path.stem.isdigit() ) if not frames: return numbers = [int(path.stem) for path in frames] expected = list(range(numbers[0], numbers[0] + len(numbers))) if numbers != expected: raise RuntimeError( f"Depth frame sequence has gaps for {source_dir}: " f"expected consecutive files, got {frames[0].name}..{frames[-1].name}" ) row = rows_by_id.get(str(episode_id or "")) if episode_id else None if row is None and episode_id and str(episode_id).isdigit(): row = rows_by_index.get(int(episode_id)) if row is None and len(rows) == 1: row = rows[0] if row is None: raise RuntimeError( f"Cannot map depth directory {source_dir} to an episode row" ) sequences.append((str(source_dir.name), source_dir, frames, row)) # Canonical migrated data is namespaced per episode. Also accept the # single-episode flat form produced by early v2.1 migrations. direct = [path for path in sorted(depth_root.iterdir()) if path.is_dir() and any(p.is_file() and p.suffix.lower() == ".png" and p.stem.isdigit() for p in path.iterdir())] if direct: for source_dir in direct: add_sequence(None, source_dir) for episode_dir in sorted(depth_root.iterdir()): if not episode_dir.is_dir() or episode_dir in direct: continue for source_dir in sorted(episode_dir.iterdir()): if source_dir.is_dir(): add_sequence(episode_dir.name, source_dir) if not sequences: return {"changed": False, "project": project_root.name, "reason": "no numeric PNG depth sequences"} # Duplicate source+episode pairs would overwrite one another and are not # safe to infer. Stop before touching the active dataset in that case. markers = [(int(row["episode_index"]), source) for source, _directory, _frames, row in sequences] if len(markers) != len(set(markers)): raise RuntimeError("Duplicate depth source for the same episode") info_path = project_root / "meta" / "info.json" info = _read_json(info_path, {}) features = dict(info.get("features") or {}) extensions = dict(info.get("extensions") or {}) video_extensions = dict(info.get("video_extensions") or {}) output_records: list[dict[str, Any]] = [] fps_default = float(info.get("fps") or 30.0) for source, depth_dir, frames, row in sequences: episode_index = int(row["episode_index"]) chunk_index = episode_chunk_for_index(episode_index) destination = (project_root / "videos" / f"observation.images.{source}" / f"chunk-{chunk_index:03d}" / f"episode_{episode_index:06d}.mp4") probe = encode_depth_png_sequence( depth_dir, frames, destination, fps=fps_default, ) feature = _source_feature(info, source, probe, True) video_info = feature.setdefault("video_info", {}) video_info.update({ "video.channels": 1, "video.is_depth_map": True, "video.is_depth_visualization": False, "video.depth_scale": 0.001, "video.depth_encoding": DEPTH_VIDEO_ENCODING, "video.depth_min_mm": DEPTH_MIN_MM, "video.depth_max_mm": DEPTH_MAX_MM, "video.depth_qmax": DEPTH_QMAX, "video.depth_qp": DEPTH_QP, "video.depth_quantization": "log", "has_audio": False, }) feature["shape"] = [int(probe["height"]), int(probe["width"]), 1] features[f"observation.images.{source}"] = feature video_extensions[source] = "mp4" key = f"videos/observation.images.{source}/" count = int(probe.get("frames") or len(frames)) fps = float(probe.get("fps") or fps_default) row[f"videos/observation.images.{source}/chunk_index"] = chunk_index row[f"videos/observation.images.{source}/file_index"] = episode_index row[f"videos/observation.images.{source}/from_timestamp"] = 0.0 row[f"videos/observation.images.{source}/to_timestamp"] = ( count / fps if fps else 0.0 ) output_records.append({ "episode_id": _episode_id_for_row(row, project_root.name), "episode_index": episode_index, "source": source, "frames": count, "path": str(destination.relative_to(project_root)), "codec": probe.get("codec"), "pix_fmt": probe.get("pix_fmt"), }) # Only metadata after all output videos were verified. The old PNG tree # is moved, not deleted, so rollback remains possible. info["features"] = features info["video_extensions"] = video_extensions info["format"] = "lerobot_v2.1" info["codebase_version"] = "v2.1" info["chunks_size"] = PROJECT_CHUNK_SIZE info["extensions"] = { **extensions, "depth_storage": "videos/observation.images./chunk-{episode_chunk:03d}/episode_.mp4", "depth_encoding": ( "HEVC gray12le log codes, qp=6, range=full, " "depth_mm=[100,5000], qmax=4095, depth_scale=0.001" ), } timestamp = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ") backup_root = (project_root.parent.parent / ".backups" / "session-migrations" / f"{project_root.name}-depth-png-{timestamp}") if backup_root.exists(): raise RuntimeError(f"Backup destination already exists: {backup_root}") write_project_episode_index(project_root, rows) _write_json(info_path, info) backup_depth = backup_root / "meta" / "depth" backup_depth.parent.mkdir(parents=True, exist_ok=True) shutil.move(str(depth_root), str(backup_depth)) if not keep_backup: shutil.rmtree(backup_root, ignore_errors=True) verification = verify_project_dataset(project_root) if not verification["passed"]: raise RuntimeError(f"Depth migration verification failed: {verification['errors']}") return {"changed": True, "project": project_root.name, "converted": output_records, "backup": str(backup_root) if keep_backup else None, "verification": verification} def _source_names(project_root: Path) -> list[Path]: return sorted( path for path in Path(project_root).iterdir() if path.is_dir() and not path.name.startswith(".") and path.name not in _CANONICAL_ROOTS ) def _clone_source_payload(source: Path, target: Path) -> None: for name in ("data", "meta", "videos", "calibration", "depth"): candidate = source / name if candidate.is_dir(): clone_tree(candidate, target / name) for name in ("metadata.json", "timestamps.json", "upload_manifest.json"): candidate = source / name if candidate.is_file(): _link_or_copy(candidate, target / name) def _preserve_sidecars(source: Path, storage_root: Path, project_name: str, episode_id: str) -> None: for source_name, target_name in (("processed", "processed"), ("original", "archives")): source_dir = source / source_name if not source_dir.is_dir(): continue clone_tree(source_dir, Path(storage_root) / target_name / project_name / episode_id) def allocate_project_episode_id(project_root: Path, incoming_name: str, project_name: str) -> str: """Choose a stable episode ID without creating a batch subdirectory.""" incoming = str(incoming_name or "").strip() existing = project_episode_rows(project_root) existing_ids = { str(row.get("episode_id") or row.get("source_batch") or "") for row in existing } if incoming and incoming in existing_ids: return incoming if incoming and episode_index_from_name(incoming) is not None: return incoming used = [int(row.get("episode_index")) for row in existing if _episode_index(row.get("episode_index")) is not None] next_index = max(used, default=-1) + 1 prefix = str(project_name or project_root.name or "episode").strip() # Old projects may already contain a malformed pair such as # episode_index=0 / episode_id=Project_000001. Never reuse that ID for a # new upload; the storage index and the public ID are separate concerns. while f"{prefix}_{next_index:06d}" in existing_ids: next_index += 1 return f"{prefix}_{next_index:06d}" def append_project_episode(project_root: Path, episode_root: Path, episode_id: str, *, replace: bool = False) -> dict[str, Any]: """Atomically append/replace one episode in a project-level dataset.""" project_root = Path(project_root) episode_root = Path(episode_root) parent = project_root.parent stamp = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%S%fZ") temp_root = parent / f".{project_root.name}.append-{stamp}" backup_root = (project_root.parent.parent / ".backups" / "session-uploads" / f"{project_root.name}-{stamp}") shutil.rmtree(temp_root, ignore_errors=True) temp_root.mkdir(parents=True, exist_ok=False) try: if project_root.is_dir() and not is_project_dataset(project_root): legacy = [path for path in project_root.iterdir() if path.is_dir() and not path.name.startswith(".") and path.name not in _CANONICAL_ROOTS] if legacy: raise RuntimeError( f"Project still uses legacy batch directories; migrate first: {project_root}" ) if project_root.is_dir(): clone_tree(project_root, temp_root) rows = project_episode_rows(temp_root) existing = next((row for row in rows if str(row.get("episode_id") or row.get("source_batch") or "") == str(episode_id)), None) if existing is None: used = {_episode_index(row.get("episode_index")) for row in rows if _episode_index(row.get("episode_index")) is not None} # ★ 优先沿用**名字里的编号**，让存储索引与公开 ID 一致。 # # 此前一律 max(used)+1，于是"名字说 000005、索引是 3"这种对不上的 # 情况会出现 —— 实测 119 个带编号批次里 **21 个**中招。它的直接 # 危害在 _match_episode_row 的名字回退：按名字推出索引后会撞上 # 别的批次（模拟：删 ..._000002 时若它自己的行已缺失，会去删 # index=2 那个批次的数据和视频）。 # # 编号允许空洞（见本模块 delete_project_episode 的说明 # "gaps are valid"），所以直接采用名字里的编号是安全的。 # 两种情况仍回落到 max+1：名字里没有编号；或该编号已被别的批次 # 占用（撞号）。 named = episode_index_from_name(episode_id) if named is not None and named not in used: target_episode_index = named else: target_episode_index = max(used, default=-1) + 1 chunk_index = episode_chunk_for_index(target_episode_index) else: if not replace: raise RuntimeError(f"Episode already exists: {episode_id}") target_episode_index = _episode_index(existing.get("episode_index")) if target_episode_index is None: raise RuntimeError(f"Existing episode has no episode_index: {episode_id}") chunk_index = _chunk_index(existing.get("data/chunk_index")) result = merge_normalized_episode( temp_root, episode_root, episode_id, chunk_index, replace=replace, target_episode_index=target_episode_index, ) verification = verify_project_dataset(temp_root) if not verification["passed"]: raise RuntimeError(f"Upload verification failed: {verification['errors']}") backup_root.parent.mkdir(parents=True, exist_ok=True) project_existed = project_root.is_dir() if project_existed: project_root.rename(backup_root) try: temp_root.rename(project_root) except Exception: if project_existed and backup_root.exists(): backup_root.rename(project_root) raise # ★ 换入之后必须失效【project_root】的缓存，不能只靠 # write_project_episode_index 里那次 —— 那次失效的是 **temp_root** # 的 key，而内容是靠 rename 换过来的。 # # 不失效的后果有两条： # (1) 紧跟的 verify_project_dataset(project_root) 读到的还是**追加前** # 的旧行 —— 校验等于在验旧树，"通过"是空头支票（实测：连续三次 # append 全部通过，而盘上索引早已是 [0,1,2,5,6,7]）； # (2) 缓存有 60 秒的盲目 TTL（TTL 内连指纹都不查），上传后前端 # 最长 60 秒看不到新批次。 invalidate_project_cache(project_root) result["verification"] = verify_project_dataset(project_root) result["backup"] = str(backup_root) if project_existed else None return result except Exception: shutil.rmtree(temp_root, ignore_errors=True) raise def migrate_project_dataset(project_root: Path, *, keep_backup: bool = True) -> dict[str, Any]: """Flatten legacy batch directories into one project-level dataset. The active project is atomically replaced only after all episodes and metadata validate. The old tree is retained under ``data/.backups``. """ project_root = Path(project_root) if not project_root.is_dir(): raise FileNotFoundError(project_root) batches = _source_names(project_root) if not batches: return {"changed": False, "project": project_root.name, "verification": verify_project_dataset(project_root)} storage_root = project_root.parent.parent stamp = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ") temp_root = project_root.parent / f".{project_root.name}.migrate-{stamp}" backup_root = (storage_root / ".backups" / "session-migrations" / f"{project_root.name}-{stamp}") shutil.rmtree(temp_root, ignore_errors=True) temp_root.mkdir(parents=True, exist_ok=False) mappings: list[dict[str, Any]] = [] try: for target_episode_index, batch in enumerate(batches): source_stage = temp_root / f".source-{target_episode_index:03d}" _clone_source_payload(batch, source_stage) normalized = normalize_extracted_dataset(source_stage, batch.name) mappings.append(merge_normalized_episode( temp_root, source_stage, batch.name, episode_chunk_for_index(target_episode_index), target_episode_index=target_episode_index, )) _preserve_sidecars(batch, storage_root, project_root.name, batch.name) shutil.rmtree(source_stage, ignore_errors=True) verification = verify_project_dataset(temp_root) if not verification["passed"]: raise RuntimeError(f"Project migration verification failed: {verification['errors']}") backup_root.parent.mkdir(parents=True, exist_ok=True) if backup_root.exists(): raise RuntimeError(f"Backup destination already exists: {backup_root}") project_root.rename(backup_root) try: temp_root.rename(project_root) except Exception: backup_root.rename(project_root) raise if not keep_backup: # This option is intentionally opt-in; callers should normally # retain the backup until the UI and worker are verified. shutil.rmtree(backup_root, ignore_errors=True) return { "changed": True, "project": project_root.name, "backup": str(backup_root) if keep_backup else None, "episodes": mappings, "verification": verify_project_dataset(project_root), } except Exception: shutil.rmtree(temp_root, ignore_errors=True) raise