Fix deconnexion MQTT

PC/MQTT_MySQL.py

+import MySQLdb
+import paho.mqtt.client as mqtt
+import time
+import sys
+
+db = MySQLdb.connect(host="pil-09.univlehavre.lan",   
+                     user="firediag",         
+                     passwd="firediag$!",  
+                     db="firediag");
+
+global topics ; global frequenceSeuils; # frequenceSeuils => Map utilisee pour verifier l'effet de seuil
+topics = []; frequenceSeuils = {};
+
+topic_changeCapteur = "/system/changeCapteur";
+
+def on_connect(client, userdata, flags, rc):
+    if rc == 0:
+        print("Connected to broker");
+	client.subscribe(topic_changeCapteur);
+
+        global Connected;
+        Connected = True;
+ 
+    else:
+        print("Connection failed");
+
+def on_message(client, userdata, msg):
+	global topics ; global frequenceSeuils ;
+
+	print(msg.topic + " Payload -> " +str(msg.payload));
+	message_topic = msg.topic;
+	message_valeur = msg.payload;
+
+	if message_topic == topic_changeCapteur: # Mise a jour des topics auxquelles on s'abonne
+		for topic in topics:
+			client.unsubscribe(topic);
+		
+		update_capteur();
+		print("Update des topics => "+" ".join(topics));
+	else :					# Arrivee d'une valeur d'un capteur
+		cursor = db.cursor();
+		cursor.execute('SELECT topic, id, idUser, nom, idGroupe FROM capteur WHERE visible = TRUE');
+		for capteur in cursor.fetchall():
+	   		if capteur[0] == message_topic:
+				try:
+					cursor.execute("INSERT INTO `valeur` (`temps`, `valeur`, `idCapteur`) VALUES ('"+
+						time.strftime('%Y-%m-%d %H:%M:%S')+"', '"+str(message_valeur)+"', '"+str(capteur[1])+"')");
+					
+					print("Nouvelle valeur ('"+time.strftime('%Y-%m-%d %H:%M:%S')+"', '"+str(message_valeur)+"', '"+str(capteur[1])+"')");
+
+					if unicode(str(message_valeur).strip(), 'utf-8').isnumeric(): # Si c'est une valeur numerique
+						# Gestion des notifications des utilisateurs (ajout des alertes)
+						cursorNotif = db.cursor();
+						cursorNotif.execute('SELECT id, nom, valeur, regle, frequence FROM seuil WHERE idCapteur = '+str(capteur[1]));
+						
+						for seuil in cursorNotif.fetchall():
+							frequence = int(seuil[0]);
+							if	(seuil[3] == "sup" and int(message_valeur) > int(seuil[2])) or \
+								(seuil[3] == "inf" and int(message_valeur) < int(seuil[2])) or \
+								(seuil[3] == "eq" and int(message_valeur) == int(seuil[2])):
+
+								if frequence not in frequenceSeuils:	# on ajoute le seuil a la map 	
+									frequenceSeuils[frequence] = 1;
+								else: # sinon on l'incremente
+									frequenceSeuils[frequence] = frequenceSeuils[frequence]+1;
+
+								# Si l'effet de seuil est depassee, on lance l'alerte
+								if frequenceSeuils[frequence] >= seuil[4]:
+
+									print(" Alerte  \""+seuil[1]+"\" => "+message_valeur+" "+seuil[3]+" "+str(int(seuil[2])));
+									titre = "Alerte sur "+str(capteur[3]);
+									message = "Le seuil \""+str(seuil[1])+"\" a ete franchi ("+message_valeur ;
+									if seuil[3] == "sup" : message += " > " ;
+									elif seuil[3] == "inf" : message += " < " ;
+									else : message += " = " ;
+									message += str(int(seuil[2]))+")" ;
+
+									cursorNotif.execute("INSERT INTO `notification` (`temps`, `titre`, `message`, `idUser`, `idCapteur`, `vu`) VALUES ('"+
+										time.strftime('%Y-%m-%d %H:%M:%S')+"', '"+titre+"', '"+message+"', '"+str(capteur[2])+"', '"+str(capteur[1])+"', false)")
+									
+									# notification du changement
+									client.publish("/system/notification/"+str(capteur[2]), str(capteur[4])+"/"+str(capteur[1]));
+									
+									frequenceSeuils[frequence] = 0;
+							else:
+								frequenceSeuils[frequence] = 0; 
+							
+					db.commit();
+				except Exception as e:
+					db.rollback();
+					print("rollback ");
+					print(e);
+				
+
+def update_capteur():
+	global topics ;
+
+	topics = [];
+	cursor = db.cursor();
+	cursor.execute('SELECT topic FROM capteur WHERE visible = TRUE');
+	
+	for capteur in cursor.fetchall():
+		topics.append(capteur[0]);
+		client.subscribe(capteur[0]);
+	db.commit();
+	
+Connected = False;
+
+client = mqtt.Client(transport="websockets");
+client.on_connect = on_connect;
+client.on_message = on_message;
+client.connect("pil-09.univlehavre.lan", 9001, 60);
+ 
+client.loop_start() ;
+
+while Connected != True: 
+    time.sleep(0.1);
+
+update_capteur();
+
+try:
+	while True:
+		time.sleep(1);
+		 
+except KeyboardInterrupt:
+	print "exiting";
+	client.disconnect();
+	client.loop_stop();
+	db.close();

