Hugging Face's logo

Tayrrr
/
KaliSupport

GGUF
conversational
Model card Files
xet
 Community
1
KaliSupport / voice.py
Tayrrr's picture
Tayrrr
Upload folder using huggingface_hub
7ea1cf8 verified
Raw History Blame Contribute Delete
17.9 kB
"""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))