| import io, wave, re, asyncio | |
| import numpy as np | |
| import sounddevice as sd | |
| import httpx | |
| from config import CFG | |
| from vlog import log, exception | |
| class Ears: | |
| def __init__(self): | |
| self.frames = [] | |
| self.stream = None | |
| self.active = False | |
| def start(self): | |
| self.frames.clear() | |
| self.active = True | |
| self.stream = sd.InputStream( | |
| samplerate=CFG.sample_rate, | |
| channels=CFG.channels, | |
| dtype="int16", | |
| blocksize=1600, | |
| callback=self._callback, | |
| ) | |
| self.stream.start() | |
| def _callback(self, indata, frames, time_info, status): | |
| if self.active: | |
| self.frames.append(indata.copy()) | |
| def stop(self): | |
| self.active = False | |
| if self.stream: | |
| try: | |
| self.stream.stop() | |
| self.stream.close() | |
| finally: | |
| self.stream = None | |
| if not self.frames: | |
| return b"" | |
| pcm = np.concatenate(self.frames, axis=0).astype(np.int16) | |
| bio = io.BytesIO() | |
| with wave.open(bio, "wb") as w: | |
| w.setnchannels(1) | |
| w.setsampwidth(2) | |
| w.setframerate(CFG.sample_rate) | |
| w.writeframes(pcm.tobytes()) | |
| return bio.getvalue() | |
| async def transcribe(self, wav_bytes: bytes) -> str: | |
| if not wav_bytes: | |
| return "" | |
| try: | |
| async with httpx.AsyncClient(timeout=60) as client: | |
| r = await client.post( | |
| CFG.whisper_url, | |
| files={"file": ("speech.wav", wav_bytes, "audio/wav")}, | |
| data={"model": CFG.whisper_model, "language": "en", | |
| "response_format": "json"}, | |
| ) | |
| r.raise_for_status() | |
| data = r.json() | |
| text = data.get("text", "") | |
| # Legacy/open-mic Whisper markers. | |
| text = re.sub(r"\[(?:SIGHS|BLANK_AUDIO|SILENCE|MUSIC|NOISE)\]", "", text, | |
| flags=re.I) | |
| return re.sub(r"\s+", " ", text).strip() | |
| except Exception as e: | |
| exception("transcription_error", e) | |
| return "" | |
Xet Storage Details
- Size:
- 2.22 kB
- Xet hash:
- 68041532af98aab89b3b6ac55ace4b88dc1a09d3247dd9350660824b5877462b
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.