# main_assistant.py import speech_recognition as sr import pyttsx3 import threading import json import os import time import logging from datetime import datetime from typing import Dict, List, Optional, Callable import queue import re class VirtualAssistant: def __init__(self, name: str = "JARVIS"): self.name = name self.is_listening = False self.is_active = True self.wake_words = ["jarvis", "hey jarvis", "assistant"] # Initialize components self._setup_logging() self._initialize_speech_recognition() self._initialize_text_to_speech() self._load_configuration() self._initialize_modules() # Command queue for thread safety self.command_queue = queue.Queue() # User preferences and memory self.user_preferences = self._load_user_preferences() self.conversation_history = [] self.logger.info(f"{self.name} initialized successfully") def _setup_logging(self): """Setup logging configuration""" logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', handlers=[ logging.FileHandler('assistant.log'), logging.StreamHandler() ] ) self.logger = logging.getLogger(self.name) def _initialize_speech_recognition(self): """Initialize speech recognition components""" self.recognizer = sr.Recognizer() self.microphone = sr.Microphone() # Adjust for ambient noise with self.microphone as source: self.recognizer.adjust_for_ambient_noise(source, duration=1) # Configure recognition parameters self.recognizer.energy_threshold = 300 self.recognizer.dynamic_energy_threshold = True self.recognizer.pause_threshold = 0.8 self.recognizer.phrase_threshold = 0.3 def _initialize_text_to_speech(self): """Initialize text-to-speech engine""" self.tts_engine = pyttsx3.init() # Configure voice properties voices = self.tts_engine.getProperty('voices') if voices: # Prefer female voice if available for voice in voices: if 'female' in voice.name.lower() or 'zira' in voice.name.lower(): self.tts_engine.setProperty('voice', voice.id) break self.tts_engine.setProperty('rate', 180) # Speech rate self.tts_engine.setProperty('volume', 0.8) # Volume level def _load_configuration(self): """Load system configuration""" config_file = 'config.json' default_config = { "wake_words": ["jarvis", "hey jarvis", "assistant"], "response_timeout": 5, "listening_timeout": 10, "voice_feedback": True, "debug_mode": False } if os.path.exists(config_file): try: with open(config_file, 'r') as f: self.config = json.load(f) except Exception as e: self.logger.warning(f"Error loading config: {e}") self.config = default_config else: self.config = default_config self._save_configuration() def _save_configuration(self): """Save current configuration""" try: with open('config.json', 'w') as f: json.dump(self.config, f, indent=4) except Exception as e: self.logger.error(f"Error saving config: {e}") def _load_user_preferences(self) -> Dict: """Load user preferences""" prefs_file = 'user_preferences.json' default_prefs = { "name": "User", "preferred_units": "metric", "time_format": "24h", "reminders": [], "favorite_topics": [] } if os.path.exists(prefs_file): try: with open(prefs_file, 'r') as f: return json.load(f) except Exception as e: self.logger.warning(f"Error loading preferences: {e}") return default_prefs def _save_user_preferences(self): """Save user preferences""" try: with open('user_preferences.json', 'w') as f: json.dump(self.user_preferences, f, indent=4) except Exception as e: self.logger.error(f"Error saving preferences: {e}") def speak(self, text: str, priority: bool = False): """Convert text to speech""" if not self.config.get('voice_feedback', True): print(f"{self.name}: {text}") return try: self.logger.info(f"Speaking: {text}") if priority: # Stop current speech for priority messages self.tts_engine.stop() self.tts_engine.say(text) self.tts_engine.runAndWait() except Exception as e: self.logger.error(f"TTS Error: {e}") print(f"{self.name}: {text}") def listen(self, timeout: int = None) -> Optional[str]: """Listen for voice input""" if timeout is None: timeout = self.config.get('listening_timeout', 10) try: with self.microphone as source: self.logger.info("Listening...") audio = self.recognizer.listen(source, timeout=timeout, phrase_time_limit=5) # Recognize speech text = self.recognizer.recognize_google(audio, language='en-US') self.logger.info(f"Recognized: {text}") return text.lower() except sr.WaitTimeoutError: self.logger.debug("Listening timeout") return None except sr.UnknownValueError: self.logger.debug("Could not understand audio") return None except sr.RequestError as e: # Fallback to offline recognition if available try: text = self.recognizer.recognize_sphinx(audio) self.logger.info(f"Offline recognition: {text}") return text.lower() except: self.logger.error(f"Speech recognition error: {e}") return None def _initialize_modules(self): """Initialize task modules""" from modules.system_info import SystemInfoModule from modules.file_operations import FileOperationsModule from modules.calculator import CalculatorModule from modules.weather import WeatherModule from modules.reminders import ReminderModule from modules.knowledge_base import KnowledgeBaseModule self.modules = { 'system': SystemInfoModule(self), 'files': FileOperationsModule(self), 'calculator': CalculatorModule(self), 'weather': WeatherModule(self), 'reminders': ReminderModule(self), 'knowledge': KnowledgeBaseModule(self) } def process_command(self, command: str) -> bool: """Process voice command""" if not command: return True # Add to conversation history self.conversation_history.append({ 'timestamp': datetime.now().isoformat(), 'user_input': command, 'type': 'voice_command' }) # Check for exit commands exit_commands = ['exit', 'quit', 'goodbye', 'shut down', 'turn off'] if any(exit_cmd in command for exit_cmd in exit_commands): self.speak("Goodbye! Have a great day!") return False # Check for wake word if not actively listening if not self.is_listening: if not any(wake_word in command for wake_word in self.wake_words): return True # Remove wake word from command for wake_word in self.wake_words: command = command.replace(wake_word, '').strip() if not command: self.speak("Yes, how can I help you?") self.is_listening = True return True # Process command through modules handled = False for module_name, module in self.modules.items(): if module.can_handle(command): try: response = module.handle(command) if response: self.speak(response) handled = True break except Exception as e: self.logger.error(f"Module {module_name} error: {e}") self.speak("I encountered an error processing that request.") if not handled: self.speak("I'm not sure how to help with that. Could you try rephrasing?") return True def run(self): """Main assistant loop""" self.speak(f"Hello! {self.name} is now active. How can I assist you today?") while self.is_active: try: command = self.listen() if command: if not self.process_command(command): break # Reset listening state after period of inactivity if self.is_listening: time.sleep(0.1) # Small delay to prevent excessive CPU usage except KeyboardInterrupt: self.speak("Shutting down...") break except Exception as e: self.logger.error(f"Main loop error: {e}") time.sleep(1) self.shutdown() def shutdown(self): """Cleanup and shutdown""" self.is_active = False self._save_user_preferences() self._save_configuration() self.logger.info(f"{self.name} shutting down")