settled-reach/tooling/domains/assets/router.py

"""Transport for the `assets` domain — args in, delegate, format out.

A verb's result is printed to stdout as JSON with the same keys the old
connector scripts printed, so a skill that reads `file`, `ogg_file` or
`size_bytes` keeps working. What changed is failure: it is a non-zero exit
with a remedy on the event stream, never `{"ok": false}` on stdout — so a
caller checks the exit status, not a field.
"""

from __future__ import annotations

import json
from pathlib import Path

import typer

from tooling.core import cli, console
from tooling.core.command import command
from tooling.core.errors import ReachError, unknown_choice

app = cli.domain("assets", "Asset generators — Stable Audio, Gemini images, Trellis 3D.")
audio_app = cli.domain("audio", "Stable Audio Open on tower-of-joy :11500 (kept OFF — VRAM).")
post_app = cli.domain("post", "ffmpeg post-processing: trim, normalize, convert.")
image_app = cli.domain("image", "Gemini image generation — every generate COSTS MONEY.")
trellis_app = cli.domain("trellis", "Trellis image-to-3D on tower-of-joy :11510 (kept OFF — VRAM).")


@app.callback()
def _domain() -> None:
    """Keeps `assets` a group (Typer collapses a single-command app)."""


@audio_app.callback()
def _audio() -> None:
    """Keeps `audio` a group."""


@post_app.callback()
def _post() -> None:
    """Keeps `post` a group."""


@image_app.callback()
def _image() -> None:
    """Keeps `image` a group."""


@trellis_app.callback()
def _trellis() -> None:
    """Keeps `trellis` a group."""


audio_app.add_typer(post_app, name="post")
app.add_typer(audio_app, name="audio")
app.add_typer(image_app, name="image")
app.add_typer(trellis_app, name="trellis")


def _emit(result: dict) -> None:
    console.out(json.dumps(result, indent=2))


def _emit_summary(summary: dict, what: str) -> None:
    """Print a batch summary, then fail the command if any item failed."""
    _emit(summary)
    if summary.get("failed"):
        raise ReachError(
            f"{what}: {summary['failed']} of {summary['total']} failed (details in the summary above)",
            fix="fix the failing items, then re-run with --skip-existing to leave the rest alone",
        )
    console.verdict(f"{what}: {summary.get('success', 0)} done, {summary.get('skipped', 0)} skipped")


# --- audio ----------------------------------------------------------------


@audio_app.command("health")
@command
def audio_health() -> None:
    """Is Stable Audio up? OFF is its normal resting state — this says which."""
    from tooling.domains.assets import audio

    _emit(audio.health())


@audio_app.command("generate")
@command
def audio_generate(
    prompt: str = typer.Argument(..., help="What the audio should sound like."),
    duration: float = typer.Option(10.0, "--duration", help="Seconds, 0-47."),
    steps: int = typer.Option(100, "--steps", help="Diffusion steps; fewer is faster and worse."),
    cfg: float = typer.Option(7.0, "--cfg", help="Classifier-free guidance scale."),
    output: Path = typer.Option(None, "--output", help="WAV path (default: named from the prompt)."),
    timeout: int = typer.Option(600, "--timeout", help="Seconds to wait for the result."),
    post: bool = typer.Option(False, "--post", help="Also trim + normalize + convert to OGG."),
    output_ogg: Path = typer.Option(None, "--output-ogg", help="OGG path; implies --post."),
) -> None:
    """Generate audio from a prompt; prints the result JSON."""
    from tooling.domains.assets import audio

    _emit(
        audio.generate(
            prompt,
            duration=duration,
            steps=steps,
            cfg=cfg,
            output=str(output) if output else None,
            timeout=timeout,
            post=post,
            output_ogg=str(output_ogg) if output_ogg else None,
        )
    )


@audio_app.command("batch")
@command
def audio_batch(
    manifest: Path = typer.Argument(..., help="Manifest .json (see the audio-gen skill)."),
    dry_run: bool = typer.Option(False, "--dry-run", help="List what would be generated."),
    only: str = typer.Option(None, "--only", help="Comma-separated asset ids."),
    skip_existing: bool = typer.Option(False, "--skip-existing", help="Leave existing OGGs alone."),
) -> None:
    """Generate every asset in a manifest; fails if any asset failed."""
    from tooling.domains.assets import audio_batch as service

    summary = service.run(
        str(manifest),
        dry_run=dry_run,
        only=set(only.split(",")) if only else None,
        skip_existing=skip_existing,
    )
    _emit_summary(summary, "audio batch")


@post_app.command("convert")
@command
def post_convert(
    input: Path = typer.Argument(..., help="Input WAV."),
    output: Path = typer.Option(None, "--output", "-o", help="Output (default: same name .ogg)."),
    quality: int = typer.Option(6, "--quality", "-q", help="Vorbis quality 0-10."),
) -> None:
    """WAV → OGG (libvorbis)."""
    from tooling.domains.assets import audio_post

    _emit(audio_post.convert(str(input), str(output) if output else None, quality))


@post_app.command("normalize")
@command
def post_normalize(
    input: Path = typer.Argument(..., help="Input audio file."),
    output: Path = typer.Option(None, "--output", "-o", help="Output (default: <name>_norm)."),
    lufs: float = typer.Option(-16, "--lufs", help="Target loudness, LUFS."),
) -> None:
    """LUFS-normalize an audio file."""
    from tooling.domains.assets import audio_post

    _emit(audio_post.normalize(str(input), str(output) if output else None, lufs))


