cli

pyproject.toml

@@ -66,9 +66,8 @@ Homepage = "https://git.levineuwirth.org/neuwirth/neuropose"
 Repository = "https://git.levineuwirth.org/neuwirth/neuropose"
 Issues = "https://git.levineuwirth.org/neuwirth/neuropose/issues"
 
-# CLI entry point will be wired up in commit 7 once src/neuropose/cli.py lands.
+[project.scripts]
-# [project.scripts]
+neuropose = "neuropose.cli:run"
-# neuropose = "neuropose.cli:app"
 
 # ---------------------------------------------------------------------------
 # Dependency groups (PEP 735). Install a group with `uv sync --group dev` or

src/neuropose/cli.py

@@ -0,0 +1,265 @@
+"""NeuroPose command-line interface.
+
+Three subcommands:
+
+- ``neuropose watch`` — run the :class:`~neuropose.interfacer.Interfacer`
+  daemon against the configured input directory.
+- ``neuropose process <video>`` — run the estimator on a single video and
+  write the predictions JSON to disk.
+- ``neuropose analyze <results>`` — stubbed placeholder pending the
+  analyzer rewrite in commit 10.
+
+User-facing error handling
+--------------------------
+This module takes responsibility for turning internal exceptions into
+clear, short messages on stderr and non-zero exit codes. Users of the CLI
+should never see a raw Python traceback for expected failure modes:
+
+===============================  ===============  ==========================
+Exception                        Exit code        User-facing message
+===============================  ===============  ==========================
+``FileNotFoundError`` (config)   2                "config file not found: ..."
+``ValidationError`` (config)     2                "invalid config: ..."
+``AlreadyRunningError``          2                "another daemon is running"
+``NotImplementedError``          3                "pending commit N: ..."
+``KeyboardInterrupt``            130              (silent, matches shell)
+===============================  ===============  ==========================
+
+Internal errors (bugs) still surface as tracebacks — we only catch the
+exception classes we expect from the layers below.
+"""
+
+from __future__ import annotations
+
+import logging
+from pathlib import Path
+from typing import Annotated
+
+import typer
+from pydantic import ValidationError
+
+from neuropose import __version__
+from neuropose.config import Settings
+from neuropose.estimator import Estimator
+from neuropose.interfacer import AlreadyRunningError, Interfacer
+from neuropose.io import save_video_predictions
+
+logger = logging.getLogger(__name__)
+
+# Exit codes. Kept as module constants so tests can import and compare.
+EXIT_OK = 0
+EXIT_USAGE = 2
+EXIT_PENDING = 3
+EXIT_INTERRUPTED = 130
+
+app = typer.Typer(
+    name="neuropose",
+    help="NeuroPose — 3D human pose estimation pipeline.",
+    no_args_is_help=True,
+    add_completion=False,
+)
+
+
+# ---------------------------------------------------------------------------
+# Global callback
+# ---------------------------------------------------------------------------
+
+
+def _version_callback(value: bool) -> None:
+    """Print the package version and exit, when ``--version`` is given."""
+    if value:
+        typer.echo(f"neuropose {__version__}")
+        raise typer.Exit()
+
+
+def _configure_logging(verbose: bool, quiet: bool) -> None:
+    """Set up the root logger based on ``--verbose``/``--quiet`` flags."""
+    if verbose and quiet:
+        raise typer.BadParameter("--verbose and --quiet are mutually exclusive")
+    if verbose:
+        level = logging.DEBUG
+    elif quiet:
+        level = logging.WARNING
+    else:
+        level = logging.INFO
+    logging.basicConfig(
+        level=level,
+        format="%(asctime)s %(levelname)-8s %(name)s: %(message)s",
+        datefmt="%Y-%m-%d %H:%M:%S",
+        force=True,
+    )
+
+
+def _load_settings(config: Path | None) -> Settings:
+    """Load settings from ``config`` or from env vars / defaults."""
+    if config is None:
+        try:
+            return Settings()
+        except ValidationError as exc:
+            typer.echo(f"error: invalid environment configuration: {exc}", err=True)
+            raise typer.Exit(code=EXIT_USAGE) from exc
+    try:
+        return Settings.from_yaml(config)
+    except FileNotFoundError as exc:
+        typer.echo(f"error: config file not found: {config}", err=True)
+        raise typer.Exit(code=EXIT_USAGE) from exc
+    except ValidationError as exc:
+        typer.echo(f"error: invalid config {config}: {exc}", err=True)
+        raise typer.Exit(code=EXIT_USAGE) from exc
+    except ValueError as exc:
+        typer.echo(f"error: invalid config {config}: {exc}", err=True)
+        raise typer.Exit(code=EXIT_USAGE) from exc
+
+
+@app.callback()
+def main(
+    ctx: typer.Context,
+    config: Annotated[
+        Path | None,
+        typer.Option(
+            "--config",
+            "-c",
+            help="Path to a YAML configuration file. If omitted, configuration "
+            "is read from NEUROPOSE_* environment variables and defaults.",
+            exists=False,
+            dir_okay=False,
+            readable=True,
+        ),
+    ] = None,
+    verbose: Annotated[
+        bool,
+        typer.Option("--verbose", "-v", help="Enable debug logging."),
+    ] = False,
+    quiet: Annotated[
+        bool,
+        typer.Option("--quiet", "-q", help="Suppress info logging."),
+    ] = False,
+    version: Annotated[
+        bool,
+        typer.Option(
+            "--version",
+            help="Show the package version and exit.",
+            is_eager=True,
+            callback=_version_callback,
+        ),
+    ] = False,
+) -> None:
+    """NeuroPose — 3D human pose estimation pipeline."""
+    del version  # Handled eagerly in _version_callback.
+    _configure_logging(verbose, quiet)
+    ctx.obj = _load_settings(config)
+
+
+# ---------------------------------------------------------------------------
+# watch
+# ---------------------------------------------------------------------------
+
+
+@app.command()
+def watch(ctx: typer.Context) -> None:
+    """Run the NeuroPose job-processing daemon.
+
+    Watches the configured input directory for new job subdirectories,
+    processes each one in order, and persists status to disk. Blocks until
+    SIGINT, SIGTERM, or another instance of the daemon takes over.
+    """
+    settings: Settings = ctx.obj
+    estimator = Estimator(
+        device=settings.device,
+        default_fov_degrees=settings.default_fov_degrees,
+    )
+    interfacer = Interfacer(settings=settings, estimator=estimator)
+    try:
+        interfacer.run()
+    except AlreadyRunningError as exc:
+        typer.echo(f"error: {exc}", err=True)
+        raise typer.Exit(code=EXIT_USAGE) from exc
+    except NotImplementedError as exc:
+        typer.echo(f"error: {exc}", err=True)
+        raise typer.Exit(code=EXIT_PENDING) from exc
+    except KeyboardInterrupt as exc:
+        raise typer.Exit(code=EXIT_INTERRUPTED) from exc
+
+
+# ---------------------------------------------------------------------------
+# process
+# ---------------------------------------------------------------------------
+
+
+@app.command()
+def process(
+    ctx: typer.Context,
+    video: Annotated[
+        Path,
+        typer.Argument(
+            exists=True,
+            file_okay=True,
+            dir_okay=False,
+            readable=True,
+            help="Path to an input video file (.mp4, .avi, .mov, .mkv).",
+        ),
+    ],
+    output: Annotated[
+        Path | None,
+        typer.Option(
+            "--output",
+            "-o",
+            help="Where to write the predictions JSON. Defaults to "
+            "<video-stem>_predictions.json in the current working directory.",
+            dir_okay=False,
+            writable=True,
+        ),
+    ] = None,
+) -> None:
+    """Run pose estimation on a single video and write the JSON result.
+
+    Convenience entry point for one-off processing outside the daemon
+    workflow. The ``watch`` subcommand is the right choice for batch
+    processing of job directories.
+    """
+    settings: Settings = ctx.obj
+    estimator = Estimator(
+        device=settings.device,
+        default_fov_degrees=settings.default_fov_degrees,
+    )
+    try:
+        estimator.load_model(cache_dir=settings.model_cache_dir)
+    except NotImplementedError as exc:
+        typer.echo(f"error: {exc}", err=True)
+        raise typer.Exit(code=EXIT_PENDING) from exc
+    result = estimator.process_video(video)
+    out_path = output if output is not None else Path.cwd() / f"{video.stem}_predictions.json"
+    save_video_predictions(out_path, result.predictions)
+    typer.echo(f"wrote {out_path} ({result.frame_count} frames)")
+
+
+# ---------------------------------------------------------------------------
+# analyze (stub)
+# ---------------------------------------------------------------------------
+
+
+@app.command()
+def analyze(
+    ctx: typer.Context,
+    results: Annotated[
+        Path,
+        typer.Argument(help="Path to a results.json produced by watch or process."),
+    ],
+) -> None:
+    """Run the analyzer subpackage against a results.json (pending commit 10)."""
+    del ctx, results
+    typer.echo(
+        "error: the analyzer subpackage is pending commit 10. "
+        "Until it lands, use neuropose.io to load results.json from Python.",
+        err=True,
+    )
+    raise typer.Exit(code=EXIT_PENDING)
+
+
+def run() -> None:
+    """Entry point referenced by ``pyproject.toml``'s ``[project.scripts]``."""
+    app()
+
+
+if __name__ == "__main__":
+    run()

