225 lines
7.4 KiB
Python
225 lines
7.4 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Meeting Assistant - Real-time AI Copilot for Meetings
|
|
"""
|
|
|
|
import sys
|
|
import os
|
|
import yaml
|
|
import threading
|
|
import logging
|
|
from pathlib import Path
|
|
from datetime import datetime
|
|
|
|
sys.path.insert(0, str(Path(__file__).parent))
|
|
|
|
from src.audio_listener import AudioListener
|
|
from src.ai_engine import AIEngine
|
|
from src.overlay import InvisibleOverlay
|
|
from src.context_manager import ContextManager
|
|
from src.utils import setup_logging, get_platform
|
|
|
|
# Question words that indicate the speaker wants an answer
|
|
_QUESTION_WORDS = (
|
|
"what", "why", "how", "when", "where", "who", "which",
|
|
"can you", "could you", "would you", "do you", "did you",
|
|
"is there", "are there", "is it", "are you",
|
|
"tell me", "explain", "describe", "define",
|
|
"your name", "help me",
|
|
)
|
|
|
|
# Phrases to skip even if they contain question words (noise)
|
|
_NOISE_PHRASES = (
|
|
"timer cannot", "qobject", "qml", "pyside", "pyqt",
|
|
"def ", "class ", "import ", "return ", "self.",
|
|
)
|
|
|
|
|
|
class MeetingAssistant:
|
|
def __init__(self):
|
|
"""Initialize all components"""
|
|
print("🚀 Starting Meeting Assistant...")
|
|
print(f"💻 Platform: {get_platform()}")
|
|
|
|
# Load config
|
|
config_path = Path(__file__).parent / "config.yaml"
|
|
with open(config_path, 'r') as f:
|
|
self.config = yaml.safe_load(f)
|
|
|
|
# Setup logging
|
|
self.logger = setup_logging()
|
|
|
|
# Initialize components
|
|
self.ai_engine = AIEngine(self.config)
|
|
self.context_manager = ContextManager(self.config)
|
|
self.overlay = InvisibleOverlay(self.config)
|
|
|
|
# Audio listener with callback
|
|
self.audio_listener = AudioListener(self.config, self.on_audio_transcript)
|
|
|
|
# State management
|
|
self.current_answer = None
|
|
self.answering = False
|
|
self.answer_lock = threading.Lock()
|
|
self._last_question = ""
|
|
|
|
print("✅ Meeting Assistant Ready!")
|
|
print("==================================================")
|
|
print("🎤 Listening for questions via microphone")
|
|
print(" Answers appear in the overlay (top-right)")
|
|
print(" Type a question here + Enter to test AI")
|
|
print(" Ctrl+C to stop")
|
|
print("==================================================")
|
|
|
|
def on_audio_transcript(self, text, timestamp):
|
|
"""Audio callback — runs in the audio listener thread"""
|
|
if not text or len(text.strip()) < 2:
|
|
return
|
|
|
|
self.context_manager.add_audio_context(text, timestamp)
|
|
|
|
if self._is_question(text):
|
|
print(f"\n❓ Question detected: {text}")
|
|
self._generate_answer(text)
|
|
else:
|
|
print(f" Context (not a question): {text}")
|
|
|
|
def _is_question(self, text):
|
|
"""Enhanced question detection for faster responses"""
|
|
if len(text.split()) < 2: # Reduced from 3 for shorter questions
|
|
return False
|
|
|
|
text_lower = text.lower().strip()
|
|
|
|
# Skip obvious noise / code text
|
|
if any(p in text_lower for p in _NOISE_PHRASES):
|
|
return False
|
|
|
|
# Ends with "?" → always a question
|
|
if text.rstrip().endswith("?"):
|
|
return True
|
|
|
|
# Check for math expressions (2+2, two plus two)
|
|
import re
|
|
math_patterns = [
|
|
r'\d+\s*[\+\-\*\/]\s*\d+', # 2+2, 5-3
|
|
r'\d+\s*plus\s*\d+', # 2 plus 2
|
|
r'\d+\s*minus\s*\d+', # 5 minus 3
|
|
r'\d+\s*times\s*\d+', # 4 times 5
|
|
r'\d+\s*divided by\s*\d+', # 10 divided by 2
|
|
]
|
|
for pattern in math_patterns:
|
|
if re.search(pattern, text_lower):
|
|
return True
|
|
|
|
# Check for short implicit questions (2-5 words without filler)
|
|
words = text_lower.split()
|
|
if 2 <= len(words) <= 5:
|
|
filler = {'um', 'uh', 'like', 'so', 'well', 'actually', 'basically'}
|
|
content_words = [w for w in words if w not in filler]
|
|
if len(content_words) >= 2:
|
|
# "two plus two", "capital france", "python example"
|
|
return True
|
|
|
|
# Starts with or contains a question word
|
|
if any(text_lower.startswith(w) or f" {w} " in text_lower for w in _QUESTION_WORDS):
|
|
return True
|
|
|
|
# Check for command starters
|
|
command_starters = ['tell me', 'explain', 'describe', 'show me', 'give me', 'find', 'search']
|
|
for starter in command_starters:
|
|
if text_lower.startswith(starter):
|
|
return True
|
|
|
|
return False
|
|
|
|
def _generate_answer(self, question):
|
|
"""Generate answer with interruption support"""
|
|
# Debounce: skip exact repeat within same run
|
|
if question.lower() == self._last_question.lower():
|
|
return
|
|
self._last_question = question
|
|
|
|
# Check if we should interrupt current answer
|
|
with self.answer_lock:
|
|
if self.answering:
|
|
print("🔄 Interrupting previous answer...")
|
|
if hasattr(self.ai_engine, 'interrupt'):
|
|
self.ai_engine.interrupt()
|
|
|
|
# Start answer generation in background thread
|
|
threading.Thread(target=self._answer_worker, args=(question,), daemon=True).start()
|
|
|
|
def _answer_worker(self, question):
|
|
"""Worker thread for answer generation"""
|
|
with self.answer_lock:
|
|
self.answering = True
|
|
|
|
try:
|
|
# Get context and generate answer
|
|
context = self.context_manager.get_context()
|
|
answer = self.ai_engine.answer_question(question, context)
|
|
|
|
# Show answer in overlay
|
|
self.overlay.show_answer(answer, question)
|
|
self.logger.info(f"Q: {question}")
|
|
self.logger.info(f"A: {answer}")
|
|
print(f"\n💡 Answer: {answer}\n")
|
|
|
|
except Exception as e:
|
|
print(f"❌ Error generating answer: {e}")
|
|
finally:
|
|
with self.answer_lock:
|
|
self.answering = False
|
|
|
|
# ------------------------------------------------------------------
|
|
# Manual terminal input for testing
|
|
# ------------------------------------------------------------------
|
|
|
|
def _terminal_input_loop(self):
|
|
print("\n Type a question and press Enter to test the AI directly.")
|
|
print(" (Leave blank to skip)\n")
|
|
while True:
|
|
try:
|
|
line = input()
|
|
if line.strip():
|
|
self._generate_answer(line.strip())
|
|
except (EOFError, KeyboardInterrupt):
|
|
break
|
|
|
|
# ------------------------------------------------------------------
|
|
# Run / shutdown
|
|
# ------------------------------------------------------------------
|
|
|
|
def run(self):
|
|
print("\n🎤 Starting audio listener...")
|
|
self.overlay.start()
|
|
self.audio_listener.start()
|
|
|
|
# Terminal input runs in a background daemon thread
|
|
t = threading.Thread(target=self._terminal_input_loop, daemon=True)
|
|
t.start()
|
|
|
|
try:
|
|
if self.overlay.app:
|
|
sys.exit(self.overlay.app.exec_())
|
|
except KeyboardInterrupt:
|
|
self.shutdown()
|
|
|
|
def shutdown(self):
|
|
print("\n🛑 Shutting down...")
|
|
if hasattr(self, 'audio_listener'):
|
|
self.audio_listener.stop()
|
|
if hasattr(self, 'overlay'):
|
|
self.overlay.stop()
|
|
print("👋 Goodbye!")
|
|
sys.exit(0)
|
|
|
|
|
|
def main():
|
|
assistant = MeetingAssistant()
|
|
assistant.run()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main() |