"""Sesli komut girisi (STT). Tasarim ------- * Mikrofon yakalama : `sounddevice` (PortAudio sarmalayicisi, pip ile gelir) * Konusma -> metin : `faster-whisper` (yerel; CPU int8 / GPU float16) * Konusma sonu tespiti: basit RMS esigi + sessizlik suresi (silence detection) * Modulin yuklenmesi : tembel; faster-whisper modeli ilk `listen()` cagrisinda indirilip belleme alinir. Bu modul agent.py'da REPL'in `:v` komutuyla cagrilir. `is_available()` False donerse kullaniciya net "voice kapali" mesaji gosterilir; agent yine de calisir. Cevre degiskenleri (config.py / VoiceConfig): VOICE_ENABLED, VOICE_MODEL, VOICE_DEVICE, VOICE_COMPUTE_TYPE, VOICE_LANGUAGE, VOICE_SAMPLERATE, VOICE_SILENCE_THRESHOLD, VOICE_SILENCE_SECONDS, VOICE_MAX_SECONDS, VOICE_INITIAL_TIMEOUT, VOICE_INPUT_DEVICE, VOICE_CONFIRM """ from __future__ import annotations import os import queue import sys import time from typing import TYPE_CHECKING, Optional import numpy as np from config import VoiceConfig if TYPE_CHECKING: from logger import SessionLogger # ----------------------------------------------------------------------------- # huggingface_hub'in symlink uyarisi Windows'ta cok gurultulu; bizim icin # sembolik link sart degil (cache yine calisir). Faster-whisper import edilmeden # once env var'i set ediyoruz. # ----------------------------------------------------------------------------- os.environ.setdefault("HF_HUB_DISABLE_SYMLINKS_WARNING", "1") # ----------------------------------------------------------------------------- # Opsiyonel bagimliliklarin tembel ithali # ----------------------------------------------------------------------------- # sounddevice / faster-whisper olmadan da agent.py calismali. Bu yuzden # import hatalarini sessizce tutup `is_available()` ile dosabilecegiz. try: import sounddevice as sd # type: ignore[import-not-found] _SD_OK = True _SD_ERR: Optional[Exception] = None except Exception as exc: # noqa: BLE001 sd = None # type: ignore[assignment] _SD_OK = False _SD_ERR = exc try: from faster_whisper import WhisperModel # type: ignore[import-not-found] _FW_OK = True _FW_ERR: Optional[Exception] = None except Exception as exc: # noqa: BLE001 WhisperModel = None # type: ignore[assignment] _FW_OK = False _FW_ERR = exc # ----------------------------------------------------------------------------- # Konsol renkleri (agent.py ile ayni palet) # ----------------------------------------------------------------------------- _C_RESET = "\033[0m" _C_DIM = "\033[2m" _C_RED = "\033[31m" _C_GREEN = "\033[32m" _C_YELLOW = "\033[33m" _C_CYAN = "\033[36m" # ----------------------------------------------------------------------------- # Hatalar # ----------------------------------------------------------------------------- class VoiceUnavailableError(RuntimeError): """Kutuphaneler eksik veya cihaz acilamadigi durum.""" class VoiceCancelled(RuntimeError): """Kullanici Ctrl+C ile dinlemeyi iptal etti.""" # ----------------------------------------------------------------------------- # Yardimci: cihaz secimi # ----------------------------------------------------------------------------- def _resolve_device(spec: str) -> Optional[int | str]: """`VOICE_INPUT_DEVICE` degerini sounddevice'in bekledigi forma cevir. Bos -> None (sistem default) Tam sayi -> int (device index) String -> sounddevice'in kendi name-substring resolution'i kullanilir. """ if not spec: return None s = spec.strip() try: return int(s) except ValueError: return s def list_input_devices() -> list[tuple[int, str, int]]: """Mevcut input cihazlarini listele. (index, ad, max_input_channels).""" if not _SD_OK: raise VoiceUnavailableError( "sounddevice yuklu degil. `pip install sounddevice` ile kur." ) devices = sd.query_devices() # type: ignore[union-attr] out: list[tuple[int, str, int]] = [] for i, d in enumerate(devices): ch = int(d.get("max_input_channels", 0)) if ch <= 0: continue out.append((i, str(d.get("name", "?")), ch)) return out # ----------------------------------------------------------------------------- # Ana sinif # ----------------------------------------------------------------------------- class VoiceInput: """Push-to-record stili sesli komut girisi. Kullanim: voice = VoiceInput(VoiceConfig.from_env(), logger=log) if voice.is_available(): text = voice.listen_once() # mikrofona konus, transkript don """ def __init__( self, cfg: VoiceConfig, *, logger: Optional["SessionLogger"] = None, ) -> None: self.cfg = cfg self.logger = logger self._model: Optional["WhisperModel"] = None # tembel yukle # --- saglik kontrolu ------------------------------------------------- def is_available(self) -> tuple[bool, Optional[str]]: """(ok, sebep) doner. ok=False ise `sebep` insan-okur aciklamadir.""" if not self.cfg.enabled: return False, "VOICE_ENABLED=false (.env icinde acabilirsin)." if not _SD_OK: return False, ( "sounddevice yuklu degil veya yuklenemedi. " "Kurulum: `pip install sounddevice` " f"(orijinal hata: {_SD_ERR})" ) if not _FW_OK: return False, ( "faster-whisper yuklu degil. " "Kurulum: `pip install faster-whisper` " f"(orijinal hata: {_FW_ERR})" ) return True, None # --- model yuklemesi (tembel) --------------------------------------- def warmup(self) -> bool: """Modeli onceden yukle (agent baslarken cagirilirsa ilk :v'de beklemeyi azaltir). Hata olursa False doner; ana akis durmaz. """ ok, why = self.is_available() if not ok: return False try: self._ensure_model() except VoiceUnavailableError as exc: print(f"{_C_YELLOW}[voice] model on-yukleme basarisiz: {exc}{_C_RESET}") return False return True def _ensure_model(self) -> "WhisperModel": if self._model is not None: return self._model ok, why = self.is_available() if not ok: raise VoiceUnavailableError(why or "voice kullanilamiyor") print( f"{_C_DIM}[voice] faster-whisper yukleniyor " f"(model={self.cfg.model_size}, device={self.cfg.device}, " f"compute_type={self.cfg.compute_type}; " f"ilk seferde HF'den indirilebilir, bekle){_C_RESET}", flush=True, ) if self.logger: self.logger.info( "voice_model_load", model=self.cfg.model_size, device=self.cfg.device, compute_type=self.cfg.compute_type, ) t0 = time.monotonic() # Onemli: download veya init sirasinda Ctrl+C gelirse self._model # None kalsin ki sonraki cagri yeniden denesin (tutarli durum). local_model: Optional["WhisperModel"] = None try: local_model = WhisperModel( # type: ignore[misc] self.cfg.model_size, device=self.cfg.device, compute_type=self.cfg.compute_type, ) except KeyboardInterrupt: print(f"\n{_C_YELLOW}[voice] model yuklemesi iptal edildi.{_C_RESET}") if self.logger: self.logger.warn("voice_model_load_cancelled") raise VoiceCancelled("model yuklemesi iptal") from None except Exception as exc: # noqa: BLE001 if self.logger: self.logger.error("voice_model_load_failed", detail=str(exc)) raise VoiceUnavailableError( f"faster-whisper modeli yuklenemedi: {exc}" ) from exc self._model = local_model dt = time.monotonic() - t0 print( f"{_C_DIM}[voice] model hazir ({dt:.1f}s).{_C_RESET}", flush=True, ) if self.logger: self.logger.info("voice_model_ready", load_seconds=round(dt, 2)) return self._model # --- ana akis: kayit + transkript ----------------------------------- def listen_once(self) -> str: """Bir tur dinle, transkripti string olarak don. - Konusma baslamazsa `initial_timeout` icinde abort - Konusma bittikten sonra `silence_seconds` sessizlikten sonra dur - `max_seconds` sert ust sinir (kotu calar / surekli gurultu) - Ctrl+C -> VoiceCancelled """ ok, why = self.is_available() if not ok: raise VoiceUnavailableError(why or "voice kullanilamiyor") audio = self._record_until_silence() if audio.size == 0: return "" # faster-whisper float32 mono, 16 kHz beklemiyor; resample gerekmez # cunku biz zaten 16 kHz mono float32 kayit alacagiz. text = self._transcribe(audio) if self.logger: self.logger.info("voice_transcript", text=text, samples=int(audio.size)) return text # --- kayit: silence detection ile ---------------------------------- def _record_until_silence(self) -> np.ndarray: if not _SD_OK: raise VoiceUnavailableError("sounddevice yok") cfg = self.cfg device = _resolve_device(cfg.input_device) block_seconds = 0.05 # 50 ms blok -> akiskan VAD blocksize = int(cfg.samplerate * block_seconds) buf: queue.Queue[np.ndarray] = queue.Queue() def _cb(indata, frames, time_info, status): # noqa: ARG001 if status: # Underrun/overflow gibi durumlarda sessizce gec; hatayi # logger'a yazip akisi bozmamak istiyoruz. if self.logger: self.logger.warn("voice_stream_status", detail=str(status)) buf.put(indata.copy()) collected: list[np.ndarray] = [] spoke = False silence_started_at: Optional[float] = None start = time.monotonic() print( f"{_C_GREEN}[voice] dinleniyor... " f"{_C_DIM}(konusmaya basla; {cfg.silence_seconds:.1f}s sessiz " f"kalinca otomatik kapanir, Ctrl+C iptal){_C_RESET}", flush=True, ) if self.logger: self.logger.info( "voice_record_start", samplerate=cfg.samplerate, device=str(device) if device is not None else "default", ) try: with sd.InputStream( # type: ignore[union-attr] samplerate=cfg.samplerate, channels=1, dtype="float32", blocksize=blocksize, device=device, callback=_cb, ): while True: elapsed = time.monotonic() - start if elapsed >= cfg.max_seconds: print( f"\n{_C_YELLOW}[voice] max_seconds ({cfg.max_seconds:.0f}s) " f"asildi, kapaniyor.{_C_RESET}" ) break if not spoke and elapsed >= cfg.initial_timeout: print( f"\n{_C_YELLOW}[voice] {cfg.initial_timeout:.0f}s icinde " f"konusma duyulmadi, iptal.{_C_RESET}" ) return np.zeros(0, dtype=np.float32) try: chunk = buf.get(timeout=0.1) except queue.Empty: continue rms = float(np.sqrt(np.mean(np.square(chunk, dtype=np.float64)))) loud = rms >= cfg.silence_threshold # Mikro VU-meter: konsola tek karakter yaz bar = "#" if loud else "." sys.stdout.write(bar) sys.stdout.flush() if loud: if not spoke: spoke = True silence_started_at = None collected.append(chunk) else: if spoke: collected.append(chunk) if silence_started_at is None: silence_started_at = time.monotonic() elif (time.monotonic() - silence_started_at) >= cfg.silence_seconds: break # konusma henuz baslamadiysa bos chunk'lari atlat except KeyboardInterrupt as exc: print(f"\n{_C_YELLOW}[voice] iptal edildi.{_C_RESET}") if self.logger: self.logger.warn("voice_cancelled") raise VoiceCancelled("kullanici dinlemeyi iptal etti") from exc except Exception as exc: # noqa: BLE001 if self.logger: self.logger.error("voice_stream_error", detail=str(exc)) raise VoiceUnavailableError( f"Mikrofon akisi acilamadi: {exc}. " "VOICE_INPUT_DEVICE'i kontrol et veya `python -m voice devices` ile listele." ) from exc print() # vu-meter satirini kapat if not collected: return np.zeros(0, dtype=np.float32) audio = np.concatenate(collected, axis=0).flatten().astype(np.float32, copy=False) duration = audio.size / cfg.samplerate rms_avg = float(np.sqrt(np.mean(np.square(audio, dtype=np.float64)))) print( f"{_C_DIM}[voice] kayit bitti: {duration:.1f}s, " f"rms={rms_avg:.3f} (esik={cfg.silence_threshold:.3f}){_C_RESET}", flush=True, ) if self.logger: self.logger.info( "voice_record_done", duration_s=round(duration, 2), rms_avg=rms_avg, ) return audio # --- transkripsiyon ------------------------------------------------- def _transcribe(self, audio: np.ndarray) -> str: model = self._ensure_model() lang = None if self.cfg.language.lower() in {"auto", ""} else self.cfg.language # Kullanici sessizlikte bekledigini sansin diye kucuk bir not. print(f"{_C_DIM}[voice] transkript ediliyor...{_C_RESET}", flush=True) t0 = time.monotonic() try: segments, info = model.transcribe( audio, language=lang, beam_size=5, vad_filter=True, vad_parameters={"min_silence_duration_ms": 300}, ) text_parts: list[str] = [] for seg in segments: if seg.text: text_parts.append(seg.text.strip()) text = " ".join(p for p in text_parts if p).strip() except KeyboardInterrupt: print(f"\n{_C_YELLOW}[voice] transkript iptal edildi.{_C_RESET}") if self.logger: self.logger.warn("voice_transcribe_cancelled") raise VoiceCancelled("transkript iptal") from None except Exception as exc: # noqa: BLE001 if self.logger: self.logger.error("voice_transcribe_error", detail=str(exc)) raise VoiceUnavailableError(f"Transkripsiyon hatasi: {exc}") from exc dt = time.monotonic() - t0 detected = getattr(info, "language", None) prob = getattr(info, "language_probability", None) prob_str = f", lang={detected} ({prob:.2f})" if detected and prob is not None else "" print( f"{_C_DIM}[voice] transkript {dt:.1f}s{prob_str}{_C_RESET}" ) return text # ----------------------------------------------------------------------------- # CLI yardimcisi: # python -m voice devices -> mikrofonlari listele # python -m voice listen -> tek seferlik dinle, transkripti yaz # ----------------------------------------------------------------------------- def _cli(argv: list[str]) -> int: cmd = argv[1] if len(argv) > 1 else "devices" if cmd in {"devices", "list"}: try: for idx, name, ch in list_input_devices(): print(f" [{idx:>2}] {name} (in_ch={ch})") except VoiceUnavailableError as exc: print(f"{_C_RED}{exc}{_C_RESET}", file=sys.stderr) return 2 return 0 if cmd in {"listen", "test"}: cfg = VoiceConfig.from_env() v = VoiceInput(cfg) ok, why = v.is_available() if not ok: print(f"{_C_RED}voice kullanilamiyor: {why}{_C_RESET}", file=sys.stderr) return 3 try: text = v.listen_once() except VoiceCancelled: return 130 except VoiceUnavailableError as exc: print(f"{_C_RED}{exc}{_C_RESET}", file=sys.stderr) return 4 print(f"\n{_C_CYAN}>> transkript:{_C_RESET} {text!r}") return 0 print(__doc__) print("\nKomutlar: devices | listen", file=sys.stderr) return 1 if __name__ == "__main__": sys.exit(_cli(sys.argv))