eye_guard_2020.py

"""
EyeGuard 20-20-20: Robust Eye Rest Enforcement System
=======================================================

Enforces the 20-20-20 rule for eye health:
  - Every 20 minutes of screen time
  - Look at something 20 feet away
  - For at least 20 seconds

Anti-spoofing features:
  - BOTH eyes must be closed for the full 20 seconds
  - One-eye-closed tricks are detected and rejected
  - Face must remain visible throughout the rest period
  - Blink detection prevents counting short blinks as rest

Uses MediaPipe Face Mesh for precise per-eye landmark tracking,
with EAR (Eye Aspect Ratio) geometric calculation for each eye independently.
"""

import cv2
import mediapipe as mp
import numpy as np
import time
from dataclasses import dataclass, field
from enum import Enum, auto
from typing import Optional, Tuple, List
import warnings
warnings.filterwarnings("ignore")

# ---------------------------------------------------------------------------
# Constants
# ---------------------------------------------------------------------------

# MediaPipe Face Mesh eye landmark indices
LEFT_EYE_IDX = [362, 385, 387, 263, 373, 380]
RIGHT_EYE_IDX = [33, 160, 158, 133, 153, 144]

# EAR thresholds (tuned for typical webcam distances)
EAR_OPEN_THRESHOLD = 0.22
EAR_CLOSED_THRESHOLD = 0.18

# Timing constants (20-20-20 rule)
SCREEN_TIME_ALERT_SECONDS = 20 * 60
REST_DURATION_SECONDS = 20
MIN_CONSECUTIVE_CLOSED_FRAMES = 15
BLINK_MAX_DURATION_FRAMES = 10


class Status(Enum):
    SCREENING = auto()
    ALERT = auto()
    RESTING = auto()
    REST_COMPLETE = auto()
    SPOOFING = auto()
    NO_FACE = auto()


# ---------------------------------------------------------------------------
# Geometry helpers
# ---------------------------------------------------------------------------

def euclidean(p1, p2) -> float:
    return np.sqrt((p1[0] - p2[0]) ** 2 + (p1[1] - p2[1]) ** 2)


def eye_aspect_ratio(landmarks, eye_indices, img_w, img_h) -> float:
    pts = [(landmarks[i].x * img_w, landmarks[i].y * img_h) for i in eye_indices]
    v1 = euclidean(pts[1], pts[5])
    v2 = euclidean(pts[2], pts[4])
    h = euclidean(pts[0], pts[3])
    if h < 1e-6:
        return 0.0
    return (v1 + v2) / (2.0 * h)


# ---------------------------------------------------------------------------
# Tracker state
# ---------------------------------------------------------------------------

@dataclass
class EyeState:
    ear: float = 1.0
    is_closed: bool = False
    closed_start: Optional[float] = None
    closed_frames: int = 0


@dataclass
class TrackerState:
    status: Status = Status.SCREENING
    screen_timer_start: float = field(default_factory=time.time)
    rest_start_time: Optional[float] = None
    rest_completed_time: Optional[float] = None
    last_face_seen: float = field(default_factory=time.time)
    left_eye: EyeState = field(default_factory=EyeState)
    right_eye: EyeState = field(default_factory=EyeState)
    ear_history_left: List[float] = field(default_factory=list)
    ear_history_right: List[float] = field(default_factory=list)
    history_maxlen: int = 5
    spoofing_detected_at: Optional[float] = None
    spoofing_count: int = 0
    blink_count: int = 0
    last_blink_end: float = 0.0
    total_rests_completed: int = 0
    total_rest_seconds: float = 0.0
    failed_rests: int = 0
    fps: float = 30.0
    frame_times: List[float] = field(default_factory=list)


def update_eye_state(eye, ear, threshold_open, threshold_closed):
    eye.ear = ear
    if eye.is_closed:
        if ear > threshold_open:
            eye.is_closed = False
            eye.closed_start = None
            eye.closed_frames = 0
    else:
        if ear < threshold_closed:
            eye.is_closed = True
            eye.closed_start = time.time()
            eye.closed_frames = 1
        else:
            eye.closed_frames = 0
    if eye.is_closed:
        eye.closed_frames += 1
    return eye


# ---------------------------------------------------------------------------
# Main Tracker
# ---------------------------------------------------------------------------

