PYTHON

1def process_event(assistant, led, event):
2    # Different events trigger different actions:
3    if event.type == EventType.ON_START_FINISHED:
4        # Ready to listen
5    elif event.type == EventType.ON_CONVERSATION_TURN_STARTED:
6        # You started talking
7    elif event.type == EventType.ON_RECOGNIZING_SPEECH_FINISHED:
8        # It understood what you said - THIS IS WHERE WE ADD COMMANDS!

PYTHON

1def get_temp():
2    # Gets the temperature and returns a message
3    return "My CPU temperature is 45 degrees Celsius"
4
5def get_uptime():
6    # Checks how long the system has been running
7    return "The system has been up for 2 hours"

PYTHON

1elif text == 'cpu temperature':
2    assistant.stop_conversation()  # Stop Google from answering
3    tts.say(get_temp())           # Use OUR function instead
4
5elif text == 'uptime':
6    assistant.stop_conversation()
7    tts.say(get_uptime())

PYTHON

1
2#!/usr/bin/env python3
3# Copyright 2017 Google Inc.
4#
5# Licensed under the Apache License, Version 2.0 (the "License");
6# you may not use this file except in compliance with the License.
7# You may obtain a copy of the License at
8#
9#     http://www.apache.org/licenses/LICENSE-2.0
10#
11# Unless required by applicable law or agreed to in writing, software
12# distributed under the License is distributed on an "AS IS" BASIS,
13# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14# See the License for the specific language governing permissions and
15# limitations under the License.
16
17"""Run a recognizer using the Google Assistant Library.
18
19The Google Assistant Library has direct access to the audio API, so this Python
20code doesn't need to record audio. Hot word detection "OK, Google" is supported.
21
22It is available for Raspberry Pi 2/3 only; Pi Zero is not supported.
23"""
24
25import logging
26import platform
27import subprocess
28import sys
29import requests
30import random
31import re
32
33from google.assistant.library.event import EventType
34
35from aiy.assistant import auth_helpers
36from aiy.assistant.library import Assistant
37from aiy.board import Board, Led
38from aiy.voice import tts
39
40def power_off_pi():
41    tts.say('Good bye!')
42    subprocess.call('sudo shutdown now', shell=True)
43
44
45def reboot_pi():
46    tts.say('See you in a bit!')
47    subprocess.call('sudo reboot', shell=True)
48
49
50def say_ip():
51    ip_address = subprocess.check_output("hostname -I | cut -d' ' -f1", shell=True)
52    tts.say('My IP address is %s' % ip_address.decode('utf-8'))
53
54
55def get_temp():
56    try:
57        cpu_temp_output = subprocess.check_output("vcgencmd measure_temp", shell=True).decode('utf-8')
58        # Extract just the temperature number (removes "temp=" and "'C\n")
59        temp_value = cpu_temp_output.replace("temp=", "").replace("'C\n", "")
60        return f'My CPU temperature is {temp_value} degrees Celsius'
61    except subprocess.CalledProcessError:
62        return "Sorry, I couldn't read the CPU temperature"
63    
64def get_uptime():
65    output = subprocess.check_output("uptime -p", shell=True, universal_newlines=True)
66    # The output is like "up 2 hours, 30 minutes"
67    # Remove the "up" and return
68    uptime_str = output.strip().replace("up ", "")
69    return f"The system has been up for {uptime_str}"
70
71
72def get_public_ip():
73    services = [
74        'https://api.ipify.org',
75        'https://icanhazip.com',
76        'https://ident.me',
77        'https://checkip.amazonaws.com'
78    ]
79    
80    for service in services:
81        try:
82            import requests
83            response = requests.get(service, timeout=5)
84            if response.status_code == 200:
85                public_ip = response.text.strip()
86                return f'My public IP address is {public_ip}'
87        except Exception as e:
88            logging.warning(f"Failed to get IP from {service}: {e}")
89            continue
90    
91    return "Sorry, I couldn't retrieve the public IP address from any service"
92    
93    
94def get_usb_devices():
95    output = subprocess.check_output("lsusb", shell=True, universal_newlines=True)
96    # Count the lines
97    devices = output.strip().splitlines()
98    count = len(devices)
99    return f"There are {count} USB devices connected"
100
101
102def get_system_load():
103    output = subprocess.check_output("uptime", shell=True, universal_newlines=True)
104    # The output has load averages at the end
105    # Example: " 18:28:42 up 1:23, 1 user, load average: 0.00, 0.00, 0.00"
106    parts = output.split()
107    load_avgs = parts[-3], parts[-2], parts[-1]
108    return f"The load averages are {load_avgs[0]}, {load_avgs[1]}, and {load_avgs[2]} for the past 1, 5, and 15 minutes"
109   
110    
111    
112def process_event(assistant, led, event):
113    logging.info(event)
114    if event.type == EventType.ON_START_FINISHED:
115        led.state = Led.BEACON_DARK  # Ready.
116        print('Say "OK, Google" then speak, or press Ctrl+C to quit...')
117    elif event.type == EventType.ON_CONVERSATION_TURN_STARTED:
118        led.state = Led.ON  # Listening.
119    elif event.type == EventType.ON_RECOGNIZING_SPEECH_FINISHED and event.args:
120        print('You said:', event.args['text'])
121        text = event.args['text'].lower()
122        if text == 'power off':
123            assistant.stop_conversation()
124            power_off_pi()
125        elif text == 'reboot':
126            assistant.stop_conversation()
127            reboot_pi()
128        elif text == 'ip address':
129            assistant.stop_conversation()
130            say_ip()
131        elif text == 'cpu temperature':  # Added this condition
132            assistant.stop_conversation()
133            tts.say(get_temp())
134        elif text == 'uptime':  # Added this condition
135            assistant.stop_conversation()
136            tts.say(get_uptime())
137        elif text == 'public ip address':  # Added this condition
138            assistant.stop_conversation()
139            tts.say(get_public_ip())
140        elif text == 'usb devices':  # Added this condition
141            assistant.stop_conversation()
142            tts.say(get_usb_devices())
143        elif text == 'system load':  # Added this condition
144            assistant.stop_conversation()
145            tts.say(get_system_load())       
146
147            
148    elif event.type == EventType.ON_END_OF_UTTERANCE:
149        led.state = Led.PULSE_QUICK  # Thinking.
150    elif (event.type == EventType.ON_CONVERSATION_TURN_FINISHED
151          or event.type == EventType.ON_CONVERSATION_TURN_TIMEOUT
152          or event.type == EventType.ON_NO_RESPONSE):
153        led.state = Led.BEACON_DARK  # Ready.
154    elif event.type == EventType.ON_ASSISTANT_ERROR and event.args and event.args['is_fatal']:
155        sys.exit(1)
156
157
158def main():
159    logging.basicConfig(level=logging.INFO)
160
161    credentials = auth_helpers.get_assistant_credentials()
162    with Board() as board, Assistant(credentials) as assistant:
163        for event in assistant.start():
164            process_event(assistant, board.led, event)
165
166
167if __name__ == '__main__':
168    main()
169