refactor: Update face location handling to DeepFace format across the codebase

This commit refactors the handling of face location data to exclusively use the DeepFace format ({x, y, w, h}) instead of the legacy tuple format (top, right, bottom, left). Key changes include updating method signatures, modifying internal logic for face quality score calculations, and ensuring compatibility in the GUI components. Additionally, configuration settings for face detection have been adjusted to allow for smaller face sizes and lower confidence thresholds, enhancing the system's ability to detect faces in various conditions. All relevant tests have been updated to reflect these changes, ensuring continued functionality and performance.
This commit is contained in:
tanyar09
2025-10-17 12:55:11 -04:00
parent 68673ccdbe
commit 2828b9966b
9 changed files with 299 additions and 206 deletions
+15 -3
View File
@@ -194,15 +194,27 @@ class GUICore:
return badge_frame
def create_face_crop_image(self, photo_path: str, face_location: tuple,
def create_face_crop_image(self, photo_path: str, face_location: dict,
face_id: int, crop_size: int = 100) -> Optional[str]:
"""Create a face crop image for display"""
try:
# Parse location tuple from string format
# Parse location from string format (DeepFace format only)
if isinstance(face_location, str):
face_location = eval(face_location)
top, right, bottom, left = face_location
# DeepFace format: {x, y, w, h}
if not isinstance(face_location, dict):
raise ValueError(f"Expected DeepFace dict format, got {type(face_location)}")
x = face_location.get('x', 0)
y = face_location.get('y', 0)
w = face_location.get('w', 0)
h = face_location.get('h', 0)
left = x
top = y
right = x + w
bottom = y + h
# Load the image
with Image.open(photo_path) as image:
+202 -98
View File
@@ -40,6 +40,10 @@ class IdentifyPanel:
self.identify_data_cache = {}
self.current_face_crop_path = None
# Caching system for all faces data
self.all_faces_cache = [] # Cache all faces from database
self.cache_loaded = False # Flag to track if cache is loaded
# GUI components
self.components = {}
self.main_frame = None
@@ -248,16 +252,10 @@ class IdentifyPanel:
self.components['unique_var'].set(False)
return
else:
# Reload the original unfiltered face list
print("🔄 Reloading all faces...")
# Reload the original unfiltered face list from cache
print("🔄 Reloading all faces from cache...")
self.main_frame.update() # Update UI to show the message
# Get current date filters
date_from = self.components['date_from_var'].get().strip() or None
date_to = self.components['date_to_var'].get().strip() or None
date_processed_from = self.components['date_processed_from_var'].get().strip() or None
date_processed_to = self.components['date_processed_to_var'].get().strip() or None
# Get batch size
try:
batch_size = int(self.components['batch_var'].get().strip())
@@ -272,12 +270,12 @@ class IdentifyPanel:
sort_display = self.components['sort_var'].get()
sort_by = self.sort_value_map.get(sort_display, "quality")
# Reload faces with current filters and sort option
self.current_faces = self._get_unidentified_faces(batch_size, date_from, date_to,
date_processed_from, date_processed_to,
min_quality_score, sort_by)
print(f"✅ Reloaded: {len(self.current_faces)} faces")
# Reload faces with current filters and sort option from cache
if self.cache_loaded:
self.current_faces = self._filter_cached_faces(min_quality_score, sort_by, batch_size)
print(f"✅ Reloaded: {len(self.current_faces)} faces from cache")
else:
print("⚠️ Cache not loaded - please click 'Start Identification' first")
# Reset to first face and update display
self.current_face_index = 0
@@ -361,17 +359,11 @@ class IdentifyPanel:
# Add sort change handler
def on_sort_change(event=None):
"""Handle sort option change - refresh face list if identification is active"""
if self.is_active and self.current_faces:
if self.is_active and self.cache_loaded:
# Show progress message
print("🔄 Refreshing face list with new sort order...")
self.main_frame.update()
# Get current filters
date_from = self.components['date_from_var'].get().strip() or None
date_to = self.components['date_to_var'].get().strip() or None
date_processed_from = self.components['date_processed_from_var'].get().strip() or None
date_processed_to = self.components['date_processed_to_var'].get().strip() or None
# Get batch size
try:
batch_size = int(self.components['batch_var'].get().strip())
@@ -386,10 +378,8 @@ class IdentifyPanel:
sort_display = self.components['sort_var'].get()
sort_by = self.sort_value_map.get(sort_display, "quality")
# Reload faces with new sort order
self.current_faces = self._get_unidentified_faces(batch_size, date_from, date_to,
date_processed_from, date_processed_to,
min_quality_score, sort_by)
# Apply new sort order to cached data
self.current_faces = self._filter_cached_faces(min_quality_score, sort_by, batch_size)
# Reset to first face and update display
self.current_face_index = 0
@@ -557,7 +547,7 @@ class IdentifyPanel:
no_compare_label.pack(pady=20)
def _start_identification(self):
"""Start the identification process"""
"""Start the identification process - loads all faces into cache and applies initial filtering"""
try:
batch_size = int(self.components['batch_var'].get().strip())
if batch_size <= 0:
@@ -580,13 +570,20 @@ class IdentifyPanel:
sort_display = self.components['sort_var'].get()
sort_by = self.sort_value_map.get(sort_display, "quality")
# Get unidentified faces with quality filter and sort option
self.current_faces = self._get_unidentified_faces(batch_size, date_from, date_to,
date_processed_from, date_processed_to,
min_quality_score, sort_by)
# Load all faces into cache (only database access point)
print("🔄 Loading faces from database...")
self._load_all_faces_cache(date_from, date_to, date_processed_from, date_processed_to)
if not self.all_faces_cache:
messagebox.showinfo("No Faces", "🎉 All faces have been identified!")
return
# Apply quality filtering and sorting to cached data
print("🔄 Applying quality filter and sorting...")
self.current_faces = self._filter_cached_faces(min_quality_score, sort_by, batch_size)
if not self.current_faces:
messagebox.showinfo("No Faces", "🎉 All faces have been identified!")
messagebox.showinfo("No Faces", f"No faces found with quality >= {min_quality}%.\nTry lowering the quality filter.")
return
# Pre-fetch data for optimal performance
@@ -605,15 +602,16 @@ class IdentifyPanel:
self._update_button_states()
self.is_active = True
print(f"✅ Started identification with {len(self.current_faces)} faces (from {len(self.all_faces_cache)} total cached)")
def _get_unidentified_faces(self, batch_size: int, date_from: str = None, date_to: str = None,
date_processed_from: str = None, date_processed_to: str = None,
min_quality_score: float = 0.0, sort_by: str = "quality") -> List[Tuple]:
"""Get unidentified faces from database with optional date and quality filtering"""
"""Get unidentified faces from database with optional date filtering (no quality filtering at DB level)"""
with self.db.get_db_connection() as conn:
cursor = conn.cursor()
# Build the SQL query with optional date filtering
# Build the SQL query with optional date filtering only
# Include DeepFace metadata: face_confidence, quality_score, detector_backend, model_name
query = '''
SELECT f.id, f.photo_id, p.path, p.filename, f.location,
@@ -624,11 +622,6 @@ class IdentifyPanel:
'''
params = []
# Add quality filtering if specified
if min_quality_score > 0.0:
query += ' AND f.quality_score >= ?'
params.append(min_quality_score)
# Add date taken filtering if specified
if date_from:
query += ' AND p.date_taken >= ?'
@@ -673,6 +666,128 @@ class IdentifyPanel:
}
return sort_clauses.get(sort_by, "f.quality_score DESC") # Default to quality DESC
def _load_all_faces_cache(self, date_from: str = None, date_to: str = None,
date_processed_from: str = None, date_processed_to: str = None) -> None:
"""Load all unidentified faces into cache with date filtering only"""
with self.db.get_db_connection() as conn:
cursor = conn.cursor()
# Build the SQL query with optional date filtering only
query = '''
SELECT f.id, f.photo_id, p.path, p.filename, f.location,
f.face_confidence, f.quality_score, f.detector_backend, f.model_name,
p.date_taken, p.date_added
FROM faces f
JOIN photos p ON f.photo_id = p.id
WHERE f.person_id IS NULL
'''
params = []
# Add date taken filtering if specified
if date_from:
query += ' AND p.date_taken >= ?'
params.append(date_from)
if date_to:
query += ' AND p.date_taken <= ?'
params.append(date_to)
# Add date processed filtering if specified
if date_processed_from:
query += ' AND DATE(p.date_added) >= ?'
params.append(date_processed_from)
if date_processed_to:
query += ' AND DATE(p.date_added) <= ?'
params.append(date_processed_to)
# Order by quality DESC by default (can be re-sorted later)
query += ' ORDER BY f.quality_score DESC'
cursor.execute(query, params)
self.all_faces_cache = cursor.fetchall()
self.cache_loaded = True
if self.verbose > 0:
print(f"📦 Loaded {len(self.all_faces_cache)} faces into cache")
def _filter_cached_faces(self, min_quality_score: float = 0.0, sort_by: str = "quality",
batch_size: int = None) -> List[Tuple]:
"""Filter cached faces by quality and apply sorting"""
if not self.cache_loaded:
return []
# Filter by quality
filtered_faces = []
for face_data in self.all_faces_cache:
face_id, photo_id, photo_path, filename, location, face_conf, quality, detector, model, date_taken, date_added = face_data
quality_score = quality if quality is not None else 0.0
if quality_score >= min_quality_score:
# Return in the same format as the original method (without date_taken, date_added)
filtered_faces.append((face_id, photo_id, photo_path, filename, location, face_conf, quality, detector, model))
# Apply sorting
if sort_by == "quality":
filtered_faces.sort(key=lambda x: x[6] if x[6] is not None else 0.0, reverse=True)
elif sort_by == "quality_asc":
filtered_faces.sort(key=lambda x: x[6] if x[6] is not None else 0.0, reverse=False)
elif sort_by == "date_taken":
# Need to get date_taken from cache for sorting
filtered_faces_with_dates = []
for face_data in self.all_faces_cache:
face_id, photo_id, photo_path, filename, location, face_conf, quality, detector, model, date_taken, date_added = face_data
quality_score = quality if quality is not None else 0.0
if quality_score >= min_quality_score:
filtered_faces_with_dates.append((face_data, date_taken))
filtered_faces_with_dates.sort(key=lambda x: x[1] or "", reverse=True)
filtered_faces = [x[0][:9] for x in filtered_faces_with_dates] # Remove date fields
elif sort_by == "date_taken_asc":
filtered_faces_with_dates = []
for face_data in self.all_faces_cache:
face_id, photo_id, photo_path, filename, location, face_conf, quality, detector, model, date_taken, date_added = face_data
quality_score = quality if quality is not None else 0.0
if quality_score >= min_quality_score:
filtered_faces_with_dates.append((face_data, date_taken))
filtered_faces_with_dates.sort(key=lambda x: x[1] or "", reverse=False)
filtered_faces = [x[0][:9] for x in filtered_faces_with_dates]
elif sort_by == "date_added":
filtered_faces_with_dates = []
for face_data in self.all_faces_cache:
face_id, photo_id, photo_path, filename, location, face_conf, quality, detector, model, date_taken, date_added = face_data
quality_score = quality if quality is not None else 0.0
if quality_score >= min_quality_score:
filtered_faces_with_dates.append((face_data, date_added))
filtered_faces_with_dates.sort(key=lambda x: x[1] or "", reverse=True)
filtered_faces = [x[0][:9] for x in filtered_faces_with_dates]
elif sort_by == "date_added_asc":
filtered_faces_with_dates = []
for face_data in self.all_faces_cache:
face_id, photo_id, photo_path, filename, location, face_conf, quality, detector, model, date_taken, date_added = face_data
quality_score = quality if quality is not None else 0.0
if quality_score >= min_quality_score:
filtered_faces_with_dates.append((face_data, date_added))
filtered_faces_with_dates.sort(key=lambda x: x[1] or "", reverse=False)
filtered_faces = [x[0][:9] for x in filtered_faces_with_dates]
elif sort_by == "filename":
filtered_faces.sort(key=lambda x: x[3] or "", reverse=False)
elif sort_by == "filename_desc":
filtered_faces.sort(key=lambda x: x[3] or "", reverse=True)
elif sort_by == "confidence":
filtered_faces.sort(key=lambda x: x[5] if x[5] is not None else 0.0, reverse=True)
elif sort_by == "confidence_asc":
filtered_faces.sort(key=lambda x: x[5] if x[5] is not None else 0.0, reverse=False)
# Apply batch size limit if specified
if batch_size and batch_size > 0:
filtered_faces = filtered_faces[:batch_size]
return filtered_faces
def _prefetch_identify_data(self, faces: List[Tuple]) -> Dict:
"""Pre-fetch all needed data to avoid repeated database queries"""
cache = {
@@ -1468,12 +1583,11 @@ class IdentifyPanel:
self._load_more_faces()
def _load_more_faces(self):
"""Load more faces if available"""
# Get current date filters
date_from = self.components['date_from_var'].get().strip() or None
date_to = self.components['date_to_var'].get().strip() or None
date_processed_from = self.components['date_processed_from_var'].get().strip() or None
date_processed_to = self.components['date_processed_to_var'].get().strip() or None
"""Load more faces from cache if available"""
if not self.cache_loaded:
messagebox.showinfo("Complete", "🎉 All faces have been identified!")
self._quit_identification()
return
# Get quality filter
min_quality = self.components['quality_filter_var'].get()
@@ -1483,17 +1597,26 @@ class IdentifyPanel:
sort_display = self.components['sort_var'].get()
sort_by = self.sort_value_map.get(sort_display, "quality")
# Get more faces
more_faces = self._get_unidentified_faces(DEFAULT_BATCH_SIZE, date_from, date_to,
date_processed_from, date_processed_to,
min_quality_score, sort_by)
# Get batch size
try:
batch_size = int(self.components['batch_var'].get().strip())
except Exception:
batch_size = DEFAULT_BATCH_SIZE
if more_faces:
# Add to current faces
self.current_faces.extend(more_faces)
# Get more faces from cache (extend current batch)
current_batch_size = len(self.current_faces)
new_batch_size = current_batch_size + batch_size
more_faces = self._filter_cached_faces(min_quality_score, sort_by, new_batch_size)
if len(more_faces) > current_batch_size:
# Add new faces to current faces
new_faces = more_faces[current_batch_size:]
self.current_faces.extend(new_faces)
self.current_face_index += 1
self._update_current_face()
self._update_button_states()
print(f"✅ Loaded {len(new_faces)} more faces from cache")
else:
# No more faces
messagebox.showinfo("Complete", "🎉 All faces have been identified!")
@@ -1636,6 +1759,10 @@ class IdentifyPanel:
if hasattr(self, 'similar_face_vars'):
self.similar_face_vars = []
# Clear cache
self.all_faces_cache = []
self.cache_loaded = False
# Clear right panel content
scrollable_frame = self.components['similar_scrollable_frame']
for widget in scrollable_frame.winfo_children():
@@ -1655,59 +1782,35 @@ class IdentifyPanel:
self.components['face_canvas'].delete("all")
def _apply_date_filters(self):
"""Apply date and quality filters by jumping to next qualifying face"""
"""Apply quality and sort filters to cached data (no database access)"""
# Check if cache is loaded
if not self.cache_loaded:
messagebox.showinfo("Start Identification First",
"Please click 'Start Identification' to load faces before applying filters.")
return
# Get quality filter
min_quality = self.components['quality_filter_var'].get()
min_quality_score = min_quality / 100.0
# If we have faces loaded, find the next face that meets quality criteria
if self.current_faces:
# Start from current position and find next qualifying face
self._find_next_qualifying_face(min_quality)
else:
# No faces loaded, need to reload
date_from = self.components['date_from_var'].get().strip() or None
date_to = self.components['date_to_var'].get().strip() or None
date_processed_from = self.components['date_processed_from_var'].get().strip() or None
date_processed_to = self.components['date_processed_to_var'].get().strip() or None
# Get batch size
try:
batch_size = int(self.components['batch_var'].get().strip())
except Exception:
batch_size = DEFAULT_BATCH_SIZE
# Quality filter is already extracted above in min_quality
min_quality_score = min_quality / 100.0
# Get sort option
sort_display = self.components['sort_var'].get()
sort_by = self.sort_value_map.get(sort_display, "quality")
# Reload faces with new filters and sort option
self.current_faces = self._get_unidentified_faces(batch_size, date_from, date_to,
date_processed_from, date_processed_to,
min_quality_score, sort_by)
if not self.current_faces:
messagebox.showinfo("No Faces Found", "No unidentified faces found with the current filters.")
return
# Reset state
self.current_face_index = 0
self.face_status = {}
self.face_person_names = {}
self.face_selection_states = {}
# Pre-fetch data
self.identify_data_cache = self._prefetch_identify_data(self.current_faces)
# Find first qualifying face
self._find_next_qualifying_face(min_quality)
# Get sort option
sort_display = self.components['sort_var'].get()
sort_by = self.sort_value_map.get(sort_display, "quality")
self.is_active = True
# Get batch size
try:
batch_size = int(self.components['batch_var'].get().strip())
except Exception:
batch_size = DEFAULT_BATCH_SIZE
# Apply filtering to cached data
print("🔄 Applying filters to cached data...")
self.current_faces = self._filter_cached_faces(min_quality_score, sort_by, batch_size)
if not self.current_faces:
messagebox.showinfo("No Faces Found", "No unidentified faces found with the current date filters.")
messagebox.showinfo("No Faces Found",
f"No faces found with quality >= {min_quality}%.\n"
f"Try lowering the quality filter or click 'Start Identification' to reload from database.")
return
# Reset state
@@ -1724,6 +1827,7 @@ class IdentifyPanel:
self._update_button_states()
self.is_active = True
print(f"✅ Applied filters: {len(self.current_faces)} faces (from {len(self.all_faces_cache)} total cached)")
def _find_next_qualifying_face(self, min_quality: int):
"""Find the next face that meets the quality criteria"""