Explain What Happened: Pegasus 1.6 Summaries and Segments

1Use my brief and selected files for chapter 3 only. Plan one small analysis2increment and wait for approval before applying it. Cite current official3Pegasus API contracts, not invented fields. Preserve schemas, task-ID persistence,4timestamp validation, uncertainty, and rejection of truncated/missing-end streams.5Run strict local tests first; inspect actual responses/source footage only after6an approved live run. Debug with redacted errors and explain each change.7Do not read secrets, commit/push, provision, or make paid calls without my8explicit approval. Do not jump chapters. I review the Git diff and commit.

1Review the chapter-3 diff and local tests against current official API contracts.2Check typed inputs, response schema, missing/length stream endings, failed tasks,3invalid timestamps, and preserved uncertainty. Distinguish mock evidence from4actual responses; require human checks of any approved live output.5Report gaps with redacted errors. Do not read secrets or jump chapters.6Do not commit/push, provision, or make paid calls without my explicit approval.7Explain the Git diff for my review; I commit only what I understand.

1"""Chapter 3: Pegasus 1.6 streamed summary.2 3Documentation-checked against the TwelveLabs Python SDK sync-analysis4reference; live validation pending. Contract notes:5- analyze_stream(model_name="pegasus1.6", video=VideoContext_AssetId(asset_id=...), prompt=...)6- Iterate stream events; inspect the stream_end event's finish_reason.7  "length" or a missing reason means the output is incomplete — do not ship it.8"""9import json10import os11from pathlib import Path12 13from dotenv import load_dotenv14from twelvelabs import TwelveLabs15from twelvelabs.types import VideoContext_AssetId16 17load_dotenv()18client = TwelveLabs(api_key=os.environ["TWELVELABS_API_KEY"])19 20ROOT = Path(__file__).parent21RESULTS = ROOT / "local-results"22RESULTS.mkdir(exist_ok=True)23state = json.loads((ROOT / "state.json").read_text())24 25PROMPT = (26    "Summarize what happens in this clip in three sentences. "27    "Describe visible actions in order and state uncertainty explicitly."28)29 30 31def summarize(clip_id: str) -> str:32    asset_id = state["assets"][clip_id]["asset_id"]33    text_parts: list[str] = []34    finish_reason: str | None = None35    stream = client.analyze_stream(36        model_name="pegasus1.6",37        video=VideoContext_AssetId(asset_id=asset_id),38        prompt=PROMPT,39    )40    for event in stream:41        event_type = getattr(event, "event_type", None) or getattr(event, "type", None)42        if event_type == "text_generation":43            text_parts.append(getattr(event, "text", ""))44        elif event_type == "stream_end":45            finish_reason = getattr(event, "metadata", None) and getattr(46                event.metadata, "finish_reason", None47            ) or getattr(event, "finish_reason", None)48    if finish_reason is None:49        raise RuntimeError("stream ended without a finish_reason — treat output as incomplete")50    if finish_reason != "stop":51        raise RuntimeError(f"incomplete output: finish_reason={finish_reason!r}")52    return "".join(text_parts)53 54 55if __name__ == "__main__":56    for clip_id in state["assets"]:57        text = summarize(clip_id)58        (RESULTS / f"{clip_id}-summary.txt").write_text(text)59        print(f"{clip_id}: {len(text)} chars")

