Spaces:
Running
Running
File size: 4,976 Bytes
702222d | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 | """Turn step-level video ETL output (src/video_steps.py) into index documents.
doc_type stays ``action_chunk`` so retrieval, citation ids (``<clip>::clip::NNNN``),
the media resolver and the UI keep working unchanged; what changes is the unit:
one *step* (sentence-aligned, real absolute timestamps, English) instead of a fixed
5-10 s window with a model-guessed timestamp.
Two texts per step:
* ``text`` stored in the payload; what the diagnosis LLM and the UI see
(narration + visual description + parts/tools/on-screen text).
* ``embed_text`` what gets vectorised: narration-first. On AutoRepair-VideoQA,
appending the visual description to the embedded text lowered
retrieval accuracy, so it is kept for answering/display only.
"""
from __future__ import annotations
import json
import os
import shutil
from dataclasses import dataclass
from pathlib import PurePosixPath
from .indexer import RAGDocument, stable_doc_id
from .ref_id import clip_ref_id
from .video_steps import clean_on_screen, mmss, step_text
@dataclass(frozen=True)
class IndexedDoc:
doc: RAGDocument
embed_text: str
def _posix(path: str) -> str:
return str(PurePosixPath(path.replace("\\", "/")))
def build_step_documents(
*,
steps_json: str,
metadata_path: str,
base_metadata: dict,
video_path: str,
frames_out_root: str,
) -> list[IndexedDoc]:
"""``metadata_path`` / ``video_path`` should be project-root-relative; they are stored
POSIX-style so an index built on Windows resolves on a Linux Space."""
data = json.load(open(steps_json, encoding="utf-8"))
# sentence-level ASR timeline (transcript.json sits next to steps.json); lets an answer point
# at "00:39-00:58 of this step" instead of the whole step
tr_path = os.path.join(os.path.dirname(steps_json), "transcript.json")
asr_segments = []
if os.path.isfile(tr_path):
asr_segments = json.load(open(tr_path, encoding="utf-8")).get("segments", [])
clip_id = data["clip_id"]
task = data["task"]
steps = data["steps"]
meta_rel = _posix(metadata_path)
video_rel = _posix(video_path)
frames_dir = os.path.join(frames_out_root, clip_id)
os.makedirs(frames_dir, exist_ok=True)
out: list[IndexedDoc] = []
for s in steps:
idx = int(s["idx"])
# copy the step's sampled frames next to the index so the app can show thumbnails
copied: list[str] = []
for k, fr in enumerate(s.get("frames") or []):
src = fr["path"]
if not os.path.isfile(src):
continue
dst = os.path.join(frames_dir, f"step_{idx:03d}_{k}.jpg")
if not os.path.exists(dst):
shutil.copyfile(src, dst)
copied.append(_posix(os.path.relpath(dst, start=os.getcwd())))
mid = copied[len(copied) // 2] if copied else None
step = dict(s)
step["on_screen_text"] = clean_on_screen(s.get("on_screen_text") or [])
full = step_text(step, task=task, include_visual=True)
narr_only = step_text(step, task=task, include_visual=False)
embed = narr_only if len((s.get("narration") or "").strip()) >= 40 else full
timeline = [
[round(float(g["start"]), 1), round(float(g["end"]), 1), str(g["text"]).strip()[:160]]
for g in asr_segments
if float(g["end"]) > float(s["start"]) and float(g["start"]) < float(s["end"])
][:16]
meta = dict(base_metadata or {})
meta.update(
{
"doc_type": "action_chunk",
"timeline": timeline,
"schema_version": 2,
"clip_id": clip_id,
"chunk_index": idx,
"step_count": len(steps),
"ref_id": clip_ref_id(meta_rel, chunk_index=idx),
"metadata_path": meta_rel,
"source_video": video_rel,
"start_time": f"{float(s['start']):.2f}",
"end_time": f"{float(s['end']):.2f}",
"start_seconds": float(s["start"]),
"end_seconds": float(s["end"]),
"task": base_metadata.get("task") or task,
"step_title": s.get("title") or "",
"title": f"{task} · {s.get('title') or 'step ' + str(idx + 1)} ({mmss(s['start'])}-{mmss(s['end'])})",
"video_title": base_metadata.get("title"),
"narration": s.get("narration") or "",
"visual_summary": s.get("visual_summary") or "",
"frame_path": mid,
"frame_paths": copied,
}
)
doc_id = stable_doc_id("action_chunk_v2", clip_id, f"{idx:04d}")
out.append(IndexedDoc(RAGDocument(id=doc_id, text=full, metadata=meta), embed))
return out
|