59 lines
2.1 KiB
Python
59 lines
2.1 KiB
Python
import pyaudio
|
||
import threading
|
||
|
||
from .config import AUDIO_INPUT_DEVICE_INDEX, AUDIO_INPUT_DEVICE_NAME
|
||
|
||
|
||
class AudioManager:
|
||
_instance = None
|
||
_lock = threading.Lock()
|
||
|
||
def __new__(cls):
|
||
with cls._lock:
|
||
if cls._instance is None:
|
||
cls._instance = super(AudioManager, cls).__new__(cls)
|
||
cls._instance.pa = pyaudio.PyAudio()
|
||
print("🔊 AudioManager: PyAudio initialized (Global)")
|
||
return cls._instance
|
||
|
||
def get_pyaudio(self):
|
||
return self.pa
|
||
|
||
def resolve_input_device(self):
|
||
"""
|
||
Возвращает индекс и информацию о входном устройстве.
|
||
Если индекс/имя не заданы в .env, используем системное default устройство.
|
||
"""
|
||
if AUDIO_INPUT_DEVICE_INDEX is not None:
|
||
info = self.pa.get_device_info_by_index(AUDIO_INPUT_DEVICE_INDEX)
|
||
if info.get("maxInputChannels", 0) <= 0:
|
||
raise ValueError(
|
||
f"AUDIO_INPUT_DEVICE_INDEX={AUDIO_INPUT_DEVICE_INDEX} не является входным устройством."
|
||
)
|
||
return AUDIO_INPUT_DEVICE_INDEX, info
|
||
|
||
if AUDIO_INPUT_DEVICE_NAME:
|
||
wanted = AUDIO_INPUT_DEVICE_NAME.lower()
|
||
for index in range(self.pa.get_device_count()):
|
||
info = self.pa.get_device_info_by_index(index)
|
||
if info.get("maxInputChannels", 0) <= 0:
|
||
continue
|
||
name = str(info.get("name", ""))
|
||
if wanted in name.lower():
|
||
return index, info
|
||
raise ValueError(
|
||
f"Не найдено входное устройство по AUDIO_INPUT_DEVICE_NAME={AUDIO_INPUT_DEVICE_NAME!r}."
|
||
)
|
||
|
||
default_info = self.pa.get_default_input_device_info()
|
||
return None, default_info
|
||
|
||
def cleanup(self):
|
||
if self.pa:
|
||
self.pa.terminate()
|
||
self.pa = None
|
||
|
||
|
||
def get_audio_manager():
|
||
return AudioManager()
|