refactor: Remove speaker monitor capture and channel selection features
- Remove monitor/speaker capture UI from sidebar - Remove channel selection dropdown - Remove monitor-related state (monitor_enabled, monitor_index, monitor_list, channel_select) - Remove monitor-related functions from audio_devices.py - Simplify resample_to_mono_16k to always mix channels - Clean up bridge.py device selection logic - Remove debug output on startup
+15 −194 · 7 files changed
-
src/core.py+1 −3 -
src/core_parts/audio_devices.py+7 −126 -
src/core_parts/processing.py+2 −3 -
src/whispering_ui/bridge.py+3 −22 -
src/whispering_ui/components/sidebar.py+1 −26 -
src/whispering_ui/main.py+1 −8 -
src/whispering_ui/state.py+0 −6
@@ -3,10 +3,8 @@ from core_parts.paragraph_detector import ParagraphDetector from core_parts.audio_devices import ( TARGET_SAMPLE_RATE, SAMPLE_WIDTH, CHUNK_DURATION, - CHANNEL_MIX, CHANNEL_LEFT, CHANNEL_RIGHT, get_preferred_hostapi_index, get_mic_names, get_default_device_index, - get_mic_index, get_device_info, audio_to_wav_bytes, resample_to_mono_16k, - get_monitor_names, get_all_input_devices + get_mic_index, get_device_info, audio_to_wav_bytes, resample_to_mono_16k ) from core_parts.processing import ( translate, parse_ai_proofread_translate, ai_translate, proc @@ -8,11 +8,6 @@ TARGET_SAMPLE_RATE = 16000 SAMPLE_WIDTH = 2 # 16-bit audio CHUNK_DURATION = 0.1 # seconds per chunk -# Channel selection options -CHANNEL_MIX = "mix" # Mix all channels to mono (default) -CHANNEL_LEFT = "left" # Use only left channel -CHANNEL_RIGHT = "right" # Use only right channel - def get_preferred_hostapi_index(): """Find the best host API: ALSA only (JACK causes crashes).""" @@ -59,109 +54,6 @@ def get_mic_names(): return mics -def _extract_friendly_name(device_name): - """Extract a friendly display name from a PulseAudio/PipeWire device name. - - Examples: - 'alsa_output.usb-MOTU_828_828E0238SD-00.HiFi__Headphones1__sink.monitor' - -> '828 Headphones (1)' - - 'Monitor of 828 Main Out A (1-2)' - -> '828 Main Out A (1-2)' - """ - name = device_name - - # If it starts with "Monitor of ", extract the rest - if name.lower().startswith('monitor of '): - return name[11:] # Remove "Monitor of " - - # Try to extract from PulseAudio-style names - # Pattern: ...HiFi__DeviceName__sink.monitor - import re - - # Look for pattern like __SomeName__sink - match = re.search(r'__([^_]+\d*)__sink', name) - if match: - friendly = match.group(1) - # Convert camelCase/numbers to spaced format - # Headphones1 -> Headphones (1) - friendly = re.sub(r'([a-zA-Z])(\d+)$', r'\1 (\2)', friendly) - # Add device prefix if we can find it (like "828") - prefix_match = re.search(r'usb-([^_]+)_(\d+)', name) - if prefix_match: - model = prefix_match.group(2) - return f"{model} {friendly}" - return friendly - - # Fallback: just return the original but truncated if too long - if len(name) > 50: - return name[:47] + "..." - return name - - -def debug_list_all_devices(): - """Print all audio devices for debugging purposes.""" - devices_list = sd.query_devices() - hostapis = sd.query_hostapis() - - print("\n=== All Audio Devices ===") - for i, d in enumerate(devices_list): - api_name = hostapis[d['hostapi']]['name'] - in_ch = d['max_input_channels'] - out_ch = d['max_output_channels'] - print(f" [{i}] {d['name']}") - print(f" API: {api_name}, In: {in_ch}, Out: {out_ch}") - print("========================\n") - - -def get_monitor_names(debug=False): - """Get list of speaker/output monitor device names (for capturing system audio output). - - Monitor devices capture what's being played through speakers/headphones. - Returns tuples of (device_index, friendly_display_name). - """ - devices_list = sd.query_devices() - - if debug: - debug_list_all_devices() - - monitors = [] - - for i, d in enumerate(devices_list): - if d['max_input_channels'] > 0: - name = d['name'] - name_lower = name.lower() - api_name = sd.query_hostapis(d['hostapi'])['name'].lower() - - # Skip JACK (causes crashes) - if 'jack' in api_name: - continue - - # Look for monitor devices (PulseAudio/PipeWire convention) - if 'monitor' in name_lower: - friendly_name = _extract_friendly_name(name) - monitors.append((i, friendly_name)) - if debug: - print(f" Found monitor: [{i}] {name} -> '{friendly_name}'") - - if debug and not monitors: - print(" No monitor devices found!") - - return monitors - - -def get_all_input_devices(): - """Get both microphones and monitor devices, categorized. - - Returns: - dict with 'mics' and 'monitors' keys, each containing list of (index, name) tuples - """ - return { - 'mics': get_mic_names(), - 'monitors': get_monitor_names() - } - - def get_default_device_index(): """Get the best default device (pipewire or pulse, not the actual 'default').""" devices_list = sd.query_devices() @@ -202,20 +94,20 @@ def get_device_info(device_index): if device_index is None: # Use smart default (pipewire/pulse) device_index = get_default_device_index() - + if device_index is None: # Ultimate fallback return 48000, 1 - + device_info = sd.query_devices(device_index) - + # Get native sample rate (or use default) sample_rate = int(device_info.get('default_samplerate', 48000)) - + # Get channels - limit to 2 to avoid issues with high channel count devices max_channels = int(device_info.get('max_input_channels', 1)) channels = min(2, max(1, max_channels)) # Use 1-2 channels max - + return sample_rate, channels @@ -231,31 +123,20 @@ def audio_to_wav_bytes(audio_data, sample_rate, sample_width, channels=1): return buffer -def resample_to_mono_16k(data, orig_rate, orig_channels, channel_select=CHANNEL_MIX): +def resample_to_mono_16k(data, orig_rate, orig_channels): """Convert audio to mono 16kHz for Whisper. Args: data: Raw audio data (numpy array or bytes) orig_rate: Original sample rate orig_channels: Number of channels in original audio - channel_select: Which channel to use - CHANNEL_MIX (average all), - CHANNEL_LEFT (left only), or CHANNEL_RIGHT (right only) """ # Ensure we have a copy to avoid memory issues audio = np.array(data, dtype=np.float32, copy=True) / 32768.0 # Convert to mono if stereo/multi-channel if orig_channels > 1 and len(audio.shape) > 1: - if channel_select == CHANNEL_LEFT: - # Use only left channel (index 0) - audio = audio[:, 0] - elif channel_select == CHANNEL_RIGHT: - # Use only right channel (index 1, or last channel if mono) - channel_idx = min(1, audio.shape[1] - 1) - audio = audio[:, channel_idx] - else: - # Default: mix all channels - audio = audio.mean(axis=1) + audio = audio.mean(axis=1) else: audio = audio.flatten() @@ -20,7 +20,6 @@ except ImportError: from .paragraph_detector import ParagraphDetector from .audio_devices import ( TARGET_SAMPLE_RATE, SAMPLE_WIDTH, CHUNK_DURATION, - CHANNEL_MIX, CHANNEL_LEFT, CHANNEL_RIGHT, get_default_device_index, get_device_info, audio_to_wav_bytes, resample_to_mono_16k ) @@ -108,7 +107,7 @@ def ai_translate(text, ai_processor): return (text, f"AI processing error: {str(e)}") -def proc(index, model, vad, memory, patience, timeout, prompt, source, target, tsres_queue, tlres_queue, ready, device="cpu", error=None, level=None, para_detect=True, para_threshold_std=1.5, para_min_pause=0.8, para_max_chars=500, para_max_words=100, ai_processor=None, ai_process_interval=2, ai_process_words=None, ai_trigger_mode="time", silence_timeout=60, prres_queue=None, auto_stop_enabled=False, auto_stop_minutes=5, manual_trigger=None, use_google_translate=True, channel_select=CHANNEL_MIX): +def proc(index, model, vad, memory, patience, timeout, prompt, source, target, tsres_queue, tlres_queue, ready, device="cpu", error=None, level=None, para_detect=True, para_threshold_std=1.5, para_min_pause=0.8, para_max_chars=500, para_max_words=100, ai_processor=None, ai_process_interval=2, ai_process_words=None, ai_trigger_mode="time", silence_timeout=60, prres_queue=None, auto_stop_enabled=False, auto_stop_minutes=5, manual_trigger=None, use_google_translate=True): # Create paragraph detector if enabled para_detector = ParagraphDetector( threshold_std=para_threshold_std, @@ -463,7 +462,7 @@ def proc(index, model, vad, memory, patience, timeout, prompt, source, target, t level[0] = min(100, int(rms / 328 * 100)) # Convert to mono 16kHz for Whisper - mono_16k = resample_to_mono_16k(data_copy, sample_rate, channels, channel_select) + mono_16k = resample_to_mono_16k(data_copy, sample_rate, channels) frame_queue.put(mono_16k) frame_queue.put(None) @@ -88,17 +88,10 @@ class ProcessingBridge: # Initialize AI processor if enabled self.ai_processor = self._create_ai_processor() - # Get device index (mic or monitor based on mode) + # Get mic device index mic_index = None try: - if self.state.monitor_enabled and self.state.monitor_list: - # Use monitor device for speaker capture - if self.state.monitor_index == 0 or self.state.monitor_index > len(self.state.monitor_list): - # First monitor in list (or fallback if index out of range) - mic_index = self.state.monitor_list[0][0] - else: - mic_index = self.state.monitor_list[self.state.monitor_index - 1][0] - elif self.state.mic_index == 0 or not self.state.mic_list: + if self.state.mic_index == 0 or not self.state.mic_list: mic_index = core.get_default_device_index() else: idx = min(self.state.mic_index - 1, len(self.state.mic_list) - 1) @@ -107,13 +100,6 @@ class ProcessingBridge: print(f"[WARN] Device selection error: {e}, falling back to default") mic_index = core.get_default_device_index() - # Get channel selection constant - channel_select = core.CHANNEL_MIX - if self.state.channel_select == "left": - channel_select = core.CHANNEL_LEFT - elif self.state.channel_select == "right": - channel_select = core.CHANNEL_RIGHT - # Get translation target (None if "none") target = None if self.state.target_language == "none" else self.state.target_language source = None if self.state.source_language == "auto" else self.state.source_language @@ -177,8 +163,7 @@ class ProcessingBridge: 'auto_stop_enabled': self.state.auto_stop_enabled, 'auto_stop_minutes': self.state.auto_stop_minutes, 'manual_trigger': self.manual_trigger_requested, - 'use_google_translate': use_google_translate, - 'channel_select': channel_select + 'use_google_translate': use_google_translate }, daemon=True ).start() @@ -584,10 +569,6 @@ class ProcessingBridge: """Refresh the list of available microphones.""" self.state.mic_list = core.get_mic_names() - def refresh_monitors(self): - """Refresh the list of available monitor (speaker capture) devices.""" - self.state.monitor_list = core.get_monitor_names() - def _get_config_for_logging(self) -> dict: """Get current configuration for logging purposes.""" config = { @@ -39,37 +39,12 @@ def create_sidebar(state: AppState, bridge: ProcessingBridge, output_container=N def refresh_mics(): bridge.refresh_mics() - bridge.refresh_monitors() mic_select.options = ["(system default)"] + [name for idx, name in state.mic_list] mic_select.update() - monitor_select.options = ["(first available)"] + [name for idx, name in state.monitor_list] - monitor_select.update() ui.button(icon='refresh', on_click=refresh_mics).props('flat dense round size=sm') - # === SPEAKER CAPTURE (MONITOR) SECTION === - with ui.row().classes('items-center w-full gap-1'): - monitor_cb = ui.checkbox('Speaker', value=state.monitor_enabled).props('dense') - monitor_cb.on_value_change(lambda e: setattr(state, 'monitor_enabled', e.value)) - - monitor_display = ["(first available)"] + [name for idx, name in state.monitor_list] - monitor_select = ui.select( - options=monitor_display, - value=monitor_display[0] if monitor_display else None - ).classes('flex-grow').props('dense') - monitor_select.on_value_change(lambda e: setattr(state, 'monitor_index', - monitor_select.options.index(e.value) if e.value in monitor_select.options else 0)) - - # Channel selection - with ui.row().classes('items-center w-full gap-1'): - ui.label('Channel:').classes('text-xs w-14') - channel_select = ui.select( - options=["mix", "left", "right"], - value=state.channel_select - ).classes('flex-grow').props('dense') - channel_select.on_value_change(lambda e: setattr(state, 'channel_select', e.value)) - - # === CONTROL SECTION (MOVED UP) === + # === CONTROL SECTION === control_btn = ui.button( 'Start', on_click=lambda: _toggle_recording(state, bridge, control_btn, level_progress, status_label) @@ -56,20 +56,13 @@ def main(): state.auto_stop_enabled = settings.get("auto_stop_enabled", False) state.auto_stop_minutes = settings.get("auto_stop_minutes", 5) - # Initialize microphone and monitor lists + # Initialize microphone list try: state.mic_list = core.get_mic_names() except Exception as e: print(f"Error getting mic list: {e}") state.mic_list = [] - try: - # Use debug=True to see all devices on startup - state.monitor_list = core.get_monitor_names(debug=True) - except Exception as e: - print(f"Error getting monitor list: {e}") - state.monitor_list = [] - # Check AI availability try: from ai_config import load_ai_config @@ -16,12 +16,6 @@ class AppState: mic_index: int = 0 mic_list: List[Tuple[int, str]] = field(default_factory=list) - # === Monitor (Speaker Capture) Settings === - monitor_enabled: bool = False # Whether to capture speaker output instead of mic - monitor_index: int = 0 - monitor_list: List[Tuple[int, str]] = field(default_factory=list) - channel_select: str = "mix" # "mix", "left", or "right" - # === Model Settings === model: str = "large-v3" vad_enabled: bool = True