Spaces:
Running
Running
| """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 | |
| 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 | |