jarvis/main.py at main · SumitWagdare/jarvis · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
import time
# Optimized for low latency performance on Mac M2
import threading
import os
from core.listener import ClapListener
from core.speaker import Speaker
from core.stt_engine import STTEngine
from core.actions import JarvisActions
from core.brain import JarvisBrain
from core.weather import WeatherEngine
from ui.dashboard import JarvisUI

class Jarvis:
    def __init__(self):
        self.is_awake = False
        self.speaker = Speaker()
        self.listener = ClapListener(threshold=0.3)
        self.stt = STTEngine()
        self.actions = JarvisActions()
        self.brain = JarvisBrain() # Uses GOOGLE_API_KEY from env if available
        self.weather = WeatherEngine()
        self.ui = JarvisUI('ui/index.html')
        self.running = True

    def run_audio_loop(self):
        """Monitor for claps in the background."""
        self.listener.start()
        print(">>> JARVIS: STANDBY MODE (Clap to wake up) <<<")

        try:
            while self.running:
                if not self.is_awake:
                    if self.listener.listen():
                        self.wake_up()
                time.sleep(0.01)
        except Exception as e:
            print(f"Error in audio loop: {e}")

    def run_hud_update_loop(self):
        """Background loop to update the Visual HUD stats and weather."""
        weather_timer = 0
        while self.running:
            # Update System Stats every 2 seconds
            stats = self.actions.get_system_stats()
            self.ui.update_stats(stats['cpu'], stats['ram'])

            # Update Weather every 10 minutes (600 seconds)
            if weather_timer <= 0:
                w_data = self.weather.get_current_weather()
                if w_data:
                    cond = self.weather.get_condition_string(w_data['code'])
                    self.ui.update_weather(w_data['temp'], cond)
                weather_timer = 600

            weather_timer -= 2
            time.sleep(2)

    def wake_up(self):
        self.is_awake = True
        print(">>> JARVIS: AWAKE <<<")
        self.ui.trigger_activation_animation()
        self.ui.toggle_hud(True) # Show HUD on wake-up
        self.speaker.speak("I am online, Sumit. How can I assist you today?")

        # Start command listening cycle
        self.listen_for_command()

    def listen_for_command(self):
        """Handle the AI-powered command cycle."""
        command_text = self.stt.listen_for_command()

        if command_text:
            self.ui.trigger_thinking_animation()
            # Use the AI Brain to reason about the command
            intent = self.brain.reason(command_text)
            self.execute_intent(intent)
        else:
            self.speaker.speak("I didn't catch that. Returning to standby.")
            self.go_to_sleep()

    def execute_intent(self, intent):
        """Map AI-parsed intents to system actions."""
        if not intent:
            return

        self.ui.trigger_executing_animation()
        action = intent.get('action')
        params = intent.get('params')
        response = intent.get('response', "Processing command.")

        # Logic mapping
        if action == "open_app":
            self.speaker.speak(response)
            self.actions.open_app(params)
        elif action == "set_volume":
            self.speaker.speak(response)
            self.actions.set_volume(params)
        elif action == "start_work":
            self.speaker.speak(response)
            self.actions.start_work_flow()
        elif action == "get_stats":
            stats = self.actions.get_system_stats()
            self.speaker.speak(f"CPU is at {stats['cpu']} percent, RAM is at {stats['ram']} percent.")
        elif action == "get_weather":
            w_data = self.weather.get_current_weather()
            if w_data:
                cond = self.weather.get_condition_string(w_data['code'])
                self.speaker.speak(f"The current temperature is {w_data['temp']} degrees Celsius with {cond}.")
            else:
                self.speaker.speak("I'm unable to reach the SkyPulse weather service right now.")
        elif action == "go_to_sleep":
            self.speaker.speak(response)
            self.go_to_sleep()
        else:
            # General Chat response
            self.speaker.speak(response)

        # Stay awake for a moment or go back to sleep
        time.sleep(2)
        self.go_to_sleep()

    def go_to_sleep(self):
        self.is_awake = False
        print(">>> JARVIS: STANDBY <<<")
        self.ui.trigger_standby_animation()
        # Slide HUD out after a delay or keep it visible based on preference
        # self.ui.toggle_hud(False)

    def start(self):
        # Start background threads
        threading.Thread(target=self.run_audio_loop, daemon=True).start()
        threading.Thread(target=self.run_hud_update_loop, daemon=True).start()

        # Start UI
        self.ui.start()

if __name__ == "__main__":
    jarvis = Jarvis()
    try:
        jarvis.start()
    except KeyboardInterrupt:
        print("\nShutting down Jarvis...")
        jarvis.running = False