"""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]}