diff options
Diffstat (limited to '')
| -rw-r--r-- | .gitignore | 1 | ||||
| -rw-r--r-- | cs2pov/cli.py | 245 | ||||
| -rw-r--r-- | cs2pov/settings.py | 280 | ||||
| -rw-r--r-- | example_cs2pov.json | 31 | ||||
| -rwxr-xr-x | scripts/recorder | 17 |
5 files changed, 551 insertions, 23 deletions
@@ -6,3 +6,4 @@ .venv/ demos/ recordings/ +cs2pov.json diff --git a/cs2pov/cli.py b/cs2pov/cli.py index 537b665..09457e4 100644 --- a/cs2pov/cli.py +++ b/cs2pov/cli.py @@ -24,6 +24,10 @@ 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 @@ -752,50 +756,55 @@ def create_parser() -> argparse.ArgumentParser: 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=True, metavar="COMMAND") + 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", required=True, type=Path, + 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", required=True, + 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", required=True, type=Path, + 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="1920x1080", + recording_args.add_argument("-r", "--resolution", default=None, help="Recording resolution (default: 1920x1080)") - recording_args.add_argument("-f", "--framerate", type=int, default=60, + recording_args.add_argument("-f", "--framerate", type=int, default=None, help="Recording framerate (default: 60)") - recording_args.add_argument("--display", type=int, default=0, + recording_args.add_argument("--display", type=int, default=None, help="X display number (default: 0)") - recording_args.add_argument("--no-hud", action="store_true", + recording_args.add_argument("--no-hud", action="store_true", default=None, help="Hide HUD elements") - recording_args.add_argument("--no-audio", action="store_true", + recording_args.add_argument("--no-audio", action="store_true", default=None, help="Disable audio recording") - recording_args.add_argument("--audio-device", + recording_args.add_argument("--audio-device", default=None, help="PulseAudio device (auto-detected)") - recording_args.add_argument("--cs2-path", type=Path, + recording_args.add_argument("--cs2-path", type=Path, default=None, help="Custom CS2 installation path") - recording_args.add_argument("--tick-nav", action="store_true", + recording_args.add_argument("--tick-nav", action="store_true", 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="store_true", + verbose_args.add_argument("-v", "--verbose", action="store_true", default=None, help="Verbose output") # INFO command @@ -816,7 +825,7 @@ Examples: 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", + pov_parser.add_argument("--no-trim", action="store_true", default=None, help="Skip post-processing trim") # RECORD command @@ -847,6 +856,13 @@ Examples: trim_parser.add_argument("--startup-time", type=float, help="Override startup time (seconds from video start to demo start)") + # 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 @@ -854,6 +870,160 @@ Examples: # Command Handlers # ============================================================================= +def _validate_required_args(args, command: str) -> Optional[str]: + """Validate that required args (demo, player, output) 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 missing: + return f"Missing required arguments for '{command}': {', '.join(missing)}" + return None + + +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 + 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() @@ -887,8 +1057,8 @@ def cmd_info(args) -> int: return 0 -def cmd_pov(args) -> int: - """Handle 'pov' command - full recording pipeline.""" +def _run_single_pov(args) -> int: + """Run a single POV recording job (record + trim).""" # Check dependencies missing = check_dependencies() if missing: @@ -951,8 +1121,13 @@ def cmd_pov(args) -> int: return 0 if result.success else 1 -def cmd_record(args) -> int: - """Handle 'record' command - raw recording without trimming.""" +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: @@ -1006,8 +1181,36 @@ def cmd_record(args) -> int: 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 + + return _run_batch(args, "pov") + + def cmd_trim(args) -> int: """Handle 'trim' command - post-process existing video.""" + # Apply config defaults for verbose if not explicitly set + try: + config_path = find_config(getattr(args, "config", None)) + if config_path: + project = load_config(config_path) + if args.verbose is None: + args.verbose = project.defaults.get("verbose", False) + except ConfigError: + pass # Config errors are non-fatal for trim + if args.verbose is None: + args.verbose = False + video_path = args.video.resolve() if not video_path.exists(): print(f"Error: Video not found: {video_path}", file=sys.stderr) @@ -1085,7 +1288,11 @@ def main() -> int: args = parser.parse_args() try: - if args.command == "info": + 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) diff --git a/cs2pov/settings.py b/cs2pov/settings.py new file mode 100644 index 0000000..41b316f --- /dev/null +++ b/cs2pov/settings.py @@ -0,0 +1,280 @@ +"""Config file support for cs2pov. + +Loads cs2pov.json config files with defaults and batch job queues. +Priority: CLI explicit flag > job override > config defaults > hardcoded defaults. +""" + +import json +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any, Optional + +from .exceptions import CS2POVError + + +class ConfigError(CS2POVError): + """Invalid config file.""" + pass + + +# Maps 1:1 to CLI flags. These are the final fallback values. +HARDCODED_DEFAULTS: dict[str, Any] = { + "resolution": "1920x1080", + "framerate": 60, + "display": 0, + "no_hud": False, + "no_audio": False, + "audio_device": None, + "cs2_path": None, + "tick_nav": False, + "no_trim": False, + "verbose": False, +} + +# Keys valid in "defaults" and as job overrides +VALID_DEFAULT_KEYS = set(HARDCODED_DEFAULTS.keys()) + +# Keys required in each job +REQUIRED_JOB_KEYS = {"demo", "player", "output"} + +# Keys valid in a job entry (required + overridable + job-only) +VALID_JOB_KEYS = REQUIRED_JOB_KEYS | VALID_DEFAULT_KEYS | {"type"} + +# Valid job types +VALID_JOB_TYPES = {"pov", "record"} + +CONFIG_FILENAME = "cs2pov.json" +CURRENT_VERSION = 1 + + +@dataclass +class JobConfig: + """A single recording job from config.""" + demo: str + player: str + output: str + type: str = "pov" + overrides: dict[str, Any] = field(default_factory=dict) + + +@dataclass +class ProjectConfig: + """Parsed config file.""" + version: int + defaults: dict[str, Any] + jobs: list[JobConfig] + config_path: Path + + +def find_config(override_path: Optional[Path] = None) -> Optional[Path]: + """Find config file. Check override path, then cwd/cs2pov.json.""" + if override_path is not None: + path = Path(override_path).resolve() + if not path.exists(): + raise ConfigError(f"Config file not found: {path}") + return path + + cwd_config = Path.cwd() / CONFIG_FILENAME + if cwd_config.exists(): + return cwd_config + + return None + + +def load_config(path: Path) -> ProjectConfig: + """Parse and validate a cs2pov.json config file.""" + try: + raw = json.loads(path.read_text()) + except json.JSONDecodeError as e: + raise ConfigError(f"Invalid JSON in {path}: {e}") + + if not isinstance(raw, dict): + raise ConfigError(f"Config must be a JSON object, got {type(raw).__name__}") + + # Version check + version = raw.get("version") + if version is None: + raise ConfigError("Config missing 'version' field") + if not isinstance(version, int) or version < 1: + raise ConfigError(f"Invalid config version: {version}") + if version > CURRENT_VERSION: + raise ConfigError( + f"Config version {version} is newer than supported ({CURRENT_VERSION}). " + f"Please update cs2pov." + ) + + # Parse defaults + defaults_raw = raw.get("defaults", {}) + if not isinstance(defaults_raw, dict): + raise ConfigError("'defaults' must be an object") + + unknown_keys = set(defaults_raw.keys()) - VALID_DEFAULT_KEYS + if unknown_keys: + raise ConfigError(f"Unknown keys in 'defaults': {', '.join(sorted(unknown_keys))}") + + defaults = dict(defaults_raw) + + # Validate types in defaults + _validate_setting_types(defaults, "defaults") + + # Parse jobs + jobs_raw = raw.get("jobs", []) + if not isinstance(jobs_raw, list): + raise ConfigError("'jobs' must be an array") + + jobs: list[JobConfig] = [] + for i, job_raw in enumerate(jobs_raw): + if not isinstance(job_raw, dict): + raise ConfigError(f"Job {i+1} must be an object") + + missing = REQUIRED_JOB_KEYS - set(job_raw.keys()) + if missing: + raise ConfigError(f"Job {i+1} missing required keys: {', '.join(sorted(missing))}") + + unknown = set(job_raw.keys()) - VALID_JOB_KEYS + if unknown: + raise ConfigError(f"Job {i+1} has unknown keys: {', '.join(sorted(unknown))}") + + # Extract and validate job type + job_type = job_raw.get("type", "pov") + if not isinstance(job_type, str): + raise ConfigError(f"Job {i+1}: 'type' must be a string") + if job_type not in VALID_JOB_TYPES: + raise ConfigError( + f"Job {i+1}: invalid type '{job_type}', must be one of: {', '.join(sorted(VALID_JOB_TYPES))}" + ) + + overrides = {k: v for k, v in job_raw.items() if k not in REQUIRED_JOB_KEYS and k != "type"} + _validate_setting_types(overrides, f"job {i+1}") + + jobs.append(JobConfig( + demo=job_raw["demo"], + player=job_raw["player"], + output=job_raw["output"], + type=job_type, + overrides=overrides, + )) + + return ProjectConfig( + version=version, + defaults=defaults, + jobs=jobs, + config_path=path.resolve(), + ) + + +def _validate_setting_types(settings: dict[str, Any], context: str) -> None: + """Validate types of setting values.""" + type_checks: dict[str, tuple[type, ...]] = { + "resolution": (str,), + "framerate": (int,), + "display": (int,), + "no_hud": (bool,), + "no_audio": (bool,), + "audio_device": (str, type(None)), + "cs2_path": (str, type(None)), + "tick_nav": (bool,), + "no_trim": (bool,), + "verbose": (bool,), + } + + for key, value in settings.items(): + expected = type_checks.get(key) + if expected and not isinstance(value, expected): + expected_names = " or ".join(t.__name__ for t in expected) + raise ConfigError( + f"In {context}: '{key}' must be {expected_names}, got {type(value).__name__}" + ) + + +def resolve_job(job: JobConfig, defaults: dict[str, Any]) -> dict[str, Any]: + """Merge job overrides onto defaults. Returns flat dict with all keys.""" + merged = dict(defaults) + merged.update(job.overrides) + merged["demo"] = job.demo + merged["player"] = job.player + merged["output"] = job.output + return merged + + +def resolve_paths(job_dict: dict[str, Any], config_dir: Path) -> dict[str, Any]: + """Make relative paths in job_dict absolute against config_dir.""" + result = dict(job_dict) + + for key in ("demo", "output"): + if key in result and result[key] is not None: + p = Path(result[key]) + if not p.is_absolute(): + result[key] = str((config_dir / p).resolve()) + + if "cs2_path" in result and result["cs2_path"] is not None: + p = Path(result["cs2_path"]) + if not p.is_absolute(): + result["cs2_path"] = str((config_dir / p).resolve()) + + return result + + +def merge_args_with_config(args, job_dict: dict[str, Any]): + """Layer CLI > config > hardcoded onto args namespace. + + For each config-overridable key: + - If CLI explicitly set it (not None), keep CLI value + - Else if config provides it, use config value + - Else use hardcoded default + """ + # Keys that should be Path objects when set + path_keys = {"cs2_path"} + + for key, hardcoded in HARDCODED_DEFAULTS.items(): + cli_val = getattr(args, key, None) + config_val = job_dict.get(key) + + if cli_val is not None: + # CLI wins - already set + continue + elif config_val is not None: + if key in path_keys and isinstance(config_val, str): + setattr(args, key, Path(config_val)) + else: + setattr(args, key, config_val) + else: + setattr(args, key, hardcoded) + + # Set demo/player/output from job if not on CLI + for key in ("demo", "player", "output"): + 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"): + setattr(args, key, Path(job_val)) + else: + setattr(args, key, job_val) + + return args + + +def generate_default_config(cs2_path: Optional[str] = None) -> dict: + """Generate a template cs2pov.json config.""" + config: dict[str, Any] = { + "version": CURRENT_VERSION, + "defaults": { + "resolution": "1920x1080", + "framerate": 60, + "tick_nav": False, + "verbose": False, + }, + "jobs": [ + { + "demo": "./demos/example.dem", + "player": "PlayerName", + "output": "./recordings/example.mp4", + } + ], + } + + if cs2_path: + config["defaults"]["cs2_path"] = cs2_path + + return config diff --git a/example_cs2pov.json b/example_cs2pov.json new file mode 100644 index 0000000..b005bdc --- /dev/null +++ b/example_cs2pov.json @@ -0,0 +1,31 @@ +{ + "version": 1, + "defaults": { + "resolution": "1920x1080", + "framerate": 60, + "tick_nav": true, + "verbose": false, + "cs2_path": "/path/to/Counter-Strike Global Offensive" + }, + "jobs": [ + { + "type": "pov", + "demo": "./demos/match1.dem", + "player": "PlayerName", + "output": "./recordings/match1_pov.mp4" + }, + { + "type": "record", + "demo": "./demos/match2.dem", + "player": "PlayerName", + "output": "./recordings/match2_raw.mp4" + }, + { + "demo": "./demos/match3.dem", + "player": "PlayerName", + "output": "./recordings/match3_pov.mp4", + "no_hud": true, + "tick_nav": false + } + ] +} diff --git a/scripts/recorder b/scripts/recorder index f1831a1..80873bd 100755 --- a/scripts/recorder +++ b/scripts/recorder @@ -1,5 +1,14 @@ -# POV recorder -cs2pov pov -d ./demos/260224/nuke.dem -p "schark" -o ./recordings/260224/schark_nuke.mp4 --tick-nav -sleep 10 -cs2pov pov -d ./demos/260224/ancient.dem -p "schark" -o ./recordings/260224/schark_ancient.mp4 --tick-nav +#!/bin/bash +# DEPRECATED: Use cs2pov.json config file instead. +# +# Create a config: +# cs2pov init +# +# Edit cs2pov.json with your jobs, then run: +# cs2pov pov +# +# See cs2pov.json for the config format. +echo "This script is deprecated. Use 'cs2pov init' to create a cs2pov.json config," +echo "then run 'cs2pov pov' to process all jobs." +exit 1 |