Amazing Open Source Raspberry Pi Pico 2 Synth, Cutting Edge Power over Skin Research, and much more!

0:00:00 / -

Watch now on YouTube Watch now

Remote Gesture Controller With Mqtt

Made by The Tactigon / Artificial intelligence / Robotics / Wearables / IoT

About the project

Wouldn’t be nice to control a robotic arm on the other side of the world and make it follow your gestures in the most natural manner?

Project info

Difficulty: Easy

Platforms: The Tactigon, MQTT

Estimated time: 1 hour

License: MIT license (MIT)

Items used in this project

Hardware components

Tactigon SKIN SDK

x 1

Software apps and online services

MQTT

Story

Wouldn’t be nice to control a robotic arm on the other side of the world and make it follow your gestures in the most natural manner?

With our Tactigon Skin and recently developed SDK this can be achieved with very few lines of code. In this article we will show how to use the T-Skin to recognize gestures in conjunction with our AI-based SDK and the MQTT protocol allowing to send messages wherever we want. The python code illustrated below is importing modules from our SDK which are in charge of BLE data collection from one or two T-SKIN (Tactigon_BLE) and gesture recognition (Tactigon_Gesture).

Thanks to those modules, recognized gestures are pushed in a pipe which can be pulled by the application process (Application_MQTT) and published on an MQTT server. Gestures to be recognized,

MQTT server details and topic names can all be customized by using our SDK configuration files.

T-Gear SDK is use for collecting new raw data, send the data to server, ask server to train a model using the raw data, and download the model from server and finally use the model for testing real-time gesture recognition.

The T-Gear SDK includes several end applications that can be customized using config files or used as reference for development.

Code

Application_Gesture_to_MQTT.py

import multiprocessing
import logging
import time
import paho.mqtt.client as mqtt

import sys
from os import path

# resource directory - need to be referenced to exe path when used with exe bundle
if getattr(sys, "frozen", False):
    CLIENT_PY_PATH = sys._MEIPASS
else:
    CLIENT_PY_PATH = path.join(path.dirname(__file__), "../../")
    sys.path.insert(0, path.join(CLIENT_PY_PATH, "utilities"))
    sys.path.insert(0, path.join(CLIENT_PY_PATH, "hal"))
    sys.path.insert(0, path.join(CLIENT_PY_PATH, "middleware"))

from Config_Manager import Config_Manager
from Tactigon_BLE import Tactigon_BLE
from Tactigon_Gesture import Tactigon_Gesture


class Application_MQTT(multiprocessing.Process):
    def __init__(self, gesture_r_pipe=False, gesture_l_pipe=False):

        super(Application_MQTT, self).__init__(
            target=self.loop_iterator, args=(gesture_r_pipe, gesture_l_pipe)
        )

    def loop_iterator(self, gesture_r_pipe, gesture_l_pipe):

        self.gesture_r_pipe = gesture_r_pipe
        self.gesture_l_pipe = gesture_l_pipe
        self.app_config = Config_Manager.from_file(
            "apps/application_gesture_to_mqtt.json"
        )

        self.client = mqtt.Client()
        self.client.connect(
            self.app_config.get("MQTT_URL"), self.app_config.get("MQTT_PORT"), 60
        )
        self.client.loop_start()

        print("Application process started, type any key to close the application")

        while True:
            self.loop()

    def loop(self):
        """Gesture recognition loop routine"""

        if (self.gesture_r_pipe != False) and (self.gesture_r_pipe.poll()):
            r_gesture, _ = self.gesture_r_pipe.recv()
            print("Publishing RIGHT topic")
            self.client.publish(self.app_config.get("MQTT_RIGHT_TOPIC"), r_gesture)

        if (self.gesture_l_pipe != False) and (self.gesture_l_pipe.poll()):
            l_gesture, _ = self.gesture_l_pipe.recv()
            print("Publishing LEFT topic")
            self.client.publish(self.app_config.get("MQTT_LEFT_TOPIC"), l_gesture)


def run(debug=False):

    if debug:
        # logging
        multiprocessing.log_to_stderr()
        logger = multiprocessing.get_logger()
        logger.setLevel(logging.DEBUG)

    # import json config
    hal_config = Config_Manager.from_file("hal.json")

    R_add = hal_config.get("BLE_RIGHT_ADDRESS")
    L_add = hal_config.get("BLE_LEFT_ADDRESS")
    R_en = hal_config.get("BLE_RIGHT_ENABLE")
    L_en = hal_config.get("BLE_LEFT_ENABLE")
    num_sample = hal_config.get("NUM_SAMPLE")

    # data pipe variables
    if R_en == "True":
        rx_sensor_r_pipe, tx_sensor_r_pipe = multiprocessing.Pipe(duplex=False)
        rx_gesture_r_pipe, tx_gesture_r_pipe = multiprocessing.Pipe(duplex=False)
    else:
        rx_sensor_r_pipe = False
        rx_gesture_r_pipe = False

    if L_en == "True":
        rx_sensor_l_pipe, tx_sensor_l_pipe = multiprocessing.Pipe(duplex=False)
        rx_gesture_l_pipe, tx_gesture_l_pipe = multiprocessing.Pipe(duplex=False)
    else:
        rx_sensor_l_pipe = False
        rx_gesture_l_pipe = False

    # create  processes
    if R_en == "True":
        pro_in_r = Tactigon_BLE("RIGHT", R_add, tx_sensor_r_pipe, debug=debug)
        pro_g_r = Tactigon_Gesture(
            "RIGHT",
            num_sample,
            rx_sensor_r_pipe,
            tx_gesture_r_pipe,
            confidence_th=5,
            debug=debug,
        )

    if L_en == "True":
        pro_in_l = Tactigon_BLE("LEFT", L_add, tx_sensor_l_pipe, debug=debug)
        pro_g_l = Tactigon_Gesture(
            "LEFT",
            num_sample,
            rx_sensor_l_pipe,
            tx_gesture_l_pipe,
            confidence_th=5,
            debug=debug,
        )

    pro_app = Application_MQTT(rx_gesture_r_pipe, rx_gesture_l_pipe)

    pro_app.start()

    if R_en == "True":
        pro_g_r.start()
        pro_in_r.start()

    if L_en == "True":
        pro_g_l.start()
        pro_in_l.start()

    input()

    if R_en == "True":
        pro_in_r.terminate()
        pro_g_r.terminate()

    if L_en == "True":
        pro_in_l.terminate()
        pro_g_l.terminate()

    pro_app.terminate()


# Start point of the application
if __name__ == "__main__":

    run(False)

Credits

The Tactigon

The Tactigon is a brand of Next Industries Milano. Next Industries is a born of a passion to develop devices and sensors fit for Iot Technology. Our team is made of incredible experts in movement detection for big structural monitoring systems, software development and wearable device.

Shop

Project Hub

Video

Blog

Shop home

3D printing

Single Board Computers

Development kits

Sensors

Internet of things

Robotics

Accessories

Product of the Week

Electromaker Educator

The Electromaker Show

The Electromaker Podcast

Remote Gesture Controller With Mqtt

About the project

Project info