updated to DNN for face detections
This commit is contained in:
+20
-6
@@ -127,6 +127,8 @@ class Camera:
|
||||
if ii not in seen:
|
||||
seen.add(ii)
|
||||
ordered.append(ii)
|
||||
self._requested_indices = ordered
|
||||
self._probe_timeout = float(config.get("camera_probe_timeout_seconds", 2.5))
|
||||
|
||||
# Camera obstruction detection settings
|
||||
obs = config.get("camera_obscured_detection", {}) or {}
|
||||
@@ -134,20 +136,32 @@ class Camera:
|
||||
self.brightness_threshold = float(obs.get("brightness_threshold", 35))
|
||||
self.variance_threshold = float(obs.get("variance_threshold", 10))
|
||||
|
||||
probe_timeout = float(config.get("camera_probe_timeout_seconds", 2.5))
|
||||
|
||||
self.streams: List[CameraStream] = []
|
||||
for idx in ordered:
|
||||
self._open_streams()
|
||||
|
||||
def _open_streams(self) -> None:
|
||||
"""Internal: (re)probe and open the originally requested camera indices."""
|
||||
self.streams = []
|
||||
for idx in self._requested_indices:
|
||||
stream = CameraStream(idx)
|
||||
if stream.open(probe_timeout=probe_timeout):
|
||||
if stream.open(probe_timeout=self._probe_timeout):
|
||||
print(f"[Camera] index {idx} OPEN")
|
||||
self.streams.append(stream)
|
||||
else:
|
||||
print(f"[Camera] index {idx} DEAD (skipped after {probe_timeout}s probe)")
|
||||
# keep the dead stream object out of the list entirely
|
||||
print(f"[Camera] index {idx} DEAD (skipped after {self._probe_timeout}s probe)")
|
||||
if not self.alive_streams:
|
||||
print("[Camera] WARNING: no working cameras detected")
|
||||
|
||||
def reopen(self) -> None:
|
||||
"""Re-acquire camera streams after a previous release(). Safe to call multiple times."""
|
||||
# Close any lingering handles first
|
||||
for s in self.streams:
|
||||
try:
|
||||
s.release()
|
||||
except Exception:
|
||||
pass
|
||||
self._open_streams()
|
||||
|
||||
# ---- discovery / accessors ----
|
||||
|
||||
@property
|
||||
|
||||
@@ -0,0 +1,323 @@
|
||||
"""DNN-based face recognition using the face_recognition library (dlib + FaceNet-style model).
|
||||
|
||||
This provides significantly higher accuracy than the LBPH backend, especially
|
||||
against look-alikes. It requires the `face-recognition` package (which depends
|
||||
on dlib).
|
||||
|
||||
Installation on Windows is non-trivial:
|
||||
- Install Visual Studio Build Tools + CMake first.
|
||||
- Then: pip install face-recognition
|
||||
|
||||
See README for detailed instructions.
|
||||
"""
|
||||
|
||||
from pathlib import Path
|
||||
from typing import Optional, Tuple, List
|
||||
import pickle
|
||||
import numpy as np
|
||||
|
||||
try:
|
||||
import face_recognition
|
||||
FACE_RECOGNITION_AVAILABLE = True
|
||||
except ImportError:
|
||||
FACE_RECOGNITION_AVAILABLE = False
|
||||
|
||||
class DNNFaceRecognizer:
|
||||
"""Face recognition using deep neural network embeddings (FaceNet style).
|
||||
|
||||
Much more robust than LBPH for distinguishing similar-looking people.
|
||||
"""
|
||||
|
||||
def __init__(self, tolerance: float = 0.6, allowed_labels: Optional[List[str]] = None, faces_dir: str = "faces"):
|
||||
self.tolerance = float(tolerance)
|
||||
self.allowed_labels = set(allowed_labels) if allowed_labels else None
|
||||
self.faces_dir = Path(faces_dir)
|
||||
self.encodings_file = self.faces_dir / "face_encodings.pkl"
|
||||
self.labels_file = self.faces_dir / "dnn_labels.json"
|
||||
|
||||
self._known_encodings: list = []
|
||||
self._known_labels: list = []
|
||||
self._loaded = False
|
||||
self._load()
|
||||
|
||||
@property
|
||||
def available(self) -> bool:
|
||||
# Model must be loaded. dlib (face_recognition) itself only needs to be
|
||||
# importable in the isolated worker subprocess, not in this process.
|
||||
return self._loaded
|
||||
|
||||
def _load(self) -> None:
|
||||
if not self.encodings_file.exists() or not self.labels_file.exists():
|
||||
print("[DNNFaceRecognizer] No trained DNN model found. Run enrollment with DNN backend.")
|
||||
return
|
||||
|
||||
try:
|
||||
with open(self.encodings_file, "rb") as f:
|
||||
self._known_encodings = pickle.load(f)
|
||||
|
||||
import json
|
||||
with open(self.labels_file, "r", encoding="utf-8") as f:
|
||||
self._known_labels = json.load(f)
|
||||
|
||||
self._loaded = True
|
||||
print(f"[DNNFaceRecognizer] Loaded DNN model with {len(self._known_labels)} identities")
|
||||
except Exception as e:
|
||||
print(f"[DNNFaceRecognizer] Failed to load model: {e}")
|
||||
|
||||
def verify_frame(self, frame) -> Tuple[bool, Optional[str], Optional[float]]:
|
||||
"""Verify a frame using DNN embeddings.
|
||||
|
||||
Returns (matched, label, distance). Lower distance = better match.
|
||||
Extremely defensive to avoid crashing the Tkinter white screen thread.
|
||||
"""
|
||||
if not self.available or frame is None:
|
||||
return False, None, None
|
||||
|
||||
try:
|
||||
# Convert BGR to RGB
|
||||
rgb_frame = frame[:, :, ::-1]
|
||||
|
||||
# Multiple safety checks
|
||||
if rgb_frame is None or rgb_frame.size == 0:
|
||||
return False, None, None
|
||||
if len(rgb_frame.shape) != 3 or rgb_frame.shape[2] != 3:
|
||||
return False, None, None
|
||||
if rgb_frame.dtype != np.uint8:
|
||||
rgb_frame = rgb_frame.astype(np.uint8)
|
||||
|
||||
# Detect faces
|
||||
face_locations = face_recognition.face_locations(rgb_frame)
|
||||
if not face_locations:
|
||||
return False, None, None
|
||||
|
||||
face_encodings = face_recognition.face_encodings(rgb_frame, face_locations)
|
||||
if not face_encodings:
|
||||
return False, None, None
|
||||
|
||||
best_label = None
|
||||
best_distance = None
|
||||
|
||||
for encoding in face_encodings:
|
||||
try:
|
||||
distances = face_recognition.face_distance(self._known_encodings, encoding)
|
||||
if len(distances) == 0:
|
||||
continue
|
||||
|
||||
min_distance_idx = int(np.argmin(distances))
|
||||
min_distance = float(distances[min_distance_idx])
|
||||
label = self._known_labels[min_distance_idx]
|
||||
|
||||
if self.allowed_labels and label not in self.allowed_labels:
|
||||
continue
|
||||
|
||||
if best_distance is None or min_distance < best_distance:
|
||||
best_distance = min_distance
|
||||
best_label = label
|
||||
except Exception as inner_e:
|
||||
print(f"[DNNFaceRecognizer] inner distance error: {inner_e}")
|
||||
continue
|
||||
|
||||
if best_label is None or best_distance is None:
|
||||
return False, None, None
|
||||
|
||||
matched = best_distance <= self.tolerance
|
||||
return matched, best_label, best_distance
|
||||
|
||||
except Exception as e:
|
||||
print(f"[DNNFaceRecognizer] verify_frame error: {type(e).__name__}: {e}")
|
||||
import traceback
|
||||
traceback.print_exc(limit=3)
|
||||
return False, None, None
|
||||
|
||||
def verify_from_camera(self, camera, attempts: int = 3) -> Tuple[bool, Optional[str], Optional[float]]:
|
||||
"""Capture several frames from ALL alive cameras and verify them in ONE
|
||||
isolated subprocess.
|
||||
|
||||
Frames are captured here in the (safe) calling process, then the dlib
|
||||
matching is delegated to a single subprocess so that (a) a native dlib
|
||||
crash can never take down the main ICEYOU app, and (b) the model is
|
||||
loaded only once per attempt (fast -> button re-enables quickly).
|
||||
|
||||
IMPORTANT: we sample EVERY working camera (not just the primary). On
|
||||
multi-cam rigs the user-facing camera is often not index 0, so reading
|
||||
only the primary stream caused face unlock to never see the user.
|
||||
"""
|
||||
if not self.available:
|
||||
return False, None, None
|
||||
|
||||
import cv2
|
||||
import tempfile
|
||||
import os
|
||||
|
||||
rounds = max(1, attempts)
|
||||
tmp_paths: list = []
|
||||
try:
|
||||
captured = 0
|
||||
|
||||
def _save(frame) -> None:
|
||||
nonlocal captured
|
||||
if frame is None:
|
||||
return
|
||||
fd, tmp_path = tempfile.mkstemp(suffix=".png", prefix="iceyou_dnn_")
|
||||
os.close(fd)
|
||||
if cv2.imwrite(tmp_path, frame):
|
||||
tmp_paths.append(tmp_path)
|
||||
captured += 1
|
||||
else:
|
||||
try:
|
||||
os.remove(tmp_path)
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
# Prefer per-camera capture so we sample every working stream.
|
||||
read_all = getattr(camera, "read_all_frames", None)
|
||||
for _ in range(rounds):
|
||||
if callable(read_all):
|
||||
frames = read_all() or {}
|
||||
for _idx, frame in frames.items():
|
||||
_save(frame)
|
||||
else:
|
||||
_save(camera.read_frame())
|
||||
|
||||
cam_count = 0
|
||||
try:
|
||||
cam_count = len(getattr(camera, "alive_streams", []) or [])
|
||||
except Exception:
|
||||
cam_count = 0
|
||||
print(f"[DNNFaceRecognizer] Captured {captured} frame(s) across {cam_count or 1} camera(s) x {rounds} round(s) for face unlock")
|
||||
if not tmp_paths:
|
||||
print("[DNNFaceRecognizer] No usable camera frames; face unlock will report no match")
|
||||
return False, None, None
|
||||
return self._verify_images_subprocess(tmp_paths)
|
||||
finally:
|
||||
for p in tmp_paths:
|
||||
try:
|
||||
os.remove(p)
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
def _verify_images_subprocess(self, image_paths: list) -> Tuple[bool, Optional[str], Optional[float]]:
|
||||
"""Run dlib matching over the given images in a single subprocess.
|
||||
|
||||
Returns (matched, label, distance). On ANY failure (including a native
|
||||
subprocess crash / non-zero exit) returns a safe no-match instead of
|
||||
propagating, so the parent process never dies.
|
||||
"""
|
||||
import subprocess
|
||||
import os
|
||||
import sys
|
||||
|
||||
try:
|
||||
allowed_csv = ",".join(sorted(self.allowed_labels)) if self.allowed_labels else ""
|
||||
print(f"[DNNFaceRecognizer] Launching isolated DNN worker for {len(image_paths)} frame(s) (face recognition in subprocess)...")
|
||||
|
||||
cmd = [
|
||||
sys.executable,
|
||||
"-m", "iceyou.dnn_verify_worker",
|
||||
"--encodings", str(self.encodings_file),
|
||||
"--labels", str(self.labels_file),
|
||||
"--tolerance", str(self.tolerance),
|
||||
]
|
||||
if allowed_csv:
|
||||
cmd += ["--allowed", allowed_csv]
|
||||
cmd += list(image_paths)
|
||||
|
||||
# Ensure the worker can import the iceyou package
|
||||
env = dict(os.environ)
|
||||
src_dir = str(Path(__file__).resolve().parent.parent)
|
||||
existing = env.get("PYTHONPATH", "")
|
||||
env["PYTHONPATH"] = src_dir + (os.pathsep + existing if existing else "")
|
||||
|
||||
# Suppress the console window flash on Windows
|
||||
creationflags = 0
|
||||
if os.name == "nt":
|
||||
creationflags = getattr(subprocess, "CREATE_NO_WINDOW", 0)
|
||||
|
||||
proc = subprocess.run(
|
||||
cmd,
|
||||
capture_output=True,
|
||||
text=True,
|
||||
timeout=30,
|
||||
env=env,
|
||||
creationflags=creationflags,
|
||||
)
|
||||
|
||||
if proc.returncode != 0:
|
||||
# Native crash / segfault in dlib -> isolated, parent survives
|
||||
print(f"[DNNFaceRecognizer] worker exited with code {proc.returncode} (isolated, app safe)")
|
||||
if proc.stderr:
|
||||
print(f"[DNNFaceRecognizer] worker stderr: {proc.stderr.strip()[:300]}")
|
||||
return False, None, None
|
||||
|
||||
# Always surface worker diagnostics (face counts, closest distance)
|
||||
if proc.stderr and proc.stderr.strip():
|
||||
for line in proc.stderr.strip().splitlines():
|
||||
print(f"[DNNFaceRecognizer] {line.strip()}")
|
||||
|
||||
out = (proc.stdout or "").strip().splitlines()
|
||||
if not out:
|
||||
return False, None, None
|
||||
|
||||
import json
|
||||
result = json.loads(out[-1])
|
||||
matched = bool(result.get("matched", False))
|
||||
label = result.get("label")
|
||||
distance = result.get("distance")
|
||||
if matched:
|
||||
print(f"[DNNFaceRecognizer] MATCH: label={label} distance={distance:.3f} (tolerance={self.tolerance:.3f})")
|
||||
elif distance is not None:
|
||||
print(f"[DNNFaceRecognizer] Worker result: best_distance={distance:.3f} (tolerance={self.tolerance:.3f}) — no match")
|
||||
else:
|
||||
print("[DNNFaceRecognizer] Worker result: no allowed face match (see diagnostics above)")
|
||||
return matched, label, distance
|
||||
|
||||
except subprocess.TimeoutExpired:
|
||||
print("[DNNFaceRecognizer] worker timed out (isolated, app safe)")
|
||||
return False, None, None
|
||||
except Exception as e:
|
||||
print(f"[DNNFaceRecognizer] subprocess verify error: {type(e).__name__}: {e}")
|
||||
return False, None, None
|
||||
|
||||
|
||||
def train_dnn_model(faces_dir: str = "faces") -> bool:
|
||||
"""Train a DNN model from images in faces/<label>/ directories.
|
||||
This is a helper function for future enrollment script updates.
|
||||
"""
|
||||
if not FACE_RECOGNITION_AVAILABLE:
|
||||
print("face_recognition not available. Cannot train DNN model.")
|
||||
return False
|
||||
|
||||
faces_path = Path(faces_dir)
|
||||
encodings = []
|
||||
labels = []
|
||||
|
||||
for label_dir in faces_path.iterdir():
|
||||
if not label_dir.is_dir():
|
||||
continue
|
||||
label = label_dir.name
|
||||
for img_path in label_dir.glob("*.png"):
|
||||
try:
|
||||
image = face_recognition.load_image_file(str(img_path))
|
||||
face_encodings = face_recognition.face_encodings(image)
|
||||
if face_encodings:
|
||||
encodings.append(face_encodings[0])
|
||||
labels.append(label)
|
||||
except Exception as e:
|
||||
print(f"Failed to process {img_path}: {e}")
|
||||
|
||||
if not encodings:
|
||||
print("No valid face encodings found.")
|
||||
return False
|
||||
|
||||
encodings_file = faces_path / "face_encodings.pkl"
|
||||
labels_file = faces_path / "dnn_labels.json"
|
||||
|
||||
with open(encodings_file, "wb") as f:
|
||||
pickle.dump(encodings, f)
|
||||
|
||||
import json
|
||||
with open(labels_file, "w", encoding="utf-8") as f:
|
||||
json.dump(labels, f, indent=2)
|
||||
|
||||
print(f"Trained DNN model with {len(labels)} images across {len(set(labels))} identities.")
|
||||
return True
|
||||
@@ -0,0 +1,143 @@
|
||||
"""Standalone DNN face-verification worker.
|
||||
|
||||
Run as a SUBPROCESS so that any native crash inside dlib / face_recognition
|
||||
(which Python cannot catch and which would otherwise kill the whole ICEYOU
|
||||
app) is contained here. If this process segfaults, the parent simply sees a
|
||||
non-zero exit code and carries on.
|
||||
|
||||
Processes MULTIPLE candidate frames in a single invocation (one model load)
|
||||
and returns as soon as one matches, so the parent stays responsive.
|
||||
|
||||
Usage:
|
||||
python -m iceyou.dnn_verify_worker \
|
||||
--encodings faces/face_encodings.pkl \
|
||||
--labels faces/dnn_labels.json \
|
||||
--tolerance 0.6 [--allowed owner,admin] \
|
||||
img1.png img2.png img3.png
|
||||
|
||||
Output (stdout, single line JSON):
|
||||
{"matched": bool, "label": str|null, "distance": float|null}
|
||||
"""
|
||||
|
||||
import sys
|
||||
import json
|
||||
import pickle
|
||||
import argparse
|
||||
|
||||
|
||||
def main() -> int:
|
||||
no_match = {"matched": False, "label": None, "distance": None}
|
||||
try:
|
||||
parser = argparse.ArgumentParser()
|
||||
parser.add_argument("--encodings", required=True)
|
||||
parser.add_argument("--labels", required=True)
|
||||
parser.add_argument("--tolerance", type=float, default=0.6)
|
||||
parser.add_argument("--allowed", default="")
|
||||
parser.add_argument("images", nargs="+")
|
||||
args = parser.parse_args()
|
||||
|
||||
allowed = None
|
||||
if args.allowed.strip():
|
||||
allowed = set(x for x in args.allowed.split(",") if x)
|
||||
|
||||
def _is_allowed(label: str) -> bool:
|
||||
# Accept exact matches OR augmentation variants (e.g. an allowed
|
||||
# label "owner" matches "owner", "owner_glasses", "owner_dnn", ...).
|
||||
if allowed is None:
|
||||
return True
|
||||
for a in allowed:
|
||||
if label == a or label.startswith(a + "_"):
|
||||
return True
|
||||
return False
|
||||
|
||||
import numpy as np
|
||||
import face_recognition
|
||||
|
||||
with open(args.encodings, "rb") as f:
|
||||
known_encodings = pickle.load(f)
|
||||
with open(args.labels, "r", encoding="utf-8") as f:
|
||||
known_labels = json.load(f)
|
||||
|
||||
best_label = None
|
||||
best_distance = None
|
||||
best_distance_any = None # closest match regardless of allowed filter
|
||||
best_label_any = None
|
||||
frames_with_faces = 0
|
||||
|
||||
for image_path in args.images:
|
||||
try:
|
||||
image = face_recognition.load_image_file(image_path)
|
||||
except Exception:
|
||||
continue
|
||||
|
||||
# Upsample once so smaller / farther / slightly-angled faces are
|
||||
# still detected (default upsample=0 misses many webcam faces).
|
||||
face_locations = face_recognition.face_locations(image, number_of_times_to_upsample=1)
|
||||
if not face_locations:
|
||||
continue
|
||||
|
||||
encodings = face_recognition.face_encodings(image, face_locations)
|
||||
if not encodings:
|
||||
continue
|
||||
|
||||
frames_with_faces += 1
|
||||
|
||||
for enc in encodings:
|
||||
distances = face_recognition.face_distance(known_encodings, enc)
|
||||
if len(distances) == 0:
|
||||
continue
|
||||
idx = int(np.argmin(distances))
|
||||
dist = float(distances[idx])
|
||||
label = known_labels[idx]
|
||||
# Track the global closest match (for diagnostics)
|
||||
if best_distance_any is None or dist < best_distance_any:
|
||||
best_distance_any = dist
|
||||
best_label_any = label
|
||||
if not _is_allowed(label):
|
||||
continue
|
||||
if best_distance is None or dist < best_distance:
|
||||
best_distance = dist
|
||||
best_label = label
|
||||
|
||||
# Early exit once we have a confident match
|
||||
if best_distance is not None and best_distance <= args.tolerance:
|
||||
break
|
||||
|
||||
sys.stderr.write(
|
||||
f"[dnn_verify_worker] frames={len(args.images)} frames_with_faces={frames_with_faces} "
|
||||
f"closest_any={best_label_any}:{best_distance_any if best_distance_any is None else round(best_distance_any,3)}\n"
|
||||
)
|
||||
|
||||
if best_label is None or best_distance is None:
|
||||
# No ALLOWED match. If a face was seen but filtered out by allowed
|
||||
# labels, surface that distance so the parent can explain it.
|
||||
if best_distance_any is not None:
|
||||
sys.stderr.write(
|
||||
f"[dnn_verify_worker] a face was detected but no ALLOWED label matched "
|
||||
f"(closest='{best_label_any}' dist={best_distance_any:.3f}); check allowed_labels\n"
|
||||
)
|
||||
elif frames_with_faces == 0:
|
||||
sys.stderr.write("[dnn_verify_worker] NO face detected in any frame (camera angle/lighting?)\n")
|
||||
print(json.dumps(no_match))
|
||||
return 0
|
||||
|
||||
matched = best_distance <= args.tolerance
|
||||
result = {
|
||||
"matched": bool(matched),
|
||||
"label": best_label,
|
||||
"distance": best_distance,
|
||||
}
|
||||
if not matched:
|
||||
# Helpful for debugging why DNN did not match
|
||||
sys.stderr.write(f"[dnn_verify_worker] best_distance={best_distance:.3f} > tolerance={args.tolerance:.3f} (no match)\n")
|
||||
print(json.dumps(result))
|
||||
return 0
|
||||
|
||||
except Exception as e:
|
||||
sys.stderr.write(f"[dnn_verify_worker] {type(e).__name__}: {e}\n")
|
||||
print(json.dumps(no_match))
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
+13
-1
@@ -192,6 +192,10 @@ class Monitor:
|
||||
self.is_away = away
|
||||
self._log_event("manual_state", f"Manually set away={away}")
|
||||
if away:
|
||||
# If we were paused (camera released), resume first so the camera
|
||||
# reactivates for motion recording and face unlock on the white screen.
|
||||
if self.paused:
|
||||
self.resume_monitoring()
|
||||
# Manual force: lock until explicit unlock
|
||||
self.manual_lock = True
|
||||
self.snooze_until = 0.0
|
||||
@@ -226,12 +230,20 @@ class Monitor:
|
||||
self.white_screen.hide()
|
||||
|
||||
def resume_monitoring(self) -> None:
|
||||
"""Resume monitoring after being paused."""
|
||||
"""Resume monitoring after being paused. Re-opens the camera so face
|
||||
unlock, motion recording, and obstruction detection work again.
|
||||
"""
|
||||
if not self.paused:
|
||||
return
|
||||
self.paused = False
|
||||
self._log_event("monitoring_resumed", "Monitoring resumed")
|
||||
|
||||
# Re-acquire cameras (they were released in pause_monitoring)
|
||||
try:
|
||||
self.camera.reopen()
|
||||
except Exception as e:
|
||||
print(f"[Monitor] camera reopen failed: {e}")
|
||||
|
||||
# Restart hooks
|
||||
self.input_hooks.start()
|
||||
if self.config.get("device_monitoring", True):
|
||||
|
||||
+42
-11
@@ -37,21 +37,43 @@ class TrayApp:
|
||||
self.actions = Actions(self.config, self.monitor)
|
||||
self.monitor.on_intrusion = self._on_intrusion
|
||||
|
||||
# Optional face recognition
|
||||
self.face_recognizer: Optional[FaceRecognizer] = None
|
||||
# Optional face recognition (supports "lbph" or "dnn" backend)
|
||||
self.face_recognizer = None
|
||||
face_cfg = self.config.get("face_unlock", {}) or {}
|
||||
face_enabled = bool(face_cfg.get("enabled", self.config.get("face_recognition_enabled", False)))
|
||||
if face_enabled:
|
||||
backend = self.config.get("face_recognition_backend", "lbph").lower()
|
||||
threshold = float(face_cfg.get("confidence_threshold", 70.0))
|
||||
allowed = face_cfg.get("allowed_labels") # None = any enrolled
|
||||
allowed = face_cfg.get("allowed_labels")
|
||||
min_matches = int(face_cfg.get("min_successful_matches", 2))
|
||||
self.face_recognizer = FaceRecognizer(
|
||||
confidence_threshold=threshold,
|
||||
allowed_labels=allowed,
|
||||
min_successful_matches=min_matches,
|
||||
)
|
||||
if not self.face_recognizer.available:
|
||||
self.face_recognizer = None # Enrol first; fall back to password only
|
||||
|
||||
if backend == "dnn":
|
||||
try:
|
||||
from .dnn_face_recognizer import DNNFaceRecognizer
|
||||
faces_dir = face_cfg.get("enrollment_images_dir", "faces")
|
||||
# DNN tolerance is a face DISTANCE (~0.6, lower = stricter),
|
||||
# NOT the LBPH confidence_threshold. Read it from the
|
||||
# face_recognition config block.
|
||||
dnn_cfg = self.config.get("face_recognition", {}) or {}
|
||||
dnn_tolerance = float(dnn_cfg.get("tolerance", 0.6))
|
||||
self.face_recognizer = DNNFaceRecognizer(
|
||||
tolerance=dnn_tolerance,
|
||||
allowed_labels=allowed,
|
||||
faces_dir=faces_dir,
|
||||
)
|
||||
except Exception as e:
|
||||
print(f"[TrayApp] Failed to load DNNFaceRecognizer: {e}")
|
||||
self.face_recognizer = None
|
||||
else:
|
||||
# Default: LBPH
|
||||
self.face_recognizer = FaceRecognizer(
|
||||
confidence_threshold=threshold,
|
||||
allowed_labels=allowed,
|
||||
min_successful_matches=min_matches,
|
||||
)
|
||||
|
||||
if self.face_recognizer and not self.face_recognizer.available:
|
||||
self.face_recognizer = None
|
||||
|
||||
# Wire WhiteScreen callbacks
|
||||
ws = self.monitor.white_screen
|
||||
@@ -105,6 +127,7 @@ class TrayApp:
|
||||
if self.face_recognizer is None or not self.face_recognizer.available:
|
||||
return False, None, None
|
||||
attempts = int(self.config.get("face_unlock", {}).get("frames_per_attempt", 4))
|
||||
print("[TrayApp] Face unlock attempt: reading camera frames for DNN verification...")
|
||||
return self.face_recognizer.verify_from_camera(self.monitor.camera, attempts=attempts)
|
||||
|
||||
def _on_motion_event(self, clip_path: str) -> None:
|
||||
@@ -247,7 +270,15 @@ class TrayApp:
|
||||
def _toggle_white_screen(self, icon=None, item=None) -> None:
|
||||
ws = self.monitor.white_screen
|
||||
ws.enabled = not ws.enabled
|
||||
if not ws.enabled:
|
||||
if ws.enabled:
|
||||
# If the user is enabling the white screen while monitoring is paused
|
||||
# (camera released), resume so the camera reactivates for face unlock
|
||||
# and motion recording. Otherwise the white screen would have no camera.
|
||||
if self.monitor.paused:
|
||||
self.monitor.resume_monitoring()
|
||||
if not ws.is_shown:
|
||||
ws.show()
|
||||
else:
|
||||
ws.hide()
|
||||
self.icon.notify(f"White screen {'enabled' if ws.enabled else 'disabled'}", "ICEYOU")
|
||||
|
||||
|
||||
+24
-11
@@ -522,13 +522,16 @@ class WhiteScreen:
|
||||
else:
|
||||
matched = bool(result)
|
||||
except Exception as e:
|
||||
print(f"[WhiteScreen] face_unlock error: {e}")
|
||||
# Hand result back to Tk thread
|
||||
if self._root:
|
||||
try:
|
||||
print(f"[WhiteScreen] face_unlock error: {type(e).__name__}: {e}")
|
||||
import traceback
|
||||
traceback.print_exc(limit=2)
|
||||
# Hand result back to Tk thread - never let this crash the UI thread
|
||||
try:
|
||||
if self._root:
|
||||
self._root.after(0, lambda: self._face_unlock_result(matched, label, conf, silent, manual))
|
||||
except Exception:
|
||||
self._face_busy = False
|
||||
except Exception as e:
|
||||
print(f"[WhiteScreen] face result delivery error: {e}")
|
||||
self._face_busy = False
|
||||
|
||||
threading.Thread(target=worker, daemon=True).start()
|
||||
|
||||
@@ -542,11 +545,18 @@ class WhiteScreen:
|
||||
# Fire attempt log: always for manual clicks, only on success for auto (avoid flooding)
|
||||
if manual or matched:
|
||||
try:
|
||||
conf_value = None
|
||||
if conf is not None:
|
||||
try:
|
||||
conf_value = float(conf)
|
||||
except (ValueError, TypeError):
|
||||
conf_value = None
|
||||
|
||||
self.on_attempt({
|
||||
"method": "face",
|
||||
"success": matched,
|
||||
"label": label,
|
||||
"confidence": float(conf) if conf is not None else None,
|
||||
"confidence": conf_value,
|
||||
})
|
||||
except Exception as e:
|
||||
print(f"[WhiteScreen] on_attempt error: {e}")
|
||||
@@ -556,12 +566,15 @@ class WhiteScreen:
|
||||
self.on_unlocked()
|
||||
except Exception as e:
|
||||
print(f"[WhiteScreen] on_unlocked error: {e}")
|
||||
self._hide_window()
|
||||
try:
|
||||
self._hide_window()
|
||||
except Exception as e:
|
||||
print(f"[WhiteScreen] _hide_window error after face success: {e}")
|
||||
return
|
||||
if not silent and self._status_var:
|
||||
if self._status_var:
|
||||
if conf is not None:
|
||||
self._status_var.set(
|
||||
f"Face not recognised (best={label or '?'}, conf={conf:.1f}). Try the password."
|
||||
f"Face not recognised (best={label or '?'}, dist={conf:.2f}). Try the password. (DNN: lower distance = better match)"
|
||||
)
|
||||
else:
|
||||
elif not silent:
|
||||
self._status_var.set("No face detected. Try the password.")
|
||||
|
||||
Reference in New Issue
Block a user