Store action playback times from final video frames

Dockerfile

@@ -17,7 +17,7 @@ COPY requirements.txt ./
 RUN pip install --no-cache-dir -r requirements.txt
 COPY --from=node-build --chown=appuser:appuser /app/dist ./dist
 COPY --from=node-build --chown=appuser:appuser /app/node_modules ./node_modules
-COPY --chown=appuser:appuser package.json app.py start.py storage_listing.py ./
+COPY --chown=appuser:appuser package.json app.py start.py storage_listing.py playback_timing.py backfill_playback_times.py ./
 USER appuser
 EXPOSE 7860
 CMD ["python", "start.py"]

README.md

@@ -104,6 +104,12 @@ Brevo is configured as Supabase Auth's SMTP provider; the Express API does not s
 
 Brevo credentials stay in Supabase SMTP settings and are not added to the browser or committed to `.env`.
 
+## Action playback timestamps
+
+Run `Database/add_action_playback_times.sql` in the Supabase SQL Editor before relying on saved action playback times. New completed actions store the presentation timestamps of their first frame and the frame after their last frame from the finalized camera video. Older actions remain playable using the FPS calculation until backfilled.
+
+To fill older completed sessions, run `python backfill_playback_times.py` from this directory with `SUPABASE_URL`, `SUPABASE_SERVICE_ROLE_KEY`, `STORAGE_REPO_ID`, and `HF_TOKEN` set in `Backend/.env` or the environment. Add `--session-id UUID` to process one session. The script downloads one camera video per session and updates only actions whose playback times are missing, so the first run can take time for large videos. Re-running it is safe.
+
 ## Routes
 
 - `POST /api/auth/signup` — `{ fname, lname, email, password, academy? }`; sends Supabase's confirmation email when confirmation is enabled.

app.py

@@ -31,6 +31,7 @@ from fastapi.responses import FileResponse, JSONResponse
 from huggingface_hub import HfApi, hf_hub_download, snapshot_download
 from huggingface_hub.errors import RemoteEntryNotFoundError
 from storage_listing import session_files
+from playback_timing import action_times, frame_times
 
 load_dotenv(Path(__file__).with_name(".env"))
 jobs = {}
@@ -438,7 +439,7 @@ def finalize_videos(work, cameras, job):
             "ffmpeg", "-hide_banner", "-loglevel", "error", "-nostats", "-y", "-i", str(source),
             "-map", "0:v:0", "-an", "-sn", "-dn", "-vf", "scale=trunc(iw/2)*2:trunc(ih/2)*2",
             "-c:v", "libx264", "-preset", "veryfast", "-crf", "23", "-pix_fmt", "yuv420p",
-            "-movflags", "+faststart", "-progress", "pipe:1", str(temporary),
+            "-fps_mode", "passthrough", "-movflags", "+faststart", "-progress", "pipe:1", str(temporary),
         ]
         process = subprocess.Popen(command, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, bufsize=1)
         stderr_lines = []
@@ -518,13 +519,21 @@ def run_job(client_id, work, cameras):
         finalize_videos(work, cameras, job)
         if job.get("cancel"):
             raise InterruptedError()
+        try:
+            playback_boundaries = frame_times(work / "videos" / f"camera_{min(cameras)}.mp4")
+        except (OSError, ValueError, subprocess.CalledProcessError):
+            logger.exception("Unable to index final video timestamps for session %s", session_id)
+            playback_boundaries = None
         job.update(progress=98, phase="Saving results", stage="saving", videoProgress=100)
         hub().upload_folder(repo_id=os.environ["STORAGE_REPO_ID"], repo_type="dataset",
             folder_path=str(work), path_in_repo=f"sessions/{session_id}",
             allow_patterns=["results.json", "videos/*", "posters/*.webp"])
         saved = True
         completion = {"status": "completed", "path": f"sessions/{session_id}", "actions": [
-            {"index": index, **{key: json_value(report[key]) for key in ("action", "start_frame", "end_frame", "fps", "error") if key in report}}
+            {"index": index, **{key: json_value(report[key]) for key in ("action", "start_frame", "end_frame", "fps", "error") if key in report},
+             **(action_times(playback_boundaries, int(report["start_frame"]), int(report["end_frame"]))
+                if playback_boundaries and not report.get("error")
+                and int(report["end_frame"]) + 1 < len(playback_boundaries) else {})}
             for index, report in enumerate(reports)]}
         job["completion"] = completion
         backend(f"/{session_id}", job.get("ownerUserId") or job["userId"], "PUT", completion, is_admin=job.get("actorIsAdmin", False))
@@ -862,6 +871,14 @@ def update_action(session_id: UUID, action_id: UUID, payload: dict, request: Req
         and result_action_id(session_id, index) == str(action_id)), None)
     if matching_index is None:
         raise HTTPException(404, "Action not found")