@post_app.command("trim")
@command
def post_trim(
    input: Path = typer.Argument(..., help="Input audio file."),
    output: Path = typer.Option(None, "--output", "-o", help="Output (default: <name>_trimmed)."),
    threshold: int = typer.Option(-50, "--threshold", help="Silence threshold, dB."),
) -> None:
    """Trim leading and trailing silence."""
    from tooling.domains.assets import audio_post

    _emit(audio_post.trim(str(input), str(output) if output else None, threshold))


@post_app.command("pipeline")
@command
def post_pipeline(
    input: Path = typer.Argument(..., help="Input WAV."),
    output: Path = typer.Option(None, "--output", "-o", help="Output OGG (default: same name .ogg)."),
    quality: int = typer.Option(6, "--quality", "-q", help="Vorbis quality 0-10."),
    lufs: float = typer.Option(-16, "--lufs", help="Target loudness, LUFS."),
    threshold: int = typer.Option(-50, "--threshold", help="Silence threshold, dB."),
) -> None:
    """trim → normalize → convert to OGG, the full chain."""
    from tooling.domains.assets import audio_post

    _emit(audio_post.pipeline(str(input), str(output) if output else None, quality, lufs, threshold))


# --- image ----------------------------------------------------------------


@image_app.command("health")
@command
def image_health() -> None:
    """Is the Gemini API reachable with GEMINI_API_KEY? Free — lists models."""
    from tooling.domains.assets import image

    _emit(image.health())


@image_app.command("generate")
@command
def image_generate(
    prompt: str = typer.Argument(..., help="What to draw."),
    output: Path = typer.Option(None, "--output", help="PNG path (default: ~/Pictures/mcp-images/)."),
    aspect: str = typer.Option("1:1", "--aspect", help="Aspect ratio, e.g. 1:1, 16:9, 3:4."),
    size: str = typer.Option(None, "--size", help="Resolution hint (1K/2K/4K) — may be ignored."),
    input: Path = typer.Option(None, "--input", help="Source image, for image-to-image."),
) -> None:
    """Generate one image. COSTS MONEY per call."""
    from tooling.domains.assets import image

    if aspect not in image.ASPECT_RATIOS:
        raise unknown_choice("aspect ratio", aspect, image.ASPECT_RATIOS)
    _emit(
        image.generate(
            prompt,
            output=str(output) if output else None,
            aspect_ratio=aspect,
            image_size=size,
            input_image=str(input) if input else None,
        )
    )


# --- trellis --------------------------------------------------------------


@trellis_app.command("health")
@command
def trellis_health() -> None:
    """Is Trellis up? OFF is its normal resting state — this says which."""
    from tooling.domains.assets import trellis

    _emit(trellis.health())


@trellis_app.command("generate")
@command
def trellis_generate(
    image: Path = typer.Argument(..., help="Input image (PNG recommended)."),
    output: Path = typer.Option(None, "--output", help=".glb path (default: named from the image)."),
    simplify: float = typer.Option(0.95, "--simplify", help="Mesh simplification, 0.9-0.98."),
    texture_size: int = typer.Option(1024, "--texture-size", help="Texture resolution, 512-2048."),
    seed: int = typer.Option(0, "--seed", help="Random seed."),
    timeout: int = typer.Option(600, "--timeout", help="Seconds for the 3D step."),
) -> None:
    """Image → .glb; prints the result JSON."""
    from tooling.domains.assets import trellis

    _emit(
        trellis.generate(
            str(image),
            output=str(output) if output else None,
            simplify=simplify,
            texture_size=texture_size,
            seed=seed,
            timeout=timeout,
        )
    )


@trellis_app.command("batch")
@command
def trellis_batch(
    input_dir: Path = typer.Option(None, "--input-dir", help="Directory of .png inputs (default: the character bodies)."),
    output_dir: Path = typer.Option(None, "--output-dir", help="Where the .glb files go."),
    names: str = typer.Option(None, "--names", help="Comma-separated stems; default: every .png in --input-dir."),
    simplify: float = typer.Option(0.95, "--simplify", help="Mesh simplification."),
    texture_size: int = typer.Option(1024, "--texture-size", help="Texture resolution."),
    cooldown: float = typer.Option(15, "--cooldown", help="Seconds between successful jobs."),
    retries: int = typer.Option(3, "--retries", help="Attempts per item."),
    retry_delay: float = typer.Option(60, "--retry-delay", help="Seconds before a retry."),
) -> None:
    """One .glb per image, gently — long; consider --detach. Fails if any item failed."""
    from tooling.domains.assets import trellis

    summary = trellis.batch(
        str(input_dir) if input_dir else trellis.BODIES_INPUT,
        str(output_dir) if output_dir else trellis.BODIES_OUTPUT,
        names=names.split(",") if names else None,
        simplify=simplify,
        texture_size=texture_size,
        cooldown=cooldown,
        max_retries=retries,
        retry_delay=retry_delay,
    )
    _emit_summary(summary, "trellis batch")


# --- local synthesis ------------------------------------------------------


@app.command("synth-ui")
@command
def synth_ui(
    output_dir: Path = typer.Option(None, "--output-dir", help="Where the WAVs go (default: client/assets/audio)."),
) -> None:
    """Synthesize the four insert-tech UI sounds locally — no service needed."""
    from tooling.domains.assets import synth_ui as service

    paths = service.run(str(output_dir) if output_dir else service.OUTPUT_DIR)
    console.out("\n".join(paths))
    console.verdict(
        f"synth-ui: wrote {len(paths)} WAVs — convert with `reach assets audio post convert <file.wav>`"
    )