Vibe voice ASR 7B audio transcription utility for MLX audio

vibetranscribe.py

#!/usr/bin/env python3
# /// script
# requires-python = ">=3.10"
# dependencies = ["mlx-audio==0.5.1", "numpy", "soundfile>=0.13.1", "soxr"]
# ///
"""Transcribe a long recording with MLX VibeVoice-ASR into grouped Markdown.

    uv run --python 3.12 vibetranscribe.py meeting.flac --dry-run
    uv run --python 3.12 vibetranscribe.py meeting.flac -o meeting.md
    uv run --python 3.12 vibetranscribe.py meeting.flac \
        --speaker-names '1:0=Peter' '2:1=Peter'

The default is the 8-bit, non-streaming VibeVoice-ASR checkpoint, with roughly
10-minute chunks. Quiet boundaries can move by up to 15 seconds; every actual
chunk stays below the backend's 59-minute trimming limit. No source frames are
omitted or overlapped. Only one chunk, not the entire recording, is decoded
into memory for inference.

Speaker IDs are local to each chunk, including ID 0. Unmapped speakers remain
part-scoped in statistics. Names change presentation only; context and hotwords
change recognition. Completed, clean chunks are reused on identical reruns.
Changed recognition settings invalidate the affected cache entries. Failed
chunks are attempted again on the next run, never retried within the same run.

The token budget is a duration-based heuristic, not a completeness guarantee.
Malformed output, invalid timestamps, empty results, and exhausted token budgets
are visible in the transcript and cause exit status 1. Raw output is preserved
in the chunk cache and, on a problem, OUTPUT.raw.txt. A clean exit establishes
structural success, not transcription accuracy or complete speech coverage.
"""

from __future__ import annotations

import argparse
import json
import logging
import math
import platform
import re
import sys
import time
from collections.abc import Sequence
from dataclasses import asdict, dataclass, field
from importlib.metadata import PackageNotFoundError, version
from pathlib import Path
from typing import Any

import numpy as np
import soundfile as sf
import soxr

__version__ = "3.0.0"
SAMPLE_RATE = 24_000
MAX_SECONDS = 59 * 60
DEFAULT_MODEL = "mlx-community/VibeVoice-ASR-8bit"
logger = logging.getLogger("vibetranscribe")
SpeakerMap = dict[tuple[int | None, str], str]


@dataclass(frozen=True)
class Chunk:
    index: int
    start_frame: int
    end_frame: int
    sample_rate: int

    @property
    def start(self) -> float:
        return self.start_frame / self.sample_rate

    @property
    def duration(self) -> float:
        return (self.end_frame - self.start_frame) / self.sample_rate

    @property
    def end(self) -> float:
        return self.end_frame / self.sample_rate


@dataclass
class Utterance:
    start: float | None
    end: float | None
    speaker: str
    text: str
    chunk: int


@dataclass
class Result:
    chunk: Chunk
    raw_text: str
    generation_tokens: int
    max_tokens: int
    elapsed: float = 0.0
    peak_gib: float = 0.0
    utterances: list[Utterance] = field(default_factory=list)
    issues: list[str] = field(default_factory=list)
    cached: bool = False


@dataclass(frozen=True)
class Budget:
    tokens_per_minute: int
    floor: int
    ceiling: int

    def for_duration(self, seconds: float) -> int:
        wanted = math.ceil(seconds / 60 * self.tokens_per_minute) + 1024
        return max(self.floor, min(self.ceiling, wanted))


def read_mono(audio: sf.SoundFile, start: int, end: int) -> np.ndarray:
    """Read a frame-exact interval without decoding the rest of the source."""
    audio.seek(start)
    data = audio.read(end - start, dtype="float32", always_2d=True)
    return data.mean(axis=1, dtype=np.float32)


