diff options
| -rw-r--r-- | cs2pov/automation.py | 37 | ||||
| -rw-r--r-- | cs2pov/cli.py | 139 | ||||
| -rw-r--r-- | cs2pov/navigation.py | 91 | ||||
| -rw-r--r-- | cs2pov/trim.py | 388 |
4 files changed, 77 insertions, 578 deletions
diff --git a/cs2pov/automation.py b/cs2pov/automation.py index dbc90b8..ee8347d 100644 --- a/cs2pov/automation.py +++ b/cs2pov/automation.py @@ -447,40 +447,3 @@ def calibrate_tick_offset( return 0, log_position -def check_death_in_console( - console_log_path: Path, - player_slot: int, - last_position: int, -) -> tuple[bool, int]: - """Check console.log for player death (Shutdown prediction). - - Incrementally reads console.log looking for: - [Prediction] Shutdown prediction for player slot {player_slot} - - Args: - console_log_path: Path to CS2 console.log - player_slot: 0-based player slot index - last_position: File position to start reading from - - Returns: - (detected, new_position) - """ - if not console_log_path.exists(): - return False, last_position - - pattern = re.compile( - rf"\[Prediction\] Shutdown prediction for player slot {player_slot}\b" - ) - - try: - with open(console_log_path, 'r', errors='ignore') as f: - f.seek(last_position) - content = f.read() - new_position = f.tell() - - if pattern.search(content): - return True, new_position - - return False, new_position - except Exception: - return False, last_position diff --git a/cs2pov/cli.py b/cs2pov/cli.py index 09457e4..faa440d 100644 --- a/cs2pov/cli.py +++ b/cs2pov/cli.py @@ -18,7 +18,7 @@ from pathlib import Path from typing import Optional from . import __version__ -from .automation import send_key, check_demo_ended, wait_for_cs2_window, wait_for_demo_ready, parse_demo_end_info +from .automation import send_key, send_console_command, check_demo_ended, wait_for_cs2_window, wait_for_demo_ready, parse_demo_end_info from .loading import LoadingAnimation from .capture import FFmpegCapture, get_default_audio_monitor from .config import RecordingConfig, generate_recording_cfg @@ -31,7 +31,6 @@ from .settings import ( from .navigation import GotoTransition, NavigationState, recording_loop_tick_nav from .parser import DemoInfo, PlayerInfo, find_player, get_player_index, parse_demo from .preprocessor import DemoTimeline, preprocess_demo, get_trim_periods -from .trim import extract_death_periods # ============================================================================= @@ -44,8 +43,6 @@ class RecordingResult: success: bool video_path: Path console_log_path: Path - recording_start_time: float - player_slot: int exit_reason: str # "demo_ended", "timeout", "ffmpeg_stopped", "interrupted", "segments_complete" timeline: Optional[DemoTimeline] = None transitions: Optional[list[GotoTransition]] = None @@ -187,7 +184,7 @@ def record_demo( print(f" Total alive time: {total_alive:.1f}s") except Exception as e: print(f" Warning: Preprocessing failed: {e}") - print(f" Will fall back to console.log parsing for trim") + print(f" Trimming will be skipped (no timeline data)") timeline = None # Prepare directories @@ -241,7 +238,6 @@ def record_demo( cs2_process: Optional[CS2Process] = None ffmpeg: Optional[FFmpegCapture] = None - recording_start_time = 0.0 exit_reason = "unknown" try: @@ -264,6 +260,13 @@ def record_demo( else: print(" Warning: CS2 window not detected, continuing anyway") + # Determine tick-nav mode early (needed for startup sequence) + use_tick_nav = ( + tick_nav + and timeline is not None + and timeline.alive_segments + ) + # Wait for demo to be ready print(" Waiting for demo to load...") if wait_for_demo_ready(console_log_path, timeout=180): @@ -271,11 +274,13 @@ def record_demo( if verbose: print(" Waiting 20s before hiding demo UI...") time.sleep(20) - if window_id and send_key("shift+F2", display_str, window_id): - if verbose: - print(" Sent Shift+F2 to hide demo UI") - elif window_id: - print(" Warning: Failed to send Shift+F2 to hide demo UI") + # In tick-nav mode, defer Shift+F2 until after the seek (gototick resets UI state) + if not use_tick_nav: + if window_id and send_key("shift+F2", display_str, window_id): + if verbose: + print(" Sent Shift+F2 to hide demo UI") + elif window_id: + print(" Warning: Failed to send Shift+F2 to hide demo UI") else: print(" Warning: Demo ready state not detected, continuing anyway") @@ -289,6 +294,24 @@ def record_demo( else: print(" Warning: Could not detect audio device, recording video only") + # In tick-nav mode: pause, seek to first segment, then start FFmpeg + # This synchronizes the demo clock with the recording clock from frame 1 + if use_tick_nav and window_id: + first_tick = timeline.alive_segments[0].start_tick + if verbose: + print(f" Tick-nav: seeking to first segment start tick {first_tick}") + send_key("F7", display_str, window_id) # Pause demo + time.sleep(0.5) + send_console_command(f"demo_gototick {first_tick}", display_str, window_id) + time.sleep(2.0) # Wait for seek to complete + send_key("F5", display_str, window_id) # Re-lock spectator + # Send Shift+F2 after seek (gototick resets UI state, swallowing earlier sends) + if send_key("shift+F2", display_str, window_id): + if verbose: + print(" Sent Shift+F2 to hide demo UI") + else: + print(" Warning: Failed to send Shift+F2 to hide demo UI") + # Start FFmpeg capture ffmpeg = FFmpegCapture( display=display_str, @@ -304,20 +327,17 @@ def record_demo( else: print(f" FFmpeg capture started (video only)") - recording_start_time = time.time() - transitions = None + # Unpause after FFmpeg is rolling (tick-nav only) + if use_tick_nav and window_id: + send_key("F6", display_str, window_id) # Resume demo + if verbose: + print(" Tick-nav: demo unpaused, recording synchronized") - # Use tick-based navigation if enabled and timeline available - use_tick_nav = ( - tick_nav - and timeline is not None - and timeline.alive_segments - ) + transitions = None if use_tick_nav: nav_state = NavigationState( timeline=timeline, - player_slot=player_slot - 1, # Convert 1-based spec slot to 0-based console slot ) exit_reason, transitions = recording_loop_tick_nav( display=display_str, @@ -374,8 +394,6 @@ def record_demo( success=success, video_path=output_path, console_log_path=console_log_path, - recording_start_time=recording_start_time, - player_slot=player_index, exit_reason=exit_reason, timeline=timeline, transitions=transitions, @@ -385,8 +403,6 @@ def record_demo( def postprocess_video( video_path: Path, console_log_path: Path, - player_slot: int, - recording_start_time: float, verbose: bool = False, timeline: Optional[DemoTimeline] = None, startup_time_override: Optional[float] = None, @@ -403,6 +419,7 @@ def postprocess_video( 3. Extract and concatenate only the alive segments 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) @@ -533,52 +550,6 @@ def postprocess_video( if verbose: print(f" Keep: {video_start:.2f}s - {video_end:.2f}s ({video_end - video_start:.2f}s)") - # Fall back to console.log parsing (legacy method) - if not video_segments: - if timeline is not None: - print(" No alive segments found, falling back to console.log") - else: - print(" Using console.log parsing (legacy method)") - - if verbose: - print(f" Console log: {console_log_path}") - print(f" Player slot: {player_slot}") - print(f" Recording start: {recording_start_time}") - - if not console_log_path.exists(): - print(" Console log not found, skipping trim") - print(f"\nRecording saved: {video_path} ({raw_size_mb:.1f} MB)") - return video_path - - # Use legacy death period extraction - death_periods = extract_death_periods( - log_path=console_log_path, - player_slot=player_slot, - recording_start_time=recording_start_time, - verbose=verbose - ) - - if death_periods: - # Convert death periods to video segments (inverse) - try: - video_duration = get_video_duration(video_path) - except Exception as e: - print(f" Error getting video duration: {e}") - return video_path - - # Sort death periods and compute alive segments - death_periods_sorted = sorted(death_periods, key=lambda p: p.death_time) - current_pos = 0.0 - - for dp in death_periods_sorted: - if dp.death_time > current_pos: - video_segments.append((current_pos, dp.death_time)) - current_pos = max(current_pos, dp.respawn_time) - - # Add final segment - if current_pos < video_duration: - video_segments.append((current_pos, video_duration)) - # Execute trimming if video_segments: total_keep_time = sum(end - start for start, end in video_segments) @@ -846,13 +817,6 @@ Examples: trim_parser.add_argument("video", type=Path, help="Input video file") trim_parser.add_argument("-o", "--output", type=Path, help="Output file (default: adds _trimmed suffix)") - # Fallback options - trim_parser.add_argument("--console-log", type=Path, - help="Console.log file (fallback when demo unavailable)") - trim_parser.add_argument("--player-slot", type=int, - help="Player slot, 0-based (fallback)") - trim_parser.add_argument("--recording-start-time", type=float, - help="Recording start timestamp (fallback)") trim_parser.add_argument("--startup-time", type=float, help="Override startup time (seconds from video start to demo start)") @@ -1103,8 +1067,6 @@ def _run_single_pov(args) -> int: postprocess_video( video_path=result.video_path, console_log_path=result.console_log_path, - player_slot=result.player_slot, - recording_start_time=result.recording_start_time, verbose=args.verbose, timeline=result.timeline, transitions=result.transitions, @@ -1232,7 +1194,6 @@ def cmd_trim(args) -> int: try: demo_info = parse_demo(demo_path) player = find_player(demo_info, args.player) - player_slot = get_player_index(demo_info, player) except CS2POVError as e: print(f"Error: {e}", file=sys.stderr) return 1 @@ -1245,20 +1206,10 @@ def cmd_trim(args) -> int: print(f" {len(timeline.alive_segments)} alive segments to keep") except Exception as e: print(f"Warning: Could not preprocess demo: {e}") - print("Falling back to console.log method") - # Validate fallback parameters if needed if timeline is None: - if args.console_log is None: - print("Error: --console-log required when demo preprocessing fails", file=sys.stderr) - return 1 - if args.player_slot is None: - print("Error: --player-slot required when demo preprocessing fails", file=sys.stderr) - return 1 - if args.recording_start_time is None: - print("Error: --recording-start-time required when demo preprocessing fails", file=sys.stderr) - return 1 - player_slot = args.player_slot + print("Error: Demo preprocessing failed, cannot trim without timeline data", file=sys.stderr) + return 1 # Copy video to output path first (postprocess_video expects to rename) if video_path != output_path: @@ -1267,9 +1218,7 @@ def cmd_trim(args) -> int: # Run trimming postprocess_video( video_path=output_path, - console_log_path=args.console_log.resolve() if args.console_log else Path("/dev/null"), - player_slot=player_slot, - recording_start_time=args.recording_start_time or 0.0, + console_log_path=Path("/dev/null"), verbose=args.verbose, timeline=timeline, startup_time_override=args.startup_time, diff --git a/cs2pov/navigation.py b/cs2pov/navigation.py index b2462a3..fc07401 100644 --- a/cs2pov/navigation.py +++ b/cs2pov/navigation.py @@ -1,8 +1,9 @@ """Tick-based demo navigation - skip death periods in real-time. Instead of recording the full demo and trimming in post-processing, -this module detects deaths via console.log and uses demo_gototick to -skip to the next alive segment. Transition artifacts (the pause/seek/unpause +this module uses time-based segment tracking (from preprocessor alive +segments) to detect when a segment ends and uses demo_gototick to skip +to the next alive segment. Transition artifacts (the pause/seek/unpause period) are trimmed in lightweight post-processing. """ @@ -12,7 +13,6 @@ from pathlib import Path from typing import Optional from .automation import ( - check_death_in_console, check_demo_ended_tick_aware, calibrate_tick_offset, find_cs2_window, @@ -41,7 +41,6 @@ class GotoTransition: class NavigationState: """State for tick-based navigation during recording.""" timeline: DemoTimeline - player_slot: int # 0-based, for console.log pattern matching tick_offset: int = 0 # Calibrated offset (actual - expected) current_segment_index: int = 0 transitions: list[GotoTransition] = field(default_factory=list) @@ -166,8 +165,8 @@ def recording_loop_tick_nav( """Recording loop with tick-based navigation (skips death periods). Replaces the standard recording_loop when --tick-nav is enabled. - Detects deaths via console.log and uses demo_gototick to skip to - the next alive segment. + Uses time-based segment tracking to detect when an alive segment ends + and demo_gototick to skip to the next alive segment. Args: display: X display string @@ -186,7 +185,6 @@ def recording_loop_tick_nav( state.recording_start_time = start_time last_spec_lock = 0.0 log_position = 0 - death_log_position = 0 window_id = None window_found_logged = False last_status_time = start_time @@ -246,68 +244,43 @@ def recording_loop_tick_nav( print(" Demo end detected in console.log") return "demo_ended", state.transitions - # Check for death (only if we have a window to navigate with) - if window_id: - death_triggered = False - death_detected, death_log_position = check_death_in_console( - console_log_path, state.player_slot, death_log_position - ) - if death_detected: + # Time-based segment end: skip forward when alive segment duration elapsed + if window_id and state.current_segment_index < len(state.timeline.alive_segments): + current_seg = state.timeline.alive_segments[state.current_segment_index] + + # Determine when to trigger the skip + target_end_time = current_seg.end_time + if current_seg.reason_ended == "round_end": + # Extend to next round's prestart (freeze time begins), + # minus 0.2s to avoid bleeding into the next round + next_idx = state.current_segment_index + 1 + if next_idx < len(state.timeline.alive_segments): + next_seg = state.timeline.alive_segments[next_idx] + for r in state.timeline.rounds: + if r.round_num == next_seg.round_num and r.prestart_time is not None: + target_end_time = r.prestart_time - 0.2 + break + + expected_duration = target_end_time - current_seg.start_time + + segment_elapsed = time.time() - state.segment_start_wall_time + if segment_elapsed >= expected_duration: if verbose: - print(f" Death detected (segment {state.current_segment_index + 1})") - time.sleep(0.9) # Brief delay to show death context - has_more, death_log_position = handle_death( + print(f" Segment {state.current_segment_index + 1} timer expired " + f"({expected_duration:.1f}s elapsed, reason: {current_seg.reason_ended})") + if current_seg.reason_ended == "death": + time.sleep(0.9) # Brief delay to show death context + has_more, log_position = handle_death( state, display, window_id, console_log_path, - death_log_position, verbose + log_position, verbose ) # Advance past any messages generated during gototick seek eof = console_log_path.stat().st_size - death_log_position = eof log_position = eof state.segment_start_wall_time = time.time() if not has_more: print(" All alive segments complete") return "segments_complete", state.transitions - death_triggered = True - - # Time-based segment end: skip forward when alive segment duration elapsed - if not death_triggered and state.current_segment_index < len(state.timeline.alive_segments): - current_seg = state.timeline.alive_segments[state.current_segment_index] - - # Determine when to trigger the skip - target_end_time = current_seg.end_time - if current_seg.reason_ended == "round_end": - # Extend to next round's prestart (freeze time begins), - # minus 0.1s to avoid bleeding into the next round - next_idx = state.current_segment_index + 1 - if next_idx < len(state.timeline.alive_segments): - next_seg = state.timeline.alive_segments[next_idx] - for r in state.timeline.rounds: - if r.round_num == next_seg.round_num and r.prestart_time is not None: - target_end_time = r.prestart_time - 0.2 - break - - expected_duration = target_end_time - current_seg.start_time - - segment_elapsed = time.time() - state.segment_start_wall_time - if segment_elapsed >= expected_duration: - if verbose: - print(f" Segment {state.current_segment_index + 1} timer expired " - f"({expected_duration:.1f}s elapsed, reason: {current_seg.reason_ended})") - if current_seg.reason_ended == "death": - time.sleep(0.9) # Brief delay to show death context - has_more, death_log_position = handle_death( - state, display, window_id, console_log_path, - death_log_position, verbose - ) - # Advance past any messages generated during gototick seek - eof = console_log_path.stat().st_size - death_log_position = eof - log_position = eof - state.segment_start_wall_time = time.time() - if not has_more: - print(" All alive segments complete") - return "segments_complete", state.transitions # Send spec_lock (F5) every 3 seconds if elapsed - last_spec_lock >= 3.0: diff --git a/cs2pov/trim.py b/cs2pov/trim.py index 0edda07..ea2abdd 100644 --- a/cs2pov/trim.py +++ b/cs2pov/trim.py @@ -1,170 +1,11 @@ -"""Post-processing: trim death periods from recorded video.""" +"""Post-processing: trim segments from recorded video.""" -import re import subprocess import tempfile -from dataclasses import dataclass -from datetime import datetime from pathlib import Path -from typing import Optional, TYPE_CHECKING from .exceptions import CaptureError -if TYPE_CHECKING: - from .preprocessor import DemoTimeline - - -@dataclass -class DeathPeriod: - """A period when the player was dead.""" - death_time: float # Video timestamp in seconds - respawn_time: float # Video timestamp in seconds - - @property - def duration(self) -> float: - return self.respawn_time - self.death_time - - -@dataclass -class TrimPeriod: - """A period to trim from the video.""" - start_time: float # Video timestamp in seconds - end_time: float # Video timestamp in seconds - - @property - def duration(self) -> float: - return self.end_time - self.start_time - - -def parse_log_timestamp(timestamp_str: str, reference_date: Optional[datetime] = None) -> datetime: - """Parse a console.log timestamp in MM/DD HH:mm:ss format. - - Args: - timestamp_str: Timestamp string like "01/28 15:30:45" - reference_date: Reference date for year (defaults to current year) - - Returns: - datetime object - """ - if reference_date is None: - reference_date = datetime.now() - - # Parse MM/DD HH:mm:ss - parsed = datetime.strptime(timestamp_str, "%m/%d %H:%M:%S") - # Add the year from reference date - return parsed.replace(year=reference_date.year) - - -def extract_death_periods( - log_path: Path, - player_slot: int, - recording_start_time: float, - verbose: bool = False -) -> list[DeathPeriod]: - """Extract death periods from console.log. - - Finds periods between "Shutdown prediction for player slot X" and - "Added TrueView prediction for player slot X" lines. - - Also trims from the start of the recording until the player's POV is - first selected (first TrueView prediction). - - Args: - log_path: Path to console.log file - player_slot: Player slot index (0-based) - recording_start_time: Unix timestamp when recording started - verbose: Print debug output - - Returns: - List of DeathPeriod objects with video-relative timestamps - """ - if not log_path.exists(): - if verbose: - print(f" [Trim] Console log not found: {log_path}") - return [] - - # Patterns for death and respawn - # Format: MM/DD HH:mm:ss [Prediction] Shutdown prediction for player slot X. ... - death_pattern = re.compile( - rf"^(\d{{2}}/\d{{2}} \d{{2}}:\d{{2}}:\d{{2}}).*" - rf"\[Prediction\] Shutdown prediction for player slot {player_slot}\b" - ) - respawn_pattern = re.compile( - rf"^(\d{{2}}/\d{{2}} \d{{2}}:\d{{2}}:\d{{2}}).*" - rf"\[Prediction\] Added TrueView prediction for player slot {player_slot}\b" - ) - - # Reference date for parsing (use recording start time) - reference_date = datetime.fromtimestamp(recording_start_time) - - death_periods: list[DeathPeriod] = [] - pending_death_time: Optional[float] = None - first_pov_time: Optional[float] = None # Track first time POV is selected - - if verbose: - print(f" [Trim] Parsing log: {log_path}") - print(f" [Trim] Looking for player slot {player_slot}") - - with open(log_path, 'r', errors='ignore') as f: - for line in f: - # Check for death - death_match = death_pattern.match(line) - if death_match: - timestamp_str = death_match.group(1) - try: - log_time = parse_log_timestamp(timestamp_str, reference_date) - video_time = log_time.timestamp() - recording_start_time - if video_time >= 0: # Only consider events after recording started - pending_death_time = video_time - if verbose: - print(f" [Trim] Death at video time {video_time:.2f}s") - except ValueError as e: - if verbose: - print(f" [Trim] Failed to parse timestamp: {timestamp_str}: {e}") - continue - - # Check for respawn/POV selection - respawn_match = respawn_pattern.match(line) - if respawn_match: - timestamp_str = respawn_match.group(1) - try: - log_time = parse_log_timestamp(timestamp_str, reference_date) - video_time = log_time.timestamp() - recording_start_time - - # Track first POV selection time - if first_pov_time is None and video_time >= 0: - first_pov_time = video_time - if verbose: - print(f" [Trim] First POV selection at video time {video_time:.2f}s") - - # Handle respawn after death - if pending_death_time is not None and video_time > pending_death_time: - death_periods.append(DeathPeriod( - death_time=pending_death_time, - respawn_time=video_time - )) - if verbose: - print(f" [Trim] Respawn at video time {video_time:.2f}s " - f"(dead for {video_time - pending_death_time:.2f}s)") - pending_death_time = None - except ValueError as e: - if verbose: - print(f" [Trim] Failed to parse timestamp: {timestamp_str}: {e}") - - # Add initial period from start until first POV selection - if first_pov_time is not None and first_pov_time > 0.5: # Only if > 0.5s to avoid tiny trims - death_periods.insert(0, DeathPeriod( - death_time=0.0, - respawn_time=first_pov_time - )) - if verbose: - print(f" [Trim] Adding start trim: 0.0s - {first_pov_time:.2f}s") - - if verbose: - print(f" [Trim] Found {len(death_periods)} periods to trim") - - return death_periods - def get_video_duration(video_path: Path) -> float: """Get video duration in seconds using ffprobe. @@ -200,233 +41,6 @@ def get_video_duration(video_path: Path) -> float: raise CaptureError(f"Failed to get video duration: {e}") -def trim_death_periods( - input_path: Path, - output_path: Path, - death_periods: list[DeathPeriod], - verbose: bool = False -) -> bool: - """Trim death periods from video using FFmpeg concat demuxer. - - Creates segments for "alive" periods and concatenates them. - - Args: - input_path: Path to input video - output_path: Path for output video - death_periods: List of death periods to remove - verbose: Print debug output - - Returns: - True if trimming was performed, False if no trimming needed - """ - if not death_periods: - if verbose: - print(" [Trim] No death periods to trim") - return False - - # Get video duration - try: - video_duration = get_video_duration(input_path) - except CaptureError as e: - if verbose: - print(f" [Trim] Failed to get video duration: {e}") - return False - - if verbose: - print(f" [Trim] Video duration: {video_duration:.2f}s") - - # Sort death periods by start time - death_periods = sorted(death_periods, key=lambda p: p.death_time) - - # Calculate "alive" segments (inverse of death periods) - alive_segments: list[tuple[float, float]] = [] - current_pos = 0.0 - - for period in death_periods: - # Skip if death starts before current position (overlapping) - if period.death_time <= current_pos: - current_pos = max(current_pos, period.respawn_time) - continue - - # Add alive segment before this death - if period.death_time > current_pos: - alive_segments.append((current_pos, period.death_time)) - - current_pos = period.respawn_time - - # Add final segment after last death - if current_pos < video_duration: - alive_segments.append((current_pos, video_duration)) - - if not alive_segments: - if verbose: - print(" [Trim] No alive segments found") - return False - - if verbose: - print(f" [Trim] Found {len(alive_segments)} alive segments") - total_alive = sum(end - start for start, end in alive_segments) - total_dead = sum(p.duration for p in death_periods) - print(f" [Trim] Total alive time: {total_alive:.2f}s, dead time: {total_dead:.2f}s") - - # Create concat file and segment files in temp directory - with tempfile.TemporaryDirectory() as tmpdir: - tmpdir_path = Path(tmpdir) - concat_file = tmpdir_path / "concat.txt" - segment_paths: list[Path] = [] - - # Extract each alive segment - for i, (start, end) in enumerate(alive_segments): - segment_path = tmpdir_path / f"segment_{i:03d}.mp4" - segment_paths.append(segment_path) - - duration = end - start - if verbose: - print(f" [Trim] Extracting segment {i}: {start:.2f}s - {end:.2f}s ({duration:.2f}s)") - - cmd = [ - "ffmpeg", - "-y", - "-ss", str(start), - "-i", str(input_path), - "-t", str(duration), - "-c", "copy", # No re-encoding - "-avoid_negative_ts", "make_zero", - str(segment_path) - ] - - try: - result = subprocess.run( - cmd, - capture_output=True, - text=True, - timeout=300 # 5 min per segment - ) - if result.returncode != 0: - if verbose: - print(f" [Trim] Segment extraction failed: {result.stderr}") - return False - except subprocess.TimeoutExpired: - if verbose: - print(f" [Trim] Segment extraction timed out") - return False - - # Create concat file - with open(concat_file, 'w') as f: - for segment_path in segment_paths: - # Escape single quotes in path - escaped_path = str(segment_path).replace("'", "'\\''") - f.write(f"file '{escaped_path}'\n") - - if verbose: - print(f" [Trim] Concatenating {len(segment_paths)} segments") - - # Concatenate segments - cmd = [ - "ffmpeg", - "-y", - "-f", "concat", - "-safe", "0", - "-i", str(concat_file), - "-c", "copy", - str(output_path) - ] - - try: - result = subprocess.run( - cmd, - capture_output=True, - text=True, - timeout=600 # 10 min for concat - ) - if result.returncode != 0: - if verbose: - print(f" [Trim] Concatenation failed: {result.stderr}") - return False - except subprocess.TimeoutExpired: - if verbose: - print(f" [Trim] Concatenation timed out") - return False - - if verbose: - if output_path.exists(): - output_duration = get_video_duration(output_path) - print(f" [Trim] Output video: {output_duration:.2f}s") - - return True - - -def get_trim_periods_from_timeline( - timeline: "DemoTimeline", - first_spawn_video_time: float, - verbose: bool = False, -) -> list[TrimPeriod]: - """Get trim periods from preprocessed timeline data. - - Uses the timeline's death/spawn events to compute trim periods. - The first_spawn_video_time parameter tells us when the first spawn - occurred in the video, allowing us to align demo times with video times. - - Args: - timeline: Preprocessed DemoTimeline with death/spawn events - first_spawn_video_time: Video timestamp when first spawn occurred - verbose: Print debug output - - Returns: - List of TrimPeriod objects with video-relative timestamps - """ - from .preprocessor import get_trim_periods_for_video - - trim_periods = [] - - # Get demo-relative trim periods and convert to video-relative - video_periods = get_trim_periods_for_video(timeline, first_spawn_video_time) - - for start, end in video_periods: - trim_periods.append(TrimPeriod(start_time=start, end_time=end)) - if verbose: - print(f" [Trim] Period: {start:.2f}s - {end:.2f}s ({end - start:.2f}s)") - - if verbose: - print(f" [Trim] Found {len(trim_periods)} periods to trim from timeline") - - return trim_periods - - -def trim_video_with_periods( - input_path: Path, - output_path: Path, - trim_periods: list[TrimPeriod], - verbose: bool = False, -) -> bool: - """Trim specified periods from video using FFmpeg concat demuxer. - - This is a generalized version that takes TrimPeriod objects directly. - Creates segments for periods to keep and concatenates them. - - Args: - input_path: Path to input video - output_path: Path for output video - trim_periods: List of periods to remove from video - verbose: Print debug output - - Returns: - True if trimming was performed, False if no trimming needed - """ - if not trim_periods: - if verbose: - print(" [Trim] No periods to trim") - return False - - # Convert TrimPeriod to DeathPeriod for compatibility with existing logic - death_periods = [ - DeathPeriod(death_time=p.start_time, respawn_time=p.end_time) - for p in trim_periods - ] - - return trim_death_periods(input_path, output_path, death_periods, verbose) - - def trim_goto_transitions( input_path: Path, output_path: Path, |