SE5 ECEAI/eceai 2023/2024/ValleeSellali
Groupe Vallée-Sellali
Séance n°1:
Machine virtuelle créée sur chassiron :
Hostname : ValSel
Distribution : bookworm
Stockage : 10G
Mémoire vive : 1G
Mot de passe habituel
Les fichiers /etc/network/interfaces, /etc/resolv.conf et /etc/apt/sources.list ont été modifiés comme demandés.
Raspberry 4 :
Communication via le port série possible
Login : valsel Pwd : pasglop
STM32F401RE : Nanoedge et IDE installés, conception d'un code pour tester le capteur en cours
Séance n°2 (18/12/2023) :
Raspberry connectée au WiFi SE5
Connexion ipv6 possible entre la VM et la raspberry
Données du capteur lues et envoyées au port série
Création d'un programme python client à destination de la raspberry qui : lit le port série, stocke les données reçues, envoie les données reçues en http vers le "serveur" présent sur la VM
Création d'un programme python serveur à destination de la VM qui reçoit et stocke les données
Séance 3 (19/12/2023) :
Création des datasets : main fermée et main ouverte (Une ligne = 16 * 16 données, environ 30 lignes par fichier)
Choix de garder les capteurs en 4x4 pour un compromis vitesse de création des datasets/temps d'entrainement du modèle/efficacité du capteur de mouvement.
Entrainement du modèle avec les datasets sur NanoEdge
Séance 4 (20/12/2023)
Création de nouveaux datasets et de fichiers de tests : (100 lignes de 16 * 16 données)
Création d'un script Python utilisant la bibliothèque sklearn pour implémenter un algorithme de classification SVM, à destination de la Raspberry et de la VM
Le choix de main ouverte/main fermée se questionne, on utilise pas l'aspect mouvement que permet les algorithmes.
Les algorithmes donnent un bon résultat avec une précision proche de 70%. La communication entre la VM et la zabeth sur laquelle on teste les fichiers pythons fonctionne.
Du microcontrôleur, on reçoit les données brutes et l'estimation de l'état de la main. Le fichier python sur la zabeth, à destination future de la raspberry, reçoit ces données via le port série, les stocke dans un fichier txt, fait une estimation de l'état de la main, et envoie les données brutes à la VM en utilisant TCP/IP avec l'adresse IPV6 de la VM.
La VM reçoit les données mais ne les traite pas encore.
Travail hors des séances et rendu final :
Extension du programme sur la STM aux 64 capteurs.
Nouveau choix de test qui prend en compte les mouvements.
Création de nouveaux datasets adaptés : lignes de 64 * 16 données, une ligne correspond à une itération du mouvement, environ 170 lignes par datasets pour une prise en compte de plusieurs cas
(passage rapide, passage lent, hauteurs différentes). (Utilisation du code annexe GestionData sans la communication TCP pour la création des fichiers)
Les mouvements sont :
- Mouvement de gauche à droite
- Mouvement de droite à gauche
- Aucune présence
Pour pouvoir tester les modèles résultant de ces datasets, 3 fichiers de tests sont créés contenant chacun une trentaine d'itérations du mouvement.
Avec Nanoedge, entraînement d'un nouveau modèle à destination de la STM. Le modèle le plus efficace avec nos fichiers de tests sont, contrairement aux premières séances, un modèle MLP (multilayer
perceptron). Résultats avec nos fichiers de tests : précision de 90%.
Intégration de la bibliothèque libneai.a créée par Nanoedge dans le projet STM et adaptation du code de la STM pour y intégrer l'estimation à partir des données des capteurs. (Annexe app_tof.c)
En effet, les modèles SVM ne semblent pas adaptés à nos datasets => Adaptation du code python de la raspberry pour utiliser lui aussi un modèle MLP, ce qui est plus efficace. (entraînement du modèle avec le programme mlpAlgo.py présent en annexe)
Modèle résultant avec comme input les fichiers de tests : 70% de précision sur Droite à gauche, 68% de gauche à droite, 100% si aucune présence.
Intégration du code GestionData.py dans la raspberry ainsi que du modèle créé par mlpAlgo.py (uniquement l'output)
Réseau :
- La STM32 envoie les données brutes + l'estimation du mouvement issue de son modèle intégré, par le port série (ap_tof.c disponible en annexe)
- La Raspberry Pi reçoit ces données brutes, les stocke dans un fichier texte, estime le mouvement effectué, et affiche l'estimation en plus de celle de la STM. Ensuite, les données brutes sont transmises à la VM par TCP/IP en IPV6, à la condition que 1024 données soient présentes à chaque lecture. Si ce n'est pas le cas, les données "corrompues" ne sont pas transmises puisqu'inexploitables. (Voir GestionData.py en annexe)
- La VM reçoit les données par paquet de 1024 entiers, stocke les données dans un fichier txt, et fait une estimation du mouvement en utilisant le même modèle que celui présent dans la raspberry pi.
Lors des tests, nous avons constaté que la communication TCP ne semblait pas permise depuis un ordinateur externe au réseau des zabeths vers la VM : le ping est possible, mais la communication ne semble pas. Nous avons donc fait le choix d'utiliser un ordinateur personnel pour montrer l'efficacité du projet : le fonctionnement est similaire, les données sont envoyées sur la machine personnelle en IPV6.
Vidéo :
Image de l'écran capturé dans la vidéo : DemoVideo.mp4
On constate : Si les données reçues ne contiennent pas 1024 données avant le \n, alors ces données ne sont pas transmises car non exploitables par les modèles de la RPi/VM. On obtient tout de même l'estimation de la STM32.
Comme montré sur la vidéo, on constate que lorsqu'il n'y a aucun mouvement, les estimations sont bonnes.
Pour les trois tests de gauche à droite, les estimations sont bonnes.
Pour les trois tests de droite à gauche, les estimations sont bonnes, à l'exception de la dernière estimation donnée par la RPi et par la VM (vu que les modèles sont les mêmes et qu'ils utilisent les mêmes données à ce moment) : cela est du aux 70% de précision observés lors des tests.
Annexe
Tous les codes rendus en annexe sont dans leur version finale et ont été utilisés au cours du projet.
Codes externes ayant servis à l'entraînement :
Les fichiers de type data ou test ont été réalisés en lisant le port série et en insérant directement les données dans le fichier texte à condition que les données ne soient pas corrompues.
mlpAlgo.py : programme permettant de générer un modèle utilisant l'algorithme MLP
#Import mlp model
from sklearn import neural_network
from sklearn import metrics
import joblib
GtoD_path = 'data_GtoD.txt'
DtoG_path = 'data_DtoG.txt'
GtoD_test_path = 'test_GtoD.txt'
DtoG_test_path = 'test_DtoG.txt'
rien_path = 'data_rien.txt'
rien_test_path = 'test_rien.txt'
GtoD_train = []
DtoG_train = []
GtoD_target = []
DtoG_target = []
rien_train = []
rien_target = []
test = []
data = []
targets = []
target_test = []
def ReadAndFill(nom_fichier, target_tab, target_int, taille_sous_liste=1024):
try:
# Ouverture du fichier en mode lecture
with open(nom_fichier, 'r') as fichier:
# Lecture du contenu du fichier ligne par ligne
lignes = fichier.readlines()
# Pour chaque ligne, conversion en liste d'entiers
tableau = [[int(nombre) for nombre in ligne.split()] for ligne in lignes]
tableau_filtre = [sous_liste for sous_liste in tableau if len(sous_liste) == taille_sous_liste]
target_tab = len(tableau_filtre) * [target_int]
return tableau_filtre, target_tab
except FileNotFoundError:
print(f"Le fichier {nom_fichier} n'a pas été trouvé.")
return []
except Exception as e:
print(f"Une erreur s'est produite : {e}")
return []
# Exemple d'utilisation
GtoD_train, GtoD_target = ReadAndFill(GtoD_path, GtoD_target, 2)
DtoG_train, DtoG_target = ReadAndFill(DtoG_path, DtoG_target, 1)
rien_train, rien_target = ReadAndFill(rien_path, rien_target, 0)
data = GtoD_train + DtoG_train + rien_train
targets = GtoD_target + DtoG_target + rien_target
#Create a svm Classifier
clf = neural_network.MLPClassifier(max_iter=5000000)
#Train the model using the training sets
clf.fit(data, targets)
joblib.dump(clf, 'modele_mlp2.pkl')
test, target_test = ReadAndFill(GtoD_test_path, target_test, 2)
#Predict the response for test dataset
pred = clf.predict(test)
print(pred)
# Model Accuracy: how often is the classifier correct?
print("Fichier de test actuel :",GtoD_test_path)
print("Accuracy:",metrics.accuracy_score(target_test, pred))
test, target_test = ReadAndFill(DtoG_test_path, target_test, 1)
#Predict the response for test dataset
pred = clf.predict(test)
print(pred)
# Model Accuracy: how often is the classifier correct?
print("Fichier de test actuel :",DtoG_test_path)
print("Accuracy:",metrics.accuracy_score(target_test, pred))
test, target_test = ReadAndFill(rien_test_path, target_test, 0)
#Predict the response for test dataset
pred = clf.predict(test)
print(pred)
# Model Accuracy: how often is the classifier correct?
print("Fichier de test actuel :",rien_path)
print("Accuracy:",metrics.accuracy_score(target_test, pred))
mlpPredic.py : Code similaire servant à tester de charger un modèle créé et sauvegardé auparavant
#Import mlp model
from sklearn import neural_network
from sklearn import metrics
import joblib
GtoD_test_path = 'test_GtoD.txt'
DtoG_test_path = 'test_DtoG.txt'
rien_test_path = 'test_rien.txt'
test = []
data = []
targets = []
target_test = []
def ReadAndFill(nom_fichier, target_tab, target_int, taille_sous_liste=1024):
try:
# Ouverture du fichier en mode lecture
with open(nom_fichier, 'r') as fichier:
# Lecture du contenu du fichier ligne par ligne
lignes = fichier.readlines()
# Pour chaque ligne, conversion en liste d'entiers
tableau = [[int(nombre) for nombre in ligne.split()] for ligne in lignes]
tableau_filtre = [sous_liste for sous_liste in tableau if len(sous_liste) == taille_sous_liste]
target_tab = len(tableau_filtre) * [target_int]
return tableau_filtre, target_tab
except FileNotFoundError:
print(f"Le fichier {nom_fichier} n'a pas été trouvé.")
return []
except Exception as e:
print(f"Une erreur s'est produite : {e}")
return []
#Create a svm Classifier
clf = joblib.load('modele_mlp.pkl')
test, target_test = ReadAndFill(GtoD_test_path, target_test, 2)
#Predict the response for test dataset
pred = clf.predict(test)
print(pred)
# Model Accuracy: how often is the classifier correct?
print("Fichier de test actuel :",GtoD_test_path)
print("Accuracy:",metrics.accuracy_score(target_test, pred))
test, target_test = ReadAndFill(DtoG_test_path, target_test, 1)
#Predict the response for test dataset
pred = clf.predict(test)
print(pred)
# Model Accuracy: how often is the classifier correct?
print("Fichier de test actuel :",DtoG_test_path)
print("Accuracy:",metrics.accuracy_score(target_test, pred))
test, target_test = ReadAndFill(rien_test_path, target_test, 0)
#Predict the response for test dataset
pred = clf.predict(test)
print(pred)
# Model Accuracy: how often is the classifier correct?
print("Fichier de test actuel :",rien_test_path)
print("Accuracy:",metrics.accuracy_score(target_test, pred))
Codes internes à la STM :
app_tof.c : Code du process. Acquisition des données des 64 capteurs, 16 fois, envoi en temps réel des données et stockage temporaire des données pour faire une estimation avec le modèle interne à la stm32. Une fois les 1024 données récupérées, print d'un \n pour que "serial.readline()" des commandes python se termine.
Envoi de l'estimation après 1024 données.
static void MX_53L5A1_SimpleRanging_Process(void)
{
uint32_t Id;
int8_t compteur = 0;
uint8_t zones_per_line;
VL53L5A1_RANGING_SENSOR_ReadID(VL53L5A1_DEV_CENTER, &Id);
VL53L5A1_RANGING_SENSOR_GetCapabilities(VL53L5A1_DEV_CENTER, &Cap);
Profile.RangingProfile = RS_PROFILE_8x8_CONTINUOUS;
Profile.TimingBudget = TIMING_BUDGET; /* 5 ms < TimingBudget < 100 ms */
Profile.Frequency = RANGING_FREQUENCY; /* Ranging frequency Hz (shall be consistent with TimingBudget value) */
Profile.EnableAmbient = 0; /* Enable: 1, Disable: 0 */
Profile.EnableSignal = 0; /* Enable: 1, Disable: 0 */
uint16_t id_class = 0;
float input_user_buffer[DATA_INPUT_USER * AXIS_NUMBER]; // Buffer of input values
float output_class_buffer[CLASS_NUMBER];
zones_per_line = ((Profile.RangingProfile == RS_PROFILE_8x8_AUTONOMOUS) ||
(Profile.RangingProfile == RS_PROFILE_8x8_CONTINUOUS)) ? 8 : 4;
enum neai_state error_code = neai_classification_init(knowledge);
if (error_code != NEAI_OK)
{
printf("Erreur lib \n");
}
/* set the profile if different from default one */
VL53L5A1_RANGING_SENSOR_ConfigProfile(VL53L5A1_DEV_CENTER, &Profile);
status = VL53L5A1_RANGING_SENSOR_Start(VL53L5A1_DEV_CENTER, RS_MODE_BLOCKING_CONTINUOUS);
if (status != BSP_ERROR_NONE)
{
printf("VL53L5A1_RANGING_SENSOR_Start failed\n");
while (1);
}
while (1)
{
/* polling mode */
status = VL53L5A1_RANGING_SENSOR_GetDistance(VL53L5A1_DEV_CENTER, &Result);
int sous_buffer[DATA_INPUT_USER];
if (status == BSP_ERROR_NONE)
{
print_result(&Result, sous_buffer);
//afficher_tableau(sous_buffer, DATA_INPUT_USER);
fill_buffer(input_user_buffer, sous_buffer, compteur);
}
if (compteur == 15)
{
//printf("Class number :%d Data user input : %d axis : %d multiply : %d",CLASS_NUMBER, DATA_INPUT_USER, AXIS_NUMBER, AXIS_NUMBER*DATA_INPUT_USER);
neai_classification(input_user_buffer, output_class_buffer, &id_class);
//afficher_tableau(input_user_buffer, DATA_INPUT_USER*AXIS_NUMBER);
printf("\n");
printf("%s\n",id2class[id_class]);
compteur = 0;
}
else
{
compteur++;//printf("Compteur = %d",compteur);
}
if (com_has_data())
{
handle_cmd(get_key());
}
HAL_Delay(POLLING_PERIOD);
}
}
static void print_result(RANGING_SENSOR_Result_t *Result, int sous_buffer[DATA_INPUT_USER])
{
int8_t j;
int8_t k;
int8_t l;
int8_t buffer_position = 0;
int8_t total_values = 0;
uint8_t zones_per_line;
zones_per_line = ((Profile.RangingProfile == RS_PROFILE_8x8_AUTONOMOUS) ||
(Profile.RangingProfile == RS_PROFILE_8x8_CONTINUOUS)) ? 8 : 4;
for (j = 0; j < Result->NumberOfZones; j += zones_per_line)
{
for (l = 0; l < RANGING_SENSOR_NB_TARGET_PER_ZONE; l++)
{
/* Print distance and status */
for (k = (zones_per_line - 1); k >= 0; k--)
{
if (Result->ZoneResult[j+k].NumberOfTargets > 0)
{
printf("%d ",(int)Result->ZoneResult[j+k].Distance[l]);
sous_buffer[buffer_position] = (int)Result->ZoneResult[j+k].Distance[l];
buffer_position++;
}
else
printf("%d ", '10');
}
}
}
//printf("total : %d ",total_values);
}
Codes internes à la Raspberry pi :
GestionData.py : Lis le port série, stocke les données, effectue une estimation du mouvement, affiche l'estimation provenant de la STM32 et envoie les données brutes à la VM via TCP
COM3 pour les test depuis la machine windows, ACM0 pour l'application finale sur la raspberry
IPV6 de la VM laissée en commentaire
import serial
import datetime
import csv
import socket
import joblib
import requests
import struct
port = 'COM3'#'/dev/ttyACM0'
baudrate = 460800
ser = serial.Serial(port, baudrate)
csv_filename = 'donnees_raspberry.txt'
max_lines = 200
clf = joblib.load('modele_mlp.pkl')
server_address = '2001:861:3541:4260:c69f:9d4:ccf5:7673'#'2001:660:4401:6050:216:3eff:febc:550c'
server_port = 12345
classes = ["Aucune presence", "Mouvement de droite a gauche","Mouvement de gauche a droite"]
def DataToTab(ligne, taille_sous_liste=1024):
tableau = [int(nombre) for nombre in ligne.split()]
if (len(tableau)!=taille_sous_liste):
return tableau, 1
return tableau, 0
with open(csv_filename, 'w', newline='') as csv_file:
csv_writer = csv.writer(csv_file)
line_count = 0
client_socket = socket.socket(socket.AF_INET6,socket.SOCK_STREAM)
client_socket.connect((server_address, server_port))
try:
while line_count < max_lines:
data = ser.readline().decode('utf-8').strip()
#print("Ligne : ", line_count,"\n")
timestamp = datetime.datetime.now().strftime('%H:%M:%S')
#print(f"{timestamp} - Données lues: {data}")
headers = {'Content-type': 'text/plain'}
body = f"{timestamp} - Données lues: {data}"
tableau, error = DataToTab(data)
if error == 1:
print("Taille de tableau non conforme")
else:
pred = clf.predict([tableau])
print("Classe identifiee sur la Raspberry:", classes[pred[0]])
try:
donnees_brutes = struct.pack('!{}i'.format(len(tableau)), *tableau)
client_socket.sendall(donnees_brutes)
except Exception as e:
print(f"Erreur : {e}")
csv_writer.writerow([data])
classe = ser.readline().decode('utf-8').strip()
print("Classe identifiee par la STM:", classe);
line_count += 1
except KeyboardInterrupt:
pass
finally:
client_socket.close()
ser.close()
Codes internes à la VM (présents en réalité sur la machine personnelle) :
server_tcp.py : Ecoute sur le port donné, reçois les données brutes, les stocke, et estime le mouvement effectué
import socket
import struct
import csv
import joblib
import datetime
packet_size = 1024
classes = ["Aucune presence", "Mouvement de droite a gauche","Mouvement de gauche a droite"]
csv_filename = 'donnees_recues.txt'
max_lines = 200
clf = joblib.load('modele_mlp.pkl')
def start_tcp_server(ip, port):
with open(csv_filename, 'w', newline='') as csv_file:
csv_writer = csv.writer(csv_file)
server_socket = socket.socket(socket.AF_INET6, socket.SOCK_STREAM)
server_socket.bind((ip, port))
server_socket.listen(1)
print(f"Serveur tcp en ecoute sur {ip}:{port}")
client_socket, client_address = server_socket.accept()
line_count = 0
try:
while line_count < max_lines:
donnees_brutes = client_socket.recv(packet_size * 4)
liste_entiers = list(struct.unpack('!{}i'.format(packet_size), donnees_brutes))
#print("Recu :", liste_entiers)
csv_writer.writerow(liste_entiers)
pred = clf.predict([liste_entiers])
timestamp = datetime.datetime.now().strftime('%H:%M:%S')
print(f"{timestamp}")
print("Classe identifiee sur la VM:", classes[pred[0]])
line_count+=1
except KeyboardInterrupt:
pass
finally:
client_socket.close
if __name__ == "__main__":
server_ip = '2001:861:5642:4610:46fb:e35c:a75d:2b74'
server_port = 12345
start_tcp_server(server_ip, server_port)