127 lines
4.3 KiB
Python
127 lines
4.3 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()
|
|
cls._instance._input_device_index = None
|
|
cls._instance._input_device_resolved = False
|
|
print("🔊 AudioManager: PyAudio initialized (Global)")
|
|
return cls._instance
|
|
|
|
def get_pyaudio(self):
|
|
return self.pa
|
|
|
|
def get_input_device_index(self):
|
|
"""
|
|
Returns PortAudio input device index or None (let PortAudio pick default).
|
|
Raises a RuntimeError with a helpful message if no input devices exist.
|
|
"""
|
|
if self._input_device_resolved:
|
|
return self._input_device_index
|
|
|
|
self._input_device_index = self._resolve_input_device_index()
|
|
self._input_device_resolved = True
|
|
return self._input_device_index
|
|
|
|
def _resolve_input_device_index(self):
|
|
if self.pa is None:
|
|
return None
|
|
|
|
device_count = int(self.pa.get_device_count() or 0)
|
|
|
|
def is_input_device(idx: int) -> bool:
|
|
try:
|
|
info = self.pa.get_device_info_by_index(idx)
|
|
except Exception:
|
|
return False
|
|
return int(info.get("maxInputChannels") or 0) > 0
|
|
|
|
if AUDIO_INPUT_DEVICE_INDEX is not None:
|
|
idx = int(AUDIO_INPUT_DEVICE_INDEX)
|
|
if 0 <= idx < device_count and is_input_device(idx):
|
|
return idx
|
|
raise RuntimeError(
|
|
"Audio input initialization failed: invalid AUDIO_INPUT_DEVICE_INDEX="
|
|
f"{AUDIO_INPUT_DEVICE_INDEX}. Available input devices:\n"
|
|
+ self.describe_input_devices()
|
|
)
|
|
|
|
if AUDIO_INPUT_DEVICE_NAME:
|
|
needle = AUDIO_INPUT_DEVICE_NAME.lower()
|
|
for idx in range(device_count):
|
|
if not is_input_device(idx):
|
|
continue
|
|
try:
|
|
name = str(self.pa.get_device_info_by_index(idx).get("name") or "")
|
|
except Exception:
|
|
continue
|
|
if needle in name.lower():
|
|
return idx
|
|
|
|
raise RuntimeError(
|
|
"Audio input initialization failed: could not find an input device "
|
|
f"matching AUDIO_INPUT_DEVICE_NAME={AUDIO_INPUT_DEVICE_NAME!r}. "
|
|
"Available input devices:\n"
|
|
+ self.describe_input_devices()
|
|
)
|
|
|
|
# Default input device (if PortAudio has one).
|
|
try:
|
|
default_info = self.pa.get_default_input_device_info()
|
|
default_idx = int(default_info.get("index"))
|
|
if 0 <= default_idx < device_count and is_input_device(default_idx):
|
|
return default_idx
|
|
except Exception:
|
|
pass
|
|
|
|
# Fallback: first input device.
|
|
for idx in range(device_count):
|
|
if is_input_device(idx):
|
|
return idx
|
|
|
|
raise RuntimeError(
|
|
"Audio input initialization failed: no input devices found. "
|
|
"Check microphone connection and PipeWire/PulseAudio. "
|
|
"PortAudio devices:\n"
|
|
+ self.describe_input_devices()
|
|
)
|
|
|
|
def describe_input_devices(self, limit: int = 20) -> str:
|
|
if self.pa is None:
|
|
return "<PyAudio not initialized>"
|
|
|
|
items = []
|
|
count = int(self.pa.get_device_count() or 0)
|
|
for idx in range(count):
|
|
try:
|
|
info = self.pa.get_device_info_by_index(idx)
|
|
except Exception:
|
|
continue
|
|
max_in = int(info.get("maxInputChannels") or 0)
|
|
if max_in <= 0:
|
|
continue
|
|
name = str(info.get("name") or "").strip()
|
|
items.append(f"[{idx}] {name} (in={max_in})")
|
|
if len(items) >= limit:
|
|
break
|
|
return "\n".join(items) if items else "<no input devices>"
|
|
|
|
def cleanup(self):
|
|
if self.pa:
|
|
self.pa.terminate()
|
|
self.pa = None
|
|
|
|
|
|
def get_audio_manager():
|
|
return AudioManager()
|