diff --git a/README.md b/README.md index 4e38ff8..0be41eb 100644 --- a/README.md +++ b/README.md @@ -101,6 +101,10 @@ dg -o json listen standup.mp3 \ # Live microphone with interim (partial) results dg listen --mic --model nova-3 --interim +# Redact sensitive numbers and spell numbers as digits (files or live) +# Flux STT (v2) accepts --redact numbers|aggressive_numbers; v1 also pci, ssn, … +dg listen call.wav --redact numbers --numerals + # Raw audio stream from ffmpeg ffmpeg -i video.mp4 -f s16le -ar 16000 -ac 1 - \ | dg listen --encoding linear16 @@ -186,10 +190,18 @@ dg speak "Hello from Flux" -o hello.wav # end-of-stream notice (the audio is complete). dg speak "Hello from Flux" | ffplay -loglevel error -nodisp -autoexit - +# Flux TTS streaming controls (flux-* only): --speed 0.85–1.15 (0.05 steps). +# --expressivity -2..2 is beta; its default 0 is nominal delivery. +dg speak "A little slower" --speed 0.9 --expressivity 1 -o slow.wav + # Aura (v1, batch REST) — opt in with -m aura-*; needed for MP3 output dg speak "Welcome to Deepgram" -o welcome.mp3 -m aura-2-asteria-en dg speak --file script.txt -o output.mp3 -m aura-2-luna-en echo "Hello" | dg speak -o greeting.mp3 -m aura-2-asteria-en + +# Aura-2 also has Spanish voices (e.g. aura-2-selena-es); run `dg models` +# for the full, current list. +dg speak "Hola, bienvenido a Deepgram" -o hola.mp3 -m aura-2-selena-es ``` ### Text Intelligence diff --git a/packages/deepctl-cmd-listen/pyproject.toml b/packages/deepctl-cmd-listen/pyproject.toml index 8e2af4f..5b35fb3 100644 --- a/packages/deepctl-cmd-listen/pyproject.toml +++ b/packages/deepctl-cmd-listen/pyproject.toml @@ -21,7 +21,7 @@ classifiers = [ keywords = ["deepgram", "cli", "stt", "live", "streaming", "listen"] requires-python = ">=3.10" dependencies = [ - "deepctl-core>=0.1.10", + "deepctl-core>=0.2.15", "deepctl-shared-utils>=0.1.10", "click>=8.0.0", "rich>=13.0.0", diff --git a/packages/deepctl-cmd-listen/src/deepctl_cmd_listen/command.py b/packages/deepctl-cmd-listen/src/deepctl_cmd_listen/command.py index b198013..f0f0e31 100644 --- a/packages/deepctl-cmd-listen/src/deepctl_cmd_listen/command.py +++ b/packages/deepctl-cmd-listen/src/deepctl_cmd_listen/command.py @@ -21,6 +21,7 @@ from typing import Any from urllib.parse import urlencode +import click from deepctl_core import ( AuthManager, BaseCommand, @@ -54,6 +55,11 @@ out = Console() +# Flux STT (listen v2) only recognises these two --redact values; the v1 REST +# vocabulary ("pci", "ssn", …) earns an opaque HTTP 400 from the v2 endpoint. +_FLUX_REDACT = ("numbers", "aggressive_numbers") + + def _is_url(s: str) -> bool: return s.startswith(("http://", "https://")) @@ -65,6 +71,18 @@ def _ws_base(client: DeepgramClient) -> str: return base.replace("https://", "wss://").replace("http://", "ws://") +async def _cancel_and_drain(*tasks: asyncio.Task[Any]) -> None: + """Cancel tasks and consume their terminal exceptions during teardown.""" + for task in tasks: + if not task.done(): + task.cancel() + for task in tasks: + try: + await task + except BaseException: + pass + + class ListenCommand(BaseCommand): """Unified speech-to-text command supporting files, URLs, mic, and streams.""" @@ -86,6 +104,8 @@ class ListenCommand(BaseCommand): "dg listen https://example.com/call.mp3 --diarize", "dg listen --mic --model nova-3 --interim", "dg listen audio.mp3 --diarize --summarize --save-to transcript.txt", + "dg listen call.wav --redact numbers --numerals", + "dg listen --mic --model flux-general-en --redact aggressive_numbers", "dg -o json listen audio.mp3 | jq '.results.channels[0].alternatives[0].transcript'", "ffmpeg -i video.mp4 -f s16le -ar 16000 -ac 1 - | dg listen --encoding linear16", "dg listen - # read raw audio from stdin interactively", @@ -175,6 +195,27 @@ def get_arguments(self) -> list[dict[str, Any]]: "is_flag": True, "is_option": True, }, + { + "names": ["--redact"], + "help": ( + "Redact sensitive content. Flux STT (v2) accepts 'numbers' or " + "'aggressive_numbers'; v1 models also accept 'pci', 'ssn', " + "etc. Repeatable on v1 (e.g. --redact pci --redact numbers). " + "Applies to files and live streams." + ), + "type": str, + "multiple": True, + "is_option": True, + }, + { + "names": ["--numerals"], + "help": ( + "Convert spoken numbers to digits ('four twenty' -> '420'). " + "Applies to files and live streams." + ), + "is_flag": True, + "is_option": True, + }, # ── Live streaming options ──────────────────────────────── { "names": ["--interim"], @@ -293,7 +334,7 @@ def handle( language = kwargs.get("language") or "en-US" # flux-* models require listen.v2; everything else uses v1. - api_version = 2 if model.startswith("flux") else 1 + api_version = 2 if model.startswith("flux-") else 1 diarize = kwargs.get("diarize", False) smart_format = kwargs.get("smart_format", True) punctuate = kwargs.get("punctuate", True) @@ -301,13 +342,71 @@ def handle( topics = kwargs.get("topics", False) sentiment = kwargs.get("sentiment", False) interim = kwargs.get("interim", False) + # --redact is repeatable (click multiple=True → tuple). Normalise so the + # rest of the flow always sees a tuple: click gives (), direct callers + # (tests) may pass a bare string or None. + _redact_raw = kwargs.get("redact") + if _redact_raw is None: + redact: tuple[str, ...] = () + elif isinstance(_redact_raw, str): + redact = (_redact_raw,) if _redact_raw else () + else: + redact = tuple(_redact_raw) + numerals = kwargs.get("numerals", False) encoding = kwargs.get("encoding") sample_rate = kwargs.get("sample_rate") or 16000 - channels = kwargs.get("channels") or 1 + channels = kwargs.get("channels") + if channels is None: + channels = 1 + if channels < 1: + raise click.ClickException("--channels must be at least 1.") + + # Flux STT (listen v2) is turn-based and has no diarization; --diarize + # is dropped from the v2 param set (sending it earns an HTTP 400). It + # defaults to False, so if it's set the user asked for it explicitly — + # say we're ignoring it rather than letting it vanish silently. + # (smart_format / punctuate default to True and can't be told apart + # from an explicit flag, so they stay silent; --interim still gates + # client-side display.) + if api_version >= 2 and diarize: + status.print( + "[yellow]Note:[/yellow] --diarize is not supported by Flux STT " + "(listen v2) models; ignoring it." + ) + diarize = False save_to = kwargs.get("save_to") probe = kwargs.get("probe", False) no_validate = kwargs.get("no_validate", False) + # Flux STT (listen v2) is streaming-only — there is no v2 pre-recorded + # REST endpoint, so a file/URL routes to /v1/listen and the server + # rejects it ("Flux models are not supported on /v1/listen") wrapped in + # a header dump. Say so up front instead. + if api_version >= 2 and mode in ("prerecorded_file", "prerecorded_url"): + raise click.ClickException( + f"Flux STT ({model}) is streaming-only and cannot transcribe " + "a file or URL. Use a live source (--mic, stdin, or '-'), or " + "pick a v1 model (e.g. nova-3) for pre-recorded audio." + ) + + if api_version >= 2 and channels != 1: + raise click.ClickException( + f"--channels {channels} is not supported by Flux STT ({model}); " + "listen v2 accepts mono audio only (--channels 1)." + ) + + # Flux STT (listen v2) only accepts a narrow --redact vocabulary; the + # v1 values ("pci", "ssn", …) come back as an opaque HTTP 400. Validate + # up front, mirroring how `speak` guards its Flux-only flags. + if api_version >= 2: + bad = [r for r in redact if r not in _FLUX_REDACT] + if bad: + allowed = " or ".join(f"'{v}'" for v in _FLUX_REDACT) + raise click.ClickException( + f"--redact {', '.join(bad)} is not supported by Flux STT " + f"({model}); listen v2 accepts only {allowed}." + ) + # ── Caption format ───────────────────────────────────────────── want_webvtt = kwargs.get("webvtt", False) want_srt = kwargs.get("srt", False) @@ -342,6 +441,8 @@ def handle( summarize=summarize, topics=topics, sentiment=sentiment, + redact=redact, + numerals=numerals, save_to=save_to, probe=probe, no_validate=no_validate, @@ -363,6 +464,8 @@ def handle( summarize=summarize, topics=topics, sentiment=sentiment, + redact=redact, + numerals=numerals, save_to=save_to, probe=False, no_validate=no_validate, @@ -379,6 +482,8 @@ def handle( smart_format=smart_format, punctuate=punctuate, interim=interim, + redact=redact, + numerals=numerals, sample_rate=sample_rate, channels=channels, save_to=save_to, @@ -394,6 +499,8 @@ def handle( smart_format=smart_format, punctuate=punctuate, interim=interim, + redact=redact, + numerals=numerals, encoding=encoding, sample_rate=sample_rate, channels=channels, @@ -465,6 +572,8 @@ def _prerecorded( summarize: bool, topics: bool, sentiment: bool, + redact: tuple[str, ...], + numerals: bool, save_to: str | None, probe: bool, no_validate: bool, @@ -533,6 +642,12 @@ def _prerecorded( options["topics"] = "true" if sentiment: options["sentiment"] = "true" + if redact: + # Fern's query encoder expands a list into repeated params + # (redact=pci&redact=numbers) but leaves a tuple unexpanded. + options["redact"] = list(redact) + if numerals: + options["numerals"] = "true" # ── Call API ─────────────────────────────────────────────────── status.print(f"[dim]Transcribing[/dim] {source}") @@ -544,7 +659,10 @@ def _prerecorded( else: result_dict = client.transcribe_file(source, options) except Exception as e: - return BaseResult(status="error", message=f"Transcription failed: {e}") + # Raise (not return an error result) so an API rejection — e.g. a + # --redact value the v1 endpoint refuses — exits non-zero and is + # visible, rather than printing nothing and exiting 0. + raise click.ClickException(f"Transcription failed: {e}") # ── Format transcript ────────────────────────────────────────── if diarize: @@ -607,6 +725,8 @@ def _stream_mic( smart_format: bool, punctuate: bool, interim: bool, + redact: tuple[str, ...], + numerals: bool, sample_rate: int, channels: int, save_to: str | None, @@ -641,6 +761,8 @@ def _stream_mic( smart_format=smart_format, punctuate=punctuate, interim=interim, + redact=redact, + numerals=numerals, sample_rate=sample_rate, channels=channels, caption_writer=caption_writer, @@ -649,6 +771,8 @@ def _stream_mic( except KeyboardInterrupt: status.print("\n[yellow]Stopped.[/yellow]") result = ListenResult(status="success", source="mic", mode="live") + except click.ClickException: + raise except Exception as e: err = str(e) msg = f"Microphone error: {err}" @@ -690,6 +814,8 @@ async def _ws_mic( smart_format: bool, punctuate: bool, interim: bool, + redact: tuple[str, ...], + numerals: bool, sample_rate: int, channels: int, caption_writer: StreamingCaptionWriter | None = None, @@ -707,12 +833,15 @@ async def _ws_mic( smart_format=smart_format, punctuate=punctuate, interim=interim, + redact=redact, + numerals=numerals, encoding="linear16", sample_rate=sample_rate, channels=channels, ) api_key = client.auth_manager.get_api_key() full_transcript: list[str] = [] + v2_state = self._new_v2_state() if api_version >= 2 else None stop_event = threading.Event() async with websockets.connect( @@ -756,6 +885,7 @@ async def recv_transcripts() -> None: diarize=diarize, interim=interim, caption_writer=caption_writer, + v2_state=v2_state, ) send_task = asyncio.create_task(send_audio()) @@ -764,8 +894,15 @@ async def recv_transcripts() -> None: await asyncio.gather(send_task, recv_task) except (KeyboardInterrupt, asyncio.CancelledError): stop_event.set() - send_task.cancel() - recv_task.cancel() + await _cancel_and_drain(send_task, recv_task) + except BaseException: + stop_event.set() + await _cancel_and_drain(send_task, recv_task) + raise + + # Flux may close mid-turn without an EndOfTurn; emit what we have. + if v2_state is not None: + self._flush_v2(v2_state, full_transcript, caption_writer=caption_writer) return ListenResult( status="success", @@ -791,6 +928,8 @@ def _stream_stdin( smart_format: bool, punctuate: bool, interim: bool, + redact: tuple[str, ...], + numerals: bool, encoding: str | None, sample_rate: int, channels: int, @@ -831,6 +970,8 @@ def _stream_stdin( smart_format=smart_format, punctuate=punctuate, interim=interim, + redact=redact, + numerals=numerals, encoding=resolved_encoding, sample_rate=sample_rate, channels=channels, @@ -867,6 +1008,8 @@ async def _ws_stdin( smart_format: bool, punctuate: bool, interim: bool, + redact: tuple[str, ...], + numerals: bool, encoding: str, sample_rate: int, channels: int, @@ -883,25 +1026,63 @@ async def _ws_stdin( smart_format=smart_format, punctuate=punctuate, interim=interim, + redact=redact, + numerals=numerals, encoding=encoding, sample_rate=sample_rate, channels=channels, ) api_key = client.auth_manager.get_api_key() full_transcript: list[str] = [] + v2_state = self._new_v2_state() if api_version >= 2 else None async with websockets.connect( url, additional_headers={"Authorization": f"Token {api_key}"} ) as ws: + loop = asyncio.get_running_loop() + audio_queue: asyncio.Queue[bytes | Exception | None] = asyncio.Queue() + stop_reader = threading.Event() + + def post_audio(item: bytes | Exception | None) -> None: + if stop_reader.is_set(): + return + try: + loop.call_soon_threadsafe(audio_queue.put_nowait, item) + except RuntimeError: + # The event loop already closed after a stream failure. + pass + + def read_stdin() -> None: + try: + while not stop_reader.is_set(): + data = sys.stdin.buffer.read(4096) + if stop_reader.is_set(): + return + post_audio(data or None) + if not data: + return + except Exception as exc: + post_audio(exc) + + reader_thread = threading.Thread( + target=read_stdin, + name="deepctl-stdin-reader", + daemon=True, + ) async def send_audio() -> None: - loop = asyncio.get_event_loop() - while True: - data = await loop.run_in_executor(None, sys.stdin.buffer.read, 4096) - if not data: - break - await ws.send(data) - await ws.send(json.dumps({"type": "CloseStream"})) + reader_thread.start() + try: + while True: + item = await audio_queue.get() + if item is None: + break + if isinstance(item, Exception): + raise item + await ws.send(item) + await ws.send(json.dumps({"type": "CloseStream"})) + finally: + stop_reader.set() async def recv_transcripts() -> None: async for msg in ws: @@ -911,9 +1092,26 @@ async def recv_transcripts() -> None: diarize=diarize, interim=interim, caption_writer=caption_writer, + v2_state=v2_state, ) - await asyncio.gather(send_audio(), recv_transcripts()) + send_task = asyncio.create_task(send_audio()) + recv_task = asyncio.create_task(recv_transcripts()) + try: + await asyncio.gather(send_task, recv_task) + except (KeyboardInterrupt, asyncio.CancelledError): + # Mirror the mic path: on Ctrl-C drain the tasks but do NOT + # re-raise, so we fall through to _flush_v2 + the return below + # and the partial transcript (and --save-to) survive. Fatal + # errors still propagate via the BaseException clause. + await _cancel_and_drain(send_task, recv_task) + except BaseException: + await _cancel_and_drain(send_task, recv_task) + raise + + # Flux may close mid-turn without an EndOfTurn; emit what we have. + if v2_state is not None: + self._flush_v2(v2_state, full_transcript, caption_writer=caption_writer) return ListenResult( status="success", @@ -942,22 +1140,40 @@ def _ws_url( encoding: str, sample_rate: int, channels: int, + redact: tuple[str, ...] = (), + numerals: bool = False, ) -> str: + # v1 and v2 (Flux) have different query-param vocabularies. The v2 + # endpoint rejects v1-only params (language, smart_format, punctuate, + # channels, diarize, interim_results) with HTTP 400, so build the + # param set per version rather than sending the v1 shape to both. params: dict[str, Any] = { "model": model, - "language": language, - "smart_format": "true" if smart_format else "false", - "punctuate": "true" if punctuate else "false", "encoding": encoding, "sample_rate": sample_rate, - "channels": channels, } - if diarize: - params["diarize"] = "true" - if interim: - params["interim_results"] = "true" + if api_version >= 2: + # Flux (listen v2): turn-based, no interim/diarize/smart_format; + # language is encoded in the model name (e.g. flux-general-en). + pass + else: + params["language"] = language + params["smart_format"] = "true" if smart_format else "false" + params["punctuate"] = "true" if punctuate else "false" + params["channels"] = channels + if diarize: + params["diarize"] = "true" + if interim: + params["interim_results"] = "true" + # redact / numerals are valid on both versions. redact is repeatable; + # doseq=True expands a sequence into redact=pci&redact=numbers (and + # leaves a bare string as a single scalar param). + if redact: + params["redact"] = redact + if numerals: + params["numerals"] = "true" base = _ws_base(client) - return f"{base}/v{api_version}/listen?{urlencode(params)}" + return f"{base}/v{api_version}/listen?{urlencode(params, doseq=True)}" def _handle_ws_message( self, @@ -967,14 +1183,40 @@ def _handle_ws_message( diarize: bool, interim: bool, caption_writer: StreamingCaptionWriter | None = None, + v2_state: dict[str, Any] | None = None, ) -> None: - """Parse one WebSocket message and print/accumulate the transcript.""" + """Parse one WebSocket message and print/accumulate the transcript. + + Handles both v1 (`Results`) and Flux/v2 (`TurnInfo`) message shapes; + other control frames (Connected, Metadata, …) are ignored. Flux turns + are stateful, so v2 callers must pass a ``v2_state`` dict (see + ``_new_v2_state``) and call ``_flush_v2`` once the stream closes. + """ try: data = json.loads(raw_msg) except Exception: return - if data.get("type") != "Results": + msg_type = data.get("type") + if msg_type == "TurnInfo": + if v2_state is not None: + self._handle_v2_turn( + data, + transcript_acc, + v2_state, + interim=interim, + caption_writer=caption_writer, + ) + return + + # Flux STT (v2) fatal errors arrive as a control frame (type "Error"). + # Raising propagates the failure through the stream and Click exit status. + if msg_type == "Error" and v2_state is not None: + code = data.get("code") or "UNKNOWN_ERROR" + description = data.get("description") or "No description provided" + raise click.ClickException(f"Flux STT error ({code}): {description}") + + if msg_type != "Results": return channel = data.get("channel", {}) @@ -1014,6 +1256,131 @@ def _handle_ws_message( if transcript: print(f"\r{transcript} ", end="", flush=True) + @staticmethod + def _new_v2_state() -> dict[str, Any]: + """Per-stream state for Flux (v2) turn tracking. See ``_handle_v2_turn``.""" + return {"turns": {}, "order": []} + + def _handle_v2_turn( + self, + data: dict[str, Any], + transcript_acc: list[str], + v2_state: dict[str, Any], + *, + interim: bool, + caption_writer: StreamingCaptionWriter | None = None, + ) -> None: + """Render a Flux (listen v2) ``TurnInfo`` message. + + Flux is turn-based: a turn's transcript grows across ``Update`` / + ``StartOfTurn`` events and is finalized by ``EndOfTurn`` (the analogue + of v1's ``is_final``). But a finite file/stdin stream often ends + mid-turn, so the final turn may never get an ``EndOfTurn`` — we keep + the latest transcript per turn and ``_flush_v2`` emits any turn left + unfinalized when the socket closes. Diarization is not a v2 feature, + so there are no speaker labels here. + """ + turn_index = data.get("turn_index", 0) + transcript = data.get("transcript", "") + event = data.get("event") + + turns = v2_state["turns"] + st = turns.get(turn_index) + if st is None: + st = { + "transcript": "", + "words": [], + "final": False, + "start": 0.0, + "end": 0.0, + } + turns[turn_index] = st + v2_state["order"].append(turn_index) + + # Keep the most complete transcript seen for this turn, plus the turn's + # audio window — the only timing Flux guarantees. Per-word start/end are + # optional on TurnInfo and usually absent, so captions key off the window. + if transcript: + st["transcript"] = transcript + st["words"] = data.get("words", []) + st["start"] = data.get("audio_window_start", st["start"]) + st["end"] = data.get("audio_window_end", st["end"]) + + if event == "EndOfTurn": + self._emit_v2_turn(st, transcript_acc, caption_writer=caption_writer) + elif event == "Update" and interim and not caption_writer and transcript: + print(f"\r{transcript} ", end="", flush=True) + + def _emit_v2_turn( + self, + st: dict[str, Any], + transcript_acc: list[str], + *, + caption_writer: StreamingCaptionWriter | None, + ) -> None: + """Finalize one Flux turn: print it (or route words to captions) once.""" + if st["final"]: + return + st["final"] = True + transcript = st["transcript"] + if not transcript: + return + if caption_writer: + start, end = st["start"], st["end"] + words = self._timed_v2_words(st["words"], transcript, start, end) + caption_writer.write_entry(words, start, end) + transcript_acc.append(transcript) + else: + transcript_acc.append(transcript) + print(transcript, flush=True) + + @staticmethod + def _timed_v2_words( + words: list[dict[str, Any]], + transcript: str, + start: float, + end: float, + ) -> list[dict[str, Any]]: + """Give Flux turn words the start/end the caption converter requires. + + ``captions_from_words`` raises ``KeyError: 'start'`` on words without + timings, and Flux ``TurnInfo`` words carry only ``{word, confidence}``. + Spread the turn's audio window evenly across the words (keeping any real + per-word timings Flux does send), and synthesise a single word from the + transcript if the turn arrived without a word list at all. + """ + if not words: + return [{"word": transcript, "start": start, "end": end}] + if all(w.get("start") is not None and w.get("end") is not None for w in words): + return words + n = len(words) + span = max(0.0, end - start) + step = span / n if n else 0.0 + timed: list[dict[str, Any]] = [] + for i, w in enumerate(words): + tw = dict(w) + if tw.get("start") is None: + tw["start"] = start + i * step + if tw.get("end") is None: + tw["end"] = start + (i + 1) * step + timed.append(tw) + return timed + + def _flush_v2( + self, + v2_state: dict[str, Any], + transcript_acc: list[str], + *, + caption_writer: StreamingCaptionWriter | None = None, + ) -> None: + """Emit any Flux turns the stream closed without an ``EndOfTurn``.""" + for turn_index in v2_state["order"]: + self._emit_v2_turn( + v2_state["turns"][turn_index], + transcript_acc, + caption_writer=caption_writer, + ) + # ── Output rendering ─────────────────────────────────────────────── def output_result(self, result: Any, config: Config) -> None: diff --git a/packages/deepctl-cmd-listen/tests/unit/test_captions.py b/packages/deepctl-cmd-listen/tests/unit/test_captions.py index e1977c3..920a074 100644 --- a/packages/deepctl-cmd-listen/tests/unit/test_captions.py +++ b/packages/deepctl-cmd-listen/tests/unit/test_captions.py @@ -14,8 +14,7 @@ def _words(*entries: tuple[str, float, float]) -> list[dict]: """Build word dicts from (text, start, end) tuples.""" return [ - {"word": t, "punctuated_word": t, "start": s, "end": e} - for t, s, e in entries + {"word": t, "punctuated_word": t, "start": s, "end": e} for t, s, e in entries ] diff --git a/packages/deepctl-cmd-listen/tests/unit/test_formatters.py b/packages/deepctl-cmd-listen/tests/unit/test_formatters.py index abceb9e..acb0c8f 100644 --- a/packages/deepctl-cmd-listen/tests/unit/test_formatters.py +++ b/packages/deepctl-cmd-listen/tests/unit/test_formatters.py @@ -52,7 +52,13 @@ def test_speaker_change_mid_sequence(self): def test_uses_punctuated_word_when_available(self): words = [ - {"word": "hello", "punctuated_word": "Hello,", "start": 0.0, "end": 0.5, "speaker": 0} + { + "word": "hello", + "punctuated_word": "Hello,", + "start": 0.0, + "end": 0.5, + "speaker": 0, + } ] assert "Hello," in format_diarized_words(words) @@ -64,7 +70,9 @@ def test_empty_words_returns_empty_string(self): assert format_diarized_words([]) == "" def test_skips_words_with_no_text(self): - words = [{"word": "", "punctuated_word": "", "start": 0.0, "end": 0.5, "speaker": 0}] + words = [ + {"word": "", "punctuated_word": "", "start": 0.0, "end": 0.5, "speaker": 0} + ] assert format_diarized_words(words) == "" def test_multiple_speaker_changes(self): @@ -82,11 +90,7 @@ def test_multiple_speaker_changes(self): class TestFormatDiarizedTranscript: def _api_result(self, words: list[dict]) -> dict: - return { - "results": { - "channels": [{"alternatives": [{"words": words}]}] - } - } + return {"results": {"channels": [{"alternatives": [{"words": words}]}]}} def test_extracts_and_formats_speakers(self): words = _words(("Hello", 0.0, 0.5, 0), ("Hi", 0.6, 1.0, 1)) @@ -111,7 +115,9 @@ class TestExtractPlainTranscript: def test_extracts_from_channel_alternatives(self): result = { "results": { - "channels": [{"alternatives": [{"transcript": "Hello world", "words": []}]}] + "channels": [ + {"alternatives": [{"transcript": "Hello world", "words": []}]} + ] } } assert extract_plain_transcript(result) == "Hello world" diff --git a/packages/deepctl-cmd-listen/tests/unit/test_listen_command.py b/packages/deepctl-cmd-listen/tests/unit/test_listen_command.py index 9ef74f6..b790c00 100644 --- a/packages/deepctl-cmd-listen/tests/unit/test_listen_command.py +++ b/packages/deepctl-cmd-listen/tests/unit/test_listen_command.py @@ -2,10 +2,62 @@ from unittest.mock import Mock, patch +import click import pytest +from click.testing import CliRunner from deepctl_cmd_listen.command import ListenCommand from deepctl_cmd_listen.models import ListenResult -from deepctl_core import AuthManager, BaseResult, Config, DeepgramClient +from deepctl_core import ( + AuthManager, + BaseResult, + Config, + DeepgramClient, + PluginManager, +) + + +def _install_flux_error_websockets(monkeypatch, ready=None): + import asyncio + import json + import sys + import types + + error_frame = json.dumps( + { + "type": "Error", + "code": "INVALID_AUDIO", + "description": "audio stream is invalid", + } + ) + + class _FakeWS: + def __init__(self): + self.messages = iter([error_frame]) + + async def send(self, _data): + return None + + def __aiter__(self): + return self + + async def __anext__(self): + while ready is not None and not ready(): + await asyncio.sleep(0) + try: + return next(self.messages) + except StopIteration: + raise StopAsyncIteration from None + + class _FakeConnect: + async def __aenter__(self): + return _FakeWS() + + async def __aexit__(self, *_args): + return False + + fake_websockets = types.ModuleType("websockets") + fake_websockets.connect = lambda *_args, **_kwargs: _FakeConnect() + monkeypatch.setitem(sys.modules, "websockets", fake_websockets) class TestListenCommand: @@ -29,6 +81,20 @@ def mock_auth_manager(self): def mock_client(self): return Mock(spec=DeepgramClient) + def invoke_click(self, command, mock_config, args, *, input=None): + mock_config.get.side_effect = lambda key, default=None: ( + True if key == "output.quiet" else default + ) + mock_config.get_profile.return_value.base_url = "https://api.deepgram.com" + click_command = PluginManager()._create_click_command(command) + with patch("deepctl_core.base_command.AuthManager.guard"): + return CliRunner().invoke( + click_command, + args, + input=input, + obj={"config": mock_config}, + ) + def test_command_properties(self, command): assert command.name == "listen" assert command.requires_auth is True @@ -50,12 +116,29 @@ def test_get_arguments(self, command): option_names.extend(arg["names"]) for expected in [ - "--mic", "--model", "-m", "--language", "-l", - "--diarize", "--smart-format", "--punctuate", - "--summarize", "--topics", "--sentiment", - "--interim", "--encoding", "--sample-rate", "--channels", - "--save-to", "-s", "--probe", "--no-validate", - "--webvtt", "--srt", + "--mic", + "--model", + "-m", + "--language", + "-l", + "--diarize", + "--smart-format", + "--punctuate", + "--summarize", + "--topics", + "--sentiment", + "--redact", + "--numerals", + "--interim", + "--encoding", + "--sample-rate", + "--channels", + "--save-to", + "-s", + "--probe", + "--no-validate", + "--webvtt", + "--srt", ]: assert expected in option_names, f"Missing option: {expected}" @@ -85,7 +168,11 @@ def test_handle_mic_no_sounddevice( """--mic without sounddevice installed returns an error.""" mock_sys.stdin.isatty.return_value = True - original_import = __builtins__.__import__ if hasattr(__builtins__, "__import__") else __import__ + original_import = ( + __builtins__.__import__ + if hasattr(__builtins__, "__import__") + else __import__ + ) def mock_import(name, *args, **kwargs): if name == "sounddevice": @@ -111,7 +198,9 @@ def test_handle_stdin_routes_to_stream_stdin( mock_sys.stdin.isatty.return_value = False expected = ListenResult(status="success", source="stdin", mode="live") - with patch.object(command, "_stream_stdin", return_value=expected) as mock_stream: + with patch.object( + command, "_stream_stdin", return_value=expected + ) as mock_stream: result = command.handle( config=mock_config, auth_manager=mock_auth_manager, @@ -133,6 +222,115 @@ def test_handle_stdin_routes_to_stream_stdin( assert call_kwargs["encoding"] == "linear16" assert result.source == "stdin" + @patch("deepctl_cmd_listen.command.status") + @patch("deepctl_cmd_listen.command.sys") + def test_handle_warns_diarize_ignored_on_flux( + self, + mock_sys, + mock_status, + command, + mock_config, + mock_auth_manager, + mock_client, + ): + """--diarize on a Flux STT (v2) model warns instead of vanishing silently.""" + mock_sys.stdin.isatty.return_value = True + + async def fake_ws_mic(_client, **kwargs): + return ListenResult( + status="success", + source="mic", + mode="live", + diarized=kwargs["diarize"], + ) + + with patch.object(command, "_ws_mic", side_effect=fake_ws_mic) as mock_ws: + result = command.handle( + config=mock_config, + auth_manager=mock_auth_manager, + client=mock_client, + mic=True, + model="flux-general-en", + diarize=True, + ) + + printed = " ".join( + str(c.args[0]) for c in mock_status.print.call_args_list if c.args + ) + assert "not supported by Flux STT" in printed + assert "Speaker labels enabled" not in printed + assert mock_ws.call_args.kwargs["diarize"] is False + assert result.diarized is False + + @patch("deepctl_cmd_listen.command.status") + @patch("deepctl_cmd_listen.command.sys") + def test_handle_no_diarize_warning_on_v1( + self, + mock_sys, + mock_status, + command, + mock_config, + mock_auth_manager, + mock_client, + ): + """v1 models keep diarization — no spurious warning.""" + mock_sys.stdin.isatty.return_value = True + expected = ListenResult(status="success", source="mic", mode="live") + + with patch.object(command, "_stream_mic", return_value=expected): + command.handle( + config=mock_config, + auth_manager=mock_auth_manager, + client=mock_client, + mic=True, + model="nova-3", + diarize=True, + ) + + printed = " ".join( + str(c.args[0]) for c in mock_status.print.call_args_list if c.args + ) + assert "not supported by Flux STT" not in printed + + @pytest.mark.parametrize("source", ["call.wav", "https://example.com/call.wav"]) + def test_click_flux_prerecorded_source_errors(self, source, command, mock_config): + """Flux STT file/URL guards are visible and exit non-zero.""" + result = self.invoke_click( + command, + mock_config, + [source, "--model", "flux-general-en"], + ) + assert result.exit_code == 1 + assert "streaming-only" in result.stderr + + def test_click_flux_invalid_redact_errors(self, command, mock_config): + """An invalid Flux redact guard is visible and exits non-zero.""" + result = self.invoke_click( + command, + mock_config, + ["--mic", "--model", "flux-general-en", "--redact", "pci"], + ) + assert result.exit_code == 1 + assert "pci" in result.stderr + assert "aggressive_numbers" in result.stderr + + def test_click_flux_error_frame_exits_nonzero( + self, command, mock_config, monkeypatch + ): + """A stdin Error frame propagates through the stream and Click boundary.""" + _install_flux_error_websockets(monkeypatch) + + result = self.invoke_click( + command, + mock_config, + ["-", "--model", "flux-general-en", "--encoding", "linear16"], + input=b"", + ) + + assert result.exit_code == 1 + assert "INVALID_AUDIO" in result.stderr + assert "audio stream is invalid" in result.stderr + @patch("deepctl_cmd_listen.command.sys") def test_handle_mic_routes_to_stream_mic( self, mock_sys, command, mock_config, mock_auth_manager, mock_client @@ -169,13 +367,19 @@ def test_handle_file_source_routes_to_prerecorded( """A file path routes to _prerecorded with is_url=False.""" mock_sys.stdin.isatty.return_value = True expected = ListenResult( - status="success", source="file", mode="prerecorded", + status="success", + source="file", + mode="prerecorded", transcript="hello world", ) with patch.object(command, "_prerecorded", return_value=expected) as mock_pre: # Skip interactive feature selection - with patch.object(command, "_interactive_features", return_value=(False, False, False, False)): + with patch.object( + command, + "_interactive_features", + return_value=(False, False, False, False), + ): result = command.handle( config=mock_config, auth_manager=mock_auth_manager, @@ -191,6 +395,205 @@ def test_handle_file_source_routes_to_prerecorded( assert call_kwargs["is_url"] is False assert result.source == "file" + @patch("deepctl_cmd_listen.command._agentic", False) + @patch("deepctl_cmd_listen.command.sys") + def test_handle_passes_redact_and_numerals_to_prerecorded( + self, mock_sys, command, mock_config, mock_auth_manager, mock_client + ): + """--redact / --numerals reach _prerecorded.""" + mock_sys.stdin.isatty.return_value = True + expected = ListenResult(status="success", source="file", mode="prerecorded") + + with patch.object(command, "_prerecorded", return_value=expected) as mock_pre: + with patch.object( + command, + "_interactive_features", + return_value=(False, False, False, False), + ): + command.handle( + config=mock_config, + auth_manager=mock_auth_manager, + client=mock_client, + source="audio.mp3", + mic=False, + model="nova-3", + language="en-US", + redact=("numbers",), + numerals=True, + ) + + call_kwargs = mock_pre.call_args.kwargs + assert call_kwargs["redact"] == ("numbers",) + assert call_kwargs["numerals"] is True + + @patch("deepctl_cmd_listen.command._agentic", False) + @patch("deepctl_cmd_listen.command.sys") + def test_handle_passes_multiple_redact_to_prerecorded( + self, mock_sys, command, mock_config, mock_auth_manager, mock_client + ): + """A repeated --redact (v1) reaches _prerecorded as a tuple of values.""" + mock_sys.stdin.isatty.return_value = True + expected = ListenResult(status="success", source="file", mode="prerecorded") + + with patch.object(command, "_prerecorded", return_value=expected) as mock_pre: + with patch.object( + command, + "_interactive_features", + return_value=(False, False, False, False), + ): + command.handle( + config=mock_config, + auth_manager=mock_auth_manager, + client=mock_client, + source="audio.mp3", + mic=False, + model="nova-3", + language="en-US", + redact=("pci", "numbers"), + ) + + assert mock_pre.call_args.kwargs["redact"] == ("pci", "numbers") + + def test_prerecorded_builds_multi_redact_and_numerals_options( + self, command, mock_config + ): + """_prerecorded sends redact as a LIST (so Fern repeats the query param) + and numerals as the string 'true'.""" + client = Mock() + client.transcribe_file.return_value = { + "results": {"channels": [{"alternatives": [{"transcript": "hi"}]}]} + } + result = command._prerecorded( + client, + "audio.wav", + is_url=False, + model="nova-3", + language="en-US", + api_version=1, + diarize=False, + smart_format=True, + punctuate=True, + summarize=False, + topics=False, + sentiment=False, + redact=("pci", "numbers"), + numerals=True, + save_to=None, + probe=False, + no_validate=True, + caption_format=None, + config=mock_config, + ) + + assert result.status == "success" + opts = client.transcribe_file.call_args.args[1] + assert opts["redact"] == ["pci", "numbers"] # list, not tuple + assert opts["numerals"] == "true" + + def test_prerecorded_api_error_exits_nonzero(self, command, mock_config): + """A v1 REST rejection (e.g. a --redact value the endpoint refuses) + raises ClickException so the command exits non-zero and is visible — + not a returned error result that prints nothing and still exits 0.""" + client = Mock() + client.transcribe_file.side_effect = Exception("Bad Request: invalid redact") + with pytest.raises(click.ClickException) as exc_info: + command._prerecorded( + client, + "audio.wav", + is_url=False, + model="nova-3", + language="en-US", + api_version=1, + diarize=False, + smart_format=True, + punctuate=True, + summarize=False, + topics=False, + sentiment=False, + redact=("bogus",), + numerals=False, + save_to=None, + probe=False, + no_validate=True, + caption_format=None, + config=mock_config, + ) + assert "Transcription failed" in str(exc_info.value) + + def test_ws_url_expands_multiple_redact(self, command): + """Repeated redact values expand to repeated query params (doseq).""" + ws_client = Mock() + ws_client.config.get_profile.return_value = Mock( + base_url="https://api.deepgram.com" + ) + url = command._ws_url( + ws_client, + api_version=1, + model="nova-3", + language="en-US", + diarize=False, + smart_format=True, + punctuate=True, + interim=False, + encoding="linear16", + sample_rate=16000, + channels=1, + redact=("pci", "numbers"), + ) + assert "redact=pci" in url + assert "redact=numbers" in url + + def test_ws_url_includes_redact_and_numerals(self, command): + """redact / numerals become query params on the streaming URL.""" + ws_client = Mock() + ws_client.config.get_profile.return_value = Mock( + base_url="https://api.deepgram.com" + ) + + url = command._ws_url( + ws_client, + api_version=2, + model="flux-general-en", + language="en-US", + diarize=False, + smart_format=True, + punctuate=True, + interim=False, + encoding="linear16", + sample_rate=16000, + channels=1, + redact="aggressive_numbers", + numerals=True, + ) + + assert url.startswith("wss://api.deepgram.com/v2/listen?") + assert "redact=aggressive_numbers" in url + assert "numerals=true" in url + + def test_ws_url_omits_redact_and_numerals_when_unset(self, command): + """Unset redact / numerals are not sent (defaults).""" + ws_client = Mock() + ws_client.config.get_profile.return_value = Mock( + base_url="https://api.deepgram.com" + ) + + url = command._ws_url( + ws_client, + api_version=1, + model="nova-3", + language="en-US", + diarize=False, + smart_format=True, + punctuate=True, + interim=False, + encoding="linear16", + sample_rate=16000, + channels=1, + ) + + assert "redact=" not in url + assert "numerals=" not in url + @patch("deepctl_cmd_listen.command._agentic", False) @patch("deepctl_cmd_listen.command.sys") def test_handle_url_source_routes_to_prerecorded( @@ -199,12 +602,18 @@ def test_handle_url_source_routes_to_prerecorded( """A URL routes to _prerecorded with is_url=True.""" mock_sys.stdin.isatty.return_value = True expected = ListenResult( - status="success", source="url", mode="prerecorded", + status="success", + source="url", + mode="prerecorded", transcript="hello", ) with patch.object(command, "_prerecorded", return_value=expected) as mock_pre: - with patch.object(command, "_interactive_features", return_value=(False, False, False, False)): + with patch.object( + command, + "_interactive_features", + return_value=(False, False, False, False), + ): result = command.handle( config=mock_config, auth_manager=mock_auth_manager, @@ -228,7 +637,9 @@ def test_handle_explicit_stdin_dash( mock_sys.stdin.isatty.return_value = True # would normally trigger interactive expected = ListenResult(status="success", source="stdin", mode="live") - with patch.object(command, "_stream_stdin", return_value=expected) as mock_stream: + with patch.object( + command, "_stream_stdin", return_value=expected + ) as mock_stream: result = command.handle( config=mock_config, auth_manager=mock_auth_manager, @@ -257,34 +668,30 @@ def common_kwargs(self): } def test_url_arg_skips_both_prompts(self, command, common_kwargs): - with patch.object(command, "_interactive_features") as feat, patch.object( - command, "_interactive_select_source" - ) as src, patch.object( - command, "_prerecorded", return_value=BaseResult(status="ok") + with ( + patch.object(command, "_interactive_features") as feat, + patch.object(command, "_interactive_select_source") as src, + patch.object(command, "_prerecorded", return_value=BaseResult(status="ok")), ): - command.handle( - **common_kwargs, source="https://example.com/audio.wav" - ) + command.handle(**common_kwargs, source="https://example.com/audio.wav") assert feat.call_count == 0 assert src.call_count == 0 def test_file_arg_skips_both_prompts(self, command, common_kwargs): - with patch.object(command, "_interactive_features") as feat, patch.object( - command, "_interactive_select_source" - ) as src, patch.object( - command, "_prerecorded", return_value=BaseResult(status="ok") + with ( + patch.object(command, "_interactive_features") as feat, + patch.object(command, "_interactive_select_source") as src, + patch.object(command, "_prerecorded", return_value=BaseResult(status="ok")), ): command.handle(**common_kwargs, source="/tmp/audio.wav") assert feat.call_count == 0 assert src.call_count == 0 - def test_url_arg_with_diarize_skips_both_prompts( - self, command, common_kwargs - ): - with patch.object(command, "_interactive_features") as feat, patch.object( - command, "_interactive_select_source" - ) as src, patch.object( - command, "_prerecorded", return_value=BaseResult(status="ok") + def test_url_arg_with_diarize_skips_both_prompts(self, command, common_kwargs): + with ( + patch.object(command, "_interactive_features") as feat, + patch.object(command, "_interactive_select_source") as src, + patch.object(command, "_prerecorded", return_value=BaseResult(status="ok")), ): command.handle( **common_kwargs, @@ -294,38 +701,35 @@ def test_url_arg_with_diarize_skips_both_prompts( assert feat.call_count == 0 assert src.call_count == 0 - def test_bare_invocation_runs_full_guided_flow( - self, command, common_kwargs - ): - with patch.object( - command, - "_interactive_features", - return_value=(False, False, False, False), - ) as feat, patch.object( - command, - "_interactive_select_source", - return_value=("prerecorded_url", "https://x.com/a.wav"), - ) as src, patch.object( - command, "_prerecorded", return_value=BaseResult(status="ok") - ), patch( - "sys.stdin" - ) as mock_stdin, patch( - "deepctl_cmd_listen.command._agentic", False + def test_bare_invocation_runs_full_guided_flow(self, command, common_kwargs): + with ( + patch.object( + command, + "_interactive_features", + return_value=(False, False, False, False), + ) as feat, + patch.object( + command, + "_interactive_select_source", + return_value=("prerecorded_url", "https://x.com/a.wav"), + ) as src, + patch.object(command, "_prerecorded", return_value=BaseResult(status="ok")), + patch("sys.stdin") as mock_stdin, + patch("deepctl_cmd_listen.command._agentic", False), ): mock_stdin.isatty.return_value = True command.handle(**common_kwargs) assert src.call_count == 1 assert feat.call_count == 1 - def test_cancelled_source_select_returns_cancelled( - self, command, common_kwargs - ): - with patch.object( - command, "_interactive_select_source", return_value=(None, None) - ), patch.object(command, "_interactive_features") as feat, patch( - "sys.stdin" - ) as mock_stdin, patch( - "deepctl_cmd_listen.command._agentic", False + def test_cancelled_source_select_returns_cancelled(self, command, common_kwargs): + with ( + patch.object( + command, "_interactive_select_source", return_value=(None, None) + ), + patch.object(command, "_interactive_features") as feat, + patch("sys.stdin") as mock_stdin, + patch("deepctl_cmd_listen.command._agentic", False), ): mock_stdin.isatty.return_value = True result = command.handle(**common_kwargs) @@ -360,3 +764,569 @@ def test_listen_result_defaults(self): assert result.mode == "" assert result.diarized is False assert result.full_result is None + + +class TestFluxV2TurnHandling: + """Flux (listen v2) TurnInfo parsing and finalization.""" + + @pytest.fixture + def command(self): + return ListenCommand() + + def _turn(self, event, transcript, *, turn_index=0, words=None, window=(0.0, 0.0)): + import json as _json + + return _json.dumps( + { + "type": "TurnInfo", + "event": event, + "turn_index": turn_index, + "transcript": transcript, + "words": words or [], + "audio_window_start": window[0], + "audio_window_end": window[1], + } + ) + + def test_end_of_turn_finalizes_transcript(self, command, capsys): + acc: list[str] = [] + state = command._new_v2_state() + + command._handle_ws_message( + self._turn("Update", "hello"), + acc, + diarize=False, + interim=False, + v2_state=state, + ) + # Update alone does not finalize. + assert acc == [] + + command._handle_ws_message( + self._turn("EndOfTurn", "hello world"), + acc, + diarize=False, + interim=False, + v2_state=state, + ) + assert acc == ["hello world"] + assert "hello world" in capsys.readouterr().out + + def test_flush_emits_unfinalized_final_turn(self, command, capsys): + """A stream that closes mid-turn still yields the latest transcript.""" + acc: list[str] = [] + state = command._new_v2_state() + + # Turn grows across Updates but never gets an EndOfTurn. + for text in ("my", "my account", "my account number"): + command._handle_ws_message( + self._turn("Update", text), + acc, + diarize=False, + interim=False, + v2_state=state, + ) + assert acc == [] # nothing finalized yet + + command._flush_v2(state, acc) + assert acc == ["my account number"] + + def test_flush_does_not_double_emit_finalized_turn(self, command): + acc: list[str] = [] + state = command._new_v2_state() + + command._handle_ws_message( + self._turn("EndOfTurn", "done"), + acc, + diarize=False, + interim=False, + v2_state=state, + ) + command._flush_v2(state, acc) + assert acc == ["done"] # not duplicated + + def test_multiple_turns_accumulate_in_order(self, command): + acc: list[str] = [] + state = command._new_v2_state() + + command._handle_ws_message( + self._turn("EndOfTurn", "first turn", turn_index=0), + acc, + diarize=False, + interim=False, + v2_state=state, + ) + command._handle_ws_message( + self._turn("Update", "second turn", turn_index=1), + acc, + diarize=False, + interim=False, + v2_state=state, + ) + command._flush_v2(state, acc) + assert acc == ["first turn", "second turn"] + + def test_turninfo_ignored_without_state(self, command): + """A v2 message with no state (v1 caller) is a no-op, not a crash.""" + acc: list[str] = [] + command._handle_ws_message( + self._turn("EndOfTurn", "ignored"), + acc, + diarize=False, + interim=False, + v2_state=None, + ) + assert acc == [] + + def test_captions_use_audio_window_and_survive_batch_save(self, command): + """Flux words lack per-word timings: captions must key off the turn's + audio window, and the end-of-stream batch save must not KeyError.""" + from deepctl_cmd_listen.captions import ( + StreamingCaptionWriter, + captions_from_words, + ) + + writer = StreamingCaptionWriter("srt") + acc: list[str] = [] + state = command._new_v2_state() + + # TurnInfo words carry only {word, confidence} — no start/end. + words = [ + {"word": "my", "confidence": 0.9}, + {"word": "account", "confidence": 0.9}, + ] + command._handle_ws_message( + self._turn("EndOfTurn", "my account", words=words, window=(1.0, 2.5)), + acc, + diarize=False, + interim=False, + v2_state=state, + caption_writer=writer, + ) + + assert acc == ["my account"] + # Live cue span comes from the audio window, not 00:00:00. + assert writer.accumulated_words # words were captured for batch save + # The batch save path used to raise KeyError: 'start' on Flux words. + batch = captions_from_words(writer.accumulated_words, "srt") + assert "my account" in batch + assert "00:00:01,000 --> 00:00:02,500" in batch + + def test_ws_mic_flushes_final_turn_on_keyboard_interrupt( + self, command, monkeypatch + ): + """Ctrl-C during a Flux mic stream must still flush the in-flight turn. + + The interrupt surfaces at the ``asyncio.gather`` await; ``_ws_mic`` has + to catch it, run ``_flush_v2``, and return the accumulated transcript + (rather than letting the final turn vanish).""" + import asyncio as _asyncio + import sys + import types + from unittest.mock import MagicMock + + # _ws_mic imports these at call time; stub them so no hardware/net is hit. + for name in ("sounddevice", "numpy"): + monkeypatch.setitem(sys.modules, name, types.ModuleType(name)) + + class _FakeWS: + async def send(self, *a): + return None + + def __aiter__(self): + return self + + async def __anext__(self): + raise StopAsyncIteration + + class _FakeConnect: + async def __aenter__(self): + return _FakeWS() + + async def __aexit__(self, *a): + return False + + fake_ws = types.ModuleType("websockets") + fake_ws.connect = lambda *a, **k: _FakeConnect() + monkeypatch.setitem(sys.modules, "websockets", fake_ws) + + # A turn received but never finalized (no EndOfTurn before the interrupt). + seeded = { + "turns": { + 0: { + "transcript": "hello world", + "words": [], + "final": False, + "start": 0.0, + "end": 1.0, + } + }, + "order": [0], + } + monkeypatch.setattr(command, "_new_v2_state", lambda: seeded) + + async def _interrupt(*a, **k): + raise KeyboardInterrupt + + monkeypatch.setattr("deepctl_cmd_listen.command.asyncio.gather", _interrupt) + + client = MagicMock() + client.config.get_profile.return_value.base_url = "https://api.deepgram.com" + client.auth_manager.get_api_key.return_value = "k" + + # Drive on a plain loop (not asyncio.run) so the patched gather can't + # interfere with run()'s own shutdown machinery. + loop = _asyncio.new_event_loop() + try: + result = loop.run_until_complete( + command._ws_mic( + client, + model="flux-general-en", + language="en-US", + api_version=2, + diarize=False, + smart_format=True, + punctuate=True, + interim=False, + redact=(), + numerals=False, + sample_rate=16000, + channels=1, + ) + ) + finally: + loop.close() + + assert result.transcript == "hello world" + + def test_end_of_turn_with_empty_transcript_is_noop(self, command, capsys): + """An EndOfTurn that never carried text emits nothing (no blank line).""" + acc: list[str] = [] + state = command._new_v2_state() + command._handle_ws_message( + self._turn("EndOfTurn", ""), + acc, + diarize=False, + interim=False, + v2_state=state, + ) + assert acc == [] + assert capsys.readouterr().out == "" + + def test_v2_update_prints_interim(self, command, capsys): + """An Update event with --interim shows a carriage-return partial.""" + acc: list[str] = [] + state = command._new_v2_state() + command._handle_ws_message( + self._turn("Update", "partial text"), + acc, + diarize=False, + interim=True, + v2_state=state, + ) + out = capsys.readouterr().out + assert "partial text" in out + assert "\r" in out + assert acc == [] # interim never accumulates + + def test_timed_v2_words_backfills_and_preserves(self, command): + """_timed_v2_words: synthesize when empty, spread the window when words + lack timings, and pass real per-word timings through untouched.""" + # No words → one synthetic word spanning the whole turn window. + assert command._timed_v2_words([], "hello world", 1.0, 2.0) == [ + {"word": "hello world", "start": 1.0, "end": 2.0} + ] + # Words already carrying timings are returned unchanged. + real = [{"word": "a", "start": 0.1, "end": 0.2}] + assert command._timed_v2_words(real, "a", 0.0, 5.0) is real + # Timing-less words get the window spread evenly across them. + spread = command._timed_v2_words( + [{"word": "a"}, {"word": "b"}], "a b", 0.0, 2.0 + ) + assert (spread[0]["start"], spread[0]["end"]) == (0.0, 1.0) + assert (spread[1]["start"], spread[1]["end"]) == (1.0, 2.0) + + def test_timed_v2_words_backfills_partial_and_null_timings(self, command): + partial = command._timed_v2_words( + [{"word": "a", "start": 0.1}, {"word": "b", "end": 1.8}], + "a b", + 0.0, + 2.0, + ) + assert partial == [ + {"word": "a", "start": 0.1, "end": 1.0}, + {"word": "b", "start": 1.0, "end": 1.8}, + ] + + explicit_null = command._timed_v2_words( + [{"word": "a", "start": None, "end": None}], + "a", + 3.0, + 4.0, + ) + assert explicit_null == [{"word": "a", "start": 3.0, "end": 4.0}] + + def test_fatal_error_frame_raises_code_and_description(self, command): + """A Flux STT fatal error carries both server fields to the caller.""" + import json as _json + + acc: list[str] = [] + with pytest.raises(click.ClickException) as exc_info: + command._handle_ws_message( + _json.dumps( + { + "type": "Error", + "code": "INTERNAL_SERVER_ERROR", + "description": "something went wrong", + } + ), + acc, + diarize=False, + interim=False, + v2_state=command._new_v2_state(), + ) + assert acc == [] + assert "INTERNAL_SERVER_ERROR" in exc_info.value.message + assert "something went wrong" in exc_info.value.message + + def test_stream_mic_flux_error_frame_does_not_return_success( + self, command, monkeypatch + ): + import sys + import types + from unittest.mock import MagicMock + + events = {"started": False, "stopped": False, "closed": False} + _install_flux_error_websockets(monkeypatch, ready=lambda: events["started"]) + + class _FakeInputStream: + def __init__(self, **_kwargs): + pass + + def start(self): + events["started"] = True + + def stop(self): + events["stopped"] = True + + def close(self): + events["closed"] = True + + fake_sounddevice = types.ModuleType("sounddevice") + fake_sounddevice.RawInputStream = _FakeInputStream + fake_numpy = types.ModuleType("numpy") + fake_numpy.int16 = "int16" + monkeypatch.setitem(sys.modules, "sounddevice", fake_sounddevice) + monkeypatch.setitem(sys.modules, "numpy", fake_numpy) + + client = MagicMock() + client.config.get_profile.return_value.base_url = "https://api.deepgram.com" + client.auth_manager.get_api_key.return_value = "test-key" + + with pytest.raises(click.ClickException) as exc_info: + command._stream_mic( + client, + model="flux-general-en", + language="en-US", + api_version=2, + diarize=False, + smart_format=True, + punctuate=True, + interim=False, + redact=(), + numerals=False, + sample_rate=16000, + channels=1, + save_to=None, + caption_format=None, + ) + + assert "INVALID_AUDIO" in exc_info.value.message + assert "audio stream is invalid" in exc_info.value.message + assert events == {"started": True, "stopped": True, "closed": True} + + def test_stream_stdin_flux_error_does_not_wait_for_blocked_input( + self, command, monkeypatch + ): + import asyncio as _asyncio + import json as _json + import sys + import threading + import types + from unittest.mock import MagicMock + + read_started = threading.Event() + release_read = threading.Event() + + class _BlockingBuffer: + def read(self, _size): + read_started.set() + release_read.wait() + return b"" + + fake_stdin = types.SimpleNamespace(buffer=_BlockingBuffer()) + monkeypatch.setattr( + "deepctl_cmd_listen.command.sys.stdin", fake_stdin, raising=False + ) + + error_frame = _json.dumps( + { + "type": "Error", + "code": "INVALID_AUDIO", + "description": "audio stream is invalid", + } + ) + + class _FakeWS: + def __init__(self): + self.sent_error = False + + async def send(self, _data): + return None + + def __aiter__(self): + return self + + async def __anext__(self): + if self.sent_error: + raise StopAsyncIteration + while not read_started.is_set(): + await _asyncio.sleep(0) + self.sent_error = True + return error_frame + + class _FakeConnect: + async def __aenter__(self): + return _FakeWS() + + async def __aexit__(self, *_args): + return False + + fake_websockets = types.ModuleType("websockets") + fake_websockets.connect = lambda *_args, **_kwargs: _FakeConnect() + monkeypatch.setitem(sys.modules, "websockets", fake_websockets) + + client = MagicMock() + client.config.get_profile.return_value.base_url = "https://api.deepgram.com" + client.auth_manager.get_api_key.return_value = "test-key" + errors = [] + + def run_stream(): + try: + command._stream_stdin( + client, + model="flux-general-en", + language="en-US", + api_version=2, + diarize=False, + smart_format=True, + punctuate=True, + interim=False, + redact=(), + numerals=False, + encoding="linear16", + sample_rate=16000, + channels=1, + save_to=None, + caption_format=None, + ) + except BaseException as exc: + errors.append(exc) + + stream_thread = threading.Thread(target=run_stream, daemon=True) + stream_thread.start() + assert read_started.wait(timeout=1) + stream_thread.join(timeout=1) + exited_before_stdin = not stream_thread.is_alive() + release_read.set() + stream_thread.join(timeout=1) + + assert exited_before_stdin + assert len(errors) == 1 + assert isinstance(errors[0], click.ClickException) + assert "INVALID_AUDIO" in errors[0].message + + def test_stream_stdin_keyboard_interrupt_preserves_transcript_and_save_to( + self, command, monkeypatch, tmp_path + ): + """Ctrl-C on `dg listen -` (stdin) keeps the partial transcript and + writes --save-to, mirroring the mic path. The stdin path used to + re-raise the interrupt and drop both.""" + import sys + import types + from unittest.mock import MagicMock + + # No real stdin read: return EOF at once if the reader thread starts. + fake_stdin = types.SimpleNamespace( + buffer=types.SimpleNamespace(read=lambda _n: b"") + ) + monkeypatch.setattr( + "deepctl_cmd_listen.command.sys.stdin", fake_stdin, raising=False + ) + + class _FakeWS: + async def send(self, *a): + return None + + def __aiter__(self): + return self + + async def __anext__(self): + raise StopAsyncIteration + + class _FakeConnect: + async def __aenter__(self): + return _FakeWS() + + async def __aexit__(self, *a): + return False + + fake_ws = types.ModuleType("websockets") + fake_ws.connect = lambda *a, **k: _FakeConnect() + monkeypatch.setitem(sys.modules, "websockets", fake_ws) + + # A turn received but never finalized before the interrupt. + seeded = { + "turns": { + 0: { + "transcript": "hello world", + "words": [], + "final": False, + "start": 0.0, + "end": 1.0, + } + }, + "order": [0], + } + monkeypatch.setattr(command, "_new_v2_state", lambda: seeded) + + async def _interrupt(*a, **k): + raise KeyboardInterrupt + + monkeypatch.setattr("deepctl_cmd_listen.command.asyncio.gather", _interrupt) + + client = MagicMock() + client.config.get_profile.return_value.base_url = "https://api.deepgram.com" + client.auth_manager.get_api_key.return_value = "k" + + save_to = tmp_path / "out.txt" + result = command._stream_stdin( + client, + model="flux-general-en", + language="en-US", + api_version=2, + diarize=False, + smart_format=True, + punctuate=True, + interim=False, + redact=(), + numerals=False, + encoding="linear16", + sample_rate=16000, + channels=1, + save_to=str(save_to), + caption_format=None, + ) + + assert result.transcript == "hello world" + assert save_to.read_text().strip() == "hello world" diff --git a/packages/deepctl-cmd-listen/tests/unit/test_ws_and_routing.py b/packages/deepctl-cmd-listen/tests/unit/test_ws_and_routing.py index b1510bc..f173ec4 100644 --- a/packages/deepctl-cmd-listen/tests/unit/test_ws_and_routing.py +++ b/packages/deepctl-cmd-listen/tests/unit/test_ws_and_routing.py @@ -6,6 +6,7 @@ import json from unittest.mock import MagicMock, Mock, patch +import click import pytest from deepctl_cmd_listen.captions import StreamingCaptionWriter from deepctl_cmd_listen.command import ListenCommand @@ -43,18 +44,18 @@ def mock_auth_manager(): class TestWsUrl: def _url(self, command, mock_client, **overrides): - defaults = dict( - api_version=1, - model="nova-3", - language="en-US", - diarize=False, - smart_format=True, - punctuate=True, - interim=False, - encoding="linear16", - sample_rate=16000, - channels=1, - ) + defaults = { + "api_version": 1, + "model": "nova-3", + "language": "en-US", + "diarize": False, + "smart_format": True, + "punctuate": True, + "interim": False, + "encoding": "linear16", + "sample_rate": 16000, + "channels": 1, + } defaults.update(overrides) return command._ws_url(mock_client, **defaults) @@ -71,6 +72,50 @@ def test_v1_path(self, command, mock_client): def test_v2_path(self, command, mock_client): assert "/v2/listen?" in self._url(command, mock_client, api_version=2) + def test_v2_omits_v1_only_params(self, command, mock_client): + """Flux (v2) rejects v1-only params with HTTP 400, so they must not be + sent. This locks in the fix; a regression would silently break Flux STT. + """ + url = self._url( + command, + mock_client, + api_version=2, + diarize=True, + interim=True, + ) + for banned in ( + "language=", + "smart_format=", + "punctuate=", + "channels=", + "diarize=", + "interim_results=", + ): + assert banned not in url, f"v2 URL must not contain {banned!r}: {url}" + # The params v2 does accept are still present. + assert "model=" in url + assert "encoding=" in url + assert "sample_rate=" in url + + def test_v1_includes_v1_params(self, command, mock_client): + """v1 keeps sending the classic params (contrast with v2).""" + url = self._url( + command, + mock_client, + api_version=1, + diarize=True, + interim=True, + ) + for expected in ( + "language=", + "smart_format=", + "punctuate=", + "channels=", + "diarize=true", + "interim_results=true", + ): + assert expected in url, f"v1 URL should contain {expected!r}: {url}" + def test_model_param(self, command, mock_client): assert "model=nova-3" in self._url(command, mock_client, model="nova-3") @@ -91,7 +136,9 @@ def test_interim_param_absent_when_disabled(self, command, mock_client): def test_custom_base_url(self, command): client = MagicMock() - client.config.get_profile.return_value.base_url = "https://custom.api.example.com" + client.config.get_profile.return_value.base_url = ( + "https://custom.api.example.com" + ) assert self._url(command, client).startswith("wss://custom.api.example.com") def test_sample_rate_param(self, command, mock_client): @@ -105,16 +152,24 @@ def test_channels_param(self, command, mock_client): class TestFluxModelAutoVersion: - def _handle_with_source(self, command, mock_config, mock_auth_manager, mock_client, **kwargs): - defaults = dict( - source="audio.mp3", - mic=False, - model="nova-3", - language="en-US", - ) + def _handle_with_source( + self, command, mock_config, mock_auth_manager, mock_client, **kwargs + ): + defaults = { + "source": "audio.mp3", + "mic": False, + "model": "nova-3", + "language": "en-US", + } defaults.update(kwargs) - with patch.object(command, "_prerecorded", return_value=ListenResult(status="success")) as mock_pre: - with patch.object(command, "_interactive_features", return_value=(False, False, False, False)): + with patch.object( + command, "_prerecorded", return_value=ListenResult(status="success") + ) as mock_pre: + with patch.object( + command, + "_interactive_features", + return_value=(False, False, False, False), + ): command.handle( config=mock_config, auth_manager=mock_auth_manager, @@ -123,27 +178,138 @@ def _handle_with_source(self, command, mock_config, mock_auth_manager, mock_clie ) return mock_pre + def _handle_with_mic( + self, command, mock_config, mock_auth_manager, mock_client, **kwargs + ): + defaults = {"mic": True, "model": "nova-3", "language": "en-US"} + defaults.update(kwargs) + with patch.object( + command, "_stream_mic", return_value=ListenResult(status="success") + ) as mock_stream: + command.handle( + config=mock_config, + auth_manager=mock_auth_manager, + client=mock_client, + **defaults, + ) + return mock_stream + @patch("deepctl_cmd_listen.command._agentic", False) @patch("deepctl_cmd_listen.command.sys") - def test_flux_model_uses_v2(self, mock_sys, command, mock_config, mock_auth_manager, mock_client): + def test_flux_model_file_is_error( + self, mock_sys, command, mock_config, mock_auth_manager, mock_client + ): + # Flux STT (v2) is streaming-only: a file must not reach _prerecorded. mock_sys.stdin.isatty.return_value = True - mock_pre = self._handle_with_source( - command, mock_config, mock_auth_manager, mock_client, model="flux-general-en" + with ( + patch.object( + command, + "_interactive_features", + return_value=(False, False, False, False), + ), + pytest.raises(click.ClickException, match="streaming-only"), + ): + command.handle( + config=mock_config, + auth_manager=mock_auth_manager, + client=mock_client, + source="audio.mp3", + mic=False, + model="flux-general-en", + language="en-US", + ) + + @patch("deepctl_cmd_listen.command.sys") + def test_flux_model_streaming_uses_v2( + self, mock_sys, command, mock_config, mock_auth_manager, mock_client + ): + mock_sys.stdin.isatty.return_value = True + mock_stream = self._handle_with_mic( + command, + mock_config, + mock_auth_manager, + mock_client, + model="flux-general-en", ) - assert mock_pre.call_args.kwargs["api_version"] == 2 + assert mock_stream.call_args.kwargs["api_version"] == 2 - @patch("deepctl_cmd_listen.command._agentic", False) @patch("deepctl_cmd_listen.command.sys") - def test_flux_prefix_variant_uses_v2(self, mock_sys, command, mock_config, mock_auth_manager, mock_client): + def test_flux_prefix_variant_streaming_uses_v2( + self, mock_sys, command, mock_config, mock_auth_manager, mock_client + ): mock_sys.stdin.isatty.return_value = True - mock_pre = self._handle_with_source( + mock_stream = self._handle_with_mic( command, mock_config, mock_auth_manager, mock_client, model="flux-2-en" ) - assert mock_pre.call_args.kwargs["api_version"] == 2 + assert mock_stream.call_args.kwargs["api_version"] == 2 + + @pytest.mark.parametrize("model", ["flux", "fluxfoo"]) + @patch("deepctl_cmd_listen.command.sys") + def test_bare_or_typo_flux_model_uses_v1( + self, mock_sys, model, command, mock_config, mock_auth_manager, mock_client + ): + mock_sys.stdin.isatty.return_value = True + mock_stream = self._handle_with_mic( + command, mock_config, mock_auth_manager, mock_client, model=model + ) + assert mock_stream.call_args.kwargs["api_version"] == 1 + assert mock_stream.call_args.kwargs["model"] == model + + @pytest.mark.parametrize( + ("source_kwargs", "stream_method"), + [({"mic": True}, "_stream_mic"), ({"source": "-"}, "_stream_stdin")], + ) + def test_flux_multichannel_rejected_before_streaming( + self, + source_kwargs, + stream_method, + command, + mock_config, + mock_auth_manager, + mock_client, + ): + with ( + patch.object(command, stream_method) as mock_stream, + pytest.raises(click.ClickException, match="mono audio only"), + ): + command.handle( + config=mock_config, + auth_manager=mock_auth_manager, + client=mock_client, + model="flux-general-en", + channels=2, + **source_kwargs, + ) + mock_stream.assert_not_called() + + @pytest.mark.parametrize("channels", [0, -1]) + def test_nonpositive_channels_rejected( + self, + channels, + command, + mock_config, + mock_auth_manager, + mock_client, + ): + with ( + patch.object(command, "_stream_mic") as mock_stream, + pytest.raises(click.ClickException, match="at least 1"), + ): + command.handle( + config=mock_config, + auth_manager=mock_auth_manager, + client=mock_client, + model="flux-general-en", + channels=channels, + mic=True, + ) + mock_stream.assert_not_called() @patch("deepctl_cmd_listen.command._agentic", False) @patch("deepctl_cmd_listen.command.sys") - def test_nova3_uses_v1(self, mock_sys, command, mock_config, mock_auth_manager, mock_client): + def test_nova3_uses_v1( + self, mock_sys, command, mock_config, mock_auth_manager, mock_client + ): mock_sys.stdin.isatty.return_value = True mock_pre = self._handle_with_source( command, mock_config, mock_auth_manager, mock_client, model="nova-3" @@ -152,7 +318,9 @@ def test_nova3_uses_v1(self, mock_sys, command, mock_config, mock_auth_manager, @patch("deepctl_cmd_listen.command._agentic", False) @patch("deepctl_cmd_listen.command.sys") - def test_enhanced_uses_v1(self, mock_sys, command, mock_config, mock_auth_manager, mock_client): + def test_enhanced_uses_v1( + self, mock_sys, command, mock_config, mock_auth_manager, mock_client + ): mock_sys.stdin.isatty.return_value = True mock_pre = self._handle_with_source( command, mock_config, mock_auth_manager, mock_client, model="enhanced" @@ -165,7 +333,9 @@ def test_enhanced_uses_v1(self, mock_sys, command, mock_config, mock_auth_manage class TestCaptionFlagExclusivity: @patch("deepctl_cmd_listen.command.sys") - def test_both_flags_is_error(self, mock_sys, command, mock_config, mock_auth_manager, mock_client): + def test_both_flags_is_error( + self, mock_sys, command, mock_config, mock_auth_manager, mock_client + ): mock_sys.stdin.isatty.return_value = True result = command.handle( config=mock_config, @@ -181,37 +351,74 @@ def test_both_flags_is_error(self, mock_sys, command, mock_config, mock_auth_man @patch("deepctl_cmd_listen.command._agentic", False) @patch("deepctl_cmd_listen.command.sys") - def test_webvtt_alone_passes_format_to_prerecorded(self, mock_sys, command, mock_config, mock_auth_manager, mock_client): + def test_webvtt_alone_passes_format_to_prerecorded( + self, mock_sys, command, mock_config, mock_auth_manager, mock_client + ): mock_sys.stdin.isatty.return_value = True - with patch.object(command, "_prerecorded", return_value=ListenResult(status="success")) as mock_pre: - with patch.object(command, "_interactive_features", return_value=(False, False, False, False)): + with patch.object( + command, "_prerecorded", return_value=ListenResult(status="success") + ) as mock_pre: + with patch.object( + command, + "_interactive_features", + return_value=(False, False, False, False), + ): command.handle( - config=mock_config, auth_manager=mock_auth_manager, client=mock_client, - source="audio.mp3", mic=False, webvtt=True, srt=False, + config=mock_config, + auth_manager=mock_auth_manager, + client=mock_client, + source="audio.mp3", + mic=False, + webvtt=True, + srt=False, ) assert mock_pre.call_args.kwargs["caption_format"] == "webvtt" @patch("deepctl_cmd_listen.command._agentic", False) @patch("deepctl_cmd_listen.command.sys") - def test_srt_alone_passes_format_to_prerecorded(self, mock_sys, command, mock_config, mock_auth_manager, mock_client): + def test_srt_alone_passes_format_to_prerecorded( + self, mock_sys, command, mock_config, mock_auth_manager, mock_client + ): mock_sys.stdin.isatty.return_value = True - with patch.object(command, "_prerecorded", return_value=ListenResult(status="success")) as mock_pre: - with patch.object(command, "_interactive_features", return_value=(False, False, False, False)): + with patch.object( + command, "_prerecorded", return_value=ListenResult(status="success") + ) as mock_pre: + with patch.object( + command, + "_interactive_features", + return_value=(False, False, False, False), + ): command.handle( - config=mock_config, auth_manager=mock_auth_manager, client=mock_client, - source="audio.mp3", mic=False, webvtt=False, srt=True, + config=mock_config, + auth_manager=mock_auth_manager, + client=mock_client, + source="audio.mp3", + mic=False, + webvtt=False, + srt=True, ) assert mock_pre.call_args.kwargs["caption_format"] == "srt" @patch("deepctl_cmd_listen.command._agentic", False) @patch("deepctl_cmd_listen.command.sys") - def test_no_caption_flag_passes_none(self, mock_sys, command, mock_config, mock_auth_manager, mock_client): + def test_no_caption_flag_passes_none( + self, mock_sys, command, mock_config, mock_auth_manager, mock_client + ): mock_sys.stdin.isatty.return_value = True - with patch.object(command, "_prerecorded", return_value=ListenResult(status="success")) as mock_pre: - with patch.object(command, "_interactive_features", return_value=(False, False, False, False)): + with patch.object( + command, "_prerecorded", return_value=ListenResult(status="success") + ) as mock_pre: + with patch.object( + command, + "_interactive_features", + return_value=(False, False, False, False), + ): command.handle( - config=mock_config, auth_manager=mock_auth_manager, client=mock_client, - source="audio.mp3", mic=False, + config=mock_config, + auth_manager=mock_auth_manager, + client=mock_client, + source="audio.mp3", + mic=False, ) assert mock_pre.call_args.kwargs["caption_format"] is None @@ -220,25 +427,39 @@ def test_no_caption_flag_passes_none(self, mock_sys, command, mock_config, mock_ class TestHandleWsMessage: - def _msg(self, transcript, words=None, is_final=True, msg_type="Results", start=0.0, duration=1.0): - return json.dumps({ - "type": msg_type, - "channel": { - "alternatives": [{"transcript": transcript, "words": words or []}] - }, - "is_final": is_final, - "start": start, - "duration": duration, - }) + def _msg( + self, + transcript, + words=None, + is_final=True, + msg_type="Results", + start=0.0, + duration=1.0, + ): + return json.dumps( + { + "type": msg_type, + "channel": { + "alternatives": [{"transcript": transcript, "words": words or []}] + }, + "is_final": is_final, + "start": start, + "duration": duration, + } + ) def test_final_transcript_printed_to_stdout(self, command, capsys): acc = [] - command._handle_ws_message(self._msg("Hello world"), acc, diarize=False, interim=False) + command._handle_ws_message( + self._msg("Hello world"), acc, diarize=False, interim=False + ) assert "Hello world" in capsys.readouterr().out def test_final_transcript_accumulated(self, command, capsys): acc = [] - command._handle_ws_message(self._msg("Hello world"), acc, diarize=False, interim=False) + command._handle_ws_message( + self._msg("Hello world"), acc, diarize=False, interim=False + ) capsys.readouterr() assert acc == ["Hello world"] @@ -269,11 +490,24 @@ def test_interim_not_printed_when_flag_off(self, command, capsys): def test_non_results_type_ignored(self, command, capsys): acc = [] command._handle_ws_message( - json.dumps({"type": "Metadata", "data": "x"}), acc, diarize=False, interim=False + json.dumps({"type": "Metadata", "data": "x"}), + acc, + diarize=False, + interim=False, ) assert acc == [] assert capsys.readouterr().out == "" + def test_error_type_ignored_without_flux_state(self, command): + acc = [] + command._handle_ws_message( + json.dumps({"type": "Error", "code": "V1_ERROR"}), + acc, + diarize=False, + interim=False, + ) + assert acc == [] + def test_empty_transcript_not_accumulated(self, command, capsys): acc = [] command._handle_ws_message(self._msg(""), acc, diarize=False, interim=False) @@ -287,8 +521,20 @@ def test_invalid_json_does_not_raise(self, command): def test_diarized_final_uses_speaker_labels(self, command, capsys): words = [ - {"word": "hello", "punctuated_word": "Hello", "start": 0.0, "end": 0.5, "speaker": 0}, - {"word": "there", "punctuated_word": "there", "start": 0.6, "end": 1.0, "speaker": 1}, + { + "word": "hello", + "punctuated_word": "Hello", + "start": 0.0, + "end": 0.5, + "speaker": 0, + }, + { + "word": "there", + "punctuated_word": "there", + "start": 0.6, + "end": 1.0, + "speaker": 1, + }, ] acc = [] command._handle_ws_message( @@ -299,7 +545,13 @@ def test_diarized_final_uses_speaker_labels(self, command, capsys): def test_diarized_line_accumulated(self, command, capsys): words = [ - {"word": "hi", "punctuated_word": "Hi", "start": 0.0, "end": 0.5, "speaker": 0}, + { + "word": "hi", + "punctuated_word": "Hi", + "start": 0.0, + "end": 0.5, + "speaker": 0, + }, ] acc = [] command._handle_ws_message( @@ -342,7 +594,7 @@ def test_caption_writer_suppresses_plain_text(self, command, capsys): # Should see the caption timestamp, not a bare "Hi\n" assert "-->" in out # The bare transcript line should not appear on its own - lines = [l for l in out.splitlines() if l.strip() == "Hi"] + lines = [line for line in out.splitlines() if line.strip() == "Hi"] assert len(lines) == 0 or "-->" in out # caption mode def test_interim_suppressed_in_caption_mode(self, command, capsys): @@ -360,7 +612,11 @@ def test_interim_suppressed_in_caption_mode(self, command, capsys): def test_multiple_messages_accumulate(self, command, capsys): acc = [] - command._handle_ws_message(self._msg("First"), acc, diarize=False, interim=False) - command._handle_ws_message(self._msg("Second"), acc, diarize=False, interim=False) + command._handle_ws_message( + self._msg("First"), acc, diarize=False, interim=False + ) + command._handle_ws_message( + self._msg("Second"), acc, diarize=False, interim=False + ) capsys.readouterr() assert acc == ["First", "Second"] diff --git a/packages/deepctl-cmd-login/README.md b/packages/deepctl-cmd-login/README.md index 6d5b15c..9ff99fa 100644 --- a/packages/deepctl-cmd-login/README.md +++ b/packages/deepctl-cmd-login/README.md @@ -38,7 +38,7 @@ pipx run deepctl --help - `click>=8.0.0` - `rich>=13.0.0` -- `deepgram-sdk>=6.0.0rc2` +- `deepgram-sdk>=7.7.0,<8` - `pydantic>=2.0.0` ## License diff --git a/packages/deepctl-cmd-login/pyproject.toml b/packages/deepctl-cmd-login/pyproject.toml index e6b0106..a8a0669 100644 --- a/packages/deepctl-cmd-login/pyproject.toml +++ b/packages/deepctl-cmd-login/pyproject.toml @@ -24,7 +24,7 @@ dependencies = [ "deepctl-core>=0.1.10", "click>=8.0.0", "rich>=13.0.0", - "deepgram-sdk>=7.5.0", + "deepgram-sdk>=7.7.0,<8", "pydantic>=2.0.0", ] diff --git a/packages/deepctl-cmd-projects/README.md b/packages/deepctl-cmd-projects/README.md index 49bc8cf..1ef69b0 100644 --- a/packages/deepctl-cmd-projects/README.md +++ b/packages/deepctl-cmd-projects/README.md @@ -35,7 +35,7 @@ pipx run deepctl --help - `click>=8.0.0` - `rich>=13.0.0` -- `deepgram-sdk>=6.0.0rc2` +- `deepgram-sdk>=7.7.0,<8` - `pydantic>=2.0.0` ## License diff --git a/packages/deepctl-cmd-projects/pyproject.toml b/packages/deepctl-cmd-projects/pyproject.toml index d2d2b8a..56aee09 100644 --- a/packages/deepctl-cmd-projects/pyproject.toml +++ b/packages/deepctl-cmd-projects/pyproject.toml @@ -24,7 +24,7 @@ dependencies = [ "deepctl-core>=0.1.10", "click>=8.0.0", "rich>=13.0.0", - "deepgram-sdk>=7.5.0", + "deepgram-sdk>=7.7.0,<8", "pydantic>=2.0.0", ] diff --git a/packages/deepctl-cmd-speak/pyproject.toml b/packages/deepctl-cmd-speak/pyproject.toml index dc274e9..c66313c 100644 --- a/packages/deepctl-cmd-speak/pyproject.toml +++ b/packages/deepctl-cmd-speak/pyproject.toml @@ -21,7 +21,7 @@ classifiers = [ keywords = ["deepgram", "cli", "tts", "text-to-speech", "speak"] requires-python = ">=3.10" dependencies = [ - "deepctl-core>=0.1.10", + "deepctl-core>=0.2.15", "click>=8.0.0", "rich>=13.0.0", "pydantic>=2.0.0", diff --git a/packages/deepctl-cmd-speak/src/deepctl_cmd_speak/command.py b/packages/deepctl-cmd-speak/src/deepctl_cmd_speak/command.py index e45eaa3..705a32b 100644 --- a/packages/deepctl-cmd-speak/src/deepctl_cmd_speak/command.py +++ b/packages/deepctl-cmd-speak/src/deepctl_cmd_speak/command.py @@ -27,6 +27,13 @@ console = Console(stderr=True) +# Flux (Speak v2) streaming controls, per the /v2/speak API. `speed` is a +# 0.05-increment multiplier and `expressivity` is a small integer range; both +# are validated up front so we fail with a clear message instead of surfacing +# a raw SPEED_OUT_OF_RANGE / server error mid-stream. +_FLUX_SPEEDS = (0.85, 0.90, 0.95, 1.00, 1.05, 1.10, 1.15) +_FLUX_EXPRESSIVITY = (-2, -1, 0, 1, 2) + def _fmt_bytes(n: int) -> str: """Human-readable byte count for progress display.""" @@ -170,11 +177,17 @@ class SpeakCommand(BaseCommand): 'dg speak "Hello world" -o hello.wav', "dg speak --file message.txt -o output.wav", 'dg speak "Hello" | ffplay -loglevel error -nodisp -autoexit -', + # Flux TTS streaming controls: --speed (0.85-1.15) and beta + # --expressivity (-2..2, default 0). + 'dg speak "A little slower, please" --speed 0.9 -o slow.wav', + 'dg speak "So exciting!" --expressivity 2 -o lively.wav # beta; default: 0', # Aura (Speak v1, batch REST) — opt in with -m aura-*; needed for # containerized formats like mp3. 'dg speak "Hello" -m aura-2-asteria-en -o hello.mp3', 'dg speak "Hello" -m aura-2-luna-en -o hello.wav --encoding linear16 --container wav', 'echo "Hello" | dg speak -o hello.mp3 -m aura-2-asteria-en', + # Aura-2 Spanish voice (run `dg models` for the full list). + 'dg speak "Hola, mundo" -m aura-2-selena-es -o hola.mp3', ] agent_help = ( "Convert text to speech using Deepgram's TTS API. " @@ -184,7 +197,10 @@ class SpeakCommand(BaseCommand): "v2, streaming over WebSocket and emitting raw audio; linear16 output is " "wrapped in a WAV container so it is directly playable. Pass an aura-* " "model to use Speak v1 (batch REST), which supports containerized " - "formats like mp3. Supports model selection and audio format options." + "formats like mp3. Supports model selection and audio format options. " + "Flux TTS models also accept --speed (0.85–1.15) and beta " + "--expressivity (-2..2; default 0 = nominal) streaming controls; these " + "are rejected for other models." ) def get_arguments(self) -> list[dict[str, Any]]: @@ -237,6 +253,25 @@ def get_arguments(self) -> list[dict[str, Any]]: "type": float, "is_option": True, }, + { + "names": ["--speed"], + "help": ( + "Flux TTS (v2) only. Speech-rate multiplier: 0.85, 0.90, 0.95, " + "1.00, 1.05, 1.10, or 1.15 (1.00 = nominal)." + ), + "type": float, + "is_option": True, + }, + { + "names": ["--expressivity"], + "help": ( + "Flux TTS (v2) only, beta. Expressive range: -2, -1, 0, 1, " + "or 2 (default 0 = nominal; negative flatter, positive more " + "animated). Fixed for the connection." + ), + "type": int, + "is_option": True, + }, { "names": ["--file", "-f"], "help": "Read text from file", @@ -258,6 +293,8 @@ def handle( encoding = kwargs.get("encoding") container = kwargs.get("container") sample_rate = kwargs.get("sample_rate") + speed = kwargs.get("speed") + expressivity = kwargs.get("expressivity") file_path = kwargs.get("file") # Resolve text input: arg > --file > stdin @@ -285,8 +322,29 @@ def handle( message="No output specified. Use -o/--output to save to file, or pipe stdout.", ) - # Flux models stream over the WebSocket (speak.v2); Aura uses REST (speak.v1). - is_flux = model.lower().startswith("flux") + # Only the documented flux-* namespace uses speak.v2. Aura and unknown + # model names pass through to the REST API so the service can resolve them. + is_flux = model.lower().startswith("flux-") + + # speed / expressivity are Flux (Speak v2) connect controls; reject them + # for other models rather than silently dropping them. Raise (not return) + # so the failure exits non-zero in every output mode. + if not is_flux and (speed is not None or expressivity is not None): + raise click.ClickException( + "--speed and --expressivity are only supported for Flux TTS " + "(Speak v2) models (flux-*). They are not available for " + f"Speak v1 model '{model}'." + ) + if speed is not None and speed not in _FLUX_SPEEDS: + allowed = ", ".join(f"{s:.2f}" for s in _FLUX_SPEEDS) + raise click.ClickException( + f"--speed must be one of: {allowed} (got {speed})." + ) + if expressivity is not None and expressivity not in _FLUX_EXPRESSIVITY: + allowed = ", ".join(str(e) for e in _FLUX_EXPRESSIVITY) + raise click.ClickException( + f"--expressivity must be one of: {allowed} (got {expressivity})." + ) if is_flux: # WebSocket streaming path. Streaming output is raw audio, so @@ -313,6 +371,8 @@ def handle( model=model, encoding=eff_encoding, sample_rate=eff_sample_rate, + speed=speed, + expressivity=expressivity, ) ) @@ -395,7 +455,7 @@ def handle( # to the stderr console. return None - # REST path (Aura v1) — unchanged. + # REST path (Speak v1, including Aura and unknown-model pass-through). try: console.print(f"[blue]Generating speech with {model}...[/blue]") @@ -443,6 +503,11 @@ def handle( # summary above went to the stderr console. return None + except click.ClickException: + raise except Exception as e: - console.print(f"[red]Error generating speech:[/red] {e}") - return BaseResult(status="error", message=str(e)) + # Raise (not return an error result) so the failure exits non-zero + # in every output mode — a returned BaseResult is only printed and + # still exits 0. Reachable now that unknown models (e.g. a bare + # "flux" typo) route here instead of the raising v2 path. + raise click.ClickException(f"Error generating speech: {e}") diff --git a/packages/deepctl-cmd-speak/tests/unit/test_speak_command.py b/packages/deepctl-cmd-speak/tests/unit/test_speak_command.py index b84831b..f37952b 100644 --- a/packages/deepctl-cmd-speak/tests/unit/test_speak_command.py +++ b/packages/deepctl-cmd-speak/tests/unit/test_speak_command.py @@ -71,6 +71,12 @@ def test_command_properties(self, command): assert command.name == "speak" assert command.requires_auth is True assert command.ci_friendly is True + assert "beta" in command.agent_help + assert "default 0" in command.agent_help + assert any( + "--expressivity" in example and "beta" in example + for example in command.examples + ) def test_get_arguments(self, command): """Test command arguments configuration.""" @@ -94,6 +100,8 @@ def test_get_arguments(self, command): assert "--encoding" in option_names assert "--container" in option_names assert "--sample-rate" in option_names + assert "--speed" in option_names + assert "--expressivity" in option_names assert "--file" in option_names assert "-f" in option_names @@ -339,6 +347,117 @@ def test_handle_flux_streams_and_wraps_wav( assert data[8:12] == b"WAVE" assert pcm in data + @pytest.mark.parametrize("model", ["flux", "fluxfoo"]) + @patch("deepctl_cmd_speak.command.sys") + def test_handle_non_flux_prefix_uses_v1_pass_through( + self, + mock_sys, + model, + command, + mock_config, + mock_auth_manager, + mock_client, + tmp_path, + ): + """Bare and typo flux names do not enter the flux-* Speak v2 route.""" + mock_sys.stdin.isatty.return_value = True + mock_sys.stdout.isatty.return_value = True + mock_client.speak_text.return_value = iter([b"audio"]) + + command.handle( + config=mock_config, + auth_manager=mock_auth_manager, + client=mock_client, + text="Hello", + output=str(tmp_path / "output.mp3"), + model=model, + encoding=None, + container=None, + sample_rate=None, + file=None, + ) + + mock_client.speak_text.assert_called_once_with( + text="Hello", + model=model, + encoding=None, + container=None, + sample_rate=None, + ) + mock_client.speak_text_stream.assert_not_called() + + @pytest.mark.parametrize("model", ["flux", "fluxfoo", "aura-2-asteria-en"]) + @patch("deepctl_cmd_speak.command.sys") + def test_handle_rest_api_error_exits_nonzero( + self, + mock_sys, + model, + command, + mock_config, + mock_auth_manager, + mock_client, + tmp_path, + ): + """A Speak v1 (REST) API failure raises ClickException so the command + exits non-zero. A returned error result would only print and still exit + 0 — the sink that bare/typo `flux` names (now routed to REST, not the + raising v2 path) would otherwise fall into.""" + mock_sys.stdin.isatty.return_value = True + mock_sys.stdout.isatty.return_value = True + mock_client.speak_text.side_effect = Exception("model not found") + + with pytest.raises(click.ClickException) as exc_info: + command.handle( + config=mock_config, + auth_manager=mock_auth_manager, + client=mock_client, + text="Hello", + output=str(tmp_path / "output.mp3"), + model=model, + encoding=None, + container=None, + sample_rate=None, + file=None, + ) + + assert "model not found" in str(exc_info.value) + + @pytest.mark.parametrize("model", ["flux", "fluxfoo"]) + @patch("deepctl_cmd_speak.command.sys") + def test_handle_non_flux_prefix_control_error_is_model_neutral( + self, + mock_sys, + model, + command, + mock_config, + mock_auth_manager, + mock_client, + tmp_path, + ): + """Unknown models are not mislabeled as Aura in control validation.""" + mock_sys.stdin.isatty.return_value = True + mock_sys.stdout.isatty.return_value = True + + with pytest.raises(click.ClickException) as exc_info: + command.handle( + config=mock_config, + auth_manager=mock_auth_manager, + client=mock_client, + text="Hello", + output=str(tmp_path / "output.wav"), + model=model, + encoding=None, + container=None, + sample_rate=None, + speed=1.0, + file=None, + ) + + assert model in str(exc_info.value) + assert "Aura" not in str(exc_info.value) + mock_client.speak_text.assert_not_called() + mock_client.speak_text_stream.assert_not_called() + @patch("deepctl_cmd_speak.command.sys") def test_handle_flux_rejects_non_raw_encoding( self, @@ -373,6 +492,136 @@ def test_handle_flux_rejects_non_raw_encoding( mock_client.speak_text_stream.assert_not_called() + @patch("deepctl_cmd_speak.command.sys") + def test_handle_flux_forwards_speed_and_expressivity( + self, + mock_sys, + command, + mock_config, + mock_auth_manager, + mock_client, + tmp_path, + ): + """--speed / --expressivity reach speak_text_stream for flux-* models.""" + mock_sys.stdin.isatty.return_value = True + mock_sys.stdout.isatty.return_value = True + + pcm = b"\x01\x00\x02\x00" + mock_client.speak_text_stream.return_value = iter([pcm]) + + command.handle( + config=mock_config, + auth_manager=mock_auth_manager, + client=mock_client, + text="Hello", + output=str(tmp_path / "hello.wav"), + model="flux-alexis-en", + encoding=None, + container=None, + sample_rate=None, + speed=0.9, + expressivity=2, + file=None, + ) + + _, kwargs = mock_client.speak_text_stream.call_args + assert kwargs["speed"] == 0.9 + assert kwargs["expressivity"] == 2 + + @patch("deepctl_cmd_speak.command.sys") + def test_handle_speed_rejected_for_aura( + self, + mock_sys, + command, + mock_config, + mock_auth_manager, + mock_client, + tmp_path, + ): + """speed / expressivity are Flux-only; using them with Aura fails loudly.""" + mock_sys.stdin.isatty.return_value = True + mock_sys.stdout.isatty.return_value = True + + with pytest.raises(click.ClickException, match="only supported for Flux"): + command.handle( + config=mock_config, + auth_manager=mock_auth_manager, + client=mock_client, + text="Hello", + output=str(tmp_path / "x.mp3"), + model="aura-2-asteria-en", + encoding=None, + container=None, + sample_rate=None, + speed=1.0, + file=None, + ) + + mock_client.speak_text_stream.assert_not_called() + mock_client.speak_text.assert_not_called() + + @patch("deepctl_cmd_speak.command.sys") + def test_handle_invalid_speed_rejected( + self, + mock_sys, + command, + mock_config, + mock_auth_manager, + mock_client, + tmp_path, + ): + """An off-grid --speed value fails before opening a stream.""" + mock_sys.stdin.isatty.return_value = True + mock_sys.stdout.isatty.return_value = True + + with pytest.raises(click.ClickException, match="--speed must be one of"): + command.handle( + config=mock_config, + auth_manager=mock_auth_manager, + client=mock_client, + text="Hello", + output=str(tmp_path / "x.wav"), + model="flux-alexis-en", + encoding=None, + container=None, + sample_rate=None, + speed=1.3, + file=None, + ) + + mock_client.speak_text_stream.assert_not_called() + + @patch("deepctl_cmd_speak.command.sys") + def test_handle_invalid_expressivity_rejected( + self, + mock_sys, + command, + mock_config, + mock_auth_manager, + mock_client, + tmp_path, + ): + """An out-of-range --expressivity value fails before opening a stream.""" + mock_sys.stdin.isatty.return_value = True + mock_sys.stdout.isatty.return_value = True + + with pytest.raises(click.ClickException, match="--expressivity must be one of"): + command.handle( + config=mock_config, + auth_manager=mock_auth_manager, + client=mock_client, + text="Hello", + output=str(tmp_path / "x.wav"), + model="flux-alexis-en", + encoding=None, + container=None, + sample_rate=None, + expressivity=5, + file=None, + ) + + mock_client.speak_text_stream.assert_not_called() + @patch("deepctl_cmd_speak.command.sys") def test_handle_flux_empty_audio_fails( self, diff --git a/packages/deepctl-cmd-usage/README.md b/packages/deepctl-cmd-usage/README.md index 3a0659b..82771a6 100644 --- a/packages/deepctl-cmd-usage/README.md +++ b/packages/deepctl-cmd-usage/README.md @@ -35,7 +35,7 @@ pipx run deepctl --help - `click>=8.0.0` - `rich>=13.0.0` -- `deepgram-sdk>=6.0.0rc2` +- `deepgram-sdk>=7.7.0,<8` - `pydantic>=2.0.0` ## License diff --git a/packages/deepctl-cmd-usage/pyproject.toml b/packages/deepctl-cmd-usage/pyproject.toml index b8213c3..95d351a 100644 --- a/packages/deepctl-cmd-usage/pyproject.toml +++ b/packages/deepctl-cmd-usage/pyproject.toml @@ -25,7 +25,7 @@ dependencies = [ "deepctl-shared-utils>=0.1.10", "click>=8.0.0", "rich>=13.0.0", - "deepgram-sdk>=7.5.0", + "deepgram-sdk>=7.7.0,<8", "pydantic>=2.0.0", ] diff --git a/packages/deepctl-core/README.md b/packages/deepctl-core/README.md index 701a74c..2cf36ca 100644 --- a/packages/deepctl-core/README.md +++ b/packages/deepctl-core/README.md @@ -30,7 +30,7 @@ pipx run deepctl --help ## Dependencies - `click>=8.0.0` -- `deepgram-sdk>=6.0.0rc2` +- `deepgram-sdk>=7.7.0,<8` - `pydantic>=2.0.0` - `rich>=13.0.0` - `httpx>=0.24.0` diff --git a/packages/deepctl-core/pyproject.toml b/packages/deepctl-core/pyproject.toml index e484083..cbfb847 100644 --- a/packages/deepctl-core/pyproject.toml +++ b/packages/deepctl-core/pyproject.toml @@ -23,7 +23,7 @@ keywords = ["deepgram", "core", "auth", "config", "client"] requires-python = ">=3.10" dependencies = [ "click>=8.0.0", - "deepgram-sdk>=7.5.0", + "deepgram-sdk>=7.7.0,<8", "pydantic>=2.0.0", "rich>=13.0.0", "httpx>=0.24.0", diff --git a/packages/deepctl-core/src/deepctl_core/client.py b/packages/deepctl-core/src/deepctl_core/client.py index 4239c56..8c842e0 100644 --- a/packages/deepctl-core/src/deepctl_core/client.py +++ b/packages/deepctl-core/src/deepctl_core/client.py @@ -189,12 +189,17 @@ def speak_text_stream( model: str, encoding: str | None = None, sample_rate: float | None = None, + speed: float | None = None, + expressivity: int | None = None, ) -> Iterator[bytes]: """Stream TTS audio over the Flux v2 WebSocket (speak.v2.connect). Yields raw audio chunks as they arrive. The streaming transport emits raw (non-containerized) audio, so only linear16/mulaw/alaw encodings apply and sample_rate is sent as the string the streaming API expects. + + ``speed`` (0.85-1.15) and beta ``expressivity`` (-2..2, default 0) are + Flux connect query parameters; they are only sent when set. """ from deepgram.speak.v2.types.speak_v2speak import SpeakV2Speak @@ -203,6 +208,10 @@ def speak_text_stream( connect_kwargs["encoding"] = encoding if sample_rate: connect_kwargs["sample_rate"] = str(int(sample_rate)) + if speed is not None: + connect_kwargs["speed"] = speed + if expressivity is not None: + connect_kwargs["expressivity"] = expressivity try: with self.client.speak.v2.connect(**connect_kwargs) as conn: diff --git a/packages/deepctl-core/src/deepctl_core/skill_generator.py b/packages/deepctl-core/src/deepctl_core/skill_generator.py index 842ca28..f6a9e10 100644 --- a/packages/deepctl-core/src/deepctl_core/skill_generator.py +++ b/packages/deepctl-core/src/deepctl_core/skill_generator.py @@ -417,7 +417,8 @@ def render_developer_guide( lines.append("## Text-to-Speech (TTS)") lines.append("") lines.append( - "Generate natural-sounding speech from text using Deepgram's Aura voices." + "Generate natural-sounding speech from text using Deepgram's Aura and " + "Flux voices." ) lines.append("") lines.append("### Models") @@ -463,6 +464,10 @@ def render_developer_guide( lines.append("") lines.append("### Flux TTS — Speak v2, WebSocket streaming (Python)") lines.append("") + lines.append( + "`expressivity` is beta and defaults to `0` (nominal delivery) when omitted." + ) + lines.append("") lines.append("```python") lines.append("from deepgram import DeepgramClient") lines.append("from deepgram.speak.v2.types.speak_v2speak import SpeakV2Speak") @@ -470,7 +475,11 @@ def render_developer_guide( lines.append('client = DeepgramClient(api_key="DEEPGRAM_API_KEY")') lines.append("") lines.append("with client.speak.v2.connect(") - lines.append(' model="flux-alexis-en", encoding="linear16", sample_rate="24000"') + lines.append(' model="flux-alexis-en",') + lines.append(' encoding="linear16",') + lines.append(' sample_rate="24000",') + lines.append(" speed=1.0, # 0.85–1.15 in 0.05 steps (optional)") + lines.append(" expressivity=0, # beta; -2..2, default 0 = nominal (optional)") lines.append(") as conn:") lines.append( ' conn.send_speak(SpeakV2Speak(type="Speak", text="Hello from Flux!"))' diff --git a/packages/deepctl-core/tests/unit/test_client.py b/packages/deepctl-core/tests/unit/test_client.py index 4cab216..c8fd4af 100644 --- a/packages/deepctl-core/tests/unit/test_client.py +++ b/packages/deepctl-core/tests/unit/test_client.py @@ -343,6 +343,52 @@ def test_speak_text(self, mock_dg_client, client): text="Hello world", model="aura-2-asteria-en" ) + def test_speak_text_stream_forwards_flux_controls(self, client): + """Speed and expressivity reach the SDK's Speak v2 connect call.""" + mock_sdk_client = MagicMock() + mock_connection = MagicMock() + mock_connection.__iter__.return_value = iter([b"audio"]) + mock_sdk_client.speak.v2.connect.return_value.__enter__.return_value = ( + mock_connection + ) + client._client = mock_sdk_client + + result = list( + client.speak_text_stream( + "Hello world", + model="flux-alexis-en", + encoding="linear16", + sample_rate=24000, + speed=0.9, + expressivity=2, + ) + ) + + assert result == [b"audio"] + mock_sdk_client.speak.v2.connect.assert_called_once_with( + model="flux-alexis-en", + encoding="linear16", + sample_rate="24000", + speed=0.9, + expressivity=2, + ) + + def test_speak_text_stream_omits_unset_flux_controls(self, client): + """Unset controls are omitted rather than sent as null query values.""" + mock_sdk_client = MagicMock() + mock_connection = MagicMock() + mock_connection.__iter__.return_value = iter([]) + mock_sdk_client.speak.v2.connect.return_value.__enter__.return_value = ( + mock_connection + ) + client._client = mock_sdk_client + + assert ( + list(client.speak_text_stream("Hello world", model="flux-alexis-en")) == [] + ) + + mock_sdk_client.speak.v2.connect.assert_called_once_with(model="flux-alexis-en") + @patch("deepctl_core.client.DGClient") def test_analyze_text(self, mock_dg_client, client): """Test analyzing text.""" diff --git a/packages/deepctl-core/tests/unit/test_skill_generator.py b/packages/deepctl-core/tests/unit/test_skill_generator.py index 90f7f95..9074564 100644 --- a/packages/deepctl-core/tests/unit/test_skill_generator.py +++ b/packages/deepctl-core/tests/unit/test_skill_generator.py @@ -131,7 +131,10 @@ def test_contains_tts_content(self): content = render_developer_guide("1.0.0") assert "Text-to-Speech" in content assert "Aura-2" in content + assert "Aura and Flux voices" in content assert "aura-2-andromeda-en" in content + assert "`expressivity` is beta" in content + assert "defaults to `0`" in content def test_contains_audio_intelligence(self): content = render_developer_guide("1.0.0") diff --git a/pyproject.toml b/pyproject.toml index 2a0e7c3..01bca0c 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -34,8 +34,8 @@ keywords = [ requires-python = ">=3.10" dependencies = [ "click>=8.0.0", - "deepgram-sdk>=7.5.0", - "deepctl-core>=0.1.10", + "deepgram-sdk>=7.7.0,<8", + "deepctl-core>=0.2.15", "deepctl-cmd-login>=0.1.10", "deepctl-cmd-projects>=0.1.10", "deepctl-cmd-transcribe>=0.1.10", @@ -54,10 +54,10 @@ dependencies = [ "deepctl-cmd-skills>=0.0.1", "deepctl-cmd-init>=0.0.1", "deepctl-cmd-models>=0.0.1", - "deepctl-cmd-speak>=0.0.1", + "deepctl-cmd-speak>=0.0.4", "deepctl-cmd-keys>=0.0.1", "deepctl-cmd-read>=0.0.1", - "deepctl-cmd-listen>=0.0.1", + "deepctl-cmd-listen>=0.0.14", "deepctl-cmd-requests>=0.0.1", "deepctl-cmd-billing>=0.0.1", "deepctl-cmd-members>=0.0.1", @@ -82,6 +82,7 @@ dev = [ "pytest-asyncio>=0.21.0", "pytest-cov>=4.0.0", "pytest-mock>=3.10.0", + "pytest-timeout>=2.3.1,<3", "responses>=0.23.0", # Code Quality "ruff>=0.8.0", @@ -232,6 +233,7 @@ testing = [ "pytest-asyncio>=0.21.0", "pytest-cov>=4.0.0", "pytest-mock>=3.10.0", + "pytest-timeout>=2.3.1,<3", "responses>=0.23.0", "deepctl-plugin-example", ] diff --git a/tests/e2e/__init__.py b/tests/e2e/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/e2e/conftest.py b/tests/e2e/conftest.py new file mode 100644 index 0000000..c91c1b6 --- /dev/null +++ b/tests/e2e/conftest.py @@ -0,0 +1,107 @@ +"""Fixtures and opt-in gate for the live end-to-end suite. + +These tests drive real command ``handle()`` methods against a live Deepgram API +(in-process, not via subprocess), so they require credentials and network +access. They run only when ``DEEPGRAM_API_KEY`` and ``RUN_LIVE_E2E=1`` are set. +The target must also be explicit: set ``DEEPGRAM_BASE_URL`` for staging or a +custom endpoint, or set ``RUN_LIVE_E2E_PRODUCTION=1`` to confirm use of the +default production endpoint. A normally exported API key alone is never enough +to enable this suite. +""" + +from __future__ import annotations + +import io +import os +import types +from typing import TYPE_CHECKING +from urllib.parse import urlparse + +import pytest + +if TYPE_CHECKING: + from collections.abc import Mapping + +# Capture credentials at import time — the root autouse ``_clean_deepgram_env`` +# fixture strips every ``DEEPGRAM_*`` var before each test runs, so reading them +# inside a test/fixture would always come back empty. We re-inject the captured +# values per test in ``live_client`` below. +LIVE_API_KEY = os.environ.get("DEEPGRAM_API_KEY") +LIVE_BASE_URL = os.environ.get("DEEPGRAM_BASE_URL") + + +def _is_production_target(base_url: str | None) -> bool: + """Return whether the configured target resolves to Deepgram production.""" + if not base_url: + return True + candidate = base_url if "://" in base_url else f"https://{base_url}" + return (urlparse(candidate).hostname or "").lower() == "api.deepgram.com" + + +def _live_e2e_skip_reason(environ: Mapping[str, str]) -> str | None: + """Return why live e2e is disabled, without exposing environment values.""" + if not environ.get("DEEPGRAM_API_KEY"): + return "DEEPGRAM_API_KEY is not set; live e2e tests require credentials" + if environ.get("RUN_LIVE_E2E") != "1": + return "RUN_LIVE_E2E must be set to 1; live e2e tests are disabled" + if ( + _is_production_target(environ.get("DEEPGRAM_BASE_URL")) + and environ.get("RUN_LIVE_E2E_PRODUCTION") != "1" + ): + return ( + "set DEEPGRAM_BASE_URL for a staging/custom target or set " + "RUN_LIVE_E2E_PRODUCTION=1 to confirm production" + ) + return None + + +LIVE_E2E_SKIP_REASON = _live_e2e_skip_reason(os.environ) + +# Applied at module level by each e2e test module. +requires_live_e2e = pytest.mark.skipif( + LIVE_E2E_SKIP_REASON is not None, + reason=LIVE_E2E_SKIP_REASON or "live e2e gate satisfied", +) + + +@pytest.fixture +def live_client(monkeypatch): + """A real (Config, AuthManager, DeepgramClient) wired to the live API. + + Re-injects the credentials the root autouse fixture stripped, then builds + the same object graph the CLI framework constructs at runtime. + """ + if LIVE_E2E_SKIP_REASON: + pytest.skip(LIVE_E2E_SKIP_REASON) + + monkeypatch.setenv("DEEPGRAM_API_KEY", LIVE_API_KEY or "") + if LIVE_BASE_URL: + monkeypatch.setenv("DEEPGRAM_BASE_URL", LIVE_BASE_URL) + + from deepctl_core import AuthManager, Config, DeepgramClient + + config = Config() + auth = AuthManager(config) + client = DeepgramClient(config, auth) + return config, auth, client + + +@pytest.fixture +def feed_stdin(monkeypatch): + """Return a helper that pipes raw bytes into the listen command's stdin. + + The stdin streaming path reads ``sys.stdin.buffer`` (via the listen + module's ``sys``); swap in a BytesIO-backed fake so an in-process run + behaves like ``… | dg listen -``. + """ + + def _feed(pcm: bytes) -> None: + fake_stdin = types.SimpleNamespace( + buffer=io.BytesIO(pcm), + isatty=lambda: False, + ) + monkeypatch.setattr( + "deepctl_cmd_listen.command.sys.stdin", fake_stdin, raising=False + ) + + return _feed diff --git a/tests/e2e/test_flux_live.py b/tests/e2e/test_flux_live.py new file mode 100644 index 0000000..590a757 --- /dev/null +++ b/tests/e2e/test_flux_live.py @@ -0,0 +1,196 @@ +"""Live end-to-end tests for Flux TTS / STT and Aura multilingual. + +Each test calls a command's ``handle()`` in-process against the real Deepgram +API, exercising the full transport (SDK WebSocket for Flux TTS, raw WebSocket +for Flux/nova STT, REST for Aura) plus the CLI's own parsing and assembly. + +Live execution requires ``DEEPGRAM_API_KEY``, ``RUN_LIVE_E2E=1``, and an +explicit target. Set ``DEEPGRAM_BASE_URL`` for staging/custom testing. To use +the default production endpoint, set ``RUN_LIVE_E2E_PRODUCTION=1`` as a second +confirmation. See conftest for the complete gate. ASR wording is +non-deterministic, so assertions stay loose: transcripts must be non-empty and +show the specific transformation under test (digits for numerals, ``*`` for +number redaction). Each live case has a two-minute deadline so a stalled +transport cannot occupy the runner indefinitely. +""" + +from __future__ import annotations + +import pytest +from deepctl_cmd_listen.command import ListenCommand +from deepctl_cmd_speak.command import SpeakCommand +from deepctl_cmd_speak.models import SpeakResult + +from .conftest import requires_live_e2e + +pytestmark = [ + pytest.mark.integration, + pytest.mark.requires_auth, + pytest.mark.requires_network, + pytest.mark.slow, + pytest.mark.timeout(120), + requires_live_e2e, +] + +# Flux TTS emits 24 kHz linear16; feed STT the same rate so no resampling is +# needed and the tests carry no ffmpeg dependency. +SAMPLE_RATE = 24000 +NUMBERS_PHRASE = "My account number is four five six seven." + + +def _synth_pcm(client, text: str) -> bytes: + """Synthesize raw linear16 PCM via Flux TTS (used as STT input).""" + pcm = bytearray() + for chunk in client.speak_text_stream( + text=text, + model="flux-alexis-en", + encoding="linear16", + sample_rate=float(SAMPLE_RATE), + ): + pcm.extend(chunk) + return bytes(pcm) + + +# ── Speak (Flux TTS + Aura) ──────────────────────────────────────────────── + + +def test_speak_flux_speed_and_expressivity(live_client, tmp_path): + """dg speak with Flux + --speed/--expressivity writes a valid WAV.""" + config, auth, client = live_client + out = tmp_path / "flux.wav" + + result = SpeakCommand().handle( + config=config, + auth_manager=auth, + client=client, + text="Testing Flux speed and expressivity end to end.", + model="flux-alexis-en", + speed=0.9, + expressivity=2, + output=str(out), + ) + + assert isinstance(result, SpeakResult) + assert result.status == "success" + assert result.bytes_written > 1000 + data = out.read_bytes() + assert data[:4] == b"RIFF" + assert data[8:12] == b"WAVE" + + +def test_speak_aura_spanish_voice(live_client, tmp_path): + """Aura-2 Spanish voice (7.6.0) round-trips over the REST path to MP3.""" + config, auth, client = live_client + out = tmp_path / "hola.mp3" + + result = SpeakCommand().handle( + config=config, + auth_manager=auth, + client=client, + text="Hola, bienvenido a Deepgram.", + model="aura-2-selena-es", + output=str(out), + ) + + assert result.status == "success" + data = out.read_bytes() + assert len(data) > 1000 + # MP3: ID3 tag or an MPEG audio frame sync (0xFF Ex/Fx). + assert data[:3] == b"ID3" or (data[0] == 0xFF and data[1] & 0xE0 == 0xE0) + + +# ── Listen (Flux STT v2 + nova v1) ───────────────────────────────────────── + + +def test_listen_flux_numerals(live_client, feed_stdin): + """Flux STT (v2) streaming with --numerals spells numbers as digits.""" + config, auth, client = live_client + feed_stdin(_synth_pcm(client, NUMBERS_PHRASE)) + + result = ListenCommand().handle( + config=config, + auth_manager=auth, + client=client, + source="-", + model="flux-general-en", + encoding="linear16", + sample_rate=SAMPLE_RATE, + numerals=True, + ) + + assert result.status == "success" + assert result.transcript.strip() + assert any(ch.isdigit() for ch in result.transcript), result.transcript + + +def test_listen_flux_numerals_and_redact(live_client, feed_stdin): + """--redact numbers replaces the digits (Deepgram uses ``*``).""" + config, auth, client = live_client + feed_stdin(_synth_pcm(client, NUMBERS_PHRASE)) + + result = ListenCommand().handle( + config=config, + auth_manager=auth, + client=client, + source="-", + model="flux-general-en", + encoding="linear16", + sample_rate=SAMPLE_RATE, + numerals=True, + redact="numbers", + ) + + assert result.status == "success" + assert result.transcript.strip() + assert "*" in result.transcript, result.transcript + + +def test_listen_nova3_v1_baseline(live_client, feed_stdin): + """nova-3 (v1) streaming still works — guards against a v2-fix regression.""" + config, auth, client = live_client + feed_stdin(_synth_pcm(client, NUMBERS_PHRASE)) + + result = ListenCommand().handle( + config=config, + auth_manager=auth, + client=client, + source="-", + model="nova-3", + encoding="linear16", + sample_rate=SAMPLE_RATE, + ) + + assert result.status == "success" + assert result.transcript.strip() + + +def test_listen_flux_srt_captions_have_real_timestamps(live_client, feed_stdin, capsys): + """Flux STT (v2) --srt emits well-formed cues with non-zero timestamps. + + Flux ``TurnInfo`` words carry no per-word timings, so captions must key off + the turn's ``audio_window_*``; a regression would print ``00:00:00,000`` for + every cue (or crash the end-of-stream save with ``KeyError: 'start'``). + """ + import re + + config, auth, client = live_client + feed_stdin(_synth_pcm(client, NUMBERS_PHRASE)) + + result = ListenCommand().handle( + config=config, + auth_manager=auth, + client=client, + source="-", + model="flux-general-en", + encoding="linear16", + sample_rate=SAMPLE_RATE, + srt=True, + ) + + assert result.status == "success" + out = capsys.readouterr().out + stamps = re.findall(r"\d\d:\d\d:\d\d,\d\d\d", out) + assert " --> " in out, out + assert stamps, out + # The audio window is real, so at least one boundary must be non-zero. + assert any(s != "00:00:00,000" for s in stamps), out diff --git a/tests/e2e/test_live_gate.py b/tests/e2e/test_live_gate.py new file mode 100644 index 0000000..e915d76 --- /dev/null +++ b/tests/e2e/test_live_gate.py @@ -0,0 +1,89 @@ +"""Deterministic tests for the live e2e opt-in gate.""" + +from __future__ import annotations + +import pytest + +from .conftest import _live_e2e_skip_reason + + +@pytest.mark.parametrize( + ("environ", "expected_reason"), + [ + ({}, "DEEPGRAM_API_KEY is not set"), + ( + {"DEEPGRAM_API_KEY": "test-key"}, + "RUN_LIVE_E2E must be set to 1", + ), + ( + { + "DEEPGRAM_API_KEY": "test-key", + "RUN_LIVE_E2E": "true", + "DEEPGRAM_BASE_URL": "https://staging.example", + }, + "RUN_LIVE_E2E must be set to 1", + ), + ( + { + "DEEPGRAM_API_KEY": "test-key", + "RUN_LIVE_E2E": "1", + }, + "RUN_LIVE_E2E_PRODUCTION=1 to confirm production", + ), + ( + { + "DEEPGRAM_API_KEY": "test-key", + "RUN_LIVE_E2E": "1", + "RUN_LIVE_E2E_PRODUCTION": "true", + }, + "RUN_LIVE_E2E_PRODUCTION=1 to confirm production", + ), + ( + { + "DEEPGRAM_API_KEY": "test-key", + "RUN_LIVE_E2E": "1", + "DEEPGRAM_BASE_URL": "https://api.deepgram.com", + }, + "RUN_LIVE_E2E_PRODUCTION=1 to confirm production", + ), + ( + { + "DEEPGRAM_API_KEY": "test-key", + "RUN_LIVE_E2E": "1", + "DEEPGRAM_BASE_URL": "HTTPS://API.DEEPGRAM.COM/", + }, + "RUN_LIVE_E2E_PRODUCTION=1 to confirm production", + ), + ], +) +def test_live_e2e_gate_rejects_incomplete_opt_in(environ, expected_reason): + reason = _live_e2e_skip_reason(environ) + + assert reason is not None + assert expected_reason in reason + assert "test-key" not in reason + + +@pytest.mark.parametrize( + "environ", + [ + { + "DEEPGRAM_API_KEY": "test-key", + "RUN_LIVE_E2E": "1", + "DEEPGRAM_BASE_URL": "https://staging.example", + }, + { + "DEEPGRAM_API_KEY": "test-key", + "RUN_LIVE_E2E": "1", + "RUN_LIVE_E2E_PRODUCTION": "1", + }, + { + "DEEPGRAM_API_KEY": "test-key", + "RUN_LIVE_E2E": "1", + "DEEPGRAM_BASE_URL": "https://api.deepgram.com/", + "RUN_LIVE_E2E_PRODUCTION": "1", + }, + ], +) +def test_live_e2e_gate_accepts_explicit_target(environ): + assert _live_e2e_skip_reason(environ) is None