felixmandClaude Opus 5
9d04427816
Transcripts came back short, but the log showed the whole clip going up: captured samples matched wall clock and upload duration matched the recording minus exactly the fixed 150 ms head trim. So nothing was being cut in transit — the losses had to be at the edges or inside the audio, and neither was visible. Trim the stream-open click by measurement instead of by the clock: cut only the leading frames that are genuinely louder than the rest of the clip, up to the old cap. Mics that do not click now keep their first syllable. Measure each frame against its own mean so a steady DC bias — or a stream frozen on one value — reads as the silence it is, and report the longest run of digital silence as dead=. A live mic never goes below DEAD_AIR_DBFS, so anything there is input that never arrived, which is what following the system default source invites. Warn on startup when no device is pinned, log the noise floor next to the speech level, and log how much of the clip the output limiter had to clamp. BLURT_SAVE_CAPTURES keeps the raw and uploaded WAVs per clip, so the next short transcript can be checked against what the API actually got. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
101 lines
3.6 KiB
Python
101 lines
3.6 KiB
Python
"""Diagnose Blurt's capture path: record, measure, transcribe, save the WAV.
|
|
|
|
Usage: uv run mic_check.py [device] [seconds]
|
|
device is a sounddevice index or name substring; omitted = whatever
|
|
Blurt itself is configured to use (device.txt, else system default).
|
|
"""
|
|
|
|
import io
|
|
import sys
|
|
import tempfile
|
|
from pathlib import Path
|
|
|
|
import numpy as np
|
|
import sounddevice as sd
|
|
from scipy.io import wavfile
|
|
|
|
import blurt
|
|
|
|
OUT_WAV = Path(tempfile.gettempdir()) / "blurt_mic_check.wav"
|
|
|
|
|
|
def report(label: str, samples: np.ndarray) -> None:
|
|
s = blurt.audio_stats(samples)
|
|
frame = blurt.SAMPLE_RATE // 50
|
|
energy = np.array([
|
|
np.sqrt(np.mean(np.square(samples[i:i + frame])) + 1e-20)
|
|
for i in range(0, max(len(samples) - frame, 1), frame)
|
|
])
|
|
snr = 20 * np.log10(np.percentile(energy, 90) / (np.percentile(energy, 10) + 1e-12))
|
|
print(f" {label}")
|
|
print(f" rms {s['rms_dbfs']:+7.1f} dBFS (speech wants -30..-15)")
|
|
print(f" peak {s['peak_dbfs']:+7.1f} dBFS")
|
|
print(f" dc bias {s['dc']:+7.4f} (wants ~0.0000)")
|
|
print(f" clipped {s['clipped_pct']:7.3f} %")
|
|
print(f" loud/quiet spread {snr:.1f} dB (real speech wants > 25)")
|
|
|
|
|
|
def main() -> None:
|
|
args = sys.argv[1:]
|
|
device = blurt.load_input_device_name() or None
|
|
if args and args[0]:
|
|
try:
|
|
device = int(args[0])
|
|
except ValueError:
|
|
device = args[0]
|
|
seconds = float(args[1]) if len(args) > 1 else 6.0
|
|
|
|
print("Input devices:")
|
|
for i, d in enumerate(sd.query_devices()):
|
|
if d["max_input_channels"] > 0:
|
|
print(f" {i:>3} {d['name']}")
|
|
resolved = sd.query_devices(device, "input") if device else sd.query_devices(kind="input")
|
|
rate, channels = blurt.input_stream_settings(device)
|
|
print(f"\nRecording {seconds:g}s from: {resolved['name']} at {rate} Hz / {channels} ch — speak now.")
|
|
|
|
frames = sd.rec(
|
|
int(seconds * rate),
|
|
samplerate=rate,
|
|
channels=channels,
|
|
dtype=blurt.DTYPE,
|
|
device=device,
|
|
)
|
|
sd.wait()
|
|
audio = frames.mean(axis=1) if channels > 1 else frames[:, 0]
|
|
audio = blurt.resample_to_target(audio, rate)
|
|
click = blurt.leading_click_samples(audio)
|
|
audio = audio[click:]
|
|
|
|
print(f"\nCapture (trimmed {click * 1000 / blurt.SAMPLE_RATE:.0f} ms of stream-open click):")
|
|
report("raw", audio)
|
|
conditioned = blurt.condition_audio(audio)
|
|
report("after Blurt conditioning", conditioned)
|
|
|
|
buffer = io.BytesIO()
|
|
wavfile.write(buffer, blurt.SAMPLE_RATE, (conditioned * 32767).astype(np.int16))
|
|
wav_bytes = buffer.getvalue()
|
|
OUT_WAV.write_bytes(wav_bytes)
|
|
print(f"\nWAV saved: {OUT_WAV} ({len(wav_bytes)} bytes) — play it to hear what the API hears")
|
|
|
|
app = blurt.BlurtApp.__new__(blurt.BlurtApp)
|
|
app.api_keys = {"openai": blurt.load_api_key(), "openrouter": blurt.load_openrouter_key()}
|
|
app.transcription_model = blurt.load_transcription_model()
|
|
app.transcription_language = blurt.load_transcription_language()
|
|
app.session = blurt.requests.Session()
|
|
provider = app._provider_for_model(app.transcription_model)
|
|
if not app.api_keys.get(provider):
|
|
print(f"\nNo {provider} key available; skipping transcription.")
|
|
return
|
|
print(f"\nTranscribing with {app.transcription_model} ({app.transcription_language})…")
|
|
result = (
|
|
app._transcribe_openrouter(wav_bytes)
|
|
if provider == "openrouter"
|
|
else app._transcribe_openai(wav_bytes)
|
|
)
|
|
print("\n--- transcript ---")
|
|
print(result.get("text", result))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|