from flask import Flask, Response, jsonify import cv2 import numpy as np import face_recognition import os import subprocess import threading import time import RPi.GPIO as GPIO # NOUVEAU import paho.mqtt.client as mqtt # NOUVEAU # ── LM393 GPIO ─────────────────────────────────────────────────── NOUVEAU GPIO.setmode(GPIO.BCM) GPIO.setup(17, GPIO.IN) # ── MQTT pour LM393 ────────────────────────────────────────────── NOUVEAU def connecter_mqtt_lm(): try: client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION1) client.connect("broker.hivemq.com", 1883, keepalive=60) client.loop_start() print("MQTT LM393 connecte") return client except Exception as e: print("MQTT LM393 erreur:", e) return None mqtt_lm = connecter_mqtt_lm() # NOUVEAU # ── Thread publication LM393 ───────────────────────────────────── NOUVEAU def publier_lumiere(): global mqtt_lm while True: try: valeur = GPIO.input(17) if mqtt_lm: mqtt_lm.publish("berceau/135701/lumiere", str(valeur)) print(f"Lumiere publiee : {valeur}") else: mqtt_lm = connecter_mqtt_lm() except Exception as e: print("Erreur LM393:", e) mqtt_lm = connecter_mqtt_lm() time.sleep(2) threading.Thread(target=publier_lumiere, daemon=True).start() # NOUVEAU # ── Chargement des visages connus ────────────────────────────────────────── encodings_connus = [] noms_connus = [] DOSSIER = "/home/younes/visages_connus" if not os.path.exists(DOSSIER): os.makedirs(DOSSIER) for file in os.listdir(DOSSIER): if file.endswith((".jpg", ".png")): chemin_f = os.path.join(DOSSIER, file) if os.path.getsize(chemin_f) == 0: continue try: img = face_recognition.load_image_file(chemin_f) encodages = face_recognition.face_encodings(img) if encodages: encodings_connus.append(encodages[0]) noms_connus.append(os.path.splitext(file)[0]) print(f"Profil charge : {os.path.splitext(file)[0]}") except Exception as e: print(f"Erreur {file}: {e}") # ── Statut reconnaissance (partagé) ─────────────────────────────────────── recognition_status = { "nom": "Aucune detection", "alerte": False, "nb_visages": 0, "timestamp": "" } # ── Caméra partagée ─────────────────────────────────────────────────────── class SharedCamera: def __init__(self): self.frame = None self.lock = threading.Lock() self.thread = threading.Thread(target=self._capture_loop, daemon=True) self.thread.start() def _capture_loop(self): global recognition_status pipeline_cmd = [ "rpicam-vid", "-t", "0", "--inline", "-n", "--width", "640", "--height", "480", "--framerate", "15", "--codec", "yuv420", "-o", "-" ] process = subprocess.Popen(pipeline_cmd, stdout=subprocess.PIPE, bufsize=10**7) taille_frame = 640 * 480 * 3 // 2 frame_count = 0 while True: raw_frame = process.stdout.read(taille_frame) if len(raw_frame) != taille_frame: print("Erreur flux camera, relance...") time.sleep(2) process = subprocess.Popen(pipeline_cmd, stdout=subprocess.PIPE, bufsize=10**7) continue yuv = np.frombuffer(raw_frame, dtype=np.uint8).reshape((720, 640)) frame = cv2.cvtColor(yuv, cv2.COLOR_YUV2BGR_I420) frame = cv2.flip(frame, -1) frame_count += 1 if frame_count % 5 == 0: small_frame = cv2.resize(frame, (0, 0), fx=0.25, fy=0.25) rgb_small_frame = cv2.cvtColor(small_frame, cv2.COLOR_BGR2RGB) face_locations = face_recognition.face_locations(rgb_small_frame) face_encodings_fr = face_recognition.face_encodings(rgb_small_frame, face_locations) alerte = len(face_locations) == 0 noms_detectes = [] for (top, right, bottom, left), face_enc in zip(face_locations, face_encodings_fr): nom = "Inconnu" if encodings_connus: matches = face_recognition.compare_faces(encodings_connus, face_enc, tolerance=0.55) distances = face_recognition.face_distance(encodings_connus, face_enc) best = np.argmin(distances) if matches[best]: nom = noms_connus[best] if nom == "Inconnu": alerte = True noms_detectes.append(nom) top *= 4; right *= 4; bottom *= 4; left *= 4 couleur = (0, 220, 100) if nom != "Inconnu" else (0, 0, 255) cv2.rectangle(frame, (left, top), (right, bottom), couleur, 2) cv2.putText(frame, nom.upper(), (left, top - 10), cv2.FONT_HERSHEY_SIMPLEX, 0.75, couleur, 2) if alerte and face_locations: overlay = frame.copy() overlay[:] = (0, 0, 200) cv2.addWeighted(overlay, 0.25, frame, 0.75, 0, frame) cv2.putText(frame, "ACCES NON AUTORISE", (30, 50), cv2.FONT_HERSHEY_TRIPLEX, 0.8, (255, 255, 255), 2) recognition_status["nb_visages"] = len(face_locations) recognition_status["alerte"] = alerte recognition_status["timestamp"] = time.strftime("%H:%M:%S") if noms_detectes: recognition_status["nom"] = ", ".join(noms_detectes) else: recognition_status["nom"] = "Personne detectee" ts = time.strftime("%d/%m/%Y %H:%M:%S") cv2.putText(frame, ts, (10, 470), cv2.FONT_HERSHEY_SIMPLEX, 0.5, (200, 200, 200), 1) _, buffer = cv2.imencode('.jpg', frame, [cv2.IMWRITE_JPEG_QUALITY, 75]) with self.lock: self.frame = buffer.tobytes() def get_frame(self): with self.lock: return self.frame camera = SharedCamera() print("Camera initialisee, attente du premier frame...") time.sleep(3) # ── Flask ────────────────────────────────────────────────────────────────── app = Flask(__name__) def generate_frames(): while True: frame = camera.get_frame() if frame: yield (b'--frame\r\nContent-Type: image/jpeg\r\n\r\n' + frame + b'\r\n') time.sleep(0.067) @app.route('/video_feed') def video_feed(): return Response(generate_frames(), mimetype='multipart/x-mixed-replace; boundary=frame') @app.route('/api/person') def api_person(): return jsonify(recognition_status) @app.route('/') def index(): return '' if __name__ == '__main__': print("Stream sur http://192.168.68.123:8080/video_feed") app.run(host='0.0.0.0', port=8080, debug=False, threaded=True)