chore: Add Streamlit and MQTT implementation for light mixing control panel

app.py

@@ -1,19 +1,12 @@
-"""Based on
-https://github.com/AccelerationConsortium/ac-training-lab/blob/main/src/ac_training_lab/picow/digital-pipette/scripts/streamlit_webapp.py
-
-permalink:
-https://github.com/AccelerationConsortium/ac-training-lab/blob/230c72f3d9d6e8a5d0b9cece044515dfd386acdc/src/ac_training_lab/picow/digital-pipette/scripts/streamlit_webapp.py]
-
-Streamlit was preferred over the gradio implementation
-"""
-
 import json
+import queue
+import threading
 
 import paho.mqtt.client as mqtt
 import streamlit as st
 
 # Initialize Streamlit app
-st.title("Actuator Control Panel")
+st.title("RGB Command and Sensor Data Panel")
 
 # MQTT Configuration
 HIVEMQ_HOST = st.text_input("Enter your HiveMQ host:", "", type="password")
@@ -26,66 +19,80 @@ PORT = st.number_input(
 # User input for the Pico ID
 pico_id = st.text_input("Enter your Pico ID:", "", type="password")
 
-# Slider for position value
-position = st.slider(
-    "Select the position value:", min_value=1.1, max_value=1.9, value=1.5
-)
+# Sliders for RGB values
+R = st.slider("Select the Red value:", min_value=0, max_value=255, value=0)
+G = st.slider("Select the Green value:", min_value=0, max_value=255, value=0)
+B = st.slider("Select the Blue value:", min_value=0, max_value=255, value=0)
 
+# Initialize session state for messages
+if "messages" not in st.session_state:
+    st.session_state.messages = []
 
-# singleton: https://docs.streamlit.io/develop/api-reference/caching-and-state/st.cache_resource
+# 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(hostname, username, password=None, port=8883, tls=True):
+def get_paho_client(
+    sensor_data_topic, hostname, username, password=None, port=8883, tls=True
+):
+
+    client = mqtt.Client(protocol=mqtt.MQTTv5)  # create new instance
 
-    client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, protocol=mqtt.MQTTv5)
+    def on_message(client, userdata, msg):
+        sensor_data_queue.put(json.loads(msg.payload))
 
-    # The callback for when the client receives a CONNACK response from the server.
     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
 
-    # enable TLS for secure connection
     if tls:
-        client.tls_set()
-    # set username and password
+        client.tls_set(tls_version=mqtt.ssl.PROTOCOL_TLS_CLIENT)
     client.username_pw_set(username, password)
-    # connect to HiveMQ Cloud on port 8883 (default for MQTT)
     client.connect(hostname, port)
     client.loop_start()  # Use a non-blocking loop
 
     return client
 
 
-def send_command(client, pico_id, position):
-    # Topic
-    command_topic = f"digital-pipette/picow/{pico_id}/L16-R"
+def send_and_receive(client, command_topic, msg, queue_timeout=60):
+    client.publish(command_topic, json.dumps(msg), qos=2)
 
-    # Create and send command
-    command = {"position": position}
-
-    try:
-        result = client.publish(command_topic, json.dumps(command), qos=1)
-        result.wait_for_publish()  # Ensure the message is sent
-        if result.rc == mqtt.MQTT_ERR_SUCCESS:
-            return f"Command sent: {command} to topic {command_topic}"
-        else:
-            return f"Failed to send command: {result.rc}"
-    except Exception as e:
-        return f"An error occurred: {e}"
+    while True:
+        sensor_data = sensor_data_queue.get(True, queue_timeout)
+        return sensor_data
 
 
 # Publish button
-if st.button("Send Command"):
+if st.button("Send RGB Command"):
     if not pico_id or not HIVEMQ_HOST or not HIVEMQ_USERNAME or not HIVEMQ_PASSWORD:
         st.error("Please enter all required fields.")
     else:
+        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,
         )
-        success_msg = send_command(client, pico_id, position)
-        st.success(success_msg)
+
+        command_msg = {"R": R, "G": G, "B": B}
+        sensor_data = send_and_receive(
+            client, command_topic, command_msg, queue_timeout=30
+        )
+
+        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}")

requirements.txt

@@ -1 +1,2 @@
-paho-mqtt
+paho-mqtt
+streamlit

scripts/digital_pipette_picow_basic.py

@@ -1,28 +0,0 @@
-import utime
-from machine import PWM, Pin
-
-# Setup PWM
-pwm = PWM(Pin(0))  # Use the appropriate GPIO pin
-pwm.freq(50)  # 50 Hz frequency
-
-
-def set_position(pulse_ms):
-    duty = int((pulse_ms / 20.0) * 65535)
-    pwm.duty_u16(duty)
-
-
-# Example to set the actuator to different positions
-set_position(1.1)  # Almost full retraction
-utime.sleep(5)
-set_position(1.5)  # Halfway
-utime.sleep(5)
-set_position(1.9)  # Almost full extension
-utime.sleep(5)
-set_position(1.1)  # Almost full retraction
-utime.sleep(5)
-set_position(1.5)  # Halfway
-utime.sleep(5)
-
-# Add your logic to set it to the desired intermediate positions
-
-pwm.deinit()  # Deinitialize PWM

scripts/onboard_led_temp.py

