feat: harden audio device compatibility across machines
This commit is contained in:
@@ -1,7 +1,12 @@
|
||||
import pyaudio
|
||||
import threading
|
||||
|
||||
from .config import AUDIO_INPUT_DEVICE_INDEX, AUDIO_INPUT_DEVICE_NAME
|
||||
from .config import (
|
||||
AUDIO_INPUT_DEVICE_INDEX,
|
||||
AUDIO_INPUT_DEVICE_NAME,
|
||||
AUDIO_OUTPUT_DEVICE_INDEX,
|
||||
AUDIO_OUTPUT_DEVICE_NAME,
|
||||
)
|
||||
|
||||
|
||||
class AudioManager:
|
||||
@@ -14,7 +19,9 @@ class AudioManager:
|
||||
cls._instance = super(AudioManager, cls).__new__(cls)
|
||||
cls._instance.pa = pyaudio.PyAudio()
|
||||
cls._instance._input_device_index = None
|
||||
cls._instance._output_device_index = None
|
||||
cls._instance._input_device_resolved = False
|
||||
cls._instance._output_device_resolved = False
|
||||
print("🔊 AudioManager: PyAudio initialized (Global)")
|
||||
return cls._instance
|
||||
|
||||
@@ -33,22 +40,84 @@ class AudioManager:
|
||||
self._input_device_resolved = True
|
||||
return self._input_device_index
|
||||
|
||||
def get_output_device_index(self):
|
||||
"""
|
||||
Returns PortAudio output device index or None (let PortAudio pick default).
|
||||
Raises a RuntimeError with a helpful message if no output devices exist.
|
||||
"""
|
||||
if self._output_device_resolved:
|
||||
return self._output_device_index
|
||||
|
||||
self._output_device_index = self._resolve_output_device_index()
|
||||
self._output_device_resolved = True
|
||||
return self._output_device_index
|
||||
|
||||
def _get_device_count(self) -> int:
|
||||
if self.pa is None:
|
||||
return 0
|
||||
return int(self.pa.get_device_count() or 0)
|
||||
|
||||
def _is_input_device(self, 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
|
||||
|
||||
def _is_output_device(self, idx: int) -> bool:
|
||||
try:
|
||||
info = self.pa.get_device_info_by_index(idx)
|
||||
except Exception:
|
||||
return False
|
||||
return int(info.get("maxOutputChannels") or 0) > 0
|
||||
|
||||
def _find_device_by_name(self, needle: str, input_kind: bool):
|
||||
if not needle:
|
||||
return None
|
||||
lowered = needle.lower()
|
||||
count = self._get_device_count()
|
||||
for idx in range(count):
|
||||
if input_kind and not self._is_input_device(idx):
|
||||
continue
|
||||
if not input_kind and not self._is_output_device(idx):
|
||||
continue
|
||||
try:
|
||||
name = str(self.pa.get_device_info_by_index(idx).get("name") or "")
|
||||
except Exception:
|
||||
continue
|
||||
if lowered in name.lower():
|
||||
return idx
|
||||
return None
|
||||
|
||||
def _get_default_input_index(self):
|
||||
try:
|
||||
info = self.pa.get_default_input_device_info()
|
||||
idx = int(info.get("index"))
|
||||
if self._is_input_device(idx):
|
||||
return idx
|
||||
except Exception:
|
||||
pass
|
||||
return None
|
||||
|
||||
def _get_default_output_index(self):
|
||||
try:
|
||||
info = self.pa.get_default_output_device_info()
|
||||
idx = int(info.get("index"))
|
||||
if self._is_output_device(idx):
|
||||
return idx
|
||||
except Exception:
|
||||
pass
|
||||
return None
|
||||
|
||||
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
|
||||
device_count = self._get_device_count()
|
||||
|
||||
if AUDIO_INPUT_DEVICE_INDEX is not None:
|
||||
idx = int(AUDIO_INPUT_DEVICE_INDEX)
|
||||
if 0 <= idx < device_count and is_input_device(idx):
|
||||
if 0 <= idx < device_count and self._is_input_device(idx):
|
||||
return idx
|
||||
raise RuntimeError(
|
||||
"Audio input initialization failed: invalid AUDIO_INPUT_DEVICE_INDEX="
|
||||
@@ -57,16 +126,9 @@ class AudioManager:
|
||||
)
|
||||
|
||||
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
|
||||
match_idx = self._find_device_by_name(AUDIO_INPUT_DEVICE_NAME, input_kind=True)
|
||||
if match_idx is not None:
|
||||
return match_idx
|
||||
|
||||
raise RuntimeError(
|
||||
"Audio input initialization failed: could not find an input device "
|
||||
@@ -76,17 +138,13 @@ class AudioManager:
|
||||
)
|
||||
|
||||
# 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
|
||||
default_idx = self._get_default_input_index()
|
||||
if default_idx is not None:
|
||||
return default_idx
|
||||
|
||||
# Fallback: first input device.
|
||||
for idx in range(device_count):
|
||||
if is_input_device(idx):
|
||||
if self._is_input_device(idx):
|
||||
return idx
|
||||
|
||||
raise RuntimeError(
|
||||
@@ -96,12 +154,185 @@ class AudioManager:
|
||||
+ self.describe_input_devices()
|
||||
)
|
||||
|
||||
def _resolve_output_device_index(self):
|
||||
if self.pa is None:
|
||||
return None
|
||||
|
||||
device_count = self._get_device_count()
|
||||
|
||||
if AUDIO_OUTPUT_DEVICE_INDEX is not None:
|
||||
idx = int(AUDIO_OUTPUT_DEVICE_INDEX)
|
||||
if 0 <= idx < device_count and self._is_output_device(idx):
|
||||
return idx
|
||||
raise RuntimeError(
|
||||
"Audio output initialization failed: invalid AUDIO_OUTPUT_DEVICE_INDEX="
|
||||
f"{AUDIO_OUTPUT_DEVICE_INDEX}. Available output devices:\n"
|
||||
+ self.describe_output_devices()
|
||||
)
|
||||
|
||||
if AUDIO_OUTPUT_DEVICE_NAME:
|
||||
match_idx = self._find_device_by_name(
|
||||
AUDIO_OUTPUT_DEVICE_NAME, input_kind=False
|
||||
)
|
||||
if match_idx is not None:
|
||||
return match_idx
|
||||
raise RuntimeError(
|
||||
"Audio output initialization failed: could not find an output device "
|
||||
f"matching AUDIO_OUTPUT_DEVICE_NAME={AUDIO_OUTPUT_DEVICE_NAME!r}. "
|
||||
"Available output devices:\n"
|
||||
+ self.describe_output_devices()
|
||||
)
|
||||
|
||||
default_idx = self._get_default_output_index()
|
||||
if default_idx is not None:
|
||||
return default_idx
|
||||
|
||||
for idx in range(device_count):
|
||||
if self._is_output_device(idx):
|
||||
return idx
|
||||
|
||||
raise RuntimeError(
|
||||
"Audio output initialization failed: no output devices found. "
|
||||
"Check speaker connection and PipeWire/PulseAudio. "
|
||||
"PortAudio devices:\n"
|
||||
+ self.describe_output_devices()
|
||||
)
|
||||
|
||||
def _ordered_input_candidates(self, preferred_index=None):
|
||||
candidates = []
|
||||
|
||||
def add(idx):
|
||||
if idx not in candidates:
|
||||
candidates.append(idx)
|
||||
|
||||
if preferred_index is not None:
|
||||
add(preferred_index)
|
||||
else:
|
||||
try:
|
||||
add(self.get_input_device_index())
|
||||
except Exception:
|
||||
pass
|
||||
add(self._get_default_input_index())
|
||||
add(None) # Let PortAudio decide default path.
|
||||
for idx in range(self._get_device_count()):
|
||||
if self._is_input_device(idx):
|
||||
add(idx)
|
||||
|
||||
return [idx for idx in candidates if idx is None or self._is_input_device(idx)]
|
||||
|
||||
def _ordered_output_candidates(self, preferred_index=None):
|
||||
candidates = []
|
||||
|
||||
def add(idx):
|
||||
if idx not in candidates:
|
||||
candidates.append(idx)
|
||||
|
||||
if preferred_index is not None:
|
||||
add(preferred_index)
|
||||
else:
|
||||
try:
|
||||
add(self.get_output_device_index())
|
||||
except Exception:
|
||||
pass
|
||||
add(self._get_default_output_index())
|
||||
add(None) # Let PortAudio decide default path.
|
||||
for idx in range(self._get_device_count()):
|
||||
if self._is_output_device(idx):
|
||||
add(idx)
|
||||
|
||||
return [idx for idx in candidates if idx is None or self._is_output_device(idx)]
|
||||
|
||||
def open_input_stream(
|
||||
self,
|
||||
*,
|
||||
rate: int,
|
||||
channels: int,
|
||||
format,
|
||||
frames_per_buffer: int,
|
||||
preferred_index=None,
|
||||
fallback_rates=None,
|
||||
):
|
||||
if self.pa is None:
|
||||
raise RuntimeError("PyAudio is not initialized")
|
||||
|
||||
fallback_rates = fallback_rates or []
|
||||
rates = [int(rate)] + [int(r) for r in fallback_rates if int(r) > 0 and int(r) != int(rate)]
|
||||
errors = []
|
||||
|
||||
for device_idx in self._ordered_input_candidates(preferred_index):
|
||||
for attempt_rate in rates:
|
||||
fb = max(
|
||||
64, int(round(frames_per_buffer * attempt_rate / max(1, int(rate))))
|
||||
)
|
||||
kwargs = {
|
||||
"rate": attempt_rate,
|
||||
"channels": channels,
|
||||
"format": format,
|
||||
"input": True,
|
||||
"frames_per_buffer": fb,
|
||||
}
|
||||
if device_idx is not None:
|
||||
kwargs["input_device_index"] = device_idx
|
||||
try:
|
||||
stream = self.pa.open(**kwargs)
|
||||
return stream, device_idx, attempt_rate
|
||||
except Exception as exc:
|
||||
errors.append(
|
||||
f"device={device_idx!r}, rate={attempt_rate}: {exc}"
|
||||
)
|
||||
|
||||
joined_errors = "\n".join(errors[:12])
|
||||
raise RuntimeError(
|
||||
"Audio input initialization failed. Tried multiple devices/rates.\n"
|
||||
f"{joined_errors}\nAvailable input devices:\n{self.describe_input_devices()}"
|
||||
)
|
||||
|
||||
def open_output_stream(
|
||||
self,
|
||||
*,
|
||||
rate: int,
|
||||
channels: int,
|
||||
format,
|
||||
preferred_index=None,
|
||||
fallback_rates=None,
|
||||
):
|
||||
if self.pa is None:
|
||||
raise RuntimeError("PyAudio is not initialized")
|
||||
|
||||
fallback_rates = fallback_rates or []
|
||||
rates = [int(rate)] + [int(r) for r in fallback_rates if int(r) > 0 and int(r) != int(rate)]
|
||||
errors = []
|
||||
|
||||
for device_idx in self._ordered_output_candidates(preferred_index):
|
||||
for attempt_rate in rates:
|
||||
kwargs = {
|
||||
"rate": attempt_rate,
|
||||
"channels": channels,
|
||||
"format": format,
|
||||
"output": True,
|
||||
}
|
||||
if device_idx is not None:
|
||||
kwargs["output_device_index"] = device_idx
|
||||
try:
|
||||
stream = self.pa.open(**kwargs)
|
||||
return stream, device_idx, attempt_rate
|
||||
except Exception as exc:
|
||||
errors.append(
|
||||
f"device={device_idx!r}, rate={attempt_rate}: {exc}"
|
||||
)
|
||||
|
||||
joined_errors = "\n".join(errors[:12])
|
||||
raise RuntimeError(
|
||||
"Audio output initialization failed. Tried multiple devices/rates.\n"
|
||||
f"{joined_errors}\nAvailable output devices:\n{self.describe_output_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)
|
||||
count = self._get_device_count()
|
||||
for idx in range(count):
|
||||
try:
|
||||
info = self.pa.get_device_info_by_index(idx)
|
||||
@@ -116,6 +347,26 @@ class AudioManager:
|
||||
break
|
||||
return "\n".join(items) if items else "<no input devices>"
|
||||
|
||||
def describe_output_devices(self, limit: int = 20) -> str:
|
||||
if self.pa is None:
|
||||
return "<PyAudio not initialized>"
|
||||
|
||||
items = []
|
||||
count = self._get_device_count()
|
||||
for idx in range(count):
|
||||
try:
|
||||
info = self.pa.get_device_info_by_index(idx)
|
||||
except Exception:
|
||||
continue
|
||||
max_out = int(info.get("maxOutputChannels") or 0)
|
||||
if max_out <= 0:
|
||||
continue
|
||||
name = str(info.get("name") or "").strip()
|
||||
items.append(f"[{idx}] {name} (out={max_out})")
|
||||
if len(items) >= limit:
|
||||
break
|
||||
return "\n".join(items) if items else "<no output devices>"
|
||||
|
||||
def cleanup(self):
|
||||
if self.pa:
|
||||
self.pa.terminate()
|
||||
|
||||
Reference in New Issue
Block a user