reverberation.py (27541B)
1 #!/usr/bin/env python3 2 import tkinter as tk 3 from tkinter import ttk 4 import threading 5 import queue 6 import subprocess 7 import sys 8 import os 9 import pyaudio 10 import wave 11 import tempfile 12 import time 13 from faster_whisper import WhisperModel 14 import numpy as np 15 import logging 16 from datetime import datetime 17 import json 18 from pathlib import Path 19 20 # Set up logging 21 log_file = f"/tmp/whisper_transcribe_{datetime.now().strftime('%Y%m%d_%H%M%S')}.log" 22 logging.basicConfig( 23 level=logging.DEBUG, 24 format='%(asctime)s - %(levelname)s - %(message)s', 25 handlers=[ 26 logging.FileHandler(log_file), 27 logging.StreamHandler() 28 ] 29 ) 30 logger = logging.getLogger(__name__) 31 logger.info(f"Starting whisper-transcribe, log file: {log_file}") 32 33 # Model configurations 34 # Note: int8 is the quantized version for CPU, float16 requires CUDA 35 # Available compute types: int8 (CPU), float16 (GPU), float32 (both) 36 MODELS = [ 37 {"name": "tiny.en", "model": "tiny.en", "device": "cpu", "compute_type": "int8"}, 38 {"name": "base.en", "model": "base.en", "device": "cpu", "compute_type": "int8"}, 39 {"name": "small.en", "model": "small.en", "device": "cpu", "compute_type": "int8"} 40 ] 41 42 # Config file path 43 CONFIG_PATH = Path.home() / ".config" / "reverberation" / "config.json" 44 45 def load_config(): 46 """Load configuration from file""" 47 if CONFIG_PATH.exists(): 48 try: 49 with open(CONFIG_PATH, 'r') as f: 50 return json.load(f) 51 except: 52 pass 53 return {"model_index": 0} 54 55 def save_config(config): 56 """Save configuration to file""" 57 CONFIG_PATH.parent.mkdir(parents=True, exist_ok=True) 58 with open(CONFIG_PATH, 'w') as f: 59 json.dump(config, f) 60 61 class TranscribeWindow: 62 def __init__(self): 63 logger.info("Initializing TranscribeWindow") 64 self.root = tk.Tk() 65 66 # Hide window initially to prevent flash 67 self.root.withdraw() 68 69 self.root.title("reverberation") 70 logger.info("Created Tk root window") 71 72 # Remove window decorations and make it stay on top 73 self.root.overrideredirect(True) 74 self.root.attributes('-topmost', True) 75 76 # Set window type for i3 to treat it as floating 77 self.root.wm_attributes('-type', 'dialog') 78 79 # Style configuration (dmenu-like) 80 self.bg_color = "#222222" 81 self.fg_color = "#eeeeee" 82 self.highlight_color = "#005577" 83 84 # Set window size and center it (increased height for model selection) 85 window_width = 600 86 window_height = 500 87 88 # Update window first to get accurate screen dimensions 89 self.root.update_idletasks() 90 91 # Get screen dimensions 92 screen_width = self.root.winfo_screenwidth() 93 screen_height = self.root.winfo_screenheight() 94 95 # Calculate position 96 x = (screen_width - window_width) // 2 97 y = (screen_height - window_height) // 2 98 99 self.root.geometry(f"{window_width}x{window_height}+{x}+{y}") 100 self.root.configure(bg=self.bg_color) 101 102 # Create main frame 103 self.main_frame = tk.Frame(self.root, bg=self.bg_color) 104 self.main_frame.pack(fill=tk.BOTH, expand=True, padx=10, pady=10) 105 106 # Status label 107 self.status_label = tk.Label( 108 self.main_frame, 109 text="loading tiny.en .", 110 bg=self.bg_color, 111 fg=self.fg_color, 112 font=("monospace", 10) 113 ) 114 self.status_label.pack(pady=5) 115 116 # Buffer indicator frame 117 self.buffer_frame = tk.Frame(self.main_frame, bg=self.bg_color) 118 self.buffer_frame.pack(pady=2) 119 120 # Create 9 dots for buffer visualization 121 self.buffer_dots = [] 122 for i in range(9): 123 dot = tk.Label( 124 self.buffer_frame, 125 text="○", # Empty circle 126 bg=self.bg_color, 127 fg="#444444", # Dark gray for empty 128 font=("monospace", 12) 129 ) 130 dot.pack(side=tk.LEFT, padx=1) 131 self.buffer_dots.append(dot) 132 133 # Buffer tracking variables 134 self.buffer_progress = 0 135 self.is_processing = False 136 137 # Help text (pack at bottom first) 138 self.help_label = tk.Label( 139 self.main_frame, 140 text="[Tab] Switch model | [ESC] Cancel | [Enter] Insert text | [Shift+Enter] Copy to clipboard", 141 bg=self.bg_color, 142 fg="#888888", 143 font=("monospace", 9) 144 ) 145 self.help_label.pack(side=tk.BOTTOM, pady=2) 146 147 # Model selection frame (pack at bottom second) 148 self.model_frame = tk.Frame(self.main_frame, bg="#333333", height=50) # Different bg to see it 149 self.model_frame.pack(side=tk.BOTTOM, fill=tk.X, pady=8) 150 self.model_frame.pack_propagate(False) # Maintain fixed height 151 152 # Load saved config 153 self.config = load_config() 154 self.current_model_index = self.config.get("model_index", 0) 155 156 # Create model labels with better visibility 157 self.model_labels = [] 158 for i, model in enumerate(MODELS): 159 label = tk.Label( 160 self.model_frame, 161 text=f" {model['name']} ", 162 bg=self.bg_color, 163 fg=self.fg_color, 164 font=("monospace", 11, "bold"), 165 padx=15, 166 pady=5, 167 relief="solid", 168 borderwidth=1 169 ) 170 label.pack(side=tk.LEFT, padx=8) 171 self.model_labels.append(label) 172 173 # Update model highlighting 174 self.update_model_highlight() 175 176 # Text display (pack after bottom elements are in place) 177 self.text_frame = tk.Frame(self.main_frame, bg=self.bg_color) 178 self.text_frame.pack(fill=tk.BOTH, expand=True, pady=(10, 5)) 179 180 self.text_display = tk.Text( 181 self.text_frame, 182 bg=self.bg_color, 183 fg=self.fg_color, 184 font=("monospace", 12), 185 wrap=tk.WORD, 186 insertbackground=self.fg_color, 187 selectbackground=self.highlight_color, 188 selectforeground=self.fg_color, 189 borderwidth=0, 190 highlightthickness=0 191 ) 192 self.text_display.pack(fill=tk.BOTH, expand=True) 193 194 # Scrollbar 195 scrollbar = tk.Scrollbar(self.text_frame, command=self.text_display.yview) 196 scrollbar.pack(side=tk.RIGHT, fill=tk.Y) 197 self.text_display.config(yscrollcommand=scrollbar.set) 198 199 # Bind keys - use root window and all children 200 logger.info("Setting up key bindings") 201 self.root.bind_all('<Escape>', self.on_escape) 202 self.root.bind_all('<Return>', self.on_return) 203 self.root.bind_all('<Shift-Return>', self.on_shift_return) 204 self.root.bind_all('<Tab>', self.on_tab) 205 206 # Also bind to text display specifically 207 self.text_display.bind('<Escape>', self.on_escape) 208 self.text_display.bind('<Return>', self.on_return) 209 self.text_display.bind('<Shift-Return>', self.on_shift_return) 210 self.text_display.bind('<Tab>', self.on_tab) 211 212 # Debug: print when keys are pressed 213 self.root.bind_all('<Key>', self.on_any_key) 214 215 # We'll set focus and grab after window is mapped 216 self.root.after(10, self.setup_focus_and_grab) 217 218 # Audio and model setup 219 self.audio_queue = queue.Queue() 220 self.text_queue = queue.Queue() 221 self.is_recording = False 222 self.model = None 223 self.audio_thread = None 224 self.transcribe_thread = None 225 self.loading_dots = 1 226 self.is_loading = True 227 self.is_reloading = False 228 229 # Start loading model in background 230 threading.Thread(target=self.load_model, daemon=True).start() 231 232 # Update UI periodically 233 self.update_ui() 234 235 # Start loading animation 236 self.animate_loading() 237 238 def setup_focus_and_grab(self): 239 """Set up focus and keyboard grab after window is mapped""" 240 logger.info("Setting up focus and keyboard grab") 241 242 # First ensure window is visible and mapped 243 self.root.update_idletasks() 244 self.root.lift() 245 246 # Set focus 247 self.root.focus_force() 248 self.text_display.focus_set() 249 250 # Wait a bit more then grab keyboard 251 self.root.after(100, self.grab_keyboard) 252 253 def grab_keyboard(self): 254 """Grab keyboard input exclusively""" 255 try: 256 logger.info("Attempting to grab keyboard") 257 self.root.grab_set() 258 self.root.grab_set_global() # This grabs ALL keyboard input 259 logger.info(f"Keyboard grabbed successfully") 260 logger.info(f"Grab current: {self.root.grab_current()}") 261 logger.info(f"Focus: {self.root.focus_get()}") 262 except Exception as e: 263 logger.error(f"Failed to grab keyboard: {e}") 264 265 def update_model_highlight(self): 266 """Update model label highlighting""" 267 for i, label in enumerate(self.model_labels): 268 if i == self.current_model_index: 269 # Highlight selected model 270 label.config( 271 bg=self.highlight_color, 272 fg=self.bg_color, 273 relief="solid", 274 borderwidth=2 275 ) 276 else: 277 # Normal appearance 278 label.config( 279 bg=self.bg_color, 280 fg=self.fg_color, 281 relief="solid", 282 borderwidth=1 283 ) 284 285 def animate_loading(self): 286 """Animate the loading dots""" 287 if self.is_loading: 288 dots = "." * self.loading_dots 289 model_name = MODELS[self.current_model_index]["name"] 290 self.status_label.config(text=f"loading {model_name} {dots}") 291 self.loading_dots = (self.loading_dots % 4) + 1 292 self.root.after(500, self.animate_loading) 293 294 def load_model(self): 295 try: 296 model_config = MODELS[self.current_model_index] 297 logger.info(f"Loading model: {model_config['name']}") 298 299 self.model = WhisperModel( 300 model_config["model"], 301 device=model_config["device"], 302 compute_type=model_config["compute_type"] 303 ) 304 305 self.is_loading = False 306 self.text_queue.put(("status", "reverberation")) 307 308 # Start recording if not reloading 309 if not self.is_reloading: 310 self.start_recording() 311 self.is_reloading = False 312 313 except Exception as e: 314 self.is_loading = False 315 self.is_reloading = False 316 self.text_queue.put(("error", f"Error loading model: {str(e)}")) 317 318 def start_recording(self): 319 self.is_recording = True 320 self.audio_thread = threading.Thread(target=self.record_audio, daemon=True) 321 self.transcribe_thread = threading.Thread(target=self.transcribe_audio, daemon=True) 322 self.audio_thread.start() 323 self.transcribe_thread.start() 324 325 def update_buffer_indicator(self, progress, processing=False): 326 """Update the buffer progress dots""" 327 if processing: 328 # Show all dots filled when processing 329 for dot in self.buffer_dots: 330 dot.config(text="●", fg=self.highlight_color) 331 else: 332 # Show progress normally 333 for i, dot in enumerate(self.buffer_dots): 334 if i < progress: 335 dot.config(text="●", fg=self.fg_color) # Filled dot 336 else: 337 dot.config(text="○", fg="#444444") # Empty dot 338 339 def record_audio(self): 340 CHUNK = 1024 # Back to larger chunks for cleaner audio 341 FORMAT = pyaudio.paInt16 342 CHANNELS = 1 343 RATE = 16000 344 345 p = pyaudio.PyAudio() 346 stream = None 347 348 try: 349 stream = p.open( 350 format=FORMAT, 351 channels=CHANNELS, 352 rate=RATE, 353 input=True, 354 frames_per_buffer=CHUNK 355 ) 356 357 audio_buffer = [] 358 frames_per_chunk = RATE * 10 # 10 second chunks for complete thoughts 359 overlap_frames = int(RATE * 0.5) # Short overlap to avoid duplication 360 361 while self.is_recording: 362 try: 363 data = stream.read(CHUNK, exception_on_overflow=False) 364 audio_buffer.append(data) 365 366 # Update buffer progress indicator (make dots fill slightly earlier) 367 progress = min(9, (len(audio_buffer) * CHUNK * 10) // frames_per_chunk) 368 self.text_queue.put(("buffer_progress", progress)) 369 370 # Process chunks with small overlap to preserve word boundaries 371 if len(audio_buffer) >= frames_per_chunk // CHUNK: 372 # Signal processing state 373 self.text_queue.put(("buffer_processing", True)) 374 375 audio_data = b''.join(audio_buffer) 376 self.audio_queue.put(audio_data) 377 378 # Keep small overlap (0.25s) to preserve word boundaries 379 overlap_chunks = overlap_frames // CHUNK 380 if len(audio_buffer) > overlap_chunks: 381 audio_buffer = audio_buffer[-overlap_chunks:] 382 else: 383 audio_buffer = [] 384 385 except Exception as e: 386 if self.is_recording: # Only log if we're still supposed to be recording 387 logger.error(f"Audio error: {e}") 388 break 389 390 finally: 391 logger.info("Cleaning up audio stream") 392 if stream: 393 try: 394 stream.stop_stream() 395 stream.close() 396 except: 397 pass 398 p.terminate() 399 400 def transcribe_audio(self): 401 last_segments = [] # Track recent segments to filter repetitions 402 last_chunk_words = [] # Track words from overlap region 403 recent_text_context = [] # Track recent text for context prompts 404 405 while self.is_recording: 406 try: 407 if not self.audio_queue.empty(): 408 audio_data = self.audio_queue.get() 409 410 # Convert audio bytes to numpy array 411 audio_np = np.frombuffer(audio_data, dtype=np.int16).astype(np.float32) / 32768.0 412 413 # Audio preprocessing for better sensitivity 414 # Normalize audio to use full dynamic range 415 if np.max(np.abs(audio_np)) > 0: 416 audio_np = audio_np / (np.max(np.abs(audio_np)) * 0.8) # Normalize with headroom 417 418 # Apply additional gain for quiet speech 419 audio_gain = 2.0 # Moderate boost after normalization 420 audio_np = np.clip(audio_np * audio_gain, -1.0, 1.0) 421 422 # Check audio level after gain 423 audio_level = np.abs(audio_np).mean() 424 logger.debug(f"Audio level (after {audio_gain}x gain): {audio_level:.4f}") 425 426 # Even more aggressive - process almost all audio 427 if audio_level < 0.0005: # Extremely sensitive 428 logger.debug("Skipping very quiet audio") 429 continue 430 431 # Build dynamic context prompt - emphasize literal transcription 432 base_prompt = "This is a highly technical statement. Transcribe only the exact words that are spoken. Do not add, interpret, or complete sentences. If speech is unclear or incomplete, transcribe only what is clearly audible." 433 434 if recent_text_context: 435 # Add recent context (last 2 sentences) 436 context_text = " ".join(recent_text_context[-2:]) 437 full_prompt = f"{base_prompt} Previous context: \"{context_text}\"" 438 else: 439 full_prompt = base_prompt 440 441 logger.debug(f"Using prompt: {full_prompt[:100]}...") 442 443 # Balanced transcription settings with dynamic context 444 segments, _ = self.model.transcribe( 445 audio_np, 446 beam_size=12, # Maximum beams for comprehensive search 447 best_of=5, # More candidates for comprehensive coverage 448 temperature=0.0, # Fully deterministic for literal transcription 449 condition_on_previous_text=False, # Disable audio context - using text context instead 450 no_speech_threshold=0.1, # Very low - catch almost everything 451 compression_ratio_threshold=3.0, # More lenient to avoid dropping speech 452 log_prob_threshold=-1.0, # Balanced confidence requirement 453 word_timestamps=False, 454 suppress_tokens=[-1], # Suppress special tokens 455 repetition_penalty=1.05, # Light repetition penalty to avoid cutting off speech 456 initial_prompt=full_prompt 457 ) 458 459 # Combine all segments into one text block for better flow 460 full_text = "" 461 for segment in segments: 462 if segment.text.strip(): 463 full_text += segment.text 464 465 if full_text.strip(): 466 logger.debug(f"Full transcription: '{full_text.strip()}'") 467 # Simple word-based overlap filtering 468 words = full_text.strip().split() 469 470 if last_chunk_words and len(words) > 0: 471 # Find overlap by comparing first few words with last chunk's end 472 overlap_size = 0 473 for i in range(min(len(last_chunk_words), len(words))): 474 if words[i] == last_chunk_words[-(len(last_chunk_words)-i)]: 475 overlap_size = len(last_chunk_words) - i 476 break 477 478 # Send only the new part 479 if overlap_size > 0 and overlap_size < len(words): 480 new_words = words[overlap_size:] 481 new_text = " " + " ".join(new_words) 482 else: 483 new_text = " " + " ".join(words) 484 else: 485 # First chunk 486 new_text = " ".join(words) 487 488 # Clean up text and send with better repetition filtering 489 new_text = new_text.strip() 490 491 # Check for repetitive patterns 492 words = new_text.split() 493 if len(words) > 3: 494 # Simple repetition detection - check if same word repeated >3 times 495 word_counts = {} 496 for word in words: 497 word_counts[word] = word_counts.get(word, 0) + 1 498 max_count = max(word_counts.values()) if word_counts else 0 499 500 if max_count > 5: # More lenient - allow some repetition 501 logger.debug(f"Filtered repetitive text: '{new_text}'") 502 new_text = "" 503 504 if new_text and new_text not in last_segments: 505 logger.debug(f"Transcribed: '{new_text}'") 506 self.text_queue.put(("text", " " + new_text)) 507 508 # Reset buffer indicator after transcription 509 self.text_queue.put(("buffer_reset", True)) 510 511 # Track recent text (last 5 for better context tracking) 512 last_segments.append(new_text) 513 if len(last_segments) > 5: 514 last_segments.pop(0) 515 516 # Add to context buffer (keep last 3 for prompts) 517 recent_text_context.append(new_text) 518 if len(recent_text_context) > 3: 519 recent_text_context.pop(0) 520 521 # Store last 3 words for overlap detection 522 last_chunk_words = words[-3:] if len(words) >= 3 else words 523 segment_count = 1 524 else: 525 segment_count = 0 526 527 if segment_count == 0: 528 logger.debug("No speech detected in this chunk") 529 530 else: 531 time.sleep(0.1) # Standard sleep 532 533 except Exception as e: 534 if self.is_recording: 535 logger.error(f"Transcription error: {e}") 536 537 def update_ui(self): 538 try: 539 while not self.text_queue.empty(): 540 msg_type, content = self.text_queue.get_nowait() 541 542 if msg_type == "status": 543 self.status_label.config(text=content) 544 elif msg_type == "text": 545 self.text_display.insert(tk.END, content + " ") 546 self.text_display.see(tk.END) 547 elif msg_type == "error": 548 self.status_label.config(text=content, fg="#ff0000") 549 elif msg_type == "buffer_progress": 550 self.update_buffer_indicator(content, processing=False) 551 elif msg_type == "buffer_processing": 552 self.update_buffer_indicator(9, processing=True) 553 elif msg_type == "buffer_reset": 554 self.update_buffer_indicator(0, processing=False) 555 556 except queue.Empty: 557 pass 558 559 self.root.after(100, self.update_ui) 560 561 def on_any_key(self, event): 562 logger.debug(f"Key pressed: {event.keysym} (state: {event.state}, keycode: {event.keycode})") 563 print(f"Key pressed: {event.keysym} (state: {event.state})") 564 565 def on_escape(self, event): 566 logger.info("Escape pressed!") 567 print("Escape pressed!") 568 self.cancel() 569 return "break" 570 571 def on_return(self, event): 572 logger.info("Return pressed!") 573 print("Return pressed!") 574 self.insert_text() 575 return "break" 576 577 def on_shift_return(self, event): 578 logger.info("Shift+Return pressed!") 579 print("Shift+Return pressed!") 580 self.copy_to_clipboard() 581 return "break" 582 583 def on_tab(self, event): 584 logger.info("Tab pressed!") 585 # Cycle to next model 586 self.current_model_index = (self.current_model_index + 1) % len(MODELS) 587 self.update_model_highlight() 588 589 # Save config 590 self.config["model_index"] = self.current_model_index 591 save_config(self.config) 592 593 # Reload model 594 self.reload_model() 595 return "break" 596 597 def reload_model(self): 598 """Reload the model with new selection""" 599 logger.info(f"Reloading model to: {MODELS[self.current_model_index]['name']}") 600 601 # Stop current recording 602 self.is_recording = False 603 time.sleep(0.5) # Give threads time to stop 604 605 # Clear current text 606 self.text_display.delete("1.0", tk.END) 607 608 # Set loading state 609 self.is_loading = True 610 self.is_reloading = True 611 self.loading_dots = 1 612 613 # Start loading animation again 614 self.animate_loading() 615 616 # Load new model in background 617 threading.Thread(target=self.load_model, daemon=True).start() 618 619 # Restart recording after model loads 620 threading.Thread(target=self._restart_recording, daemon=True).start() 621 622 def _restart_recording(self): 623 """Helper to restart recording after model reload""" 624 # Wait for model to load 625 while self.is_loading: 626 time.sleep(0.1) 627 628 # Start recording again 629 if not self.is_recording: 630 self.start_recording() 631 632 def get_text(self): 633 return self.text_display.get("1.0", tk.END).strip() 634 635 def cancel(self): 636 logger.info("Cancelling and closing window") 637 self.is_recording = False 638 639 # Give threads time to finish 640 logger.info("Waiting for threads to finish...") 641 time.sleep(0.5) 642 643 try: 644 self.root.grab_release() # Release keyboard grab 645 except: 646 pass 647 648 try: 649 self.root.quit() # Exit mainloop first 650 self.root.destroy() # Then destroy window 651 except: 652 pass 653 654 logger.info("Exiting application") 655 sys.exit(0) 656 657 def insert_text(self): 658 text = self.get_text() 659 if text: 660 # Use xdotool to type the text 661 self.root.withdraw() # Hide window first 662 time.sleep(0.1) # Small delay 663 subprocess.run(['xdotool', 'type', '--clearmodifiers', text]) 664 self.cancel() 665 666 def copy_to_clipboard(self): 667 text = self.get_text() 668 if text: 669 # Use xclip to copy to clipboard 670 process = subprocess.Popen(['xclip', '-selection', 'clipboard'], stdin=subprocess.PIPE) 671 process.communicate(text.encode('utf-8')) 672 self.cancel() 673 674 def run(self): 675 # Show window now that everything is configured 676 self.root.deiconify() 677 678 # Ensure window has focus when starting 679 # Focus and grab are now handled by setup_focus_and_grab 680 logger.info("Window mainloop starting") 681 self.root.mainloop() 682 683 if __name__ == "__main__": 684 try: 685 logger.info("Starting application") 686 app = TranscribeWindow() 687 app.run() 688 except Exception as e: 689 logger.error(f"Application error: {e}", exc_info=True) 690 raise