Source code

src/scripts/presentation/cli.py

This file is included in this documentation build. It is displayed as code and is not executed.

Download source

"""Provide portable setup, validation, and video generation commands.

@author Laurent Declercq l.declercq@agon-innovation.ch
@version 20260921
"""

import argparse
import os
from pathlib import Path
import subprocess
import sys
import venv

from pipeline import ROOT, SOURCE, build_language, load_project, output_paths

TOOLS = ROOT / ".build/presentation-tools"
PYTHON = TOOLS / ("Scripts/python.exe" if os.name == "nt" else "bin/python")


def main():
    """Validate CLI inputs and run the requested operation with isolated dependencies."""
    parser = argparse.ArgumentParser(
        description="Rebuild narrated Atrium videos from versioned slides and editable TXT narration."
    )
    parser.add_argument("command", choices=["setup", "check", "build"])
    parser.add_argument(
        "--locale",
        action="append",
        help="Language to process. Repeat for several languages. Defaults to all.",
    )
    parser.add_argument(
        "--source", type=Path, default=SOURCE, help="Presentation source directory."
    )
    parser.add_argument(
        "--output",
        type=Path,
        help="Final video directory. Defaults to src/showcase/assets.",
    )
    parser.add_argument(
        "--artifacts",
        type=Path,
        help="Project-only posters, captions and transcripts. Defaults to SOURCE/artifacts, or a sibling OUTPUT-artifacts directory for a custom --output.",
    )
    parser.add_argument(
        "--cache",
        type=Path,
        default=ROOT / ".build/presentation",
        help="Disposable content-addressed cache.",
    )
    parser.add_argument(
        "--offline",
        action="store_true",
        help="Forbid speech requests. All required speech must be cached.",
    )
    parser.add_argument(
        "--force", action="store_true", help="Regenerate all selected speech and clips."
    )
    args = parser.parse_args()
    if args.offline and args.force:
        parser.error("--offline and --force cannot be combined.")
    output, artifacts = output_paths(args.source, args.output, args.artifacts)
    if args.command == "setup":
        venv.EnvBuilder(with_pip=True).create(TOOLS)
        subprocess.run(
            [
                str(PYTHON),
                "-m",
                "pip",
                "install",
                "--disable-pip-version-check",
                "-r",
                str(Path(__file__).with_name("requirements.txt")),
            ],
            check=True,
        )
        print(f"Presentation tools installed in {TOOLS}.")
        return
    settings, project = load_project(args.source.resolve(), args.locale)
    if args.command == "check":
        print(
            f'Validated narration and 1920 × 1080 source slides for: {", ".join(project)}.'
        )
        return
    if Path(sys.prefix).resolve() != TOOLS.resolve():
        if not PYTHON.is_file():
            parser.error("Run yarn presentation:setup once before building videos.")
        result = subprocess.run(
            [str(PYTHON), str(Path(__file__).resolve()), *sys.argv[1:]]
        )
        raise SystemExit(result.returncode)
    import imageio_ffmpeg

    ffmpeg = os.environ.get("ATRIUM_FFMPEG") or imageio_ffmpeg.get_ffmpeg_exe()
    args.cache.mkdir(parents=True, exist_ok=True)
    # A held cache lock prevents concurrent encoders from replacing shared artifacts.
    lock = args.cache / ".lock"
    try:
        descriptor = os.open(lock, os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0o600)
    except FileExistsError:
        parser.error(
            f"A presentation build is already using {args.cache}. Remove {lock} only after confirming the earlier process has ended."
        )
    try:
        os.write(descriptor, str(os.getpid()).encode())
        for locale, slides in project.items():
            build_language(
                locale,
                slides,
                settings,
                output,
                artifacts,
                args.cache.resolve(),
                ffmpeg,
                args.offline,
                args.force,
            )
    finally:
        os.close(descriptor)
        lock.unlink()


if __name__ == "__main__":
    try:
        main()
    except (OSError, ValueError, RuntimeError, subprocess.CalledProcessError) as error:
        print(f"Presentation build failed: {error}", file=sys.stderr)
        raise SystemExit(1) from error