diff --git a/timelapse.py b/image_processing.py similarity index 75% rename from timelapse.py rename to image_processing.py index 676ffa6..7c3b8d1 100644 --- a/timelapse.py +++ b/image_processing.py @@ -1,7 +1,6 @@ -# timelapse.py +# image_processing.py import os import io -import requests import concurrent.futures from datetime import datetime from dataclasses import dataclass @@ -12,6 +11,7 @@ import dlib from tqdm import tqdm import logging from typing import Tuple +from immich_api import get_assets_with_person, download_asset class TqdmLoggingHandler(logging.Handler): def __init__(self, level=logging.NOTSET): @@ -32,8 +32,6 @@ formatter = logging.Formatter('%(asctime)s - %(levelname)s - %(message)s', datef tqdm_handler.setFormatter(formatter) logger.addHandler(tqdm_handler) -face_predictor = None - @dataclass class AppConfig: @@ -50,71 +48,6 @@ class AppConfig: date_from: str date_to: str -def draw_landmarks(image, landmarks, face_rect, output_path): - """ - Draws facial landmarks and face rectangle on the image and saves it. - - Args: - image (numpy.ndarray): The input image in BGR format. - landmarks (dict): Dictionary containing facial landmarks. - face_rect (tuple): Face rectangle coordinates (x1, y1, x2, y2). - output_path (str): Path to save the output image. - """ - # Create a copy of the image to draw on - img = image.copy() - - # Draw face rectangle - x1, y1, x2, y2 = face_rect - cv2.rectangle(img, (x1, y1), (x2, y2), (0, 255, 0), 2) - - # Draw landmarks if available - if landmarks is not None: - for i, (x, y) in landmarks.items(): - cv2.circle(img, (int(x), int(y)), 2, (0, 0, 255), -1) - - # Create output directory if it doesn't exist - os.makedirs(os.path.dirname(output_path), exist_ok=True) - - # Save the image - cv2.imwrite(output_path, img) - -def validate_immich_connection(api_key, base_url): - """ - Validates that the provided Immich API key and base URL are working. - - Args: - api_key (str): API key for authentication. - base_url (str): Base URL of the API. - - Returns: - tuple: (bool, str) - (is_valid, error_message) - """ - if not api_key or not base_url: - return False, "API key and base URL are required." - - try: - headers = { - 'Accept': 'application/json', - 'x-api-key': api_key, - } - # Try a simple ping to the server via the user endpoint - url = f"{base_url}/server/about" - response = requests.get(url, headers=headers, timeout=5) - - if response.status_code == 200: - return True, "Connection successful." - elif response.status_code == 401: - return False, "Authentication failed. Invalid API key." - else: - return False, f"Server error: Status code {response.status_code}" - - except requests.exceptions.ConnectionError: - return False, "Connection error. Check the base URL." - except requests.exceptions.Timeout: - return False, "Connection timed out. Server might be down." - except Exception as e: - return False, f"Unexpected error: {str(e)}" - def initialize_worker(landmark_model_path: str) -> None: """Initialize worker process with face predictor. @@ -126,142 +59,6 @@ def initialize_worker(landmark_model_path: str) -> None: face_predictor = dlib.shape_predictor(landmark_model_path) -def get_assets_with_person(api_key, base_url, person_id, date_from=None, date_to=None): - """ - Retrieve all image assets containing the specified person by querying the API. - - Args: - api_key (str): API key for authentication. - base_url (str): Base URL of the API. - person_id (str): ID of the person to search for. - date_from (str, optional): Start date in ISO format (YYYY-MM-DD). - date_to (str, optional): End date in ISO format (YYYY-MM-DD). - - Returns: - list: List of asset dictionaries. - """ - headers = { - 'Content-Type': 'application/json', - 'Accept': 'application/json', - 'x-api-key': api_key, - } - url = f"{base_url}/search/metadata" - all_assets = [] - payload = { - "page": 1, - "type": "IMAGE", - "personIds": [person_id], - "withArchived": False, - "withDeleted": True, - "withExif": True, - "withPeople": True, - "withStacked": True, - } - - if date_from: - payload["takenAfter"] = f"{date_from}T00:00:00.000Z" - - if date_to: - payload["takenBefore"] = f"{date_to}T23:59:59.999Z" - - while payload["page"] is not None: - response = requests.post(url, headers=headers, json=payload) - if response.status_code != 200: - logger.info(f"Error fetching page {payload['page']}: {response.status_code} - {response.text}") - break - data = response.json() - if not data: - break - all_assets.extend(data['assets']['items']) - logger.info(f"Fetched page {payload['page']} with {len(data['assets']['items'])} assets") - payload["page"] = data['assets'].get('nextPage') - return all_assets - - -def download_asset(api_key, base_url, asset_id): - """ - Downloads the original image asset from the API. - - Args: - api_key (str): API key for authentication. - base_url (str): Base URL of the API. - asset_id (str): The asset's ID. - - Returns: - bytes: The content of the downloaded image. - """ - headers = {'x-api-key': api_key} - response = requests.get(f'{base_url}/assets/{asset_id}/original', headers=headers) - response.raise_for_status() - return response.content - - -def get_head_pose(landmarks, image): - """ - Estimates the head pose (pitch, yaw, roll) using facial landmarks. - Based on https://learnopencv.com/head-pose-estimation-using-opencv-and-dlib/ - - Args: - landmarks (dict): Dictionary containing facial landmarks in numpy arrays. - image (PIL.Image): The input image. - - Returns: - tuple or None: (pitch, yaw, roll) in degrees if successful; otherwise None. - """ - # Get image size - img_width, img_height = image.size - - image_points = np.array([ - landmarks['nose_tip'], - landmarks['chin'], - landmarks['left_eye'][0], - landmarks['right_eye'][3], - landmarks['left_mouth'], - landmarks['right_mouth'] - ], dtype="double") - - model_points = np.array([ - (0.0, 0.0, 0.0), # Nose tip - (0.0, -330.0, -65.0), # Chin - (-225.0, 170.0, -135.0), # Left eye left corner - (225.0, 170.0, -135.0), # Right eye right corner - (-150.0, -150.0, -125.0), # Left Mouth corner - (150.0, -150.0, -125.0) # Right mouth corner - ]) - - focal_length = img_width - center = (img_width / 2, img_height / 2) - camera_matrix = np.array( - [[focal_length, 0, center[0]], - [0, focal_length, center[1]], - [0, 0, 1]], dtype="double" - ) - dist_coeffs = np.zeros((4, 1)) - success, rotation_vector, translation_vector = cv2.solvePnP( - model_points, image_points, camera_matrix, dist_coeffs, flags=cv2.SOLVEPNP_ITERATIVE - ) - if not success: - logger.info("Head pose estimation failed in solvePnP.") - return None - rotation_matrix, _ = cv2.Rodrigues(rotation_vector) - proj_matrix = np.hstack((rotation_matrix, translation_vector)) - _, _, _, _, _, _, euler_angles = cv2.decomposeProjectionMatrix(proj_matrix) - pitch, yaw, roll = [float(angle) for angle in euler_angles] - - # Normalize angles to be between -180 and 180 degrees - pitch = (pitch + 180) % 360 - 180 - yaw = (yaw + 180) % 360 - 180 - roll = (roll + 180) % 360 - 180 - - # Adjust pitch to be between -90 and 90 degrees - if pitch > 90: - pitch = 180 - pitch - elif pitch < -90: - pitch = -180 - pitch - - return pitch, yaw, roll - - def detect_landmarks(image, face_data, face_resolution_threshold): """ Detects facial landmarks in the image, resizing if necessary for better detection. @@ -376,6 +173,72 @@ def check_eye_visibility(left_eye, right_eye, ear_threshold=0.2) -> bool: return True +def get_head_pose(landmarks, image): + """ + Estimates the head pose (pitch, yaw, roll) using facial landmarks. + Based on https://learnopencv.com/head-pose-estimation-using-opencv-and-dlib/ + + Args: + landmarks (dict): Dictionary containing facial landmarks in numpy arrays. + image (PIL.Image): The input image. + + Returns: + tuple or None: (pitch, yaw, roll) in degrees if successful; otherwise None. + """ + # Get image size + img_width, img_height = image.size + + image_points = np.array([ + landmarks['nose_tip'], + landmarks['chin'], + landmarks['left_eye'][0], + landmarks['right_eye'][3], + landmarks['left_mouth'], + landmarks['right_mouth'] + ], dtype="double") + + model_points = np.array([ + (0.0, 0.0, 0.0), # Nose tip + (0.0, -330.0, -65.0), # Chin + (-225.0, 170.0, -135.0), # Left eye left corner + (225.0, 170.0, -135.0), # Right eye right corner + (-150.0, -150.0, -125.0), # Left Mouth corner + (150.0, -150.0, -125.0) # Right mouth corner + ]) + + focal_length = img_width + center = (img_width / 2, img_height / 2) + camera_matrix = np.array( + [[focal_length, 0, center[0]], + [0, focal_length, center[1]], + [0, 0, 1]], dtype="double" + ) + dist_coeffs = np.zeros((4, 1)) + success, rotation_vector, translation_vector = cv2.solvePnP( + model_points, image_points, camera_matrix, dist_coeffs, flags=cv2.SOLVEPNP_ITERATIVE + ) + if not success: + logger.info("Head pose estimation failed in solvePnP.") + return None + rotation_matrix, _ = cv2.Rodrigues(rotation_vector) + proj_matrix = np.hstack((rotation_matrix, translation_vector)) + _, _, _, _, _, _, euler_angles = cv2.decomposeProjectionMatrix(proj_matrix) + pitch, yaw, roll = [float(angle) for angle in euler_angles] + + # Normalize angles to be between -180 and 180 degrees + pitch = (pitch + 180) % 360 - 180 + yaw = (yaw + 180) % 360 - 180 + roll = (roll + 180) % 360 - 180 + + # Adjust pitch to be between -90 and 90 degrees + if pitch > 90: + pitch = 180 - pitch + elif pitch < -90: + pitch = -180 - pitch + + return pitch, yaw, roll + + def calculate_eye_alignment_transform( left_eye_center: np.ndarray, right_eye_center: np.ndarray, @@ -479,19 +342,6 @@ def crop_and_align_face(image, face_data, resize_size, face_resolution_threshold PIL.Image or None: The aligned face image if successful, None otherwise. """ try: - # DEBUG : remove after testing - face_img_width = face_data.get("imageWidth") - face_img_height = face_data.get("imageHeight") - img_width, img_height = image.size - scale_x = img_width / face_img_width - scale_y = img_height / face_img_height - x1 = int(face_data.get("boundingBoxX1", 0) * scale_x) - x2 = int(face_data.get("boundingBoxX2", 0) * scale_x) - y1 = int(face_data.get("boundingBoxY1", 0) * scale_y) - y2 = int(face_data.get("boundingBoxY2", 0) * scale_y) - w = x2 - x1 - h = y2 - y1 - # Convert image to numpy array for OpenCV processing img_np = np.array(image) img_np = cv2.cvtColor(img_np, cv2.COLOR_RGB2BGR) @@ -505,21 +355,18 @@ def crop_and_align_face(image, face_data, resize_size, face_resolution_threshold # Check if both eyes are visible if not check_eye_visibility(landmarks['left_eye'], landmarks['right_eye']): logger.info("Eyes are too closed") - # draw_landmarks(img_np, landmarks['all_landmarks'], (x1, y1, x2, y2), f"discarded/eyes_closed_{face_data.get('id')}.jpg") return None # Get head pose pose = get_head_pose(landmarks, image) if not pose: logger.info("Could not estimate head pose") - draw_landmarks(img_np, landmarks['all_landmarks'], (x1, y1, x2, y2), "discarded/no_pose.jpg") return None # Check if head pose is within acceptable range pitch, yaw, roll = pose if abs(yaw) > pose_threshold: logger.info(f"Head pose exceeds threshold: pitch={pitch:.1f}°, yaw={yaw:.1f}°, roll={roll:.1f}°") - draw_landmarks(img_np, landmarks['all_landmarks'], (x1, y1, x2, y2), f"discarded/pose_yaw_{yaw:.1f}.jpg") return None # Get eye positions diff --git a/immich_api.py b/immich_api.py new file mode 100644 index 0000000..a20b1e3 --- /dev/null +++ b/immich_api.py @@ -0,0 +1,112 @@ +import requests +import logging + +logger = logging.getLogger(__name__) + + +def validate_immich_connection(api_key, base_url): + """ + Validates that the provided Immich API key and base URL are working. + + Args: + api_key (str): API key for authentication. + base_url (str): Base URL of the API. + + Returns: + tuple: (bool, str) - (is_valid, error_message) + """ + if not api_key or not base_url: + return False, "API key and base URL are required." + + try: + headers = { + 'Accept': 'application/json', + 'x-api-key': api_key, + } + # Try a simple ping to the server via the user endpoint + url = f"{base_url}/server/about" + response = requests.get(url, headers=headers, timeout=5) + + if response.status_code == 200: + return True, "Connection successful." + elif response.status_code == 401: + return False, "Authentication failed. Invalid API key." + else: + return False, f"Server error: Status code {response.status_code}" + + except requests.exceptions.ConnectionError: + return False, "Connection error. Check the base URL." + except requests.exceptions.Timeout: + return False, "Connection timed out. Server might be down." + except Exception as e: + return False, f"Unexpected error: {str(e)}" + + +def get_assets_with_person(api_key, base_url, person_id, date_from=None, date_to=None): + """ + Retrieve all image assets containing the specified person by querying the API. + + Args: + api_key (str): API key for authentication. + base_url (str): Base URL of the API. + person_id (str): ID of the person to search for. + date_from (str, optional): Start date in ISO format (YYYY-MM-DD). + date_to (str, optional): End date in ISO format (YYYY-MM-DD). + + Returns: + list: List of asset dictionaries. + """ + headers = { + 'Content-Type': 'application/json', + 'Accept': 'application/json', + 'x-api-key': api_key, + } + url = f"{base_url}/search/metadata" + all_assets = [] + payload = { + "page": 1, + "type": "IMAGE", + "personIds": [person_id], + "withArchived": False, + "withDeleted": True, + "withExif": True, + "withPeople": True, + "withStacked": True, + } + + if date_from: + payload["takenAfter"] = f"{date_from}T00:00:00.000Z" + + if date_to: + payload["takenBefore"] = f"{date_to}T23:59:59.999Z" + + while payload["page"] is not None: + response = requests.post(url, headers=headers, json=payload) + if response.status_code != 200: + logger.info(f"Error fetching page {payload['page']}: {response.status_code} - {response.text}") + break + data = response.json() + if not data: + break + all_assets.extend(data['assets']['items']) + logger.info(f"Fetched page {payload['page']} with {len(data['assets']['items'])} assets") + payload["page"] = data['assets'].get('nextPage') + return all_assets + + +def download_asset(api_key, base_url, asset_id): + """ + Downloads the original image asset from the API. + + Args: + api_key (str): API key for authentication. + base_url (str): Base URL of the API. + asset_id (str): The asset's ID. + + Returns: + bytes: The content of the downloaded image. + """ + headers = {'x-api-key': api_key} + response = requests.get(f'{base_url}/assets/{asset_id}/original', headers=headers) + response.raise_for_status() + return response.content diff --git a/main.py b/main.py index 5ea5477..d862e14 100644 --- a/main.py +++ b/main.py @@ -6,7 +6,8 @@ from dataclasses import dataclass from typing import Callable, Dict, List, Tuple from flask import Flask, jsonify, render_template, request -from timelapse import process_faces, validate_immich_connection +from image_processing import process_faces +from immich_api import validate_immich_connection # Configure logging logger = logging.getLogger(__name__)