Workshop IAAC Documentation

import paho.mqtt.client as paho

# MQTT Callbacks definitions
def on_connect(client, userdata, flags, rc, properties=None):
    print("CONNACK received with code %s." % rc)

def on_publish(client, userdata, mid, properties=None):
    print("mid: " + str(mid))

def on_subscribe(client, userdata, mid, granted_qos, properties=None):
    print("Subscribed: " + str(mid) + " " + str(granted_qos))

def on_message(client, userdata, msg):
    print(msg.topic + " " + str(msg.qos) + " " + str(msg.payload))

def on_disconnect(client, userdata, rc, properties=None):
    print("MQTT Disconnected")

def setup(client, topic, previousCheck):
    # MQTT Setup
    client.on_connect = on_connect
    client.on_publish = on_publish
    client.on_subscribe = on_subscribe
    client.on_message = on_message
    client.on_disconnect = on_disconnect

    # Last Will and Testament (LWT) must be set before connect
    client.will_set(topic, payload=previousCheck, qos=1, retain=True) 
    
    # Connect to HiveMQ Cloud on port 1883
    client.connect("mqtt-dashboard.com", 1883)
    
    # Send an Initial Value to the Broker
    client.publish(topic, payload=previousCheck, qos=1, retain=True)
    print("Data published to MQTT broker - " + str(previousCheck))

def publish(value, client, topic, qos=1):
    try:
        if not client.is_connected():
            client.reconnect()
        client.publish(topic, payload=value, qos=qos, retain=True)
        print("Data published to MQTT broker - " + str(value))
    except Exception as e:
        print(f"Failed to publish message: {e}")

pip install paho-mqtt

from flask import Flask, Response, jsonify, render_template
import sys

# Add path to custom modules
sys.path.append('02 Code/01 Python/02 Camera/01 Gestures')
from hand_track import HandTrack
from robot_behaviour import *

app = Flask(__name__)
my_hand_track = HandTrack()

if __name__ == '__main__':
    # host='0.0.0.0' makes the server accessible from any network interface
    app.run(host='0.0.0.0', port=5000)

<!DOCTYPE html>
<html lang="en">
<head>
    <link rel="stylesheet" href="https://maxcdn.bootstrapcdn.com/bootstrap/3.4.1/css/bootstrap.min.css">
    <title>Title of the Document</title>
    <style>This is the style setting</style>
</head>

<body>
    <div class="container">
	<h1>This is the Heading</h1>
	<p>This is the paragraph</p>
    </div>
    <script>This is the script setting</script>
</body>
</html>

pip install -r requirements.txt

# Convert normalized coordinates to pixels
frame_height, frame_width = current_frame.shape[:2]
x_min_px = int(x_min * frame_width)
y_min_px = int(y_min * frame_height)

# Draw Gesture Label
if recognition_result_list[0].gestures:
    gesture = recognition_result_list[0].gestures[hand_index]
    category_name = gesture[0].category_name
    score = round(gesture[0].score, 2)
    result_text = f'{category_name} ({score})'
    cv2.putText(current_frame, result_text, (x_min_px, y_min_px - 10), 
                cv2.FONT_HERSHEY_DUPLEX, 1, (255, 255, 255), 2, cv2.LINE_AA)

# Draw Landmarks
hand_landmarks_proto = landmark_pb2.NormalizedLandmarkList()
hand_landmarks_proto.landmark.extend([
    landmark_pb2.NormalizedLandmark(x=landmark.x, y=landmark.y, z=landmark.z) 
    for landmark in hand_landmarks
])
mp_drawing.draw_landmarks(
    current_frame, 
    hand_landmarks_proto, 
    mp_hands.HAND_CONNECTIONS, 
    mp_drawing_styles.get_default_hand_landmarks_style(),
    mp_drawing_styles.get_default_hand_connections_style()
)

import paho.mqtt.client as paho
import mqtt
from mqtt import *

# MQTT Parameters
TOPIC = "IAACWorkshop/Gestures/Test"
client = paho.Client(client_id="", protocol=paho.MQTTv311)
previous_gesture = "None"

def run():
    global client, previous_gesture
    # Initialize MQTT connection
    mqtt.setup(client, TOPIC, previous_gesture)
    
    try:
        # ... application logic ...
        # Example: Publish when a gesture is recognized
        if(previous_gesture != category_name and score > 0.55):
            mqtt.publish(category_name + "Lasath", client, TOPIC, 0)
            previous_gesture = category_name
    finally:
        # Clean disconnection
        client.disconnect()