class EyeGuard2020:
    def __init__(self, ear_open=EAR_OPEN_THRESHOLD, ear_closed=EAR_CLOSED_THRESHOLD):
        self.ear_open = ear_open
        self.ear_closed = ear_closed
        self.state = TrackerState()
        self.mp_face_mesh = mp.solutions.face_mesh
        self.face_mesh = self.mp_face_mesh.FaceMesh(
            max_num_faces=1, refine_landmarks=True,
            min_detection_confidence=0.5, min_tracking_confidence=0.5
        )
        self.mp_drawing = mp.solutions.drawing_utils
        self.mp_drawing_styles = mp.solutions.drawing_styles
        self.running = False

    def _smooth_ear(self, history, new_val):
        history.append(new_val)
        if len(history) > self.state.history_maxlen:
            history.pop(0)
        return np.median(history)

    def process_frame(self, frame: np.ndarray) -> Tuple[np.ndarray, TrackerState]:
        h, w = frame.shape[:2]
        now = time.time()
        self.state.frame_times.append(now)
        if len(self.state.frame_times) > 30:
            self.state.frame_times.pop(0)
        if len(self.state.frame_times) > 1:
            self.state.fps = len(self.state.frame_times) / (self.state.frame_times[-1] - self.state.frame_times[0])

        rgb = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
        rgb.flags.writeable = False
        results = self.face_mesh.process(rgb)
        rgb.flags.writeable = True
        face_visible = results.multi_face_landmarks is not None

        if not face_visible:
            self.state.last_face_seen = now
            if self.state.status in (Status.RESTING, Status.SPOOFING):
                if self.state.status == Status.RESTING:
                    self.state.failed_rests += 1
                self.state.status = Status.NO_FACE
                self.state.rest_start_time = None
            elif self.state.status == Status.ALERT:
                pass
            else:
                self.state.status = Status.NO_FACE
            return self._draw_overlay(frame), self.state

        self.state.last_face_seen = now
        landmarks = results.multi_face_landmarks[0].landmark

        ear_left_raw = eye_aspect_ratio(landmarks, LEFT_EYE_IDX, w, h)
        ear_right_raw = eye_aspect_ratio(landmarks, RIGHT_EYE_IDX, w, h)
        ear_left = self._smooth_ear(self.state.ear_history_left, ear_left_raw)
        ear_right = self._smooth_ear(self.state.ear_history_right, ear_right_raw)

        self.state.left_eye = update_eye_state(self.state.left_eye, ear_left, self.ear_open, self.ear_closed)
        self.state.right_eye = update_eye_state(self.state.right_eye, ear_right, self.ear_open, self.ear_closed)

        self._draw_eyes(frame, landmarks, w, h, ear_left, ear_right)

        screen_elapsed = now - self.state.screen_timer_start
        both_eyes_closed = self.state.left_eye.is_closed and self.state.right_eye.is_closed
        one_eye_closed = (self.state.left_eye.is_closed and not self.state.right_eye.is_closed) or \
                         (not self.state.left_eye.is_closed and self.state.right_eye.is_closed)

        if self.state.status == Status.RESTING and one_eye_closed:
            self.state.spoofing_detected_at = now
            self.state.spoofing_count += 1
            self.state.status = Status.SPOOFING
            self.state.failed_rests += 1
            self.state.rest_start_time = None
            return self._draw_overlay(frame), self.state

        if self.state.status == Status.ALERT and one_eye_closed:
            self.state.spoofing_detected_at = now
            self.state.spoofing_count += 1

        if self.state.status == Status.SCREENING:
            if screen_elapsed >= SCREEN_TIME_ALERT_SECONDS:
                self.state.status = Status.ALERT

        elif self.state.status == Status.ALERT:
            if both_eyes_closed and self.state.left_eye.closed_frames >= MIN_CONSECUTIVE_CLOSED_FRAMES:
                self.state.status = Status.RESTING
                self.state.rest_start_time = now

        elif self.state.status == Status.RESTING:
            if not both_eyes_closed:
                self.state.failed_rests += 1
                self.state.status = Status.ALERT
                self.state.rest_start_time = None
            else:
                rest_elapsed = now - self.state.rest_start_time
                if rest_elapsed >= REST_DURATION_SECONDS:
                    self.state.status = Status.REST_COMPLETE
                    self.state.rest_completed_time = now
                    self.state.total_rests_completed += 1
                    self.state.total_rest_seconds += rest_elapsed
                    self.state.screen_timer_start = now

        elif self.state.status == Status.REST_COMPLETE:
            if now - self.state.rest_completed_time > 3.0:
                self.state.status = Status.SCREENING

        elif self.state.status == Status.SPOOFING:
            if now - self.state.spoofing_detected_at > 3.0:
                self.state.status = Status.ALERT

        elif self.state.status == Status.NO_FACE:
            if face_visible:
                if screen_elapsed >= SCREEN_TIME_ALERT_SECONDS:
                    self.state.status = Status.ALERT
                else:
                    self.state.status = Status.SCREENING

        return self._draw_overlay(frame), self.state

    def _draw_eyes(self, frame, landmarks, w, h, ear_left, ear_right):
        left_pts = np.array([
            [int(landmarks[i].x * w), int(landmarks[i].y * h)] for i in LEFT_EYE_IDX
        ], np.int32)
        color_left = (0, 255, 0) if not self.state.left_eye.is_closed else (0, 0, 255)
        cv2.polylines(frame, [left_pts], True, color_left, 2)
        cv2.putText(frame, f"L-EAR: {ear_left:.3f}", (left_pts[0][0] - 30, left_pts[0][1] - 10),
                    cv2.FONT_HERSHEY_SIMPLEX, 0.5, color_left, 2)

        right_pts = np.array([
            [int(landmarks[i].x * w), int(landmarks[i].y * h)] for i in RIGHT_EYE_IDX
        ], np.int32)
        color_right = (0, 255, 0) if not self.state.right_eye.is_closed else (0, 0, 255)
        cv2.polylines(frame, [right_pts], True, color_right, 2)
        cv2.putText(frame, f"R-EAR: {ear_right:.3f}", (right_pts[0][0] - 30, right_pts[0][1] - 10),
                    cv2.FONT_HERSHEY_SIMPLEX, 0.5, color_right, 2)

    def _draw_overlay(self, frame):
        h, w = frame.shape[:2]
        now = time.time()
        overlay = frame.copy()
        cv2.rectangle(overlay, (0, 0), (w, 90), (0, 0, 0), -1)
        frame = cv2.addWeighted(overlay, 0.6, frame, 0.4, 0)

        cv2.putText(frame, "EyeGuard 20-20-20", (10, 30),
                    cv2.FONT_HERSHEY_SIMPLEX, 0.9, (0, 255, 255), 2)
        cv2.putText(frame, f"FPS: {self.state.fps:.1f}", (w - 120, 30),
                    cv2.FONT_HERSHEY_SIMPLEX, 0.6, (255, 255, 255), 1)

        screen_elapsed = now - self.state.screen_timer_start
        screen_left = max(0, SCREEN_TIME_ALERT_SECONDS - screen_elapsed)
        mins, secs = divmod(int(screen_left), 60)
        timer_color = (0, 255, 0) if screen_left > 60 else (0, 165, 255) if screen_left > 30 else (0, 0, 255)
        cv2.putText(frame, f"Next break in: {mins:02d}:{secs:02d}", (10, 60),
                    cv2.FONT_HERSHEY_SIMPLEX, 0.7, timer_color, 2)

        banner_y = h - 80
        if self.state.status == Status.SCREENING:
            msg, color = "Keep working! Eyes healthy.", (0, 255, 0)
        elif self.state.status == Status.ALERT:
            msg, color = "TIME FOR A BREAK! Close BOTH eyes for 20 seconds", (0, 0, 255)
            if int(now * 2) % 2 == 0:
                cv2.rectangle(frame, (0, banner_y - 10), (w, banner_y + 40), (0, 0, 255), -1)
        elif self.state.status == Status.RESTING:
            rest_elapsed = now - self.state.rest_start_time
            rest_left = max(0, REST_DURATION_SECONDS - rest_elapsed)
            msg, color = f"RESTING... Keep both eyes closed: {rest_left:.1f}s left", (255, 255, 0)
            cv2.rectangle(frame, (0, banner_y - 10), (w, banner_y + 40), (255, 255, 0), -1)
        elif self.state.status == Status.REST_COMPLETE:
            msg, color = "Great job! Break complete. Back to work!", (0, 255, 0)
            cv2.rectangle(frame, (0, banner_y - 10), (w, banner_y + 40), (0, 255, 0), -1)
        elif self.state.status == Status.SPOOFING:
            msg, color = "SPOOFING DETECTED! One eye closed trick won't work!", (0, 0, 255)
            cv2.rectangle(frame, (0, banner_y - 10), (w, banner_y + 40), (0, 0, 255), -1)
        elif self.state.status == Status.NO_FACE:
            msg, color = "No face detected. Please position yourself in front of camera.", (128, 128, 128)
        else:
            msg, color = "Unknown state", (128, 128, 128)

        text_size = cv2.getTextSize(msg, cv2.FONT_HERSHEY_SIMPLEX, 0.8, 2)[0]
        text_x = (w - text_size[0]) // 2
        bg_color = (0, 0, 0) if self.state.status in (
            Status.ALERT, Status.RESTING, Status.REST_COMPLETE, Status.SPOOFING) else None
        cv2.putText(frame, msg, (text_x, banner_y + 20),
                    cv2.FONT_HERSHEY_SIMPLEX, 0.8, (0, 0, 0) if bg_color else color, 2)

        eye_status_y = 110
        le_status = "CLOSED" if self.state.left_eye.is_closed else "OPEN"
        re_status = "CLOSED" if self.state.right_eye.is_closed else "OPEN"
        le_color = (0, 0, 255) if self.state.left_eye.is_closed else (0, 255, 0)
        re_color = (0, 0, 255) if self.state.right_eye.is_closed else (0, 255, 0)

        cv2.putText(frame, f"Left Eye:  {le_status}", (w - 220, eye_status_y),
                    cv2.FONT_HERSHEY_SIMPLEX, 0.6, le_color, 2)
        cv2.putText(frame, f"Right Eye: {re_status}", (w - 220, eye_status_y + 25),
                    cv2.FONT_HERSHEY_SIMPLEX, 0.6, re_color, 2)

        stats_y = h - 10
        cv2.putText(frame,
                    f"Completed: {self.state.total_rests_completed} | Failed: {self.state.failed_rests} | Spoofs: {self.state.spoofing_count}",
                    (10, stats_y), cv2.FONT_HERSHEY_SIMPLEX, 0.5, (255, 255, 255), 1)

        return frame

    def run(self, source=0):
        cap = cv2.VideoCapture(source)
        if not cap.isOpened():
            print(f"Failed to open video source: {source}")
            return

        print("=" * 60)
        print("EyeGuard 20-20-20 Started!")
        print("- Every 20 minutes, close BOTH eyes for 20 seconds")
        print("- One-eye tricks are detected and rejected!")
        print("- Press 'q' to quit, 'r' to reset screen timer, 't' for test mode")
        print("=" * 60)

        self.running = True
        try:
            while self.running:
                ret, frame = cap.read()
                if not ret:
                    break
                frame = cv2.flip(frame, 1)
                annotated, state = self.process_frame(frame)
                cv2.imshow("EyeGuard 20-20-20", annotated)
                key = cv2.waitKey(1) & 0xFF
                if key == ord('q'):
                    break
                elif key == ord('r'):
                    self.state.screen_timer_start = time.time()
                    print("Timer reset!")
                elif key == ord('t'):
                    self.state.screen_timer_start = time.time() - SCREEN_TIME_ALERT_SECONDS - 1
                    print("Test mode: forcing alert!")
        finally:
            self.running = False
            cap.release()
            cv2.destroyAllWindows()
            self.face_mesh.close()
            self._print_summary()

    def _print_summary(self):
        print("\n" + "=" * 60)
        print("SESSION SUMMARY")
        print("=" * 60)
        print(f"Total rests completed: {self.state.total_rests_completed}")
        print(f"Total rest time:       {self.state.total_rest_seconds:.1f} seconds")
        print(f"Failed rests:            {self.state.failed_rests}")
        print(f"Spoofing attempts:       {self.state.spoofing_count}")
        print("=" * 60)


if __name__ == "__main__":
    import argparse
    parser = argparse.ArgumentParser(description="EyeGuard 20-20-20 Rule Enforcer")
    parser.add_argument("--camera", type=int, default=0)
    parser.add_argument("--ear-open", type=float, default=EAR_OPEN_THRESHOLD)
    parser.add_argument("--ear-closed", type=float, default=EAR_CLOSED_THRESHOLD)
    args = parser.parse_args()
    guard = EyeGuard2020(ear_open=args.ear_open, ear_closed=args.ear_closed)
    guard.run(source=args.camera)