375 lines
14 KiB
Python
375 lines
14 KiB
Python
import os
|
|
import subprocess
|
|
import sys
|
|
import threading
|
|
import time
|
|
import webbrowser
|
|
from datetime import date, datetime
|
|
|
|
import pyautogui
|
|
import speech_recognition as sr
|
|
from PySide6.QtCore import QObject, Signal
|
|
from PySide6.QtGui import QIcon
|
|
from PySide6.QtWidgets import QApplication, QMenu, QSystemTrayIcon
|
|
|
|
from audio.tts import speak
|
|
from bs.verity import ouiouioui
|
|
|
|
# External file imports
|
|
from core.ui import RisaAssistantPopup
|
|
from utils.ai_client import ask_ai
|
|
from utils.time_utils import time_to_words
|
|
from utils.weather import get_local_weather
|
|
|
|
popup_window = None
|
|
trigger_bridge = None
|
|
tray_icon = None
|
|
|
|
recognizer = sr.Recognizer()
|
|
microphone = sr.Microphone()
|
|
|
|
recognizer.dynamic_energy_threshold = True
|
|
recognizer.pause_threshold = 0.8
|
|
|
|
class VoiceTriggerBridge(QObject):
|
|
hide_window = Signal()
|
|
show_input_panel = Signal()
|
|
show_thinking_panel = Signal(str)
|
|
update_display_card = Signal(str, dict)
|
|
show_timeout_card = Signal(str)
|
|
update_state = Signal(str)
|
|
|
|
# Commands
|
|
def process_command(command: str) -> bool:
|
|
global trigger_bridge
|
|
command = command.lower().strip()
|
|
if not command:
|
|
return False
|
|
|
|
print(f"Executing command: '{command}'")
|
|
|
|
# Time Commands
|
|
if any(word in command for word in ["time", "what time", "current time"]):
|
|
now = datetime.now()
|
|
time_words = time_to_words(now)
|
|
if trigger_bridge:
|
|
time_data = {
|
|
"time": now.strftime("%I:%M %p").lstrip('0'),
|
|
"day": now.strftime("%A"),
|
|
"date": now.strftime("%B %d, %Y")
|
|
}
|
|
trigger_bridge.update_display_card.emit("time", time_data)
|
|
speak(f"The time is {time_words}")
|
|
return True
|
|
|
|
elif any(word in command for word in ["date", "what's the date"]):
|
|
today_date = date.today().strftime("%B %d, %Y")
|
|
if trigger_bridge:
|
|
trigger_bridge.update_display_card.emit("general", {"text": f"Today is {today_date}."})
|
|
speak(f"The date is {today_date}")
|
|
return True
|
|
|
|
# Weather
|
|
elif any(word in command for word in ["weather", "weather today", "what is the weather"]):
|
|
weather = get_local_weather()
|
|
if weather:
|
|
city = weather['location']['city']
|
|
temp_unit = weather['units']['temp']
|
|
current_temp = weather['current_temp']
|
|
high_temp = weather['high_temp']
|
|
low_temp = weather['low_temp']
|
|
clouds = weather['cloud_cover_pct']
|
|
rain = weather['precipitation']
|
|
|
|
if clouds < 10: cloud_phrase = "Clear skies"
|
|
elif clouds < 40: cloud_phrase = "Partly cloudy skies"
|
|
elif clouds < 75: cloud_phrase = "Mostly cloudy skies"
|
|
else: cloud_phrase = "Completely overcast"
|
|
|
|
if rain > 0:
|
|
cloud_phrase += ", rain soon."
|
|
|
|
if trigger_bridge:
|
|
trigger_bridge.update_display_card.emit("weather", {
|
|
"temp": str(current_temp),
|
|
"humidity": str(clouds),
|
|
"high": str(high_temp),
|
|
"low": str(low_temp),
|
|
"condition": cloud_phrase,
|
|
"location": f"{city} NJ"
|
|
})
|
|
print(f"Right now in {city}, it is {current_temp}{temp_unit} with {cloud_phrase.lower()}. Today's high will be {high_temp} degrees, and the low will be {low_temp} degrees.")
|
|
speak(f"Right now in {city}, it is {current_temp}{temp_unit} with {cloud_phrase.lower()}. Today's high will be {high_temp} degrees, and the low will be {low_temp} degrees.")
|
|
return True
|
|
return False
|
|
|
|
# Identity
|
|
elif any(phrase in command for phrase in ["who are you", "your name", "what do you do"]):
|
|
if trigger_bridge:
|
|
trigger_bridge.update_display_card.emit("general", {"text": "Hi, my name is Risa. I'm your voice assistant."})
|
|
speak("Hi, my name is Risa. I'm your voice assistant.")
|
|
return True
|
|
elif any(phrase in command for phrase in ["what is the capital of france", "What is the capital of france", "What's the capital of france"]):
|
|
if trigger_bridge:
|
|
trigger_bridge.update_display_card.emit("general", {"text": "Oui oui oui! It paris"})
|
|
ouiouioui()
|
|
return True
|
|
# Media Controls
|
|
elif any(phrase in command for phrase in ["pause", "pause music", "pause video", "play", "resume"]):
|
|
pyautogui.press('playpause')
|
|
speak("Done")
|
|
return True
|
|
|
|
elif "volume up" in command:
|
|
for _ in range(5): pyautogui.press("volumeup")
|
|
speak("Volume up")
|
|
return True
|
|
|
|
elif "volume down" in command:
|
|
for _ in range(5): pyautogui.press("volumedown")
|
|
speak("Volume down")
|
|
return True
|
|
|
|
elif "mute" in command:
|
|
pyautogui.press("volumemute")
|
|
return True
|
|
|
|
# Web Controls
|
|
elif "open google" in command:
|
|
speak("Opening Google")
|
|
webbrowser.open("https://google.com")
|
|
return True
|
|
|
|
elif "open youtube" in command:
|
|
speak("Opening Youtube")
|
|
webbrowser.open("https://youtube.com")
|
|
return True
|
|
|
|
elif "search for" in command:
|
|
query = command.split("search for")[-1].strip()
|
|
if query:
|
|
if trigger_bridge:
|
|
trigger_bridge.update_display_card.emit("general", {"text": f"Searching Google for: {query}"})
|
|
speak(f"Searching for {query}")
|
|
webbrowser.open(f"https://www.google.com/search?q={query}")
|
|
return True
|
|
|
|
# Apps
|
|
elif "open calculator" in command:
|
|
speak("Opening Calculator")
|
|
if os.name == 'nt': subprocess.Popen('calc.exe')
|
|
else: subprocess.Popen('gnome-calculator' if os.name == 'posix' else 'open -a Calculator')
|
|
return True
|
|
# Sleep
|
|
elif any(phrase in command for phrase in ["stop listening", "go to sleep", "hide"]):
|
|
speak("Going to sleep.")
|
|
if popup_window and trigger_bridge:
|
|
trigger_bridge.hide_window.emit()
|
|
trigger_bridge.update_state.emit("idle")
|
|
return False
|
|
|
|
|
|
# Commands List
|
|
elif any(phrase in command for phrase in ["list commands", "what are your commands", "show commands"]):
|
|
commands_html = """
|
|
<div style="line-height: 140%;">
|
|
<b style="color: #ff8c00; font-size: 15px;">System Capabilities:</b><br>
|
|
<table width="100%" style="margin-top: 5px; font-size: 13px; color: #111111;">
|
|
<tr><td>• <b>Time / Date:</b> "What time is it?", "What's the date?"</td></tr>
|
|
<tr><td>• <b>Weather:</b> "What is the weather today?"</td></tr>
|
|
<tr><td>• <b>Controls:</b> "Volume up/down", "Mute", "Pause/Play"</td></tr>
|
|
<tr><td>• <b>Apps & Web:</b> "Open Google", "Search for [query]"</td></tr>
|
|
<tr><td>• <b>AI Core:</b> "Ask AI [your question]"</td></tr>
|
|
<tr><td>• <b>Sleep:</b> "Stop listening", "Go to sleep"</td></tr>
|
|
</table>
|
|
</div>
|
|
"""
|
|
if trigger_bridge:
|
|
trigger_bridge.update_display_card.emit("general", {"text": commands_html})
|
|
speak("Here is a list of commands.")
|
|
return True
|
|
|
|
else:
|
|
# AI Fallback Query
|
|
prompt = command
|
|
if prompt.startswith("ask ai") or prompt.startswith("ask a i"):
|
|
for phrase in ["ask ai", "ask a i"]:
|
|
if prompt.startswith(phrase):
|
|
prompt = prompt.replace(phrase, "", 1).strip(" :")
|
|
break
|
|
|
|
if prompt:
|
|
if trigger_bridge:
|
|
trigger_bridge.show_thinking_panel.emit(prompt)
|
|
trigger_bridge.update_state.emit("thinking")
|
|
|
|
speak("Thinking about it...")
|
|
ai_response = ask_ai(prompt)
|
|
|
|
if trigger_bridge and ai_response:
|
|
trigger_bridge.update_display_card.emit("general", {"text": str(ai_response)})
|
|
elif trigger_bridge:
|
|
trigger_bridge.update_display_card.emit("general", {"text": f"Processed request: '{prompt}'"})
|
|
|
|
return True
|
|
else:
|
|
speak("What would you like to ask?")
|
|
return True
|
|
|
|
# Voice listener loop
|
|
def voice_listener_loop():
|
|
global trigger_bridge
|
|
|
|
with microphone as source:
|
|
recognizer.adjust_for_ambient_noise(source, duration=1.5)
|
|
print("\nRisa is active... Say 'Hey Risa [your command]' in one sentence!")
|
|
in_conversation = False
|
|
timeout_emitted = False
|
|
|
|
if trigger_bridge:
|
|
trigger_bridge.update_state.emit("idle")
|
|
|
|
while True:
|
|
try:
|
|
if not in_conversation:
|
|
if trigger_bridge:
|
|
trigger_bridge.update_state.emit("idle")
|
|
print(".", end="", flush=True)
|
|
|
|
audio = recognizer.listen(source, timeout=None, phrase_time_limit=10)
|
|
raw_text = recognizer.recognize_google(audio).lower()
|
|
|
|
if "hey risa" in raw_text or "hey lisa" in raw_text:
|
|
print(f"\n[Triggered]: '{raw_text}'")
|
|
timeout_emitted = False
|
|
|
|
if trigger_bridge:
|
|
trigger_bridge.show_input_panel.emit()
|
|
trigger_bridge.update_state.emit("active")
|
|
|
|
if "risa" in raw_text:
|
|
command_text = raw_text.split("risa")[-1].strip()
|
|
else:
|
|
command_text = raw_text.split("lisa")[-1].strip()
|
|
|
|
if command_text:
|
|
executed = process_command(command_text)
|
|
if executed:
|
|
in_conversation = True
|
|
else:
|
|
speak("How can I help?")
|
|
in_conversation = True
|
|
else:
|
|
print("\n[Follow-up Mode: Listening...]")
|
|
if trigger_bridge:
|
|
trigger_bridge.update_state.emit("active")
|
|
time.sleep(0.4)
|
|
|
|
audio = recognizer.listen(source, timeout=6, phrase_time_limit=6)
|
|
command_text = recognizer.recognize_google(audio).lower()
|
|
print(f"[Follow-up Command Received]: '{command_text}'")
|
|
|
|
timeout_emitted = False
|
|
executed = process_command(command_text)
|
|
if not executed:
|
|
in_conversation = False
|
|
|
|
except (sr.UnknownValueError, sr.WaitTimeoutError, sr.RequestError) as e:
|
|
if in_conversation:
|
|
print("\n[Follow-up timeout or unrecognized audio: resetting context]")
|
|
in_conversation = False
|
|
|
|
if not timeout_emitted:
|
|
if trigger_bridge:
|
|
error_msg = "Risa could not hear you clearly or the request timed out."
|
|
if isinstance(e, sr.WaitTimeoutError):
|
|
error_msg = "Follow-up timed out. No speech detected."
|
|
|
|
html_error = f"""
|
|
<div style="line-height: 140%;">
|
|
<span style="color: #d9534f; font-size: 12px; font-weight: bold; text-transform: uppercase; letter-spacing: 1px;">Timeout</span><br>
|
|
<div style="padding-top: 6px; color: #333333; font-weight: 500;">{error_msg} Please reinitiate Risa by saying: Hey Risa</div>
|
|
</div>
|
|
"""
|
|
trigger_bridge.update_display_card.emit("general", {"text": html_error})
|
|
trigger_bridge.update_state.emit("idle")
|
|
timeout_emitted = True
|
|
|
|
except Exception as e:
|
|
print(f"\nUnexpected error: {e}")
|
|
in_conversation = False
|
|
if trigger_bridge:
|
|
trigger_bridge.update_state.emit("idle")
|
|
|
|
def handle_typed_command(command_text: str):
|
|
def run():
|
|
if trigger_bridge:
|
|
trigger_bridge.show_input_panel.emit()
|
|
trigger_bridge.update_state.emit("active")
|
|
|
|
process_command(command_text)
|
|
|
|
threading.Thread(target=run, daemon=True).start()
|
|
|
|
def trigger_from_tray():
|
|
if popup_window:
|
|
popup_window.show()
|
|
popup_window.raise_()
|
|
popup_window.activateWindow()
|
|
if trigger_bridge:
|
|
trigger_bridge.show_input_panel.emit()
|
|
trigger_bridge.update_state.emit("active")
|
|
|
|
def setup_system_tray(app):
|
|
global tray_icon
|
|
|
|
app.setQuitOnLastWindowClosed(False)
|
|
|
|
icon = QIcon("ico/icon.png")
|
|
|
|
tray_icon = QSystemTrayIcon(icon, app)
|
|
tray_icon.setToolTip("Risa Assistant")
|
|
|
|
tray_menu = QMenu()
|
|
|
|
show_action = tray_menu.addAction("Open Risa")
|
|
show_action.triggered.connect(trigger_from_tray)
|
|
|
|
tray_menu.addSeparator()
|
|
|
|
quit_action = tray_menu.addAction("Quit")
|
|
quit_action.triggered.connect(app.quit)
|
|
|
|
tray_icon.setContextMenu(tray_menu)
|
|
|
|
def on_tray_activated(reason):
|
|
if reason in (QSystemTrayIcon.Trigger, QSystemTrayIcon.DoubleClick):
|
|
trigger_from_tray()
|
|
|
|
tray_icon.activated.connect(on_tray_activated)
|
|
tray_icon.show()
|
|
|
|
def main():
|
|
global popup_window, trigger_bridge
|
|
app = QApplication(sys.argv)
|
|
|
|
setup_system_tray(app)
|
|
|
|
trigger_bridge = VoiceTriggerBridge()
|
|
popup_window = RisaAssistantPopup()
|
|
|
|
trigger_bridge.show_input_panel.connect(popup_window.show_initial_input)
|
|
trigger_bridge.show_thinking_panel.connect(popup_window.show_thinking)
|
|
trigger_bridge.update_display_card.connect(popup_window.display_command_card)
|
|
trigger_bridge.hide_window.connect(popup_window.hide)
|
|
trigger_bridge.update_state.connect(popup_window.set_listening_state)
|
|
|
|
popup_window.text_submitted.connect(handle_typed_command)
|
|
|
|
listener_thread = threading.Thread(target=voice_listener_loop, daemon=True)
|
|
listener_thread.start()
|
|
|
|
sys.exit(app.exec())
|
|
|
|
if __name__ == "__main__":
|
|
main() |