SE5 ECEAI/eceai 2023/2024/NaudotRiffaut
L’objectif de ce projet est de réaliser un système d'objet connectés pour une application d'Intelligence Artificielle. Notre application permettra de détecter la direction d'un mouvement passant devant un capteur de distance,: gauche, droite, haut et bas. L’aboutissement de cette application permettrait de réaliser un système de suivi de mouvement. Le système d'objet connecté comportera une VM (Machine Virtuelle) héberger sur le serveur chassiron qui sera connecté à une raspbery PI en wifi. Un capteur de distance sera connecté à notre raspbery PI pour réaliser notre application.
Le binôme
Le binôme est constitué de 2 élèves de SE5-SC :
- Mathis RIFFAUT
- François NAUDOT
Compte rendu des séances
Séance du 04/12/2023
Une machine virtuelle a été crée sur le serveur chassiron avec la commande :
xen-create-image --hostname fnrm --force --dist bookworm --size 10G --memory 1G --dir /usr/local/xen --password glopglop --dhcp --bridge bridgeStudents
La machine virtuelle possède les propriétés suivantes:
Hostname : fnrm
Distribution : bookworm
Stockage : 10G
Mémoire vive : 1G
Mot de passe : glopglop
Les fichier /etc/network/interfaces, /etc/resolv.conf et /etc/apt/sources.list on étés configurés comme demandé dans le sujet.
La VM est connecté à internet par le BridgeStudents.
On vérifie la connexion à internet à l'aide d'un ping www.google.com.
Le STM32F401RE et le capteur de distance (Nucleo-53L5A1):
Nous avons tester le fonctionnement du capteur avec minicom et nous avons observé que chaque lignes est constitué de 64 valeurs séparé par un espace et une ",". Cela convenant pour le logiciel nanoedge nous n'avons pas flaché de nouveau code de le stm32. Si nous avions du le faire, nous aurions utilisé l'ide stm32cube pour modifier la fonction printresult() et changer l'affichage des données afin quelles soient exploitable par l'application nanoedge.
Séance du 18/12/2023
Communication/Protocole :
Pour ce projet nous utilisons un serveur et un client http. Cela nous permet d'afficher par la suite sur une page web les résultats du capteur après analyse de ces données.
Les données transmises entre la Raspberry et la VM xen le seront sous un format JSON qui est pratique car universelle et facile à utiliser avec le protocole HTTP.
Bien que ce protocole et ce format ne soit pas idéal en terme de performance, ces sont des formats pratique, souvent utilisé et facilement utilisable.
Partie Client (Raspberry):
Nous avons résolu le problème de la Raspberry qui ne répondait pas. Cela était dû à l'alimentation : brancher la carte en USBC résoud le problème. Cependant, nous recevons une erreur l'erreur EXT4-fs error (device mmcblk0p2). Le problème venait des arguments saisis lors du lancement de Minicom.
Mot de passe : pasglop
Login : pifou
La communication entre la Raspberry et la machine virtuelle a été vérifiée avec une commande de ping.
La Raspberry Pi 4 a été configurée et connectée au WiFi WIFI_IE_1.
On se connecte a la Raspberry Pi 4 avec une liaison série et on l'alimente en la branchant avec l'alimentation officielle qui se branche sur secteur.
On vérifie la connexion avec un ping www.google.com
Avec la commande suivante, il est à présent possible de se connecter sur la Raspberry : minicom -D /dev/ttyUSB0 -b 115200
On installe un environnement web Python de manière à définir un client simple et de tester la connexion.
On a également défini un programme Python3 qui envoie des requêtes au serveur :
import requests
import json
import sys
def send_post_request(url, data):
try:
response = requests.post(url, data=json.dumps(data), headers={'Content-T
ype': 'application/json'})
print("Réponse du serveur :", response.text)
except requests.exceptions.RequestException as e:
print("Erreur lors de l'envoi de la requête POST :", e)
def main():
if len(sys.argv) < 2:
print("Usage: python script.py '{\"key\": \"value\"}'")
return
try:
data = json.loads(sys.argv[1])
except json.JSONDecodeError as e:
print("Erreur de décodage JSON :", e)
return
# L'adresse du serveur (remplacez avec l'adresse IPv6 de votre serveur et le
port)
url = 'http://[2001:660:4401:6050:216:3eff:fe4c:2673]:8888'
send_post_request(url, data)
if __name__ == "__main__":
main()
Code Python3 côté client
J'améliore ce code de manière à ce qu'il permette d'être relié au capteur et transmette au serveur ce que capte le capteur par un json :
import requests
import json
import sys
import serial
import time
def read_from_sensor(ser):
# Lire les données du capteur
if ser.in_waiting > 0:
line = ser.readline().decode('utf-8').rstrip()
return line
return None
def send_post_request(url, data):
try:
response = requests.post(url, data=json.dumps(data), headers={'Content-T
ype': 'application/json'})
print("Réponse du serveur :", response.text)
except requests.exceptions.RequestException as e:
print("Erreur lors de l'envoi de la requête POST :", e)
def main():
# Configuration du port série (remplacez '/dev/ttyUSB0' par votre port série
)
ser = serial.Serial('/dev/ttyUSB0', 9600, timeout=1)
ser.flush()
while True:
data = read_from_sensor(ser)
if data:
try:
json_data = json.loads(data)
# L'adresse du serveur (remplacez avec l'adresse IPv6 de votre s
erveur et le port)
url = 'http://[2001:660:4401:6050:216:3eff:fe4c:2673]:8888'
send_post_request(url, json_data)
except json.JSONDecodeError as e:
print("Erreur de décodage JSON :", e)
time.sleep(1) # Délai entre les lectures
if __name__ == "__main__":
main()
Code Python3 côté client permettant la liaison USB avec le capteur
Partie Serveur (VM Debian):
On installe Python 3 sur la VM, on installe ufw comme Pare feu, on lui autorise donc les connexions HTTP
On configure un fichier app.py contenant un code de réception simple en Python capable de récupérer des fichiers JSON
On autorise les ipv6 sur le port 8888 de l'adresse : "2001:660:4401:6050:216:3eff:fe4c:2673" dans un programme Python app.py qui lance également une page html index.html qui affiche les requetes.
On a testé le fonctionnement de ce serveur en lui ajoutant les bibliothèques JSON, HTTP et socket de manière à faire apparaître sur une page html, des requetes POST envoyés.
On définit sur ce serveur des méthodes do_GET() et do_POST() de manière les prendres en compte, on définit également des fonctions pour configurer l'ipv6 et lancer la page Web :
# -*- coding: utf-8 -*-
from http.server import HTTPServer, BaseHTTPRequestHandler
import json
import socket
# Stocker les requêtes POST reçues
post_requests = []
class CustomHTTPHandler(BaseHTTPRequestHandler):
def do_GET(self):
if self.path == '/':
self.path = '/index.html'
if self.path == '/data':
self.send_response(200)
self.send_header('Content-type', 'application/json')
self.end_headers()
self.wfile.write(bytes(json.dumps(post_requests), 'utf-8'))
return
if self.path == '/index.html':
self.send_response(200)
self.send_header('Content-type', 'text/html')
self.end_headers()
# Créer le contenu HTML avec les données POST
content = '<html><body><h1>Voici la liste des donnees POST:</h1><ul>'
for item in post_requests:
content += f'<li>{item}</li>'
content += '</ul></body></html>'
self.wfile.write(content.encode('utf-8'))
return
try:
file_to_open = open(self.path[1:]).read()
self.send_response(200)
self.send_header('Content-type', 'text/html')
self.end_headers()
self.wfile.write(bytes(file_to_open, 'utf-8'))
except FileNotFoundError:
self.send_error(404, 'File Not Found: %s' % self.path)
def do_POST(self):
content_length = int(self.headers['Content-Length'])
post_data = self.rfile.read(content_length).decode('utf-8')
# Ajouter les données POST à la liste
post_requests.append(post_data)
# Envoyer une réponse HTTP
self.send_response(200)
self.end_headers()
self.wfile.write(b"POST request processed")
class HTTPServerV6(HTTPServer):
address_family = socket.AF_INET6
if __name__ == '__main__':
server_address = ('::', 8888) # Écoute sur toutes les interfaces IPv6, port 8888
httpd = HTTPServerV6(server_address, CustomHTTPHandler)
print("Serveur actif sur le port", 8888)
httpd.serve_forever()
Code Python 3 du serveur
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<title>Page d'Accueil</title>
</head>
<body>
<h1>Bienvenue sur Mon Serveur</h1>
<h2>Données POST reçues:</h2>
<ul id="data">
<!-- Les données POST seront listées ici -->
</ul>
</body>
</html>
Code html de la page web
Entraînement Capteur STMicroelectronics :
Avec le logiciel Nanoedgeaistudio, on commence à entraîner notre IA en lui faisant 4 premiers signaux, dans lesquels on lui montre des mouvements allant :
-Vers la Gauche
-Vers la Droite
-Vers le Haut
-Vers le Bas
Nous avons fait un premier Benchmark comprenant 2 signaux différents :
-Vers la gauche
-Vers la droite
Avec une centaine de points par signaux.
On commence par essayer un signal avec 64 axes mais il n'y a pas assez de points par axes par rapport au nombre total de points, et NanoEdgeStudio nous envoie plusieurs messages d'avertissement.
On prend donc le maximum d'axe pour une ligne 64 données, c'est à dire 4 puisqu’il faut au minimum 16 données par axe pour NanoEdge.
Séance du 19/12/2023
Entraînement Capteur STMicroelectronics :
On utilise le logiciel Nanoedgeaistudio, on configure donc cette fois-ci avec 4 axes.
Nous a lancé une nouvelle acquisition qui a durée toute la nuit, pour que celle-ci échoue au dernier moment car trop de valeur, temps d'execution trop long.
Partie Client (Raspberry) :
On modifie le programme de la Raspberry de manière à ce qu'elle accepte la connexion avec le capteur et on a élaboré une ébauche de traitement de données avant envoie de requêtes brut en JSON au serveur.
import requests
import json
import sys
import serial
import time
def read_from_sensor(ser):
if ser.in_waiting > 0:
line = ser.readline() # Lire les données en tant que bytes
return line
return None
def send_post_request(url, data):
try:
response = requests.post(url, data=json.dumps(data), headers={'Content-)
print("Réponse du serveur :", response.text)
except requests.exceptions.RequestException as e:
print("Erreur lors de l'envoi de la requête POST :", e)
def convert_data_to_json(data):
try:
# Convertir les données binaires en chaîne de caractères
string_data = data.decode('utf-8', 'ignore').strip()
# Séparer les nombres et convertir en entiers ou flottants
numbers = [int(num.strip()) for num in string_data.split(',') if num.st]
# Créer une structure de données JSON (exemple simple)
json_data = {"values": numbers}
return json_data
except Exception as e:
print("Erreur lors de la conversion des données :", e)
return None
def main():
try:
ser = serial.Serial('/dev/ttyACM0', 460800, timeout=1)
ser.flush()
url = 'http://[2001:660:4401:6050:216:3eff:fe4c:2673]:8888'
while True:
data = read_from_sensor(ser)
if data:
print("Données lues du capteur (transmises dans les logs):", da)
json_data = convert_data_to_json(data)
if json_data:
send_post_request(url, json_data)
time.sleep(1)
except serial.SerialException as se:
print("Erreur lors de l'ouverture du port série :", se)
sys.exit(1)
if __name__ == "__main__":
main()
Test de la communication (avec récupération des données du capteur) :
On observe à droite de l'image, les logs du programme client géré par connexion ssh à l'intérieur de la Raspberry, celle-ci étant connectée au capteur. Tandis qu'à gauche, on observe les logs de la partie serveur localisé dans la VM à l'intérieur de Chassiron.
On est donc capable de transmettre des données du capteur jusqu'au serveur en passant par la Raspberry qui ne fait pour l'instant aucun traitement.
Séance du 20/12/2023
Au cours de cette séance, nous avons eu des difficulté à exploité le Benchmark effectué à la séance précédente. Ne pouvant donc pas exploité ces résultats qui nous ont pris du temps à avoir, nous avons décidé de passer sur un projet de reconnaissance "statique". L'objectif sera de détecter si une personne porte des lunettes ou non.
NanoEdgeAiStudio :
On a donc relancé un nouveau benchmark composé de 2 signaux avec cette fois 64 axes pour utiliser chaque valeur de la matrice indépendamment (même si cela ne plaît pas à NanoEdgeStudio). Pour l’entraîner, nous nous sommes placé à une distance du capteur quasiment constante mais nous avons changer la position de notre tête sur les côté et nous avons aussi entraîné avec deux personnes différentes afin que l'intelligence ne soit pas entraîner pour une unique personne.
Les signaux utilisé ne comporte environ 75 lignes chacune ce qui n'est pas énorme mais qui nous a tout de même demandé 30min afin de réaliser le Benchmark. Nous aurions pu réaliser quelque chose de plus précis avec plus de lignes par signaux seulement cela n'était pas réalisable en terme de temps.
Communication :
En parallèle de la réalisation du benchmark nous avons modifié le code côté serveur de manière à afficher la page html à qui on transmet les valeurs et non afficher le code Python directement :
# -*- coding: utf-8 -*-
from http.server import HTTPServer, BaseHTTPRequestHandler
import json
import socket
import logging
import os
# Configurer le système de journalisation
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
# Stocker les requêtes POST reçues
post_requests = []
class CustomHTTPHandler(BaseHTTPRequestHandler):
def do_GET(self):
logger.info(f"Requête GET reçue pour {self.path}")
try:
if self.path == '/':
self.path = '/index.html'
if self.path == '/data':
self.send_response(200)
self.send_header('Content-type', 'application/json')
self.end_headers()
self.wfile.write(bytes(json.dumps(post_requests), 'utf-8'))
return
# Gérer l'affichage de index.html
if self.path == '/index.html':
file_path = os.path.join(os.getcwd(), self.path[1:])
if os.path.isfile(file_path):
with open(file_path, 'r') as file:
content = file.read()
self.send_response(200)
self.send_header('Content-type', 'text/html')
self.end_headers()
self.wfile.write(content.encode('utf-8'))
else:
self.send_error(404, 'File Not Found: %s' % self.path)
return
except FileNotFoundError:
self.send_error(404, 'File Not Found: %s' % self.path)
def do_POST(self):
content_length = int(self.headers['Content-Length'])
post_data = self.rfile.read(content_length).decode('utf-8')
# Ajouter les données POST à la liste
post_requests.append(post_data)
# Envoyer une réponse HTTP
self.send_response(200)
self.end_headers()
self.wfile.write(b"POST request processed")
logger.info("Requête POST reçue et traitée avec succès.")
class HTTPServerV6(HTTPServer):
address_family = socket.AF_INET6
if __name__ == '__main__':
server_address = ('::', 8888) # Écoute sur toutes les interfaces IPv6, port 8888
httpd = HTTPServerV6(server_address, CustomHTTPHandler)
print("Serveur actif sur le port", 8888)
logger.info("Serveur actif sur le port 8888")
httpd.serve_forever()
Déploiement de l'application :
Une fois le Benchmark achevé, nous avons tester le résultat et la détection fonctionnait correctement.
Ensuite nous avons copié les fichiers librairies qui sont sorties du benchmark du logiciel NanoEdgeAiStudio dans la VM ainsi que dans la Raspberry PI de manière à implémenter les parties traitements dans ces 2 cas, pour pouvoir par la suite, comparer les performances temporelle en termes de l'execution mais également les performances au niveau consommation énergétique.
Afin d'utiliser les libraries télécharger correctement nous avons suivis la documentation joint : https://wiki.st.com/stm32mcu/wiki/AI:NanoEdge_AI_Emulator_for_n-class_classification_(nCC)
Nous avons observer un problème lors de l'analyse des datas sur la Raspberry: l’exécutable d'analyse ne pouvait pas se lancer à cause d'une erreur.
Pour procéder autrement, nous envoyons directement les datas brut depuis le client au serveur et c'est le serveur qui traite les datas pour indiquer sur notre page web si la personne observé porte des lunettes ou non.
Mais sur la VM, cela fonctionne correctement.
Dans notre VM, nous avons prévu un traitement adaptable à chaque système de reconnaissance d'IA, ici le programme du serveur exécute le programme IA issue de NanoEdgeAiStudio avec les valeurs reçues de la Raspberry client en argument et les envoie à une page index.html qui va donc afficher le résultat comme cela :
from http.server import HTTPServer, BaseHTTPRequestHandler
import json
import socket
import logging
import os
import subprocess
import re
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
tmp = []
result = "" # To store the processed result
class CustomHTTPHandler(BaseHTTPRequestHandler):
def do_GET(self):
global result
logger.info(f"Requête GET reçue pour {self.path}")
if self.path == '/':
self.path = '/index.html'
if self.path == '/data':
self.send_response(200)
self.send_header('Content-type', 'application/json')
self.end_headers()
self.wfile.write(bytes(json.dumps([result]), 'utf-8')) # Send the processed result
elif self.path.endswith(".html"):
try:
file_path = os.path.join(os.getcwd(), self.path[1:])
if os.path.isfile(file_path):
with open(file_path, 'r') as file:
self.send_response(200)
self.send_header('Content-type', 'text/html')
self.end_headers()
self.wfile.write(file.read().encode())
else:
self.send_error(404, 'File Not Found: %s' % self.path)
except FileNotFoundError:
self.send_error(404, 'File Not Found: %s' % self.path)
else:
self.send_error(404, 'Resource Not Found: %s' % self.path)
def do_POST(self):
content_length = int(self.headers['Content-Length'])
post_data = self.rfile.read(content_length).decode('utf-8')
if post_data:
try:
data = json.loads(post_data)
print(f"Voici mon data : {data}")
numbers = self.extract_numbers(data)
global tmp
tmp = numbers
self.send_response(200)
self.end_headers()
self.wfile.write(b"POST request processed")
logger.info("Requête POST reçue et traitée avec succès.")
run_nanoegeai_class_emulator()
except json.JSONDecodeError:
logger.error("Invalid JSON received")
self.send_error(400, "Invalid JSON")
else:
logger.error("Empty POST data received")
self.send_error(400, "Empty POST data")
def extract_numbers(self, data):
numbers = []
if "values" in data and isinstance(data["values"], list):
for value in data["values"]:
if isinstance(value, (int, float)):
numbers.append(value)
return numbers
def run_nanoegeai_class_emulator():
global tmp, result
data_str = ' '.join(map(str, tmp))
print(f"Voici mon data_str : {data_str}")
cmd = f"./libneai_avecetsanslunettes_2/emulators/NanoEdgeAI_Class_Emulator neai_classification --knowledge_path libneai_avecetsanslunettes_2/emulators/knowledge.dat --array {data_str}"
try:
with subprocess.Popen(cmd, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True) as process:
stdout, stderr = process.communicate()
if process.returncode == 0:
logger.info("NanoEdgeAI Class Emulator executed successfully")
logger.info(stdout)
# Verifier la sortie pour "Aveclunettes" ou "SansLunettes"
if "AvecLunettes" in stdout:
result = "Vous portez des lunettes"
elif "SansLunettes" in stdout:
result = "Vous ne portez pas de lunettes"
else:
result = "Unknown"
logger.error("Aucun résultat valide trouvé dans la sortie.")
else:
logger.error(f"Error in NanoEdgeAI Class Emulator: {stderr}")
result = "Error"
except Exception as e:
logger.error(f"Exception occurred: {e}")
result = "Exception"
class HTTPServerV6(HTTPServer):
address_family = socket.AF_INET6
if __name__ == '__main__':
server_address = ('::', 8888)
httpd = HTTPServerV6(server_address, CustomHTTPHandler)
print("Serveur actif sur le port", 8888)
logger.info("Serveur actif sur le port 8888")
httpd.serve_forever()
On récupère à la réception des requêtes POST par la VM et donc en sortie de la Raspberry PI, des signaux comme cela :
{'values': [1877, 1914, 1909, 1922, 1928, 1932, 1951, 2020, 1871, 1885, 1898, 1898, 1903, 1913, 1933, 1949, 1877, 1879, 1888, 1898, 1902, 1902, 1924, 1937, 1862, 1882, 1889, 1902, 1904, 1902, 1910, 1921, 1870, 1876, 1877, 1888, 1895, 1891, 1897, 1908, 1870, 1879, 1871, 1883, 1922, 1918, 1941, 1924, 1864, 1857, 1863, 1869, 1869, 1888, 1898, 1945, 1875, 1859, 1857, 1857, 1868, 1880, 1917, 1936]}
Qui sont ensuite transformés en signal pour mettre en argument de la commande ci-dessous
1877 1914 1909 1922 1928 1932 1951 2020 1871 1885 1898 1898 1903 1913 1933 1949 1877 1879 1888 1898 1902 1902 1924 1937 1862 1882 1889 1902 1904 1902 1910 1921 1870 1876 1877 1888 1895 1891 1897 1908 1870 1879 1871 1883 1922 1918 1941 1924 1864 1857 1863 1869 1869 1888 1898 1945 1875 1859 1857 1857 1868 1880 1917 1936
Le programme effectue donc cette commande :
./libneai_avecetsanslunettes_2/emulators/NanoEdgeAI_Class_Emulator neai_classification --knowledge_path libneai_avecetsanslunettes_2/emulators/knowledge.dat --array {data_str}
Le programme récupère donc une réponse de cette forme :
"status": "classification",
"input": "array",
"results": [
{"signal": 1, "line": "null", "class_status": 1, "class_name": "AvecLunettes", "class_proba": [0.50, 0.50]}
],
"classification_summary": {"signals": 1, "classified": [1, 0], "unclassified": 0}
}
Celui-ci récupère ensuite le class_name : "AvecLunettes" ou "SansLunettes" de manière à ensuite l'afficher sur l'index.html :
