"""Face landmark detection using MediaPipe FaceMesh for passport photo positioning.""" import sys import json import os def emit_progress(percent, stage): """Emit structured progress to stderr for bridge.ts to capture.""" print(json.dumps({"progress": percent, "stage": stage}), file=sys.stderr, flush=True) # ── Landmark extraction (shared by both APIs) ────────────────────── # MediaPipe face mesh indices for key points LEFT_EYE_INDICES = [33, 133, 159, 145, 160, 144, 158, 153] RIGHT_EYE_INDICES = [362, 263, 386, 374, 385, 373, 387, 380] CHIN_INDEX = 152 FOREHEAD_INDEX = 10 NOSE_INDEX = 1 def extract_key_points(lms): """Extract passport-relevant points from a list of (x, y) normalized landmarks.""" left_eye_x = sum(lms[i][0] for i in LEFT_EYE_INDICES) / len(LEFT_EYE_INDICES) left_eye_y = sum(lms[i][1] for i in LEFT_EYE_INDICES) / len(LEFT_EYE_INDICES) right_eye_x = sum(lms[i][0] for i in RIGHT_EYE_INDICES) / len(RIGHT_EYE_INDICES) right_eye_y = sum(lms[i][1] for i in RIGHT_EYE_INDICES) / len(RIGHT_EYE_INDICES) eye_center_x = (left_eye_x + right_eye_x) / 2 eye_center_y = (left_eye_y + right_eye_y) / 2 chin_x, chin_y = lms[CHIN_INDEX] forehead_x, forehead_y = lms[FOREHEAD_INDEX] nose_x, nose_y = lms[NOSE_INDEX] forehead_chin_dist = chin_y - forehead_y crown_y = forehead_y - (forehead_chin_dist * 0.15) crown_x = forehead_x face_center_x = (nose_x + eye_center_x) / 2 return { "leftEye": {"x": round(left_eye_x, 6), "y": round(left_eye_y, 6)}, "rightEye": {"x": round(right_eye_x, 6), "y": round(right_eye_y, 6)}, "eyeCenter": {"x": round(eye_center_x, 6), "y": round(eye_center_y, 6)}, "chin": {"x": round(chin_x, 6), "y": round(chin_y, 6)}, "forehead": {"x": round(forehead_x, 6), "y": round(forehead_y, 6)}, "crown": {"x": round(crown_x, 6), "y": round(crown_y, 6)}, "nose": {"x": round(nose_x, 6), "y": round(nose_y, 6)}, "faceCenterX": round(face_center_x, 6), } # ── Old API: mp.solutions (mediapipe < 0.10.30) ─────────────────── def detect_with_solutions(img_array, max_faces=1): """Use the legacy mp.solutions.face_mesh API.""" import mediapipe as mp mp_face_mesh = mp.solutions.face_mesh face_mesh = mp_face_mesh.FaceMesh( static_image_mode=True, max_num_faces=max_faces, refine_landmarks=True, min_detection_confidence=0.5, ) results = face_mesh.process(img_array) face_mesh.close() if not results.multi_face_landmarks: return None face_lm = results.multi_face_landmarks[0] return [(lm.x, lm.y) for lm in face_lm.landmark] # ── New API: mp.tasks (mediapipe >= 0.10.30) ─────────────────────── _MODELS_BASE = os.environ.get("MODELS_PATH", "/opt/models") MODEL_URL = "https://storage.googleapis.com/mediapipe-models/face_landmarker/face_landmarker/float16/latest/face_landmarker.task" _DOCKER_MODEL_PATH = os.path.join(_MODELS_BASE, "mediapipe", "face_landmarker.task") MODEL_DIR = os.path.join(os.path.dirname(__file__), "..", "..", "..", ".models") MODEL_PATH = os.path.join(MODEL_DIR, "face_landmarker.task") def ensure_model(): """Resolve face landmarker model. Docker path first, then local dev.""" if os.path.exists(_DOCKER_MODEL_PATH): return _DOCKER_MODEL_PATH if os.path.exists(MODEL_PATH): return MODEL_PATH os.makedirs(MODEL_DIR, exist_ok=True) import urllib.request emit_progress(15, "Downloading face model") urllib.request.urlretrieve(MODEL_URL, MODEL_PATH) return MODEL_PATH def detect_with_tasks(img_path, max_faces=1): """Use the new mp.tasks.vision.FaceLandmarker API.""" import mediapipe as mp model_path = ensure_model() options = mp.tasks.vision.FaceLandmarkerOptions( base_options=mp.tasks.BaseOptions(model_asset_path=model_path), running_mode=mp.tasks.vision.RunningMode.IMAGE, num_faces=max_faces, min_face_detection_confidence=0.5, output_face_blendshapes=False, output_facial_transformation_matrixes=False, ) landmarker = mp.tasks.vision.FaceLandmarker.create_from_options(options) mp_image = mp.Image.create_from_file(img_path) result = landmarker.detect(mp_image) landmarker.close() if not result.face_landmarks: return None face_lm = result.face_landmarks[0] return [(lm.x, lm.y) for lm in face_lm] # ── Main ─────────────────────────────────────────────────────────── def main(): input_path = sys.argv[1] output_path = sys.argv[2] # unused but kept for bridge.ts compatibility settings = json.loads(sys.argv[3]) if len(sys.argv) > 3 else {} max_faces = settings.get("max_num_faces", 1) try: emit_progress(10, "Loading image") from PIL import Image img = Image.open(input_path).convert("RGB") iw, ih = img.size try: import mediapipe as mp import numpy as np emit_progress(20, "Initializing face mesh") landmarks_list = None try: img_array = np.array(img) emit_progress(30, "Detecting face landmarks") landmarks_list = detect_with_solutions(img_array, max_faces) except AttributeError: emit_progress(30, "Detecting face landmarks") landmarks_list = detect_with_tasks(input_path, max_faces) if landmarks_list is None: print(json.dumps({ "success": True, "faceDetected": False, "landmarks": None, })) return emit_progress(60, "Extracting key points") key_points = extract_key_points(landmarks_list) emit_progress(90, "Done") print(json.dumps({ "success": True, "faceDetected": True, "landmarks": key_points, "imageWidth": iw, "imageHeight": ih, })) except ImportError: print(json.dumps({ "success": False, "error": "Face landmark detection requires MediaPipe. Install with: pip install mediapipe", })) sys.exit(1) except ImportError: print(json.dumps({ "success": False, "error": "Pillow is not installed. Install with: pip install Pillow", })) sys.exit(1) except Exception as e: print(json.dumps({"success": False, "error": str(e)})) sys.exit(1) if __name__ == "__main__": main()