diff --git a/README.md b/README.md index 98f4ff9..15c19fe 100644 --- a/README.md +++ b/README.md @@ -50,38 +50,52 @@ Set from the tray menu, or via files / env vars: Devices that refuse 16 kHz (raw ALSA and PipeWire's JACK endpoints do) are captured at their native rate and resampled. -Every clip is conditioned before upload — the stream-open click is trimmed off -the front (only as much of it as is actually there, up to 150 ms), DC bias is -filtered out, and the level is normalised towards -20 dBFS. Near-silent clips -are refused instead of being sent, since transcription models answer silence -with confident hallucinations. +By default, Blurt transcribes while you speak. It conditions the incoming audio, +then seals and uploads phrases as pauses provide natural boundaries. A chunk is +at least about 8 seconds long; early cuts require roughly 800 ms of quiet, while +the required pause falls towards 500 ms as the chunk approaches the 20-second +target. After 45 seconds it uses the quietest recent point even without a full +pause, so one uninterrupted passage cannot hold up the whole recording. Pause +detection follows the microphone's noise floor and only skips near-digital +silence; ordinary quiet speech and the silence around phrase boundaries remain +in the audio. -Every clip is written to `spool/` (gitignored) before it is uploaded, and -deleted again as soon as its text reaches the clipboard. If transcription fails — -timeout, connection error, server error, empty response — the WAV stays there -and the notification says so; **Retry failed transcriptions (N)** in the tray -menu re-uploads the spooled clips oldest first, so a network hiccup never costs -you the words. A rejected request (any 4xx other than 408 and 429) is the -exception: the same audio can only be rejected again, so it is not kept, and the -notification says that too. Retried transcripts are copied to the clipboard -together, oldest first, once the whole run is done — and never dispatched to an -agent, since a spooled clip may be one part of a longer recording. The upload -timeout scales with the clip: 30 s plus 30 s per minute of audio, capped at -5 minutes. Clips longer than 5 minutes are split into chunks of about -4 minutes — each cut lands in a pause, not mid-word — and the chunks are -uploaded in parallel and stitched back together in order. The tray icon fills -clockwise as each chunk comes back, so you can see how far along the clip is. -If only some chunks come back, the transcript is copied with ` […] ` where the -gaps are, and just the failed chunks are kept in the spool as `…-partNN.wav` -for the retry. +Up to three chunks are transcribed concurrently. Their results are retained by +position and delivered once, in recording order, after you stop and the final +chunk finishes. Starting another recording does not mix the two sessions, and +each session keeps the model, language, provider, and API key selection it had +when recording began even if you change the tray settings meanwhile. The tray +icon shows progress as chunks finish. +While recording, the red icon brightens and its three level bars rise with +microphone volume, settling back down during pauses. This shows incoming audio +activity; it does not indicate that the transcription service has recognized words. -Nothing else is cut: every second of the recording is uploaded. If a transcript comes -back missing words, check `blurt.log` — each clip logs its length, speech level, -noise floor, and `dead=`, the longest run of digital silence inside it. A -non-zero `dead=` means the input dropped out mid-clip, which is what following -the system default source invites; pin a device instead. Set -`BLURT_SAVE_CAPTURES=1` (or to a directory) to keep the raw and uploaded WAVs -for every clip and hear exactly what the API got. +Each session has a recovery directory under `spool/` (gitignored). Sealed chunks, +completed text, and a checkpoint of the unsealed tail are kept there until the +complete transcript reaches the clipboard or agent. The tail is checkpointed +about once a second; an abrupt crash can lose audio since the last checkpoint. +**Retry failed transcriptions (N)** re-uploads unfinished chunks and reconstructs +the saved recording, including text that succeeded before a network error, +timeout, server error, or empty response. A partial result is copied with ` […] ` +and retained for retry; retried results are +copied rather than dispatched to an agent. The upload timeout scales with each +chunk: 30 seconds plus 30 seconds per minute of audio, capped at five minutes. + +Set `BLURT_STREAMING=0` before starting Blurt to use the legacy whole-clip flow. +That mode trims up to 150 ms of stream-open click, filters DC bias, normalises +towards -20 dBFS, and rejects near-silent clips. It spools the complete clip +before upload, splits recordings longer than five minutes near pauses into +roughly four-minute parallel uploads, retains retryable failures, and marks +missing chunks with ` […] `. Rejected requests (4xx other than 408 and 429) are +discarded because retrying the same request cannot fix them. + +If a transcript comes back missing words, check `blurt.log`; streaming mode logs +chunk boundaries, noise floor, and `dead=`, the longest run of digital silence +per chunk, plus the time from Stop to delivery. A non-zero `dead=` can indicate +an input dropout. Pinning an +input device avoids the system default moving between sources. In legacy mode, +set `BLURT_SAVE_CAPTURES=1` (or to a directory) to retain raw and uploaded WAVs +and hear exactly what the API received. ## Checking the microphone diff --git a/blurt.py b/blurt.py index 8934d00..1cc3564 100644 --- a/blurt.py +++ b/blurt.py @@ -28,6 +28,9 @@ from PIL import Image, ImageDraw from scipy import signal as dsp from scipy.io import wavfile +from blurt_client import TranscriptionClient, TranscriptionConfig +from blurt_session import RecordingSession, RecoveryStore, TranscriptionScheduler + OPENAI_TRANSCRIPTIONS_URL = "https://api.openai.com/v1/audio/transcriptions" OPENROUTER_TRANSCRIPTIONS_URL = "https://openrouter.ai/api/v1/audio/transcriptions" OPENROUTER_APP_URL = "https://git.felixm.de/felixm/blurt" @@ -835,6 +838,21 @@ class BlurtApp: self.chunks_total = 0 self.chunks_done = 0 self._inflight_spool: set[Path] = set() + self.streaming = os.environ.get("BLURT_STREAMING", "1") != "0" + self._client = TranscriptionClient() + self._scheduler = TranscriptionScheduler(self._client.transcribe, workers=CHUNK_WORKERS) + self._recovery = RecoveryStore(SPOOL_DIR) + self._live_lock = threading.RLock() + self._delivery_lock = threading.Lock() + self._live_sessions: list[RecordingSession] = [] + self._recording_session: RecordingSession | None = None + self._retrying = False + self._retry_thread: threading.Thread | None = None + self._quitting = False + self._meter_stop = threading.Event() + self._meter_frame = None + self._meter_level = 0.0 + self._meter_icons = [self._create_recording_image(i / 11) for i in range(12)] spooled = self._spooled_files() if spooled: logging.warning( @@ -914,6 +932,16 @@ class BlurtApp: if self.recording: color = (220, 40, 40) # red state = "Recording..." + active = self._recording_session + if active is not None: + state = f"Recording · {active.done}/{active.total} chunks processed" + elif self._live_sessions: + color = (40, 100, 220) + sessions = tuple(self._live_sessions) + done = sum(s.done for s in sessions) + total = sum(s.total for s in sessions) + state = f"Finishing {done}/{total} chunks..." + progress = done / total if total else None elif self.pending_transcriptions > 0: color = (40, 100, 220) # blue if total > 1: @@ -924,15 +952,79 @@ class BlurtApp: else: color = (120, 120, 120) # gray state = "Ready" - self.icon.icon = self._create_icon_image(color, progress) + self.icon.icon = ( + self._meter_icons[round(self._meter_level * 11)] + if self.recording else self._create_icon_image(color, progress) + ) self.icon.title = f"{APP_NAME} - {state}" self.icon.update_menu() def _audio_callback(self, indata, frames, time_info, status): + if hasattr(self, "_meter_stop"): + # Keep only a small, owned snapshot. Level calculation and tray + # drawing happen outside PortAudio's callback. + self._meter_frame = ( + time.monotonic(), indata[::max(1, frames // 512)].copy(), + ) + active = self._recording_session + if active is not None: + if status: + active.report_input_status(str(status)) + active.feed(indata) + return if status: logging.warning("sounddevice input status: %s", status) self.audio_frames.append(indata.copy()) + def _create_recording_image(self, level: float) -> Image.Image: + color = (round(160 + 80 * level), round(28 + 32 * level), 40) + img = self._create_icon_image(color) + draw = ImageDraw.Draw(img) + for x, scale in ((21, 0.6), (30, 1.0), (39, 0.75)): + height = round(4 + 30 * level * scale) + draw.rounded_rectangle( + (x, 32 - height // 2, x + 4, 32 + height // 2), + radius=2, fill=(255, 235, 235), + ) + return img + + @staticmethod + def _microphone_level(samples: np.ndarray) -> float: + mono = samples.mean(axis=1) if samples.ndim == 2 else samples + if not len(mono): + return 0.0 + centered = mono - mono.mean() + rms = float(np.sqrt(np.mean(centered * centered))) + dbfs = 20 * np.log10(max(rms, 1e-6)) + return float(np.clip((dbfs + 60) / 42, 0, 1)) + + def _animate_microphone(self): + previous = None + while not self._meter_stop.wait(0.1): + snapshot = self._meter_frame + target = 0.0 + if self.recording and snapshot is not None: + captured_at, samples = snapshot + if time.monotonic() - captured_at < 0.3: + target = self._microphone_level(samples) + # Fast attack, gentle release: speech is obvious without strobing. + self._meter_level = max(target, self._meter_level * 0.65) if self.recording else 0.0 + step = round(self._meter_level * 11) + if not self.recording: + previous = None + continue + if step == previous: + continue + try: + with self._icon_lock: + if self.recording and self.icon and not self._quitting: + self.icon.icon = self._meter_icons[step] + previous = step + # Deliberately leave menu/title updates to state transitions. + except Exception: + logging.exception("Could not update microphone activity icon") + return + def _provider_for_model(self, model: str) -> str: for model_id, _, provider in TRANSCRIPTION_MODELS: if model_id == model: @@ -992,6 +1084,8 @@ class BlurtApp: def _start_recording(self): if not self._require_api_key("Recording"): return + self._meter_frame = None + self._meter_level = 0.0 self.audio_frames = [] try: self.capture_rate, self.capture_channels = input_stream_settings(self.device_index) @@ -1406,10 +1500,71 @@ class BlurtApp: return [path for path in self._spooled_files() if path not in inflight] def _retry_spooled(self, icon=None, item=None): + if self.streaming: + with self._live_lock: + if self._retrying or self._quitting: + return + self._retrying = True + self._retry_thread = threading.Thread(target=self._retry_sessions, daemon=True, name="RetrySessions") + self._retry_thread.start() + return if not self._require_api_key("Retry"): return threading.Thread(target=self._retry_spooled_worker, daemon=True, name="RetrySpool").start() + def _retry_count(self): + if self.streaming: + return len(self._recovery.discover()) + len(self._retryable_files()) + return len(self._retryable_files()) + + def _retry_sessions(self): + collected = [] + recovered = [] + attempted = [] + try: + entries = [(r.created_at, r, None) for r in self._recovery.discover()] + entries.extend((p.stat().st_mtime, None, p) for p in self._retryable_files()) + for _, record, path in sorted(entries, key=lambda entry: entry[0]): + if self._quitting: + break + try: + config = self._transcription_config(record.config if record else None) + if not config.api_key: + self.platform.notify("Error", f"Missing {config.provider} key for saved recording.") + continue + if record is not None: + active = self._recovery.resume(record, config, self._scheduler) + else: + active = RecordingSession.from_wav(path, config, SPOOL_DIR, self._scheduler) + attempted.append(active) + result = active.join() + if result and result.text: + collected.append(result.text) + recovered.append(active) + elif result and result.complete: + active.acknowledge_delivery() # recovered silence needs no clipboard + except Exception: + logging.exception("Could not recover recording %s", record or path) + if collected and not self._quitting: + with self._delivery_lock: + pyperclip.copy("\n\n".join(collected)) + for active in recovered: + active.acknowledge_delivery() + self.platform.notify("Copied", f"Recovered {len(collected)} recording(s).") + elif not self._quitting: + self.platform.notify("Retry", "No transcripts recovered; audio remains saved.") + except Exception: + logging.exception("Session retry failed; recovery data retained") + if not self._quitting: + self.platform.notify("Error", "Retry failed; audio remains saved.") + finally: + for active in attempted: + active.release_for_retry() + with self._live_lock: + self._retrying = False + if not self._quitting: + self._update_icon() + def _retry_spooled_worker(self): """Re-upload spooled clips one at a time: they failed once already, and a burst of parallel requests is the least likely thing to fix that. Their @@ -1494,6 +1649,9 @@ class BlurtApp: logging.error("Transcription error response: %s", response.text[:1000]) def toggle_recording(self): + if self.streaming: + self._toggle_streaming() + return with self.lock: if not self.recording: self._start_recording() @@ -1514,6 +1672,128 @@ class BlurtApp: else: self.platform.notify("Error", "No audio recorded.") + def _transcription_config(self, metadata=None): + metadata = metadata or {} + model = metadata.get("model", self.transcription_model) + provider = metadata.get("provider", self._provider_for_model(model)) + return TranscriptionConfig( + model=model, + language=metadata.get("language", self.transcription_language), + provider=provider, + api_key=self.api_keys.get(provider, ""), + ) + + def _session_update(self, session): + if not self._quitting: + if (session is self._recording_session and session.result is not None + and session.result.errors and not getattr(session, "capture_warned", False)): + session.capture_warned = True + self.platform.notify("Capture interrupted", "Stop recording; available audio is kept for retry.") + self._update_icon() + + def _session_complete(self, session): + # Sessions may finish out of order. Deliver in recording order, and keep + # delivery outside the microphone/start-stop lock. + with self._delivery_lock: + while not self._quitting: + with self._live_lock: + if not self._live_sessions or self._live_sessions[0].result is None: + break + if self._live_sessions[0] is self._recording_session: + # An input failure can finish the producer before the + # user stops the microphone. Never deliver mid-recording. + break + ready = self._live_sessions.pop(0) + result = ready.result + try: + if result.text: + if result.complete: + self._deliver(result.text) + else: + pyperclip.copy(result.text) + self.platform.notify("Partial", "Copied with gaps; audio kept for retry.") + ready.acknowledge_delivery() + stopped_at = getattr(ready, "stopped_at", None) + if stopped_at is not None: + logging.info( + "Session delivered: id=%s stop_to_delivery=%.3fs complete=%s", + result.session_id, time.monotonic() - stopped_at, result.complete, + ) + elif result.errors: + ready.release_for_retry() + self.platform.notify("Error", "Transcription failed; audio kept for retry.") + else: + ready.acknowledge_delivery() + self.platform.notify("Recording", "No speech detected.") + except Exception: + logging.exception("Session delivery failed; recovery data retained") + ready.release_for_retry() + self.platform.notify("Error", "Could not deliver transcript; kept for retry.") + self._session_update(session) + + def _toggle_streaming(self): + with self.lock: + if self._quitting: + return + if self.recording: + self._stop_streaming() + return + if not self._require_api_key("Recording"): + return + self._meter_frame = None + self._meter_level = 0.0 + active = None + try: + self.capture_rate, self.capture_channels = input_stream_settings(self.device_index) + active = RecordingSession( + self.capture_rate, self.capture_channels, + self._transcription_config(), SPOOL_DIR, self._scheduler, + on_update=self._session_update, on_complete=self._session_complete, + ) + with self._live_lock: + self._live_sessions.append(active) + self._recording_session = active + self.stream = sd.InputStream( + device=self.device_index, samplerate=self.capture_rate, + channels=self.capture_channels, dtype=DTYPE, + callback=self._audio_callback, + ) + self.stream.start() + self.recording = True + logging.info( + "Streaming capture started: session=%s device=%s rate=%d channels=%d", + active.session_id, self.device_name or "", + self.capture_rate, self.capture_channels, + ) + except Exception as error: + logging.exception("Could not start streaming capture") + if self.stream is not None: + self.stream.close() + self.stream = None + self._recording_session = None + if active is not None: + active.stop() + self.platform.notify("Error", f"Could not start recording: {error}") + return + self._update_icon() + self.platform.notify("Recording", "Transcribing as you speak. Press hotkey to stop.") + + def _stop_streaming(self): + # PortAudio must finish its last callback before the session receives EOF. + if self.stream is not None: + self.stream.stop() + self.stream.close() + self.stream = None + active = self._recording_session + self._recording_session = None + self.recording = False + if active is not None: + active.stopped_at = time.monotonic() + active.stop() + if active.result is not None: + self._session_complete(active) + self._update_icon() + def _on_click(self, icon, item): self.toggle_recording() @@ -1659,9 +1939,9 @@ class BlurtApp: pystray.MenuItem(lambda item: self._api_key_label(), None, enabled=False), pystray.MenuItem("Language", pystray.Menu(*language_items)), pystray.MenuItem( - lambda item: f"Retry failed transcriptions ({len(self._retryable_files())})", + lambda item: f"Retry failed transcriptions ({self._retry_count()})", self._retry_spooled, - enabled=lambda item: len(self._retryable_files()) > 0, + enabled=lambda item: not self._retrying and self._retry_count() > 0, ), pystray.MenuItem("Show Logs", self._show_logs), *awake_items, @@ -1670,15 +1950,36 @@ class BlurtApp: ) def _on_quit(self, icon, item): + self._quitting = True + self._meter_stop.set() if self.recording: - self._stop_recording() + if self.streaming: + self._stop_streaming() + else: + self._stop_recording() if self._awake_keeper and self._awake_keeper.is_active: self._awake_keeper.stop() self.platform.cleanup_toggle_trigger() + # A non-daemon coordinator keeps Python alive after the tray event loop + # exits, until capture checkpoints and pending responses are durable. + threading.Thread(target=self._finish_shutdown, name="BlurtShutdown", daemon=False).start() icon.stop() + def _finish_shutdown(self): + if self.streaming: + # Finish checkpointing and uploads without delivering into an app + # the user has closed. Undelivered text remains recoverable. + for active in tuple(self._live_sessions): + active.join() + active.release_for_retry() + if self._retry_thread is not None: + self._retry_thread.join() + self._scheduler.shutdown(wait=True) + self._client.close() + def _setup(self, icon): icon.visible = True + threading.Thread(target=self._animate_microphone, name="MicrophoneMeter", daemon=True).start() def run(self): self.platform.setup_toggle_trigger(self) diff --git a/blurt_audio.py b/blurt_audio.py new file mode 100644 index 0000000..2dfcf31 --- /dev/null +++ b/blurt_audio.py @@ -0,0 +1,313 @@ +"""Streaming capture conditioning and pause-aware chunking for Blurt. + +The class deliberately has no UI or sounddevice dependencies. ``feed`` may be +called with whatever block size the input device supplies; all public sample +positions are in the 16 kHz output timebase. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from math import gcd +import logging + +import numpy as np +from scipy import signal + + +OUTPUT_RATE = 16_000 +FRAME = OUTPUT_RATE // 50 # 20 ms +MIN_SECONDS = 8.0 +TARGET_SECONDS = 20.0 +MAX_SECONDS = 45.0 +EARLY_PAUSE_SECONDS = 0.8 +LATE_PAUSE_SECONDS = 0.5 +HIGHPASS_HZ = 70.0 +TARGET_RMS = 10 ** (-20.0 / 20.0) +PEAK_CEILING = 10 ** (-1.0 / 20.0) +MAX_GAIN = 10 ** (30.0 / 20.0) + + +@dataclass(frozen=True) +class AudioChunk: + """A contiguous section of conditioned audio in the session timebase.""" + + samples: np.ndarray + start: int + end: int + speech: bool + + +def _mono(frames: np.ndarray, channels: int) -> np.ndarray: + data = np.asarray(frames, dtype=np.float32) + if data.ndim == 1: + if channels != 1: + raise ValueError("one-dimensional frames are valid only for mono capture") + return data + if data.ndim != 2 or data.shape[1] != channels: + raise ValueError(f"expected frames shaped (n, {channels})") + return data.mean(axis=1, dtype=np.float32) + + +def _levels(samples: np.ndarray) -> np.ndarray: + count = len(samples) // FRAME + if not count: + return np.empty(0, dtype=np.float32) + framed = samples[: count * FRAME].reshape(count, FRAME) + framed = framed - framed.mean(axis=1, keepdims=True) + rms = np.sqrt(np.mean(framed * framed, axis=1)) + return (20.0 * np.log10(np.maximum(rms, 1e-7))).astype(np.float32) + + +def _normalize(samples: np.ndarray) -> np.ndarray: + """Normalize one sealed, already high-pass-filtered region.""" + if not len(samples): + return np.empty(0, dtype=np.int16) + filtered = samples + rms = float(np.sqrt(np.mean(filtered * filtered))) + peak = float(np.percentile(np.abs(filtered), 99.9)) + gain = min(TARGET_RMS / rms, PEAK_CEILING / peak, MAX_GAIN) if rms and peak else 1.0 + scaled = np.clip(filtered * gain, -1.0, 1.0) + return np.rint(scaled * 32767.0).astype(np.int16) + + +class StreamingAudio: + """Convert native-rate blocks into pause-aware 16 kHz mono chunks. + + Native input is resampled in one-second, phase-aligned batches with 50 ms + of context on both sides. Thus callback block size does not affect output + length, while filter-edge errors stay outside the retained middle. A final + partial batch is length-corrected to the rounded session duration. Startup + audio is deliberately retained: without evidence of a device click, an + automatic trim risks deleting the first quiet consonant. + """ + + _GUARD = FRAME * 2 + _CONTEXT_SECONDS = 0.05 + + def __init__(self, rate: int, channels: int): + if rate <= 0 or channels <= 0: + raise ValueError("rate and channels must be positive") + self.rate = int(rate) + self.channels = int(channels) + self._pending: list[np.ndarray] = [] + self._pending_len = 0 + self._left_native = np.empty(0, dtype=np.float32) + self._resampled = np.empty(0, dtype=np.float32) + self._base = 0 + self._start = 0 # absolute 16 kHz sample position + self._output_total = 0 + self._sos = signal.butter( + 2, HIGHPASS_HZ, btype="highpass", fs=OUTPUT_RATE, output="sos" + ) + self._filter_zi = None + self._finished = False + + def feed(self, frames: np.ndarray) -> list[AudioChunk]: + if self._finished: + raise RuntimeError("cannot feed a finished stream") + mono = _mono(frames, self.channels) + previous_available = self._base + len(self._resampled) + if len(mono): + self._pending.append(mono.copy()) + self._pending_len += len(mono) + self._refresh(final=False) + # Completed native batches already used right-side resampler context. + stable = self._base + len(self._resampled) + if stable == previous_available: + return [] + return self._emit(stable, finishing=False) + + def finish(self) -> list[AudioChunk]: + if self._finished: + return [] + self._finished = True + self._refresh(final=True) + chunks = self._emit(self._base + len(self._resampled), finishing=True) + end = self._base + len(self._resampled) + if self._start < end: + chunks.append(self._make_chunk(self._start, end)) + self._start = end + self._discard_sealed() + return chunks + + def checkpoint(self) -> np.ndarray: + """Return a conditioned copy of the currently unsealed audio.""" + preview = self._preview() + local = self._start - self._base + return _normalize(preview[local:]).copy() + + def _resample(self, native: np.ndarray) -> np.ndarray: + if self.rate == OUTPUT_RATE: + return native.astype(np.float32, copy=True) + divisor = gcd(self.rate, OUTPUT_RATE) + return signal.resample_poly( + native, OUTPUT_RATE // divisor, self.rate // divisor + ).astype(np.float32) + + def _refresh(self, final: bool) -> None: + batch = self.rate + context = round(self.rate * self._CONTEXT_SECONDS) + while self._pending_len >= batch + (0 if final else context): + pending = np.concatenate(self._pending) + take = min(batch, self._pending_len) + right = min(context, self._pending_len - take) + left = len(self._left_native) + source = np.concatenate((self._left_native, pending[: take + right])) + converted = self._resample(source) + offset = round(left * OUTPUT_RATE / self.rate) + wanted = round(take * OUTPUT_RATE / self.rate) + self._append_filtered(converted[offset : offset + wanted]) + consumed = pending[:take] + self._left_native = np.concatenate((self._left_native, consumed))[-context:] + remainder = pending[take:] + self._pending = [remainder] if len(remainder) else [] + self._pending_len = len(remainder) + if final and self._pending_len: + pending = np.concatenate(self._pending) + left = len(self._left_native) + converted = self._resample(np.concatenate((self._left_native, pending))) + offset = round(left * OUTPUT_RATE / self.rate) + wanted = round(len(pending) * OUTPUT_RATE / self.rate) + tail = converted[offset : offset + wanted] + self._append_filtered(np.pad(tail, (0, max(0, wanted - len(tail))))) + self._pending = [] + self._pending_len = 0 + + def _append_filtered(self, samples: np.ndarray) -> None: + if not len(samples): + return + if self._filter_zi is None: + self._filter_zi = signal.sosfilt_zi(self._sos) * float(samples[0]) + filtered, self._filter_zi = signal.sosfilt( + self._sos, samples, zi=self._filter_zi + ) + self._resampled = np.concatenate((self._resampled, filtered.astype(np.float32))) + self._output_total += len(samples) + + def _preview(self) -> np.ndarray: + """Resample the outstanding tail without advancing stream state.""" + if not self._pending_len: + return self._resampled + context = round(self.rate * self._CONTEXT_SECONDS) + pending = np.concatenate(self._pending) + left = len(self._left_native) + source = np.concatenate((self._left_native, pending)) + converted = self._resample(source) + offset = round(left * OUTPUT_RATE / self.rate) + wanted = round(len(pending) * OUTPUT_RATE / self.rate) + tail = converted[offset : offset + wanted] + zi = self._filter_zi + if zi is None and len(tail): + zi = signal.sosfilt_zi(self._sos) * float(tail[0]) + if zi is not None and len(tail): + tail = signal.sosfilt(self._sos, tail, zi=zi)[0] + return np.concatenate((self._resampled, tail.astype(np.float32))) + + def _emit(self, available: int, finishing: bool) -> list[AudioChunk]: + chunks: list[AudioChunk] = [] + while True: + cut = self._next_cut(available) + if cut is None: + break + chunks.append(self._make_chunk(self._start, cut)) + self._start = cut + self._discard_sealed() + # finish() leaves the sub-minimum tail to its caller so it is emitted + # once, after every pause/forced boundary has been handled. + return chunks + + def _next_cut(self, available: int) -> int | None: + length = available - self._start + minimum = int(MIN_SECONDS * OUTPUT_RATE) + maximum = int(MAX_SECONDS * OUTPUT_RATE) + if length < minimum: + return None + if length >= maximum: + search_end = self._start + maximum + else: + search_end = available + + segment = self._resampled[self._start - self._base : search_end - self._base] + levels = _levels(segment) + if not len(levels): + return None + # The lower quartile follows different microphones/noise floors. The + # ceiling prevents ordinary quiet speech from being labelled a pause. + floor = float(np.percentile(levels, 25)) + quiet_below = min(-45.0, floor + 6.0) + quiet = levels <= quiet_below + run = 0 + first_eligible = int(MIN_SECONDS * 50) + for frame_index, is_quiet in enumerate(quiet): + run = run + 1 if is_quiet else 0 + age = (frame_index + 1) / 50.0 + fraction = np.clip( + (age - MIN_SECONDS) / (TARGET_SECONDS - MIN_SECONDS), 0.0, 1.0 + ) + required = EARLY_PAUSE_SECONDS + fraction * ( + LATE_PAUSE_SECONDS - EARLY_PAUSE_SECONDS + ) + required_frames = int(np.ceil(required * 50)) + if frame_index + 1 >= first_eligible and run >= required_frames: + # Cut in the middle of the accepted pause. Both adjacent + # ranges still retain their half, so coverage remains exact. + cut_frame = frame_index + 1 - run // 2 + cut = self._start + max(minimum, cut_frame * FRAME) + logging.info( + "Streaming chunk pause cut at %.2fs (age %.2fs, pause %.2fs, " + "threshold %.1f dBFS)", + cut / OUTPUT_RATE, + age, + run / 50.0, + quiet_below, + ) + return cut + if length >= maximum: + # Prefer the quietest 20 ms frame in the final second when speech + # has been uninterrupted, while strictly respecting the limit. + end_frame = maximum // FRAME + begin_frame = max(first_eligible, end_frame - 50) + chosen = begin_frame + int(np.argmin(levels[begin_frame:end_frame])) + cut = self._start + min((chosen + 1) * FRAME, maximum) + logging.info( + "Streaming chunk forced cut at %.2fs after %.2fs without a pause", + cut / OUTPUT_RATE, + length / OUTPUT_RATE, + ) + return cut + return None + + def _make_chunk(self, start: int, end: int) -> AudioChunk: + raw = self._resampled[start - self._base : end - self._base] + levels = _levels(raw) + # -80 dBFS distinguishes actual/near digital silence while preserving + # quiet but valid speech. Consumers spool false chunks but skip upload. + if len(levels): + speech_level = float(np.max(levels)) + elif len(raw): + centered = raw - float(np.mean(raw)) + speech_level = float(20 * np.log10(max(np.sqrt(np.mean(centered * centered)), 1e-7))) + else: + speech_level = -140.0 + speech = speech_level > -80.0 + floor = float(np.percentile(levels, 10)) if len(levels) else -140.0 + longest = run = 0 + for level in levels: + run = run + 1 if level < -75.0 else 0 + longest = max(longest, run) + logging.info( + "Streaming chunk %.2f-%.2fs speech=%s floor=%.1f dBFS dead=%.2fs", + start / OUTPUT_RATE, + end / OUTPUT_RATE, + speech, + floor, + longest / 50.0, + ) + return AudioChunk(_normalize(raw), start, end, speech) + + def _discard_sealed(self) -> None: + count = self._start - self._base + if count > 0: + self._resampled = self._resampled[count:] + self._base = self._start diff --git a/blurt_client.py b/blurt_client.py new file mode 100644 index 0000000..5bfa4f0 --- /dev/null +++ b/blurt_client.py @@ -0,0 +1,170 @@ +"""UI-free client for Blurt's transcription providers.""" + +from __future__ import annotations + +import base64 +import io +import logging +import threading +from dataclasses import dataclass, field +from typing import Callable + +import numpy as np +import requests +from scipy.io import wavfile + +OPENAI_TRANSCRIPTIONS_URL = "https://api.openai.com/v1/audio/transcriptions" +OPENROUTER_TRANSCRIPTIONS_URL = "https://openrouter.ai/api/v1/audio/transcriptions" +OPENROUTER_APP_URL = "https://git.felixm.de/felixm/blurt" +OPENROUTER_APP_TITLE = "Blurt" +SAMPLE_RATE = 16_000 + +TIMEOUT_CONNECT_SECONDS = 10.0 +TIMEOUT_BASE_SECONDS = 30.0 +TIMEOUT_PER_AUDIO_MINUTE = 30.0 +TIMEOUT_MAX_SECONDS = 300.0 + + +@dataclass(frozen=True) +class TranscriptionConfig: + model: str + language: str + provider: str + api_key: str = field(repr=False) + + +class TranscriptionError(Exception): + """The provider answered but returned no usable transcript.""" + + +def transcription_timeout(seconds_of_audio: float) -> tuple[float, float]: + read = TIMEOUT_BASE_SECONDS + TIMEOUT_PER_AUDIO_MINUTE * (seconds_of_audio / 60.0) + return TIMEOUT_CONNECT_SECONDS, min(read, TIMEOUT_MAX_SECONDS) + + +class TranscriptionClient: + """Encode audio and send requests using one HTTP session per worker thread. + + ``close`` must be called only after callers have stopped their worker pool. + """ + + def __init__(self, session_factory: Callable[[], requests.Session] = requests.Session): + self._session_factory = session_factory + self._local = threading.local() + self._sessions: list[requests.Session] = [] + self._sessions_lock = threading.Lock() + self._closed = False + + def _session(self) -> requests.Session: + if self._closed: + raise RuntimeError("TranscriptionClient is closed") + session = getattr(self._local, "session", None) + if session is not None: + return session + with self._sessions_lock: + if self._closed: + raise RuntimeError("TranscriptionClient is closed") + session = self._session_factory() + self._sessions.append(session) + self._local.session = session + return session + + def transcribe(self, audio: np.ndarray, config: TranscriptionConfig) -> str: + if audio.dtype != np.int16 or audio.ndim != 1: + raise ValueError("audio must be a one-dimensional int16 array at 16 kHz") + if config.provider not in {"openai", "openrouter"}: + raise ValueError(f"Unsupported transcription provider: {config.provider}") + + buffer = io.BytesIO() + wavfile.write(buffer, SAMPLE_RATE, audio) + wav_bytes = buffer.getvalue() + seconds = len(audio) / SAMPLE_RATE + timeout = transcription_timeout(seconds) + logging.info( + "Starting transcription: provider=%s model=%s language=%s audio_seconds=%.1f " + "wav_bytes=%d connect_timeout=%.0fs read_timeout=%.0fs", + config.provider, + config.model, + config.language, + seconds, + len(wav_bytes), + timeout[0], + timeout[1], + ) + + if config.provider == "openrouter": + result = self._transcribe_openrouter(wav_bytes, config, timeout) + else: + result = self._transcribe_openai(wav_bytes, config, timeout) + + text = result.get("text", "").lstrip() + logging.info("Transcription text length: %d", len(text)) + if "usage" in result: + logging.info("Transcription usage: %s", result["usage"]) + if not text: + error = result.get("error") + message = ( + error.get("message", "Unknown error") + if isinstance(error, dict) + else "Unknown error" + ) + logging.error("Transcription response did not include text") + raise TranscriptionError(f"Transcription failed: {message}") + return text + + def _transcribe_openai(self, wav_bytes, config, timeout) -> dict: + data = {"model": config.model, "language": config.language} + if config.model == "gpt-4o-transcribe-diarize": + data.update(response_format="diarized_json", chunking_strategy="auto") + response = self._session().post( + OPENAI_TRANSCRIPTIONS_URL, + headers={"Authorization": f"Bearer {config.api_key}"}, + files={"file": ("blurt.wav", wav_bytes, "audio/wav")}, + data=data, + timeout=timeout, + ) + self._log_response(response) + response.raise_for_status() + return response.json() + + def _transcribe_openrouter(self, wav_bytes, config, timeout) -> dict: + response = self._session().post( + OPENROUTER_TRANSCRIPTIONS_URL, + headers={ + "Authorization": f"Bearer {config.api_key}", + "HTTP-Referer": OPENROUTER_APP_URL, + "X-OpenRouter-Title": OPENROUTER_APP_TITLE, + }, + json={ + "model": config.model, + "input_audio": { + "data": base64.b64encode(wav_bytes).decode("ascii"), + "format": "wav", + }, + "language": config.language, + }, + timeout=timeout, + ) + self._log_response(response) + response.raise_for_status() + return response.json() + + @staticmethod + def _log_response(response: requests.Response) -> None: + logging.info( + "Transcription response: status=%s content_type=%s bytes=%d", + response.status_code, + response.headers.get("content-type", ""), + len(response.content), + ) + if not response.ok: + logging.error("Transcription error response: %s", response.text[:1000]) + + def close(self) -> None: + with self._sessions_lock: + if self._closed: + return + self._closed = True + sessions, self._sessions = self._sessions, [] + for session in sessions: + session.close() diff --git a/blurt_session.py b/blurt_session.py new file mode 100644 index 0000000..ba1f63a --- /dev/null +++ b/blurt_session.py @@ -0,0 +1,591 @@ +"""Concurrent, crash-recoverable recording and transcription sessions.""" + +from __future__ import annotations + +import json +import logging +import queue +import shutil +import threading +import time +import uuid +import wave +from collections import deque +from dataclasses import asdict, dataclass, is_dataclass, replace +from pathlib import Path +from types import MappingProxyType +from typing import Any, Callable, Mapping + +import numpy as np + +from blurt_audio import StreamingAudio + + +_ACTIVE_PATHS: set[Path] = set() +_ACTIVE_LOCK = threading.Lock() + + +def _atomic_json(path: Path, value: dict[str, Any]) -> None: + temporary = path.with_suffix(path.suffix + ".tmp") + temporary.write_text(json.dumps(value, indent=2), encoding="utf-8") + temporary.replace(path) + + +def _write_wav(path: Path, samples: np.ndarray) -> None: + temporary = path.with_suffix(path.suffix + ".tmp") + with wave.open(str(temporary), "wb") as output: + output.setnchannels(1) + output.setsampwidth(2) + output.setframerate(16000) + output.writeframes(np.asarray(samples, dtype=" np.ndarray: + with wave.open(str(path), "rb") as source: + if source.getnchannels() != 1 or source.getsampwidth() != 2: + raise ValueError(f"unsupported recovery WAV format: {path}") + data = source.readframes(source.getnframes()) + rate = source.getframerate() + samples = np.frombuffer(data, dtype=" dict[str, Any]: + if is_dataclass(config): + return asdict(config) + if isinstance(config, Mapping): + return dict(config) + return dict(vars(config)) + + +def _public_config(config: Any) -> dict[str, Any]: + blocked = ("key", "token", "secret", "password", "auth", "credential") + return { + str(k): v + for k, v in _config_dict(config).items() + if not any(word in str(k).lower() for word in blocked) + and isinstance(v, (str, int, float, bool, type(None))) + } + + +@dataclass(frozen=True) +class SessionResult: + session_id: str + text: str + complete: bool + errors: tuple[str, ...] = () + + +@dataclass(frozen=True) +class RecoveryRecord: + session_id: str + path: Path + state: str + config: Mapping[str, Any] + done: int + total: int + created_at: float = 0.0 + + +@dataclass +class _Job: + session: "RecordingSession" + index: int + path: Path + config: Any + + +class TranscriptionScheduler: + """A shared worker pool whose pending payloads live as WAV files on disk.""" + + def __init__(self, request: Callable[[np.ndarray, Any], str], workers: int = 3, + queue_size: int = 96): + if workers < 1: + raise ValueError("workers must be positive") + self.request = request + self._ready: queue.Queue[_Job | None] = queue.Queue(maxsize=max(workers, queue_size)) + self._overflow: deque[_Job] = deque() + self._lock = threading.Lock() + self._closed = False + self._threads = [threading.Thread(target=self._worker, name=f"blurt-upload-{i}", daemon=True) + for i in range(workers)] + for thread in self._threads: + thread.start() + + def submit(self, session: "RecordingSession", index: int, path: Path, + config: Any) -> None: + job = _Job(session, index, path, config) + with self._lock: + if self._closed: + raise RuntimeError("transcription scheduler is shut down") + try: + self._ready.put_nowait(job) + except queue.Full: + # Only tiny disk references are retained here; audio remains on disk. + self._overflow.append(job) + + def _refill(self) -> None: + with self._lock: + while self._overflow: + try: + self._ready.put_nowait(self._overflow[0]) + except queue.Full: + break + self._overflow.popleft() + + def _worker(self) -> None: + while True: + try: + job = self._ready.get(timeout=0.1) + except queue.Empty: + with self._lock: + if self._closed and not self._overflow: + return + continue + text = None + error = None + try: + samples = _read_wav(job.path) + text = self.request(samples, job.config) + if not isinstance(text, str) or not text.strip(): + raise ValueError("transcription returned empty text") + text = text.strip() + except Exception as exception: + error = f"{type(exception).__name__}: {exception}" + try: + job.session._job_finished(job.index, text, error) + except Exception: + # Session persistence/callback failures must not terminate the pool. + logging.exception("Could not record transcription result for %s", job.path) + finally: + self._ready.task_done() + self._refill() + + def shutdown(self, wait: bool = True) -> None: + with self._lock: + self._closed = True + if wait: + for thread in self._threads: + thread.join() + + +class RecordingSession: + """Owns capture, chunk persistence, ordered results, and one recording lifecycle.""" + + def __init__(self, rate: int, channels: int, config: Any, spool_dir: str | Path, + scheduler: TranscriptionScheduler, on_update: Callable[["RecordingSession"], None] | None = None, + on_complete: Callable[["RecordingSession"], None] | None = None, + input_queue_size: int = 64): + self.session_id = uuid.uuid4().hex + self.path = Path(spool_dir) / self.session_id + self.path.mkdir(parents=True, exist_ok=False) + self.config = config + self.scheduler = scheduler + self.on_update = on_update + self.on_complete = on_complete + self.state = "recording" + self.done = 0 + self.total = 0 + self.result: SessionResult | None = None + self.created_at = time.time() + self._entries: list[dict[str, Any]] = [] + self._submitted: set[int] = set() + self._capture_persisted_ok = False + self._lock = threading.RLock() + self._condition = threading.Condition(self._lock) + self._input: queue.Queue[np.ndarray] = queue.Queue(maxsize=input_queue_size) + self._stop = threading.Event() + self._producer_done = threading.Event() + self._overflow_error = threading.Event() + self._input_statuses: queue.SimpleQueue[str] = queue.SimpleQueue() + self._audio = StreamingAudio(rate, channels) + self._last_checkpoint = time.monotonic() + self._tail_start = 0 + self._manifest_path = self.path / "session.json" + with _ACTIVE_LOCK: + _ACTIVE_PATHS.add(self.path.resolve()) + try: + self._save() + except Exception: + self.release_for_retry() + raise + self._producer = threading.Thread(target=self._produce, name=f"blurt-capture-{self.session_id[:6]}", daemon=True) + self._producer.start() + + def feed(self, indata: np.ndarray) -> bool: + if self.state != "recording": + return False + try: + self._input.put_nowait(np.array(indata, copy=True)) + return True + except queue.Full: + # The sound callback must never wait for a lock or filesystem. + self.state = "capture_failed" + self._overflow_error.set() + self._stop.set() + return False + + def stop(self) -> None: + with self._lock: + if self.state == "recording": + self.state = "finishing" + self._stop.set() + self._save() + + def report_input_status(self, status: object) -> None: + """Record a PortAudio status without blocking its callback thread.""" + if status: + self._input_statuses.put(str(status)) + + def producer_join(self, timeout: float | None = None) -> bool: + self._producer.join(timeout) + return not self._producer.is_alive() + + def join(self, timeout: float | None = None) -> SessionResult | None: + deadline = None if timeout is None else time.monotonic() + timeout + with self._condition: + while self.result is None: + remaining = None if deadline is None else deadline - time.monotonic() + if remaining is not None and remaining <= 0: + return None + self._condition.wait(remaining) + return self.result + + def _produce(self) -> None: + try: + while True: + try: + block = self._input.get(timeout=0.05) + except queue.Empty: + if self._stop.is_set(): + break + continue + try: + for chunk in self._audio.feed(block): + self._seal(chunk) + if time.monotonic() - self._last_checkpoint >= 1.0: + self._checkpoint() + finally: + self._input.task_done() + for chunk in self._audio.finish(): + self._seal(chunk) + self._remove_tail() + if self._overflow_error.is_set(): + with self._lock: + self._entries.append({"index": len(self._entries), "speech": False, + "status": "failed", "error": "capture input queue overflowed; audio was lost"}) + self._save() + self._persist_input_statuses() + self._capture_persisted_ok = True + except Exception as error: + with self._lock: + self.state = "capture_failed" + for entry in self._entries: + if (entry.get("speech") and entry.get("status") == "pending" + and int(entry["index"]) not in self._submitted): + entry["status"] = "failed" + entry["error"] = f"capture: {type(error).__name__}: {error}" + self.done += 1 + self._entries.append({"index": len(self._entries), "speech": False, + "status": "failed", + "error": f"capture: {type(error).__name__}: {error}"}) + try: + self._save() + except OSError: + pass + finally: + self._producer_done.set() + self._maybe_complete() + + def _persist_input_statuses(self) -> None: + statuses = [] + while not self._input_statuses.empty(): + statuses.append(self._input_statuses.get()) + if statuses: + with self._lock: + self._entries.append({"index": len(self._entries), "speech": False, + "status": "failed", "error": "input stream status: " + ", ".join(statuses)}) + self._save() + + @staticmethod + def _chunk_value(chunk: Any, name: str, default: Any = None) -> Any: + return getattr(chunk, name, default) + + def _seal(self, chunk: Any) -> None: + samples = np.asarray(self._chunk_value(chunk, "samples"), dtype=np.int16) + speech = bool(self._chunk_value(chunk, "speech", True)) + with self._lock: + index = len(self._entries) + filename = f"chunk-{index:06d}.wav" + entry = {"index": index, "file": filename, "speech": speech, + "start": self._chunk_value(chunk, "start", None), + "end": self._chunk_value(chunk, "end", None), "status": "pending"} + if not speech: + entry["status"] = "skipped" + entry["text"] = "" + self._entries.append(entry) + self.total += int(speech) + _write_wav(self.path / filename, samples) + # Commit the sealed file before removing the previous recovery tail. + self._save() + self._remove_tail_locked() + self._tail_start = int(entry.get("end") or self._tail_start) + self._save() + if speech: + self.scheduler.submit(self, index, self.path / filename, self.config) + with self._lock: + self._submitted.add(index) + self._notify() + + def _checkpoint(self) -> None: + tail = np.asarray(self._audio.checkpoint(), dtype=np.int16) + with self._lock: + if tail.size: + _write_wav(self.path / "tail.wav", tail) + ends = [int(e["end"]) for e in self._entries if e.get("end") is not None] + self._tail_start = max(ends, default=0) + else: + self._remove_tail_locked() + self._last_checkpoint = time.monotonic() + self._save() + + def _remove_tail_locked(self) -> None: + (self.path / "tail.wav").unlink(missing_ok=True) + + def _remove_tail(self) -> None: + with self._lock: + self._remove_tail_locked() + self._save() + + def _job_finished(self, index: int, text: str | None, error: str | None) -> None: + with self._lock: + entry = self._entries[index] + entry["status"] = "done" if error is None else "failed" + if text is not None: + entry["text"] = text + if error is not None: + entry["error"] = error + self.done += 1 + try: + self._save() + except OSError as persistence_error: + entry.pop("text", None) + entry["status"] = "failed" + entry["error"] = ( + f"could not persist transcription: {type(persistence_error).__name__}: " + f"{persistence_error}" + ) + self._notify() + self._maybe_complete() + + def _maybe_complete(self) -> None: + callback = None + with self._condition: + if not self._producer_done.is_set() or self.done < self.total or self.result is not None: + return + errors = tuple(str(e["error"]) for e in self._entries if e.get("error")) + has_text = any(entry.get("text") for entry in self._entries) + parts = [] + gap = False + for entry in self._entries: + if entry.get("text"): + if gap and has_text: + parts.append("[…]") + gap = False + parts.append(str(entry["text"])) + elif entry.get("error"): + gap = True + if gap and has_text: + parts.append("[…]") + text = " ".join(parts) + complete = not errors + self.state = "complete" if complete else "needs_retry" + self.result = SessionResult(self.session_id, text, complete, errors) + try: + self._save() + except OSError: + complete = False + self.state = "needs_retry" + errors = errors + ("could not persist final session state",) + self.result = SessionResult(self.session_id, text, False, errors) + self._condition.notify_all() + callback = self.on_complete + if callback: + try: + callback(self) + except Exception: + logging.exception("Session completion callback failed: %s", self.session_id) + self.release_for_retry() + + def _notify(self) -> None: + if self.on_update: + try: + self.on_update(self) + except Exception: + logging.exception("Session progress callback failed: %s", self.session_id) + + def _save(self) -> None: + _atomic_json(self._manifest_path, {"version": 1, "session_id": self.session_id, + "state": self.state, "config": _public_config(self.config), "done": self.done, + "total": self.total, "created_at": self.created_at, "entries": self._entries, + "has_tail": (self.path / "tail.wav").exists(), "tail_start": self._tail_start}) + + def acknowledge_delivery(self) -> None: + with self._lock: + if self.result is None: + raise RuntimeError("session is not finished") + if self.result.complete: + shutil.rmtree(self.path) + else: + self.state = "needs_retry" + self._save() + with _ACTIVE_LOCK: + _ACTIVE_PATHS.discard(self.path.resolve()) + + def release_for_retry(self) -> None: + """Make retained data discoverable after delivery is deferred or fails.""" + with _ACTIVE_LOCK: + _ACTIVE_PATHS.discard(self.path.resolve()) + + @classmethod + def recover(cls, path: str | Path, config: Any, scheduler: TranscriptionScheduler, + on_update=None, on_complete=None) -> "RecordingSession": + return _recover_session(Path(path), config, scheduler, on_update, on_complete) + + @classmethod + def from_wav(cls, wav_path: str | Path, config: Any, spool_dir: str | Path, + scheduler: TranscriptionScheduler, on_update=None, on_complete=None) -> "RecordingSession": + samples = _read_wav(Path(wav_path)) + session = cls(16000, 1, config, spool_dir, scheduler, on_update, on_complete) + session.feed(samples.reshape(-1, 1).astype(np.float32) / 32768.0) + session.stop() + session.producer_join() + if session._capture_persisted_ok: + Path(wav_path).unlink(missing_ok=True) + return session + + +def _recover_session(path: Path, config: Any, scheduler: TranscriptionScheduler, + on_update=None, on_complete=None) -> RecordingSession: + data = json.loads((path / "session.json").read_text(encoding="utf-8")) + session = object.__new__(RecordingSession) + session.session_id, session.path = data["session_id"], path + saved = data.get("config", {}) + if is_dataclass(config): + fields = getattr(config, "__dataclass_fields__", {}) + session.config = replace(config, **{k: v for k, v in saved.items() if k in fields}) + else: + merged = dict(saved); merged.update(_config_dict(config)) + session.config = MappingProxyType(merged) + session.scheduler, session.on_update, session.on_complete = scheduler, on_update, on_complete + session.state, session.done, session.total = "finishing", 0, 0 + session.result = None + session.created_at = float(data.get("created_at", path.stat().st_mtime)) + session._entries = data.get("entries", []) + session._submitted = set() + session._capture_persisted_ok = True + session._lock = threading.RLock(); session._condition = threading.Condition(session._lock) + session._input = queue.Queue(); session._stop = threading.Event(); session._stop.set() + session._producer_done = threading.Event(); session._producer_done.set() + session._overflow_error = threading.Event() + session._input_statuses = queue.SimpleQueue() + session._audio = None; session._last_checkpoint = time.monotonic() + session._tail_start = int(data.get("tail_start", 0)) + session._manifest_path = path / "session.json" + referenced = {entry.get("file") for entry in session._entries} + orphans = [candidate for candidate in sorted(path.glob("chunk-*.wav")) + if candidate.name not in referenced] + sealed_end = max((int(e.get("end") or 0) for e in session._entries), default=0) + for orphan in orphans: + samples = _read_wav(orphan) + index = len(session._entries) + speech = bool(samples.size and np.max(np.abs(samples.astype(np.int32))) > 1) + session._entries.append({"index": index, "file": orphan.name, "speech": speech, + "status": "pending" if speech else "skipped", + "start": sealed_end, "end": sealed_end + len(samples), + "recovered_orphan": True}) + sealed_end += len(samples) + tail = path / "tail.wav" + if tail.exists(): + tail_samples = _read_wav(tail) + overlap = max(0, sealed_end - session._tail_start) + if overlap: + tail_samples = tail_samples[min(overlap, len(tail_samples)):] + index = len(session._entries) + target = path / f"chunk-{index:06d}.wav" + if len(tail_samples): + _write_wav(target, tail_samples) + speech = bool(np.max(np.abs(tail_samples.astype(np.int32))) > 1) + session._entries.append({"index": index, "file": target.name, "speech": speech, + "status": "pending" if speech else "skipped", + "start": sealed_end, "end": sealed_end + len(tail_samples), + "recovered_tail": True}) + pending = [] + for entry in session._entries: + if entry.get("speech") and entry.get("status") != "done" and entry.get("file") and (path / entry["file"]).exists(): + entry["status"] = "pending"; entry.pop("error", None) + pending.append(entry) + elif entry.get("speech") and entry.get("status") != "done": + entry["status"] = "failed" + entry["error"] = "recovery audio file is missing" + session.done += 1 + elif entry.get("speech"): + session.done += 1 + session.total = sum(bool(e.get("speech")) for e in session._entries) + with _ACTIVE_LOCK: + _ACTIVE_PATHS.add(path.resolve()) + # Commit adopted orphan/suffix entries before deleting the original tail. + try: + session._save() + tail.unlink(missing_ok=True) + session._save() + except Exception: + session.release_for_retry() + raise + + def start_recovery() -> None: + for entry in pending: + try: + scheduler.submit(session, entry["index"], path / entry["file"], session.config) + session._submitted.add(int(entry["index"])) + except Exception as error: + session._job_finished(entry["index"], None, f"Could not schedule retry: {error}") + session._maybe_complete() + + session._producer = threading.Thread(target=start_recovery, name="BlurtRecovery", daemon=True) + session._producer.start() + return session + + +class RecoveryStore: + def __init__(self, spool_dir: str | Path): + self.spool_dir = Path(spool_dir) + + def discover(self) -> list[RecoveryRecord]: + records = [] + if not self.spool_dir.exists(): + return records + for manifest in sorted(self.spool_dir.glob("*/session.json")): + try: + with _ACTIVE_LOCK: + if manifest.parent.resolve() in _ACTIVE_PATHS: + continue + data = json.loads(manifest.read_text(encoding="utf-8")) + records.append(RecoveryRecord(data["session_id"], manifest.parent, + data.get("state", "interrupted"), MappingProxyType(data.get("config", {})), + int(data.get("done", 0)), int(data.get("total", 0)), + float(data.get("created_at", manifest.stat().st_mtime)))) + except (OSError, ValueError, KeyError): + continue + return sorted(records, key=lambda record: record.created_at) + + def resume(self, record: RecoveryRecord | str, config: Any, + scheduler: TranscriptionScheduler, on_update=None, on_complete=None) -> RecordingSession: + if isinstance(record, str): + path = self.spool_dir / record + else: + path = record.path + return RecordingSession.recover(path, config, scheduler, on_update, on_complete) diff --git a/export.py b/export.py index 94923e5..31f3544 100644 --- a/export.py +++ b/export.py @@ -24,7 +24,7 @@ ARCHIVE_ROOT = "blurt" # top-level folder inside the zip # swept in, even though they sit right next to the source. EXCLUDE_DIRS = { ".git", ".venv", "__pycache__", ".claude", - "dist", "build", ".ruff_cache", ".pytest_cache", + "dist", "build", ".ruff_cache", ".pytest_cache", "spool", "captures", } EXCLUDE_NAMES = {"api_key.txt", "openrouter_key.txt"} EXCLUDE_SUFFIXES = {".pyc", ".pyo"} diff --git a/pyproject.toml b/pyproject.toml index ccb661b..3ce1cf2 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -22,3 +22,6 @@ build-backend = "hatchling.build" [project.scripts] blurt = "blurt:main" + +[tool.hatch.build.targets.wheel] +only-include = ["blurt.py", "blurt_audio.py", "blurt_client.py", "blurt_session.py"] diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/test_app_streaming.py b/tests/test_app_streaming.py new file mode 100644 index 0000000..b8e3e73 --- /dev/null +++ b/tests/test_app_streaming.py @@ -0,0 +1,162 @@ +"""Tray integration without a microphone, display, network, or clipboard.""" +import os +os.environ.setdefault("PYSTRAY_BACKEND", "dummy") + +import threading +import unittest +from types import SimpleNamespace +from unittest.mock import Mock, patch + +import blurt + + +class StreamingAppTests(unittest.TestCase): + def app(self): + app = blurt.BlurtApp.__new__(blurt.BlurtApp) + app._live_lock = threading.RLock() + app._delivery_lock = threading.Lock() + app.lock = threading.Lock() + app._live_sessions = [] + app._quitting = False + app._recording_session = None + app._update_icon = Mock() + app.platform = SimpleNamespace(notify=Mock(), dispatch=Mock()) + app.transcription_model = "whisper-1" + app.transcription_language = "en" + app.api_keys = {"openai": "test-key", "openrouter": "router-key"} + return app + + def session(self, text, complete=True): + return SimpleNamespace( + result=SimpleNamespace(session_id="test", text=text, complete=complete, + errors=[] if complete else ["failed"]), + acknowledge_delivery=Mock(), + release_for_retry=Mock(), + ) + + @patch("blurt.pyperclip.copy") + def test_results_delivered_in_recording_order(self, copy): + app = self.app() + first = self.session("first") + final_result = first.result + first.result = None + second = self.session("second") + app._live_sessions = [first, second] + app._session_complete(second) + copy.assert_not_called() + first.result = final_result + app._session_complete(first) + self.assertEqual([c.args[0] for c in copy.call_args_list], ["first", "second"]) + self.assertEqual(app._live_sessions, []) + + @patch("blurt.pyperclip.copy") + def test_partial_agent_command_never_dispatched(self, copy): + app = self.app() + active = self.session("to my agent change […]", complete=False) + app._live_sessions = [active] + app._session_complete(active) + copy.assert_called_once_with("to my agent change […]") + app.platform.dispatch.assert_not_called() + + @patch("blurt.pyperclip.copy", side_effect=RuntimeError("clipboard unavailable")) + def test_clipboard_failure_does_not_acknowledge(self, copy): + app = self.app() + active = self.session("keep these words") + app._live_sessions = [active] + app._session_complete(active) + active.acknowledge_delivery.assert_not_called() + active.release_for_retry.assert_called_once() + + @patch("blurt.pyperclip.copy") + def test_capture_failure_is_not_delivered_until_microphone_stops(self, copy): + app = self.app() + active = self.session("some words […]", complete=False) + active.stop = Mock() + app._recording_session = active + app._live_sessions = [active] + app.stream = Mock() + app.recording = True + app._session_complete(active) + copy.assert_not_called() + app._stop_streaming() + copy.assert_called_once_with("some words […]") + + @patch("blurt.pyperclip.copy") + def test_quitting_keeps_completed_recording_for_recovery(self, copy): + app = self.app() + app._quitting = True + active = self.session("keep these words") + app._live_sessions = [active] + app._session_complete(active) + copy.assert_not_called() + active.acknowledge_delivery.assert_not_called() + + def test_settings_snapshot_survives_menu_changes(self): + app = self.app() + config = app._transcription_config() + app.transcription_model = "another-model" + app.transcription_language = "de" + app.api_keys["openai"] = "replacement" + self.assertEqual((config.model, config.language, config.api_key), + ("whisper-1", "en", "test-key")) + + @patch("blurt.pyperclip.copy") + def test_retries_collect_oldest_first_and_never_dispatch(self, copy): + app = self.app() + app._retrying = True + app._scheduler = Mock() + app._retryable_files = Mock(return_value=[]) + app._recovery = Mock() + newer = SimpleNamespace(created_at=20, config={"language": "de"}) + older = SimpleNamespace(created_at=10, config={"language": "en"}) + app._recovery.discover.return_value = [newer, older] + first = self.session("to my agent first") + second = self.session("second") + first.join = Mock(return_value=first.result) + second.join = Mock(return_value=second.result) + app._recovery.resume.side_effect = [first, second] + app._retry_sessions() + copy.assert_called_once_with("to my agent first\n\nsecond") + app.platform.dispatch.assert_not_called() + self.assertIs(app._recovery.resume.call_args_list[0].args[0], older) + first.acknowledge_delivery.assert_called_once() + second.acknowledge_delivery.assert_called_once() + self.assertFalse(app._retrying) + + @patch("blurt.pyperclip.copy", side_effect=RuntimeError("unavailable")) + def test_retry_clipboard_failure_releases_records_without_deleting(self, copy): + app = self.app() + app._retrying = True + app._scheduler = Mock() + app._retryable_files = Mock(return_value=[]) + app._recovery = Mock() + app._recovery.discover.return_value = [SimpleNamespace(created_at=10, config={})] + active = self.session("recovered words") + active.join = Mock(return_value=active.result) + app._recovery.resume.return_value = active + with self.assertLogs(level="ERROR"): + app._retry_sessions() + active.acknowledge_delivery.assert_not_called() + active.release_for_retry.assert_called_once() + + def test_callback_routes_to_owning_session_and_stop_drains_first(self): + app = self.app() + order = [] + active = Mock() + active.stop.side_effect = lambda: order.append("session-eof") + app._recording_session = active + app.recording = True + app.stream = Mock() + app.stream.stop.side_effect = lambda: order.append("stream-stop") + app.stream.close.side_effect = lambda: order.append("stream-close") + frames = object() + app._audio_callback(frames, 0, None, None) + active.feed.assert_called_once_with(frames) + app._stop_streaming() + self.assertEqual(order, ["stream-stop", "stream-close", "session-eof"]) + self.assertIsNone(app._recording_session) + self.assertFalse(app.recording) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_client.py b/tests/test_client.py new file mode 100644 index 0000000..33f29f4 --- /dev/null +++ b/tests/test_client.py @@ -0,0 +1,134 @@ +import base64 +import threading +import unittest +from dataclasses import FrozenInstanceError + +import numpy as np + +from blurt_client import ( + OPENAI_TRANSCRIPTIONS_URL, + OPENROUTER_TRANSCRIPTIONS_URL, + TranscriptionClient, + TranscriptionConfig, + TranscriptionError, + transcription_timeout, +) + + +class FakeResponse: + status_code = 200 + headers = {"content-type": "application/json"} + content = b"{}" + text = "{}" + ok = True + + def __init__(self, result=None): + self.result = {"text": " transcript"} if result is None else result + + def raise_for_status(self): + return None + + def json(self): + return self.result + + +class FakeSession: + def __init__(self, result=None): + self.calls = [] + self.response = FakeResponse(result) + self.closed = False + + def post(self, url, **kwargs): + self.calls.append((url, kwargs)) + return self.response + + def close(self): + self.closed = True + + +class ClientTests(unittest.TestCase): + def setUp(self): + self.sessions = [] + + def factory(): + session = FakeSession() + self.sessions.append(session) + return session + + self.client = TranscriptionClient(factory) + self.audio = np.zeros(32_000, dtype=np.int16) + + def tearDown(self): + self.client.close() + + def test_config_is_frozen_and_hides_key(self): + config = TranscriptionConfig("model", "en", "openai", "secret") + self.assertNotIn("secret", repr(config)) + with self.assertRaises(FrozenInstanceError): + config.model = "changed" + + def test_openai_multipart_payload_and_diarization(self): + config = TranscriptionConfig("gpt-4o-transcribe-diarize", "de", "openai", "key") + self.assertEqual(self.client.transcribe(self.audio, config), "transcript") + url, call = self.sessions[0].calls[0] + self.assertEqual(url, OPENAI_TRANSCRIPTIONS_URL) + self.assertEqual(call["headers"], {"Authorization": "Bearer key"}) + self.assertEqual(call["files"]["file"][0::2], ("blurt.wav", "audio/wav")) + self.assertTrue(call["files"]["file"][1].startswith(b"RIFF")) + self.assertEqual(call["data"], { + "model": config.model, "language": "de", + "response_format": "diarized_json", "chunking_strategy": "auto", + }) + self.assertEqual(call["timeout"], (10.0, 31.0)) + + def test_openrouter_base64_payload_and_headers(self): + config = TranscriptionConfig("openai/whisper-large-v3", "en", "openrouter", "router-key") + self.client.transcribe(self.audio, config) + url, call = self.sessions[0].calls[0] + self.assertEqual(url, OPENROUTER_TRANSCRIPTIONS_URL) + self.assertEqual(call["headers"]["Authorization"], "Bearer router-key") + self.assertEqual(call["headers"]["X-OpenRouter-Title"], "Blurt") + payload = call["json"] + self.assertEqual(payload["model"], config.model) + self.assertEqual(payload["language"], "en") + self.assertEqual(payload["input_audio"]["format"], "wav") + self.assertTrue(base64.b64decode(payload["input_audio"]["data"]).startswith(b"RIFF")) + + def test_timeout_scales_and_caps(self): + self.assertEqual(transcription_timeout(0), (10.0, 30.0)) + self.assertEqual(transcription_timeout(60), (10.0, 60.0)) + self.assertEqual(transcription_timeout(10_000), (10.0, 300.0)) + + def test_thread_local_sessions_are_reused_and_closed(self): + config = TranscriptionConfig("whisper-1", "en", "openai", "key") + self.client.transcribe(self.audio, config) + self.client.transcribe(self.audio, config) + worker = threading.Thread(target=self.client.transcribe, args=(self.audio, config)) + worker.start() + worker.join() + self.assertEqual(len(self.sessions), 2) + self.assertEqual(len(self.sessions[0].calls), 2) + self.client.close() + self.assertTrue(all(session.closed for session in self.sessions)) + with self.assertRaises(RuntimeError): + self.client.transcribe(self.audio, config) + + def test_empty_response_raises_provider_message(self): + session = FakeSession({"error": {"message": "bad audio"}}) + client = TranscriptionClient(lambda: session) + with self.assertRaisesRegex(TranscriptionError, "bad audio"): + client.transcribe( + self.audio, TranscriptionConfig("whisper-1", "en", "openai", "key") + ) + client.close() + + def test_rejects_wrong_audio_shape_or_dtype(self): + config = TranscriptionConfig("whisper-1", "en", "openai", "key") + with self.assertRaises(ValueError): + self.client.transcribe(np.zeros(10, dtype=np.float32), config) + with self.assertRaises(ValueError): + self.client.transcribe(np.zeros((10, 1), dtype=np.int16), config) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_pipeline.py b/tests/test_pipeline.py new file mode 100644 index 0000000..a81291c --- /dev/null +++ b/tests/test_pipeline.py @@ -0,0 +1,54 @@ +"""Exercise real audio/session integration with deterministic provider responses.""" +import tempfile +import threading +import unittest + +import numpy as np + +from blurt_client import TranscriptionConfig +from blurt_session import RecordingSession, TranscriptionScheduler + + +class PipelineTests(unittest.TestCase): + def test_real_audio_uploads_before_stop_and_seals_final_speech(self): + with tempfile.TemporaryDirectory() as directory: + requested = threading.Event() + calls = [] + completions = [] + + def transcribe(audio, config): + calls.append((len(audio), config)) + requested.set() + return f"phrase {len(calls)}" + + scheduler = TranscriptionScheduler(transcribe) + config = TranscriptionConfig("test", "en", "openai", "fake") + session = RecordingSession(48000, 1, config, directory, scheduler, + on_complete=completions.append) + try: + tone = (0.05 * np.sin(2 * np.pi * 220 * np.arange(48000) / 48000)).astype(np.float32) + for _ in range(9): + self.assertTrue(session.feed(tone[:, None])) + for _ in range(2): + self.assertTrue(session.feed(np.zeros((48000, 1), dtype=np.float32))) + self.assertTrue(requested.wait(5), "real pause detector did not upload during capture") + self.assertIsNone(session.result) + self.assertEqual(completions, []) + self.assertTrue(session.feed(tone[:, None])) + session.stop() + result = session.join(5) + self.assertIsNotNone(result) + self.assertTrue(result.complete, result.errors) + self.assertEqual(result.text, "phrase 1 phrase 2") + self.assertEqual(sum(length for length, _ in calls), 12 * 16000) + self.assertTrue(all(snapshot is config for _, snapshot in calls)) + session.acknowledge_delivery() + self.assertFalse(session.path.exists()) + finally: + session.stop() + session.join(5) + scheduler.shutdown() + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_sessions.py b/tests/test_sessions.py new file mode 100644 index 0000000..22df17c --- /dev/null +++ b/tests/test_sessions.py @@ -0,0 +1,344 @@ +import json +import tempfile +import threading +import time +import unittest +from dataclasses import dataclass +from pathlib import Path +from unittest.mock import patch + +import numpy as np + +import blurt_session + + +@dataclass(frozen=True) +class Config: + model: str = "test-model" + language: str = "en" + api_key: str = "do-not-store" + + +@dataclass +class Chunk: + samples: np.ndarray + start: int + end: int + speech: bool = True + + +class FakeAudio: + def __init__(self, rate, channels): + self.position = 0 + self.tail = np.empty(0, dtype=np.int16) + + def feed(self, frames): + value = int(np.asarray(frames).flat[0]) + samples = np.full(100, value, dtype=np.int16) + start = self.position + self.position += len(samples) + return [Chunk(samples, start, self.position, value != 0)] + + def finish(self): + return [] + + def checkpoint(self): + return self.tail.copy() + + +class SessionTests(unittest.TestCase): + def setUp(self): + self.temp = tempfile.TemporaryDirectory() + self.addCleanup(self.temp.cleanup) + + def scheduler(self, request, workers=3): + scheduler = blurt_session.TranscriptionScheduler(request, workers=workers) + self.addCleanup(scheduler.shutdown) + return scheduler + + @patch.object(blurt_session, "StreamingAudio", FakeAudio) + def test_transcribes_before_stop_and_orders_out_of_order_results(self): + first_started = threading.Event() + release_first = threading.Event() + + def request(audio, config): + value = int(audio[0]) + if value == 1: + first_started.set() + release_first.wait(2) + return f"part {value}" + + completed = [] + scheduler = self.scheduler(request) + session = blurt_session.RecordingSession(16000, 1, Config(), self.temp.name, + scheduler, on_complete=completed.append) + self.assertTrue(session.feed(np.array([[1]], dtype=np.float32))) + self.assertTrue(first_started.wait(1), "upload did not begin while recording") + self.assertTrue(session.feed(np.array([[2]], dtype=np.float32))) + deadline = time.monotonic() + 1 + while session.done < 1 and time.monotonic() < deadline: + time.sleep(0.005) + self.assertEqual(session.state, "recording") + self.assertEqual(completed, [], "must not deliver during recording") + session.stop() + release_first.set() + result = session.join(2) + self.assertIsNotNone(result) + self.assertEqual(result.text, "part 1 part 2") + self.assertTrue(result.complete) + self.assertEqual(completed, [session]) + + @patch.object(blurt_session, "StreamingAudio", FakeAudio) + def test_failure_is_recoverable_and_partial_ack_keeps_spool(self): + scheduler = self.scheduler(lambda audio, config: (_ for _ in ()).throw(RuntimeError("offline"))) + session = blurt_session.RecordingSession(16000, 1, Config(), self.temp.name, scheduler) + session.feed(np.array([[7]], dtype=np.float32)) + session.stop() + result = session.join(2) + self.assertFalse(result.complete) + self.assertIn("offline", result.errors[0]) + self.assertEqual(result.text, "") + session.acknowledge_delivery() + self.assertTrue(session.path.exists()) + manifest = json.loads((session.path / "session.json").read_text()) + self.assertNotIn("api_key", manifest["config"]) + self.assertEqual(manifest["config"]["model"], "test-model") + + @patch.object(blurt_session, "StreamingAudio", FakeAudio) + def test_recovery_retries_failed_audio_and_preserves_successful_text(self): + calls = [] + + def initial(audio, config): + value = int(audio[0]) + if value == 2: + raise OSError("temporary") + return "one" + + first_scheduler = blurt_session.TranscriptionScheduler(initial, workers=2) + session = blurt_session.RecordingSession(16000, 1, Config(), self.temp.name, first_scheduler) + session.feed(np.array([[1]], dtype=np.float32)) + session.feed(np.array([[2]], dtype=np.float32)) + session.stop() + self.assertFalse(session.join(2).complete) + session.release_for_retry() + first_scheduler.shutdown() + + def recovered(audio, config): + calls.append(int(audio[0])) + self.assertEqual(config.language, "en") + self.assertEqual(config.api_key, "new-key") + return "two" + + second_scheduler = self.scheduler(recovered) + records = blurt_session.RecoveryStore(self.temp.name).discover() + self.assertEqual(len(records), 1) + resumed = blurt_session.RecoveryStore(self.temp.name).resume( + records[0], Config(api_key="new-key"), second_scheduler) + result = resumed.join(2) + self.assertTrue(result.complete) + self.assertEqual(result.text, "one two") + self.assertEqual(calls, [2]) + resumed.acknowledge_delivery() + self.assertFalse(resumed.path.exists()) + + @patch.object(blurt_session, "StreamingAudio", FakeAudio) + def test_silence_is_spooled_but_not_uploaded(self): + called = [] + session = blurt_session.RecordingSession( + 16000, 1, Config(), self.temp.name, + self.scheduler(lambda audio, config: called.append(True) or "bad")) + session.feed(np.array([[0]], dtype=np.float32)) + session.stop() + result = session.join(2) + self.assertTrue(result.complete) + self.assertEqual(result.text, "") + self.assertEqual(called, []) + self.assertTrue(any(session.path.glob("chunk-*.wav"))) + + def test_discovery_includes_checkpoint_tail(self): + path = Path(self.temp.name) / "crashed" + path.mkdir() + blurt_session._write_wav(path / "tail.wav", np.full(20, 2, dtype=np.int16)) + (path / "session.json").write_text(json.dumps({ + "version": 1, "session_id": "crashed", "state": "recording", + "config": {"model": "old", "language": "de"}, "done": 0, + "total": 0, "entries": [], "has_tail": True, + })) + scheduler = self.scheduler(lambda audio, config: "tail") + record = blurt_session.RecoveryStore(self.temp.name).discover()[0] + self.assertEqual(record.config["language"], "de") + resumed = blurt_session.RecordingSession.recover(path, Config(api_key="fresh"), scheduler) + self.assertEqual(resumed.join(2).text, "tail") + + def _interrupted(self, name="crash", entries=None, **extra): + path = Path(self.temp.name) / name + path.mkdir() + data = {"version": 1, "session_id": name, "state": "recording", + "config": {"model": "old", "language": "de"}, "done": 0, + "total": 0, "entries": entries or [], "has_tail": False, + "tail_start": 0, "created_at": extra.pop("created_at", time.time())} + data.update(extra) + (path / "session.json").write_text(json.dumps(data)) + return path + + def test_recovery_adopts_orphan_sealed_wav_instead_of_old_tail(self): + path = self._interrupted() + blurt_session._write_wav(path / "chunk-000000.wav", np.full(20, 4, np.int16)) + blurt_session._write_wav(path / "tail.wav", np.full(10, 2, np.int16)) + seen = [] + scheduler = self.scheduler(lambda audio, config: seen.append((len(audio), int(audio[0]))) or "sealed") + result = blurt_session.RecordingSession.recover(path, Config(), scheduler).join(2) + self.assertEqual(result.text, "sealed") + self.assertEqual(seen, [(20, 4)]) + + def test_silent_interrupted_tail_is_not_uploaded(self): + path = self._interrupted() + blurt_session._write_wav(path / "tail.wav", np.zeros(20, np.int16)) + calls = [] + session = blurt_session.RecordingSession.recover( + path, Config(), self.scheduler(lambda audio, config: calls.append(1) or "wrong")) + self.assertTrue(session.join(2).complete) + self.assertEqual(session.result.text, "") + self.assertEqual(calls, []) + + def test_missing_recovery_file_finishes_with_error(self): + entry = {"index": 0, "file": "gone.wav", "speech": True, "status": "pending"} + path = self._interrupted(entries=[entry], total=1) + session = blurt_session.RecordingSession.recover(path, Config(), self.scheduler(lambda *_: "wrong")) + result = session.join(2) + self.assertFalse(result.complete) + self.assertEqual(result.text, "") + self.assertIn("missing", result.errors[0]) + + @patch.object(blurt_session, "StreamingAudio", FakeAudio) + def test_raising_callbacks_do_not_stop_worker_pool(self): + scheduler = self.scheduler(lambda audio, config: str(int(audio[0])), workers=1) + bad = blurt_session.RecordingSession(16000, 1, Config(), self.temp.name, scheduler, + on_update=lambda _: (_ for _ in ()).throw(RuntimeError("ui")), + on_complete=lambda _: (_ for _ in ()).throw(RuntimeError("ui"))) + bad.feed(np.array([[3]], np.float32)); bad.stop() + self.assertEqual(bad.join(2).text, "3") + good = blurt_session.RecordingSession(16000, 1, Config(), self.temp.name, scheduler) + good.feed(np.array([[4]], np.float32)); good.stop() + self.assertEqual(good.join(2).text, "4") + + @patch.object(blurt_session, "StreamingAudio", FakeAudio) + def test_active_exclusion_and_release_for_retry(self): + session = blurt_session.RecordingSession(16000, 1, Config(), self.temp.name, + self.scheduler(lambda *_: "text")) + self.assertEqual(blurt_session.RecoveryStore(self.temp.name).discover(), []) + session.feed(np.array([[1]], np.float32)); session.stop(); session.join(2) + self.assertEqual(blurt_session.RecoveryStore(self.temp.name).discover(), []) + session.release_for_retry() + self.assertEqual(len(blurt_session.RecoveryStore(self.temp.name).discover()), 1) + + def test_from_wav_migrates_source_after_durable_spool(self): + source = Path(self.temp.name) / "legacy.wav" + blurt_session._write_wav(source, np.tile(np.array([-1000, 1000], np.int16), 800)) + spool = Path(self.temp.name) / "sessions"; spool.mkdir() + session = blurt_session.RecordingSession.from_wav( + source, Config(), spool, self.scheduler(lambda *_: "legacy")) + self.assertFalse(source.exists()) + self.assertTrue((session.path / "session.json").exists()) + self.assertEqual(session.join(2).text, "legacy") + + def test_discovery_is_oldest_first(self): + self._interrupted("new", created_at=20) + self._interrupted("old", created_at=10) + self.assertEqual([r.session_id for r in blurt_session.RecoveryStore(self.temp.name).discover()], + ["old", "new"]) + + def test_old_tail_overlap_keeps_only_suffix_after_orphan(self): + path = self._interrupted(tail_start=0) + blurt_session._write_wav(path / "chunk-000000.wav", np.full(6, 3, np.int16)) + blurt_session._write_wav(path / "tail.wav", np.arange(1, 11, dtype=np.int16)) + seen = [] + scheduler = self.scheduler(lambda audio, config: seen.append(audio.copy()) or "x") + session = blurt_session.RecordingSession.recover(path, Config(), scheduler) + self.assertEqual(session.join(2).text, "x x") + self.assertCountEqual([tuple(int(v) for v in x) for x in seen], + [(3, 3, 3, 3, 3, 3), (7, 8, 9, 10)]) + + @patch.object(blurt_session, "StreamingAudio", FakeAudio) + def test_wav_persistence_failure_finishes_without_waiting_forever(self): + scheduler = self.scheduler(lambda *_: "unused") + session = blurt_session.RecordingSession(16000, 1, Config(), self.temp.name, scheduler) + with patch.object(blurt_session, "_write_wav", side_effect=OSError("disk full")): + session.feed(np.array([[5]], np.float32)); session.stop() + result = session.join(2) + self.assertIsNotNone(result) + self.assertFalse(result.complete) + self.assertIn("disk full", result.errors[0]) + + @patch.object(blurt_session, "StreamingAudio", FakeAudio) + def test_inflight_job_is_not_failed_by_later_capture_error(self): + started = threading.Event(); release = threading.Event() + def request(audio, config): + started.set(); release.wait(2); return "first" + scheduler = self.scheduler(request, workers=1) + session = blurt_session.RecordingSession(16000, 1, Config(), self.temp.name, scheduler) + real_write = blurt_session._write_wav + writes = 0 + def fail_second(path, samples): + nonlocal writes + writes += 1 + if writes == 2: + raise OSError("disk full") + return real_write(path, samples) + with patch.object(blurt_session, "_write_wav", side_effect=fail_second): + session.feed(np.array([[1]], np.float32)); self.assertTrue(started.wait(1)) + session.feed(np.array([[2]], np.float32)); session.stop() + self.assertIsNone(session.join(0.05)) + release.set() + result = session.join(2) + self.assertEqual(result.text, "first […]") + self.assertEqual(session.done, session.total) + + @patch.object(blurt_session, "StreamingAudio", FakeAudio) + def test_input_status_makes_result_incomplete(self): + session = blurt_session.RecordingSession(16000, 1, Config(), self.temp.name, + self.scheduler(lambda *_: "kept")) + session.report_input_status("input overflow") + session.feed(np.array([[1]], np.float32)); session.stop() + result = session.join(2) + self.assertFalse(result.complete) + self.assertEqual(result.text, "kept […]") + + @patch.object(blurt_session, "StreamingAudio", FakeAudio) + def test_partial_ack_retains_successful_text(self): + values = iter(["saved", RuntimeError("bad")]) + def request(*_): + value = next(values) + if isinstance(value, Exception): raise value + return value + session = blurt_session.RecordingSession(16000, 1, Config(), self.temp.name, + self.scheduler(request, workers=1)) + session.feed(np.array([[1]], np.float32)); session.feed(np.array([[2]], np.float32)); session.stop() + self.assertFalse(session.join(2).complete) + session.acknowledge_delivery() + manifest = json.loads((session.path / "session.json").read_text()) + self.assertEqual(manifest["entries"][0]["text"], "saved") + + def test_feed_queue_overflow_returns_immediately_and_records_gap(self): + entered = threading.Event(); release = threading.Event() + class SlowAudio(FakeAudio): + def feed(self, frames): + entered.set(); release.wait(2) + return super().feed(frames) + with patch.object(blurt_session, "StreamingAudio", SlowAudio): + session = blurt_session.RecordingSession( + 16000, 1, Config(), self.temp.name, self.scheduler(lambda *_: "first"), + input_queue_size=1) + session.feed(np.array([[1]], np.float32)); self.assertTrue(entered.wait(1)) + self.assertTrue(session.feed(np.array([[2]], np.float32))) + started = time.monotonic() + self.assertFalse(session.feed(np.array([[3]], np.float32))) + self.assertLess(time.monotonic() - started, 0.05) + release.set() + result = session.join(2) + self.assertFalse(result.complete) + self.assertIn("overflowed", " ".join(result.errors)) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_streaming_audio.py b/tests/test_streaming_audio.py new file mode 100644 index 0000000..349d00e --- /dev/null +++ b/tests/test_streaming_audio.py @@ -0,0 +1,156 @@ +import unittest + +import numpy as np +from scipy import signal + +from blurt_audio import OUTPUT_RATE, StreamingAudio + + +def tone(rate, seconds, amplitude=0.1): + t = np.arange(round(rate * seconds)) / rate + return (amplitude * np.sin(2 * np.pi * 220 * t)).astype(np.float32) + + +class StreamingAudioTests(unittest.TestCase): + def collect(self, rate, audio, block): + stream = StreamingAudio(rate, 1) + chunks = [] + for pos in range(0, len(audio), block): + chunks.extend(stream.feed(audio[pos : pos + block])) + chunks.extend(stream.finish()) + return chunks + + def assert_coverage(self, chunks, expected): + cursor = 0 + for chunk in chunks: + self.assertEqual(chunk.start, cursor) + self.assertEqual(len(chunk.samples), chunk.end - chunk.start) + cursor = chunk.end + self.assertEqual(cursor, expected) + + def test_output_length_and_ranges_are_block_size_independent(self): + for rate in (16_000, 44_100, 48_000): + audio = tone(rate, 10.123) + expected = round(len(audio) * OUTPUT_RATE / rate) + reference = self.collect(rate, audio, len(audio)) + split = self.collect(rate, audio, 997) + self.assert_coverage(reference, expected) + self.assert_coverage(split, expected) + self.assertEqual([(c.start, c.end) for c in reference], + [(c.start, c.end) for c in split]) + + def test_strong_pause_after_minimum_cuts_during_feed(self): + audio = np.concatenate((tone(OUTPUT_RATE, 8.2), np.zeros(int(OUTPUT_RATE)))) + stream = StreamingAudio(OUTPUT_RATE, 1) + chunks = stream.feed(audio) + self.assertEqual(len(chunks), 1) + self.assertGreater(chunks[0].end, 8 * OUTPUT_RATE) + self.assertLess(chunks[0].end, 9.2 * OUTPUT_RATE) + chunks += stream.finish() + self.assert_coverage(chunks, len(audio)) + + def test_shorter_pause_is_accepted_near_target(self): + audio = np.concatenate((tone(OUTPUT_RATE, 20), np.zeros(int(.6 * OUTPUT_RATE)), + tone(OUTPUT_RATE, 1))) + chunks = self.collect(OUTPUT_RATE, audio, 777) + self.assertGreaterEqual(len(chunks), 2) + self.assertLess(chunks[0].end, int(20.7 * OUTPUT_RATE)) + self.assert_coverage(chunks, len(audio)) + + def test_uninterrupted_audio_forces_bounded_chunks(self): + audio = tone(OUTPUT_RATE, 92) + chunks = self.collect(OUTPUT_RATE, audio, 1024) + self.assertGreaterEqual(len(chunks), 3) + self.assertTrue(all(c.end - c.start <= 45 * OUTPUT_RATE for c in chunks)) + self.assert_coverage(chunks, len(audio)) + + def test_silence_is_retained_but_marked_not_speech(self): + audio = np.zeros(10 * OUTPUT_RATE, dtype=np.float32) + chunks = self.collect(OUTPUT_RATE, audio, 1000) + self.assert_coverage(chunks, len(audio)) + self.assertTrue(all(not c.speech for c in chunks)) + + def test_quiet_speech_is_preserved_and_uploaded(self): + audio = tone(OUTPUT_RATE, 9, amplitude=2e-4) + chunks = self.collect(OUTPUT_RATE, audio, 1234) + self.assert_coverage(chunks, len(audio)) + self.assertTrue(all(c.speech for c in chunks)) + self.assertTrue(all(c.end - c.start >= 8 * OUTPUT_RATE for c in chunks[:-1])) + + def test_tiny_tail_and_checkpoint_are_non_mutating(self): + stream = StreamingAudio(OUTPUT_RATE, 1) + stream.feed(tone(OUTPUT_RATE, 0.031)) + first = stream.checkpoint() + second = stream.checkpoint() + np.testing.assert_array_equal(first, second) + chunks = stream.finish() + self.assertEqual(len(chunks), 1) + self.assert_coverage(chunks, round(0.031 * OUTPUT_RATE)) + + def test_multichannel_is_mixed_to_mono(self): + mono = tone(48_000, 1) + stereo = np.column_stack((mono, mono)) + stream = StreamingAudio(48_000, 2) + stream.feed(stereo) + chunks = stream.finish() + self.assert_coverage(chunks, 16_000) + + def test_samples_are_callback_size_independent(self): + for rate in (16_000, 44_100, 48_000): + audio = np.concatenate((tone(rate, 8.3), np.zeros(round(rate * .9)), + tone(rate, 2.1))) + large = self.collect(rate, audio, len(audio)) + small = self.collect(rate, audio, 317) + self.assertEqual([(c.start, c.end) for c in large], + [(c.start, c.end) for c in small]) + np.testing.assert_array_equal(np.concatenate([c.samples for c in large]), + np.concatenate([c.samples for c in small])) + + def test_resampler_joins_agree_with_whole_clip_reference(self): + rate = 44_100 + source = tone(rate, 5.25) + 0.015 * np.sin( + 2 * np.pi * 1700 * np.arange(round(rate * 5.25)) / rate + ).astype(np.float32) + chunks = self.collect(rate, source, 271) + actual = np.concatenate([c.samples for c in chunks]).astype(np.float32) / 32767 + reference = signal.resample_poly(source, 160, 441) + sos = signal.butter(2, 70, btype="highpass", fs=OUTPUT_RATE, output="sos") + zi = signal.sosfilt_zi(sos) * float(reference[0]) + reference = signal.sosfilt(sos, reference, zi=zi)[0] + rms = float(np.sqrt(np.mean(reference * reference))) + peak = float(np.percentile(np.abs(reference), 99.9)) + gain = min(10 ** (-20 / 20) / rms, 10 ** (-1 / 20) / peak, 10 ** (30 / 20)) + reference = np.clip(reference * gain, -1, 1) + self.assertLess(float(np.mean(np.abs(actual - reference))), 2e-4) + for join in range(OUTPUT_RATE, len(actual), OUTPUT_RATE): + self.assertLess(float(np.max(np.abs(actual[join - 16:join + 16] + - reference[join - 16:join + 16]))), .003) + + def test_long_recording_discards_sealed_history(self): + stream = StreamingAudio(OUTPUT_RATE, 1) + emitted = [] + block = tone(OUTPUT_RATE, 1) + max_retained = 0 + for _ in range(100): + emitted.extend(stream.feed(block)) + max_retained = max(max_retained, len(stream._resampled)) + self.assertLessEqual(stream._pending_len, stream.rate + round(stream.rate * .05)) + emitted.extend(stream.finish()) + self.assertLessEqual(max_retained, 45 * OUTPUT_RATE) + self.assert_coverage(emitted, 100 * OUTPUT_RATE) + + def test_highpass_state_is_continuous_across_resampler_batches(self): + rate = 48_000 + source = tone(rate, 4, amplitude=.03) + np.float32(.2) + chunks = self.collect(rate, source, 113) + actual = np.concatenate([c.samples for c in chunks]).astype(np.float32) + # No repeated DC-step transient should appear at one-second batch joins. + for join in range(OUTPUT_RATE, len(actual), OUTPUT_RATE): + local = actual[join - 100:join + 100] + baseline = actual[join - 500:join - 300] + self.assertLess(abs(float(np.sqrt(np.mean(local * local))) + - float(np.sqrt(np.mean(baseline * baseline)))), 100) + + +if __name__ == "__main__": + unittest.main()