diff options
Diffstat (limited to '')
| -rw-r--r-- | cs2pov/cli.py | 191 | ||||
| -rw-r--r-- | cs2pov/comms.py | 97 | ||||
| -rw-r--r-- | cs2pov/settings.py | 16 | ||||
| -rw-r--r-- | cs2pov/trim.py | 153 |
4 files changed, 398 insertions, 59 deletions
diff --git a/cs2pov/cli.py b/cs2pov/cli.py index 498e6f8..fa43a62 100644 --- a/cs2pov/cli.py +++ b/cs2pov/cli.py @@ -389,6 +389,12 @@ def record_demo( print(f" Console log saved: {saved_log_path.name}") console_log_path = saved_log_path + # Save transitions sidecar for standalone trim + if transitions: + from .trim import save_transitions + trans_path = save_transitions(transitions, output_path) + print(f" Transitions saved: {trans_path.name}") + success = exit_reason in ("demo_ended", "cs2_exited", "segments_complete") return RecordingResult( success=success, @@ -407,7 +413,7 @@ def postprocess_video( timeline: Optional[DemoTimeline] = None, startup_time_override: Optional[float] = None, transitions: Optional[list[GotoTransition]] = None, -) -> Path: +) -> tuple[Path, Optional[list[tuple[float, float]]]]: """Post-process a recorded video to keep only alive segments. If transitions are provided (from tick-nav recording), uses lightweight @@ -418,21 +424,31 @@ def postprocess_video( 2. Convert alive_segments from demo time to video time 3. Extract and concatenate only the alive segments + Returns: + Tuple of (video_path, keep_segments). keep_segments is the list of + (start, end) tuples used for trimming, or None if no trimming occurred. + Args: console_log_path: Path to console log (used for demo end detection/duration) startup_time_override: If provided, use this value instead of calculating startup_time. Useful when the automatic calculation is wrong. transitions: If provided, use transition-based trimming (from --tick-nav) """ - from .trim import get_video_duration, extract_and_concat_segments, trim_goto_transitions + from .trim import get_video_duration, extract_and_concat_segments, trim_goto_transitions, compute_keep_segments, load_transitions if not video_path.exists(): print(f"Error: Video file not found: {video_path}") - return video_path + return video_path, None raw_size_mb = video_path.stat().st_size / (1024 * 1024) print(f"\nRaw recording: {video_path} ({raw_size_mb:.1f} MB)") + # Auto-load transitions from sidecar file if not provided + if transitions is None: + transitions = load_transitions(video_path) + if transitions: + print(f" Loaded {len(transitions)} transitions from sidecar file") + # Tick-nav transition-based trimming (lightweight) if transitions: print(f"\nPost-processing: trimming {len(transitions)} goto transitions...") @@ -440,22 +456,23 @@ def postprocess_video( video_path.rename(raw_path) print(f" Raw recording moved to: {raw_path.name}") - success = trim_goto_transitions( + keep_segments = trim_goto_transitions( input_path=raw_path, output_path=video_path, transitions=transitions, verbose=verbose, ) - if success and video_path.exists(): + if keep_segments and video_path.exists(): final_size_mb = video_path.stat().st_size / (1024 * 1024) print(f"\nFinal recording: {video_path} ({final_size_mb:.1f} MB)") else: print("\nTransition trimming failed, restoring raw recording") if not video_path.exists() and raw_path.exists(): raw_path.rename(video_path) + keep_segments = None - return video_path + return video_path, keep_segments print("\nPost-processing: calculating segments to keep...") @@ -478,7 +495,7 @@ def postprocess_video( except Exception as e: print(f" Error getting video duration: {e}") print(f"\nRecording saved: {video_path} ({raw_size_mb:.1f} MB)") - return video_path + return video_path, None # Step 2: Get demo duration for startup_time calculation demo_duration = 0.0 @@ -520,35 +537,27 @@ def postprocess_video( if demo_duration == 0.0: print(" Error: Could not determine demo duration") print(f"\nRecording saved: {video_path} ({raw_size_mb:.1f} MB)") - return video_path + return video_path, None - # Step 3: Calculate startup time (recording before demo started) + # Step 3: Calculate startup time and convert alive segments to video time if startup_time_override is not None: - startup_time = startup_time_override if verbose: - print(f" Startup time: {startup_time:.2f}s (manual override)") + print(f" Startup time: {startup_time_override:.2f}s (manual override)") else: - startup_time = video_duration - demo_duration - if startup_time < 0: - print(f" Warning: Negative startup time ({startup_time:.2f}s), using 0") - startup_time = 0.0 + calc_startup = video_duration - demo_duration + if calc_startup < 0: + print(f" Warning: Negative startup time ({calc_startup:.2f}s), using 0") if verbose: - print(f" Startup time: {startup_time:.2f}s (calculated)") + print(f" Startup time: {max(0.0, video_duration - demo_duration):.2f}s (calculated)") - # Step 4: Convert alive segments to video time - for seg in timeline.alive_segments: - video_start = startup_time + seg.start_time - video_end = startup_time + seg.end_time - - # Clamp to video bounds - video_start = max(0.0, video_start) - video_end = min(video_duration, video_end) + video_segments = compute_keep_segments( + timeline.alive_segments, video_duration, demo_duration, + startup_time_override=startup_time_override, + ) - # Only include if segment has meaningful duration - if video_end > video_start + 0.5: - video_segments.append((video_start, video_end)) - if verbose: - print(f" Keep: {video_start:.2f}s - {video_end:.2f}s ({video_end - video_start:.2f}s)") + if verbose: + for i, (vs, ve) in enumerate(video_segments): + print(f" Keep: {vs:.2f}s - {ve:.2f}s ({ve - vs:.2f}s)") # Execute trimming if video_segments: @@ -577,15 +586,17 @@ def postprocess_video( if success and video_path.exists(): final_size_mb = video_path.stat().st_size / (1024 * 1024) print(f"\nFinal recording: {video_path} ({final_size_mb:.1f} MB)") + return video_path, video_segments else: print("\nTrimming failed, restoring raw recording") if not video_path.exists() and raw_path.exists(): raw_path.rename(video_path) + return video_path, None else: print(" No segments to extract, keeping original") print(f"\nRecording saved: {video_path} ({raw_size_mb:.1f} MB)") - return video_path + return video_path, None # ============================================================================= @@ -833,6 +844,8 @@ Examples: help="Output file (default: adds _trimmed suffix)") trim_parser.add_argument("--startup-time", type=float, help="Override startup time (seconds from video start to demo start)") + trim_parser.add_argument("--tick-nav", action="store_true", + help="Video was recorded with --tick-nav (changes trim strategy)") # COMMS command comms_parser = subparsers.add_parser( @@ -852,6 +865,8 @@ Examples: help="Comms volume multiplier (default: 1.0)") comms_parser.add_argument("--game-volume", type=float, default=1.0, help="Game audio volume multiplier (default: 1.0)") + comms_parser.add_argument("--tick-nav", action="store_true", + help="Video was recorded with tick-nav mode (extends segment boundaries)") comms_parser.add_argument("--no-trim-sync", action="store_true", help="Skip alive-segment sync (for untrimmed videos)") @@ -884,6 +899,13 @@ def _validate_required_args(args, command: str) -> Optional[str]: missing.append("-d/--demo") if getattr(args, "player", None) is None: missing.append("-p/--player") + elif command == "trim": + if getattr(args, "video", None) is None: + missing.append("video") + if getattr(args, "demo", None) is None: + missing.append("-d/--demo") + if getattr(args, "player", None) is None: + missing.append("-p/--player") else: if getattr(args, "demo", None) is None: missing.append("-d/--demo") @@ -924,12 +946,39 @@ def _run_single_comms(args) -> int: return cmd_comms(args) +def _run_single_trim(args) -> int: + """Run a single trim job from config.""" + video = getattr(args, "video", None) + if video is None: + print("Error: trim job requires 'video' field", file=sys.stderr) + return 1 + args.video = Path(video) + + # startup_time can be None (auto-detect) + if not hasattr(args, "startup_time"): + args.startup_time = None + + # tick_nav defaults to False + if not hasattr(args, "tick_nav"): + args.tick_nav = False + elif isinstance(args.tick_nav, str): + args.tick_nav = args.tick_nav.lower() in ("true", "1", "yes") + + # output is optional — cmd_trim generates {stem}_trimmed if unset + if not hasattr(args, "output") or args.output is None: + args.output = None + + return cmd_trim(args) + + def _job_runner_for_type(job_type: str): """Return the single-job runner for a given job type.""" if job_type == "record": return _run_single_record if job_type == "comms": return _run_single_comms + if job_type == "trim": + return _run_single_trim return _run_single_pov @@ -1137,8 +1186,9 @@ def _run_single_pov(args) -> int: ) # Post-process + keep_segments = None if result.success and not args.no_trim: - postprocess_video( + _, keep_segments = postprocess_video( video_path=result.video_path, console_log_path=result.console_log_path, verbose=args.verbose, @@ -1175,6 +1225,8 @@ def _run_single_pov(args) -> int: comms_volume=float(getattr(args, "comms_volume", 1.0) or 1.0), is_trimmed=not args.no_trim, verbose=args.verbose, + tick_nav=args.tick_nav, + keep_segments=keep_segments, ) if comms_ok: size_mb = comms_output.stat().st_size / (1024 * 1024) @@ -1334,14 +1386,40 @@ def cmd_trim(args) -> int: if video_path != output_path: shutil.copy2(video_path, output_path) - # Run trimming - postprocess_video( - video_path=output_path, - console_log_path=Path("/dev/null"), - verbose=args.verbose, - timeline=timeline, - startup_time_override=args.startup_time, - ) + # Tick-nav path: use transitions (sidecar or reconstructed) + if getattr(args, "tick_nav", False): + from .trim import load_transitions, reconstruct_transitions, get_video_duration + + transitions = load_transitions(output_path) + if transitions: + print(f" Loaded {len(transitions)} transitions from sidecar file") + else: + video_duration = get_video_duration(output_path) + transitions = reconstruct_transitions( + timeline.alive_segments, video_duration + ) + if transitions: + print(f" Reconstructed {len(transitions)} transitions from alive segments") + else: + print(" Only 1 alive segment, nothing to trim") + return 0 + + postprocess_video( + video_path=output_path, + console_log_path=Path("/dev/null"), + verbose=args.verbose, + timeline=timeline, + transitions=transitions, + ) + else: + # Standard path: alive-segment-based trimming + postprocess_video( + video_path=output_path, + console_log_path=Path("/dev/null"), + verbose=args.verbose, + timeline=timeline, + startup_time_override=args.startup_time, + ) return 0 @@ -1392,6 +1470,39 @@ def cmd_comms(args) -> int: return 1 from .comms import apply_comms_overlay + from .trim import get_video_duration, compute_keep_segments + + # Recompute keep segments for standalone comms on an already-trimmed video. + # Skip when tick_nav=True: let compute_comms_segments use its own extension + # logic (death +0.9s, round_end → next prestart - 0.2s) which matches + # what tick-nav actually recorded. + keep_segments = None + if is_trimmed and timeline and timeline.alive_segments and not args.tick_nav: + try: + video_duration = get_video_duration(video_path) + # Estimate demo duration from timeline + demo_duration = 0.0 + if timeline.total_duration > 0: + demo_duration = timeline.total_duration + elif timeline.rounds: + for r in reversed(timeline.rounds): + if r.end_tick is not None: + demo_duration = r.end_time + 5.0 + break + if demo_duration == 0.0 and timeline.alive_segments: + demo_duration = timeline.alive_segments[-1].end_time + 5.0 + + if demo_duration > 0: + keep_segments = compute_keep_segments( + timeline.alive_segments, video_duration, demo_duration, + ) + if args.verbose: + print(f" Recomputed {len(keep_segments)} keep segments for comms sync") + except Exception as e: + if args.verbose: + print(f" Warning: Could not recompute keep segments: {e}") + elif args.tick_nav and args.verbose: + print(" Using tick-nav segment extensions (skipping keep_segments recomputation)") print(f"Overlaying comms: {comms_path.name} → {output_path.name}") success = apply_comms_overlay( @@ -1404,6 +1515,8 @@ def cmd_comms(args) -> int: comms_volume=float(args.comms_volume), is_trimmed=is_trimmed, verbose=args.verbose, + tick_nav=args.tick_nav, + keep_segments=keep_segments, ) if success: diff --git a/cs2pov/comms.py b/cs2pov/comms.py index 7d96e20..8fb1698 100644 --- a/cs2pov/comms.py +++ b/cs2pov/comms.py @@ -5,13 +5,16 @@ import tempfile from pathlib import Path from typing import Optional -from .preprocessor import AliveSegment, DemoTimeline +from .preprocessor import AliveSegment, DemoTimeline, RoundBoundary def compute_comms_segments( alive_segments: list[AliveSegment], + rounds: list[RoundBoundary], round1_freeze_end_time: float, r1_sync_time: float = 0.0, + tick_nav: bool = False, + keep_durations: Optional[list[float]] = None, ) -> list[tuple[float, float]]: """Map alive segments from demo time to comms audio time. @@ -19,18 +22,48 @@ def compute_comms_segments( r1_sync_time is the timestamp in the comms audio where round 1 starts. Segments before comms t=0 are clamped or dropped. + When keep_durations is provided (from trim's actual keep segments), those + durations override the computed end times. This ensures comms segments + match the exact durations in the trimmed video. + + When tick_nav=True and keep_durations is not available, segment end times + are extended to match the actual navigation behavior as a fallback. + Args: alive_segments: Alive segments from DemoTimeline + rounds: Round boundaries from DemoTimeline round1_freeze_end_time: Demo time (seconds) of round 1 freeze end r1_sync_time: Seconds into comms audio where round 1 begins + tick_nav: Whether tick-nav mode was used (extends segment boundaries) + keep_durations: Durations from trim's actual keep segments (overrides end time calc) Returns: List of (start, end) tuples in comms audio time (seconds) """ segments = [] - for seg in alive_segments: + for i, seg in enumerate(alive_segments): comms_start = seg.start_time - round1_freeze_end_time + r1_sync_time - comms_end = seg.end_time - round1_freeze_end_time + r1_sync_time + + if keep_durations and i < len(keep_durations): + # Use trim's actual duration — guarantees sync with trimmed video + comms_end = comms_start + keep_durations[i] + else: + # Fallback: compute end from demo timeline + end_time = seg.end_time + + if tick_nav: + # Match navigation.py segment duration extensions + if seg.reason_ended == "round_end": + next_seg = alive_segments[i + 1] if i + 1 < len(alive_segments) else None + if next_seg: + for r in rounds: + if r.round_num == next_seg.round_num and r.prestart_time is not None: + end_time = r.prestart_time - 0.2 + break + elif seg.reason_ended == "death": + end_time += 0.9 + + comms_end = end_time - round1_freeze_end_time + r1_sync_time if comms_end <= 0: continue @@ -174,6 +207,7 @@ def overlay_comms_on_video( output_path: Path, game_volume: float = 1.0, comms_volume: float = 1.0, + tempo: Optional[float] = None, verbose: bool = False, ) -> bool: """Mix comms audio onto video's existing game audio using FFmpeg amix. @@ -187,6 +221,7 @@ def overlay_comms_on_video( output_path: Output video path game_volume: Volume multiplier for game audio (default 1.0) comms_volume: Volume multiplier for comms audio (default 1.0) + tempo: Optional atempo correction factor (e.g. 1.005 to speed up slightly) verbose: Print debug info Returns: @@ -194,11 +229,16 @@ def overlay_comms_on_video( """ has_audio = _has_audio_stream(video_path) + # Build comms filter chain: volume → optional atempo + comms_filters = f"volume={comms_volume}" + if tempo is not None: + comms_filters += f",atempo={tempo}" + if has_audio: # Mix game audio + comms audio filter_complex = ( f"[0:a]volume={game_volume}[game];" - f"[1:a]volume={comms_volume}[comms];" + f"[1:a]{comms_filters}[comms];" f"[game][comms]amix=inputs=2:duration=first:dropout_transition=0[aout]" ) cmd = [ @@ -214,13 +254,14 @@ def overlay_comms_on_video( str(output_path), ] else: - # No game audio — use comms as sole audio track + # No game audio — apply filters to comms and use as sole audio cmd = [ "ffmpeg", "-y", "-i", str(video_path), "-i", str(comms_audio_path), + "-filter_complex", f"[1:a]{comms_filters}[aout]", "-map", "0:v", - "-map", "1:a", + "-map", "[aout]", "-c:v", "copy", "-c:a", "aac", "-b:a", "192k", @@ -229,7 +270,8 @@ def overlay_comms_on_video( if verbose: mode = "mixing with game audio" if has_audio else "adding as sole audio" - print(f" [Comms] Overlaying comms ({mode})") + tempo_info = f", atempo={tempo:.6f}" if tempo else "" + print(f" [Comms] Overlaying comms ({mode}{tempo_info})") try: result = subprocess.run(cmd, capture_output=True, text=True, timeout=600) @@ -255,12 +297,15 @@ def apply_comms_overlay( comms_volume: float = 1.0, is_trimmed: bool = True, verbose: bool = False, + tick_nav: bool = False, + keep_segments: Optional[list[tuple[float, float]]] = None, ) -> bool: """High-level entry point: trim comms to match alive segments, then overlay onto video. If the video is trimmed and a timeline is provided, the comms audio is cut - to match the same alive segments so it stays in sync. Otherwise, the comms - are overlaid directly with an offset seek. + to match the same alive segments so it stays in sync. When keep_segments + is provided (from trim's actual output), those durations are used instead + of recomputing from the timeline, preventing audio drift. Args: video_path: Input video (trimmed or raw) @@ -272,6 +317,8 @@ def apply_comms_overlay( comms_volume: Comms audio volume (default 1.0) is_trimmed: Whether the video has been trimmed verbose: Print debug info + tick_nav: Whether tick-nav mode was used (fallback for segment extensions) + keep_segments: Actual keep segments from trim (overrides duration calculation) Returns: True on success @@ -283,9 +330,17 @@ def apply_comms_overlay( print(" Warning: No round data found, using demo start as reference") round1_time = 0.0 + # Extract durations from trim's keep segments if available + keep_durations = None + if keep_segments: + keep_durations = [end - start for start, end in keep_segments] + if verbose: + print(f" [Comms] Using {len(keep_durations)} keep segment durations from trim") + # Compute which parts of comms audio correspond to alive segments comms_segments = compute_comms_segments( - timeline.alive_segments, round1_time, r1_sync_time + timeline.alive_segments, timeline.rounds, round1_time, + r1_sync_time, tick_nav=tick_nav, keep_durations=keep_durations, ) if not comms_segments: @@ -303,10 +358,30 @@ def apply_comms_overlay( print(" Error: Failed to extract comms segments") return False + # Compute atempo correction for residual drift + tempo = None + try: + from .trim import get_video_duration + comms_dur = get_video_duration(trimmed_comms) + video_dur = get_video_duration(video_path) + if comms_dur > 0 and video_dur > 0 and abs(comms_dur - video_dur) > 1.0: + tempo = comms_dur / video_dur + if verbose: + print(f" [Comms] Atempo correction: {tempo:.6f} " + f"(comms={comms_dur:.2f}s, video={video_dur:.2f}s, " + f"drift={comms_dur - video_dur:+.2f}s)") + elif verbose: + print(f" [Comms] No atempo needed " + f"(comms={comms_dur:.2f}s, video={video_dur:.2f}s, " + f"drift={comms_dur - video_dur:+.2f}s)") + except Exception as e: + if verbose: + print(f" [Comms] Could not compute atempo: {e}") + # Overlay trimmed comms onto trimmed video return overlay_comms_on_video( video_path, trimmed_comms, output_path, - game_volume, comms_volume, verbose, + game_volume, comms_volume, tempo, verbose, ) else: # Untrimmed or no timeline — overlay directly with offset diff --git a/cs2pov/settings.py b/cs2pov/settings.py index 78f6e92..4da4ecb 100644 --- a/cs2pov/settings.py +++ b/cs2pov/settings.py @@ -42,10 +42,10 @@ VALID_DEFAULT_KEYS = set(HARDCODED_DEFAULTS.keys()) | {"player", "demo", "output REQUIRED_MERGED_KEYS = {"demo", "player", "output"} # Keys valid in a job entry -VALID_JOB_KEYS = REQUIRED_MERGED_KEYS | VALID_DEFAULT_KEYS | {"type", "video", "audio", "no_trim_sync"} +VALID_JOB_KEYS = REQUIRED_MERGED_KEYS | VALID_DEFAULT_KEYS | {"type", "video", "audio", "no_trim_sync", "startup_time"} # Valid job types -VALID_JOB_TYPES = {"pov", "record", "comms"} +VALID_JOB_TYPES = {"pov", "record", "comms", "trim"} CONFIG_FILENAME = "cs2pov.json" CURRENT_VERSION = 1 @@ -180,6 +180,8 @@ def _validate_setting_types(settings: dict[str, Any], context: str) -> None: "player": (str,), "demo": (str,), "output": (str,), + "video": (str,), + "startup_time": (int, float, type(None)), } for key, value in settings.items(): @@ -209,7 +211,7 @@ def resolve_paths(job_dict: dict[str, Any], config_dir: Path) -> dict[str, Any]: """Make relative paths in job_dict absolute against config_dir.""" result = dict(job_dict) - for key in ("demo", "output"): + for key in ("demo", "output", "video"): if key in result and result[key] is not None: p = Path(result[key]) if not p.is_absolute(): @@ -289,6 +291,14 @@ def generate_default_config(cs2_path: Optional[str] = None) -> dict: "comms_volume": 1.0, "game_volume": 0.7, }, + { + "type": "trim", + "video": "./recordings/example_raw.mp4", + "demo": "./demos/example.dem", + "player": "PlayerName", + "output": "./recordings/example_trimmed.mp4", + "startup_time": None, + }, ], } diff --git a/cs2pov/trim.py b/cs2pov/trim.py index ea2abdd..61a7204 100644 --- a/cs2pov/trim.py +++ b/cs2pov/trim.py @@ -1,12 +1,77 @@ """Post-processing: trim segments from recorded video.""" +import json import subprocess import tempfile from pathlib import Path +from typing import Optional from .exceptions import CaptureError +def _transitions_path(video_path: Path) -> Path: + """Return the sidecar transitions JSON path for a video.""" + return video_path.parent / f"{video_path.stem}_transitions.json" + + +def save_transitions(transitions: list, video_path: Path) -> Path: + """Save GotoTransition list to a sidecar JSON file alongside the video. + + Args: + transitions: List of GotoTransition objects + video_path: Path to the video file + + Returns: + Path to the saved transitions file + """ + path = _transitions_path(video_path) + data = [ + { + "pause_video_time": t.pause_video_time, + "unpause_video_time": t.unpause_video_time, + "from_tick": t.from_tick, + "to_tick": t.to_tick, + } + for t in transitions + ] + path.write_text(json.dumps(data, indent=2)) + return path + + +def load_transitions(video_path: Path) -> Optional[list]: + """Load GotoTransition list from a sidecar JSON file. + + Args: + video_path: Path to the video file + + Returns: + List of GotoTransition objects, or None if no sidecar file exists + """ + from .navigation import GotoTransition + + path = _transitions_path(video_path) + if not path.exists(): + return None + + try: + data = json.loads(path.read_text()) + except (json.JSONDecodeError, OSError): + return None + + if not isinstance(data, list) or not data: + return None + + return [ + GotoTransition( + pause_video_time=t["pause_video_time"], + unpause_video_time=t["unpause_video_time"], + from_tick=t["from_tick"], + to_tick=t["to_tick"], + ) + for t in data + ] + + def get_video_duration(video_path: Path) -> float: """Get video duration in seconds using ffprobe. @@ -46,7 +111,7 @@ def trim_goto_transitions( output_path: Path, transitions: list, verbose: bool = False, -) -> bool: +) -> Optional[list[tuple[float, float]]]: """Trim goto transition artifacts from a tick-nav recording. When using tick-based navigation, each death triggers a pause/seek/unpause @@ -63,12 +128,12 @@ def trim_goto_transitions( verbose: Print debug output Returns: - True if trimming was performed, False on failure or nothing to trim + List of keep segments on success, None on failure or nothing to trim """ if not transitions: if verbose: print(" [Trim] No transitions to trim") - return False + return None # Get video duration try: @@ -76,7 +141,7 @@ def trim_goto_transitions( except CaptureError as e: if verbose: print(f" [Trim] Failed to get video duration: {e}") - return False + return None # Build keep segments from the gaps between transitions keep_segments: list[tuple[float, float]] = [] @@ -100,7 +165,7 @@ def trim_goto_transitions( if not keep_segments: if verbose: print(" [Trim] No segments to keep after transition removal") - return False + return None if verbose: total_keep = sum(end - start for start, end in keep_segments) @@ -108,7 +173,8 @@ def trim_goto_transitions( print(f" [Trim] {len(transitions)} transitions → {len(keep_segments)} keep segments") print(f" [Trim] Keep: {total_keep:.1f}s, Trim: {total_trim:.1f}s") - return extract_and_concat_segments(input_path, output_path, keep_segments, verbose) + success = extract_and_concat_segments(input_path, output_path, keep_segments, verbose) + return keep_segments if success else None def extract_and_concat_segments( @@ -228,3 +294,78 @@ def extract_and_concat_segments( print(f" [Trim] Output video: {output_duration:.2f}s") return True + + +def reconstruct_transitions(alive_segments, video_duration: float) -> list: + """Reconstruct GotoTransitions from alive segment durations and video duration. + + For tick-nav recordings without a sidecar file, we know: + - Each alive segment played for its full duration + - Between segments are goto artifacts of unknown but uniform duration + - total_artifact_time = video_duration - sum(segment durations) + + Args: + alive_segments: List of AliveSegment objects from DemoTimeline + video_duration: Total video duration in seconds + + Returns: + List of GotoTransition objects (empty if <= 1 segment) + """ + from .navigation import GotoTransition + + if len(alive_segments) <= 1: + return [] + + total_segment_time = sum( + seg.end_time - seg.start_time for seg in alive_segments + ) + num_gaps = len(alive_segments) - 1 + total_artifact_time = max(0.0, video_duration - total_segment_time) + artifact_per_gap = total_artifact_time / num_gaps + + transitions = [] + pos = 0.0 + for i, seg in enumerate(alive_segments[:-1]): + seg_duration = seg.end_time - seg.start_time + pause_time = pos + seg_duration + unpause_time = pause_time + artifact_per_gap + transitions.append(GotoTransition( + pause_video_time=pause_time, + unpause_video_time=unpause_time, + from_tick=seg.end_tick, + to_tick=alive_segments[i + 1].start_tick, + )) + pos = unpause_time + + return transitions + + +def compute_keep_segments( + alive_segments, + video_duration: float, + demo_duration: float, + startup_time_override: Optional[float] = None, +) -> list[tuple[float, float]]: + """Convert alive segments to video-time keep segments. + + Args: + alive_segments: List of AliveSegment objects from DemoTimeline + video_duration: Total video duration in seconds + demo_duration: Total demo duration in seconds + startup_time_override: Manual startup time override + + Returns: + List of (start, end) tuples in video time (seconds) + """ + if startup_time_override is not None: + startup_time = startup_time_override + else: + startup_time = max(0.0, video_duration - demo_duration) + + segments = [] + for seg in alive_segments: + start = max(0.0, startup_time + seg.start_time) + end = min(video_duration, startup_time + seg.end_time) + if end > start + 0.5: + segments.append((start, end)) + return segments |