"""
Vocal Helper — click-based command-line interface.
Twin of :mod:`vocal_helper.cli_argparse`: same public surface (identical
subcommand names, identical flag semantics), but implemented with
:mod:`click` so users who already have a click-native shell setup
(bash / zsh completion via ``click.shell_completion``, colored ``--help``,
nested command groups) can plug it in without friction. Installed as
the ``vocal-helper-click`` entry point in ``pyproject.toml``.
Design notes
------------
- Subcommands mirror ``vocal-helper`` (the argparse twin) so both CLIs
expose the same operations under symmetric, predictable names.
- Flags reuse the argparse names (``--whisper-model``, ``--language``,
``--diar-backend``, …) rather than the more idiomatic click positional
style — consistency across the two CLIs beats micro-idiomaticity.
- Async work is wrapped in :func:`asyncio.run` inside each command;
click itself stays sync.
Usage Example
-------------
>>> # vocal-helper-click mic --llm --jsonl
>>> # vocal-helper-click file meeting.wav --offline --language en
>>> # vocal-helper-click url "https://youtu.be/…" --language fr
>>> # vocal-helper-click transcribe clip.wav --language en
Author
------
Warith Harchaoui, Ph.D. , https://linkedin.com/in/warith-harchaoui/
"""
from __future__ import annotations
import asyncio
import json
import sys
from collections.abc import AsyncIterator, Callable, Mapping
from pathlib import Path
from typing import Any
try:
import click
except ImportError as exc: # pragma: no cover
raise ImportError(
"The click CLI requires the [cli] extra. Install with: pip install 'vocal-helper[cli]'"
) from exc
from vocal_helper.pipeline import (
OfflinePipeline,
OfflinePipelineConfig,
Pipeline,
PipelineConfig,
resolve_engine,
)
from vocal_helper.types import PcmFrame
# ---------------------------------------------------------------------------
# Shared config translation — mirrors the argparse twin's ``_build_pipeline_config``.
# We keep the click callbacks small by punching the shared kwargs through
# a plain dict rather than a Namespace.
# ---------------------------------------------------------------------------
def _pipeline_config(
*,
whisper_model: str,
language: str,
threads: int,
initial_prompt: str,
diar_backend: str,
join_threshold: float | None,
llm: bool,
llm_recent_window_s: float,
endpoint: str | None,
eot: bool,
) -> PipelineConfig:
"""Build a :class:`PipelineConfig` from the shared click options."""
# ASR dict passed straight through to WhisperStage.__init__ — keys mirror
# the argparse twin so both CLIs produce byte-identical pipeline configs.
asr_cfg: dict = {
"model": whisper_model,
"language": language,
"threads": threads,
# Coerce a possible ``None`` bias prompt to "" — whisper rejects None.
"initial_prompt": initial_prompt or "",
}
# Model weights load from the self-hosted diarization-engines bundle
# (settings.yaml ``engines.diarization_url``) — no HuggingFace token.
diar_cfg: dict = {"backend": diar_backend}
# Only override the join threshold when the user passed one — otherwise let
# the diarizer keep its tuned 0.30 default rather than pinning it here.
if join_threshold is not None:
diar_cfg["join_threshold"] = join_threshold
# LLM analyst and semantic-EOT stages are opt-in and both need an LLM. The
# model is never a flag — best-engine-ai-helper resolves it (per machine)
# from the committed ``llm.brief.yaml``. Resolve once, thread the same
# engine descriptor into whichever stage(s) are enabled.
llm_cfg: dict | None = None
eot_cfg: dict | None = None
if llm or eot:
engine = resolve_engine(endpoint=endpoint)
if llm:
llm_cfg = {"engine": engine, "recent_window_s": llm_recent_window_s}
# Keys must match SemanticEOTStage.__init__ (``engine``).
if eot:
eot_cfg = {"engine": engine}
return PipelineConfig(diar=diar_cfg, asr=asr_cfg, llm=llm_cfg, eot=eot_cfg)
def _resolve_online_backend(cfg: PipelineConfig, requested_backend: str) -> None:
"""Resolve a live command's ``"auto"`` diar backend through the router, in place.
The ``mic`` / ``url`` commands build a streaming :class:`Pipeline` directly,
so — unlike ``file`` — they never pass through :func:`_choose_file_diar`. This
routes their backend the same way (live stream, no duration → ``nemo`` per the
study, or an explicit override) and mutates ``cfg.diar['backend']`` so the
sentinel ``"auto"`` never reaches :class:`~vocal_helper.OnlineDiarStage`.
Parameters
----------
cfg : PipelineConfig
The live pipeline config whose ``diar`` backend is resolved in place.
requested_backend : str
``"auto"`` to route, or an explicit backend name to honour as override.
Returns
-------
None
``cfg.diar['backend']`` is updated as a side effect; any router note is
written to stderr.
"""
# Reuse the argparse twin's router choke-point so both CLIs decide identically.
from vocal_helper.cli_argparse import _route_backend
backend, note = _route_backend(requested_backend=requested_backend, live=True, duration_s=None)
if note:
sys.stderr.write(note + "\n")
cfg.diar["backend"] = backend
def _print_event(ev: Mapping[str, Any], jsonl: bool) -> None:
"""Emit a single pipeline event to stdout, JSONL or human-readable."""
if jsonl:
# Strip the raw PCM before serialising — cheaper log, right transport.
sys.stdout.write(json.dumps({k: v for k, v in ev.items() if k != "pcm"}) + "\n")
sys.stdout.flush()
return
# Human-readable branch — the event's shape tells us which template to use:
# a ``text`` key is an utterance line, a ``summary`` key is a rolling digest.
if "text" in ev:
sys.stdout.write(f"[{ev['t0']:7.2f}s -> {ev['t1']:7.2f}s {ev['speaker']}] {ev['text']}\n")
elif "summary" in ev:
sys.stdout.write(
f"\n--- rolling summary @ {ev['t0']:.1f}s "
f"(model={ev['model']}) ---\n{ev['summary']}\n"
f"--- recent window ---\n{ev['recent']}\n\n"
)
sys.stdout.flush()
async def _drain(pipeline: Pipeline | OfflinePipeline, jsonl: bool) -> None:
"""Consume every event emitted by ``pipeline.run()``.
Parameters
----------
pipeline : Pipeline or OfflinePipeline
A started orchestrator whose ``run()`` coroutine yields event dicts.
jsonl : bool
When ``True`` each event is printed as a single JSON line ; otherwise
it is rendered in the human-readable format.
Returns
-------
None
Events are written to stdout as a side effect ; nothing is returned.
"""
# Pull events as they land and hand each straight to the printer — the
# pipeline back-pressures us, so this loop paces the whole CLI.
async for ev in pipeline.run():
_print_event(ev, jsonl)
# ---------------------------------------------------------------------------
# Reusable option bundle — Click does not have a built-in shared-options
# decorator, but we can compose one with a helper. Every subcommand slaps
# ``@_common_options`` on top of its own signature.
# ---------------------------------------------------------------------------
def _common_options(func: Callable[..., object]) -> Callable[..., object]:
"""Decorator adding the shared VAD / diar / ASR / LLM levers.
Parameters
----------
func : Callable[..., object]
The click command callback to wrap with the shared option stack.
Returns
-------
Callable[..., object]
The same callback with every shared ``click.option`` applied, so it
can be composed on top of a subcommand's own signature.
"""
# Order matters for --help output; we mirror the argparse twin.
func = click.option(
"--jsonl", is_flag=True, default=False, help="Emit one JSON event per line."
)(func)
func = click.option(
"--eot",
is_flag=True,
default=False,
help="Enable the SemanticEOTStage (LiveKit-style turn "
"detector). Reduces mid-sentence cuts at the cost of "
"one extra LLM hop per voiced segment.",
)(func)
func = click.option(
"--endpoint",
default=None,
help="Override the LLM serving base URL passed to best-engine-ai-helper "
"(e.g. a remote Ollama / vLLM host).",
)(func)
func = click.option(
"--llm-recent-window-s",
type=float,
default=60.0,
show_default=True,
help="Verbatim window (seconds) kept out of the summary.",
)(func)
func = click.option("--llm", is_flag=True, default=False, help="Enable the LLM analyst stage.")(
func
)
func = click.option(
"--join-threshold",
type=float,
default=None,
help="Cosine-distance join threshold for online diarizer (default 0.30).",
)(func)
func = click.option(
"--diar-backend",
type=click.Choice(["auto", "pyannote", "nemo", "sherpa"]),
default="auto",
show_default=True,
help="Speaker-diarization backend. Default 'auto' delegates to the "
"study-grounded router (the aiguilleur): offline short → 'nemo', "
"offline long/unknown → 'pyannote', live → 'nemo', reporting DER "
"(quality) and RTF (speed). Pass 'pyannote' / 'nemo' / 'sherpa' to "
"override; 'sherpa' is torch-free (needs the [sherpa] extra).",
)(func)
func = click.option(
"--initial-prompt",
default="",
help="Whisper bias prompt — name the domain and a few expected "
"proper nouns. Cuts WER 15-25 pp on AMI (2026-06-30 sweep).",
)(func)
func = click.option(
"--threads", type=int, default=6, show_default=True, help="whisper.cpp CPU threads."
)(func)
func = click.option(
"--language",
default="auto",
show_default=True,
help="ISO-639-1 code or 'auto' for language ID.",
)(func)
func = click.option(
"--whisper-model",
default="large-v3-turbo-q5_0",
show_default=True,
help="pywhispercpp model tag.",
)(func)
return func
# ---------------------------------------------------------------------------
# Top-level group
# ---------------------------------------------------------------------------
@click.group(
context_settings={"help_option_names": ["-h", "--help"], "max_content_width": 100},
)
@click.version_option(package_name="vocal-helper", prog_name="vocal-helper-click")
def cli() -> None:
"""Vocal Helper — click twin of the argparse CLI. Same subcommands."""
# Nothing at the group level; every subcommand carries its own args.
# ---------------------------------------------------------------------------
# mic
# ---------------------------------------------------------------------------
@cli.command()
@_common_options
@click.option("--device", default=None, help="Substring of the microphone name.")
def mic(
whisper_model: str,
language: str,
threads: int,
initial_prompt: str,
diar_backend: str,
join_threshold: float | None,
llm: bool,
llm_recent_window_s: float,
endpoint: str | None,
eot: bool,
jsonl: bool,
device: str | None,
) -> None:
"""Live microphone input (needs the ``[mic]`` extra)."""
from vocal_helper.sources import from_microphone
cfg = _pipeline_config(
whisper_model=whisper_model,
language=language,
threads=threads,
initial_prompt=initial_prompt,
diar_backend=diar_backend,
join_threshold=join_threshold,
llm=llm,
llm_recent_window_s=llm_recent_window_s,
endpoint=endpoint,
eot=eot,
)
def factory() -> AsyncIterator[PcmFrame]:
"""Open a fresh 16 kHz / 20 ms microphone stream on each pipeline start."""
# 16 kHz mono is whisper.cpp's native rate ; 20 ms is the Silero VAD stride,
# so the source hands the pipeline frames it can consume without resampling.
return from_microphone(device_name=device, sample_rate=16_000, frame_ms=20)
# A live stream carries no duration — route the online backend (auto → nemo
# per the study) so ``"auto"`` never reaches the stage.
_resolve_online_backend(cfg, diar_backend)
pipeline = Pipeline(source=factory, config=cfg)
asyncio.run(_drain(pipeline, jsonl))
# ---------------------------------------------------------------------------
# file
# ---------------------------------------------------------------------------
@cli.command()
@_common_options
@click.argument("path", type=click.Path(exists=True))
@click.option(
"--no-real-time",
is_flag=True,
default=False,
help="Batch mode: process as fast as possible (skip real-time pacing). By "
"default auto-selects the offline pyannote diarizer when its bundle is "
"present — the most reliable path (DER ~0.12 on AMI vs ~0.50 online, "
"2026-07-16 sweep) — else falls back to the online diarizer with the global "
"re-clustering repair pass. Pass --online to force the streaming diarizer.",
)
@click.option(
"--offline",
is_flag=True,
default=False,
help="Force the OfflinePipeline (pyannote 3.1 whole-buffer, global "
"clustering) — most reliable on meetings/podcasts/lectures. Honours "
"--diar-backend. Already the default for --no-real-time when the bundle is "
"available.",
)
@click.option(
"--online",
is_flag=True,
default=False,
help="Force the online streaming diarizer for a batch file run instead of "
"auto-selecting offline (lighter, lower latency, higher DER).",
)
def file(
whisper_model: str,
language: str,
threads: int,
initial_prompt: str,
diar_backend: str,
join_threshold: float | None,
llm: bool,
llm_recent_window_s: float,
endpoint: str | None,
eot: bool,
jsonl: bool,
path: str,
no_real_time: bool,
offline: bool,
online: bool,
) -> None:
"""Replay a 16 kHz mono WAV through the pipeline."""
from vocal_helper.cli_argparse import _choose_file_diar
from vocal_helper.sources import from_wav_file, probe_duration_s
cfg = _pipeline_config(
whisper_model=whisper_model,
language=language,
threads=threads,
initial_prompt=initial_prompt,
diar_backend=diar_backend,
join_threshold=join_threshold,
llm=llm,
llm_recent_window_s=llm_recent_window_s,
endpoint=endpoint,
eot=eot,
)
def factory() -> AsyncIterator[PcmFrame]:
"""Open the WAV source ; ``--no-real-time`` skips wall-clock pacing for batch runs."""
# Real-time pacing simulates a live feed ; disabling it fires frames as fast
# as they decode, which is what you want when timing throughput on a file.
return from_wav_file(Path(path), real_time=not no_real_time)
# Shared decision with the argparse twin (see cli_argparse._choose_file_diar):
# batch runs prefer the offline whole-buffer diarizer when a backend is
# installed, with the study-grounded router choosing that backend from the
# file's real probed duration — short→nemo, long→pyannote — reporting DER +
# RTF. Falls back to the online diarizer + refine pass when no offline
# backend is available.
use_offline, diar_cfg, note = _choose_file_diar(
cfg.diar,
explicit_offline=offline,
batch=no_real_time,
force_online=online,
duration_s=probe_duration_s(Path(path)),
requested_backend=diar_backend,
)
if note:
sys.stderr.write(note + "\n")
pipeline: Pipeline | OfflinePipeline
if use_offline:
pipeline = OfflinePipeline(
source=factory,
config=OfflinePipelineConfig(diar=diar_cfg, asr=cfg.asr, llm=cfg.llm),
)
else:
cfg.diar = diar_cfg
pipeline = Pipeline(source=factory, config=cfg)
asyncio.run(_drain(pipeline, jsonl))
# ---------------------------------------------------------------------------
# url
# ---------------------------------------------------------------------------
@cli.command()
@_common_options
@click.argument("url")
def url(
whisper_model: str,
language: str,
threads: int,
initial_prompt: str,
diar_backend: str,
join_threshold: float | None,
llm: bool,
llm_recent_window_s: float,
endpoint: str | None,
eot: bool,
jsonl: bool,
url: str,
) -> None:
"""Stream from any URL yt-dlp can reach (needs the ``[stream]`` extra)."""
from vocal_helper.sources import from_url as _from_url
cfg = _pipeline_config(
whisper_model=whisper_model,
language=language,
threads=threads,
initial_prompt=initial_prompt,
diar_backend=diar_backend,
join_threshold=join_threshold,
llm=llm,
llm_recent_window_s=llm_recent_window_s,
endpoint=endpoint,
eot=eot,
)
def factory() -> AsyncIterator[PcmFrame]:
"""Open a streaming source for the given URL (yt-dlp resolves the media)."""
return _from_url(url)
# URL playback is a live stream — route the online backend (auto → nemo).
_resolve_online_backend(cfg, diar_backend)
pipeline = Pipeline(source=factory, config=cfg)
asyncio.run(_drain(pipeline, jsonl))
# ---------------------------------------------------------------------------
# transcribe — one-shot, no VAD/diarization.
# ---------------------------------------------------------------------------
@cli.command()
@click.argument("path", type=click.Path(exists=True))
@click.option("--whisper-model", default="large-v3-turbo-q5_0", show_default=True)
@click.option("--language", default="auto", show_default=True)
@click.option("--threads", type=int, default=6, show_default=True)
@click.option(
"--initial-prompt",
default="",
help="Whisper bias prompt — name the domain and a few expected proper "
"nouns. Cuts WER 15-25 pp on AMI (2026-06-30 sweep).",
)
@click.option(
"--jsonl", is_flag=True, default=False, help='Emit {"path": ..., "text": ...} JSON on stdout.'
)
def transcribe(
path: str,
whisper_model: str,
language: str,
threads: int,
initial_prompt: str,
jsonl: bool,
) -> None:
"""One-shot transcription of a WAV file (skip VAD / diarization)."""
# Lazy imports so ``--help`` never pays the numpy / audio-helper / whisper.cpp
# import cost for users who only wanted the usage text.
import numpy as np
from audio_helper import load_audio
from vocal_helper.asr import transcribe_pcm
# ffmpeg-backed decode — any format (mp3/m4a/opus/video), mono, native rate.
pcm, sr = load_audio(path, to_mono=True, to_numpy=True)
# whisper.cpp wants a contiguous float32 buffer — coerce whatever dtype we got.
pcm = np.asarray(pcm, dtype=np.float32)
text = transcribe_pcm(
pcm=pcm,
sr=int(sr),
model=whisper_model,
language=language,
threads=threads,
initial_prompt=initial_prompt or "",
)
if jsonl:
click.echo(json.dumps({"path": path, "text": text}))
else:
click.echo(text)
[docs]
def main() -> None:
"""Console entry point (``vocal-helper-click``).
click's own ``main()`` only special-cases ``ClickException``/``Abort``
(and a broken pipe); a plain library exception would otherwise
propagate as a raw Python traceback instead of a clean CLI error. This
wraps the whole invocation and translates that last case into a
one-line stderr message + exit 1 — click's own control flow (usage
errors, ``--help``, an explicit ``sys.exit`` in a subcommand) already
raises ``SystemExit``, a ``BaseException`` this does not catch, so it
passes through untouched.
"""
try:
cli()
except Exception as err: # noqa: BLE001 — last resort: see docstring
click.echo(f"Error: {err}", err=True)
sys.exit(1)
if __name__ == "__main__": # pragma: no cover
main()