scripts

scripts/{basic_rgb_led_sensor.py → basic_send_receive.py}

@@ -1,17 +1,11 @@
-"""
-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
+PICO_ID = "test-fan"  # UPDATE THIS TO YOUR ID
 
-command_topic = f"sdl-demo/picow/{PICO_ID}/GPIO/28/"
-sensor_data_topic = f"sdl-demo/picow/{PICO_ID}/as7341/"
+command_topic = f"fan-control/picow/{PICO_ID}/speed"
+sensor_data_topic = f"fan-control/picow/{PICO_ID}/rpm"
 
 HIVEMQ_USERNAME = "sgbaird"
 HIVEMQ_PASSWORD = "D.Pq5gYtejYbU#L"
@@ -20,27 +14,31 @@ 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 create_paho_client(tls=True):
+    client = mqtt.Client(protocol=mqtt.MQTTv5)
+    if tls:
+        client.tls_set(tls_version=mqtt.ssl.PROTOCOL_TLS_CLIENT)
+    return client
 
+
+# Define setup separately since sensor_data is dynamic
+def setup_paho_client(
+    client, rpm_data_topic, hostname, username, password=None, port=8883
+):
     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.subscribe(rpm_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)
+    client.loop_start()  # Use a non-blocking loop
 
     return client
 
@@ -56,12 +54,13 @@ def send_and_receive(client, command_topic, msg, queue_timeout=60):
         return sensor_data
 
 
-client = get_paho_client(
-    sensor_data_topic, HIVEMQ_HOST, HIVEMQ_USERNAME, password=HIVEMQ_PASSWORD
+client = create_paho_client(tls=True)
+
+client = setup_paho_client(
+    client, sensor_data_topic, HIVEMQ_HOST, HIVEMQ_USERNAME, password=HIVEMQ_PASSWORD
 )
 
-command_msg = {"R": 10, "G": 10, "B": 10}
+# Sending speed command
+command_msg = {"speed": 50}  # Update the speed value as needed (0-100)
 sensor_data = send_and_receive(client, command_topic, command_msg, queue_timeout=30)
 print(sensor_data)
-
-1 + 1

scripts/onboard_led_temp.py

@@ -1,73 +0,0 @@
-"""
-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)