+    video_dir = f'{row["path"]}/videos/'
+    video_files = sorted(file for file in session_files(hub(), os.environ["STORAGE_REPO_ID"], video_dir.rstrip("/"))
+                         if re.search(r"/camera_\d+\.mp4$", file))
+    if not video_files:
+        raise HTTPException(409, "Session video is unavailable")
+    video_path = hf_hub_download(repo_id=os.environ["STORAGE_REPO_ID"], repo_type="dataset",
+                                 filename=video_files[0], token=os.environ["HF_TOKEN"])
+    playback_times = action_times(frame_times(video_path), start_frame, end_frame)
     reports[matching_index]["start_frame"] = start_frame
     reports[matching_index]["end_frame"] = end_frame
     reports[matching_index]["action"] = action_type.strip()
@@ -870,6 +887,8 @@ def update_action(session_id: UUID, action_id: UUID, payload: dict, request: Req
     try:
         backend(f"/{session_id}/actions/{action_id}", request.state.user_id, "PATCH", {
             "startFrame": start_frame, "endFrame": end_frame, "type": action_type.strip(),
+            "startTimeSeconds": playback_times["start_time_seconds"],
+            "endTimeSeconds": playback_times["end_time_seconds"],
         }, is_admin=True)
     except Exception:
         try:
@@ -916,6 +935,7 @@ def results(session_id: UUID, request: Request):
             camera_files.append((int(camera_id_text), filename))
     camera_files.sort()
     fps_values = [float(report.get("fps") or 30) for report in reports if report.get("fps")]
+    saved_times = {item["action_id"]: item for item in row.get("action_times", [])}
     actions, failed = [], []
     for index, report in enumerate(reports):
         if report.get("_deleted"):
@@ -965,10 +985,17 @@ def results(session_id: UUID, request: Request):
             "poster": f"/storage/{session_id}/posters/camera_{camera_id}.webp"}
             for camera_id, filename in camera_files]
         action_id = result_action_id(session_id, index)
