summaryrefslogtreecommitdiff
diff options
context:
space:
mode:
authorSchark <jordan@schark.online>2026-03-04 12:43:08 -0500
committerSchark <jordan@schark.online>2026-03-04 12:43:08 -0500
commit5c99ce31990c02a7b00108b6a8766b17f15660ed (patch)
treee2877b094197189add3cf9079626de24f74f52de
parentc41e858565fac42480222b72863cb054ae63c39e (diff)
downloadcs2pov-5c99ce31990c02a7b00108b6a8766b17f15660ed.tar.gz
cs2pov-5c99ce31990c02a7b00108b6a8766b17f15660ed.zip
Testing new comms overlay method
-rw-r--r--cs2pov/cli.py119
-rw-r--r--cs2pov/cli.py.bk1621
2 files changed, 1713 insertions, 27 deletions
diff --git a/cs2pov/cli.py b/cs2pov/cli.py
index fa43a62..6463489 100644
--- a/cs2pov/cli.py
+++ b/cs2pov/cli.py
@@ -1470,39 +1470,104 @@ def cmd_comms(args) -> int:
return 1
from .comms import apply_comms_overlay
- from .trim import get_video_duration, compute_keep_segments
+ from .trim import get_video_duration, compute_keep_segments, load_transitions
# 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.
+ # For tick-nav mode: use GotoTransitions sidecar for actual video durations
+ # (includes polling overrun), avoiding cumulative drift.
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
+ if is_trimmed and timeline and timeline.alive_segments:
+ if args.tick_nav:
+ # Try loading transitions sidecar for actual segment durations.
+ # The sidecar is saved as {output_stem}_transitions.json during recording,
+ # where output_stem is the base name (e.g., "recording"). Try multiple
+ # candidate paths since the user may pass the trimmed or raw video.
+ try:
+ transitions = None
+ # Try: video_path itself, then strip _raw suffix, then add _raw suffix
+ candidates = [video_path]
+ stem = video_path.stem
+ if stem.endswith("_raw"):
+ # User passed raw video — sidecar is named after the base stem
+ base = video_path.parent / f"{stem[:-4]}{video_path.suffix}"
+ candidates.insert(0, base)
+ else:
+ # User passed trimmed video — sidecar might also be next to raw
+ raw_path = video_path.parent / f"{stem}_raw{video_path.suffix}"
+ candidates.append(raw_path)
+
+ for candidate in candidates:
+ transitions = load_transitions(candidate)
+ if transitions:
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 transitions:
+ sorted_transitions = sorted(transitions, key=lambda t: t.pause_video_time)
+ num_alive = len(timeline.alive_segments)
+ num_trans = len(sorted_transitions)
+
+ # Build per-alive-segment durations from transition timestamps.
+ # transition[i] fires at end of alive_segment[i], so:
+ # alive[i] video_dur = transition[i].pause - transition[i-1].unpause
+ # For alive[0]: no preceding transition (startup contaminates),
+ # use demo-time duration as fallback (~0.25s overrun, acceptable).
+ per_segment_durations: list[tuple[float, float]] = []
+ for i in range(num_alive):
+ if i == 0 or i > num_trans or i - 1 >= num_trans:
+ # Fallback: demo-time duration with tick_nav extensions
+ seg = timeline.alive_segments[i]
+ dur = seg.end_time - seg.start_time
+ if seg.reason_ended == "death":
+ dur += 0.9
+ elif seg.reason_ended == "round_end":
+ next_seg = timeline.alive_segments[i + 1] if i + 1 < num_alive else None
+ if next_seg:
+ for r in timeline.rounds:
+ if r.round_num == next_seg.round_num and r.prestart_time is not None:
+ dur = r.prestart_time - 0.2 - seg.start_time
+ break
+ per_segment_durations.append((0.0, dur))
+ else:
+ dur = sorted_transitions[i].pause_video_time - sorted_transitions[i - 1].unpause_video_time
+ per_segment_durations.append((0.0, dur))
+
+ if per_segment_durations:
+ keep_segments = per_segment_durations
+ if args.verbose:
+ durations = [end - start for start, end in keep_segments]
+ print(f" Using {len(keep_segments)} per-segment durations from transitions")
+ print(f" Total duration: {sum(durations):.1f}s")
+ print(f" Segment 0 (demo-time fallback): {durations[0]:.1f}s")
+ print(f" Segments 1+ (transition-based): avg {sum(durations[1:])/max(len(durations)-1,1):.1f}s")
+ elif args.verbose:
+ print(" No transitions sidecar found, falling back to tick-nav extension logic")
+ except Exception as e:
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" Warning: Could not load transitions: {e}, falling back to tick-nav extension logic")
+ else:
+ 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}")
print(f"Overlaying comms: {comms_path.name} → {output_path.name}")
success = apply_comms_overlay(
diff --git a/cs2pov/cli.py.bk b/cs2pov/cli.py.bk
new file mode 100644
index 0000000..b44f297
--- /dev/null
+++ b/cs2pov/cli.py.bk
@@ -0,0 +1,1621 @@
+#!/usr/bin/env python3
+"""CS2 POV Recorder - Record player POV from CS2 demo files.
+
+Subcommand-based CLI:
+ pov - Full pipeline (record + trim)
+ info - Show demo information
+ record - Raw recording only
+ trim - Post-process existing video
+"""
+
+import argparse
+import json
+import shutil
+import sys
+import time
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Optional
+
+from . import __version__
+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
+from .exceptions import CS2POVError
+from .game import CS2Process, find_cs2_path, get_cfg_dir, get_demo_dir
+from .settings import (
+ ConfigError, find_config, load_config, resolve_job, resolve_paths,
+ merge_args_with_config, generate_default_config, HARDCODED_DEFAULTS,
+)
+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
+
+
+# =============================================================================
+# Data Classes
+# =============================================================================
+
+@dataclass
+class RecordingResult:
+ """Result of recording phase with metadata for post-processing."""
+ success: bool
+ video_path: Path
+ console_log_path: Path
+ exit_reason: str # "demo_ended", "timeout", "ffmpeg_stopped", "interrupted", "segments_complete"
+ timeline: Optional[DemoTimeline] = None
+ transitions: Optional[list[GotoTransition]] = None
+
+
+# =============================================================================
+# Shared Utilities
+# =============================================================================
+
+def check_dependencies() -> list[str]:
+ """Check for required system dependencies."""
+ missing = []
+ if not shutil.which("ffmpeg"):
+ missing.append("FFmpeg - Install with: apt install ffmpeg")
+ if not shutil.which("xdotool"):
+ missing.append("xdotool - Install with: apt install xdotool")
+ return missing
+
+
+def parse_resolution(resolution_str: str) -> tuple[int, int]:
+ """Parse resolution string like '1920x1080' to tuple."""
+ try:
+ width, height = resolution_str.lower().split("x")
+ return (int(width), int(height))
+ except ValueError:
+ raise ValueError(f"Invalid resolution format: {resolution_str}. Use WxH (e.g., 1920x1080)")
+
+
+# =============================================================================
+# Recording Functions
+# =============================================================================
+
+def recording_loop(
+ display: str,
+ console_log_path: Path,
+ cs2_process: CS2Process,
+ ffmpeg: FFmpegCapture,
+ timeout: float,
+ verbose: bool = False,
+) -> str:
+ """Main recording loop - single-threaded, blocking.
+
+ Returns:
+ Exit reason: "demo_ended", "cs2_exited", "timeout", "ffmpeg_stopped"
+ """
+ start_time = time.time()
+ last_spec_lock = 0
+ log_position = 0
+ window_id = None
+ window_found_logged = False
+ last_status_time = start_time
+
+ print(" Recording loop started")
+
+ while True:
+ elapsed = time.time() - start_time
+
+ if elapsed > timeout:
+ print(f" Timeout reached ({timeout/60:.1f} min)")
+ return "timeout"
+
+ if not cs2_process.is_running():
+ print(" CS2 exited")
+ return "cs2_exited"
+
+ if not ffmpeg.is_running():
+ print(" FFmpeg stopped unexpectedly")
+ return "ffmpeg_stopped"
+
+ demo_ended, log_position = check_demo_ended(console_log_path, log_position)
+ if demo_ended:
+ print(" Demo end detected in console.log")
+ return "demo_ended"
+
+ # Send spec_lock (F5) every 3 seconds
+ if elapsed - last_spec_lock >= 3.0:
+ if window_id is None:
+ from .automation import find_cs2_window
+ window_id = find_cs2_window(display)
+ if window_id and not window_found_logged:
+ if verbose:
+ print(f" CS2 window found: {window_id}")
+ window_found_logged = True
+
+ if window_id:
+ send_key("F5", display, window_id)
+ if verbose:
+ print(f" Sending F5 (spec_lock)")
+ last_spec_lock = elapsed
+
+ if elapsed - last_status_time >= 60:
+ print(f" Still recording... ({elapsed/60:.1f} min)")
+ last_status_time = elapsed
+
+ time.sleep(0.5)
+
+
+def record_demo(
+ demo_path: Path,
+ player_identifier: str,
+ output_path: Path,
+ resolution: tuple[int, int] = (1920, 1080),
+ framerate: int = 60,
+ hide_hud: bool = True,
+ display_num: int = 0,
+ verbose: bool = False,
+ cs2_path_override: Path | None = None,
+ enable_audio: bool = True,
+ audio_device: str | None = None,
+ tick_nav: bool = False,
+) -> RecordingResult:
+ """Record a player's POV from a demo file."""
+ # Find CS2 installation
+ print("Finding CS2 installation...")
+ cs2_path = find_cs2_path(cs2_path_override)
+ print(f" Found: {cs2_path}")
+
+ # Parse demo
+ print(f"Parsing demo: {demo_path.name}")
+ demo_info = parse_demo(demo_path)
+ print(f" Map: {demo_info.map_name}, {len(demo_info.players)} players")
+ print(f" Ticks: {demo_info.total_ticks}, Rate: {demo_info.tick_rate}/s")
+
+ # Find target player
+ player = find_player(demo_info, player_identifier)
+ player_index = get_player_index(demo_info, player)
+ # player_slot is user_id + 1 for spec_player command
+ player_slot = (player.user_id + 1) if player.user_id is not None else (player_index + 1)
+ print(f"Recording POV: {player.name} (SteamID: {player.steamid}, slot: {player_slot})")
+
+ # Preprocess demo for timeline data
+ print("Preprocessing demo for timeline data...")
+ try:
+ timeline = preprocess_demo(demo_path, player.steamid, player.name)
+ print(f" Found {len(timeline.deaths)} deaths, {len(timeline.rounds)} rounds")
+ print(f" {len(timeline.alive_segments)} alive segments to keep")
+ if verbose and timeline.alive_segments:
+ total_alive = sum(seg.duration_seconds for seg in timeline.alive_segments)
+ print(f" Total alive time: {total_alive:.1f}s")
+ except Exception as e:
+ print(f" Warning: Preprocessing failed: {e}")
+ print(f" Trimming will be skipped (no timeline data)")
+ timeline = None
+
+ # Prepare directories
+ demo_dir = get_demo_dir(cs2_path)
+ demo_dir.mkdir(parents=True, exist_ok=True)
+ cfg_dir = get_cfg_dir(cs2_path)
+ cfg_dir.mkdir(parents=True, exist_ok=True)
+
+ # Copy demo to CS2 replays directory if needed
+ target_demo = demo_dir / demo_path.name
+ if not target_demo.exists():
+ print(f"Copying demo to: {target_demo}")
+ shutil.copy2(demo_path, target_demo)
+ elif not target_demo.samefile(demo_path):
+ print(f"Updating demo at: {target_demo}")
+ shutil.copy2(demo_path, target_demo)
+
+ # Generate recording config
+ cfg_path = cfg_dir / "cs2pov_recording.cfg"
+ config = RecordingConfig(
+ demo_name=demo_path.stem,
+ player_index=player_index,
+ player_name=player.name,
+ player_steamid=player.steamid,
+ player_slot=player_slot,
+ resolution=resolution,
+ hide_hud=hide_hud,
+ tick_navigation=tick_nav,
+ )
+ generate_recording_cfg(config, cfg_path)
+ print(f"Generated config: {cfg_path.name}")
+
+ # Ensure output directory exists
+ output_path.parent.mkdir(parents=True, exist_ok=True)
+
+ # Setup paths
+ display_str = f":{display_num}"
+ console_log_path = cs2_path / "game/csgo/console.log"
+ cs2_log_path = output_path.parent / f"{output_path.stem}_cs2.log"
+
+ # Calculate timeout
+ if demo_info.tick_rate > 0 and demo_info.total_ticks > 0:
+ estimated_duration = demo_info.total_ticks / demo_info.tick_rate
+ else:
+ estimated_duration = 3600
+ timeout = estimated_duration + 600
+
+ print(f"Starting recording (timeout: {timeout/60:.1f} min)...")
+ print(f" Display: {display_str}")
+ print(f" Output: {output_path}")
+
+ cs2_process: Optional[CS2Process] = None
+ ffmpeg: Optional[FFmpegCapture] = None
+ exit_reason = "unknown"
+
+ try:
+ # Delete old console.log
+ if console_log_path.exists():
+ console_log_path.unlink()
+ if verbose:
+ print(f" Deleted old console.log")
+
+ # Launch CS2
+ cs2_process = CS2Process(cs2_path, display_str, log_path=cs2_log_path)
+ cs2_process.launch("cs2pov_recording.cfg")
+ print(" CS2 launching via Steam...")
+
+ # Wait for CS2 window
+ print(" Waiting for CS2 window...")
+ window_id = wait_for_cs2_window(display_str, timeout=120)
+ if window_id:
+ print(f" CS2 window ready")
+ 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):
+ print(" Demo ready")
+ if verbose:
+ print(" Waiting 20s before hiding demo UI...")
+ time.sleep(20)
+ # 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")
+
+ # Determine audio source
+ audio_source = None
+ if enable_audio:
+ audio_source = audio_device or get_default_audio_monitor()
+ if audio_source:
+ if verbose:
+ print(f" Audio device: {audio_source}")
+ 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,
+ output_path=output_path,
+ resolution=resolution,
+ framerate=framerate,
+ audio_device=audio_source if enable_audio else None,
+ enable_audio=enable_audio and audio_source is not None,
+ )
+ ffmpeg.start()
+ if ffmpeg.enable_audio:
+ print(f" FFmpeg capture started (video + audio)")
+ else:
+ print(f" FFmpeg capture started (video only)")
+
+ # 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")
+
+ transitions = None
+
+ if use_tick_nav:
+ nav_state = NavigationState(
+ timeline=timeline,
+ )
+ exit_reason, transitions = recording_loop_tick_nav(
+ display=display_str,
+ console_log_path=console_log_path,
+ cs2_process=cs2_process,
+ ffmpeg=ffmpeg,
+ state=nav_state,
+ timeout=timeout,
+ verbose=verbose,
+ )
+ else:
+ if tick_nav and (timeline is None or not timeline.alive_segments):
+ print(" Warning: --tick-nav requires alive segments, falling back to standard recording")
+ exit_reason = recording_loop(
+ display=display_str,
+ console_log_path=console_log_path,
+ cs2_process=cs2_process,
+ ffmpeg=ffmpeg,
+ timeout=timeout,
+ verbose=verbose,
+ )
+
+ except KeyboardInterrupt:
+ print("\n Recording interrupted by user")
+ exit_reason = "interrupted"
+
+ finally:
+ print("\nCleaning up...")
+
+ if ffmpeg:
+ print(" Stopping FFmpeg...")
+ graceful = ffmpeg.stop(timeout=15)
+ if graceful:
+ print(" FFmpeg stopped gracefully")
+ else:
+ print(" FFmpeg force-stopped")
+ if ffmpeg.stderr_path and ffmpeg.stderr_path.exists():
+ print(f" FFmpeg log: {ffmpeg.stderr_path}")
+
+ if cs2_process and cs2_process.is_running():
+ print(" Terminating CS2...")
+ cs2_process.terminate()
+
+ print(" Cleanup complete")
+
+ if console_log_path.exists():
+ saved_log_path = output_path.parent / f"console_{demo_path.stem}.log"
+ shutil.copy2(console_log_path, saved_log_path)
+ 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,
+ video_path=output_path,
+ console_log_path=console_log_path,
+ exit_reason=exit_reason,
+ timeline=timeline,
+ transitions=transitions,
+ )
+
+
+def postprocess_video(
+ video_path: Path,
+ console_log_path: Path,
+ verbose: bool = False,
+ timeline: Optional[DemoTimeline] = None,
+ startup_time_override: Optional[float] = None,
+ transitions: Optional[list[GotoTransition]] = None,
+) -> 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
+ transition-based trimming instead of the full alive-segment approach.
+
+ Otherwise uses the standard approach:
+ 1. Calculate startup_time = video_duration - demo_duration
+ 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, compute_keep_segments, load_transitions
+
+ if not video_path.exists():
+ print(f"Error: Video file not found: {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...")
+ raw_path = video_path.parent / f"{video_path.stem}_raw{video_path.suffix}"
+ video_path.rename(raw_path)
+ print(f" Raw recording moved to: {raw_path.name}")
+
+ keep_segments = trim_goto_transitions(
+ input_path=raw_path,
+ output_path=video_path,
+ transitions=transitions,
+ verbose=verbose,
+ )
+
+ 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, keep_segments
+
+ print("\nPost-processing: calculating segments to keep...")
+
+ video_segments: list[tuple[float, float]] = []
+
+ # Prefer timeline data with alive_segments
+ if timeline is not None and timeline.alive_segments:
+ print(" Using alive segments from timeline (demoparser2)")
+ if verbose:
+ print(f" Alive segments: {len(timeline.alive_segments)}")
+ print(f" Deaths: {len(timeline.deaths)}, Rounds: {len(timeline.rounds)}")
+ print(f" Tickrate: {timeline.tickrate}, Total ticks: {timeline.total_ticks}")
+ for i, seg in enumerate(timeline.alive_segments):
+ print(f" Segment {i+1}: R{seg.round_num} {seg.start_time:.2f}s - {seg.end_time:.2f}s "
+ f"({seg.duration_seconds:.2f}s, ended: {seg.reason_ended})")
+
+ # Step 1: Get video duration
+ try:
+ video_duration = get_video_duration(video_path)
+ 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, None
+
+ # Step 2: Get demo duration for startup_time calculation
+ demo_duration = 0.0
+
+ # Method 1: From console.log demo end marker (most accurate)
+ demo_end_info = parse_demo_end_info(console_log_path)
+ if demo_end_info and timeline.tickrate > 0:
+ demo_duration = demo_end_info.tick / timeline.tickrate
+ if verbose:
+ print(f" Demo duration source: console.log end marker")
+
+ # Method 2: From timeline total_duration (from demo header)
+ if demo_duration == 0.0 and timeline.total_duration > 0:
+ demo_duration = timeline.total_duration
+ if verbose:
+ print(f" Demo duration source: demo header")
+
+ # Method 3: From last round end (more accurate than alive segments)
+ if demo_duration == 0.0 and timeline.rounds:
+ # Find the last round with an end_tick
+ for r in reversed(timeline.rounds):
+ if r.end_tick is not None:
+ demo_duration = r.end_time + 5.0 # Small buffer after round ends
+ if verbose:
+ print(f" Demo duration source: last round end + 5s")
+ break
+
+ # Method 4: From last alive segment end (fallback)
+ if demo_duration == 0.0 and timeline.alive_segments:
+ last_segment = timeline.alive_segments[-1]
+ demo_duration = last_segment.end_time + 5.0 # Small buffer
+ if verbose:
+ print(f" Demo duration source: last alive segment + 5s")
+
+ if verbose:
+ print(f" Video duration: {video_duration:.2f}s")
+ print(f" Demo duration: {demo_duration:.2f}s")
+
+ 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, None
+
+ # Step 3: Calculate startup time and convert alive segments to video time
+ if startup_time_override is not None:
+ if verbose:
+ print(f" Startup time: {startup_time_override:.2f}s (manual override)")
+ else:
+ 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: {max(0.0, video_duration - demo_duration):.2f}s (calculated)")
+
+ video_segments = compute_keep_segments(
+ timeline.alive_segments, video_duration, demo_duration,
+ startup_time_override=startup_time_override,
+ )
+
+ 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:
+ total_keep_time = sum(end - start for start, end in video_segments)
+ try:
+ video_duration = get_video_duration(video_path)
+ total_trim_time = video_duration - total_keep_time
+ except:
+ total_trim_time = 0.0
+
+ print(f" Found {len(video_segments)} segments to keep")
+ print(f" Total keep time: {total_keep_time:.1f}s, trim time: {total_trim_time:.1f}s")
+
+ raw_path = video_path.parent / f"{video_path.stem}_raw{video_path.suffix}"
+ video_path.rename(raw_path)
+ print(f" Raw recording moved to: {raw_path.name}")
+
+ print(" Extracting and concatenating segments...")
+ success = extract_and_concat_segments(
+ input_path=raw_path,
+ output_path=video_path,
+ segments=video_segments,
+ verbose=verbose
+ )
+
+ 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, None
+
+
+# =============================================================================
+# Info Output Functions
+# =============================================================================
+
+def print_demo_info_extended(
+ demo_info: DemoInfo,
+ timelines: dict[int, DemoTimeline],
+ verbose: bool = False
+):
+ """Print comprehensive demo information."""
+ # Get rounds and duration from timeline data (more reliable than header)
+ rounds_count = 0
+ max_tick = 0
+ tickrate = demo_info.tick_rate or 64
+
+ for timeline in timelines.values():
+ if timeline.rounds:
+ rounds_count = max(rounds_count, len(timeline.rounds))
+ # Find max tick from spawns/deaths to calculate duration
+ for spawn in timeline.spawns:
+ max_tick = max(max_tick, spawn.tick)
+ for death in timeline.deaths:
+ max_tick = max(max_tick, death.tick)
+ if timeline.tickrate > 0:
+ tickrate = timeline.tickrate
+
+ # Calculate duration from max tick
+ if max_tick > 0 and tickrate > 0:
+ duration = max_tick / tickrate
+ duration_str = f"{duration:.1f}s ({duration/60:.1f} min)"
+ elif demo_info.total_ticks > 0 and tickrate > 0:
+ duration = demo_info.total_ticks / tickrate
+ duration_str = f"{duration:.1f}s ({duration/60:.1f} min)"
+ else:
+ duration_str = "unknown"
+
+ print(f"Demo: {demo_info.path.name}")
+ print(f" Map: {demo_info.map_name}")
+ print(f" Tick rate: {tickrate}/s")
+ print(f" Duration: {duration_str}")
+ print(f" Rounds: {rounds_count}")
+
+ print(f"\nPlayers ({len(demo_info.players)}):")
+ for player in demo_info.players:
+ timeline = timelines.get(player.steamid)
+
+ print(f" {player.name}")
+ print(f" SteamID: {player.steamid}")
+ print(f" Kills: {player.kills} Deaths: {len(timeline.deaths) if timeline else 0} Assists: {player.assists}")
+
+ if timeline:
+ alive_time = sum(seg.duration_seconds for seg in timeline.alive_segments)
+ print(f" Alive segments: {len(timeline.alive_segments)}, Alive time: {alive_time:.1f}s")
+
+ if verbose and timeline.alive_segments:
+ print(f" Alive segments:")
+ for i, seg in enumerate(timeline.alive_segments, 1):
+ print(f" {i:2}. R{seg.round_num:2} {seg.start_time:7.1f}s - {seg.end_time:7.1f}s "
+ f"({seg.duration_seconds:5.1f}s) [{seg.reason_ended}]")
+
+
+def format_info_json(
+ demo_info: DemoInfo,
+ timelines: dict[int, DemoTimeline]
+) -> dict:
+ """Format demo info as JSON-serializable dict."""
+ # Get rounds and duration from timeline data (more reliable than header)
+ rounds_count = 0
+ max_tick = 0
+ tickrate = demo_info.tick_rate or 64
+
+ for timeline in timelines.values():
+ if timeline.rounds:
+ rounds_count = max(rounds_count, len(timeline.rounds))
+ for spawn in timeline.spawns:
+ max_tick = max(max_tick, spawn.tick)
+ for death in timeline.deaths:
+ max_tick = max(max_tick, death.tick)
+ if timeline.tickrate > 0:
+ tickrate = timeline.tickrate
+
+ # Calculate duration from max tick
+ if max_tick > 0 and tickrate > 0:
+ duration = max_tick / tickrate
+ elif demo_info.total_ticks > 0 and tickrate > 0:
+ duration = demo_info.total_ticks / tickrate
+ else:
+ duration = 0
+
+ players_data = []
+ for player in demo_info.players:
+ timeline = timelines.get(player.steamid)
+ player_data = {
+ "name": player.name,
+ "steamid": player.steamid,
+ "kills": player.kills,
+ "assists": player.assists,
+ }
+
+ if timeline:
+ player_data["deaths"] = len(timeline.deaths)
+ player_data["alive_time_seconds"] = sum(seg.duration_seconds for seg in timeline.alive_segments)
+ player_data["alive_segments"] = [
+ {
+ "round_num": seg.round_num,
+ "start_tick": seg.start_tick,
+ "start_time": seg.start_time,
+ "end_tick": seg.end_tick,
+ "end_time": seg.end_time,
+ "duration_seconds": seg.duration_seconds,
+ "reason_ended": seg.reason_ended,
+ }
+ for seg in timeline.alive_segments
+ ]
+
+ players_data.append(player_data)
+
+ return {
+ "demo": demo_info.path.name,
+ "map": demo_info.map_name,
+ "tick_rate": tickrate,
+ "duration_seconds": duration,
+ "rounds": rounds_count,
+ "players": players_data,
+ }
+
+
+# =============================================================================
+# Argument Parser
+# =============================================================================
+
+def create_parser() -> argparse.ArgumentParser:
+ """Create argument parser with subcommands."""
+ parser = argparse.ArgumentParser(
+ prog="cs2pov",
+ description="Record player POV from CS2 demo files",
+ formatter_class=argparse.RawDescriptionHelpFormatter,
+ epilog="""\
+Examples:
+ cs2pov # Run all jobs from cs2pov.json
+ cs2pov init # Create cs2pov.json config
+ cs2pov info demo.dem # Show demo information
+ cs2pov info demo.dem --json # Output as JSON
+ cs2pov pov -d demo.dem -p "Player" -o out.mp4
+ cs2pov pov # Run batch pov jobs from cs2pov.json
+ cs2pov record -d demo.dem -p "Player" -o raw.mp4
+ cs2pov trim raw.mp4 -d demo.dem -p "Player"
+""",
+ )
+ parser.add_argument("--version", action="version", version=f"%(prog)s {__version__}")
+ parser.add_argument("--config", type=Path, default=None,
+ help="Path to cs2pov.json config file (default: auto-detect)")
+
+ subparsers = parser.add_subparsers(dest="command", required=False, metavar="COMMAND")
+
+ # Shared argument groups
+ demo_args = argparse.ArgumentParser(add_help=False)
+ demo_args.add_argument("-d", "--demo", default=None, type=Path,
+ help="Demo file (.dem)")
+
+ player_args = argparse.ArgumentParser(add_help=False)
+ player_args.add_argument("-p", "--player", default=None,
+ help="Player name or SteamID")
+
+ output_args = argparse.ArgumentParser(add_help=False)
+ output_args.add_argument("-o", "--output", default=None, type=Path,
+ help="Output video file")
+
+ recording_args = argparse.ArgumentParser(add_help=False)
+ recording_args.add_argument("-r", "--resolution", default=None,
+ help="Recording resolution (default: 1920x1080)")
+ recording_args.add_argument("-f", "--framerate", type=int, default=None,
+ help="Recording framerate (default: 60)")
+ recording_args.add_argument("--display", type=int, default=None,
+ help="X display number (default: 0)")
+ recording_args.add_argument("--no-hud", action="store_true", default=None,
+ help="Hide HUD elements")
+ recording_args.add_argument("--hud", action="store_false", dest="no_hud",
+ help="Show HUD (override config no_hud)")
+ recording_args.add_argument("--no-audio", action="store_true", default=None,
+ help="Disable audio recording")
+ recording_args.add_argument("--audio", action="store_false", dest="no_audio",
+ help="Enable audio (override config no_audio)")
+ recording_args.add_argument("--audio-device", default=None,
+ help="PulseAudio device (auto-detected)")
+ recording_args.add_argument("--cs2-path", type=Path, default=None,
+ help="Custom CS2 installation path")
+ recording_args.add_argument("--tick-nav", action=argparse.BooleanOptionalAction, default=None,
+ help="Enable tick-based navigation (skip deaths in real-time)")
+
+ verbose_args = argparse.ArgumentParser(add_help=False)
+ verbose_args.add_argument("-v", "--verbose", action=argparse.BooleanOptionalAction, default=None,
+ help="Verbose output")
+
+ # INFO command
+ info_parser = subparsers.add_parser(
+ "info",
+ parents=[verbose_args],
+ help="Show demo information and player timeline data",
+ description="Display demo metadata, player list, and death/spawn statistics.",
+ )
+ info_parser.add_argument("demo", type=Path, help="Demo file (.dem)")
+ info_parser.add_argument("--json", action="store_true",
+ help="Output as JSON")
+
+ # POV command (full pipeline)
+ pov_parser = subparsers.add_parser(
+ "pov",
+ parents=[demo_args, player_args, output_args, recording_args, verbose_args],
+ help="Record and trim player POV (full pipeline)",
+ description="Record a player's POV from a demo and trim death periods.",
+ )
+ pov_parser.add_argument("--no-trim", action="store_true", default=None,
+ help="Skip post-processing trim")
+ pov_parser.add_argument("--trim", action="store_false", dest="no_trim",
+ help="Enable trimming (override config no_trim)")
+ pov_parser.add_argument("--comms-audio", type=Path, default=None,
+ help="Comms audio file to overlay after trim")
+ pov_parser.add_argument("--comms-r1-sync-time", type=float, default=None,
+ help="Seconds into comms audio where round 1 starts (default: 0)")
+ pov_parser.add_argument("--comms-volume", type=float, default=None,
+ help="Comms volume multiplier (default: 1.0)")
+ pov_parser.add_argument("--game-volume", type=float, default=None,
+ help="Game audio volume multiplier (default: 1.0)")
+
+ # RECORD command
+ subparsers.add_parser(
+ "record",
+ parents=[demo_args, player_args, output_args, recording_args, verbose_args],
+ help="Record player POV without trimming",
+ description="Record a player's POV from a demo without post-processing.",
+ )
+
+ # TRIM command
+ trim_parser = subparsers.add_parser(
+ "trim",
+ parents=[demo_args, player_args, verbose_args],
+ help="Trim death periods from recorded video",
+ description="Post-process an existing recording to remove death periods.",
+ )
+ 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)")
+ 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(
+ "comms",
+ parents=[demo_args, player_args, verbose_args],
+ help="Overlay comms audio on recorded video",
+ description="Mix external comms audio onto a video, syncing with alive segments.",
+ )
+ comms_parser.add_argument("video", type=Path, help="Input video file")
+ comms_parser.add_argument("-a", "--audio", type=Path, required=True,
+ help="Comms audio file (mp3/wav/flac)")
+ comms_parser.add_argument("-o", "--output", type=Path,
+ help="Output file (default: {stem}_comms{suffix})")
+ comms_parser.add_argument("--comms-r1-sync-time", type=float, default=0.0,
+ help="Seconds into comms audio where round 1 starts (default: 0)")
+ comms_parser.add_argument("--comms-volume", type=float, default=1.0,
+ 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)")
+
+ # INIT command
+ subparsers.add_parser(
+ "init",
+ help="Create a cs2pov.json config file in the current directory",
+ description="Generate a template cs2pov.json config with defaults and example jobs.",
+ )
+
+ return parser
+
+
+# =============================================================================
+# Command Handlers
+# =============================================================================
+
+def _validate_required_args(args, command: str) -> Optional[str]:
+ """Validate that required args are present after merging.
+
+ Returns error message or None if valid.
+ """
+ missing = []
+ if command == "comms":
+ if getattr(args, "video", None) is None:
+ missing.append("video")
+ if getattr(args, "audio", None) is None and getattr(args, "comms_audio", None) is None:
+ missing.append("audio")
+ if getattr(args, "demo", None) is None:
+ 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")
+ if getattr(args, "player", None) is None:
+ missing.append("-p/--player")
+ if getattr(args, "output", None) is None:
+ missing.append("-o/--output")
+
+ if missing:
+ return f"Missing required arguments for '{command}': {', '.join(missing)}"
+ return None
+
+
+def _run_single_comms(args) -> int:
+ """Run a single comms overlay job from config."""
+ # comms jobs require 'video' and 'audio' fields instead of 'output'
+ video = getattr(args, "video", None)
+ if video is None:
+ print("Error: comms job requires 'video' field", file=sys.stderr)
+ return 1
+ args.video = Path(video)
+
+ audio = getattr(args, "audio", None) or getattr(args, "comms_audio", None)
+ if audio is None:
+ print("Error: comms job requires 'audio' (or 'comms_audio') field", file=sys.stderr)
+ return 1
+ args.audio = Path(audio)
+
+ # Set defaults for comms-specific args
+ no_trim_sync = getattr(args, "no_trim_sync", False)
+ if isinstance(no_trim_sync, str):
+ no_trim_sync = no_trim_sync.lower() in ("true", "1", "yes")
+ args.no_trim_sync = bool(no_trim_sync)
+
+ if not hasattr(args, "comms_r1_sync_time"):
+ args.comms_r1_sync_time = 0.0
+
+ 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
+
+
+def _run_batch(args, command: str, run_single_job=None) -> int:
+ """Run single or batch jobs with config support.
+
+ Args:
+ args: Parsed argparse namespace
+ command: Command name for error messages
+ run_single_job: Callable(args) -> int for single job. If None,
+ dispatches per job using the job's 'type' field from config.
+ """
+ from copy import copy
+
+ # Load config
+ try:
+ config_path = find_config(getattr(args, "config", None))
+ project = load_config(config_path) if config_path else None
+ except ConfigError as e:
+ print(f"Config error: {e}", file=sys.stderr)
+ return 1
+
+ config_dir = config_path.parent if config_path else Path.cwd()
+
+ # Determine job list and per-job types
+ job_types: list[str] = []
+ if project and project.jobs and getattr(args, "demo", None) is None:
+ # Batch mode: use jobs from config
+ job_types = [j.type for j in project.jobs]
+ jobs = [resolve_job(j, project.defaults) for j in project.jobs]
+ jobs = [resolve_paths(j, config_dir) for j in jobs]
+ print(f"Running {len(jobs)} jobs from {config_path.name}\n")
+ else:
+ # Single mode: config defaults + CLI args
+ jobs = [project.defaults.copy() if project else {}]
+ job_types = [command] # Use the command as the type
+ if project:
+ jobs[0] = resolve_paths(jobs[0], config_dir)
+
+ results: list[tuple[int, str, Optional[str]]] = [] # (index, demo_name, error)
+
+ for i, job in enumerate(jobs):
+ merged = merge_args_with_config(copy(args), job)
+ job_type = job_types[i]
+
+ # Validate required fields
+ err = _validate_required_args(merged, job_type)
+ if err:
+ if len(jobs) == 1:
+ print(f"Error: {err}", file=sys.stderr)
+ if project:
+ print(f"Hint: Add jobs to {config_path.name} or pass -d, -p, -o flags", file=sys.stderr)
+ else:
+ print(f"Hint: Create a cs2pov.json with 'cs2pov init' or pass -d, -p, -o flags", file=sys.stderr)
+ return 1
+ demo_name = job.get("demo", f"job {i+1}")
+ print(f"[SKIP] Job {i+1}: {demo_name} - {err}")
+ results.append((i, str(demo_name), err))
+ continue
+
+ demo_name = str(merged.demo)
+ if len(jobs) > 1:
+ print(f"{'='*60}")
+ print(f"Job {i+1}/{len(jobs)}: {Path(demo_name).name} ({job_type})")
+ print(f"{'='*60}\n")
+
+ # Dispatch to the appropriate runner
+ runner = run_single_job if run_single_job else _job_runner_for_type(job_type)
+
+ try:
+ ret = runner(merged)
+ if ret == 0:
+ results.append((i, demo_name, None))
+ else:
+ results.append((i, demo_name, f"Exit code {ret}"))
+ except CS2POVError as e:
+ print(f"Error: {e}", file=sys.stderr)
+ results.append((i, demo_name, str(e)))
+ except Exception as e:
+ print(f"Unexpected error: {e}", file=sys.stderr)
+ results.append((i, demo_name, str(e)))
+
+ # Sleep between batch jobs
+ if len(jobs) > 1 and i < len(jobs) - 1:
+ print(f"\nWaiting 10s before next job...\n")
+ time.sleep(10)
+
+ # Print batch summary
+ if len(jobs) > 1:
+ succeeded = sum(1 for _, _, err in results if err is None)
+ print(f"\n{'='*60}")
+ print(f"Batch complete: {succeeded}/{len(results)} succeeded")
+ for i, demo_name, err in results:
+ if err:
+ print(f" [FAIL] Job {i+1}: {Path(demo_name).name} - {err}")
+ print(f"{'='*60}")
+ return 0 if succeeded == len(results) else 1
+
+ # Single job
+ if results and results[0][2] is not None:
+ return 1
+ return 0
+
+
+def cmd_init(args) -> int:
+ """Handle 'init' command - create cs2pov.json config."""
+ config_path = Path.cwd() / "cs2pov.json"
+
+ if config_path.exists():
+ print(f"Config already exists: {config_path}")
+ response = input("Overwrite? [y/N] ").strip().lower()
+ if response != "y":
+ print("Aborted.")
+ return 0
+
+ # Try to auto-detect CS2 path
+ cs2_path_str = None
+ try:
+ cs2_path = find_cs2_path()
+ cs2_path_str = str(cs2_path)
+ print(f"Detected CS2 path: {cs2_path}")
+ except CS2POVError:
+ print("CS2 path not auto-detected. You can set it manually in the config.")
+
+ config = generate_default_config(cs2_path=cs2_path_str)
+ config_path.write_text(json.dumps(config, indent=2) + "\n")
+ print(f"Created {config_path}")
+ print(f"\nEdit the 'jobs' array to add your recording jobs, then run:")
+ print(f" cs2pov pov")
+ return 0
+
+
+def cmd_info(args) -> int:
+ """Handle 'info' command - show demo information."""
+ demo_path = args.demo.resolve()
+ if not demo_path.exists():
+ print(f"Error: Demo not found: {demo_path}", file=sys.stderr)
+ return 1
+
+ # Parse basic demo info
+ with LoadingAnimation():
+ try:
+ demo_info = parse_demo(demo_path)
+ except CS2POVError as e:
+ print(f"Error: {e}", file=sys.stderr)
+ return 1
+
+ # Preprocess for timeline data (per-player)
+ player_timelines: dict[int, DemoTimeline] = {}
+ for player in demo_info.players:
+ try:
+ timeline = preprocess_demo(demo_path, player.steamid, player.name)
+ player_timelines[player.steamid] = timeline
+ except Exception:
+ pass
+
+ if args.json:
+ output = format_info_json(demo_info, player_timelines)
+ print(json.dumps(output, indent=2))
+ else:
+ print_demo_info_extended(demo_info, player_timelines, verbose=args.verbose)
+
+ return 0
+
+
+def _run_single_pov(args) -> int:
+ """Run a single POV recording job (record + trim)."""
+ # Check dependencies
+ missing = check_dependencies()
+ if missing:
+ print("Missing dependencies:", file=sys.stderr)
+ for dep in missing:
+ print(f" - {dep}", file=sys.stderr)
+ return 1
+
+ # Validate inputs
+ demo_path = args.demo.resolve()
+ if not demo_path.exists():
+ print(f"Error: Demo not found: {demo_path}", file=sys.stderr)
+ return 1
+
+ try:
+ resolution = parse_resolution(args.resolution)
+ except ValueError as e:
+ print(f"Error: {e}", file=sys.stderr)
+ return 1
+
+ output_path = args.output.resolve()
+
+ with LoadingAnimation():
+ # Record
+ result = record_demo(
+ demo_path=demo_path,
+ player_identifier=args.player,
+ output_path=output_path,
+ resolution=resolution,
+ framerate=args.framerate,
+ hide_hud=args.no_hud,
+ display_num=args.display,
+ verbose=args.verbose,
+ cs2_path_override=args.cs2_path,
+ enable_audio=not args.no_audio,
+ audio_device=args.audio_device,
+ tick_nav=args.tick_nav,
+ )
+
+ # Post-process
+ keep_segments = None
+ if result.success and not args.no_trim:
+ _, keep_segments = postprocess_video(
+ video_path=result.video_path,
+ console_log_path=result.console_log_path,
+ verbose=args.verbose,
+ timeline=result.timeline,
+ transitions=result.transitions,
+ )
+ elif result.success:
+ size_mb = result.video_path.stat().st_size / (1024 * 1024)
+ print(f"\nRecording saved: {result.video_path} ({size_mb:.1f} MB)")
+ else:
+ print(f"\nRecording ended with: {result.exit_reason}")
+ if result.video_path.exists():
+ size_mb = result.video_path.stat().st_size / (1024 * 1024)
+ print(f"Partial recording available: {result.video_path} ({size_mb:.1f} MB)")
+
+ # Overlay comms audio if provided
+ comms_audio = getattr(args, "comms_audio", None) or getattr(args, "audio", None)
+ if result.success and comms_audio:
+ from .comms import apply_comms_overlay
+
+ comms_path = Path(comms_audio).resolve()
+ if not comms_path.exists():
+ print(f"Warning: Comms audio not found: {comms_path}", file=sys.stderr)
+ else:
+ comms_output = result.video_path.parent / f"{result.video_path.stem}_comms{result.video_path.suffix}"
+ print(f"\nOverlaying comms audio: {comms_path.name}")
+ comms_ok = apply_comms_overlay(
+ video_path=result.video_path,
+ comms_audio_path=comms_path,
+ output_path=comms_output,
+ timeline=result.timeline,
+ r1_sync_time=float(getattr(args, "comms_r1_sync_time", 0.0) or 0.0),
+ game_volume=float(getattr(args, "game_volume", 1.0) or 1.0),
+ 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)
+ print(f"Comms output: {comms_output} ({size_mb:.1f} MB)")
+ else:
+ print("Warning: Comms overlay failed", file=sys.stderr)
+
+ return 0 if result.success else 1
+
+
+def cmd_pov(args) -> int:
+ """Handle 'pov' command - full recording pipeline with batch support."""
+ return _run_batch(args, "pov", _run_single_pov)
+
+
+def _run_single_record(args) -> int:
+ """Run a single raw recording job (no trim)."""
+ # Check dependencies
+ missing = check_dependencies()
+ if missing:
+ print("Missing dependencies:", file=sys.stderr)
+ for dep in missing:
+ print(f" - {dep}", file=sys.stderr)
+ return 1
+
+ # Validate inputs
+ demo_path = args.demo.resolve()
+ if not demo_path.exists():
+ print(f"Error: Demo not found: {demo_path}", file=sys.stderr)
+ return 1
+
+ try:
+ resolution = parse_resolution(args.resolution)
+ except ValueError as e:
+ print(f"Error: {e}", file=sys.stderr)
+ return 1
+
+ output_path = args.output.resolve()
+
+ with LoadingAnimation():
+ # Record only
+ result = record_demo(
+ demo_path=demo_path,
+ player_identifier=args.player,
+ output_path=output_path,
+ resolution=resolution,
+ framerate=args.framerate,
+ hide_hud=args.no_hud,
+ display_num=args.display,
+ verbose=args.verbose,
+ cs2_path_override=args.cs2_path,
+ enable_audio=not args.no_audio,
+ audio_device=args.audio_device,
+ tick_nav=args.tick_nav,
+ )
+
+ if result.success:
+ size_mb = result.video_path.stat().st_size / (1024 * 1024)
+ print(f"\nRaw recording saved: {result.video_path} ({size_mb:.1f} MB)")
+ print(f"\nTo trim later, run:")
+ print(f" cs2pov trim \"{result.video_path}\" -d \"{demo_path}\" -p \"{args.player}\"")
+ else:
+ print(f"\nRecording ended with: {result.exit_reason}")
+ if result.video_path.exists():
+ size_mb = result.video_path.stat().st_size / (1024 * 1024)
+ print(f"Partial recording available: {result.video_path} ({size_mb:.1f} MB)")
+
+ return 0 if result.success else 1
+
+
+def cmd_record(args) -> int:
+ """Handle 'record' command - raw recording with batch support."""
+ return _run_batch(args, "record", _run_single_record)
+
+
+def cmd_run(args) -> int:
+ """Handle bare 'cs2pov' command - run all jobs from config with per-job type dispatch."""
+ config_path = find_config(getattr(args, "config", None))
+ if config_path is None:
+ print("Error: No cs2pov.json config found.", file=sys.stderr)
+ print("Hint: Create one with 'cs2pov init' or specify with --config", file=sys.stderr)
+ return 1
+
+ project = load_config(config_path)
+ if not project.jobs:
+ print(f"Error: No jobs defined in {config_path.name}", file=sys.stderr)
+ print("Hint: Add jobs to the 'jobs' array in your config file", file=sys.stderr)
+ return 1
+
+ return _run_batch(args, "pov")
+
+
+def _apply_config_defaults(args) -> None:
+ """Apply config defaults to args for commands that don't use _run_batch.
+
+ Non-fatal: config errors are silently ignored.
+ """
+ try:
+ config_path = find_config(getattr(args, "config", None))
+ if config_path:
+ project = load_config(config_path)
+ for key, value in project.defaults.items():
+ if key in HARDCODED_DEFAULTS and getattr(args, key, None) is None:
+ setattr(args, key, value)
+ except ConfigError:
+ pass
+
+ # Fill remaining None values with hardcoded defaults
+ for key, value in HARDCODED_DEFAULTS.items():
+ if getattr(args, key, None) is None:
+ setattr(args, key, value)
+
+
+def cmd_trim(args) -> int:
+ """Handle 'trim' command - post-process existing video."""
+ _apply_config_defaults(args)
+
+ video_path = args.video.resolve()
+ if not video_path.exists():
+ print(f"Error: Video not found: {video_path}", file=sys.stderr)
+ return 1
+
+ demo_path = args.demo.resolve()
+ if not demo_path.exists():
+ print(f"Error: Demo not found: {demo_path}", file=sys.stderr)
+ return 1
+
+ # Determine output path
+ if args.output:
+ output_path = args.output.resolve()
+ else:
+ output_path = video_path.parent / f"{video_path.stem}_trimmed{video_path.suffix}"
+
+ with LoadingAnimation():
+ # Parse demo and find player
+ try:
+ demo_info = parse_demo(demo_path)
+ player = find_player(demo_info, args.player)
+ except CS2POVError as e:
+ print(f"Error: {e}", file=sys.stderr)
+ return 1
+
+ # Get timeline from demo
+ timeline = None
+ try:
+ timeline = preprocess_demo(demo_path, player.steamid, player.name)
+ print(f"Using demo timeline: {len(timeline.deaths)} deaths, {len(timeline.rounds)} rounds")
+ print(f" {len(timeline.alive_segments)} alive segments to keep")
+ except Exception as e:
+ print(f"Warning: Could not preprocess demo: {e}")
+
+ if timeline is None:
+ 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:
+ shutil.copy2(video_path, output_path)
+
+ # 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
+
+
+def cmd_comms(args) -> int:
+ """Handle 'comms' command - overlay comms audio on video."""
+ _apply_config_defaults(args)
+
+ video_path = args.video.resolve()
+ if not video_path.exists():
+ print(f"Error: Video not found: {video_path}", file=sys.stderr)
+ return 1
+
+ comms_path = args.audio.resolve()
+ if not comms_path.exists():
+ print(f"Error: Audio not found: {comms_path}", file=sys.stderr)
+ return 1
+
+ # Determine output
+ if args.output:
+ output_path = args.output.resolve()
+ else:
+ output_path = video_path.parent / f"{video_path.stem}_comms{video_path.suffix}"
+
+ # Parse demo for alive segments (unless --no-trim-sync)
+ timeline = None
+ is_trimmed = not args.no_trim_sync
+
+ if is_trimmed:
+ if args.demo is None:
+ print("Error: --demo is required for trim sync (use --no-trim-sync to skip)", file=sys.stderr)
+ return 1
+ demo_path = args.demo.resolve()
+ if not demo_path.exists():
+ print(f"Error: Demo not found: {demo_path}", file=sys.stderr)
+ return 1
+ if args.player is None:
+ print("Error: --player is required for trim sync (use --no-trim-sync to skip)", file=sys.stderr)
+ return 1
+
+ try:
+ demo_info = parse_demo(demo_path)
+ player = find_player(demo_info, args.player)
+ timeline = preprocess_demo(demo_path, player.steamid, player.name)
+ print(f" {len(timeline.alive_segments)} alive segments from timeline")
+ except CS2POVError as e:
+ print(f"Error: {e}", file=sys.stderr)
+ return 1
+
+ from .comms import apply_comms_overlay
+ from .trim import get_video_duration, compute_keep_segments, load_transitions
+
+ # Recompute keep segments for standalone comms on an already-trimmed video.
+ # For tick-nav mode: use GotoTransitions sidecar for actual video durations
+ # (includes polling overrun), avoiding cumulative drift.
+ keep_segments = None
+ if is_trimmed and timeline and timeline.alive_segments:
+ if args.tick_nav:
+ # Try loading transitions sidecar for actual segment durations.
+ # The sidecar is saved as {output_stem}_transitions.json during recording,
+ # where output_stem is the base name (e.g., "recording"). Try multiple
+ # candidate paths since the user may pass the trimmed or raw video.
+ try:
+ transitions = None
+ # Try: video_path itself, then strip _raw suffix, then add _raw suffix
+ candidates = [video_path]
+ stem = video_path.stem
+ if stem.endswith("_raw"):
+ # User passed raw video — sidecar is named after the base stem
+ base = video_path.parent / f"{stem[:-4]}{video_path.suffix}"
+ candidates.insert(0, base)
+ else:
+ # User passed trimmed video — sidecar might also be next to raw
+ raw_path = video_path.parent / f"{stem}_raw{video_path.suffix}"
+ candidates.append(raw_path)
+
+ for candidate in candidates:
+ transitions = load_transitions(candidate)
+ if transitions:
+ break
+
+ if transitions:
+ # Get raw video duration for the last keep segment
+ raw_candidate = video_path.parent / f"{stem}_raw{video_path.suffix}" if not stem.endswith("_raw") else video_path
+ if raw_candidate.exists():
+ source_duration = get_video_duration(raw_candidate)
+ else:
+ source_duration = get_video_duration(video_path)
+
+ # Build keep segments from transitions (same logic as trim_goto_transitions)
+ sorted_transitions = sorted(transitions, key=lambda t: t.pause_video_time)
+
+ raw_keep_segments: list[tuple[float, float]] = []
+ current_pos = 0.0
+ for t in sorted_transitions:
+ if t.pause_video_time > current_pos + 0.5:
+ raw_keep_segments.append((current_pos, t.pause_video_time))
+ current_pos = max(current_pos, t.unpause_video_time)
+ if current_pos < source_duration - 0.5:
+ raw_keep_segments.append((current_pos, source_duration))
+
+ if raw_keep_segments:
+ keep_segments = raw_keep_segments
+ if args.verbose:
+ durations = [end - start for start, end in keep_segments]
+ print(f" Using {len(keep_segments)} keep segment durations from transitions")
+ print(f" Total keep duration: {sum(durations):.1f}s")
+ elif args.verbose:
+ print(" No transitions sidecar found, falling back to tick-nav extension logic")
+ except Exception as e:
+ if args.verbose:
+ print(f" Warning: Could not load transitions: {e}, falling back to tick-nav extension logic")
+ else:
+ 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}")
+
+ print(f"Overlaying comms: {comms_path.name} → {output_path.name}")
+ success = apply_comms_overlay(
+ video_path=video_path,
+ comms_audio_path=comms_path,
+ output_path=output_path,
+ timeline=timeline,
+ r1_sync_time=float(args.comms_r1_sync_time),
+ game_volume=float(args.game_volume),
+ comms_volume=float(args.comms_volume),
+ is_trimmed=is_trimmed,
+ verbose=args.verbose,
+ tick_nav=args.tick_nav,
+ keep_segments=keep_segments,
+ )
+
+ if success:
+ size_mb = output_path.stat().st_size / (1024 * 1024)
+ print(f"\nOutput: {output_path} ({size_mb:.1f} MB)")
+ else:
+ print(f"\nComms overlay failed", file=sys.stderr)
+ return 1
+
+ return 0
+
+
+# =============================================================================
+# Entry Point
+# =============================================================================
+
+def main() -> int:
+ """Main entry point."""
+ parser = create_parser()
+ args = parser.parse_args()
+
+ try:
+ if args.command is None:
+ return cmd_run(args)
+ elif args.command == "init":
+ return cmd_init(args)
+ elif args.command == "info":
+ return cmd_info(args)
+ elif args.command == "pov":
+ return cmd_pov(args)
+ elif args.command == "record":
+ return cmd_record(args)
+ elif args.command == "trim":
+ return cmd_trim(args)
+ elif args.command == "comms":
+ return cmd_comms(args)
+ else:
+ parser.print_help()
+ return 1
+ except CS2POVError as e:
+ print(f"Error: {e}", file=sys.stderr)
+ return 1
+ except KeyboardInterrupt:
+ print("\nInterrupted", file=sys.stderr)
+ return 130
+
+
+if __name__ == "__main__":
+ sys.exit(main())