132 lines
5.6 KiB
Python
132 lines
5.6 KiB
Python
"""Тест Модуля 2: audio_io (запись / воспроизведение / ресемплинг / мгновенный stop).
|
|
|
|
Запуск:
|
|
python modules/audio_io/test_audio_io.py devices # список аудио-устройств
|
|
python modules/audio_io/test_audio_io.py auto # автоматический смоук (без участия человека)
|
|
python modules/audio_io/test_audio_io.py talk # ИНТЕРАКТИВНЫЙ: push-to-talk через Enter
|
|
|
|
Режим talk:
|
|
Enter → начать запись (говори), Enter → закончить и услышать себя,
|
|
затем записанное уходит в STT (проверка склейки Модуль 2 + Модуль 1).
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import sys
|
|
import time
|
|
from pathlib import Path
|
|
|
|
import numpy as np
|
|
|
|
try:
|
|
from modules.audio_io import Player, Recorder, list_devices, resample_to_16k
|
|
except ImportError:
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
|
|
from modules.audio_io import Player, Recorder, list_devices, resample_to_16k
|
|
|
|
SAMPLES_DIR = Path(__file__).resolve().parent / "samples"
|
|
|
|
|
|
# --------------------------------------------------------------------------
|
|
def smoke_test() -> None:
|
|
"""Автоматический смоук: ресемплинг, тоны, мгновенная остановка, короткая запись."""
|
|
print("== 1. Ресемплинг ==")
|
|
t = np.linspace(0.0, 1.0, 44100, endpoint=False, dtype=np.float32)
|
|
sine44k = 0.5 * np.sin(2 * np.pi * 440.0 * t)
|
|
out16k = resample_to_16k(sine44k, 44100)
|
|
assert len(out16k) == 16000, f"длина {len(out16k)} != 16000"
|
|
print(f" 44100 Гц ({len(sine44k)} сэмплов) → 16000 Гц ({len(out16k)}) OK")
|
|
# оценка частоты тона после ресемплинга (через число пересечений нуля)
|
|
zero_cross = np.count_nonzero(np.diff(np.signbit(out16k)))
|
|
freq = zero_cross / 2.0
|
|
print(f" тон после ресемплинга: ~{freq:.0f} Гц (ожидалось 440) OK")
|
|
|
|
print("== 2. Воспроизведение и мгновенный stop ==")
|
|
player = Player()
|
|
long_tone = 0.3 * np.sin(2 * np.pi * 440.0 * np.linspace(0, 5, 5 * 48000, endpoint=False))
|
|
player.play(long_tone, 48000)
|
|
time.sleep(0.8)
|
|
assert player.is_playing, "тон должен играть"
|
|
t0 = time.perf_counter()
|
|
player.stop()
|
|
stop_ms = (time.perf_counter() - t0) * 1000
|
|
print(f" stop() сработал за {stop_ms:.0f} мс (цель < 100 мс)")
|
|
assert not player.is_playing
|
|
|
|
print("== 3. Запись реального устройства (2 с, без участия человека) ==")
|
|
rec = Recorder(min_seconds=0.1)
|
|
rec.start()
|
|
time.sleep(2.0)
|
|
audio = rec.stop()
|
|
rms = float(np.sqrt(np.mean(np.square(audio)))) if len(audio) else 0.0
|
|
print(f" записано {len(audio)/16000:.1f} c, RMS={rms:.4f} (0 = тишина/нет данных)")
|
|
if len(audio) == 0:
|
|
print(" ВНИМАНИЕ: устройство вернуло слишком мало данных")
|
|
|
|
print("== 4. Короткая запись отбрасывается (защита) ==")
|
|
rec2 = Recorder(min_seconds=0.5)
|
|
rec2.start()
|
|
time.sleep(0.05)
|
|
empty = rec2.stop()
|
|
assert len(empty) == 0, "короткая запись должна возвращать пустой массив"
|
|
print(" OK: 50 мс → пусто")
|
|
|
|
print("\nСМОУК ПРОЙДЕН ✅")
|
|
|
|
|
|
# --------------------------------------------------------------------------
|
|
def talk_test() -> None:
|
|
"""Интерактивный push-to-talk: Enter — говорить, Enter — стоп, потом плеер и STT."""
|
|
from modules.stt import get_provider
|
|
|
|
print("Загружаю STT (faster-whisper small)...")
|
|
stt = get_provider("faster-whisper", model_size="small")
|
|
player = Player()
|
|
rec = Recorder()
|
|
|
|
while True:
|
|
print("\n[Enter] удержи-режим: нажми Enter и ГОВОРИ, затем Enter — стоп. "
|
|
"Пустая строка после фразы — выход.")
|
|
cmd = input("> ")
|
|
if cmd.strip() == "":
|
|
print("Выход.")
|
|
return
|
|
|
|
rec.start()
|
|
input("…запись идёт, Enter = закончить ")
|
|
audio = rec.stop()
|
|
if len(audio) == 0:
|
|
print("Слишком коротко — отброшено. Ещё раз.")
|
|
continue
|
|
|
|
print(f"Записано {len(audio)/16000:.1f} c (частота устройства {rec.native_samplerate} Гц).")
|
|
wav = Path(__file__).resolve().parent / "samples" / "ptt_last.wav"
|
|
wav.parent.mkdir(parents=True, exist_ok=True)
|
|
import soundfile as sf
|
|
sf.write(str(wav), audio, 16000)
|
|
print(f"Сохранено: {wav}")
|
|
|
|
print("Воспроизвожу твой голос (проверь качество, Esc/Enter — прервать)...")
|
|
player.play(audio, 16000)
|
|
try:
|
|
input(" (Enter — если хочешь прервать звук) ")
|
|
player.stop()
|
|
except KeyboardInterrupt:
|
|
player.stop()
|
|
|
|
res = stt.transcribe(audio)
|
|
print(f"STT: {res}")
|
|
|
|
|
|
def main() -> int:
|
|
mode = sys.argv[1] if len(sys.argv) > 1 else "auto"
|
|
if mode == "devices":
|
|
list_devices()
|
|
elif mode == "talk":
|
|
talk_test()
|
|
else:
|
|
smoke_test()
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main()) |