From 6df7cbfb7d74d6a0f74119c1430e8b90132900fc Mon Sep 17 00:00:00 2001 From: Carter Campbell Date: Wed, 9 Sep 2026 15:10:01 -0600 Subject: [PATCH] feat(sync): push-style live sessions via open_live() transcribe_live() takes an iterator, but most live audio arrives in a callback: sounddevice, PyAudio, WebRTC tracks, telephony media streams, websocket frame handlers. Every such caller had to build the same bridge (a queue, a stop flag, a drain generator and, on the sync side, a thread to run the blocking call in) before getting a transcript. SyncTranscriber.open_live() returns a LiveSession and AsyncSyncTranscriber.open_live() an AsyncLiveSession. The request starts the moment the session opens, on the transcriber's worker pool or as a loop task; write() queues audio without blocking (thread-safe on the sync side), close() ends it, result() waits for the transcript, and abort() drops the request. Context-manager exit closes on success and aborts if the block raised. Chunk types are validated at write() so a str shows up immediately rather than in result(). Both sessions are thin layers over transcribe_live(): no new transport code, same errors, same 120 s cap and silence-abort semantics. The iterator form stays for sources that already are iterators or file objects. README leads the live examples with the session now (sounddevice, and a websocket handler for asyncio), and keeps transcribe_live() for iterator and file-object sources. 19 tests. Co-Authored-By: Claude Fable 5.1 --- CLAUDE.md | 17 +- README.md | 111 ++++----- assemblyai/sync/v1/__init__.py | 6 +- assemblyai/sync/v1/_base.py | 9 + assemblyai/sync/v1/async_client.py | 186 ++++++++++++++- assemblyai/sync/v1/client.py | 194 +++++++++++++++- tests/unit/test_sync_live.py | 357 +++++++++++++++++++++++++++++ 7 files changed, 820 insertions(+), 60 deletions(-) diff --git a/CLAUDE.md b/CLAUDE.md index 7550613c..a2bbc688 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -44,8 +44,8 @@ aai.settings.api_key = "your-key" - `assemblyai.prerecorded.v2` — Canonical module for both transcribers, matching the `/v2/transcript` API. The top-level `aai.*` names re-export it - `aai.TranscriptionConfig` — All transcription options: `speech_models`, `speaker_labels`, `sentiment_analysis`, `entity_detection`, `auto_chapters`, `content_safety`, `language_detection`, `summarization`, `word_boost`, `disfluencies` - `aai.Transcript` — Result object with `.text`, `.status`, `.utterances`, `.words`, `.chapters`, `.entities`, `.sentiment_analysis`. Methods: `get_sentences()`, `get_paragraphs()`, `export_subtitles_srt()`, `export_subtitles_vtt()` -- `aai.SyncTranscriber` — Synchronous pre-recorded transcription: audio in, transcript out, one request (no polling). Methods: `transcribe()`, `transcribe_live()`, `transcribe_async()`, `warm()` -- `aai.AsyncSyncTranscriber` — Asyncio counterpart of `SyncTranscriber`. Same input types, config, result, and errors; `transcribe()`, `transcribe_live()` and `warm()` are coroutines. Owns an HTTP pool: use `async with` or `await aclose()`, or pass an `aai.AsyncClient` to share one +- `aai.SyncTranscriber` — Synchronous pre-recorded transcription: audio in, transcript out, one request (no polling). Methods: `transcribe()`, `transcribe_live()`, `open_live()`, `transcribe_async()`, `warm()` +- `aai.AsyncSyncTranscriber` — Asyncio counterpart of `SyncTranscriber`. Same input types, config, result, and errors; `transcribe()`, `transcribe_live()` and `warm()` are coroutines; `open_live()` returns an `AsyncLiveSession`. Owns an HTTP pool: use `async with` or `await aclose()`, or pass an `aai.AsyncClient` to share one - `aai.SyncTranscriptionConfig` — Sync options: `model` (default `universal-3-5-pro`), `prompt`, `keyterms_prompt`, `conversation_context`, `language_codes`, `timestamps`, `sample_rate`, `channels` - `aai.SyncTranscriptResponse` — Sync result: `.text`, `.words` (`SyncWord` with `confidence` always, `start`/`end` only when `timestamps=True`), `.confidence`, `.audio_duration_ms`, `.session_id`, `.request_time_ms` - `assemblyai.streaming.v3.RealTimeTranscriber` — Real-time streaming with event-based API (threaded) @@ -213,7 +213,7 @@ result = aai.SyncTranscriber().transcribe(raw_pcm_bytes, config=config) not asyncio) for fanning out a handful of files. In asyncio code use `aai.AsyncSyncTranscriber` instead (see "Asyncio sync transcription" below). -**Live upload** (`transcribe_live()`): uploads audio as it is produced instead of +**Live upload** (`open_live()` / `transcribe_live()`): uploads audio as it is produced instead of waiting for the whole clip. The request starts immediately, so authorization, the upload and every speech segment but the last resolve while the caller is still recording; what is left to wait for once they stop is the final segment. Takes an iterable of chunks or a @@ -225,6 +225,17 @@ def mic_chunks(): result = aai.SyncTranscriber().transcribe_live(mic_chunks()) ``` +For callback-driven sources (most microphone libraries, WebRTC, telephony) use the +push-style session instead; `write()` is thread-safe and never blocks: +```python +with transcriber.open_live(config) as session: + start_capture(on_audio=session.write) # e.g. sounddevice callback + wait_until_done() +result = session.result() # raises SyncTranscriptError on failure +``` +Leaving the block calls `close()`; an exception in the block calls `abort()` instead. The +asyncio twin is `AsyncSyncTranscriber.open_live()` → `AsyncLiveSession` (`write()`/`close()` +plain functions, `result()`/`abort()` coroutines; call `write()` on the loop thread). It pays off only when the audio is genuinely still being produced. Streaming a file already on disk is *slower* than `transcribe()`, which sends it in one piece — the saving comes from overlapping the recording, not from chunking. It also needs enough audio to diff --git a/README.md b/README.md index 315349c2..12c5efaa 100644 --- a/README.md +++ b/README.md @@ -102,7 +102,7 @@ aai.settings.api_key = f"{ASSEMBLYAI_API_KEY}" | --- | --- | | `assemblyai.prerecorded.v2.Transcriber` | Long-form audio, URLs, and the audio-intelligence features (speaker labels, chapters, sentiment, …), over the polled job API | | `assemblyai.prerecorded.v2.AsyncTranscriber` | The same, from asyncio code | -| `assemblyai.sync.v1.SyncTranscriber` | Short clips (≤120s, ≤40MB) where you want the transcript back in one request, at the lowest latency — from a file you already have (`transcribe()`) or uploaded while it is still being recorded (`transcribe_live()`) | +| `assemblyai.sync.v1.SyncTranscriber` | Short clips (≤120s, ≤40MB) where you want the transcript back in one request, at the lowest latency — from a file you already have (`transcribe()`) or uploaded while it is still being recorded (`open_live()`, `transcribe_live()`) | | `assemblyai.sync.v1.AsyncSyncTranscriber` | The same, from asyncio code | | `assemblyai.streaming.v3.RealTimeTranscriber` | Live audio (microphone, telephony, voice agents), transcribed as it arrives over a websocket session | | `assemblyai.streaming.v3.AsyncRealTimeTranscriber` | The same, from asyncio code | @@ -519,99 +519,108 @@ with aai.SyncTranscriber() as transcriber:
- Transcribe live audio as it is recorded (`transcribe_live()`) + Transcribe live audio as it is recorded (`open_live()`) -`transcribe()` needs the whole clip before it can send anything. `transcribe_live()` starts the request immediately and uploads audio as your code produces it, so the upload and all but the last speech segment are done by the time the speaker stops. What remains is one final segment: on a paced 7 s clip that cut the wait after the last byte roughly in half. +`transcribe()` needs the whole clip before it can send anything. A live session starts the request immediately and uploads audio as you produce it, so the upload and all but the last speech segment are done by the time the speaker stops. What remains is one final segment: on a paced 7 s clip that cut the wait after the last byte roughly in half. -It takes an iterable of `bytes` chunks or a binary file object read as it fills, and returns the same `SyncTranscriptResponse` as `transcribe()`. Pass raw S16LE PCM with `sample_rate` and `channels` set; that is what microphones and telephony hand you, and it needs no container header. +Most audio sources hand you chunks in a callback. `open_live()` is built for that: `write()` from the callback, `close()` when the speaker stops, and `result()` for the transcript, which is the same `SyncTranscriptResponse` that `transcribe()` returns. The request runs on one of the transcriber's worker threads, so `write()` never blocks and is safe from any thread. ```python -import queue -import threading - import assemblyai as aai import sounddevice as sd # pip install sounddevice aai.settings.api_key = "" RATE = 16000 -config = aai.SyncTranscriptionConfig(sample_rate=RATE, channels=1) +config = aai.SyncTranscriptionConfig(sample_rate=RATE, channels=1) # raw PCM needs both +with aai.SyncTranscriber() as transcriber: + with transcriber.open_live(config) as session: + microphone = sd.RawInputStream( + samplerate=RATE, channels=1, dtype="int16", + callback=lambda data, *_: session.write(bytes(data)), + ) + with microphone: + input("Recording, press Enter to stop... ") + # leaving the block ends the audio; the final segment is all that is left + print(session.result().text) +``` -def microphone(stop: threading.Event): - """Yields PCM chunks from the default microphone until `stop` is set.""" - chunks: queue.Queue[bytes] = queue.Queue() - with sd.RawInputStream( - samplerate=RATE, channels=1, dtype="int16", - callback=lambda data, *_: chunks.put(bytes(data)), - ): - while not stop.is_set(): - try: - yield chunks.get(timeout=0.1) - except queue.Empty: - pass +Leaving the `with` block calls `close()`. If the block raises, the session is aborted instead and `result()` raises. Call `abort()` yourself to drop a recording you no longer want transcribed. +This is **not** the streaming API. A live session returns one finished transcript when the audio ends; use `aai.RealTimeTranscriber` when you need words back while the speaker is still talking. -stop = threading.Event() -threading.Thread(target=lambda: (input("Recording, press Enter to stop... "), stop.set()), daemon=True).start() +Use it only when the audio is genuinely still being produced. Streaming a file that is already on disk is slower than `transcribe()`, which sends it in one piece. Short clips gain less: below about a minute the saving is the elided upload only. The same 120 s cap applies, and the server aborts an upload that goes silent for long, so keep writing until you are done rather than pausing. -result = aai.SyncTranscriber().transcribe_live(microphone(stop), config=config) -print(result.text) -``` +Errors that would normally arrive at the end (bad key, rate limit, capacity) can surface part-way through the upload; `result()` raises them as `SyncTranscriptError`. A bad key is reported at the first segment boundary, roughly 30 s in. `warm()` opens the connection early but does not validate the key. -Any producer works, not only a microphone: a file object on a pipe or socket, chunks arriving over a websocket, a telephony media stream. End the iterator to finish. The server aborts an upload that goes silent for long, so keep sending until you are done rather than pausing. +
-This is **not** the streaming API. `transcribe_live()` returns one finished transcript when the audio ends; use `aai.RealTimeTranscriber` when you need words back while the speaker is still talking. +
+ Transcribe live audio from an iterator or file object (`transcribe_live()`) -Use it only when the audio is genuinely still being produced. Streaming a file that is already on disk is slower than `transcribe()`, which sends it in one piece. Short clips gain less: below about a minute the saving is the elided upload only. The same 120 s cap applies. +When the audio already comes as something you can iterate, or as a file object that fills over time, hand it to `transcribe_live()` directly. It takes an iterable of `bytes` chunks or a binary file object, blocks until the audio ends, and returns the transcript. `open_live()` is this method with a queue in front of it. ```python import subprocess +import assemblyai as aai -# Audio you already hold whole: use transcribe(), which is faster for it. -result = aai.SyncTranscriber().transcribe("./call.wav") +aai.settings.api_key = "" +config = aai.SyncTranscriptionConfig(sample_rate=16000, channels=1) # A file object that fills over time, e.g. the stdout of a recorder (sox): -proc = subprocess.Popen(["rec", "-q", "-t", "raw", "-r", "16000", "-c", "1", "-b", "16", "-e", "signed", "-"], stdout=subprocess.PIPE) -result = aai.SyncTranscriber().transcribe_live(proc.stdout, config=config) +recorder = subprocess.Popen( + ["rec", "-q", "-t", "raw", "-r", "16000", "-c", "1", "-b", "16", "-e", "signed", "-"], + stdout=subprocess.PIPE, +) +result = aai.SyncTranscriber().transcribe_live(recorder.stdout, config=config) + +# Or any generator of chunks: +def from_call(call): + while call.active: + yield call.read_frame() + +result = aai.SyncTranscriber().transcribe_live(from_call(call), config=config) + +# Audio you already hold whole belongs in transcribe(), which is faster for it: +result = aai.SyncTranscriber().transcribe("./call.wav") ``` -Errors that would normally arrive at the end (bad key, rate limit, capacity) can surface part-way through the upload as a `SyncTranscriptError`. A bad key is reported at the first segment boundary, roughly 30 s in. `warm()` opens the connection early but does not validate the key. +End the iterator to finish. Anything the producer raises propagates unchanged and drops the upload.
Transcribe live audio from asyncio -`AsyncSyncTranscriber.transcribe_live()` takes an async iterable of chunks as well as a file object or plain iterable. A file object is read in a worker thread, so a blocking `read()` is fine; a plain iterable is consumed inline and must not block the loop. +`AsyncSyncTranscriber.open_live()` returns an `AsyncLiveSession`: `write()` and `close()` are plain functions so a callback can call them, and `result()` and `abort()` are coroutines. The request runs as a task on the event loop. `write()` must be called on the loop's thread; from an audio library's capture thread, schedule it with `loop.call_soon_threadsafe(session.write, chunk)`. ```python import asyncio import assemblyai as aai aai.settings.api_key = "" - config = aai.SyncTranscriptionConfig(sample_rate=16000, channels=1) -async def microphone(frames: asyncio.Queue): - """Yields PCM chunks pushed onto `frames` by a capture callback; None ends it.""" - while (chunk := await frames.get()) is not None: - yield chunk - +async def handle(websocket): + """Transcribes the PCM frames a browser sends over one websocket.""" + async with aai.AsyncSyncTranscriber() as transcriber: + async with transcriber.open_live(config) as session: + async for frame in websocket: + session.write(frame) + result = await session.result() + await websocket.send(result.text) +``` -async def main(): - frames: asyncio.Queue = asyncio.Queue() - start_capture(lambda pcm: frames.put_nowait(pcm)) # your audio callback +`AsyncSyncTranscriber.transcribe_live()` is the pull-style twin: it takes an async iterable of chunks as well as a file object or plain iterable. A file object is read in a worker thread, so a blocking `read()` is fine; a plain iterable is consumed inline and must not block the loop. - async with aai.AsyncSyncTranscriber() as transcriber: - task = asyncio.create_task(transcriber.transcribe_live(microphone(frames), config=config)) - await asyncio.sleep(5) # ... or until the user stops speaking - frames.put_nowait(None) # end the iterator; the upload finishes - result = await task - print(result.text) +```python +async def frames(websocket): + async for frame in websocket: + yield frame -asyncio.run(main()) +result = await transcriber.transcribe_live(frames(websocket), config=config) ```
@@ -619,7 +628,7 @@ asyncio.run(main())
Use it from asyncio (`AsyncSyncTranscriber`) -`aai.AsyncSyncTranscriber` is the asyncio counterpart of `aai.SyncTranscriber` — same input types, config, result, and errors, with `transcribe()`, `transcribe_live()` and `warm()` as coroutines. Use it in asyncio code (FastAPI, aiohttp, voice agents), where the threaded `transcribe()` would block the event loop and `transcribe_async()`'s `concurrent.futures.Future` is not awaitable. +`aai.AsyncSyncTranscriber` is the asyncio counterpart of `aai.SyncTranscriber` — same input types, config, result, and errors, with `transcribe()`, `transcribe_live()` and `warm()` as coroutines and `open_live()` returning an `AsyncLiveSession`. Use it in asyncio code (FastAPI, aiohttp, voice agents), where the threaded `transcribe()` would block the event loop and `transcribe_async()`'s `concurrent.futures.Future` is not awaitable. ```python import asyncio diff --git a/assemblyai/sync/v1/__init__.py b/assemblyai/sync/v1/__init__.py index 4e3691aa..de668acf 100644 --- a/assemblyai/sync/v1/__init__.py +++ b/assemblyai/sync/v1/__init__.py @@ -7,14 +7,16 @@ ) from ._base import AudioInput from ._multipart import AsyncAudioChunks, AudioChunks -from .async_client import AsyncSyncTranscriber -from .client import SyncTranscriber +from .async_client import AsyncLiveSession, AsyncSyncTranscriber +from .client import LiveSession, SyncTranscriber __all__ = [ "AsyncAudioChunks", + "AsyncLiveSession", "AsyncSyncTranscriber", "AudioChunks", "AudioInput", + "LiveSession", "SyncSpeechModel", "SyncTranscriber", "SyncTranscriptError", diff --git a/assemblyai/sync/v1/_base.py b/assemblyai/sync/v1/_base.py index 1d8ec3e1..31d2554a 100644 --- a/assemblyai/sync/v1/_base.py +++ b/assemblyai/sync/v1/_base.py @@ -122,6 +122,15 @@ def _config_to_json(config: types.SyncTranscriptionConfig) -> Optional[dict]: return data or None +class _Aborted(Exception): + """ + Raised inside a live session's producer when the session is aborted. + + Propagates out of the transport, which drops the connection, and is + swallowed by `abort()`; it never reaches the caller. + """ + + def check_chunks(data: object, *, allow_async: bool = False) -> None: """ Raises unless `data` can be streamed. diff --git a/assemblyai/sync/v1/async_client.py b/assemblyai/sync/v1/async_client.py index d87d9cc3..a910bd2f 100644 --- a/assemblyai/sync/v1/async_client.py +++ b/assemblyai/sync/v1/async_client.py @@ -4,7 +4,7 @@ import asyncio from types import TracebackType -from typing import Any, Callable, Optional, Type, TypeVar +from typing import Any, AsyncIterator, Callable, Optional, Type, TypeVar import httpx from typing_extensions import Self @@ -14,13 +14,14 @@ from . import api, async_api from ._base import ( AudioInput, + _Aborted, _config_to_json, _resolve_audio, check_chunks, check_config, stream_filename, ) -from ._multipart import AsyncAudioChunks +from ._multipart import AsyncAudioChunks, _as_bytes _T = TypeVar("_T") @@ -33,6 +34,154 @@ async def _run_in_thread(func: Callable[..., _T], *args: Any) -> _T: return await loop.run_in_executor(None, func, *args) +class AsyncLiveSession: + """ + A live upload that audio is pushed into, for asyncio code. + + Returned by `AsyncSyncTranscriber.open_live()`. The request runs as a task + on the current event loop from the moment the session opens; `write()` + hands it audio, `close()` ends the audio, and `await result()` waits for + the transcript. Built for callback-driven sources — a WebRTC track, a + websocket handler receiving frames, a telephony media stream — where the + audio arrives in a callback rather than from an async iterator you can + hand to `transcribe_live()`. + + `write()` and `close()` are plain functions so a callback can call them, + and must run on the event loop's thread. From another thread — an audio + library's capture thread, say — schedule them with + `loop.call_soon_threadsafe(session.write, chunk)`. + + Everything `transcribe_live()` says about when it pays off, the 120 s + audio cap, the server-side silence abort and errors surfacing mid-upload + applies here unchanged. + + Example: + ```python + async with aai.AsyncSyncTranscriber() as transcriber: + async with transcriber.open_live(config) as session: + async for frame in websocket: # audio frames from a browser + session.write(frame) + print((await session.result()).text) + ``` + """ + + def __init__( + self, + start: Callable[ + [AsyncIterator[bytes]], asyncio.Task[types.SyncTranscriptResponse] + ], + ) -> None: + """ + Args: + start: begins the upload from the given producer and returns the + task that will hold its outcome. Supplied by the transcriber. + """ + self._queue: "asyncio.Queue[Optional[bytes]]" = asyncio.Queue() + self._closed = False + self._aborted = False + self._task = start(self._chunks()) + + async def _chunks(self) -> AsyncIterator[bytes]: + while True: + chunk = await self._queue.get() + if chunk is None: + if self._aborted: + raise _Aborted() + return + yield chunk + + @property + def closed(self) -> bool: + """Whether the audio has ended, by `close()`, `result()` or `abort()`.""" + + return self._closed + + def write(self, chunk: bytes) -> None: + """ + Queues a piece of audio for upload. Never blocks. + + Must be called on the event loop's thread; see the class docstring for + calling from elsewhere. + + Args: + chunk: `bytes`, `bytearray` or `memoryview` of audio, in order. + + Raises: + TypeError: if `chunk` is not bytes-like. + RuntimeError: if the session is closed. + """ + if self._closed: + raise RuntimeError( + "the live session is closed; no more audio can be written" + ) + + self._queue.put_nowait(_as_bytes(chunk)) + + def close(self) -> None: + """ + Ends the audio. The server transcribes what was sent; `result()` + returns it. Idempotent, and never blocks. + """ + if not self._closed: + self._closed = True + self._queue.put_nowait(None) + + async def result(self) -> types.SyncTranscriptResponse: + """ + Waits for the transcript, ending the audio first if it is still open. + + Raises: + SyncTranscriptError: if the request failed, including a rejection + the server sent while the upload was still in flight. + RuntimeError: if the session was aborted. + """ + if self._aborted: + raise RuntimeError("the live session was aborted; there is no result") + + self.close() + + return await self._task + + async def abort(self) -> None: + """ + Drops the request without a transcript. + + The connection is closed and nothing the server may still return is + kept; `result()` raises afterwards. Idempotent, and a no-op once the + request has completed. Waits until the task has let go of the + connection. After `close()` the upload is already complete, so + aborting then waits out the in-flight response instead. + """ + if self._aborted or self._task.done(): + return + + self._aborted = True + if not self._closed: + self._closed = True + self._queue.put_nowait(None) + + try: + await self._task + except Exception: + pass + + async def __aenter__(self) -> "AsyncLiveSession": + return self + + async def __aexit__( + self, + exc_type: Optional[Type[BaseException]], + exc_value: Optional[BaseException], + traceback: Optional[TracebackType], + ) -> None: + """Ends the audio on a clean exit; aborts if the block raised.""" + + if exc_type is not None: + await self.abort() + else: + self.close() + + class AsyncSyncTranscriber: """ The asyncio counterpart of `SyncTranscriber`: audio in, transcript out, @@ -220,6 +369,39 @@ async def mic_chunks(): timeout=self._client.settings.sync_live_http_timeout, ) + def open_live( + self, + config: Optional[types.SyncTranscriptionConfig] = None, + ) -> AsyncLiveSession: + """ + Opens a live upload that audio is pushed into. + + The push-style counterpart of `transcribe_live()`, for sources that + deliver audio through a callback rather than an async iterator. The + request starts as a task on the running event loop immediately; call + `session.write(chunk)` as audio arrives, then `await session.result()` + for the transcript once the speaker stops. See `AsyncLiveSession`. + + Not a coroutine, so it can be used directly as `async with + transcriber.open_live(config) as session:`. Must be called while the + event loop is running. + + Args: + config: Options for this call. If `None`, the transcriber's default + configuration is used. Raw PCM requires `sample_rate` and + `channels`. + + Raises: + TypeError: if `config` is not a `SyncTranscriptionConfig`. + RuntimeError: if no event loop is running. + """ + check_config(type(self).__name__, config) + loop = asyncio.get_running_loop() + + return AsyncLiveSession( + lambda chunks: loop.create_task(self.transcribe_live(chunks, config=config)) + ) + async def warm(self) -> bool: """ Opens the connection to the sync API ahead of time. diff --git a/assemblyai/sync/v1/client.py b/assemblyai/sync/v1/client.py index 8c5fa74c..1eeeb2b6 100644 --- a/assemblyai/sync/v1/client.py +++ b/assemblyai/sync/v1/client.py @@ -2,14 +2,167 @@ import concurrent.futures import os -from typing import Any, Optional +import queue +from typing import Any, Callable, Iterator, Optional import httpx from ... import client as _client from ... import types from . import api -from ._base import AudioChunks, AudioInput, _SyncTranscriberImpl, check_config +from ._base import ( + AudioChunks, + AudioInput, + _Aborted, + _SyncTranscriberImpl, + check_config, +) +from ._multipart import _as_bytes + + +class LiveSession: + """ + A live upload that audio is pushed into. + + Returned by `SyncTranscriber.open_live()`. The request starts on one of + the transcriber's worker threads the moment the session opens; `write()` + hands it audio, `close()` ends the audio, and `result()` waits for the + transcript. Built for callback-driven sources — a microphone library, a + WebRTC track, a telephony media stream — where the audio arrives in a + callback rather than from an iterator you can hand to `transcribe_live()`. + + Everything `transcribe_live()` says about when it pays off, the 120 s + audio cap, the server-side silence abort and errors surfacing mid-upload + applies here unchanged. + + Example: + ```python + import sounddevice as sd + + config = aai.SyncTranscriptionConfig(sample_rate=16000, channels=1) + + with aai.SyncTranscriber() as transcriber: + with transcriber.open_live(config) as session: + stream = sd.RawInputStream( + samplerate=16000, channels=1, dtype="int16", + callback=lambda data, *_: session.write(bytes(data)), + ) + with stream: + input("Recording, press Enter to stop... ") + print(session.result().text) + ``` + """ + + def __init__( + self, + start: Callable[ + [Iterator[bytes]], concurrent.futures.Future[types.SyncTranscriptResponse] + ], + ) -> None: + """ + Args: + start: begins the upload from the given producer and returns the + future that will hold its outcome. Supplied by the transcriber. + """ + self._queue: "queue.Queue[Optional[bytes]]" = queue.Queue() + self._closed = False + self._aborted = False + self._future = start(self._chunks()) + + def _chunks(self) -> Iterator[bytes]: + while True: + chunk = self._queue.get() + if chunk is None: + if self._aborted: + raise _Aborted() + return + yield chunk + + @property + def closed(self) -> bool: + """Whether the audio has ended, by `close()`, `result()` or `abort()`.""" + + return self._closed + + def write(self, chunk: bytes) -> None: + """ + Queues a piece of audio for upload. + + Safe to call from any thread, including an audio library's capture + callback; it never blocks. + + Args: + chunk: `bytes`, `bytearray` or `memoryview` of audio, in order. + + Raises: + TypeError: if `chunk` is not bytes-like. + RuntimeError: if the session is closed. + """ + if self._closed: + raise RuntimeError( + "the live session is closed; no more audio can be written" + ) + + self._queue.put(_as_bytes(chunk)) + + def close(self) -> None: + """ + Ends the audio. The server transcribes what was sent; `result()` + returns it. Idempotent, and never blocks. + """ + if not self._closed: + self._closed = True + self._queue.put(None) + + def result(self) -> types.SyncTranscriptResponse: + """ + Waits for the transcript, ending the audio first if it is still open. + + Raises: + SyncTranscriptError: if the request failed, including a rejection + the server sent while the upload was still in flight. + RuntimeError: if the session was aborted. + """ + if self._aborted: + raise RuntimeError("the live session was aborted; there is no result") + + self.close() + + return self._future.result() + + def abort(self) -> None: + """ + Drops the request without a transcript. + + The connection is closed and nothing the server may still return is + kept; `result()` raises afterwards. Idempotent, and a no-op once the + request has completed. Blocks until the worker thread has let go of + the connection. After `close()` the upload is already complete, so + aborting then waits out the in-flight response instead. + """ + if self._aborted or self._future.done(): + return + + self._aborted = True + if not self._closed: + self._closed = True + self._queue.put(None) + + try: + self._future.result() + except Exception: + pass + + def __enter__(self) -> "LiveSession": + return self + + def __exit__(self, exc_type: Any, *_exc: Any) -> None: + """Ends the audio on a clean exit; aborts if the block raised.""" + + if exc_type is not None: + self.abort() + else: + self.close() class SyncTranscriber: @@ -187,6 +340,43 @@ def mic_chunks(): return self._impl.transcribe_live(data=data, config=config) + def open_live( + self, + config: Optional[types.SyncTranscriptionConfig] = None, + ) -> LiveSession: + """ + Opens a live upload that audio is pushed into. + + The push-style counterpart of `transcribe_live()`, for sources that + deliver audio through a callback rather than an iterator. The request + starts on a worker thread immediately; call `session.write(chunk)` from + the callback, then `session.result()` for the transcript once the + speaker stops. See `LiveSession`. + + Args: + config: Options for this call. If `None`, the transcriber's default + configuration is used. Raw PCM requires `sample_rate` and + `channels`. + + Raises: + TypeError: if `config` is not a `SyncTranscriptionConfig`. + + Example: + ```python + with transcriber.open_live(config) as session: + start_capture(on_audio=session.write) + wait_until_done() + result = session.result() + ``` + """ + check_config(type(self).__name__, config) + + return LiveSession( + lambda chunks: self._executor.submit( + self._impl.transcribe_live, data=chunks, config=config + ) + ) + def warm(self) -> bool: """ Opens the connection to the sync API ahead of time. diff --git a/tests/unit/test_sync_live.py b/tests/unit/test_sync_live.py index ab33e31f..4a5e7363 100644 --- a/tests/unit/test_sync_live.py +++ b/tests/unit/test_sync_live.py @@ -473,3 +473,360 @@ async def chunks(): await transcriber.transcribe_live(chunks()) assert exc.value.error_code == "capacity_exceeded" + + +# --- push-style sessions ------------------------------------------------------ + + +def _fake_live(monkeypatch, module, drain, error: Optional[Exception] = None) -> dict: + """ + Replaces the transport with a fake that drains the producer the way a real + transport would, recording what it saw. `drain` turns the producer into a + list (sync or async), so an abort raised by the producer propagates from it + exactly as it would from httpx. + """ + seen: dict = {"called": threading.Event(), "completed": False} + + if asyncio.iscoroutinefunction(drain): + + async def fake(client, **kwargs): + seen["called"].set() + seen["config"] = kwargs["config"] + seen["chunks"] = await drain(kwargs["chunks"]) + seen["completed"] = True + if error: + raise error + return aai.SyncTranscriptResponse.parse_obj(_OK_RESPONSE) + + else: + + def fake(client, **kwargs): + seen["called"].set() + seen["config"] = kwargs["config"] + seen["chunks"] = drain(kwargs["chunks"]) + seen["completed"] = True + if error: + raise error + return aai.SyncTranscriptResponse.parse_obj(_OK_RESPONSE) + + monkeypatch.setattr(module, "transcribe_live", fake) + return seen + + +def _drain(chunks): + return list(chunks) + + +async def _adrain(chunks): + return [chunk async for chunk in chunks] + + +def test_open_live_uploads_written_chunks_in_order(monkeypatch): + # Given a session fed from a callback-style producer + seen = _fake_live(monkeypatch, api, _drain) + + with aai.SyncTranscriber() as transcriber: + session = transcriber.open_live( + aai.SyncTranscriptionConfig(sample_rate=16000, channels=1) + ) + for piece in (b"\x00\x01", bytearray(b"\x02\x03"), memoryview(b"\x04\x05")): + session.write(piece) + + # When the audio ends and the result is awaited + result = session.result() + + # Then the chunks went out in order, as bytes, with the config, and the + # transcript came back parsed + assert seen["chunks"] == [b"\x00\x01", b"\x02\x03", b"\x04\x05"] + assert seen["config"] == {"sample_rate": 16000, "channels": 1} + assert result.text == "hello world" + + +def test_open_live_starts_the_request_before_any_audio(monkeypatch): + # Given a session that has just been opened + seen = _fake_live(monkeypatch, api, _drain) + + with aai.SyncTranscriber() as transcriber: + session = transcriber.open_live() + + # Then the request is already in flight, which is the point: the + # connection and config go out while the speaker is still talking + assert seen["called"].wait(timeout=2.0) + assert not seen["completed"] + + session.result() + + +def test_live_session_context_manager_ends_the_audio(monkeypatch): + # Given audio written inside a with-block + _fake_live(monkeypatch, api, _drain) + + with aai.SyncTranscriber() as transcriber: + with transcriber.open_live() as session: + session.write(b"RIFF") + assert not session.closed + + # Then leaving the block closes the audio and the result is waiting + assert session.closed + assert session.result().text == "hello world" + + +def test_live_session_rejects_writes_after_close(monkeypatch): + # Given a closed session + _fake_live(monkeypatch, api, _drain) + + with aai.SyncTranscriber() as transcriber: + session = transcriber.open_live() + session.close() + + # When more audio arrives, then it is refused rather than dropped silently + with pytest.raises(RuntimeError, match="closed"): + session.write(b"RIFF") + + session.result() + + +def test_live_session_rejects_text_at_write_time(monkeypatch): + # Given a producer handing out str + _fake_live(monkeypatch, api, _drain) + + with aai.SyncTranscriber() as transcriber: + session = transcriber.open_live() + + # When it is written, then the mistake is named immediately, not when + # the result is collected + with pytest.raises(TypeError, match="bytes"): + session.write("RIFF") + + session.result() + + +def test_live_session_abort_drops_the_request(monkeypatch): + # Given a session mid-upload + seen = _fake_live(monkeypatch, api, _drain) + + with aai.SyncTranscriber() as transcriber: + session = transcriber.open_live() + session.write(b"RIFF") + assert seen["called"].wait(timeout=2.0) + + # When it is aborted + session.abort() + + # Then the transport never completed the upload, there is no result, + # and a second abort is harmless + assert not seen["completed"] + assert session.closed + with pytest.raises(RuntimeError, match="aborted"): + session.result() + session.abort() + + +def test_live_session_aborts_when_the_block_raises(monkeypatch): + # Given a with-block that fails part-way through recording + seen = _fake_live(monkeypatch, api, _drain) + + with aai.SyncTranscriber() as transcriber: + with pytest.raises(ValueError): + with transcriber.open_live() as session: + session.write(b"RIFF") + raise ValueError("microphone unplugged") + + # Then the upload was dropped rather than transcribed + assert not seen["completed"] + with pytest.raises(RuntimeError, match="aborted"): + session.result() + + +def test_live_session_abort_after_completion_is_a_no_op(monkeypatch): + # Given a session whose result has already arrived + _fake_live(monkeypatch, api, _drain) + + with aai.SyncTranscriber() as transcriber: + session = transcriber.open_live() + result = session.result() + + # When it is aborted anyway, then nothing changes + session.abort() + assert session.result() is result + + +def test_live_session_rejects_job_api_config(): + # Given the job API's config type + with aai.SyncTranscriber() as transcriber: + with pytest.raises(TypeError, match="SyncTranscriptionConfig"): + transcriber.open_live(aai.TranscriptionConfig()) + + +def test_live_session_roundtrip_over_http(httpx_mock: HTTPXMock): + # Given the real transport against a mocked endpoint + _mock_ok(httpx_mock) + + with aai.SyncTranscriber() as transcriber: + with transcriber.open_live() as session: + session.write(b"RIFF") + session.write(b"fake") + + # Then the request hit the live route and parsed like the pull path + result = session.result() + + assert result.text == "hello world" + request = httpx_mock.get_requests()[0] + assert str(request.url) == STREAM_URL + assert request.headers.get("transfer-encoding") == "chunked" + + +def test_live_session_surfaces_server_errors(httpx_mock: HTTPXMock): + # Given a rejection the server may send mid-upload + httpx_mock.add_response( + url=STREAM_URL, + method="POST", + status_code=httpx.codes.SERVICE_UNAVAILABLE, + json={"status": 503, "title": "Capacity Exceeded", "detail": "no capacity"}, + ) + + with aai.SyncTranscriber() as transcriber: + session = transcriber.open_live() + session.write(b"RIFF") + + # When the result is collected, then it is the same error the pull path raises + with pytest.raises(aai.SyncTranscriptError) as exc: + session.result() + + assert exc.value.error_code == "capacity_exceeded" + + +@pytest.mark.asyncio +async def test_async_open_live_uploads_written_chunks_in_order(monkeypatch): + # Given an asyncio session fed from a callback-style producer + seen = _fake_live(monkeypatch, async_api, _adrain) + + async with aai.AsyncSyncTranscriber() as transcriber: + session = transcriber.open_live( + aai.SyncTranscriptionConfig(sample_rate=16000, channels=1) + ) + for piece in (b"\x00\x01", bytearray(b"\x02\x03")): + session.write(piece) + + # When the audio ends and the result is awaited + result = await session.result() + + # Then it matches the sync twin + assert seen["chunks"] == [b"\x00\x01", b"\x02\x03"] + assert seen["config"] == {"sample_rate": 16000, "channels": 1} + assert result.text == "hello world" + + +@pytest.mark.asyncio +async def test_async_live_session_context_manager_ends_the_audio(monkeypatch): + # Given audio written inside an async with-block + _fake_live(monkeypatch, async_api, _adrain) + + async with aai.AsyncSyncTranscriber() as transcriber: + async with transcriber.open_live() as session: + session.write(b"RIFF") + + # Then leaving the block closes the audio and the result is waiting + assert session.closed + assert (await session.result()).text == "hello world" + + +@pytest.mark.asyncio +async def test_async_live_session_rejects_writes_after_close(monkeypatch): + # Given a closed session + _fake_live(monkeypatch, async_api, _adrain) + + async with aai.AsyncSyncTranscriber() as transcriber: + session = transcriber.open_live() + session.close() + + # When more audio arrives, then it is refused + with pytest.raises(RuntimeError, match="closed"): + session.write(b"RIFF") + + await session.result() + + +@pytest.mark.asyncio +async def test_async_live_session_abort_drops_the_request(monkeypatch): + # Given a session mid-upload + seen = _fake_live(monkeypatch, async_api, _adrain) + + async with aai.AsyncSyncTranscriber() as transcriber: + session = transcriber.open_live() + session.write(b"RIFF") + await asyncio.sleep(0) # let the task start consuming + + # When it is aborted + await session.abort() + + # Then the transport never completed the upload and there is no result + assert not seen["completed"] + with pytest.raises(RuntimeError, match="aborted"): + await session.result() + await session.abort() + + +@pytest.mark.asyncio +async def test_async_live_session_aborts_when_the_block_raises(monkeypatch): + # Given an async with-block that fails mid-recording + seen = _fake_live(monkeypatch, async_api, _adrain) + + async with aai.AsyncSyncTranscriber() as transcriber: + with pytest.raises(ValueError): + async with transcriber.open_live() as session: + session.write(b"RIFF") + raise ValueError("call dropped") + + # Then the upload was dropped rather than transcribed + assert not seen["completed"] + with pytest.raises(RuntimeError, match="aborted"): + await session.result() + + +@pytest.mark.asyncio +async def test_async_live_session_roundtrip_over_http(httpx_mock: HTTPXMock): + # Given the real transport against a mocked endpoint + _mock_ok(httpx_mock) + + async with aai.AsyncSyncTranscriber() as transcriber: + async with transcriber.open_live() as session: + session.write(b"RIFF") + + result = await session.result() + + # Then the request hit the live route and parsed like the pull path + assert result.text == "hello world" + assert str(httpx_mock.get_requests()[0].url) == STREAM_URL + + +@pytest.mark.asyncio +async def test_async_live_session_surfaces_server_errors(httpx_mock: HTTPXMock): + # Given a rejection + httpx_mock.add_response( + url=STREAM_URL, + method="POST", + status_code=httpx.codes.TOO_MANY_REQUESTS, + json={"status": 429, "title": "Rate Limited", "detail": "slow down"}, + headers={"Retry-After": "3"}, + ) + + async with aai.AsyncSyncTranscriber() as transcriber: + session = transcriber.open_live() + session.write(b"RIFF") + + # When the result is collected, then it is the same error the pull path raises + with pytest.raises(aai.SyncTranscriptError) as exc: + await session.result() + + assert exc.value.status_code == 429 + assert exc.value.retry_after == 3 + + +def test_async_open_live_needs_a_running_loop(): + # Given no event loop + transcriber = aai.AsyncSyncTranscriber() + + # When a session is opened, then the mistake is named rather than deferred + with pytest.raises(RuntimeError, match="no running event loop"): + transcriber.open_live()