def quiet_boundary(audio: sf.SoundFile, target: int, lower: int, upper: int) -> int:
    """Choose a quiet 100 ms frame, breaking ties toward the nominal boundary."""
    if upper - lower < audio.samplerate // 5:
        return min(max(target, lower), upper)
    data = read_mono(audio, lower, upper)
    frame = max(1, audio.samplerate // 10)
    count = len(data) // frame
    frames = data[: count * frame].reshape(count, frame)
    energy = np.einsum("ij,ij->i", frames, frames) / frame
    threshold = float(energy.min()) + max(
        1e-9, 0.05 * float(energy.max() - energy.min())
    )
    candidates = np.flatnonzero(energy <= threshold)
    centers = lower + candidates * frame + frame // 2
    return int(centers[np.argmin(np.abs(centers - target))])


def plan_chunks(audio: sf.SoundFile, minutes: float) -> list[Chunk]:
    """Keep sample-exact coverage and constrain seams before seeking quietness."""
    total, rate = len(audio), audio.samplerate
    target_size = max(1, int(minutes * 60 * rate))
    count = (total + target_size - 1) // target_size
    # Leave one millisecond for resampling rounding at the backend's limit.
    maximum = int((MAX_SECONDS - 0.001) * rate)
    count = max(count, (total + maximum - 1) // maximum)
    radius = min(15 * rate, target_size // 4)
    boundaries = [0]
    for index in range(1, count):
        remaining = count - index
        target = round(total * index / count)
        # Each choice must leave enough room for all remaining chunks.
        lower = max(boundaries[-1] + 1, total - remaining * maximum)
        upper = min(boundaries[-1] + maximum, total - remaining)
        target = min(max(target, lower), upper)
        boundary = quiet_boundary(
            audio, target, max(lower, target - radius), min(upper, target + radius)
        )
        boundaries.append(boundary)
    boundaries.append(total)
    return [
        Chunk(i, start, end, rate)
        for i, (start, end) in enumerate(zip(boundaries, boundaries[1:]))
    ]


def read_chunk(audio: sf.SoundFile, chunk: Chunk) -> np.ndarray:
    data = read_mono(audio, chunk.start_frame, chunk.end_frame)
    if audio.samplerate != SAMPLE_RATE:
        data = soxr.resample(data, audio.samplerate, SAMPLE_RATE, quality="HQ")
    return np.ascontiguousarray(data, dtype=np.float32)


def to_seconds(value: Any) -> float | None:
    """Accept numeric seconds or MM:SS/HH:MM:SS; never accept NaN or infinity."""
    if isinstance(value, bool) or not isinstance(value, (int, float, str)):
        return None
    if isinstance(value, str):
        text = value.strip().removesuffix("s").strip()
        if ":" in text:
            pieces = text.split(":")
            if len(pieces) not in (2, 3):
                return None
            try:
                seconds = float(pieces[-1])
                minutes = int(pieces[-2])
                hours = int(pieces[0]) if len(pieces) == 3 else 0
            except ValueError:
                return None
            if not 0 <= seconds < 60 or minutes < 0 or hours < 0:
                return None
            if len(pieces) == 3 and minutes >= 60:
                return None
            value = hours * 3600 + minutes * 60 + seconds
        else:
            value = text
    try:
        result = float(value)
    except (ValueError, OverflowError):
        return None
    return result if math.isfinite(result) else None


def speaker_id(value: Any) -> str:
    if value is None or isinstance(value, bool):
        return "?"
    text = str(value).strip()
    match = re.fullmatch(r"(?:Speaker\s+)?(\d+)", text, re.IGNORECASE)
    return f"Speaker {int(match[1])}" if match else text or "?"


def parse_output(raw: str, chunk: Chunk) -> tuple[list[Utterance], list[str]]:
    """Parse raw JSON independently of mlx-audio's quote-unaware bracket scan."""
    text = raw.strip()
    # Accept a conventional code fence or a prose preamble. Decode only from
    # the first container; do not salvage inner objects from a broken array.
    if text.startswith("```"):
        _, _, text = text.partition("\n")
    positions = [p for p in (text.find("["), text.find("{")) if p >= 0]
    if not positions:
        return [], ["No JSON transcript returned."]
    text = text[min(positions) :]
    try:
        payload, stop = json.JSONDecoder().raw_decode(text)
    except json.JSONDecodeError as exc:
        return [], [f"Malformed JSON at character {exc.pos}: {exc.msg}."]
    issues: list[str] = []
    if text[stop:].strip() not in ("", "```"):
        issues.append("Extra output follows the JSON transcript; inspect raw output.")
    if isinstance(payload, dict):
        payload = [payload]
    if not isinstance(payload, list):
        return [], issues + ["The JSON transcript is not an array of utterances."]
    utterances: list[Utterance] = []
    for index, row in enumerate(payload, 1):
        if not isinstance(row, dict):
            issues.append(f"Entry {index} is not an utterance object.")
            continue
        content = row.get("Content", row.get("text"))
        if not isinstance(content, str) or not content.strip():
            issues.append(f"Entry {index} has no nonempty text; see raw output.")
            continue
        start = to_seconds(row.get("Start time", row.get("Start", row.get("start"))))
        end = to_seconds(row.get("End time", row.get("End", row.get("end"))))
        if (
            start is None
            or end is None
            or not 0 <= start <= end <= chunk.duration + 0.1
        ):
            issues.append(f"Entry {index} has invalid timestamps; shown as unknown.")
            start = end = None
        else:
            # The model rounds seconds. Permit only 100 ms of endpoint rounding.
            start = chunk.start + min(start, chunk.duration)
            end = chunk.start + min(end, chunk.duration)
        speaker = speaker_id(
            row.get("Speaker ID", row.get("Speaker", row.get("speaker_id")))
        )
        if speaker == "?":
            issues.append(f"Entry {index} has no speaker ID; shown as unknown.")
        utterances.append(Utterance(start, end, speaker, content.strip(), chunk.index))
    if not utterances:
        issues.append(
            "No utterances returned; this does not establish that the audio is silent."
        )
    return utterances, issues


def make_result(
    chunk: Chunk,
    raw_text: str,
    generation_tokens: int,
    max_tokens: int,
    elapsed: float = 0.0,
    peak_gib: float = 0.0,
    cached: bool = False,
) -> Result:
    utterances, issues = parse_output(raw_text, chunk)
    if generation_tokens >= max_tokens:
        issues.insert(
            0, "Generation reached its token cap; completion is not confirmed."
        )
    return Result(
        chunk,
        raw_text,
        generation_tokens,
        max_tokens,
        elapsed,
        peak_gib,
        utterances,
        issues,
        cached,
    )


def parse_speaker_names(specs: Sequence[str]) -> SpeakerMap:
    mapping: SpeakerMap = {}
    for spec in specs:
        key, separator, name = spec.partition("=")
        key, name = key.strip(), name.strip()
        if not separator or not key or not name:
            raise ValueError(f"expected [PART:]ID=NAME, got {spec!r}")
        part: int | None = None
        if ":" in key:
            prefix, key = key.split(":", 1)
            if not prefix.strip().isdigit() or int(prefix) < 1 or not key.strip():
                raise ValueError(f"invalid one-based part number in {spec!r}")
            part = int(prefix) - 1
        mapping[(part, speaker_id(key))] = name
    return mapping


def resolve_label(
    speaker: str, part: int, names: SpeakerMap, scoped: bool = False
) -> str:
    name = names.get((part, speaker), names.get((None, speaker)))
    return name or (f"Part {part + 1} / {speaker}" if scoped else speaker)


def group_turns(utterances: Sequence[Utterance]) -> list[list[Utterance]]:
    turns: list[list[Utterance]] = []
    for utterance in utterances:
        if (
            turns
            and turns[-1][0].speaker == utterance.speaker
            and turns[-1][0].chunk == utterance.chunk
        ):
            turns[-1].append(utterance)
        else:
            turns.append([utterance])
    return turns


def timestamp(seconds: float | None, hours: bool = True) -> str:
    if seconds is None:
        return "unknown"
    whole = int(seconds)
    h, remainder = divmod(whole, 3600)
    m, s = divmod(remainder, 60)
    return f"{h:02d}:{m:02d}:{s:02d}" if hours else f"{h * 60 + m:02d}:{s:02d}"


def markdown_label(text: str) -> str:
    text = " ".join(text.splitlines())
    return re.sub(r"([\\`*_{}\[\]<>|#])", r"\\\1", text)


def speaker_stats(
    results: Sequence[Result], names: SpeakerMap
) -> list[tuple[str, int, float]]:
    totals: dict[str, list[Any]] = {}
    for result in results:
        for turn in group_turns(result.utterances):
            first = turn[0]
            label = resolve_label(first.speaker, first.chunk, names, scoped=True)
            entry = totals.setdefault(label, [0, 0.0])
            entry[0] += 1
            entry[1] += sum(
                u.end - u.start
                for u in turn
                if u.start is not None and u.end is not None
            )
    return sorted(
        [(label, count, duration) for label, (count, duration) in totals.items()],
        key=lambda item: -item[2],
    )


def render_markdown(
    results: Sequence[Result],
    chunks: Sequence[Chunk],
    source: Path,
    model: str,
    names: SpeakerMap,
) -> str:
    complete = len(results) == len(chunks)
    problems = any(r.issues for r in results)
    status = (
        "incomplete" if not complete else "needs review" if problems else "processed"
    )
    quote = json.dumps
    lines = [
        "---",
        f"source: {quote(source.name, ensure_ascii=False)}",
        f"model: {quote(model)}",
        f"duration: {quote(timestamp(chunks[-1].end))}",
        f"status: {quote(status)}",
        f"processed_parts: {len(results)}",
        f"planned_parts: {len(chunks)}",
        "---",
        "",
        f"# Transcript — {markdown_label(source.stem)}",
        "",
        f"**Status: {status}. Processed {len(results)} of {len(chunks)} parts.**",
        "",
        "Timestamps and speakers are model predictions. Structural success does not "
        "prove that every spoken word was transcribed.",
        "",
    ]
    if len(chunks) > 1:
        lines.extend(
            [
                "Speaker IDs restart independently in each part. Unmapped IDs in the "
                "summary remain part-scoped. Names are combined only when explicitly "
                "mapped to the same label.",
                "",
            ]
        )
    stats = speaker_stats(results, names)
    if stats:
        total = sum(row[2] for row in stats)
        lines.extend(
            [
                "## Speakers",
                "",
                "| Speaker | Turns | Summed segment time | Share |",
                "| --- | ---: | ---: | ---: |",
            ]
        )
        for label, count, duration in stats:
            share = f"{duration / total:.0%}" if total else "—"
            lines.append(
                f"| {markdown_label(label)} | {count} | {timestamp(duration)} | {share} |"
            )
        lines.extend(
            [
                "",
                "Times sum model-emitted intervals; overlapping intervals can double-count "
                "time. Unknown timestamps are excluded. These are not measured talk-time shares.",
                "",
            ]
        )
    by_part = {r.chunk.index: r for r in results}
    for chunk in chunks:
        lines.extend(
            [
                f"## Part {chunk.index + 1} ({timestamp(chunk.start)}–{timestamp(chunk.end)})",
                "",
            ]
        )
        result = by_part.get(chunk.index)
        if result is None:
            lines.extend(["_Not processed._", ""])
            continue
        if result.issues:
            lines.extend(
                ["> [!WARNING]"] + [f"> {issue}" for issue in result.issues] + [""]
            )
        if not result.utterances:
            lines.extend(
                ["_No usable utterances. Inspect the raw-output sidecar._", ""]
            )
        for turn in group_turns(result.utterances):
            first = turn[0]
            label = resolve_label(first.speaker, first.chunk, names)
            known_ends = [u.end for u in turn if u.end is not None]
            end = max(known_ends) if known_ends else None
            lines.extend(
                [
                    f"### {markdown_label(label)} · {timestamp(first.start)} – {timestamp(end)}",
                    "",
                ]
            )
            for utterance in turn:
                # Indent continuation lines so multiline speech stays in its turn.
                content = utterance.text.replace("\n", "\n  ")
                lines.append(f"- `{timestamp(utterance.start)}` {content}")
            lines.append("")
    return "\n".join(lines).rstrip() + "\n"


def write_outputs(
    results: Sequence[Result],
    chunks: Sequence[Chunk],
    args: argparse.Namespace,
    names: SpeakerMap,
) -> None:
    args.output.parent.mkdir(parents=True, exist_ok=True)
    args.output.write_text(
        render_markdown(results, chunks, args.audio, args.model, names),
        encoding="utf-8",
    )
    if args.json_path:
        args.json_path.parent.mkdir(parents=True, exist_ok=True)
        args.json_path.write_text(
            json.dumps(
                [asdict(u) for r in results for u in r.utterances],
                ensure_ascii=False,
                indent=2,
                allow_nan=False,
            )
            + "\n",
            encoding="utf-8",
        )
    raw_path = args.output.with_suffix(".raw.txt")
    if any(r.issues for r in results):
        raw_path.write_text(
            "\n\n".join(
                f"=== Part {r.chunk.index + 1} ===\n{r.raw_text}" for r in results
            )
            + "\n",
            encoding="utf-8",
        )
    elif len(results) == len(chunks) and raw_path.is_file():
        raw_path.unlink()


def package_versions() -> dict[str, str]:
    installed = {}
    for package in ("mlx-audio", "mlx", "transformers", "numpy", "soundfile", "soxr"):
        try:
            installed[package] = version(package)
        except PackageNotFoundError:
            installed[package] = "not installed"
    return installed


def cache_settings(
    args: argparse.Namespace,
    chunk: Chunk,
    tokens: int,
    packages: dict[str, str],
    context: str | None,
) -> dict[str, Any]:
    stat = args.audio.stat()
    model_path = Path(args.model).expanduser()
    return {
        "script_version": __version__,
        "packages": packages,
        "source": str(args.audio.resolve()),
        "source_bytes": stat.st_size,
        "source_mtime_ns": stat.st_mtime_ns,
        "chunk": asdict(chunk),
        "model": str(model_path.resolve()) if model_path.exists() else args.model,
        "context": context,
        "max_tokens": tokens,
        "prefill_step_size": args.prefill_step_size,
        "temperature": 0.0,
    }


def load_cached(path: Path, settings: dict[str, Any], chunk: Chunk) -> Result | None:
    if not path.is_file():
        return None
    try:
        payload = json.loads(path.read_text(encoding="utf-8"))
        if payload["settings"] != settings:
            return None
        result = make_result(
            chunk,
            payload["raw_text"],
            payload["generation_tokens"],
            settings["max_tokens"],
            payload["elapsed"],
            payload["peak_gib"],
            cached=True,
        )
        # A saved failure is diagnostic evidence, not a completed chunk.
        return result if not result.issues else None
    except (OSError, ValueError, KeyError, TypeError, AttributeError):
        logger.warning("Ignoring unreadable or incompatible cache entry %s", path)
        return None


def save_cached(path: Path, settings: dict[str, Any], result: Result) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(
        json.dumps(
            {
                "settings": settings,
                "raw_text": result.raw_text,
                "generation_tokens": result.generation_tokens,
                "elapsed": result.elapsed,
                "peak_gib": result.peak_gib,
            },
            ensure_ascii=False,
            indent=2,
            allow_nan=False,
        )
        + "\n",
        encoding="utf-8",
    )


def load_backend(model_path: str) -> tuple[Any, Any]:
    # Help, planning, parsing, and clean cache rerenders never import MLX.
    if sys.platform != "darwin" or platform.machine() != "arm64":
        raise RuntimeError(
            "Inference requires native arm64 Python on an Apple Silicon Mac."
        )
    import mlx.core as mx

    from mlx_audio.stt.utils import load_model

    if not mx.metal.is_available():
        raise RuntimeError("MLX Metal is unavailable in this Python environment.")
    logger.info("Loading %s", model_path)
    path = Path(model_path).expanduser()
    model = load_model(str(path) if path.exists() else model_path, strict=True)
    info = mx.device_info()
    recommended = info.get("max_recommended_working_set_size", 0) / 1024**3
    if recommended:
        logger.info(
            "Metal recommended working set: %.1f GiB (not a free-memory reading)",
            recommended,
        )
    return model, mx


def transcribe_chunk(
    model: Any,
    mx: Any,
    audio: sf.SoundFile,
    chunk: Chunk,
    tokens: int,
    context: str | None,
    prefill_step_size: int,
    progress: bool,
) -> Result:
    waveform = read_chunk(audio, chunk)
    mx.reset_peak_memory()
    started = time.perf_counter()
    try:
        output = model.generate(
            waveform,
            sampling_rate=SAMPLE_RATE,
            context=context,
            max_tokens=tokens,
            temperature=0.0,
            prefill_step_size=prefill_step_size,
            verbose=progress,
        )
        elapsed = time.perf_counter() - started
        peak = mx.get_peak_memory() / 1024**3
        result = make_result(
            chunk,
            output.text or "",
            int(output.generation_tokens),
            tokens,
            elapsed,
            peak,
        )
    finally:
        mx.clear_cache()
    recommended = mx.device_info().get("max_recommended_working_set_size", 0) / 1024**3
    if recommended and peak > 0.85 * recommended:
        logger.warning(
            "Part %d used %.1f GiB MLX peak against %.1f GiB recommended. "
            "Check Activity Monitor memory pressure; consider shorter chunks.",
            chunk.index + 1,
            peak,
            recommended,
        )
    return result


def positive_int(value: str) -> int:
    result = int(value)
    if result <= 0:
        raise argparse.ArgumentTypeError("must be a positive integer")
    return result


def chunk_minutes(value: str) -> float:
    result = float(value)
    if not math.isfinite(result) or not 0 < result < 59:
        raise argparse.ArgumentTypeError(
            "must be finite, greater than 0, and less than 59"
        )
    return result


def parse_args(argv: Sequence[str] | None = None) -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter
    )
    parser.add_argument(
        "audio",
        type=Path,
        help="input readable by libsndfile, e.g. WAV or FLAC",
    )
    parser.add_argument(
        "-o",
        "--output",
        type=Path,
        help="Markdown path; defaults to AUDIO.md",
    )
    parser.add_argument(
        "--json",
        dest="json_path",
        type=Path,
        help="also write raw-ID utterances as JSON",
    )
    parser.add_argument(
        "--model",
        default=DEFAULT_MODEL,
        help="non-streaming MLX VibeVoice-ASR repository or path",
    )
    parser.add_argument(
        "--chunk-minutes",
        type=chunk_minutes,
        default=10.0,
        help="target minutes per part (default: 10)",
    )
    parser.add_argument(
        "--tokens-per-minute",
        type=positive_int,
        default=1600,
        help="heuristic output budget per audio minute (default: 1600)",
    )
    parser.add_argument(
        "--min-tokens",
        type=positive_int,
        default=2048,
        help="minimum output budget (default: 2048)",
    )
    parser.add_argument(
        "--max-tokens",
        type=positive_int,
        default=65536,
        help="maximum output budget (default: 65536)",
    )
    parser.add_argument(
        "--prefill-step-size",
        type=positive_int,
        default=2048,
        help="language-model prefill batch size (default: 2048)",
    )
    parser.add_argument(
        "--context",
        help="recognition context, e.g. topic and terminology",
    )
    parser.add_argument(
        "--hotwords",
        nargs="+",
        default=[],
        help="names and terms used as recognition hints",
    )
    parser.add_argument(
        "--speaker-names",
        nargs="+",
        default=[],
        metavar="[PART:]ID=NAME",
        help="presentation-only renaming; parts are one-based, speaker IDs are unchanged",
    )
    parser.add_argument("--cache-dir", type=Path, help="defaults to OUTPUT.cache")
    parser.add_argument(
        "--no-cache",
        action="store_true",
        help="do not read or write the chunk cache",
    )
    parser.add_argument(
        "-n",
        "--dry-run",
        action="store_true",
        help="print audio plan without loading MLX or writing output",
    )
    parser.add_argument(
        "-v",
        "--verbose",
        action="store_true",
        help="debug logging and exception tracebacks",
    )
    parser.add_argument(
        "-q",
        "--quiet",
        action="store_true",
        help="warnings/errors only; hide model generation progress",
    )
    parser.add_argument(
        "--version",
        action="version",
        version=f"%(prog)s {__version__}",
    )
    args = parser.parse_args(argv)
    if args.min_tokens > args.max_tokens:
        parser.error("--min-tokens exceeds --max-tokens")
    if args.cache_dir and args.no_cache:
        parser.error("--cache-dir and --no-cache cannot be combined")
    if args.quiet and args.verbose:
        parser.error("--quiet and --verbose cannot be combined")
    args.audio = args.audio.expanduser()
    args.output = (args.output or args.audio.with_suffix(".md")).expanduser()
    if args.json_path:
        args.json_path = args.json_path.expanduser()
    args.cache_dir = (args.cache_dir or args.output.with_suffix(".cache")).expanduser()
    # In the old script '-o input.flac' could overwrite the recording itself.
    paths = [args.audio, args.output, args.output.with_suffix(".raw.txt")]
    if args.json_path:
        paths.append(args.json_path)
    if len({path.resolve() for path in paths}) != len(paths):
        parser.error("input, Markdown, JSON, and raw-output paths must be different")
    try:
        args.names = parse_speaker_names(args.speaker_names)
    except ValueError as exc:
        parser.error(str(exc))
    return args


def run(args: argparse.Namespace) -> int:
    budget = Budget(args.tokens_per_minute, args.min_tokens, args.max_tokens)
    # mlx-audio 0.5.1 appends its hotwords list twice. Merge once here and use
    # the native context argument; do not also pass hotwords to model.generate.
    terms = ", ".join(word.strip() for word in args.hotwords if word.strip())
    context = "\n".join(s for s in ((args.context or "").strip(), terms) if s) or None
    packages = package_versions()
    logger.debug("Package versions: %s", packages)
    with sf.SoundFile(args.audio) as audio:
        if not len(audio):
            raise ValueError("The audio file has no frames.")
        chunks = plan_chunks(audio, args.chunk_minutes)
        logger.info(
            "%s: %s in %d part(s)",
            args.audio.name,
            timestamp(chunks[-1].end),
            len(chunks),
        )
        for chunk in chunks:
            tokens = budget.for_duration(chunk.duration)
            line = (
                f"Part {chunk.index + 1}/{len(chunks)}: {timestamp(chunk.start)}–"
                f"{timestamp(chunk.end)} ({chunk.duration / 60:.2f} min), "
                f"{tokens} output tokens maximum"
            )
            if args.dry_run:
                print(line)
            else:
                logger.info(line)
            if chunk.duration / 60 * args.tokens_per_minute + 1024 > args.max_tokens:
                logger.warning(
                    "Part %d's heuristic budget is clamped by --max-tokens.",
                    chunk.index + 1,
                )
        if args.dry_run:
            return 0
        model = mx = None
        results: list[Result] = []
        for chunk in chunks:
            tokens = budget.for_duration(chunk.duration)
            settings = cache_settings(args, chunk, tokens, packages, context)
            cache = args.cache_dir / f"chunk_{chunk.index:03d}.json"
            result = None if args.no_cache else load_cached(cache, settings, chunk)
            if result is None:
                if model is None:
                    model, mx = load_backend(args.model)
                logger.info("Transcribing part %d/%d", chunk.index + 1, len(chunks))
                result = transcribe_chunk(
                    model,
                    mx,
                    audio,
                    chunk,
                    tokens,
                    context,
                    args.prefill_step_size,
                    not args.quiet,
                )
                if not args.no_cache:
                    save_cached(cache, settings, result)
            results.append(result)
            logger.info(
                "Part %d: %d utterances, %d/%d tokens, %.1fs, %.1f GiB MLX peak%s",
                chunk.index + 1,
                len(result.utterances),
                result.generation_tokens,
                tokens,
                result.elapsed,
                result.peak_gib,
                " (cached)" if result.cached else "",
            )
            for issue in result.issues:
                logger.warning("Part %d: %s", chunk.index + 1, issue)
            write_outputs(results, chunks, args, args.names)
    observed = {(u.chunk, u.speaker) for r in results for u in r.utterances}
    for part, speaker in args.names:
        if not any(s == speaker and (part is None or p == part) for p, s in observed):
            logger.warning(
                "Unmatched speaker-name entry: part=%s, %s",
                None if part is None else part + 1,
                speaker,
            )
    print(args.output)
    return 1 if any(r.issues for r in results) else 0


def main(argv: Sequence[str] | None = None) -> int:
    args = parse_args(argv)
    logging.basicConfig(
        level=logging.WARNING
        if args.quiet
        else logging.DEBUG
        if args.verbose
        else logging.INFO,
        format="%(asctime)s %(levelname)s %(message)s",
        datefmt="%H:%M:%S",
    )
    try:
        return run(args)
    except KeyboardInterrupt:
        logger.warning(
            "Interrupted; completed parts remain in the output and, unless disabled, cache."
        )
        return 130
    except (OSError, RuntimeError, ValueError, ImportError) as exc:
        logger.error("%s", exc, exc_info=args.verbose)
        return 1


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