mirror of
https://gitlab.sectorq.eu/jaydee/python.git
synced 2025-12-13 18:24:53 +01:00
85 lines
2.2 KiB
Python
85 lines
2.2 KiB
Python
#!/usr/bin/env python
|
|
from apa102_pi.driver import apa102
|
|
from time import sleep
|
|
import paho.mqtt.client as mqtt
|
|
import RPi.GPIO as GPIO
|
|
import json
|
|
import os
|
|
|
|
rhasspyConfig = '/home/jd/.config/rhasspy/profiles/en/profile.json'
|
|
|
|
counter = 0
|
|
LED = "on"
|
|
mute = "off"
|
|
siteId = ""
|
|
MQTThost = ""
|
|
|
|
GPIO.setmode(GPIO.BCM)
|
|
GPIO.setup(5, GPIO.OUT)
|
|
GPIO.output(5, GPIO.HIGH)
|
|
|
|
|
|
|
|
|
|
with open(rhasspyConfig,'r', encoding='UTF-8') as file:
|
|
obj = json.loads(file.read())
|
|
MQTTconfig = json.dumps(obj["mqtt"])
|
|
MQTTconfig = MQTTconfig.replace("\"mqtt\": ","")
|
|
MQTTconfig = json.loads(MQTTconfig)
|
|
siteId = MQTTconfig["site_id"]
|
|
MQTThost = MQTTconfig["host"]
|
|
MQTThost = MQTThost.strip('"')
|
|
if "port" in json.dumps(MQTTconfig):
|
|
MQTTport = MQTTconfig["port"]
|
|
MQTTport = MQTTport.strip('"')
|
|
MQTTport = int(MQTTport)
|
|
else:
|
|
MQTTport = 1883
|
|
|
|
strip = apa102.APA102(num_led=12)
|
|
strip.clear_strip()
|
|
|
|
def on_connect(client, userdata, flags, rc):
|
|
print("Connected with result code " + str(rc))
|
|
client.subscribe("hermes/dialogueManager/sessionEnded/#")
|
|
client.subscribe("hermes/hotword/toggleOff/#")
|
|
client.subscribe("hermes/asr/stopListening")
|
|
for i in range(0,12):
|
|
strip.set_pixel(i,0,127,0,7)
|
|
strip.show()
|
|
sleep(2)
|
|
for i in range(0,12):
|
|
strip.set_pixel(i,0,255,0,0)
|
|
strip.show()
|
|
def on_message(client, userdata, msg):
|
|
jsonData = json.loads(msg.payload)
|
|
if msg.topic == "hermes/hotword/toggleOff" and jsonData["siteId"] == siteId and LED == "on":
|
|
for i in range(0,12):
|
|
strip.set_pixel(i,0,127,0,7)
|
|
strip.show()
|
|
elif msg.topic == "hermes/dialogueManager/sessionEnded" and jsonData["siteId"] == siteId:
|
|
for i in range(0,12):
|
|
strip.set_pixel(i,0,255,0,0)
|
|
strip.show()
|
|
|
|
elif msg.topic == "hermes/asr/stopListening":
|
|
for i in range(0,12):
|
|
strip.set_pixel(i,0,255,0,0)
|
|
strip.show()
|
|
|
|
for i in range(0,12):
|
|
strip.set_pixel(i,0,127,255,7)
|
|
sleep(0.1)
|
|
strip.show()
|
|
|
|
sleep(3)
|
|
|
|
|
|
client = mqtt.Client()
|
|
client.username_pw_set("jaydee", password="jaydee1")
|
|
client.on_connect = on_connect
|
|
client.on_message = on_message
|
|
client.connect(MQTThost, MQTTport, 60)
|
|
|
|
client.loop_forever()
|