@@ -0,0 +1,73 @@
+"""
+https://ac-microcourses.readthedocs.io/en/latest/courses/hello-world/1.4.1-onboard-led-temp.html
+
+permalink: https://github.com/AccelerationConsortium/ac-microcourses/blob/07dc9a0286ded2e21ea64f02d8ae697717e786b9/docs/courses/hello-world/1.4.1-onboard-led-temp.ipynb
+"""
+
+COURSE_ID = "<your_id_here>"  # UPDATE THIS TO YOUR ID
+
+command_topic = f"{COURSE_ID}/onboard_led"
+sensor_data_topic = f"{COURSE_ID}/onboard_temp"
+
+HIVEMQ_USERNAME = "sgbaird"
+HIVEMQ_PASSWORD = "D.Pq5gYtejYbU#L"
+HIVEMQ_HOST = "248cc294c37642359297f75b7b023374.s2.eu.hivemq.cloud"
+
+import paho.mqtt.client as mqtt
+import json
+from queue import Queue
+
+sensor_data_queue: "Queue[dict]" = Queue()
+
+
+def get_paho_client(
+    sensor_data_topic, hostname, username, password=None, port=8883, tls=True
+):
+
+    client = mqtt.Client(
+        mqtt.CallbackAPIVersion.VERSION2, protocol=mqtt.MQTTv5
+    )  # create new instance
+
+    def on_message(client, userdata, msg):
+        sensor_data_queue.put(json.loads(msg.payload))
+
+    # The callback for when the client receives a CONNACK response from the server.
+    def on_connect(client, userdata, flags, rc, properties=None):
+        if rc != 0:
+            print("Connected with result code " + str(rc))
+        # Subscribing in on_connect() means that if we lose the connection and
+        # reconnect then subscriptions will be renewed.
+        client.subscribe(sensor_data_topic, qos=1)
+
+    client.on_connect = on_connect
+    client.on_message = on_message
+
+    # enable TLS for secure connection
+    if tls:
+        client.tls_set(tls_version=mqtt.ssl.PROTOCOL_TLS_CLIENT)
+    # set username and password
+    client.username_pw_set(username, password)
+    # connect to HiveMQ Cloud on port 8883 (default for MQTT)
+    client.connect(hostname, port)
+    client.subscribe(sensor_data_topic, qos=2)
+
+    return client
+
+
+def send_and_receive(client, command_topic, msg, queue_timeout=60):
+    client.publish(command_topic, msg, qos=2)
+
+    client.loop_start()
+
+    while True:
+        sensor_data = sensor_data_queue.get(True, queue_timeout)
+        client.loop_stop()
+        return sensor_data
+
+
+client = get_paho_client(
+    sensor_data_topic, HIVEMQ_HOST, HIVEMQ_USERNAME, password=HIVEMQ_PASSWORD
+)
+
+onboard_temp = send_and_receive(client, command_topic, "toggle", queue_timeout=30)
+print(onboard_temp)

scripts/rgb_led_sensor.py

@@ -0,0 +1,65 @@
+"""
+https://ac-microcourses.readthedocs.io/en/latest/courses/hello-world/1.4.1-onboard-led-temp.html
+
+permalink: https://github.com/AccelerationConsortium/ac-microcourses/blob/07dc9a0286ded2e21ea64f02d8ae697717e786b9/docs/courses/hello-world/1.4.1-onboard-led-temp.ipynb
+"""
+
+import paho.mqtt.client as mqtt
+import json
+from queue import Queue
+
+PICO_ID = "test"  # UPDATE THIS TO YOUR ID
+
+command_topic = f"{PICO_ID}/neopixel"
+sensor_data_topic = f"{PICO_ID}/as7341"
+
+HIVEMQ_USERNAME = "sgbaird"
+HIVEMQ_PASSWORD = "D.Pq5gYtejYbU#L"
+HIVEMQ_HOST = "248cc294c37642359297f75b7b023374.s2.eu.hivemq.cloud"
+
+sensor_data_queue: "Queue[dict]" = Queue()
+
+
+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.subscribe(sensor_data_topic, qos=2)
+
+    return client
+
+
+def send_and_receive(client, command_topic, msg, queue_timeout=60):
+    client.publish(command_topic, json.dumps(msg), qos=2)
+
+    client.loop_start()
+
+    while True:
+        sensor_data = sensor_data_queue.get(True, queue_timeout)
+        client.loop_stop()
+        return sensor_data
+
+
+client = get_paho_client(
+    sensor_data_topic, HIVEMQ_HOST, HIVEMQ_USERNAME, password=HIVEMQ_PASSWORD
+)
+
+command_msg = {"R": 50, "G": 20, "B": 240}
+sensor_data = send_and_receive(client, command_topic, command_msg, queue_timeout=30)
+print(sensor_data)

scripts/single_submit.py

@@ -1,39 +0,0 @@
-"""This streamlit implementation is preferred over the gradio implementation"""
-
-import json
-
-import paho.mqtt.client as mqtt
-
-# MQTT Configuration
-HIVEMQ_HOST = ""
-HIVEMQ_USERNAME = ""
-HIVEMQ_PASSWORD = ""
-PORT = 8883
-
-# User input for the Pico ID
-pico_id = ""
-
-# Slider for position value
-position = 1.35
-
-
-def send_command(client, pico_id, position):
-    # Topic
-    command_topic = f"digital-pipette/picow/{pico_id}/L16-R"
-
-    # Create and send command
-    command = {"position": position}
-    client.publish(command_topic, json.dumps(command), qos=1)
-
-    return f"Command sent: {command} to topic {command_topic}"
-
-
-# Initialize MQTT client
-client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, protocol=mqtt.MQTTv5)
-client.tls_set()
-client.username_pw_set(HIVEMQ_USERNAME, HIVEMQ_PASSWORD)
-client.connect(HIVEMQ_HOST, PORT, 60)
-
-success_msg = send_command(client, pico_id, position)
-
-client.disconnect()