Python MQTT Setup
To keep the code clean, we have created a seperate python file (mqtt.py) that will handle all the MQTT communication. This file will be imported into the main python file (main.py) to send the data to Grasshopper.
warning
This tutorial is a continuation of Hand Gesture Recognition Example
Install Paho-MQTT
We will first need to install the MQTT library for Python. We will be using the paho-mqtt library. To install it, run the following command:
pip install paho-mqtt
The mqtt.py file
Create a new file called mqtt.py in the same folder as main.py. This file will handle all the MQTT communication.
1. Import the necessary libraries
import paho.mqtt.client as paho
2. Setup Callbacks
MQTT Callbacks definitions are required to handle the connection, disconnection, and message reception. They automatically get called when the corresponding event occurs.
# 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")
3. Connection Definition
The following will setup the connection to the MQTT broker. We will be using a free broker by HiveMQ.
The will_set function is used to set the Last Will and Testament (LWT) message. This message will be sent to the broker if the client disconnects unexpectedly.
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) and must be before connect
client.will_set(topic, payload=previousCheck, qos=1, retain=True)
## Connect to HiveMQ Cloud on port 1883 (default for MQTT over TLS)
client.connect("mqtt-dashboard.com",1883)
## Send an Intial Value to the Broker)
client.publish(topic, payload=previousCheck, qos=1, retain=True)
print("Data published to MQTT broker - " + str(previousCheck))
4. Publish Definition
This definition will publish your value to the set topic. qos is "Quality of Service"
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}")
Implementing into the main.py file
1. Import the mqtt.py file
Now that we have the mqtt.py file setup, we can import it into the main.py.
import sys
import time
# ... the rest of the libraries
import mqtt
from mqtt import *
2. Setup the MQTT parameters
# Recognition results
recognition_frame = None
recognition_result_list = []
#MQTT
TOPIC = "IAACWorkshop/Gestures/Test"
client = paho.Client(client_id="",protocol=paho.MQTTv311)
previous_gesture = "None"
2. Add the publish function to the main.py
In the def display_results(current_frame): we are going to replace the printing of the gesture with sending to MQTT
#print(f'Gesture: {category_name} ({score})') < replaced with MQTT Publish
#MQTT Publish
if(previous_gesture != category_name and score > 0.55):
mqtt.publish(category_name + "Lasath",client,TOPIC,0)
previous_gesture = category_name
3. Setup the MQTT connection
Next, we will add the code to setup the MQTT connection. This will be added to the run() definition.
def run():
global client,previous_gesture
mqtt.setup(client,TOPIC,previous_gesture)
# Initialize the camera
#...
4. Cleean disconnection
Lastly, as best practice, we will add a clean disconnection to the MQTT broker. This will be added to the end of run(): definition.
finally:
print('Releasing the camera and closing the windows')
recognizer.close()
cap.release()
cv2.destroyAllWindows()
client.disconnect() #MQTT disconnect
The full code
warning
Make sure to change the MODEL_PATH to the correct path of the gesture recognition model file.
import sys
import time
import cv2
import mediapipe as mp
from mediapipe.framework.formats import landmark_pb2
from mediapipe.tasks import python
from mediapipe.tasks.python import vision
import mqtt
from mqtt import *
# Initialize MediaPipe solutions
mp_hands = mp.solutions.hands
mp_drawing = mp.solutions.drawing_utils
mp_drawing_styles = mp.solutions.drawing_styles
# Default parameters
MODEL_PATH = r'02 Code\01 Python\02 Camera\01 Gestures\gesture_recognizer.task'
NUM_HANDS = 1
MIN_HAND_DETECTION_CONFIDENCE = 0.5
MIN_HAND_PRESENCE_CONFIDENCE = 0.5
MIN_TRACKING_CONFIDENCE = 0.5
CAMERA_ID = 0
FRAME_WIDTH = 640
FRAME_HEIGHT = 480
# Global variables for FPS calculation
COUNTER, FPS = 0, 0
START_TIME = time.time()
# Visualization parameters
ROW_SIZE = 50 # pixels
LEFT_MARGIN = 24 # pixels
TEXT_COLOUR = (0, 0, 0) # black
FONT_SIZE = 1
FONT_THICKNESS = 1
FPS_AVG_FRAME_COUNT = 10
# Label box parameters
LABEL_TEXT_COLOUR = (255, 255, 255) # white
LABEL_FONT_SIZE = 1
LABEL_THICKNESS = 2
# Recognition results
recognition_frame = None
recognition_result_list = []
#MQTT
TOPIC = "IAACWorkshop/Gestures/Test"
client = paho.Client(client_id="",protocol=paho.MQTTv311)
previous_gesture = "None"
def save_result(result: vision.GestureRecognizerResult, unused_output_image: mp.Image, timestamp_ms: int):
global FPS, COUNTER, START_TIME, recognition_result_list
if COUNTER % 10 == 0:
FPS = 10 / (time.time() - START_TIME)
START_TIME = time.time()
recognition_result_list.append(result)
COUNTER += 1
def load_model(model_path:str):
"""
Loads the gesture recognition model from the specified file path.
Parameters:
model_path (str): The path to the gesture recognition model file.
Returns:
vision.GestureRecognizer: An instance of the gesture recognizer.
"""
with open(MODEL_PATH, 'rb') as f:
model = f.read()
base_options = python.BaseOptions(model_asset_buffer=model)
options = vision.GestureRecognizerOptions(
base_options=base_options,
running_mode=vision.RunningMode.LIVE_STREAM,
num_hands=NUM_HANDS,
min_hand_detection_confidence=MIN_HAND_DETECTION_CONFIDENCE,
min_hand_presence_confidence=MIN_HAND_PRESENCE_CONFIDENCE,
min_tracking_confidence=MIN_TRACKING_CONFIDENCE,
result_callback=save_result)
return vision.GestureRecognizer.create_from_options(options)
def display_results(current_frame):
global previous_gesture
if recognition_result_list:
#_______________________________
for hand_index, hand_landmarks in enumerate(
recognition_result_list[0].hand_landmarks):
# Calculate the bounding box of the hand
x_min = min([landmark.x for landmark in hand_landmarks])
y_min = min([landmark.y for landmark in hand_landmarks])
y_max = max([landmark.y for landmark in hand_landmarks])
# Convert normalized coordinates to pixel values
frame_height, frame_width = current_frame.shape[:2]
x_min_px = int(x_min * frame_width)
y_min_px = int(y_min * frame_height)
y_max_px = int(y_max * frame_height)
# Get gesture classification results
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})'
#print(f'Gesture: {category_name} ({score})')
#MQTT Publish
if(previous_gesture != category_name and score > 0.55):
mqtt.publish(category_name + "Lasath",client,TOPIC,0)
previous_gesture = category_name
# Compute text size
text_size = \
cv2.getTextSize(result_text, cv2.FONT_HERSHEY_DUPLEX, LABEL_FONT_SIZE,
LABEL_THICKNESS)[0]
text_width, text_height = text_size
# Calculate text position (above the hand)
text_x = x_min_px
text_y = y_min_px - 10 # Adjust this value as needed
# Make sure the text is within the frame boundaries
if text_y < 0:
text_y = y_max_px + text_height
# Draw the text
cv2.putText(current_frame, result_text, (text_x, text_y),
cv2.FONT_HERSHEY_DUPLEX, LABEL_FONT_SIZE,
LABEL_TEXT_COLOUR, LABEL_THICKNESS, cv2.LINE_AA)
# Draw hand landmarks on the frame
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())
recognition_frame = current_frame
recognition_result_list.clear()
if recognition_frame is not None:
cv2.imshow('gesture_recognition', recognition_frame)
def run():
global client,previous_gesture
mqtt.setup(client,TOPIC,previous_gesture)
# Initialize the camera
print('Starting the camera...')
cap = cv2.VideoCapture(CAMERA_ID)
cap.set(cv2.CAP_PROP_FRAME_WIDTH, FRAME_WIDTH)
cap.set(cv2.CAP_PROP_FRAME_HEIGHT, FRAME_HEIGHT)
#Load gesture recognition model
print('Loading the gesture recognition model...')
recognizer = load_model(MODEL_PATH)
# Opening the camera
print('Opening the camera...')
try:
while cap.isOpened():
# Read the image from the camera
success, image = cap.read()
if not success:
sys.exit('ERROR: Unable to read from camera. Please verify your camera settings.')
# Flip the image horizontally for a later selfie-view display
image = cv2.flip(image, 1)
rgb_image = cv2.cvtColor(image, cv2.COLOR_BGR2RGB)
mp_image = mp.Image(image_format=mp.ImageFormat.SRGB, data=rgb_image)
# Process the image
recognizer.recognize_async(mp_image, time.time_ns() // 1_000_000)
# Show the FPS
fps_text = 'FPS = {:.1f}'.format(FPS)
text_location = (LEFT_MARGIN, ROW_SIZE)
current_frame = image
cv2.putText(current_frame, fps_text, text_location, cv2.FONT_HERSHEY_DUPLEX,
FONT_SIZE, TEXT_COLOUR, FONT_THICKNESS, cv2.LINE_AA)
#Display the gesture recognition results
display_results(current_frame)
if cv2.waitKey(1) & 0xFF == 27:
break
finally:
print('Releasing the camera and closing the windows')
recognizer.close()
cap.release()
client.disconnect()
cv2.destroyAllWindows()
if __name__ == '__main__':
run()