app.py

#!/usr/bin/env python3
"""
Voice Interview Assistant 
A professional voice recording and AI-powered interview assistance application.
"""

import tkinter as tk
from tkinter import scrolledtext, messagebox, ttk
import sounddevice as sd
from scipy.io.wavfile import write
import whisper
import tempfile
import threading
import requests
import json
import numpy as np
import logging
import os
import sys
from pathlib import Path
from dataclasses import dataclass, asdict
from typing import Optional, Dict, Any, Callable
from time import time
import configparser
from contextlib import contextmanager
import queue
import traceback

@dataclass
class AppConfig:
    """Application configuration with defaults"""
    # Audio settings
    sample_rate: int = 44100
    audio_channels: int = 1
    audio_dtype: str = 'int16'
    
    # Whisper model settings
    whisper_model: str = "medium.en"  
    whisper_device: str = "auto"  
    
    # AI API settings
    ai_api_url: str = "http://localhost:11434/api/chat"
    ai_model: str = "llama3"
    ai_timeout: int = 60
    
    # UI settings
    window_width: int = 600
    window_height: int = 80
    popup_width: int = 800
    popup_height: int = 600
    window_alpha: float = 0.85
    always_on_top: bool = True
    
    # Thread settings
    max_thread_join_timeout: float = 2.0
    debounce_delay: float = 0.5
    
    # Logging
    log_level: str = "INFO"
    log_file: Optional[str] = None

class ConfigManager:
    """Manages application configuration"""
    
    def __init__(self, config_file: str = "config.ini"):
        self.config_file = Path(config_file)
        self.config = AppConfig()
        self.load_config()
    
    def load_config(self) -> None:
        """Load configuration from file"""
        if not self.config_file.exists():
            self.save_config()
            return
        
        try:
            parser = configparser.ConfigParser()
            parser.read(self.config_file)
            
            if 'audio' in parser:
                audio_section = parser['audio']
                self.config.sample_rate = audio_section.getint('sample_rate', self.config.sample_rate)
                self.config.audio_channels = audio_section.getint('channels', self.config.audio_channels)
                self.config.audio_dtype = audio_section.get('dtype', self.config.audio_dtype)
            
            if 'whisper' in parser:
                whisper_section = parser['whisper']
                self.config.whisper_model = whisper_section.get('model', self.config.whisper_model)
                self.config.whisper_device = whisper_section.get('device', self.config.whisper_device)
            
            if 'ai' in parser:
                ai_section = parser['ai']
                self.config.ai_api_url = ai_section.get('api_url', self.config.ai_api_url)
                self.config.ai_model = ai_section.get('model', self.config.ai_model)
                self.config.ai_timeout = ai_section.getint('timeout', self.config.ai_timeout)
            
            if 'logging' in parser:
                log_section = parser['logging']
                self.config.log_level = log_section.get('level', self.config.log_level)
                self.config.log_file = log_section.get('file', self.config.log_file)
                
        except Exception as e:
            logging.warning(f"Failed to load config: {e}. Using defaults.")
    
    def save_config(self) -> None:
        """Save current configuration to file"""
        try:
            parser = configparser.ConfigParser()
            
            parser['audio'] = {
                'sample_rate': str(self.config.sample_rate),
                'channels': str(self.config.audio_channels),
                'dtype': self.config.audio_dtype
            }
            
            parser['whisper'] = {
                'model': self.config.whisper_model,
                'device': self.config.whisper_device
            }
            
            parser['ai'] = {
                'api_url': self.config.ai_api_url,
                'model': self.config.ai_model,
                'timeout': str(self.config.ai_timeout)
            }
            
            parser['logging'] = {
                'level': self.config.log_level,
                'file': self.config.log_file or ''
            }
            
            with open(self.config_file, 'w') as f:
                parser.write(f)
                
        except Exception as e:
            logging.error(f"Failed to save config: {e}")