1"""Chapter 3: Pegasus 1.6 async segmentation with explicit definitions.2 3Documentation-checked against the segment-videos quickstart and the4async-analysis SDK reference; live validation pending. Contract notes:5- analyze_async.tasks.create(model_name="pegasus1.6", video=VideoContext_AssetId(...),6  analysis_mode="time_based_metadata",7  response_format=AsyncResponseFormat(type="segment_definitions", segment_definitions=[...]))8- The created object exposes task.task_id (not task.id) — persist it BEFORE polling.9- On completion, parse json.loads(done.result.data), select your definition-ID key,10  and validate start_time/end_time/metadata before normalizing to start_sec/end_sec.11- canceled/failed statuses and local timeouts are handled, never silently retried.12"""13import math14import json15import os16import time17from pathlib import Path18 19from dotenv import load_dotenv20from twelvelabs import TwelveLabs21from twelvelabs.types import AsyncResponseFormat, VideoContext_AssetId22 23load_dotenv()24client = TwelveLabs(api_key=os.environ["TWELVELABS_API_KEY"])25 26ROOT = Path(__file__).parent27RESULTS = ROOT / "local-results"28RESULTS.mkdir(exist_ok=True)29STATE_PATH = ROOT / "state.json"30state = json.loads(STATE_PATH.read_text())31 32 33def save_state() -> None:34    STATE_PATH.write_text(json.dumps(state, indent=2))35 36 37DEFINITION_ID = "desk-actions"38DEFINITIONS = [{39    "id": DEFINITION_ID,40    "description": "Segment the clip into distinct visible desk actions.",41    "fields": [42        {"name": "action", "type": "string", "description": "A short visible action label."},43        {"name": "uncertainty", "type": "string", "description": "What cannot be established from the footage."},44    ],45}]46 47 48def start_segmentation(clip_id: str) -> str:49    state.setdefault("segment_tasks", {})50    if clip_id in state["segment_tasks"]:51        task_id = state["segment_tasks"][clip_id]["task_id"]52        print(f"resuming {clip_id}: task {task_id}")53        return task_id54    task = client.analyze_async.tasks.create(55        model_name="pegasus1.6",56        video=VideoContext_AssetId(asset_id=state["assets"][clip_id]["asset_id"]),57        analysis_mode="time_based_metadata",58        response_format=AsyncResponseFormat(type="segment_definitions", segment_definitions=DEFINITIONS),59    )60    # Persist BEFORE polling: a crash here must not orphan a running task.61    state["segment_tasks"][clip_id] = {"task_id": task.task_id}62    save_state()63    return task.task_id64 65 66def wait_task(task_id: str, timeout_sec: int = 1800):67    deadline = time.time() + timeout_sec68    delay = 569    while time.time() < deadline:70        done = client.analyze_async.tasks.retrieve(task_id)71        status = getattr(done, "status", "unknown")72        if status == "ready":73            return done74        if status in ("failed", "canceled"):75            raise RuntimeError(f"task {task_id} ended with status={status}")76        time.sleep(delay)77        delay = min(delay * 2, 30)78    raise TimeoutError(79        f"local timeout — remote task {task_id} may still run; re-run to resume"80    )81 82 83def parse_segments(done) -> list[dict]:84    """Parse and validate the definition payload before trusting it."""85    payload = json.loads(done.result.data)86    raw = payload.get(DEFINITION_ID) if isinstance(payload, dict) else None87    if not isinstance(raw, list):88        raise RuntimeError(f"missing definition key {DEFINITION_ID!r} in result")89    segments = []90    for i, seg in enumerate(raw):91        if not isinstance(seg, dict):92            raise RuntimeError(f"segment {i}: expected object")93        start, end = seg.get("start_time"), seg.get("end_time")94        if not isinstance(start, (int, float)) or not isinstance(end, (int, float)):95            raise RuntimeError(f"segment {i}: invalid start_time/end_time: {seg!r}")96        if isinstance(start, bool) or isinstance(end, bool) or not math.isfinite(start) or not math.isfinite(end) or start < 0 or end <= start:97            raise RuntimeError(f"segment {i}: end_time <= start_time: {seg!r}")98        metadata = seg.get("metadata")99        if not isinstance(metadata, dict) or any(not isinstance(metadata.get(k), str) for k in ("action", "uncertainty")):100            raise RuntimeError(f"segment {i}: invalid action/uncertainty metadata")101        segments.append({102            "start_sec": float(start),103            "end_sec": float(end),104            "metadata": metadata,105        })106    return segments107 108 109if __name__ == "__main__":110    for clip_id in state["assets"]:111        task_id = start_segmentation(clip_id)112        done = wait_task(task_id)113        segments = parse_segments(done)114        (RESULTS / f"{clip_id}-segments.json").write_text(json.dumps(segments, indent=2))115        print(f"{clip_id}: {len(segments)} segments")

