76 lines
2.0 KiB
Python
76 lines
2.0 KiB
Python
import random
|
|
import json
|
|
|
|
from paho.mqtt import client as mqtt_client
|
|
from server import run_bulb_action
|
|
from colormap import rgb2hex
|
|
|
|
|
|
broker = "vps.k-world.me.uk"
|
|
port = 1883
|
|
topic = "ledLight/set"
|
|
# generate client ID with pub prefix randomly
|
|
client_id = f"python-mqtt-{random.randint(0, 100)}"
|
|
username = "karl"
|
|
password = "karlmax"
|
|
|
|
|
|
def convert_to_bulb_format(colour_map):
|
|
hex = rgb2hex(
|
|
colour_map["color"]["r"], colour_map["color"]["g"], colour_map["color"]["b"]
|
|
)
|
|
hex = hex.replace("#", "")
|
|
value = "56" + hex + "00f0aa"
|
|
return value
|
|
|
|
|
|
def connect_mqtt() -> mqtt_client:
|
|
def on_connect(client, userdata, flags, rc):
|
|
if rc == 0:
|
|
print("Connected to MQTT Broker!")
|
|
else:
|
|
print("Failed to connect, return code %d\n", rc)
|
|
|
|
client = mqtt_client.Client(client_id)
|
|
client.username_pw_set(username, password)
|
|
client.on_connect = on_connect
|
|
client.connect(broker, port)
|
|
return client
|
|
|
|
|
|
def publish(client):
|
|
result = client.publish("ledLight/state", "ON")
|
|
status = result[0]
|
|
if status == 0:
|
|
print(f"Send `ON` to topic `{topic}`")
|
|
else:
|
|
print(f"Failed to send message to topic {topic}")
|
|
|
|
|
|
def subscribe(client: mqtt_client):
|
|
def on_message(client, userdata, msg):
|
|
print(f"Received `{msg.payload.decode()}` from `{msg.topic}` topic")
|
|
json_response = json.loads(msg.payload.decode())
|
|
if "color" in json_response:
|
|
hex = convert_to_bulb_format(json_response)
|
|
run_bulb_action(hex)
|
|
else:
|
|
if json_response["state"] == "OFF":
|
|
run_bulb_action("cc2433")
|
|
else:
|
|
run_bulb_action("cc2333")
|
|
|
|
client.subscribe(topic)
|
|
client.on_message = on_message
|
|
|
|
|
|
def run():
|
|
client = connect_mqtt()
|
|
publish(client)
|
|
subscribe(client)
|
|
client.loop_forever()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
run()
|