EMQ X Kuiper est un logiciel léger d'analyse de données IoT en flux, conçu pour fonctionner sur des appareils IoT à ressources limitées.

TensorFlow Lite est un ensemble d'outils permettant d'exécuter des modèles TensorFlow sur des appareils mobiles, embarqués et IoT, offrant des prédictions en machine learning rapides avec une empreinte binaire réduite.

En intégrant Kuiper et TensorFlow Lite, les utilisateurs peuvent analyser des données en flux avec des modèles IA pré-entraînés. Ce tutoriel explique comment créer un plugin Kuiper pour classer des images binaires générées par des appareils périphériques via un modèle de reconnaissance d'images TensorFlow.

Prérequis

Pour exécuter l'interpréteur TensorFlow Lite, un modèle entraîné est nécessaire. Ce tutoriel ne couvre pas l'entraînement de modèles - consultez le convetrisseur TFLite pour plus d'informations. Nous utiliserons le modèle label image du dépôt mattn/go tflite, qui fournit des liaisons Go pour l'API C TFLite. Ce dépôt servira à implémanter notre plugin.

Développement du plugin

Pour intégrer Kuiper et TFLite, nous créons un plugin personnalisé Kuiper. Exemple : la fonction EtiqueterImage prend des données binaires en entrée (image) et retourne une étiquette texte. Si l'image contient un paon, EtiqueterImage(col) renverra "paon".

Pour développer ce plugin :

1. Créer le fichier Go dans plugins/functions/etiquetage/etiquetage.go
2. Implémenter l'interface api.Fonction
3. Exporter la structure

L'implémentation clé est la fonction Exec :

func (f *etiquetage) Exec(args []interface{}, ctx api.FunctionContext) (interface{}, bool) {
    // Initialisation et validation
    
    // Décoder l'image
    img, _, err := image.Decode(bytes.NewReader(args[0]))
    if err != nil {
        return err, false
    }
    
    var erreur error
    f.once.Do(func() {
        // Charger les étiquettes, modèle TFLite et initialiser l'interpréteur
    })

    // Exécuter l'interpréteur sur l'image
    // Retourner l'étiquette la plus probable
    return resultat, true
}

Le plugin est stateless, nous exportons donc une instance unique. Toutes les règles utilisant cette fonction partageront l'instance pour éviter les coûts de création. Les chemins du modèle et des étiquettes sont définis à l'instanciation :

var Etiquetage = etiquetage{
    cheminModele: "etiquetage/mobilenet_quant_v1_224.tflite",
    cheminEtiquettes: "etiquetage/etiquettes.txt",
}

Pour les détails complets, consulter le fichier etiquetage.go.

Construction et installation

Pour utiliser le plugin, il faut le construire dans l'environnement Kuiper puis l'installer.

Installation via archive ZIP prédéfinie

Pour les images Docker Kuiper basées sur Debian avec les balises 1.1.1 ou 1.1.1-slim, utiliser l'archive ZIP prédéfinie :

POST http://{{kuiperHost:kuiperRestPort}}/plugins/functions
Content-Type: application/json

{"name":"etiquetage", "file": "https://www.emqx.io/downloads/kuiper-plugins/v1.1.2/debian/functions/etiquetage_amd64.zip"}

Construction manuelle

1. Si vous n'utilisez pas l'image Docker officielle, vous devez construire manuellement le plugin en trois étapes :
2. Construire l'API C TFLite
3. Construire le plugin etiquetage
4. Empaqueter le plugin avec le script d'installation

Construction de l'API C TFLite

Ce guide s'appuie sur la version TensorFlow v2.2.0-rc3. Pour Ubuntu :

1. Installer Python 3
2. Installer les dépendances : pip3 install -r requirements.txt
3. Installer Bazel
4. Cloner le dépôt TensorFlow et basculer sur la branche v2.2.0-rc3
5. Générer les fichiers .so :

$ cd $tensorflowSrc
$ bazel build --config monolithic -c opt //tensorflow/lite:libtensorflowlite.so
$ bazel build --config monolithic -c opt //tensorflow/lite/c:libtensorflowlite_c.so
$ mkdir lib
$ cp bazel-bin/tensorflow/lite/libtensorflowlite.so lib
$ cp bazel-bin/tensorflow/lite/c/libtensorflowlite_c.so lib

6. Installer les fichiers .so :

$ sudo vi /etc/ld.so.conf.d/tflite.conf
# Ajouter le chemin vers les .so
$ sudo ldconfig
$ ldconfig -p | grep libtensorflow

Construction du plugin etiquetage

Avec le dépôt Kuiper cloné :

$ cd {{kuiperRepoPath}}
$ export CGO_CFLAGS=-I/root/tensorflow
$ export CGO_LDFLAGS=-L/root/tensorflow/lib
$ go build -trimpath --buildmode=plugin -o plugins/functions/Etiquetage.so plugins/functions/etiquetage/*.go

Empaqueter le plugin

Créer une archive ZIP contenant :

• etc/
• etiquettes.txt
• mobilenet_quant_v1_224.tflite
• lib/
• libtensorflowlite.so
• libtensorflowlite_c.so
• install.sh
• Etiquetage.so
• tflite.conf

Utiliser l'API Kuiper pour installer comme décrit ci-dessus.

Exécution du plugin

Après installation, utiliser le plugin dans des règles Kuiper.

Définir le flux

Créer un flux tfdemo avec format binaire :

POST http://{{host}}/streams
Content-Type: application/json

{"sql":"CREATE STREAM tfdemo () WITH (DATASOURCE=\"tfdemo\", FORMAT=\"BINARY\")"}

Définir la règle

Créer une règle ruleTf exécutant l'étiquetage :

POST http://{{host}}/rules
Content-Type: application/json

{
  "id": "ruleTf",
  "sql": "SELECT etiquetage(self) FROM tfdemo",
  "actions": [
    {
      "log": {}
    }
  ]
}

Envoi des données

Programme Go envoyant des images :

package main

import (
	"fmt"
	mqtt "github.com/eclipse/paho.mqtt.golang"
	"io/ioutil"
	"time"
)

func main(){
	const SUJET = "tfdemo"

	images := []string{
		"paon.png",
		"grenouille.jpg",
	}
	opts := mqtt.NewClientOptions().AddBroker("tcp://votreserveur:1883")
	client := mqtt.NewClient(opts)
	if token := client.Connect(); token.Wait() && token.Error() != nil {
		panic(token.Error())
	}
	for _, image := range images {
		fmt.Println("Publication " + image);
		payload, err := ioutil.ReadFile(image)
		if err != nil{
			fmt.Println(err)
			continue
		}
		if token := client.Publish(SUJET, 0, false, payload); token.Wait() && token.Error() != nil {
			fmt.Println(token.Error())
		} else {
			fmt.Println("Publié " + image);
		}
		time.Sleep(1 * time.Second)
	}
	client.Disconnect(0)
}

Vérification des résultats

Les résultats sont enregistrés dans les logs. Avec les images paon.png et grenouille.jpg :

time="2021-02-05 16:23:29" level=info msg="résultat de ruleTf: [{\"etiquetage\":\"paon\"}]" file="sinks/log_sink.go:16" rule=ruleTf
time="2021-02-05 16:23:30" level=info msg="résultat de ruleTf: [{\"etiquetage\":\"grenouille\"}]" file="sinks/log_sink.go:16" rule=ruleTf