feat: Add audio file transcription feature
Add ability to transcribe audio files instead of live microphone input: - Support for WAV, MP3, FLAC, OGG, M4A, AAC, WMA, OPUS formats - Single file or batch processing of multiple files - Progress tracking with current file display - Results displayed with file headers in Whisper output - Uses same model settings as live transcription
+524 −3 · 7 files changed
-
src/core.py+3 −2 -
src/core_parts/audio_devices.py+85 −0 -
src/core_parts/processing.py+106 −1 -
src/whispering_ui/bridge.py+160 −0 -
src/whispering_ui/components/help.py+31 −0 -
src/whispering_ui/components/sidebar.py+132 −0 -
src/whispering_ui/state.py+7 −0
@@ -4,10 +4,11 @@ from core_parts.paragraph_detector import ParagraphDetector from core_parts.audio_devices import ( TARGET_SAMPLE_RATE, SAMPLE_WIDTH, CHUNK_DURATION, 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_mic_index, get_device_info, audio_to_wav_bytes, resample_to_mono_16k, + load_audio_file, get_audio_files_from_directory, is_audio_file ) from core_parts.processing import ( - translate, parse_ai_proofread_translate, ai_translate, proc + translate, parse_ai_proofread_translate, ai_translate, proc, proc_file ) # Re-export constants @@ -1,7 +1,9 @@ import io +import os import wave import numpy as np import sounddevice as sd +import soundfile as sf # Audio settings - Whisper expects 16kHz mono TARGET_SAMPLE_RATE = 16000 @@ -155,3 +157,86 @@ def resample_to_mono_16k(data, orig_rate, orig_channels): # Convert back to int16 audio = (audio * 32768.0).clip(-32768, 32767).astype(np.int16) return audio.tobytes() + + +def load_audio_file(file_path: str) -> tuple: + """ + Load an audio file and return audio data ready for Whisper. + + Supports: WAV, MP3, FLAC, OGG, M4A, and other formats supported by soundfile/libsndfile. + + Args: + file_path: Path to the audio file + + Returns: + tuple: (audio_data_bytes, sample_rate, duration_seconds) + + Raises: + FileNotFoundError: If file doesn't exist + ValueError: If file format is not supported + """ + if not os.path.exists(file_path): + raise FileNotFoundError(f"Audio file not found: {file_path}") + + try: + # Read audio file using soundfile (supports many formats) + audio_data, sample_rate = sf.read(file_path, dtype='float32') + + # Get number of channels + if len(audio_data.shape) == 1: + channels = 1 + else: + channels = audio_data.shape[1] + + # Calculate duration + duration = len(audio_data) / sample_rate + + # Convert to int16 + audio_int16 = (audio_data * 32768.0).clip(-32768, 32767).astype(np.int16) + + # Convert to mono 16kHz for Whisper + mono_16k = resample_to_mono_16k(audio_int16, sample_rate, channels) + + return mono_16k, TARGET_SAMPLE_RATE, duration + + except Exception as e: + raise ValueError(f"Failed to load audio file: {str(e)}") + + +def get_audio_files_from_directory(directory_path: str, recursive: bool = False) -> list: + """ + Get all audio files from a directory. + + Args: + directory_path: Path to directory + recursive: Whether to search subdirectories + + Returns: + List of audio file paths + """ + audio_extensions = {'.wav', '.mp3', '.flac', '.ogg', '.m4a', '.aac', '.wma', '.opus'} + audio_files = [] + + if not os.path.isdir(directory_path): + raise ValueError(f"Not a directory: {directory_path}") + + if recursive: + for root, dirs, files in os.walk(directory_path): + for file in files: + ext = os.path.splitext(file)[1].lower() + if ext in audio_extensions: + audio_files.append(os.path.join(root, file)) + else: + for file in os.listdir(directory_path): + ext = os.path.splitext(file)[1].lower() + if ext in audio_extensions: + audio_files.append(os.path.join(directory_path, file)) + + return sorted(audio_files) + + +def is_audio_file(file_path: str) -> bool: + """Check if a file is a supported audio file based on extension.""" + audio_extensions = {'.wav', '.mp3', '.flac', '.ogg', '.m4a', '.aac', '.wma', '.opus'} + ext = os.path.splitext(file_path)[1].lower() + return ext in audio_extensions @@ -20,7 +20,8 @@ except ImportError: from .paragraph_detector import ParagraphDetector from .audio_devices import ( TARGET_SAMPLE_RATE, SAMPLE_WIDTH, CHUNK_DURATION, - get_default_device_index, get_device_info, audio_to_wav_bytes, resample_to_mono_16k + get_default_device_index, get_device_info, audio_to_wav_bytes, resample_to_mono_16k, + load_audio_file ) def translate(text, source, target, timeout): @@ -474,3 +475,107 @@ def proc(index, model, vad, memory, patience, timeout, prompt, source, target, t error[0] = str(e) finally: ready[0] = None + + +def proc_file(file_paths, model, vad, source, tsres_queue, ready, device="cpu", error=None, progress=None, current_file=None, para_detect=True, para_threshold_std=1.5, para_min_pause=0.8, para_max_chars=500, para_max_words=100): + """ + Process audio files for transcription (batch mode). + + Args: + file_paths: List of audio file paths to transcribe + model: Whisper model size (e.g., 'large-v3') + vad: Enable voice activity detection + source: Source language (None for auto-detect) + tsres_queue: Queue for transcription results (done_text, curr_text) + ready: Control flag [None=stopped, False=stopping, True=running] + device: 'cpu' or 'cuda' + error: Error message container [str or None] + progress: Progress percentage container [0-100] + current_file: Currently processing file name container [str] + para_detect: Enable paragraph detection + para_threshold_std: Paragraph detection threshold + para_min_pause: Minimum pause for paragraph break + para_max_chars: Max characters before forced paragraph break + para_max_words: Max words before forced paragraph break + + Output Queue Format: + - (done_text, "") for each completed file + - None when all files are processed + """ + # Create paragraph detector if enabled + para_detector = ParagraphDetector( + threshold_std=para_threshold_std, + min_pause=para_min_pause, + max_chars=para_max_chars, + max_words=para_max_words + ) if para_detect else None + + try: + # Load model + whisper_model = WhisperModel(model, device=device) + ready[0] = True + + total_files = len(file_paths) + + for file_index, file_path in enumerate(file_paths): + # Check if stop requested + if ready[0] is False or ready[0] is None: + break + + # Update current file name + if current_file is not None: + import os + current_file[0] = os.path.basename(file_path) + + # Update progress + if progress is not None: + progress[0] = int((file_index / total_files) * 100) + + try: + # Load audio file + audio_bytes, sample_rate, duration = load_audio_file(file_path) + + # Create WAV in memory for Whisper + audio_file = audio_to_wav_bytes(audio_bytes, sample_rate, SAMPLE_WIDTH, channels=1) + + # Transcribe + segments, info = whisper_model.transcribe( + audio_file, + language=source, + vad_filter=vad + ) + + # Collect all segments + segments_list = list(segments) + + # Process with paragraph detection if enabled + if para_detector and segments_list: + transcription = para_detector.process_segments(segments_list, 0.0) + else: + transcription = "".join(segment.text for segment in segments_list) + + # Add file header + import os + file_name = os.path.basename(file_path) + result_text = f"\n--- {file_name} ---\n{transcription.strip()}\n" + + # Send result + tsres_queue.put((result_text, "")) + + except Exception as file_error: + # Send error for this file but continue with others + import os + file_name = os.path.basename(file_path) + error_text = f"\n--- {file_name} ---\n[Error: {str(file_error)}]\n" + tsres_queue.put((error_text, "")) + + # Final progress update + if progress is not None: + progress[0] = 100 + + except Exception as e: + if error is not None: + error[0] = str(e) + finally: + tsres_queue.put(None) + ready[0] = None @@ -37,10 +37,17 @@ class ProcessingBridge: # Polling timer self.poll_timer = None + self._file_poll_timer = None # File transcription polling timer # AI processor reference self.ai_processor = None + # File transcription control + self._file_ready = [None] + self._file_error = [None] + self._file_progress = [0] + self._file_current = [""] + # TTS session tracking self.tts_controller = None self.tts_session_text = "" @@ -630,3 +637,156 @@ class ProcessingBridge: self._whisper_committed = "" self._ai_committed = "" self._translation_committed = "" + + # === FILE TRANSCRIPTION METHODS === + + def start_file_transcription(self, file_paths: list): + """Start transcription of audio files. + + Args: + file_paths: List of audio file paths to transcribe + """ + if self.state.file_transcription_active or self.state.is_recording: + return + + if not file_paths: + self.state.error_message = "No audio files selected" + return + + # Reset state + self.state.file_transcription_mode = True + self.state.file_transcription_active = True + self.state.file_transcription_paths = file_paths + self.state.file_transcription_progress = 0 + self.state.file_transcription_current_file = "" + self.state.error_message = None + self.state.status_message = f"Starting transcription of {len(file_paths)} file(s)..." + + # Clear outputs + self.clear_outputs() + + # Control flags for file transcription + self._file_ready = [False] + self._file_error = [None] + self._file_progress = [0] + self._file_current = [""] + + # Get source language + source = None if self.state.source_language == "auto" else self.state.source_language + + # Start file processing thread + threading.Thread( + target=core.proc_file, + args=( + file_paths, + self.state.model, + self.state.vad_enabled, + source, + self.ts_queue, + self._file_ready, + self.state.device, + self._file_error, + self._file_progress, + self._file_current, + self.state.para_detect_enabled + ), + daemon=True + ).start() + + # Start polling + self._start_file_polling() + + def stop_file_transcription(self): + """Stop file transcription if running.""" + if not self.state.file_transcription_active: + return + + self._file_ready[0] = False + self.state.status_message = "Stopping..." + + # Stop polling + if self._file_poll_timer: + self._file_poll_timer.deactivate() + self._file_poll_timer = None + + # Wait for thread to stop + threading.Thread(target=self._wait_for_file_stop, daemon=True).start() + + def _wait_for_file_stop(self): + """Wait for file processing thread to stop.""" + while self._file_ready[0] is not None: + time.sleep(0.1) + + self.state.file_transcription_active = False + self.state.file_transcription_progress = 0 + self.state.file_transcription_current_file = "" + self.state.status_message = "File transcription stopped" + + def _start_file_polling(self): + """Start polling for file transcription results.""" + self._file_poll_timer = ui.timer(0.1, self._poll_file_queues) + + def _poll_file_queues(self): + """Poll result queues for file transcription.""" + # Check if stopped + if self._file_ready[0] is None: + if self._file_poll_timer: + self._file_poll_timer.deactivate() + self._file_poll_timer = None + self.state.file_transcription_active = False + self.state.file_transcription_progress = 100 + self.state.status_message = "File transcription complete" + + # Check for errors + if self._file_error[0]: + self.state.error_message = str(self._file_error[0]) + self.state.status_message = f"Error: {self._file_error[0]}" + + return + + # Update progress and current file + self.state.file_transcription_progress = self._file_progress[0] + self.state.file_transcription_current_file = self._file_current[0] + self.state.status_message = f"Processing: {self._file_current[0]}" if self._file_current[0] else "Processing..." + + # Poll whisper queue for file transcription results + while self.ts_queue: + res = self.ts_queue.get() + if res: + done, curr = res + if done: + # Append to whisper text + self.state.whisper_text += done + self._whisper_committed += done + + def add_files_for_transcription(self, file_paths: list): + """Add files to the transcription queue. + + Args: + file_paths: List of audio file paths + """ + valid_files = [p for p in file_paths if core.is_audio_file(p)] + self.state.file_transcription_paths.extend(valid_files) + return len(valid_files) + + def add_directory_for_transcription(self, directory_path: str, recursive: bool = False): + """Add all audio files from a directory. + + Args: + directory_path: Path to directory + recursive: Search subdirectories + + Returns: + Number of files added + """ + try: + files = core.get_audio_files_from_directory(directory_path, recursive) + self.state.file_transcription_paths.extend(files) + return len(files) + except Exception as e: + self.state.error_message = f"Error scanning directory: {str(e)}" + return 0 + + def clear_file_list(self): + """Clear the list of files to transcribe.""" + self.state.file_transcription_paths = [] @@ -67,6 +67,37 @@ See AI_SETUP.md for details.""", **Setup:** See INSTALL_TTS.md""", + "file_transcription": """**File Transcription:** Transcribe audio files (batch mode) + +**Add Files:** Select one or more audio files to transcribe +- Supports: WAV, MP3, FLAC, OGG, M4A, AAC, WMA, OPUS +- Max file size: 500MB per file +- Multiple files can be added and processed sequentially + +**Transcribe:** Start processing the selected files +- Uses same model settings as microphone transcription +- Progress shown with percentage and current file name +- Results appear in the Whisper text output + +**Stop:** Cancel file transcription (partial results preserved) + +**Clear (X):** Remove all files from the queue + +**Output Format:** +``` +--- filename.mp3 --- +[transcribed text] + +--- another_file.wav --- +[transcribed text] +``` + +**Tips:** +- Stop microphone recording before file transcription +- Large files may take longer depending on model size +- VAD helps filter silent sections +- Paragraph detection adds natural line breaks""", + "logging": """**Save logs:** Save session data to structured log files **Features:** @@ -74,6 +74,106 @@ def create_sidebar(state: AppState, bridge: ProcessingBridge, output_container=N ui.separator().classes('my-1') + # === FILE TRANSCRIPTION SECTION === + with ui.row().classes('items-center justify-between w-full'): + ui.label('File Transcription').classes('font-bold text-sm') + ui.button(icon='help_outline', on_click=lambda: show_help_dialog('file_transcription')).props('flat dense round size=sm') + + # File selection controls + file_list_label = ui.label('No files selected').classes('text-xs text-gray-400 truncate w-full') + + def update_file_list_display(): + """Update the file list display.""" + count = len(state.file_transcription_paths) + if count == 0: + file_list_label.text = 'No files selected' + elif count == 1: + import os + file_list_label.text = os.path.basename(state.file_transcription_paths[0]) + else: + file_list_label.text = f'{count} files selected' + + # File upload for multiple files + def on_file_upload(e): + """Handle file upload for transcription.""" + try: + import os + from pathlib import Path + + # Create uploads directory if needed + uploads_dir = Path("uploads") + uploads_dir.mkdir(exist_ok=True) + + # Read and save uploaded file + filename = Path(e.name).name + permanent_path = uploads_dir / filename + + with open(e.name, 'rb') as f: + content = f.read() + + with open(permanent_path, 'wb') as f: + f.write(content) + + # Add to transcription list + if core.is_audio_file(str(permanent_path)): + state.file_transcription_paths.append(str(permanent_path)) + update_file_list_display() + ui.notify(f"Added: {filename}", type='positive') + else: + ui.notify(f"Not an audio file: {filename}", type='warning') + except Exception as ex: + ui.notify(f"Error: {ex}", type='negative') + + with ui.row().classes('items-center w-full gap-1'): + # Hidden file upload + file_upload = ui.upload( + on_upload=on_file_upload, + auto_upload=True, + max_file_size=500_000_000, # 500MB + multiple=True + ).props('accept=audio/*').classes('hidden') + + ui.button('Add Files', on_click=lambda: file_upload.run_method('pickFiles')).classes('flex-grow').props('dense') + ui.button(icon='clear', on_click=lambda: _clear_file_list(state, file_list_label, update_file_list_display)).props('flat dense round size=sm') + + # Progress bar for file transcription + file_progress = ui.linear_progress(value=0, show_value=False).classes('w-full') + file_progress.set_visibility(False) + + # Current file being processed + file_current_label = ui.label('').classes('text-xs text-blue-400 truncate w-full') + + # Start/Stop file transcription buttons + with ui.row().classes('items-center w-full gap-1'): + file_start_btn = ui.button( + 'Transcribe', + on_click=lambda: _start_file_transcription(state, bridge, file_start_btn, file_stop_btn, file_progress) + ).classes('flex-grow').props('dense color=positive') + + file_stop_btn = ui.button( + 'Stop', + on_click=lambda: _stop_file_transcription(state, bridge, file_start_btn, file_stop_btn, file_progress) + ).classes('w-16').props('dense color=negative') + file_stop_btn.set_enabled(False) + + # Update file transcription UI periodically + def update_file_ui(): + if state.file_transcription_active: + file_progress.set_visibility(True) + file_progress.value = state.file_transcription_progress / 100.0 + file_current_label.text = f"Processing: {state.file_transcription_current_file}" if state.file_transcription_current_file else "" + file_start_btn.set_enabled(False) + file_stop_btn.set_enabled(True) + else: + file_progress.set_visibility(state.file_transcription_progress > 0 and state.file_transcription_progress < 100) + file_current_label.text = "" + file_start_btn.set_enabled(len(state.file_transcription_paths) > 0) + file_stop_btn.set_enabled(False) + + ui.timer(0.2, update_file_ui) + + ui.separator().classes('my-1') + # === TOGGLE TEXT BUTTON === toggle_btn = ui.button( 'Show Text ▶' if not state.text_visible else 'Hide Text ◀', @@ -543,3 +643,35 @@ def _set_controls_enabled(controls, enabled: bool): ctrl.props(remove='disable') else: ctrl.props('disable') + + +def _clear_file_list(state: AppState, file_list_label, update_fn): + """Clear the file transcription list.""" + state.file_transcription_paths = [] + update_fn() + ui.notify("File list cleared") + + +def _start_file_transcription(state: AppState, bridge: ProcessingBridge, start_btn, stop_btn, progress): + """Start file transcription.""" + if not state.file_transcription_paths: + ui.notify("No files selected", type='warning') + return + + if state.is_recording: + ui.notify("Stop microphone recording first", type='warning') + return + + bridge.start_file_transcription(state.file_transcription_paths) + start_btn.set_enabled(False) + stop_btn.set_enabled(True) + progress.set_visibility(True) + ui.notify(f"Starting transcription of {len(state.file_transcription_paths)} file(s)") + + +def _stop_file_transcription(state: AppState, bridge: ProcessingBridge, start_btn, stop_btn, progress): + """Stop file transcription.""" + bridge.stop_file_transcription() + start_btn.set_enabled(True) + stop_btn.set_enabled(False) + ui.notify("File transcription stopped") @@ -65,6 +65,13 @@ class AppState: error_message: Optional[str] = None status_message: str = "" + # === File Transcription State === + file_transcription_mode: bool = False # True when transcribing from files + file_transcription_active: bool = False # True when file transcription is running + file_transcription_paths: List[str] = field(default_factory=list) # List of file paths to transcribe + file_transcription_progress: int = 0 # Progress 0-100 + file_transcription_current_file: str = "" # Currently processing file name + # === Text Buffers === whisper_text: str = "" ai_text: str = ""