diff options
Diffstat (limited to '')
| -rw-r--r-- | cs2pov/cli.py | 182 | ||||
| -rw-r--r-- | cs2pov/comms.py | 363 | ||||
| -rw-r--r-- | cs2pov/settings.py | 28 |
3 files changed, 559 insertions, 14 deletions
diff --git a/cs2pov/cli.py b/cs2pov/cli.py index f5e09d2..498e6f8 100644 --- a/cs2pov/cli.py +++ b/cs2pov/cli.py @@ -804,6 +804,14 @@ Examples: 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( @@ -826,6 +834,27 @@ Examples: trim_parser.add_argument("--startup-time", type=float, help="Override startup time (seconds from video start to demo start)") + # 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("--no-trim-sync", action="store_true", + help="Skip alive-segment sync (for untrimmed videos)") + # INIT command subparsers.add_parser( "init", @@ -841,27 +870,66 @@ Examples: # ============================================================================= def _validate_required_args(args, command: str) -> Optional[str]: - """Validate that required args (demo, player, output) are present after merging. + """Validate that required args are present after merging. Returns error message or None if valid. """ missing = [] - 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 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") + 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 _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 return _run_single_pov @@ -1086,6 +1154,34 @@ def _run_single_pov(args) -> int: 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, + ) + 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 @@ -1250,6 +1346,76 @@ def cmd_trim(args) -> int: 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 + + 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, + ) + + 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 # ============================================================================= @@ -1272,6 +1438,8 @@ def main() -> int: 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 diff --git a/cs2pov/comms.py b/cs2pov/comms.py new file mode 100644 index 0000000..7d96e20 --- /dev/null +++ b/cs2pov/comms.py @@ -0,0 +1,363 @@ +"""Comms audio overlay: trim external audio to match alive segments, then mix onto video.""" + +import subprocess +import tempfile +from pathlib import Path +from typing import Optional + +from .preprocessor import AliveSegment, DemoTimeline + + +def compute_comms_segments( + alive_segments: list[AliveSegment], + round1_freeze_end_time: float, + r1_sync_time: float = 0.0, +) -> list[tuple[float, float]]: + """Map alive segments from demo time to comms audio time. + + Comms time = demo_time - round1_freeze_end_time + r1_sync_time. + r1_sync_time is the timestamp in the comms audio where round 1 starts. + Segments before comms t=0 are clamped or dropped. + + Args: + alive_segments: Alive segments from DemoTimeline + round1_freeze_end_time: Demo time (seconds) of round 1 freeze end + r1_sync_time: Seconds into comms audio where round 1 begins + + Returns: + List of (start, end) tuples in comms audio time (seconds) + """ + segments = [] + for seg in alive_segments: + comms_start = seg.start_time - round1_freeze_end_time + r1_sync_time + comms_end = seg.end_time - round1_freeze_end_time + r1_sync_time + + if comms_end <= 0: + continue + comms_start = max(0.0, comms_start) + segments.append((comms_start, comms_end)) + + return segments + + +def _get_round1_freeze_end_time(timeline: DemoTimeline) -> Optional[float]: + """Get round 1 freeze_end time from timeline. + + Returns the first round's freeze_end_time, skipping rounds without it. + """ + for r in timeline.rounds: + if r.freeze_end_time is not None: + return r.freeze_end_time + return None + + +def _has_audio_stream(video_path: Path) -> bool: + """Check if video file has an audio stream.""" + try: + result = subprocess.run( + [ + "ffprobe", + "-v", "error", + "-select_streams", "a", + "-show_entries", "stream=codec_type", + "-of", "default=noprint_wrappers=1:nokey=1", + str(video_path), + ], + capture_output=True, + text=True, + timeout=30, + ) + return result.returncode == 0 and result.stdout.strip() != "" + except (FileNotFoundError, subprocess.TimeoutExpired): + return False + + +def extract_comms_segments( + comms_audio_path: Path, + segments: list[tuple[float, float]], + output_path: Path, + verbose: bool = False, +) -> bool: + """Extract and concatenate segments from comms audio using FFmpeg. + + Extracts each segment as a WAV intermediate, then concatenates them. + + Args: + comms_audio_path: Path to input comms audio file + segments: List of (start, end) in comms audio seconds + output_path: Path for trimmed output audio (WAV) + verbose: Print debug info + + Returns: + True on success + """ + if not segments: + if verbose: + print(" [Comms] No segments to extract") + return False + + segments = sorted(segments, key=lambda s: s[0]) + + if verbose: + total_duration = sum(end - start for start, end in segments) + print(f" [Comms] Extracting {len(segments)} audio segments, total: {total_duration:.2f}s") + + with tempfile.TemporaryDirectory() as tmpdir: + tmpdir_path = Path(tmpdir) + concat_file = tmpdir_path / "concat.txt" + segment_paths: list[Path] = [] + + for i, (start, end) in enumerate(segments): + segment_path = tmpdir_path / f"segment_{i:03d}.wav" + segment_paths.append(segment_path) + duration = end - start + + if verbose: + print(f" [Comms] Segment {i}: {start:.2f}s - {end:.2f}s ({duration:.2f}s)") + + cmd = [ + "ffmpeg", "-y", + "-ss", str(start), + "-i", str(comms_audio_path), + "-t", str(duration), + "-c:a", "pcm_s16le", + "-ar", "48000", + "-ac", "2", + str(segment_path), + ] + + try: + result = subprocess.run(cmd, capture_output=True, text=True, timeout=300) + if result.returncode != 0: + if verbose: + print(f" [Comms] Segment extraction failed: {result.stderr}") + return False + except subprocess.TimeoutExpired: + if verbose: + print(" [Comms] Segment extraction timed out") + return False + + with open(concat_file, "w") as f: + for segment_path in segment_paths: + escaped_path = str(segment_path).replace("'", "'\\''") + f.write(f"file '{escaped_path}'\n") + + if verbose: + print(f" [Comms] Concatenating {len(segment_paths)} audio segments") + + cmd = [ + "ffmpeg", "-y", + "-f", "concat", + "-safe", "0", + "-i", str(concat_file), + "-c:a", "pcm_s16le", + str(output_path), + ] + + try: + result = subprocess.run(cmd, capture_output=True, text=True, timeout=600) + if result.returncode != 0: + if verbose: + print(f" [Comms] Concatenation failed: {result.stderr}") + return False + except subprocess.TimeoutExpired: + if verbose: + print(" [Comms] Concatenation timed out") + return False + + return True + + +def overlay_comms_on_video( + video_path: Path, + comms_audio_path: Path, + output_path: Path, + game_volume: float = 1.0, + comms_volume: float = 1.0, + verbose: bool = False, +) -> bool: + """Mix comms audio onto video's existing game audio using FFmpeg amix. + + Video stream is copied without re-encoding. Audio is re-encoded to AAC. + If the video has no audio track, comms are added as the sole audio. + + Args: + video_path: Input video (with or without game audio) + comms_audio_path: Comms audio (already trimmed to match video timeline) + output_path: Output video path + game_volume: Volume multiplier for game audio (default 1.0) + comms_volume: Volume multiplier for comms audio (default 1.0) + verbose: Print debug info + + Returns: + True on success + """ + has_audio = _has_audio_stream(video_path) + + if has_audio: + # Mix game audio + comms audio + filter_complex = ( + f"[0:a]volume={game_volume}[game];" + f"[1:a]volume={comms_volume}[comms];" + f"[game][comms]amix=inputs=2:duration=first:dropout_transition=0[aout]" + ) + cmd = [ + "ffmpeg", "-y", + "-i", str(video_path), + "-i", str(comms_audio_path), + "-filter_complex", filter_complex, + "-map", "0:v", + "-map", "[aout]", + "-c:v", "copy", + "-c:a", "aac", + "-b:a", "192k", + str(output_path), + ] + else: + # No game audio — use comms as sole audio track + cmd = [ + "ffmpeg", "-y", + "-i", str(video_path), + "-i", str(comms_audio_path), + "-map", "0:v", + "-map", "1:a", + "-c:v", "copy", + "-c:a", "aac", + "-b:a", "192k", + str(output_path), + ] + + if verbose: + mode = "mixing with game audio" if has_audio else "adding as sole audio" + print(f" [Comms] Overlaying comms ({mode})") + + try: + result = subprocess.run(cmd, capture_output=True, text=True, timeout=600) + if result.returncode != 0: + if verbose: + print(f" [Comms] Overlay failed: {result.stderr}") + return False + except subprocess.TimeoutExpired: + if verbose: + print(" [Comms] Overlay timed out") + return False + + return True + + +def apply_comms_overlay( + video_path: Path, + comms_audio_path: Path, + output_path: Path, + timeline: Optional[DemoTimeline] = None, + r1_sync_time: float = 0.0, + game_volume: float = 1.0, + comms_volume: float = 1.0, + is_trimmed: bool = True, + verbose: bool = False, +) -> bool: + """High-level entry point: trim comms to match alive segments, then overlay onto video. + + If the video is trimmed and a timeline is provided, the comms audio is cut + to match the same alive segments so it stays in sync. Otherwise, the comms + are overlaid directly with an offset seek. + + Args: + video_path: Input video (trimmed or raw) + comms_audio_path: External comms audio file + output_path: Final output path + timeline: DemoTimeline with alive segments and rounds + r1_sync_time: Seconds into comms audio where round 1 begins + game_volume: Game audio volume (default 1.0) + comms_volume: Comms audio volume (default 1.0) + is_trimmed: Whether the video has been trimmed + verbose: Print debug info + + Returns: + True on success + """ + if is_trimmed and timeline and timeline.alive_segments: + # Get round 1 freeze-end time as reference + round1_time = _get_round1_freeze_end_time(timeline) + if round1_time is None: + print(" Warning: No round data found, using demo start as reference") + round1_time = 0.0 + + # Compute which parts of comms audio correspond to alive segments + comms_segments = compute_comms_segments( + timeline.alive_segments, round1_time, r1_sync_time + ) + + if not comms_segments: + print(" Warning: No comms segments to extract (all before comms start)") + return False + + if verbose: + print(f" [Comms] {len(comms_segments)} segments mapped from {len(timeline.alive_segments)} alive segments") + + # Extract and concat the matching comms audio segments + with tempfile.TemporaryDirectory() as tmpdir: + trimmed_comms = Path(tmpdir) / "comms_trimmed.wav" + + if not extract_comms_segments(comms_audio_path, comms_segments, trimmed_comms, verbose): + print(" Error: Failed to extract comms segments") + return False + + # Overlay trimmed comms onto trimmed video + return overlay_comms_on_video( + video_path, trimmed_comms, output_path, + game_volume, comms_volume, verbose, + ) + else: + # Untrimmed or no timeline — overlay directly with offset + if verbose: + print(f" [Comms] Direct overlay (r1_sync_time: {r1_sync_time}s)") + + # For direct overlay with offset, seek into comms audio + has_audio = _has_audio_stream(video_path) + + if has_audio: + filter_complex = ( + f"[0:a]volume={game_volume}[game];" + f"[1:a]volume={comms_volume}[comms];" + f"[game][comms]amix=inputs=2:duration=first:dropout_transition=0[aout]" + ) + cmd = [ + "ffmpeg", "-y", + "-i", str(video_path), + "-ss", str(max(0.0, r1_sync_time)), + "-i", str(comms_audio_path), + "-filter_complex", filter_complex, + "-map", "0:v", + "-map", "[aout]", + "-c:v", "copy", + "-c:a", "aac", + "-b:a", "192k", + str(output_path), + ] + else: + cmd = [ + "ffmpeg", "-y", + "-i", str(video_path), + "-ss", str(max(0.0, r1_sync_time)), + "-i", str(comms_audio_path), + "-map", "0:v", + "-map", "1:a", + "-c:v", "copy", + "-c:a", "aac", + "-b:a", "192k", + str(output_path), + ] + + try: + result = subprocess.run(cmd, capture_output=True, text=True, timeout=600) + if result.returncode != 0: + if verbose: + print(f" [Comms] Direct overlay failed: {result.stderr}") + return False + except subprocess.TimeoutExpired: + if verbose: + print(" [Comms] Direct overlay timed out") + return False + + return True diff --git a/cs2pov/settings.py b/cs2pov/settings.py index f44796f..78f6e92 100644 --- a/cs2pov/settings.py +++ b/cs2pov/settings.py @@ -29,6 +29,10 @@ HARDCODED_DEFAULTS: dict[str, Any] = { "tick_nav": False, "no_trim": False, "verbose": False, + "comms_audio": None, + "comms_r1_sync_time": 0.0, + "comms_volume": 1.0, + "game_volume": 1.0, } # Keys valid in "defaults" and as job overrides (includes per-job fields that can be defaulted) @@ -38,10 +42,10 @@ VALID_DEFAULT_KEYS = set(HARDCODED_DEFAULTS.keys()) | {"player", "demo", "output REQUIRED_MERGED_KEYS = {"demo", "player", "output"} # Keys valid in a job entry -VALID_JOB_KEYS = REQUIRED_MERGED_KEYS | VALID_DEFAULT_KEYS | {"type"} +VALID_JOB_KEYS = REQUIRED_MERGED_KEYS | VALID_DEFAULT_KEYS | {"type", "video", "audio", "no_trim_sync"} # Valid job types -VALID_JOB_TYPES = {"pov", "record"} +VALID_JOB_TYPES = {"pov", "record", "comms"} CONFIG_FILENAME = "cs2pov.json" CURRENT_VERSION = 1 @@ -245,13 +249,13 @@ def merge_args_with_config(args, job_dict: dict[str, Any]): else: setattr(args, key, hardcoded) - # Set demo/player/output from job if not on CLI - for key in ("demo", "player", "output"): + # Set per-job fields from config if not on CLI + for key in ("demo", "player", "output", "video", "audio", "no_trim_sync"): cli_val = getattr(args, key, None) job_val = job_dict.get(key) if cli_val is None and job_val is not None: - # Convert path strings to Path objects for demo/output - if key in ("demo", "output"): + # Convert path strings to Path objects + if key in ("demo", "output", "video", "audio"): setattr(args, key, Path(job_val)) else: setattr(args, key, job_val) @@ -274,7 +278,17 @@ def generate_default_config(cs2_path: Optional[str] = None) -> dict: "demo": "./demos/example.dem", "player": "PlayerName", "output": "./recordings/example.mp4", - } + }, + { + "type": "comms", + "video": "./recordings/example.mp4", + "audio": "./comms/example_comms.wav", + "demo": "./demos/example.dem", + "player": "PlayerName", + "comms_r1_sync_time": 0.0, + "comms_volume": 1.0, + "game_volume": 0.7, + }, ], } |