import os import json import threading import requests from vosk import Model, KaldiRecognizer import pyaudio from tkinter import Tk, Label, PhotoImage import webbrowser import queue import pyttsx3 from PIL import Image import time # Path to the Vosk model MODEL_PATH = r"C:\Users\subash\Downloads\vosk-model-small-en-us-0.15" # Home Assistant configuration HA_URL = "http://192.168.1.106:8123/api/states" # Replace with your HA API states URL HA_TOKEN = "YOUR_LONG_LIVED_ACCESS_TOKEN" # Replace with your token # Initialize the Vosk model if not os.path.exists(MODEL_PATH): raise FileNotFoundError(f"Model path not found: {MODEL_PATH}") model = Model(MODEL_PATH) # We create one PyAudio() instance for the whole app audio = pyaudio.PyAudio() # Initialize pyttsx3 for text-to-speech engine = pyttsx3.init() # Queue for UI communication ui_queue = queue.Queue() # We'll store the UI thread here ui_thread = None def fetch_sensor_data(): """ Fetches sensor data from Home Assistant. """ headers = { "Authorization": f"Bearer {HA_TOKEN}", "Content-Type": "application/json", } try: response = requests.get(HA_URL, headers=headers) response.raise_for_status() data = response.json() # Extract specific sensor data (change entity IDs if needed) temperature = next((item["state"] for item in data if item["entity_id"] == "sensor.weather_station_bmp_temperature"), "N/A") humidity = next((item["state"] for item in data if item["entity_id"] == "sensor.weather_station_humidity"), "N/A") windspeed = next((item["state"] for item in data if item["entity_id"] == "sensor.weather_station_wind_speed"), "N/A") winddirection = next((item["state"] for item in data if item["entity_id"] == "sensor.weather_station_wind_direction"), "N/A") rainfall = next((item["state"] for item in data if item["entity_id"] == "sensor.weather_station_total_rainfall"), "N/A") airpressure = next((item["state"] for item in data if item["entity_id"] == "sensor.weather_station_bmp_pressure"), "N/A") return ( f"Temperature: {temperature}°C,\n" f"Humidity: {humidity}%,\n" f"Wind Speed: {windspeed} m/s,\n" f"Wind Direction: {winddirection},\n" f"Rainfall: {rainfall} mm,\n" f"Air Pressure: {airpressure} hPa." ) except requests.exceptions.RequestException as e: return f"Error fetching sensor data: {e}" def start_ui(): """ Starts the UI in a separate thread and updates it based on queue messages. """ root = Tk() root.title("Voice Assistant") root.geometry("400x400") root.configure(bg="black") root.resizable(False, False) # Label for displaying text or status label = Label(root, text="Listening...\n", font=("Arial", 20), fg="white", bg="black") label.pack(expand=True) label.place(x=30, y=270) # Example GIF for animation gifImage = "audio\\listening.gif" openImage = Image.open(gifImage) frames = openImage.n_frames imageObject = [PhotoImage(file=gifImage, format=f"gif -index {i}") for i in range(frames)] count = 0 def animation(count): if not hasattr(animation, "showAnimation"): animation.showAnimation = None newImage = imageObject[count] gif_Label.configure(image=newImage) gif_Label.configure(background="black") count += 1 if count == frames: count = 0 animation.showAnimation = root.after(50, lambda: animation(count)) gif_Label = Label(root, image="") gif_Label.place(x=50, y=20, width=300, height=250) def update_ui(): """ Check the queue for messages. If "CLOSE_UI" is received, destroy the root (close the UI). """ try: message = ui_queue.get_nowait() if message == "CLOSE_UI": root.destroy() return # Stop updating once closed else: label.config(text=message) except queue.Empty: pass root.after(100, update_ui) # Start everything update_ui() animation(count) root.mainloop() def command_listener(): """ Listens for specific commands after wake-up word is detected. If no command is recognized for 5 seconds, closes the UI and returns to the wake_word_listener(). """ recognizer = KaldiRecognizer(model, 16000) # Audio stream for the command phase stream = audio.open(format=pyaudio.paInt16, channels=1, rate=16000, input=True, frames_per_buffer=4000) stream.start_stream() print("Listening for commands...") TIMEOUT_SECONDS = 5 last_command_time = time.time() while True: try: data = stream.read(4000, exception_on_overflow=False) # Check inactivity for 5 seconds if time.time() - last_command_time > TIMEOUT_SECONDS: print("No commands for 5 seconds. Closing UI and returning to wake word listener...") ui_queue.put("CLOSE_UI") # Ask the UI to close stream.stop_stream() stream.close() return if recognizer.AcceptWaveform(data): result = json.loads(recognizer.Result()) text = result.get("text", "").lower().strip() print(f"Command recognized: {text}") if text: # Reset inactivity timer last_command_time = time.time() # ----------------------- # Handle your commands here # ----------------------- if "display" in text: print("Command: Show Weather Station Dashboard") ui_queue.put("Command: Show Dashboard") elif "open" in text: print("Command: Open Weather Station Dashboard") webbrowser.open("http://192.168.0.106:8123/lovelace/default_view") ui_queue.put("Opened Dashboard") elif "update" in text: print("Command: Update Weather Station Status") sensor_data = fetch_sensor_data() print(sensor_data) ui_queue.put(sensor_data) engine.say(sensor_data) engine.runAndWait() else: # Partial (interim) results partial_result = json.loads(recognizer.PartialResult()) print(f"Partial command: {partial_result.get('partial', '')}") except KeyboardInterrupt: print("\nStopping command listener...") stream.stop_stream() stream.close() audio.terminate() break except Exception as e: print(f"Command listener error: {e}") break def wake_word_listener(): """ Continuously listens for the wake word ("tom"). Whenever it's detected, start a fresh UI and call command_listener(). After command_listener() returns, loop back here to listen again. """ global ui_thread print("Entering wake_word_listener. Waiting for wake word...") while True: try: # Create a new stream for each iteration so that we can detect "tom" again recognizer = KaldiRecognizer(model, 16000) stream = audio.open(format=pyaudio.paInt16, channels=1, rate=16000, input=True, frames_per_buffer=4000) stream.start_stream() while True: data = stream.read(4000, exception_on_overflow=False) if recognizer.AcceptWaveform(data): result = json.loads(recognizer.Result()) text = result.get("text", "").lower() print(f"Wake word recognized text: {text}") # If "tom" is detected, break to handle commands if "tom" in text: print("Wake word DETECTED, opening UI and switching to command listener.") stream.stop_stream() stream.close() # Start a NEW UI thread each time ui_thread = threading.Thread(target=start_ui, daemon=True) ui_thread.start() # Now handle commands command_listener() # After command_listener() returns, break from inner loop # to re-initialize the wake word stream again break else: # Partial results partial_result = json.loads(recognizer.PartialResult()) print(f"Partial wake word: {partial_result.get('partial', '')}") except KeyboardInterrupt: print("\nStopping wake_word_listener...") stream.stop_stream() stream.close() audio.terminate() break except Exception as e: print(f"Wake_word_listener error: {e}") break if __name__ == "__main__": try: wake_word_listener() except Exception as e: print(f"Main error: {e}")