Spaces:
Sleeping
Sleeping
File size: 3,606 Bytes
758a4c9 b7afc5f 758a4c9 b7afc5f 758a4c9 b7afc5f 758a4c9 b7afc5f 758a4c9 b7afc5f 758a4c9 b7afc5f 758a4c9 b7afc5f 758a4c9 b7afc5f 758a4c9 b7afc5f 758a4c9 b7afc5f 758a4c9 |
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 |
import json
import queue
import threading
import paho.mqtt.client as mqtt
import streamlit as st
# Initialize Streamlit app
st.title("RGB Command and Sensor Data Panel")
# MQTT Configuration
HIVEMQ_HOST = st.text_input(
"Enter your HiveMQ host:",
"248cc294c37642359297f75b7b023374.s2.eu.hivemq.cloud",
type="password",
)
HIVEMQ_USERNAME = st.text_input("Enter your HiveMQ username:", "sgbaird")
HIVEMQ_PASSWORD = st.text_input(
"Enter your HiveMQ password:", "D.Pq5gYtejYbU#L", type="password"
)
PORT = st.number_input(
"Enter the port number:", min_value=1, max_value=65535, value=8883
)
# User input for the Pico ID
pico_id = st.text_input("Enter your Pico ID:", "test", type="password")
max_power = 0.3
max_value = round(max_power * 255)
# Information about the maximum power reduction
st.info(
f"The maximum power has been reduced. The upper limit for RGB values is {max_value} instead of 255."
)
# Sliders for RGB values
R = st.slider("Select the Red value:", min_value=0, max_value=max_value, value=0)
G = st.slider("Select the Green value:", min_value=0, max_value=max_value, value=0)
B = st.slider("Select the Blue value:", min_value=0, max_value=max_value, value=0)
# Initialize session state for messages and lock state
if "messages" not in st.session_state:
st.session_state.messages = []
if "locked" not in st.session_state:
st.session_state.locked = False
# Queue to hold sensor data
sensor_data_queue = queue.Queue()
# Singleton: https://docs.streamlit.io/develop/api-reference/caching-and-state/st.cache_resource
@st.cache_resource
def get_paho_client(
sensor_data_topic, hostname, username, password=None, port=8883, tls=True
):
client = mqtt.Client(protocol=mqtt.MQTTv5) # create new instance
def on_message(client, userdata, msg):
sensor_data_queue.put(json.loads(msg.payload))
def on_connect(client, userdata, flags, rc, properties=None):
if rc != 0:
print("Connected with result code " + str(rc))
client.subscribe(sensor_data_topic, qos=1)
client.on_connect = on_connect
client.on_message = on_message
if tls:
client.tls_set(tls_version=mqtt.ssl.PROTOCOL_TLS_CLIENT)
client.username_pw_set(username, password)
client.connect(hostname, port)
client.loop_start() # Use a non-blocking loop
return client
def send_and_receive(client, command_topic, msg, queue_timeout=60):
client.publish(command_topic, json.dumps(msg), qos=2)
while True:
sensor_data = sensor_data_queue.get(True, queue_timeout)
return sensor_data
# Publish button
if st.button("Send RGB Command") and not st.session_state.locked:
if not pico_id or not HIVEMQ_HOST or not HIVEMQ_USERNAME or not HIVEMQ_PASSWORD:
st.error("Please enter all required fields.")
else:
st.session_state.locked = True
command_topic = f"{pico_id}/neopixel"
sensor_data_topic = f"{pico_id}/as7341"
client = get_paho_client(
sensor_data_topic,
HIVEMQ_HOST,
HIVEMQ_USERNAME,
password=HIVEMQ_PASSWORD,
port=int(PORT),
tls=True,
)
command_msg = {"R": R, "G": G, "B": B}
sensor_data = send_and_receive(
client, command_topic, command_msg, queue_timeout=30
)
st.session_state.locked = False
st.success("Command sent successfully!")
st.write("Sensor Data Received:", sensor_data)
# Display received messages
for message in st.session_state.messages:
st.write(f"Received message: {message}")
|