def setup_logging(config: AppConfig) -> None:
    """Setup application logging"""
    log_format = '%(asctime)s - %(name)s - %(levelname)s - %(message)s'
    
    handlers = [logging.StreamHandler(sys.stdout)]
    
    if config.log_file:
        try:
            log_dir = Path(config.log_file).parent
            log_dir.mkdir(parents=True, exist_ok=True)
            handlers.append(logging.FileHandler(config.log_file))
        except Exception as e:
            print(f"Warning: Could not setup file logging: {e}")
    
    logging.basicConfig(
        level=getattr(logging, config.log_level.upper()),
        format=log_format,
        handlers=handlers
    )


class AudioManager:
    """Manages audio recording functionality"""
    
    def __init__(self, config: AppConfig):
        self.config = config
        self.logger = logging.getLogger(__name__)
        self.recording = False
        self.audio_data = []
        self.stream: Optional[sd.InputStream] = None
        self._lock = threading.Lock()
    
    def start_recording(self) -> bool:
        """Start audio recording"""
        with self._lock:
            if self.recording:
                self.logger.warning("Recording already in progress")
                return False
            
            try:
                self.audio_data.clear()
                self.recording = True
                
                def callback(indata, frames, time, status):
                    if status:
                        self.logger.warning(f"Audio callback status: {status}")
                    if self.recording:
                        self.audio_data.append(indata.copy())
                
                self.stream = sd.InputStream(
                    samplerate=self.config.sample_rate,
                    channels=self.config.audio_channels,
                    dtype=self.config.audio_dtype,
                    callback=callback
                )
                self.stream.start()
                self.logger.info("Recording started")
                return True
                
            except Exception as e:
                self.logger.error(f"Failed to start recording: {e}")
                self.recording = False
                return False
    
    def stop_recording(self) -> Optional[str]:
        """Stop recording and return audio file path"""
        with self._lock:
            if not self.recording:
                self.logger.warning("No recording in progress")
                return None
            
            self.recording = False
            
            try:
                if self.stream:
                    self.stream.stop()
                    self.stream.close()
                    self.stream = None
                
                if not self.audio_data:
                    self.logger.warning("No audio data recorded")
                    return None
                
                # Save audio to temporary file
                audio_array = np.concatenate(self.audio_data, axis=0)
                
                with tempfile.NamedTemporaryFile(delete=False, suffix=".wav") as tmp:
                    write(tmp.name, self.config.sample_rate, audio_array)
                    audio_path = tmp.name
                
                self.logger.info(f"Audio saved to: {audio_path}")
                return audio_path
                
            except Exception as e:
                self.logger.error(f"Failed to stop recording: {e}")
                return None
    
    def is_recording(self) -> bool:
        """Check if currently recording"""
        return self.recording
    
    def cleanup(self) -> None:
        """Cleanup resources"""
        if self.stream:
            try:
                self.stream.stop()
                self.stream.close()
            except Exception as e:
                self.logger.error(f"Error during cleanup: {e}")