PC/Verification.py

+import paho.mqtt.client as mqtt
+import time
+import sys
+import subprocess
+import os
+import signal
+from random import randint
+
+# Pour couper le script :
+# > ps -A | grep python
+# > sudo kill -9 ...
+ 
+global mqttTopic, errorDelay, lastMessageTime ;
+global pid_MySQL, pid_Generation, client;
+mqttTopic = "/system/mqtt";
+errorDelay = 60; # On laisse 60 secondes au serveur avant de le redemmarer
+
+def on_connect(client, userdata, flags, rc):
+    if rc == 0:
+        print("Connected to broker");
+
+        global Connected;
+        Connected = True;
+ 
+    else:
+        print("Connection failed");
+
+
+def on_message(client, userdata, msg):
+	global mqttTopic, lastMessageTime ;
+	print(msg.topic + " Payload -> " +str(msg.payload));
+	if msg.topic == mqttTopic :
+		lastMessageTime = time.time();	
+		print(" MQTT fonctionne ");
+
+
+def startBackgroundScript():
+	global pid_MySQL;#, pid_Generation;
+	global Connected, client ;
+	
+	client = mqtt.Client(transport="websockets");
+	client.on_connect = on_connect;
+	client.on_message = on_message;
+
+	try:
+		client.connect("pil-09.univlehavre.lan", 9001, 60);
+	except:
+		print("Echec connection");
+	
+	client.loop_start() ;
+
+	while Connected != True: 
+	    time.sleep(0.1);
+
+	client.subscribe(mqttTopic);
+
+	#proc = subprocess.Popen("sudo python GenerationDonneesMQTT.py", shell=True, preexec_fn=os.setsid);
+	#pid_Generation = proc.pid ;
+	proc = subprocess.Popen("sudo python MQTT_MySQL.py", shell=True, preexec_fn=os.setsid);
+	pid_MySQL = proc.pid ;
+
+	print("\n\nSTART SERVER : "+str(pid_MySQL));#+" "+str(pid_Generation));
+
+def restartServer():
+	global pid_MySQL;#, pid_Generation;
+	
+	print("\n\nERROR SERVER : "+str(pid_MySQL));#+" "+str(pid_Generation));
+	
+	os.killpg(os.getpgid(pid_MySQL), signal.SIGKILL) ;
+	#os.killpg(os.getpgid(pid_Generation), signal.SIGKILL) ;
+	
+	subprocess.Popen("sudo systemctl restart mosquitto", shell=True);
+	time.sleep(2);
+
+	startBackgroundScript();
+
+Connected = False;
+
+startBackgroundScript();
+lastMessageTime = time.time();
+
+try:
+    while True:	
+	print((time.time()-lastMessageTime));
+	if (time.time()-lastMessageTime) > errorDelay:
+		restartServer();	
+		lastMessageTime = time.time();
+
+	time.sleep(10); # 10 secondes
+
+        client.publish(mqttTopic, 0);
+	print("messsage securite");
+	
+except KeyboardInterrupt:
+    client.disconnect();
+    client.loop_stop();