+        playback = saved_times.get(action_id) or {}
+        start_seconds = playback.get("start_time_seconds")
+        end_seconds = playback.get("end_time_seconds")
+        has_playback_times = (isinstance(start_seconds, (int, float)) and isinstance(end_seconds, (int, float))
+                              and math.isfinite(start_seconds) and math.isfinite(end_seconds)
+                              and start_seconds >= 0 and end_seconds > start_seconds)
         actions.append({"id": action_id, "label": label,
             "start": str(start), "end": str(end), "fps": action_fps,
-            "startFrame": start, "endFrame": end, "startSeconds": start / action_fps,
-            "endSeconds": (end + 1) / action_fps, "totalFrames": max(0, end - start + 1),
+            "startFrame": start, "endFrame": end, "startSeconds": start_seconds if has_playback_times else start / action_fps,
+            "endSeconds": end_seconds if has_playback_times else (end + 1) / action_fps,
+            "hasPlaybackTimes": has_playback_times, "totalFrames": max(0, end - start + 1),
             "preFrames": 0, "inFrame": max(0, (end - start) // 2), "postFrames": 0,
             "cameraClips": clips, "sourceCameraClips": [], "preMetrics": [],
             "preActionMetrics": [], "inActionMetrics": metrics, "postMetrics": [],

backfill_playback_times.py

@@ -0,0 +1,81 @@
+"""Fill action playback times from the stored final MP4 videos.
+
+Run after Database/add_action_playback_times.sql:
+    python backfill_playback_times.py [--session-id UUID]
+"""
+
+import argparse
+import os
+from collections import defaultdict
+from pathlib import Path
+
+import httpx
+from dotenv import load_dotenv
+from huggingface_hub import HfApi, hf_hub_download
+
+from playback_timing import action_times, frame_times
+from storage_listing import session_files
+
+
+def main():
+    load_dotenv(Path(__file__).with_name(".env"))
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--session-id", help="Backfill one session only")
+    args = parser.parse_args()
+
+    url = os.environ["SUPABASE_URL"].rstrip("/") + "/rest/v1"
+    key = os.environ["SUPABASE_SERVICE_ROLE_KEY"]
+    repo = os.environ["STORAGE_REPO_ID"]
+    token = os.environ["HF_TOKEN"]
+    hub = HfApi(token=token)
+    headers = {"apikey": key, "Authorization": f"Bearer {key}"}
+    actions = []
+    with httpx.Client(headers=headers, timeout=60) as client:
+        offset = 0
+        while True:
+            params = {"select": "action_id,session_id,start_frame,end_frame", "start_time_seconds": "is.null",
+                      "order": "action_id", "limit": "1000", "offset": str(offset)}
+            if args.session_id:
+                params["session_id"] = f"eq.{args.session_id}"
+            response = client.get(f"{url}/actions", params=params)
+            response.raise_for_status()
+            page = response.json()
+            actions.extend(page)
+            if len(page) < 1000:
+                break
+            offset += len(page)
+
+        grouped = defaultdict(list)
+        for action in actions:
+            grouped[action["session_id"]].append(action)
+        saved = 0
+        for session_id, session_actions in grouped.items():
+            try:
+                response = client.get(f"{url}/sessions", params={"select": "path,status", "session_id": f"eq.{session_id}"})
+                response.raise_for_status()
+                sessions = response.json()
+                if len(sessions) != 1 or sessions[0]["status"] != "completed" or sessions[0]["path"] != f"sessions/{session_id}":
+                    print(f"Skipping {session_id}: completed session storage unavailable")
+                    continue
+                video_dir = f"sessions/{session_id}/videos"
+                files = sorted(file for file in session_files(hub, repo, video_dir)
+                               if Path(file).name.startswith("camera_") and Path(file).suffix.lower() == ".mp4")
+                if not files:
+                    print(f"Skipping {session_id}: no MP4 video")
+                    continue
+                video = hf_hub_download(repo_id=repo, repo_type="dataset", filename=files[0], token=token)
+                boundaries = frame_times(video)
+                for action in session_actions:
+                    times = action_times(boundaries, action["start_frame"], action["end_frame"])
+                    response = client.patch(f"{url}/actions", params={"action_id": f'eq.{action["action_id"]}',
+                                                                "start_time_seconds": "is.null"}, json=times)
+                    response.raise_for_status()
+                    saved += 1
+                print(f"Saved {len(session_actions)} actions for {session_id}")
+            except Exception as error:
+                print(f"Failed {session_id}: {error}")
+        print(f"Done: {saved}/{len(actions)} action rows updated")
+
+
+if __name__ == "__main__":
+    main()

playback_timing.py

@@ -0,0 +1,38 @@
+"""Read action playback boundaries from the presentation timestamps of a final video."""
+
+import json
+import math
+import subprocess
+
+
+def frame_times(video_path):
+    result = subprocess.run(
+        ["ffprobe", "-v", "error", "-select_streams", "v:0", "-show_frames",
+         "-show_entries", "frame=best_effort_timestamp_time,duration_time,pkt_duration_time",
+         "-of", "json", str(video_path)],
+        check=True, capture_output=True, text=True,
+    )
+    frames = json.loads(result.stdout).get("frames", [])
+    try:
+        times = [float(frame["best_effort_timestamp_time"]) for frame in frames]
+    except (KeyError, TypeError, ValueError) as error:
+        raise ValueError(f"Missing video frame timestamps: {video_path}") from error
+    if not times or not all(math.isfinite(value) and value >= 0 for value in times):
+        raise ValueError(f"Invalid video frame timestamps: {video_path}")
+    if any(after <= before for before, after in zip(times, times[1:])):
+        raise ValueError(f"Non-increasing video frame timestamps: {video_path}")
+    last = frames[-1]
+    duration = last.get("duration_time") or last.get("pkt_duration_time")
+    duration = float(duration) if duration is not None else 0
+    if duration <= 0 and len(times) > 1:
+        duration = times[-1] - times[-2]
+    if not math.isfinite(duration) or duration <= 0:
+        raise ValueError(f"Missing final frame duration: {video_path}")
+    return times + [times[-1] + duration]
+
+
+def action_times(boundaries, start_frame, end_frame):
+    if start_frame < 0 or end_frame < start_frame or end_frame + 1 >= len(boundaries):
+        raise ValueError(f"Action frame range {start_frame}-{end_frame} is outside video")
+    return {"start_time_seconds": boundaries[start_frame],
+            "end_time_seconds": boundaries[end_frame + 1]}

src/analysis.ts

@@ -13,6 +13,8 @@ const beginSchema = z.object({ clientId: z.uuid(), playerId: z.uuid(), name: z.s
 const reportSchema = z.object({
   index: z.number().int().nonnegative(), action: z.string().min(1).max(50),
   start_frame: z.number().int().nonnegative().optional(), end_frame: z.number().int().nonnegative().optional(),
+  start_time_seconds: z.number().finite().nonnegative().optional(),
+  end_time_seconds: z.number().finite().positive().optional(),
   error: z.string().max(2000).optional(),
 });
 export const completionSchema = z.discriminatedUnion("status", [
@@ -23,10 +25,14 @@ export const completionSchema = z.discriminatedUnion("status", [
 export function actionRows(sessionId: string, reports: z.infer<typeof reportSchema>[]) {
   if (new Set(reports.map((report) => report.index)).size !== reports.length) throw new Error("Duplicate action index");
   return reports.filter((report) => report.error === undefined).map((report) => {
-    const { start_frame: startFrame, end_frame: endFrame } = report;
+    const { start_frame: startFrame, end_frame: endFrame,
+      start_time_seconds: startTime, end_time_seconds: endTime } = report;
     if (startFrame === undefined || endFrame === undefined || endFrame < startFrame) throw new Error("Invalid action interval");
+    if ((startTime === undefined) !== (endTime === undefined) ||
+        (startTime !== undefined && endTime !== undefined && endTime <= startTime)) throw new Error("Invalid action playback times");
     return { action_id: stableId(`${sessionId}:${report.index}`), type: report.action,
-      start_frame: startFrame, end_frame: endFrame };
+      start_frame: startFrame, end_frame: endFrame,
+      start_time_seconds: startTime ?? null, end_time_seconds: endTime ?? null };
   });
 }
 
@@ -111,7 +117,13 @@ export function analysisRouter(db: SupabaseClient, secret: string, onCompleted?:
       if (res.locals.isAdmin !== true) query = query.eq("user_id", res.locals.userId);
       const { data, error } = await query.maybeSingle();
       if (error) return res.status(503).json({ error: "Unable to load session" });
-      return data ? res.json(data) : res.sendStatus(404);
+      if (!data) return res.sendStatus(404);
+      const { data: actionTimes, error: timesError } = await db.from("actions")
+        .select("action_id,start_time_seconds,end_time_seconds").eq("session_id", sessionId);
+      if (timesError) {
+        console.error("Unable to load action playback times", { sessionId, code: timesError.code, message: timesError.message });
+      }
+      return res.json({ ...data, action_times: actionTimes ?? [] });
     } catch (error) { next(error); }
   });
   router.delete("/:sessionId/actions/:actionId", async (req, res, next) => {
@@ -142,15 +154,27 @@ export function analysisRouter(db: SupabaseClient, secret: string, onCompleted?:
         startFrame: z.number().int().nonnegative(),
         endFrame: z.number().int().nonnegative(),
         type: z.string().trim().min(1).max(50),
+        startTimeSeconds: z.number().finite().nonnegative(),
+        endTimeSeconds: z.number().finite().positive(),
       }).strict().parse(req.body);
       if (input.endFrame < input.startFrame) return res.status(400).json({ error: "Action end frame must be after its start frame." });
+      if (input.endTimeSeconds <= input.startTimeSeconds) return res.status(400).json({ error: "Action playback interval is invalid." });
       const { data: session } = await db.from("sessions").select("status").eq("session_id", sessionId).maybeSingle();
       if (!session || ["pending", "uploading", "queued", "running"].includes(session.status)) return res.status(409).json({ error: "Actions cannot be edited while analysis is running." });
       const { data, error } = await db.from("actions").update({
         start_frame: input.startFrame, end_frame: input.endFrame, type: input.type,
+        start_time_seconds: input.startTimeSeconds, end_time_seconds: input.endTimeSeconds,
       }).eq("action_id", actionId).eq("session_id", sessionId).select("action_id").maybeSingle();
-      if (error) return res.status(503).json({ error: "Unable to update action." });
-      if (!data) return res.sendStatus(404);
+      if (error && !["42703", "PGRST204"].includes(error.code)) return res.status(503).json({ error: "Unable to update action." });
+      let updated = data;
+      if (error) {
+        const fallback = await db.from("actions").update({
+          start_frame: input.startFrame, end_frame: input.endFrame, type: input.type,
+        }).eq("action_id", actionId).eq("session_id", sessionId).select("action_id").maybeSingle();
+        if (fallback.error) return res.status(503).json({ error: "Unable to update action." });
+        updated = fallback.data;
+      }
+      if (!updated) return res.sendStatus(404);
       return res.json({ ok: true });
     } catch (error) { next(error); }
   });
@@ -195,7 +219,12 @@ export function analysisRouter(db: SupabaseClient, secret: string, onCompleted?:
           const rows = actions.map((action) => ({ ...action, session_id: sessionId }));
           if (rows.length) {
             const { error } = await db.from("actions").upsert(rows, { onConflict: "action_id" });
-            if (error) return res.status(503).json({ error: "Unable to save analysis actions." });
+            if (error && !["42703", "PGRST204"].includes(error.code)) return res.status(503).json({ error: "Unable to save analysis actions." });
+            if (error) {
+              const legacyRows = rows.map(({ start_time_seconds: _start, end_time_seconds: _end, ...row }) => row);
+              const fallback = await db.from("actions").upsert(legacyRows, { onConflict: "action_id" });
+              if (fallback.error) return res.status(503).json({ error: "Unable to save analysis actions." });
+            }
           }
           if (onCompleted && session.user_id) void onCompleted(sessionId, session.user_id).catch((error) => console.error("Automatic Agent feedback generation failed", { sessionId, error }));
         }