class AIService:
    """Handles AI API interactions"""
    
    def __init__(self, config: AppConfig):
        self.config = config
        self.logger = logging.getLogger(__name__)
        self.session = requests.Session()
        self.whisper_model = None
        self._load_whisper_model()
    
    def _load_whisper_model(self) -> None:
        """Load Whisper model"""
        try:
            device = self.config.whisper_device
            if device == "auto":
                device = "cuda" if self._check_cuda_available() else "cpu"
            
            self.logger.info(f"Loading Whisper model: {self.config.whisper_model} on {device}")
            self.whisper_model = whisper.load_model(self.config.whisper_model, device=device)
            self.logger.info("Whisper model loaded successfully")
            
        except Exception as e:
            self.logger.error(f"Failed to load Whisper model: {e}")
            raise
    
    def _check_cuda_available(self) -> bool:
        """Check if CUDA is available"""
        try:
            import torch
            return torch.cuda.is_available()
        except ImportError:
            return False
    
    def transcribe_audio(self, audio_path: str) -> str:
        """Transcribe audio file to text"""
        try:
            if not self.whisper_model:
                raise RuntimeError("Whisper model not loaded")
            
            self.logger.info(f"Transcribing audio: {audio_path}")
            result = self.whisper_model.transcribe(audio_path)
            transcription = result["text"].strip()
            
            self.logger.info(f"Transcription completed: {transcription[:100]}...")
            return transcription
            
        except Exception as e:
            self.logger.error(f"Transcription failed: {e}")
            raise
    
    def get_ai_response(self, question: str, response_callback: Callable[[str], None], 
                       cancel_event: threading.Event) -> None:
        """Get AI response with streaming"""
        try:
            prompt = self._build_prompt(question)
            
            payload = {
                "model": self.config.ai_model,
                "stream": True,
                "messages": [{"role": "user", "content": prompt}]
            }
            
            headers = {"Content-Type": "application/json"}
            
            self.logger.info("Requesting AI response")
            
            with self.session.post(
                self.config.ai_api_url,
                headers=headers,
                json=payload,
                stream=True,
                timeout=self.config.ai_timeout
            ) as response:
                response.raise_for_status()
                
                for line in response.iter_lines():
                    if cancel_event.is_set():
                        self.logger.info("AI response cancelled")
                        break
                    
                    if line:
                        try:
                            data = json.loads(line.decode("utf-8"))
                            if 'message' in data and 'content' in data['message']:
                                content = data['message']['content']
                                response_callback(content)
                        except json.JSONDecodeError as e:
                            self.logger.warning(f"Failed to parse JSON: {e}")
                            continue
                            
        except requests.exceptions.RequestException as e:
            self.logger.error(f"AI API request failed: {e}")
            response_callback(f"\n⚠️ AI service error: {e}")
        except Exception as e:
            self.logger.error(f"Unexpected error in AI response: {e}")
            response_callback(f"\n⚠️ Unexpected error: {e}")
    
    def _build_prompt(self, question: str) -> str:
        """Build AI prompt"""
        return f"""
You are an intelligent AI assistant helping a candidate during a software development job interview.

The candidate is being interviewed for roles related to:
- Flutter Development
- Android App Development  
- Backend System Design

The interviewer has asked the following question:
"{question}"

Your response should follow these steps:
1. Give a direct and concise answer in a single sentence.
2. Explain the reasoning or logic behind the answer in a short paragraph.
3. If and only if the interviewer explicitly asks to write code, then:
   - Briefly explain your approach to the code (1–2 lines).
   - Then write the complete code in the most appropriate language.
   - After the code, provide a short explanation of what the code does and how it works.

Keep your tone professional, informative, and supportive.
"""
    
    def cleanup(self) -> None:
        """Cleanup resources"""
        if self.session:
            self.session.close()
            

