File size: 4,058 Bytes
f4a2fda
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
import cv2
import dlib
import numpy as np
import base64
from fastapi import FastAPI, WebSocket
from scipy.spatial import distance as dist
from imutils import face_utils
import uvicorn
import json

app = FastAPI()

import os

# --- CONFIGURATION FOR HUGGING FACE ---
# Try multiple paths for the model file
possible_paths = [
    "shape_predictor_68_face_landmarks.dat",
    "data/models/shape_predictor_68_face_landmarks.dat",
    "/app/shape_predictor_68_face_landmarks.dat",
    "/app/data/models/shape_predictor_68_face_landmarks.dat"
]

predictor_path = None
for path in possible_paths:
    if os.path.exists(path):
        predictor_path = path
        break

if not predictor_path:
    print("[ERROR] shape_predictor_68_face_landmarks.dat not found in any of the searched paths.")
    print(f"Current working directory: {os.getcwd()}")
    print(f"Files in current directory: {os.listdir('.')}")
else:
    print(f"[INFO] Using model at: {predictor_path}")

print("[INFO] Loading AI models...")
detector = dlib.get_frontal_face_detector()
try:
    if predictor_path:
        predictor = dlib.shape_predictor(predictor_path)
        print("[SUCCESS] Model loaded!")
    else:
        print("[ERROR] Predictor path is None, cannot load model.")
except Exception as e:
    print(f"[ERROR] Failed to load model: {e}")

# Landmark indexes
(lStart, lEnd) = face_utils.FACIAL_LANDMARKS_IDXS["left_eye"]
(rStart, rEnd) = face_utils.FACIAL_LANDMARKS_IDXS["right_eye"]
(mStart, mEnd) = face_utils.FACIAL_LANDMARKS_IDXS["mouth"]
(nStart, nEnd) = face_utils.FACIAL_LANDMARKS_IDXS["nose"]

def get_ear(eye):
    a = dist.euclidean(eye[1], eye[5])
    b = dist.euclidean(eye[2], eye[4])
    c = dist.euclidean(eye[0], eye[3])
    return (a + b) / (2.0 * c)

def get_mar(mouth):
    a = dist.euclidean(mouth[13], mouth[19])
    b = dist.euclidean(mouth[14], mouth[18])
    c = dist.euclidean(mouth[15], mouth[17])
    d = dist.euclidean(mouth[12], mouth[16])
    return (a + b + c) / (2.0 * d + 1e-6)

@app.websocket("/ws/analyze")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()
    counter = 0
    print("[INFO] Mobile App connected!")
    
    try:
        while True:
            data = await websocket.receive_text()
            header, encoded = data.split(",", 1)
            nparr = np.frombuffer(base64.b64decode(encoded), np.uint8)
            frame = cv2.imdecode(nparr, cv2.IMREAD_COLOR)

            if frame is None:
                continue

            gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)
            faces = detector(gray, 0)
            
            response = {
                "drowsy": False,
                "distracted": False,
                "yawning": False,
                "ear": 0,
                "mar": 0
            }

            for rect in faces:
                shape = predictor(gray, rect)
                shape = face_utils.shape_to_np(shape)

                # Detection Logic
                ear = (get_ear(shape[lStart:lEnd]) + get_ear(shape[rStart:rEnd])) / 2.0
                mar = get_mar(shape[mStart:mEnd])
                
                nose_center = shape[nStart:nEnd].mean(axis=0)
                dist_left = dist.euclidean(shape[lStart:lEnd].mean(axis=0), nose_center)
                dist_right = dist.euclidean(shape[rStart:rEnd].mean(axis=0), nose_center)
                gaze_ratio = dist_left / (dist_right + 1e-6)

                response["ear"] = round(ear, 3)
                response["mar"] = round(mar, 3)
                
                if ear < 0.22: counter += 1
                else: counter = 0

                if counter >= 5: response["drowsy"] = True
                if gaze_ratio < 0.7 or gaze_ratio > 1.3: response["distracted"] = True
                if mar > 0.6: response["yawning"] = True

            await websocket.send_json(response)

    except Exception as e:
        print(f"[ERROR] Connection closed: {e}")

if __name__ == "__main__":
    # Port 7860 is required for Hugging Face Spaces
    uvicorn.run(app, host="0.0.0.0", port=7860)