| """Immutable OpenVid row ledger shared by frozen video references. |
| |
| Unlike prompt datasets, a feature-reference ledger retains videos whose caption |
| is empty. Feature rows are identified exclusively by the contiguous source |
| sequence and MP4 basename; captions are descriptive payloads, never keys. |
| """ |
|
|
| from __future__ import annotations |
|
|
| import csv |
| import hashlib |
| import json |
| import os |
| from dataclasses import dataclass |
| from pathlib import Path |
| import tempfile |
| from typing import Iterable, Mapping |
|
|
|
|
| LEDGER_KIND = "openvid_video_reference_ledger" |
| LEDGER_SCHEMA_VERSION = 1 |
|
|
|
|
| def file_sha256(path: str | os.PathLike, chunk_size: int = 8 << 20) -> str: |
| digest = hashlib.sha256() |
| with open(path, "rb") as handle: |
| while block := handle.read(int(chunk_size)): |
| digest.update(block) |
| return digest.hexdigest() |
|
|
|
|
| def canonical_sha256(value) -> str: |
| payload = json.dumps( |
| value, |
| ensure_ascii=False, |
| sort_keys=True, |
| separators=(",", ":"), |
| allow_nan=False, |
| ).encode("utf-8") |
| return hashlib.sha256(payload).hexdigest() |
|
|
|
|
| def _atomic_bytes(path: Path, payload: bytes) -> None: |
| path.parent.mkdir(parents=True, exist_ok=True) |
| descriptor, temporary_name = tempfile.mkstemp( |
| dir=path.parent, prefix=f".{path.name}.", suffix=".tmp") |
| temporary = Path(temporary_name) |
| try: |
| with os.fdopen(descriptor, "wb") as handle: |
| handle.write(payload) |
| handle.flush() |
| os.fsync(handle.fileno()) |
| os.replace(temporary, path) |
| finally: |
| temporary.unlink(missing_ok=True) |
|
|
|
|
| @dataclass(frozen=True) |
| class OpenVidReferenceRow: |
| sequence: int |
| role: str |
| video_name: str |
| caption: str |
| video_size_bytes: int |
| video_mtime_ns: int |
|
|
| def validate(self, *, anchor_cohort_size: int) -> None: |
| if self.sequence < 0: |
| raise ValueError("OpenVid reference sequence must be non-negative") |
| expected_role = ( |
| "anchor" if self.sequence < int(anchor_cohort_size) else "reference") |
| if self.role != expected_role: |
| raise ValueError( |
| f"row {self.sequence} role {self.role!r} != {expected_role!r}") |
| if ( |
| Path(self.video_name).name != self.video_name |
| or not self.video_name.lower().endswith(".mp4") |
| ): |
| raise ValueError(f"invalid MP4 basename: {self.video_name!r}") |
| if self.video_size_bytes <= 0 or self.video_mtime_ns <= 0: |
| raise ValueError("video size/mtime identity must be positive") |
|
|
| def to_dict(self) -> dict: |
| return { |
| "caption": self.caption, |
| "role": self.role, |
| "sequence": self.sequence, |
| "video_mtime_ns": self.video_mtime_ns, |
| "video_name": self.video_name, |
| "video_size_bytes": self.video_size_bytes, |
| } |
|
|
| @classmethod |
| def from_mapping( |
| cls, value: Mapping, *, anchor_cohort_size: int |
| ) -> "OpenVidReferenceRow": |
| row = cls( |
| sequence=int(value["sequence"]), |
| role=str(value["role"]), |
| video_name=str(value["video_name"]), |
| caption=str(value.get("caption", "")), |
| video_size_bytes=int(value["video_size_bytes"]), |
| video_mtime_ns=int(value["video_mtime_ns"]), |
| ) |
| if set(value) != set(row.to_dict()): |
| raise ValueError("OpenVid reference row schema mismatch") |
| row.validate(anchor_cohort_size=anchor_cohort_size) |
| return row |
|
|
|
|
| class OpenVidReferencePromptDataset: |
| """Prompt-only view of a frozen feature ledger. |
| |
| The nine captionless videos remain in the 77,615-row feature reference but |
| are intentionally absent from text-conditioned student sampling. |
| """ |
|
|
| def __init__( |
| self, |
| ledger_path: str | os.PathLike, |
| *, |
| expected_count: int = 77_615, |
| ) -> None: |
| rows, metadata = load_openvid_reference_ledger( |
| ledger_path, expected_count=expected_count) |
| self.metadata = metadata |
| self.rows = tuple(row for row in rows if row.caption.strip()) |
| skipped = len(rows) - len(self.rows) |
| if skipped != int(metadata["caption_empty_count"]): |
| raise ValueError("OpenVid empty-caption ledger count mismatch") |
| if not self.rows: |
| raise ValueError("OpenVid ledger has no usable prompts") |
|
|
| def __len__(self) -> int: |
| return len(self.rows) |
|
|
| def __getitem__(self, index: int) -> dict: |
| row = self.rows[int(index)] |
| return { |
| "idx": row.sequence, |
| "prompts": row.caption, |
| "sample_ids": row.video_name, |
| } |
|
|
|
|
| def _video_name_sequence_sha256(rows: Iterable[OpenVidReferenceRow]) -> str: |
| digest = hashlib.sha256() |
| for row in rows: |
| digest.update(row.video_name.encode("utf-8")) |
| digest.update(b"\n") |
| return digest.hexdigest() |
|
|
|
|
| def _local_source_identity_sha256(rows: Iterable[OpenVidReferenceRow]) -> str: |
| digest = hashlib.sha256() |
| for row in rows: |
| digest.update(row.video_name.encode("utf-8")) |
| digest.update(b"\0") |
| digest.update(str(row.video_size_bytes).encode("ascii")) |
| digest.update(b"\0") |
| digest.update(str(row.video_mtime_ns).encode("ascii")) |
| digest.update(b"\0") |
| return digest.hexdigest() |
|
|
|
|
| def freeze_openvid_reference_ledger( |
| *, |
| source_csv: str | os.PathLike, |
| video_root: str | os.PathLike, |
| ledger_path: str | os.PathLike, |
| expected_count: int = 77_615, |
| anchor_cohort_size: int = 1_024, |
| ) -> dict: |
| """Validate the CSV/video mirror and atomically publish JSONL + metadata.""" |
| source_csv = Path(source_csv).expanduser().resolve(strict=True) |
| video_root = Path(video_root).expanduser().resolve(strict=True) |
| ledger_path = Path(ledger_path).expanduser().resolve() |
| expected_count = int(expected_count) |
| anchor_cohort_size = int(anchor_cohort_size) |
| if not video_root.is_dir(): |
| raise NotADirectoryError(video_root) |
| if not 0 < anchor_cohort_size <= expected_count: |
| raise ValueError("invalid OpenVid anchor cohort size") |
|
|
| rows: list[OpenVidReferenceRow] = [] |
| seen: set[str] = set() |
| with source_csv.open("r", newline="", encoding="utf-8") as handle: |
| reader = csv.DictReader(handle) |
| required = {"sequence", "role", "video", "caption"} |
| if reader.fieldnames is None or not required.issubset(reader.fieldnames): |
| raise ValueError( |
| f"OpenVid reference CSV requires {sorted(required)}") |
| for csv_line, value in enumerate(reader, start=2): |
| try: |
| sequence = int(value["sequence"]) |
| except (TypeError, ValueError) as exc: |
| raise ValueError( |
| f"invalid sequence at CSV line {csv_line}") from exc |
| if sequence != len(rows): |
| raise ValueError( |
| "OpenVid sequence is not contiguous in CSV order: " |
| f"{sequence} != {len(rows)}") |
| video_name = str(value["video"]).strip() |
| if video_name in seen: |
| raise ValueError(f"duplicate OpenVid video: {video_name}") |
| video_path = video_root / video_name |
| if not video_path.is_file(): |
| raise FileNotFoundError(video_path) |
| stat = video_path.stat() |
| row = OpenVidReferenceRow( |
| sequence=sequence, |
| role=str(value["role"]).strip(), |
| video_name=video_name, |
| |
| caption=str(value["caption"] or ""), |
| video_size_bytes=int(stat.st_size), |
| video_mtime_ns=int(stat.st_mtime_ns), |
| ) |
| row.validate(anchor_cohort_size=anchor_cohort_size) |
| rows.append(row) |
| seen.add(video_name) |
| if len(rows) != expected_count: |
| raise ValueError( |
| f"OpenVid reference count {len(rows)} != {expected_count}") |
|
|
| lines = "".join( |
| json.dumps( |
| row.to_dict(), ensure_ascii=False, sort_keys=True, |
| separators=(",", ":"), allow_nan=False, |
| ) + "\n" |
| for row in rows |
| ).encode("utf-8") |
| manifest_sha256 = hashlib.sha256(lines).hexdigest() |
| metadata = { |
| "anchor_cohort_size": anchor_cohort_size, |
| "anchor_video_name_sequence_sha256": ( |
| _video_name_sequence_sha256(rows[:anchor_cohort_size])), |
| "caption_empty_count": sum(not row.caption.strip() for row in rows), |
| "csv_path": str(source_csv), |
| "csv_sha256": file_sha256(source_csv), |
| "kind": LEDGER_KIND, |
| "ledger_file": ledger_path.name, |
| "ledger_sha256": manifest_sha256, |
| "local_source_identity_sha256": ( |
| _local_source_identity_sha256(rows)), |
| "reference_count": len(rows), |
| "schema_version": LEDGER_SCHEMA_VERSION, |
| "video_name_sequence_sha256": _video_name_sequence_sha256(rows), |
| "video_root": str(video_root), |
| } |
| _atomic_bytes(ledger_path, lines) |
| _atomic_bytes( |
| ledger_path.with_suffix(".meta.json"), |
| (json.dumps( |
| metadata, ensure_ascii=False, sort_keys=True, indent=2, |
| allow_nan=False, |
| ) + "\n").encode("utf-8"), |
| ) |
| return metadata |
|
|
|
|
| def load_openvid_reference_ledger( |
| ledger_path: str | os.PathLike, |
| *, |
| expected_count: int | None = None, |
| ) -> tuple[list[OpenVidReferenceRow], dict]: |
| path = Path(ledger_path).expanduser().resolve(strict=True) |
| meta_path = path.with_suffix(".meta.json") |
| metadata = json.loads(meta_path.read_text(encoding="utf-8")) |
| if ( |
| metadata.get("kind") != LEDGER_KIND |
| or int(metadata.get("schema_version", -1)) != LEDGER_SCHEMA_VERSION |
| or metadata.get("ledger_file") != path.name |
| or metadata.get("ledger_sha256") != file_sha256(path) |
| ): |
| raise ValueError("OpenVid reference ledger metadata/hash mismatch") |
| anchor_count = int(metadata["anchor_cohort_size"]) |
| rows = [] |
| with path.open("r", encoding="utf-8") as handle: |
| for line_number, line in enumerate(handle, start=1): |
| if not line.strip(): |
| raise ValueError(f"blank OpenVid ledger line {line_number}") |
| row = OpenVidReferenceRow.from_mapping( |
| json.loads(line), anchor_cohort_size=anchor_count) |
| if row.sequence != len(rows): |
| raise ValueError("OpenVid ledger sequence is not contiguous") |
| rows.append(row) |
| if len(rows) != int(metadata.get("reference_count", -1)): |
| raise ValueError("OpenVid ledger row count mismatch") |
| if expected_count is not None and len(rows) != int(expected_count): |
| raise ValueError( |
| f"OpenVid ledger count {len(rows)} != {int(expected_count)}") |
| if ( |
| _video_name_sequence_sha256(rows) |
| != metadata.get("video_name_sequence_sha256") |
| or _video_name_sequence_sha256(rows[:anchor_count]) |
| != metadata.get("anchor_video_name_sequence_sha256") |
| or _local_source_identity_sha256(rows) |
| != metadata.get("local_source_identity_sha256") |
| ): |
| raise ValueError("OpenVid ledger row identity mismatch") |
| return rows, metadata |
|
|
|
|
| __all__ = [ |
| "LEDGER_KIND", |
| "LEDGER_SCHEMA_VERSION", |
| "OpenVidReferenceRow", |
| "OpenVidReferencePromptDataset", |
| "canonical_sha256", |
| "file_sha256", |
| "freeze_openvid_reference_ledger", |
| "load_openvid_reference_ledger", |
| ] |
|
|
