152 lines
4.8 KiB
Python
152 lines
4.8 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Meeting Assistant - Real-time AI Copilot for Meetings
|
|
"""
|
|
|
|
import sys
|
|
import os
|
|
import yaml
|
|
import threading
|
|
from pathlib import Path
|
|
|
|
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, config_path="config.yaml"):
|
|
print("🚀 Starting Meeting Assistant...")
|
|
|
|
config_file = Path(__file__).parent / config_path
|
|
with open(config_file, "r") as f:
|
|
self.config = yaml.safe_load(f)
|
|
|
|
self.logger = setup_logging()
|
|
print(f"💻 Platform: {get_platform()}")
|
|
|
|
self.context_manager = ContextManager(self.config)
|
|
self.ai_engine = AIEngine(self.config)
|
|
self.overlay = InvisibleOverlay(self.config)
|
|
self.audio_listener = AudioListener(self.config, self.on_audio_transcript)
|
|
|
|
self._last_question = "" # Simple debounce
|
|
|
|
# ------------------------------------------------------------------
|
|
# Audio callback — runs in the audio listener thread
|
|
# ------------------------------------------------------------------
|
|
|
|
def on_audio_transcript(self, text, timestamp):
|
|
self.context_manager.add_audio_context(text, timestamp)
|
|
|
|
if self._is_question(text):
|
|
print(f"❓ Question: {text}")
|
|
self._generate_answer(text)
|
|
else:
|
|
print(f" Context: {text}")
|
|
|
|
def _is_question(self, text):
|
|
if len(text.split()) < 3:
|
|
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
|
|
|
|
# Starts with or contains a question word
|
|
return any(text_lower.startswith(w) or f" {w} " in text_lower for w in _QUESTION_WORDS)
|
|
|
|
def _generate_answer(self, question):
|
|
# Debounce: skip exact repeat within same run
|
|
if question.lower() == self._last_question.lower():
|
|
return
|
|
self._last_question = question
|
|
|
|
context = self.context_manager.get_context()
|
|
answer = self.ai_engine.answer_question(question, context)
|
|
self.overlay.show_answer(answer, question)
|
|
self.logger.info(f"Q: {question}")
|
|
self.logger.info(f"A: {answer}")
|
|
print(f"💡 Answer: {answer}\n")
|
|
|
|
# ------------------------------------------------------------------
|
|
# 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✅ Meeting Assistant Ready!")
|
|
print("=" * 50)
|
|
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("=" * 50)
|
|
|
|
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...")
|
|
self.audio_listener.stop()
|
|
self.overlay.stop()
|
|
print("👋 Goodbye!")
|
|
|
|
|
|
def main():
|
|
assistant = MeetingAssistant()
|
|
assistant.run()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|