tests/unit/test_cli.py

@@ -0,0 +1,207 @@
+"""Tests for :mod:`neuropose.cli`.
+
+Uses ``typer.testing.CliRunner`` to invoke the CLI in-process and assert on
+exit codes and captured output. The autouse ``_isolate_environment`` fixture
+in ``conftest.py`` ensures each invocation uses an isolated ``$HOME`` /
+``$XDG_DATA_HOME`` so the daemon's lock file and data dirs cannot bleed
+between tests.
+
+The tests deliberately check ``result.output`` (combined stdout + stderr)
+rather than separate ``result.stderr``, because click's ``mix_stderr``
+plumbing varies across 8.x releases and we want the tests to be version-
+robust.
+"""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+import pytest
+import yaml
+from typer.testing import CliRunner
+
+from neuropose import __version__
+from neuropose.cli import (
+    EXIT_INTERRUPTED,
+    EXIT_OK,
+    EXIT_PENDING,
+    EXIT_USAGE,
+    app,
+)
+
+
+@pytest.fixture
+def runner() -> CliRunner:
+    """Return a default CliRunner.
+
+    We do NOT set ``mix_stderr=False`` because that parameter's semantics
+    changed between click 8.1 and 8.2. The default behaviour — combined
+    output on ``result.output`` — is stable and good enough for the
+    assertions this test suite makes.
+    """
+    return CliRunner()
+
+
+# ---------------------------------------------------------------------------
+# Top-level options
+# ---------------------------------------------------------------------------
+
+
+class TestTopLevelOptions:
+    def test_version_flag(self, runner: CliRunner) -> None:
+        result = runner.invoke(app, ["--version"])
+        assert result.exit_code == EXIT_OK
+        assert __version__ in result.output
+
+    def test_no_args_shows_help(self, runner: CliRunner) -> None:
+        result = runner.invoke(app, [])
+        # Typer's ``no_args_is_help`` exits with a help message. The exit
+        # code convention varies between click versions; accept either
+        # success or a usage-style code.
+        assert result.exit_code in (EXIT_OK, EXIT_USAGE)
+        assert "watch" in result.output
+        assert "process" in result.output
+        assert "analyze" in result.output
+
+    def test_help_flag(self, runner: CliRunner) -> None:
+        result = runner.invoke(app, ["--help"])
+        assert result.exit_code == EXIT_OK
+        assert "NeuroPose" in result.output
+
+    def test_subcommand_help(self, runner: CliRunner) -> None:
+        for subcommand in ("watch", "process", "analyze"):
+            result = runner.invoke(app, [subcommand, "--help"])
+            assert result.exit_code == EXIT_OK, f"{subcommand} --help failed"
+
+    def test_verbose_and_quiet_are_mutually_exclusive(
+        self, runner: CliRunner
+    ) -> None:
+        result = runner.invoke(app, ["--verbose", "--quiet", "watch"])
+        assert result.exit_code != EXIT_OK
+
+
+# ---------------------------------------------------------------------------
+# --config handling
+# ---------------------------------------------------------------------------
+
+
+class TestConfigOption:
+    def test_missing_config_file(self, runner: CliRunner, tmp_path: Path) -> None:
+        result = runner.invoke(app, ["--config", str(tmp_path / "nope.yaml"), "watch"])
+        assert result.exit_code == EXIT_USAGE
+        # typer validates file existence via the Option's exists/readable
+        # machinery, so the error message comes from click, not our handler.
+        # We just verify it mentions the missing file's name.
+        assert "nope.yaml" in result.output
+
+    def test_invalid_config_yaml_structure(
+        self, runner: CliRunner, tmp_path: Path
+    ) -> None:
+        path = tmp_path / "bad.yaml"
+        path.write_text("- not a mapping\n- another item\n")
+        result = runner.invoke(app, ["--config", str(path), "watch"])
+        assert result.exit_code == EXIT_USAGE
+        lowered = result.output.lower()
+        assert "invalid config" in lowered or "mapping" in lowered
+
+    def test_invalid_config_field(self, runner: CliRunner, tmp_path: Path) -> None:
+        path = tmp_path / "bad.yaml"
+        path.write_text(yaml.safe_dump({"device": "cpu"}))
+        result = runner.invoke(app, ["--config", str(path), "watch"])
+        assert result.exit_code == EXIT_USAGE
+        assert "invalid config" in result.output.lower()
+
+    def test_valid_config_reaches_subcommand(
+        self, runner: CliRunner, tmp_path: Path
+    ) -> None:
+        # A valid config should flow through the callback and into the
+        # subcommand. For ``watch``, the subcommand will then fail on the
+        # model load (NotImplementedError from the commit-11 stub), which
+        # is the behaviour we want to observe here — exit code
+        # ``EXIT_PENDING``.
+        data_dir = tmp_path / "data"
+        path = tmp_path / "good.yaml"
+        path.write_text(
+            yaml.safe_dump(
+                {
+                    "device": "/CPU:0",
+                    "data_dir": str(data_dir),
+                    "model_cache_dir": str(tmp_path / "models"),
+                }
+            )
+        )
+        result = runner.invoke(app, ["--config", str(path), "watch"])
+        assert result.exit_code == EXIT_PENDING
+        assert "commit 11" in result.output
+
+
+# ---------------------------------------------------------------------------
+# watch
+# ---------------------------------------------------------------------------
+
+
+class TestWatch:
+    def test_without_model_exits_pending(self, runner: CliRunner) -> None:
+        """The commit-11 stub raises NotImplementedError on model load.
+
+        The CLI should catch it and exit with ``EXIT_PENDING`` and a message
+        pointing at the pending commit.
+        """
+        result = runner.invoke(app, ["watch"])
+        assert result.exit_code == EXIT_PENDING
+        assert "commit 11" in result.output
+
+
+# ---------------------------------------------------------------------------
+# process
+# ---------------------------------------------------------------------------
+
+
+class TestProcess:
+    def test_missing_video_exits_usage(
+        self, runner: CliRunner, tmp_path: Path
+    ) -> None:
+        result = runner.invoke(app, ["process", str(tmp_path / "nope.mp4")])
+        # click's path existence check fires before our callback, so the
+        # exit is a usage error.
+        assert result.exit_code == EXIT_USAGE
+
+    def test_existing_video_without_model_exits_pending(
+        self,
+        runner: CliRunner,
+        synthetic_video: Path,
+    ) -> None:
+        result = runner.invoke(app, ["process", str(synthetic_video)])
+        assert result.exit_code == EXIT_PENDING
+        assert "commit 11" in result.output
+
+
+# ---------------------------------------------------------------------------
+# analyze
+# ---------------------------------------------------------------------------
+
+
+class TestAnalyze:
+    def test_analyze_stub_exits_with_pending_message(
+        self, runner: CliRunner, tmp_path: Path
+    ) -> None:
+        results_path = tmp_path / "results.json"
+        results_path.write_text("{}")
+        result = runner.invoke(app, ["analyze", str(results_path)])
+        assert result.exit_code == EXIT_PENDING
+        assert "commit 10" in result.output
+
+    def test_analyze_requires_an_argument(self, runner: CliRunner) -> None:
+        result = runner.invoke(app, ["analyze"])
+        assert result.exit_code == EXIT_USAGE
+
+
+# ---------------------------------------------------------------------------
+# Exit-code module constants
+# ---------------------------------------------------------------------------
+
+
+class TestExitCodes:
+    def test_exit_codes_are_distinct(self) -> None:
+        codes = {EXIT_OK, EXIT_USAGE, EXIT_PENDING, EXIT_INTERRUPTED}
+        assert len(codes) == 4