festifun/backend/festival4d/cli.py

"""Command-line entrypoint: ``python -m festival4d <cmd>``.

FROZEN after foundation. Lanes fill in the bodies of the functions this dispatches
to (in ``ingest.py``, ``audio_sync.py``, ``sfm.py``, ``events_ai.py``); they never
edit this file or ``api.py``.

Subcommands: ``synthetic | ingest | sync | reconstruct | events | features | direct |
capsule | serve``.

Every subcommand dispatches to a single lane entrypoint. While a lane is still a stub,
its entrypoint raises :class:`NotImplementedError`; we catch that here and print a clear
"not implemented yet" message with a non-zero exit code, rather than a traceback. Once the
lane fills the body, the same dispatch runs it for real — no change to this file needed.
"""

from __future__ import annotations

import argparse
import logging
import sys

log = logging.getLogger("festival4d.cli")


def _cmd_synthetic(args: argparse.Namespace) -> int:
    from festival4d import synthetic

    synthetic.build()
    return 0


def _cmd_ingest(args: argparse.Namespace) -> int:
    from festival4d import ingest

    ingest.run_ingest()
    return 0


def _cmd_sync(args: argparse.Namespace) -> int:
    from festival4d import audio_sync

    audio_sync.run_sync()
    return 0


def _cmd_reconstruct(args: argparse.Namespace) -> int:
    from festival4d import sfm

    sfm.run_reconstruct()
    return 0


def _cmd_events(args: argparse.Namespace) -> int:
    from festival4d import events_ai

    result = events_ai.run_events(t_global_s=args.t_global_s, window_s=args.window_s)
    log.info("events: %s", result)
    return 0


def _cmd_features(args: argparse.Namespace) -> int:
    from festival4d import audio_features

    result = audio_features.run_features()
    log.info("features: %s", result)
    return 0


def _cmd_direct(args: argparse.Namespace) -> int:
    import json

    from festival4d import director

    path = director.generate_path(top_n=args.top_n, lead_s=args.lead_s)
    print(json.dumps(path, indent=2))
    return 0


def _cmd_capsule(args: argparse.Namespace) -> int:
    from festival4d import capsule

    result = capsule.build_capsule(out_dir=args.out)
    log.info("capsule: %s", result)
    return 0


def _cmd_serve(args: argparse.Namespace) -> int:
    import uvicorn

    uvicorn.run(
        "festival4d.api:app",
        host=args.host,
        port=args.port,
        reload=args.reload,
    )
    return 0


def build_parser() -> argparse.ArgumentParser:
    from festival4d import config

    parser = argparse.ArgumentParser(
        prog="festival4d",
        description="Festival 4D — synchronized, explorable 4D concert replay.",
    )
    sub = parser.add_subparsers(dest="command", required=True)

    p_syn = sub.add_parser("synthetic", help="generate the synthetic fixture project (M0)")
    p_syn.set_defaults(func=_cmd_synthetic)

    p_ing = sub.add_parser("ingest", help="probe videos + extract audio (lane A / M1)")
    p_ing.set_defaults(func=_cmd_ingest)

    p_sync = sub.add_parser("sync", help="GCC-PHAT audio alignment (lane A / M1)")
    p_sync.set_defaults(func=_cmd_sync)

    p_rec = sub.add_parser("reconstruct", help="COLMAP SfM + pose export (lane B / M2)")
    p_rec.set_defaults(func=_cmd_reconstruct)

    p_evt = sub.add_parser("events", help="audio candidates + AI classification (lane D / M7)")
    p_evt.add_argument("--t-global-s", dest="t_global_s", type=float, default=None,
                       help="detect within a single window centered here (default: whole track)")
    p_evt.add_argument("--window-s", dest="window_s", type=float, default=None,
                       help="window width in seconds when --t-global-s is given")
    p_evt.set_defaults(func=_cmd_events)

    p_feat = sub.add_parser("features", help="beat/onset analysis of the reference audio (lane E / M10)")
    p_feat.set_defaults(func=_cmd_features)

    p_dir = sub.add_parser("direct", help="auto-director: events -> camPath JSON on stdout (lane E / M11)")
    p_dir.add_argument("--top-n", dest="top_n", type=int, default=8,
                       help="number of top-confidence events to cover (default 8)")
    p_dir.add_argument("--lead-s", dest="lead_s", type=float, default=2.0,
                       help="seconds of lead-in before each event (default 2.0)")
    p_dir.set_defaults(func=_cmd_direct)

    p_cap = sub.add_parser("capsule", help="bake a zero-backend shareable bundle (lane G / M17)")
    p_cap.add_argument("--out", default=None, help="output dir (default data/capsule/)")
    p_cap.set_defaults(func=_cmd_capsule)

    p_srv = sub.add_parser("serve", help="run the FastAPI app (M3)")
    p_srv.add_argument("--host", default=config.API_HOST)
    p_srv.add_argument("--port", type=int, default=config.API_PORT)
    p_srv.add_argument("--reload", action="store_true", help="uvicorn autoreload (dev)")
    p_srv.set_defaults(func=_cmd_serve)

    return parser


def main(argv: list[str] | None = None) -> int:
    logging.basicConfig(
        level=logging.INFO,
        format="%(asctime)s %(levelname)-7s %(name)s: %(message)s",
    )
    parser = build_parser()
    args = parser.parse_args(argv)
    try:
        return args.func(args)
    except NotImplementedError as exc:
        detail = f" ({exc})" if str(exc) else ""
        print(
            f"festival4d {args.command}: not implemented yet{detail}.\n"
            f"This subcommand is owned by a feature lane that hasn't landed. "
            f"Run `python -m festival4d synthetic` for the working demo path.",
            file=sys.stderr,
        )
        return 2


if __name__ == "__main__":
    raise SystemExit(main())