Add microphone picker and device management features

- Implement `list_input_devices` to retrieve and filter input-capable audio devices.
- Create `MicPickerDialog` for selecting and saving microphone settings.
- Add `set_env_value` function to update .env file with selected microphone index.
- Update UI components to integrate microphone selection functionality.
- Enhance tests for microphone device handling and environment variable management.
This commit is contained in:
2026-09-24 23:31:18 -06:00
parent ad9925da02
commit 54030a7b31
8 changed files with 283 additions and 245 deletions
+21
View File
@@ -126,6 +126,27 @@ def record_utterance(
return pcm
def list_input_devices() -> list[dict]:
"""Every input-capable device sounddevice can see, as
{"index", "name", "hostapi"}.
On Windows in particular, the *same* physical (or virtual) device is
normally listed once per host API — MME, DirectSound, WASAPI, WDM-KS —
all sharing (or near enough sharing) one name. That makes the name alone
an unreliable, often ambiguous way to pick a device (sounddevice matches
a string device as a substring and refuses to guess between ties), so the
tray's mic picker saves the numeric index instead, which this list exists
to let it show alongside a human-readable label."""
import sounddevice as sd
devices = []
for index, info in enumerate(sd.query_devices()):
if info.get("max_input_channels", 0) > 0:
hostapi = sd.query_hostapis(info["hostapi"])["name"]
devices.append({"index": index, "name": info["name"], "hostapi": hostapi})
return devices
def open_input_stream():
"""Real sounddevice input stream, imported lazily so pure-logic tests
(record_utterance with a fake stream) don't need PortAudio installed."""
+37 -1
View File
@@ -9,6 +9,7 @@ from __future__ import annotations
import os
import platform
import re
from pathlib import Path
HERE = Path(__file__).resolve().parent.parent # project root (one above bolt_pet/)
@@ -29,6 +30,26 @@ def _load_env() -> None:
_load_env()
def set_env_value(key: str, value: str) -> None:
"""Write *key=value* into .env, in place — replacing (or uncommenting)
an existing line for it, appending a new one otherwise. Used by the tray's
microphone picker so a device pick survives a restart without hand-editing
the file. Settings read once at import time (like MIC_DEVICE) still need
that restart to take effect; this only updates the file and os.environ."""
env_path = HERE / ".env"
lines = env_path.read_text(encoding="utf-8").splitlines() if env_path.exists() else []
pattern = re.compile(rf"^\s*#?\s*{re.escape(key)}\s*=")
new_line = f"{key}={value}"
for i, line in enumerate(lines):
if pattern.match(line):
lines[i] = new_line
break
else:
lines.append(new_line)
env_path.write_text("\n".join(lines) + "\n", encoding="utf-8")
os.environ[key] = value
def _node_name() -> str:
try:
return platform.node() or "desktop"
@@ -115,7 +136,22 @@ DIALOGUE_VOICES = os.environ.get("DIALOGUE_VOICES", "")
# ── mic / VAD (same tuning knobs as bolt_desk.py) ───────────────────────────
MIC_DEVICE = os.environ.get("MIC_DEVICE", "") or None # sounddevice name/index
def _mic_device(raw: str):
"""sounddevice treats an int index and a string name very differently: a
string is matched as a (fragile, often ambiguous) substring against every
device name, while an int is used as-is. .env only ever gives us a
string, so a numeric one — what the tray's mic picker saves, since the
same physical device usually shows up once per host API (MME,
DirectSound, WASAPI, ...) under the same name — has to be cast back to
int here, or it goes through the substring path and matches nothing (or
everything) instead."""
raw = raw.strip()
if not raw:
return None
return int(raw) if raw.lstrip("-").isdigit() else raw
MIC_DEVICE = _mic_device(os.environ.get("MIC_DEVICE", ""))
SPEAKER_DEVICE = os.environ.get("SPEAKER_DEVICE", "") or None
RMS_THRESHOLD = int(os.environ.get("VAD_RMS_THRESHOLD", "300"))
+33 -12
View File
@@ -11,13 +11,14 @@ from __future__ import annotations
import sys
from PySide6.QtCore import QThread
from PySide6.QtWidgets import QApplication
from PySide6.QtWidgets import QApplication, QMessageBox
from .. import config, updater
from ..controller import PetController
from ..hotkey import GlobalHotkey
from ..state import PetState
from .history_window import HistoryWindow
from .mic_picker import MicPickerDialog
from .pet_window import PetWindow
from .tray import PetTray
from .wake_tuner import WakeTunerWindow
@@ -57,6 +58,19 @@ def run() -> int:
window.publish_monitors()
window.copied.connect(lambda text: _log(f"Copied to clipboard: {text[:60]}"))
# The updater has already moved the checkout by the time this fires; all
# that's left is to let Qt tear down cleanly (so the mic and the tray
# icon are released) and then exec the new code. Doing the exec after
# app.exec() returns, rather than from the controller thread, is what
# guarantees the audio device is free before the new process opens it.
# Defined early so the mic picker (below) can trigger the same restart
# a self-update or self-restart does.
pending_restart = {"tag": None}
def _handle_restart(tag: str) -> None:
pending_restart["tag"] = tag
app.quit()
history_window = HistoryWindow(controller.history)
tuner_window = WakeTunerWindow(
get_threshold=controller.wake_threshold,
@@ -65,6 +79,23 @@ def run() -> int:
on_reset=controller.reset_wake_stats,
)
def _save_mic_device(value) -> None:
config.set_env_value("MIC_DEVICE", str(value) if value is not None else "")
_log(f"Microphone set to device #{value}." if value is not None else
"Microphone set to system default.")
answer = QMessageBox.question(
None, "Restart Bolt?",
"Restart now so Bolt starts listening on the new microphone?",
QMessageBox.Yes | QMessageBox.No,
)
if answer == QMessageBox.Yes:
_handle_restart("microphone change")
mic_picker = MicPickerDialog(
get_current=lambda: config.MIC_DEVICE,
on_save=_save_mic_device,
)
def _set_nap(napping: bool) -> None:
# Clicking the tray item pins the state; the schedule takes over again
# only after a restart or a `petctl nap off`.
@@ -82,6 +113,7 @@ def run() -> int:
on_set_nap=_set_nap,
on_show_history=history_window.show_refreshed,
on_show_wake_tuner=tuner_window.show_refreshed,
on_show_mic_picker=mic_picker.show_refreshed,
on_reset_voice=controller.reset_voice,
)
@@ -104,17 +136,6 @@ def run() -> int:
elif hotkey.running:
_log(f"Push-to-talk: {config.PUSH_TO_TALK_HOTKEY}")
# The updater has already moved the checkout by the time this fires; all
# that's left is to let Qt tear down cleanly (so the mic and the tray
# icon are released) and then exec the new code. Doing the exec after
# app.exec() returns, rather than from the controller thread, is what
# guarantees the audio device is free before the new process opens it.
pending_restart = {"tag": None}
def _handle_restart(tag: str) -> None:
pending_restart["tag"] = tag
app.quit()
controller.restart_requested.connect(_handle_restart)
def _shutdown() -> None:
+98
View File
@@ -0,0 +1,98 @@
"""Microphone picker — MIC_DEVICE in .env is a device index most people have
no way to look up short of running a script. This dialog lists whatever
sounddevice can actually see, lets you click one, and writes its index back
to .env so it survives a restart.
The index, not the name, is what gets saved: the same device is commonly
listed once per host API (MME, DirectSound, WASAPI, WDM-KS, ...) under the
same or a near-identical name, which makes a bare name an ambiguous pick that
sounddevice will flatly refuse to resolve (see audio/mic.list_input_devices).
The index is exact.
The mic is only ever opened once, at startup (see controller.run()), so a
pick made here still needs a restart before Bolt actually uses it — this
dialog just makes finding and saving the right device painless, and offers
to do the restart for you.
"""
from __future__ import annotations
from typing import Callable, Optional
from PySide6.QtCore import Qt
from PySide6.QtWidgets import (
QDialog, QDialogButtonBox, QHBoxLayout, QLabel, QListWidget,
QListWidgetItem, QPushButton, QVBoxLayout,
)
from ..audio.mic import list_input_devices
_DEFAULT_LABEL = "System default"
class MicPickerDialog(QDialog):
def __init__(
self,
get_current: Callable[[], Optional[int]],
on_save: Callable[[Optional[int]], None],
):
super().__init__()
self._get_current = get_current
self._on_save = on_save
self.setWindowTitle("Bolt — select microphone")
self.resize(440, 320)
self._list = QListWidget()
refresh_button = QPushButton("Refresh")
refresh_button.clicked.connect(self.refresh)
top = QHBoxLayout()
top.addWidget(QLabel("Pick the microphone Bolt should listen on:"))
top.addStretch(1)
top.addWidget(refresh_button)
buttons = QDialogButtonBox(QDialogButtonBox.Save | QDialogButtonBox.Cancel)
buttons.accepted.connect(self._save)
buttons.rejected.connect(self.reject)
layout = QVBoxLayout(self)
layout.addLayout(top)
layout.addWidget(self._list)
layout.addWidget(buttons)
def refresh(self) -> None:
self._list.clear()
default_item = QListWidgetItem(_DEFAULT_LABEL)
default_item.setData(Qt.UserRole, None)
self._list.addItem(default_item)
current = self._get_current()
selected = default_item
try:
devices = list_input_devices()
except Exception as exc:
self._list.addItem(f"Could not list devices: {exc}")
devices = []
for device in devices:
item = QListWidgetItem(
f"{device['name']} [{device['hostapi']}] (#{device['index']})"
)
item.setData(Qt.UserRole, device["index"])
self._list.addItem(item)
if current == device["index"]:
selected = item
self._list.setCurrentItem(selected)
def _save(self) -> None:
item = self._list.currentItem()
value = item.data(Qt.UserRole) if item is not None else None
self._on_save(value)
self.accept()
def show_refreshed(self) -> None:
self.refresh()
self.show()
self.raise_()
self.activateWindow()
+6
View File
@@ -47,6 +47,7 @@ class PetTray(QSystemTrayIcon):
on_set_nap: Optional[Callable[[bool], None]] = None,
on_show_history: Optional[Callable[[], None]] = None,
on_show_wake_tuner: Optional[Callable[[], None]] = None,
on_show_mic_picker: Optional[Callable[[], None]] = None,
on_reset_voice: Optional[Callable[[], None]] = None,
parent=None,
):
@@ -104,6 +105,11 @@ class PetTray(QSystemTrayIcon):
tuner_action.triggered.connect(on_show_wake_tuner)
menu.addAction(tuner_action)
if on_show_mic_picker is not None:
mic_action = QAction("Select microphone…", menu)
mic_action.triggered.connect(on_show_mic_picker)
menu.addAction(mic_action)
# Only ever enabled while a server-picked voice (speak_as) is in use —
# it's the way back from "talk like a pirate", which nothing else
# undoes short of a restart.