class VoiceInterviewAssistant:
    """Main application class"""
    
    def __init__(self):
        self.config_manager = ConfigManager()
        self.config = self.config_manager.config
        setup_logging(self.config)
        self.logger = logging.getLogger(__name__)
        
        # Initialize components
        self.audio_manager = AudioManager(self.config)
        self.ai_service = AIService(self.config)
        
        # Threading
        self.ai_thread: Optional[threading.Thread] = None
        self.cancel_event = threading.Event()
        self.processing = False
        self.thread_lock = threading.Lock()
        self.last_press_time = 0
        
        # UI components
        self.root: Optional[tk.Tk] = None
        self.popup: Optional[tk.Toplevel] = None
        self.toggle_btn: Optional[tk.Button] = None
        self.status_label: Optional[tk.Label] = None
        self.transcript_text: Optional[scrolledtext.ScrolledText] = None
        self.output_text: Optional[scrolledtext.ScrolledText] = None
        
        # Initialize UI
        self._setup_ui()
        
        self.logger.info("Voice Interview Assistant initialized")
    
    def _setup_ui(self) -> None:
        """Setup the user interface"""
        try:
            self._create_main_window()
            self._create_popup_window()
            self._bind_events()
            
        except Exception as e:
            self.logger.error(f"Failed to setup UI: {e}")
            raise
    
    def _create_main_window(self) -> None:
        """Create main control window"""
        self.root = tk.Tk()
        self.root.title("🎤 Voice Assistant")
        self.root.geometry(f"{self.config.window_width}x{self.config.window_height}+50+30")
        
        if self.config.always_on_top:
            self.root.attributes("-topmost", True)
        
        self.root.overrideredirect(True)
        self.root.configure(bg="black")
        self.root.wm_attributes("-alpha", self.config.window_alpha)
        
        # Main frame
        top_frame = tk.Frame(self.root, bg="black")
        top_frame.pack(fill="both", expand=True)
        
        # Toggle button
        self.toggle_btn = tk.Button(
            top_frame,
            text="🎙️",
            font=("Segoe UI", 12, "bold"),
            fg="white",
            bg="black",
            bd=0,
            activebackground="black",
            activeforeground="green",
            command=self._toggle_recording
        )
        self.toggle_btn.pack(side="left", padx=10)
        
        # Status label
        self.status_label = tk.Label(
            top_frame,
            text="Ready",
            font=("Segoe UI", 10),
            fg="white",
            bg="black"
        )
        self.status_label.pack(side="left", padx=10)
    
    def _create_popup_window(self) -> None:
        """Create popup window for transcript and response"""
        self.popup = tk.Toplevel(self.root)
        self.popup.geometry(f"{self.config.popup_width}x{self.config.popup_height}+50+120")
        self.popup.title("🧠 Transcript & Answer")
        self.popup.attributes("-topmost", True)
        self.popup.configure(bg="#1e1e1e")
        self.popup.withdraw()
        
        # Main container
        main_container = tk.Frame(self.popup, bg="#1e1e1e")
        main_container.pack(fill="both", expand=True, padx=15, pady=15)
        
        # Transcript section
        transcript_frame = tk.Frame(main_container, bg="#1e1e1e")
        transcript_frame.pack(fill="x", pady=(0, 15))
        
        tk.Label(
            transcript_frame,
            text="📝 Transcribed Question:",
            bg="#1e1e1e",
            fg="white",
            font=("Segoe UI", 11, "bold")
        ).pack(anchor="w", pady=(0, 5))
        
        self.transcript_text = scrolledtext.ScrolledText(
            transcript_frame,
            height=4,
            wrap=tk.WORD,
            font=("Segoe UI", 11),
            bg="#2d2d2d",
            fg="white",
            insertbackground="white",
            relief="flat",
            borderwidth=1,
            selectbackground="#4a4a4a"
        )
        self.transcript_text.pack(fill="x", pady=(0, 10))
        
        # Answer section
        answer_frame = tk.Frame(main_container, bg="#1e1e1e")
        answer_frame.pack(fill="both", expand=True)
        
        tk.Label(
            answer_frame,
            text="🤖 AI Answer:",
            bg="#1e1e1e",
            fg="white",
            font=("Segoe UI", 11, "bold")
        ).pack(anchor="w", pady=(0, 5))
        
        output_frame = tk.Frame(answer_frame, bg="#2d2d2d", relief="flat", borderwidth=1)
        output_frame.pack(fill="both", expand=True)
        
        self.output_text = scrolledtext.ScrolledText(
            output_frame,
            wrap=tk.WORD,
            font=("Segoe UI", 11),
            bg="#2d2d2d",
            fg="white",
            insertbackground="white",
            relief="flat",
            borderwidth=0,
            selectbackground="#4a4a4a",
            padx=10,
            pady=10
        )
        self.output_text.pack(fill="both", expand=True, padx=1, pady=1)
        self.output_text.insert("1.0", "AI responses will appear here...\n\n")
    
    def _bind_events(self) -> None:
        """Bind keyboard and mouse events"""
        self.root.bind("<space>", self._space_handler)
        self.toggle_btn.bind("<Double-Button-1>", lambda e: self._toggle_popup())
        
        # Handle window close
        self.root.protocol("WM_DELETE_WINDOW", self._on_closing)
        self.popup.protocol("WM_DELETE_WINDOW", lambda: self.popup.withdraw())
    
    def _space_handler(self, event) -> None:
        """Handle space key press"""
        current_time = time()
        if current_time - self.last_press_time > self.config.debounce_delay:
            self._toggle_recording()
            self.last_press_time = current_time
    
    def _toggle_popup(self) -> None:
        """Toggle popup window visibility"""
        try:
            if self.popup.state() == "withdrawn":
                self.popup.deiconify()
                self.popup.focus_force()
            else:
                self.popup.withdraw()
        except Exception as e:
            self.logger.error(f"Error toggling popup: {e}")
    
    def _toggle_recording(self) -> None:
        """Toggle recording state"""
        if self.processing:
            return
        
        self.processing = True
        
        def safe_toggle():
            try:
                with self.thread_lock:
                    if self.audio_manager.is_recording():
                        self._stop_recording()
                    else:
                        self._start_recording()
            except Exception as e:
                self.logger.error(f"Error in toggle recording: {e}")
                self._update_status(f"Error: {e}")
            finally:
                self.root.after(300, lambda: setattr(self, 'processing', False))
        
        threading.Thread(target=safe_toggle, daemon=True).start()
    
    def _start_recording(self) -> None:
        """Start recording process"""
        try:
            # Cancel any ongoing AI processing
            self.cancel_event.set()
            if self.ai_thread and self.ai_thread.is_alive():
                self.ai_thread.join(timeout=self.config.max_thread_join_timeout)
            
            # Clear previous results
            self._clear_text_widgets()
            
            # Start recording
            if self.audio_manager.start_recording():
                self._update_ui_recording_state(True)
                self._update_status("🎙️ Recording...")
            else:
                self._update_status("⚠️ Failed to start recording")
                
        except Exception as e:
            self.logger.error(f"Error starting recording: {e}")
            self._update_status(f"⚠️ Error: {e}")
    
    def _stop_recording(self) -> None:
        """Stop recording and process audio"""
        try:
            self._update_status("Processing...")
            self._update_ui_recording_state(False)
            
            audio_path = self.audio_manager.stop_recording()
            if audio_path:
                self._process_audio(audio_path)
            else:
                self._update_status("⚠️ No audio recorded")
                self._reset_ui_state()
                
        except Exception as e:
            self.logger.error(f"Error stopping recording: {e}")
            self._update_status(f"⚠️ Error: {e}")
            self._reset_ui_state()
    
    def _process_audio(self, audio_path: str) -> None:
        """Process recorded audio"""
        self.cancel_event.clear()
        
        def process_task():
            try:
                # Transcribe audio
                self._update_status("Transcribing...")
                question = self.ai_service.transcribe_audio(audio_path)
                
                # Update transcript
                self.root.after(0, lambda: self._update_transcript(question))
                self.root.after(0, lambda: self._show_popup())
                
                # Get AI response
                self._update_status("Generating answer...")
                self.ai_service.get_ai_response(
                    question,
                    self._on_ai_response_chunk,
                    self.cancel_event
                )
                
                # Cleanup
                try:
                    os.unlink(audio_path)
                except Exception as e:
                    self.logger.warning(f"Could not delete temp file: {e}")
                
                if not self.cancel_event.is_set():
                    self._update_status("Done")
                    
            except Exception as e:
                self.logger.error(f"Error processing audio: {e}")
                self.root.after(0, lambda: self._update_output(f"\n⚠️ Error: {e}"))
                self.root.after(0, lambda: self._update_status("Error"))
            finally:
                self.root.after(0, self._reset_ui_state)
        
        self.ai_thread = threading.Thread(target=process_task, daemon=True)
        self.ai_thread.start()
    
    def _on_ai_response_chunk(self, content: str) -> None:
        """Handle AI response chunk"""
        self.root.after(0, lambda: self._update_output(content))
    
    def _update_transcript(self, text: str) -> None:
        """Update transcript text widget"""
        try:
            if self.transcript_text:
                self.transcript_text.delete("1.0", tk.END)
                self.transcript_text.insert(tk.END, text + "\n")
        except Exception as e:
            self.logger.error(f"Error updating transcript: {e}")
    
    def _update_output(self, text: str) -> None:
        """Update output text widget"""
        try:
            if self.output_text:
                if text.startswith("\n⚠️"):  # Error message
                    self.output_text.delete("1.0", tk.END)
                self.output_text.insert(tk.END, text)
                self.output_text.see(tk.END)
        except Exception as e:
            self.logger.error(f"Error updating output: {e}")
    
    def _update_status(self, status: str) -> None:
        """Update status label"""
        try:
            if self.status_label:
                self.status_label.config(text=status)
        except Exception as e:
            self.logger.error(f"Error updating status: {e}")
    
    def _update_ui_recording_state(self, recording: bool) -> None:
        """Update UI for recording state"""
        try:
            if self.toggle_btn:
                if recording:
                    self.toggle_btn.config(text="⏹️")
                else:
                    self.toggle_btn.config(text="🎙️", state=tk.DISABLED)
        except Exception as e:
            self.logger.error(f"Error updating UI state: {e}")
    
    def _reset_ui_state(self) -> None:
        """Reset UI to initial state"""
        try:
            if self.toggle_btn:
                self.toggle_btn.config(state=tk.NORMAL, text="🎙️")
        except Exception as e:
            self.logger.error(f"Error resetting UI state: {e}")
    
    def _clear_text_widgets(self) -> None:
        """Clear text widgets"""
        try:
            if self.transcript_text:
                self.transcript_text.delete("1.0", tk.END)
            if self.output_text:
                self.output_text.delete("1.0", tk.END)
                self.output_text.insert("1.0", "AI responses will appear here...\n\n")
        except Exception as e:
            self.logger.error(f"Error clearing text widgets: {e}")
    
    def _show_popup(self) -> None:
        """Show popup window"""
        try:
            if self.popup:
                self.popup.deiconify()
                self.popup.lift()
        except Exception as e:
            self.logger.error(f"Error showing popup: {e}")
    
    def _on_closing(self) -> None:
        """Handle application closing"""
        try:
            self.logger.info("Application closing...")
            
            self.cancel_event.set()
            
            if self.ai_thread and self.ai_thread.is_alive():
                self.ai_thread.join(timeout=self.config.max_thread_join_timeout)
            
            self.audio_manager.cleanup()
            self.ai_service.cleanup()
            
            self.config_manager.save_config()
            
            if self.popup:
                self.popup.destroy()
            if self.root:
                self.root.quit()
                self.root.destroy()
                
        except Exception as e:
            self.logger.error(f"Error during cleanup: {e}")
        finally:
            sys.exit(0)
    
    def run(self) -> None:
        """Start the application"""
        try:
            self.logger.info("Starting Voice Interview Assistant")
            if self.root:
                self.root.mainloop()
        except KeyboardInterrupt:
            self.logger.info("Application interrupted by user")
        except Exception as e:
            self.logger.error(f"Unexpected error: {e}")
            messagebox.showerror("Fatal Error", f"Application error: {e}")
        finally:
            self._on_closing()

def main():
    """Main entry point"""
    try:
        app = VoiceInterviewAssistant()
        app.run()
    except Exception as e:
        print(f"Failed to start application: {e}")
        traceback.print_exc()
        sys.exit(1)

if __name__ == "__main__":
    main()