1"""Chapter 3 exercise: still-image understanding with Pegasus 1.6.2 3Extract one frame per clip with ffmpeg, upload it as an image asset, and4ask Pegasus to describe it. Compare the still answer with the video summary:5what does the model lose without motion and audio?6 7Documentation-checked; live validation pending.8"""9import hashlib10import time11import json12import os13import subprocess14from pathlib import Path15 16from dotenv import load_dotenv17from twelvelabs import TwelveLabs18from twelvelabs.types import AnalyzeImageInput19from managed_search import wait_asset20 21load_dotenv()22client = TwelveLabs(api_key=os.environ["TWELVELABS_API_KEY"])23 24ROOT = Path(__file__).parent25RESULTS = ROOT / "local-results"26FRAMES = ROOT / "local-results" / "frames"27FRAMES.mkdir(parents=True, exist_ok=True)28STATE_PATH = ROOT / "state.json"29state = json.loads(STATE_PATH.read_text())30 31 32 33def extract_frame(clip: dict, at_sec: float) -> Path:34    out = FRAMES / f"{clip['id']}-{int(at_sec)}s.jpg"35    if not out.exists():36        subprocess.run(37            ["ffmpeg", "-y", "-ss", str(at_sec), "-i", str(ROOT / clip["file"]),38             "-frames:v", "1", "-q:v", "3", str(out)],39            capture_output=True, check=True,40        )41    return out42 43 44def ensure_frame_asset(frame: Path) -> str:45    """Reuse the same checksummed frame on interruptions, including processing."""46    checksum = hashlib.sha256(frame.read_bytes()).hexdigest()47    key = str(frame.relative_to(ROOT))48    saved = state.setdefault("frames", {}).get(key)49    if saved and saved["sha256"] != checksum:50        raise RuntimeError("frame checksum changed — explicitly remove the stale frame entry")51    if not saved:52        with frame.open("rb") as f:53            asset = client.assets.create(method="direct", file=f)54        saved = {"asset_id": asset.id, "sha256": checksum}55        state["frames"][key] = saved56        STATE_PATH.write_text(json.dumps(state, indent=2))57    wait_asset(client, saved["asset_id"])58    return saved["asset_id"]59 60 61def describe_image(frame: Path) -> str:62    asset_id = ensure_frame_asset(frame)63    text_parts: list[str] = []64    finish_reason = None65    stream = client.analyze_stream(66        model_name="pegasus1.6",67        image=[AnalyzeImageInput(name="frame", asset_id=asset_id)],68        prompt="Describe exactly what is visible in this single frame. State what you cannot tell from a still image.",69    )70    for event in stream:71        event_type = getattr(event, "event_type", None) or getattr(event, "type", None)72        if event_type == "text_generation":73            text_parts.append(getattr(event, "text", ""))74        elif event_type == "stream_end":75            finish_reason = getattr(event, "finish_reason", None) or getattr(76                getattr(event, "metadata", None), "finish_reason", None77            )78    if finish_reason != "stop":79        raise RuntimeError(f"incomplete still-image output: finish_reason={finish_reason!r}")80    return "".join(text_parts)81 82 83if __name__ == "__main__":84    manifest = json.loads((ROOT / "media-manifest.json").read_text())85    for clip in manifest["clips"]:86        midpoint = clip["duration_sec"] / 287        frame = extract_frame(clip, midpoint)88        answer = describe_image(frame)89        video_summary = (RESULTS / f"{clip['id']}-summary.txt").read_text()90        comparison = {91            "clip": clip["id"],92            "frame": str(frame.relative_to(ROOT)),93            "still_answer": answer,94            "video_summary": video_summary,95            "your_notes": "",  # fill in: what did the still get wrong?96        }97        (RESULTS / f"{clip['id']}-still-comparison.json").write_text(98            json.dumps(comparison, indent=2)99        )100        print(f"{clip['id']}: still vs video comparison saved")

1git status && git diff --cached2git add pegasus_summaries.py pegasus_segments.py still_image.py3git commit -m "Chapter 3: Pegasus 1.6 summaries, segments, still-image exercise"4git push