import os import cv2 import time import subprocess import pickle import dlib import face_recognition import subprocess import sys import numpy as np from flask import Flask, render_template, Response, request, redirect, url_for, session, flash from functools import wraps from db_connector import log_recognition, get_recognition_logs from config import FLASK_SECRET_KEY from cryptography.fernet import Fernet app = Flask(__name__) app.secret_key = FLASK_SECRET_KEY def login_required(f): @wraps(f) def decorated_function(*args, **kwargs): if not session.get("logged_in"): flash("You need to be logged in to access that page.", "warning") return redirect(url_for("login", next=request.url)) return f(*args, **kwargs) return decorated_function @app.route("/login", methods=["GET", "POST"]) def login(): error = None if request.method == "POST": username = request.form.get("username") password = request.form.get("password") if username == "admin" and password == "secret": session["logged_in"] = True session["username"] = username flash("You were successfully logged in.", "success") return redirect(url_for("index")) else: error = "Invalid credentials. Please try again." return render_template("login.html", error=error) @app.route("/logout") def logout(): session.clear() flash("You have been logged out.", "info") return redirect(url_for("login")) @app.route("/add_face") @login_required def add_face(): return render_template("add_face.html") @app.route("/run_face", methods=["POST"]) @login_required def run_face(): face_name = request.form.get("face_name") if not face_name: flash("Please provide a name.", "danger") return redirect(url_for("add_face")) script = "enroll_no_gui.py" try: venv_python = "./biometric_system/bin/python3.10" result = subprocess.run([venv_python, script, face_name, "--auto"], capture_output=True, text=True, timeout=300) output = result.stdout if result.stdout else result.stderr except Exception as e: output = str(e) return render_template("result.html", output=output) @app.route("/run_face_live") @login_required def run_face_live(): face_name = request.args.get("face_name", "") venv_python = "/home/valer/recognition_system/src/biometric_system/bin/python3.10" def generate(): env = os.environ.copy() env["PYTHONUNBUFFERED"] = "1" try: subprocess.run(["sudo", "systemctl", "stop", "face_rec"], check=True) print("✅ face_rec_service stopped") yield "data: [face_rec_service stopped]\n\n" except subprocess.CalledProcessError as e: error_msg = f"[ERROR] Could not stop service: {e}" print(error_msg) yield f"data: {error_msg}\n\n" yield "data: [Aborting enrollment.]\n\n" return try: process = subprocess.Popen( [sys.executable, "enroll_no_gui.py", face_name, "--auto"], stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, bufsize=1, env=env ) for line in iter(process.stdout.readline, ''): print(line, end='') yield f"data: {line}\n\n" process.stdout.close() process.wait() except Exception as e: yield f"data: [ERROR during enrollment: {e}]\n\n" print(f"[ERROR] during subprocess: {e}") try: subprocess.run(["sudo", "systemctl", "start", "face_rec"], check=True) print("✅ face_rec_service started") yield "data: [face_rec_service started]\n\n" except subprocess.CalledProcessError as e: error_msg = f"[ERROR] Could not start service: {e}" print(error_msg) yield f"data: {error_msg}\n\n" return Response(generate(), mimetype='text/event-stream') @app.route("/delete_person") @login_required def delete_person(): return render_template("delete_person.html") @app.route("/run_delete_person", methods=["POST"]) @login_required def run_delete_person(): person_name = request.form.get("person_name") if not person_name: flash("Please provide a name.", "danger") return redirect(url_for("delete_person")) script = "remove_face.py" try: venv_python = "./biometric_system/bin/python3.10" result = subprocess.run([venv_python, script, person_name], capture_output=True, text=True, timeout=60) output = result.stdout if result.stdout else result.stderr except Exception as e: output = str(e) return render_template("result.html", output=output) @app.route("/list_people") @login_required def list_people(): key = os.getenv("ENCODINGS_KEY") fernet = Fernet(key) with open("encodings.pickle", "rb") as enc_file: encrypted_data = enc_file.read() try: decrypted_data = fernet.decrypt(encrypted_data) data = pickle.loads(decrypted_data) names = data.get("names", []) unique_names = sorted(set(names)) except Exception as e: print(f"[ERROR] Failed to decrypt or unpickle: {e}") names = [] #try: # with open("encodings.pickle", "rb") as f: # data = pickle.load(f) # names = data.get("names", []) #except Exception as e: # print(f"[ERROR] {e}") # names = [] #unique_names = sorted(set(names)) #print(unique_names) return render_template("list_people.html", people=unique_names) @app.route("/logs") @login_required def logs(): recognition_logs = get_recognition_logs() return render_template("logs.html", logs=recognition_logs) def gen_frames_raw(): cap = cv2.VideoCapture(4) if not cap.isOpened(): print("Error: Could not open camera.") return while True: ret, frame = cap.read() if not ret: continue # OPTIONAL # rgb_frame = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB) # face_locations = face_recognition.face_locations(rgb_frame) # for (top, right, bottom, left) in face_locations: # cv2.rectangle(frame, (left, top), (right, bottom), (0, 255, 0), 2) ret, buffer = cv2.imencode('.jpg', frame) if not ret: continue frame_bytes = buffer.tobytes() yield (b'--frame\r\n' b'Content-Type: image/jpeg\r\n\r\n' + frame_bytes + b'\r\n') time.sleep(0.1) cap.release() @app.route("/") @login_required def index(): logs = get_recognition_logs() return render_template("index.html", logs=logs) @app.route("/video_feed") @login_required def video_feed(): return Response(gen_frames_raw(), mimetype="multipart/x-mixed-replace; boundary=frame") @app.route("/open_door", methods=["GET", "POST"]) @login_required def open_door(): try: result = subprocess.run(["python3", "open_door.py"], capture_output=True, text=True, timeout=60, check=True) output = result.stdout if result.stdout else result.stderr flash("Door opened successfully!", "success") except Exception as e: output = str(e) flash("Door open failed!", "danger") return render_template("result.html", output=output) if __name__ == "__main__": app.run(host="0.0.0.0", port=5000, debug=False)