summaryrefslogtreecommitdiff
diff options
context:
space:
mode:
authorSchark <jordan@schark.online>2026-02-27 20:31:21 -0500
committerSchark <jordan@schark.online>2026-02-27 20:31:21 -0500
commite9619b313bd78cabd397a844f44e84f36e4a1458 (patch)
tree6fc087340d7b88f31a708aa33cb9d72f7f85de22
parent7c42e113ab324896dedda13943996cc32c82efec (diff)
parentd0cfb2173c0f9a782e7f9bbef782cd4ee440b0aa (diff)
downloadcs2pov-e9619b313bd78cabd397a844f44e84f36e4a1458.tar.gz
cs2pov-e9619b313bd78cabd397a844f44e84f36e4a1458.zip
Merge branch 'master' into feat/config-file
Diffstat (limited to '')
-rw-r--r--cs2pov/automation.py37
-rw-r--r--cs2pov/cli.py139
-rw-r--r--cs2pov/navigation.py91
-rw-r--r--cs2pov/trim.py388
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,