Completing the Application
Note: In this article, real people’s faces and names in illustrations have been intentionally obscured to protect privacy
Continuing our series, in this post we’ll complete the basic Application. We replace the last two menu placeholders with valuable search functions. Interactive recognition of a the person in a single photo. And searching for images of a known person inside a folder (optionally including the subfolders).
Before adding the actual menu handlers for these functions, we first make a small adjustment to the menu bar. So it behaves in a more intuitive, user-friendly way.
Clean Menubar
Here’s a clean, ‘makes-sense-to-a-user’ menu layout that scales with the application and keeps related actions grouped together. It’s based on how people think about the workflow. Set up and manage the knowledge base, then manage people, then recognize or search, and finally view logs or access help.
In ui.py we adjust the method _make_menubar() to create the menubar, including the keyboard shorcuts, tool tips and additional info with a small helper as shown below.
# menu to manage application functions
def _make_menubar(self):
mb = self.menuBar()
# helper: create action, set tips/shortcut, add to menu, keep reference
def act(menu, text, slot, *, shortcut=None, tip=None, status=None, key=None):
a = QtGui.QAction(text, self)
a.triggered.connect(slot)
if shortcut:
a.setShortcut(QtGui.QKeySequence(shortcut))
a.setShortcutContext(QtCore.Qt.ShortcutContext.ApplicationShortcut)
self.addAction(a) # ensures shortcut works even when a widget has focus
if tip:
a.setToolTip(tip)
if status:
a.setStatusTip(status)
menu.addAction(a)
if key:
setattr(self, key, a) # keep a ref (prevents GC + enables later edits)
return a
# Settings
m = mb.addMenu("Settings")
act(m, "Preferences", self._on_prefs, shortcut="Ctrl+P", tip="Open preferences",
status="Edit folders, thresholds, and model settings", key="act_prefs")
m.addSeparator()
act(m, "Quit", self.close, shortcut="Ctrl+Q", tip="Quit the application",
status="Close the application", key="act_quit")
# Knowledge Base
m = mb.addMenu("Knowledge Base")
act(m, "Initialize KB", self._kb_init, shortcut="Ctrl+I", tip="Create a new knowledge base from the dataset folder",
status="Build encodings.pkl from the persons dataset", key="act_kb_init")
act(m, "Extend KB from Folder…", self._kb_extend, shortcut="Ctrl+E", tip="Bulk-add new persons from a folder",
status="Adds new person folders and encodes them", key="act_kb_extend")
m.addSeparator()
act(m, "Show KB Stats…", self.menu_kb_stats, shortcut="Ctrl+S", tip="Show knowledge base statistics",
status="Displays counts and file size", key="act_kb_stats")
# Persons
m = mb.addMenu("&Persons")
act(m, "Add Person…", self.menu_add_person, shortcut="Ctrl+A", tip="Add a single person from a folder",
status="Copies images and appends encodings", key="act_add_person")
act(m, "Rename Person…", self.menu_rename_person, shortcut="Ctrl+R", tip="Rename a person in the dataset and encodings", status="Renames folder + updates name labels", key="act_rename_person")
act(m, "Remove Person…", self.menu_remove_person, shortcut="Ctrl+D", tip="Remove a person from the dataset and encodings", status="Deletes folder and removes encodings", key="act_remove_person")
act(m, "Re-encode Person…", self.menu_reencode_person, shortcut="Ctrl+E", tip="Rebuild encodings for a person from their folder", status="Clears old vectors and re-encodes all", key="act_reencode_person")
# Recognition
m = mb.addMenu("Recognition")
act(m, "Interactive Identify…", self.identify, shortcut="Ctrl+I", tip="Identify a single image interactively",
status="Shows face/body matches and final decision", key="act_identify")
act(m, "Find Person in Folder…", self.search_folder, shortcut="Ctrl+F", tip="Search for a person in a folder (optional recursive)", status="Copies matched images to a results folder", key="act_search_folder")
m.addSeparator()
act(m, "Recognize Folder…", self._recognize_folder, shortcut="Ctrl+R", tip="Batch recognize all images in a folder",
status="Writes/copies results using the current KB", key="act_recognize_folder")
# View
m = mb.addMenu("View")
act(m, "Clear Log", self.console.clear, shortcut="Ctrl+C", tip="Clear the log output",
status="Clears the log window", key="act_clear_log")
# Help
m = mb.addMenu("Help")
act(m, "About…", self._on_about, shortcut="Ctrl+A", tip="About this application",
status="Show application info and credits", key="act_about")
This approach keeps the menu definition compact while still allowing full control over shortcuts, tooltips, and status messages.
Implementing the new Functions
We now have a fully functional menu for our Application with every item working as intended. Highlighted in the image below are the two newly implemented functions:
- Interactive Identify, which takes a photo, analyzes face & body and then looks for the clothest match (above a set treshold) in the KB or, when failing to do so, labels it a ‘unknown’.
- Find Person in Folder, that selects a known person from the KB, then picks a directory and tries to find every image of this person in this folder including the subfolders (when ‘recursive’ is checked). Found images are copied into a folder named Results for Name under the root folder from which the search was started.
We’ll now take a closer look at how these functions are implemented.
Adding Interactive Recognition
First we replace the place holder Interactive Identify with an actual implementation The menu action already triggers a handler method in ui.py, but the real work happens in a new single-image recognition function added to kb.py. This function is UI-agnostic and purely focused on processing. The GUI merely calls it and displays the results. To keep the interface responsive, the recognition work runs in a separate worker thread.
Handler Method Identify
In ui.py, after restructuring the menu, we implement the menu handler method for Interactive Identify. This method:
- Opens a file dialog to select an image;
- Logs the action immediately;
- Spins up a worker thread;
- Displays a result dialog when finished.
# identify a single image interactively from the KB
def identify(self):
dlg = QFileDialog(self)
dlg.setNameFilter("Images (*.png *.jpg *.jpeg *.webp)")
if not dlg.exec():
return
files = dlg.selectedFiles()
if not files:
return
image_path = files[0]
_, kb_path = self._kb_paths()
# log immediately, then yield to paint
self.log(f"Identifying Image file: {image_path}")
QtWidgets.QApplication.processEvents()
# spin up worker
self._ithread = QtCore.QThread(self)
self._iworker = IdentifyWorker(
image_path=image_path,
kb_path=kb_path,
reid_model=self.cfg.reid_model,
face_tol=float(self.cfg.face_tol),
body_tol=float(self.cfg.body_tol),
valid_exts=tuple(self.cfg.valid_exts),
topk=3,
)
self._iworker.moveToThread(self._ithread)
# optional: busy cursor + status text
QtWidgets.QApplication.setOverrideCursor(QtCore.Qt.CursorShape.WaitCursor)
self.statusBar().showMessage("Identifying…")
def _on_done(result: dict):
# restore UI state
QtWidgets.QApplication.restoreOverrideCursor()
self.statusBar().showMessage("Ready")
# build your summary text (reuse your existing code)
def fmt(lines): return "\n".join(f"{n}: {s}" for n, s in (lines or [])) or "None"
diag = result.get("diagnostics", {})
summary = (
f"Final: {result['final_name']} (by {diag.get('decision','none')})\n\n"
f"Top Face Matches (distance):\n{fmt(result['face_result'])}\n\n"
f"Top Body Matches (cosine):\n{fmt(result['body_result'])}\n\n"
f"Timings (ms): open {diag.get('timings_ms',{}).get('open','n/a')}, "
f"face {diag.get('timings_ms',{}).get('face','n/a')}, "
f"body {diag.get('timings_ms',{}).get('body','n/a')}, "
f"total {diag.get('timings_ms',{}).get('total','n/a')}"
)
# show your result dialog
d = IdentifyResultDialog(image_path, summary, result, self)
d.exec()
def _on_err(msg: str):
QtWidgets.QApplication.restoreOverrideCursor()
self.statusBar().showMessage("Ready")
QMessageBox.warning(self, "Recognition Failed", msg)
self._ithread.started.connect(self._iworker.run)
self._iworker.finished.connect(_on_done)
self._iworker.error.connect(_on_err)
# cleanup
self._iworker.finished.connect(self._ithread.quit)
self._iworker.error.connect(self._ithread.quit)
self._ithread.finished.connect(lambda: setattr(self, "_ithread", None))
self._ithread.start()
Although this handler looks quite large, most of it is thread management and UI bookkeeping.
The Identify Worker
Because face and body encoding—and subsequent matching against the knowledge base—can take noticeable time, we run this logic in a worker thread to avoid freezing the GUI. The worker class itself is simple: it calls a single function (recognize_single_image) and emits either a result or an error.
Identify Result Dialog
Finally, we add a small dialog to display the recognition result. It shows:
- the selected image
- top face matches (distance)
- top body matches (cosine similarity)
- the final decision and timing diagnostics
For convenience we added also buttons to copy the result JSON and open the image’s folder to the dialog.
class IdentifyResultDialog(QtWidgets.QDialog):
def __init__(self, image_path: str, summary_text: str, result: dict | None = None, parent=None):
super().__init__(parent)
self.setWindowTitle("Recognition Result")
self.resize(900, 600)
layout = QtWidgets.QVBoxLayout(self)
# --- top: image + details
hl = QtWidgets.QHBoxLayout()
img_label = QtWidgets.QLabel(self)
img_label.setAlignment(QtCore.Qt.AlignmentFlag.AlignCenter)
pix = QtGui.QPixmap(image_path)
img_label.setPixmap(
pix.scaled(600, 600,
QtCore.Qt.AspectRatioMode.KeepAspectRatio,
QtCore.Qt.TransformationMode.SmoothTransformation)
)
hl.addWidget(img_label, 2)
details = QtWidgets.QTextEdit(self)
details.setReadOnly(True)
details.setPlainText(summary_text)
hl.addWidget(details, 1)
layout.addLayout(hl)
# --- bottom: buttons
buttons = QtWidgets.QDialogButtonBox(QtWidgets.QDialogButtonBox.StandardButton.Close, parent=self)
layout.addWidget(buttons)
# optional: Copy JSON (only if result dict provided)
if isinstance(result, dict):
btn_copy = QtWidgets.QPushButton("Copy JSON", self)
buttons.addButton(btn_copy, QtWidgets.QDialogButtonBox.ButtonRole.ActionRole)
def _copy_json():
import json
QtGui.QGuiApplication.clipboard().setText(json.dumps(result, indent=2, ensure_ascii=False))
btn_copy.clicked.connect(_copy_json)
# Open image folder
btn_open = QtWidgets.QPushButton("Open Image Folder", self)
buttons.addButton(btn_open, QtWidgets.QDialogButtonBox.ButtonRole.ActionRole)
def _open_folder():
QtGui.QDesktopServices.openUrl(QtCore.QUrl.fromLocalFile(os.path.dirname(image_path)))
btn_open.clicked.connect(_open_folder)
# Close behavior
buttons.rejected.connect(self.reject)
buttons.accepted.connect(self.reject) # Close acts like reject here
Processing in kb.py
All actual recognition logic lives in kb.py. The function recognize_single_image() makes use of several helper methods:
- a cached TorchReID body extractor
- a face embedding helper
- face and body “bank builders”
- vectorized comparison utilities
@lru_cache(maxsize=2)
def _cached_body_extractor(model_name: str):
# Lazily build TorchReID extractor once per model name
from reid_wrapper import TorchreidBodyExtractor
return TorchreidBodyExtractor(model_name=model_name, log_fn=None)
def _face_embed_from_image(img: Image.Image) -> np.ndarray | None:
# use the existing face pipeline
from face_encoder import detect_and_align_faces
chips = detect_and_align_faces(img, compute_embedding=True, embedding_model="small") or []
for r in chips:
emb = r.get("embedding")
if emb is not None and getattr(emb, "size", 0):
return np.asarray(emb, dtype=np.float32)
return None
# -----------------------------------------------------------------------------
# Bank builders (dimension-agnostic) reused everywhere
# -----------------------------------------------------------------------------
def build_face_bank(db: Dict[str, Any]) -> Tuple[np.ndarray, np.ndarray]:
vecs, names = [], []
for v, n in zip(db.get("face_encodings", []), db.get("face_names", [])):
a = np.asarray(v, dtype=np.float32).reshape(-1)
if np.isfinite(a).all():
vecs.append(a); names.append(n)
if vecs:
F = np.vstack(vecs).astype(np.float32)
else:
F = np.empty((0, 0), np.float32)
return F, np.asarray(names)
def build_body_bank(db: Dict[str, Any]) -> Tuple[np.ndarray, np.ndarray, int]:
vecs, names = [], []
for v, n in zip(db.get("body_encodings", []), db.get("body_names", [])):
a = np.asarray(v, dtype=np.float32).reshape(-1)
if np.isfinite(a).all():
vecs.append(a); names.append(n)
if vecs:
B = np.vstack(vecs).astype(np.float32)
dim = int(B.shape[1])
else:
B = np.empty((0, 0), np.float32); dim = 0
return B, np.asarray(names), dim
Fast Matrix Comparison
After normalizing the body bank we use the face and body banks for fast matrix comparison (instead of much slower loop processing).
# -----------------------------------------------------------------------------
# Unified scoring helpers (top-k + pass/fail with gap/ratio rules)
# -----------------------------------------------------------------------------
# np.argpartition avoids sorting the entire array (O(n log n)) when you only need top-k
# for k≈3–5 and n≥5k, this is often 2–6× faster than a full argsort
def topk_face(F: np.ndarray, F_names: np.ndarray, q: np.ndarray, k: int = 3):
if F.size == 0:
return [], None, None, None
d = np.linalg.norm(F - q[None, :], axis=1) # (N,)
n = d.shape[0]
if k >= n:
order = np.argsort(d) # fully sort if tiny bank
else:
idx = np.argpartition(d, k)[:k] # O(n)
order = idx[np.argsort(d[idx])] # sort only the k items
return [(F_names[i], float(d[i])) for i in order], d, order, float(d[order[0]])
def topk_body(B: np.ndarray, B_names: np.ndarray, q: np.ndarray, k: int = 3):
if B.size == 0: return [], None, None, None
qt = torch.from_numpy(q[None, :]).float()
tt = torch.from_numpy(B).float()
sims = torch.nn.functional.cosine_similarity(qt, tt).cpu().numpy()
idx = np.argsort(-sims)[:min(k, len(sims))]
return [(B_names[i], float(sims[i])) for i in idx], sims, idx, float(sims[idx[0]])
def pass_face(d_all, F_names, target, tol, gap, relax, ratio_max):
is_t = (F_names == target)
d_t = d_all[is_t]; d_o = d_all[~is_t]
if not d_t.size: return False, None, None, None
best_t = float(np.min(d_t))
imp = float(np.min(d_o)) if d_o.size else 1.0
g = imp - best_t
r = best_t / max(imp, 1e-6)
ok = (best_t <= tol and g >= gap) or (best_t <= tol + relax and r <= ratio_max)
return ok, best_t, imp, (g, r)
def pass_body(sims_all, B_names, target, tol, gap, relax, ratio_min):
is_t = (B_names == target)
s_t = sims_all[is_t]; s_o = sims_all[~is_t]
if not s_t.size: return False, None, None, None
best_t = float(np.max(s_t))
imp = float(np.max(s_o)) if s_o.size else -1.0
g = best_t - imp
r = best_t / max(imp, 1e-6)
ok = (best_t >= tol and g >= gap) or (best_t >= tol - relax and best_t >= imp * ratio_min)
return ok, best_t, imp, (g, r)
# Normalize body bank once; switch to dot-product cosine
# used in PersonSearchWorker and recognize_single_image()
def _normalize_rows(X: np.ndarray, eps: float = 1e-12) -> np.ndarray:
if X.size == 0:
return X
n = np.linalg.norm(X, axis=1, keepdims=True).astype(np.float32)
np.maximum(n, eps, out=n) # avoid divide-by-zero
return X / n
def _normalize_vec(q: np.ndarray, eps: float = 1e-12) -> np.ndarray:
n = float(np.linalg.norm(q))
if n < eps:
return q.astype(np.float32)
return (q / n).astype(np.float32)
def topk_body_dot(B_unit: np.ndarray, B_names: np.ndarray, q: np.ndarray, k: int = 3):
"""B_unit rows must already be L2-normalized. Returns standard (labels,score) list and arrays."""
if B_unit.size == 0:
return [], None, None, None
q_unit = _normalize_vec(q)
sims = B_unit @ q_unit # (N,), cosine because both sides are unit vectors
n = sims.shape[0]
if k >= n:
order = np.argsort(-sims)
else:
idx = np.argpartition(-sims, k)[:k] # O(n)
order = idx[np.argsort(-sims[idx])]
return [(B_names[i], float(sims[i])) for i in order], sims, order, float(sims[order[0]])
# -----------------------------------------------------------------------------
Manage Processing
Making use of these rather ingenious helpers, the actual method that executes processing remains relatively small.
# Single-image recognition with unified face+body logic
def recognize_single_image(
image_path: str | Path,
kb_path: str | Path,
*,
settings: Dict[str, Any] | None = None,
reid_model: str | None = None,
face_tol: float | None = None,
body_tol: float | None = None,
valid_exts: Tuple[str, ...] | None = None,
topk: int = 3,
) -> Dict[str, Any]:
"""
Single-image identify using face (distance) + body (cosine) with unified rules.
Returns dict with final_name, face_result, body_result, and diagnostics.
"""
p = Path(image_path)
pr = params_from_settings(settings)
reid_name = reid_model or pr["reid_model"]
face_thr = pr["face_tol"] if face_tol is None else face_tol
body_thr = pr["body_tol"] if body_tol is None else body_tol
exts = pr["valid_exts"] if valid_exts is None else valid_exts
if not p.exists() or p.suffix.lower() not in exts:
return {"error": "Invalid or unsupported image format."}
db = kb_load(kb_path)
F, F_names = build_face_bank(db)
B, B_names, body_dim = build_body_bank(db)
B_unit = _normalize_rows(B) if B.size else B
# open once
try:
with Image.open(p) as im:
img = im.convert("RGB")
except Exception as e:
return {"error": f"Image open failed: {e}"}
# FACE
face_name = "Unknown"; face_result = []; face_top1 = None
try:
qf = face_embed(img)
if qf is not None and F.size:
face_result, d_all, _, face_top1 = topk_face(F, F_names, qf, k=topk)
# pick best identity if confident (no target identity here; single-image mode chooses top-1 below tol)
if face_result and face_top1 < face_thr:
face_name = face_result[0][0]
except Exception as e:
return {"error": f"Face recognition failed: {e}"}
# BODY
body_name = "Unknown"; body_result = []; body_top1 = None
try:
if B.size and body_dim > 0:
qv = np.asarray(body_extractor(reid_name)(img), dtype=np.float32).reshape(-1)
if qv.shape[0] == body_dim and np.isfinite(qv).all():
body_result, sims_all, _, body_top1 = topk_body_dot(B_unit, B_names, qv, k=topk)
if body_result and body_top1 > body_thr:
body_name = body_result[0][0]
except Exception as e:
return {"error": f"Body recognition failed: {e}"}
final_name = face_name if face_name != "Unknown" else body_name
decision = "face" if face_name != "Unknown" else ("body" if body_name != "Unknown" else "none")
return {
"final_name": final_name,
"face_result": face_result or [],
"body_result": body_result or [],
"face_name": face_name,
"body_name": body_name,
"diagnostics": {
"decision": decision,
"face_top1": face_top1,
"body_top1": body_top1,
"face_threshold": float(face_thr),
"body_threshold": float(body_thr),
},
}
Note the difference in scoring: face score shows distance (lower is better), body score shows cosine similarity (higher is better).
Adding Search Folder(s)
Just like interactive identification, searching through folders is best done inside a separate thread. That said and because there’s a little less work to it, we decided to manage this worker thread directly from the menu handler method.
# search for a person in a folder of images
def search_folder(self):
if not self.cfg.dataset_dir:
QtWidgets.QMessageBox.warning(self, "No KB", "Initialize the KB first.")
return
# pick person
main_dir, kb_path = self._kb_paths()
dlg = PersonSelectDialog(main_dir, "Search Person", self)
if dlg.exec() != QtWidgets.QDialog.DialogCode.Accepted:
return
name = dlg.combo.currentText()
# pick folder
search_dir = QtWidgets.QFileDialog.getExistingDirectory(self, "Select folder to search")
if not search_dir:
return
# ask for recursion (store in bool rec)
rec = QtWidgets.QMessageBox.question(self, "Recursive search?", "Include subfolders in the search?",
QtWidgets.QMessageBox.StandardButton.Yes | QtWidgets.QMessageBox.StandardButton.No
) == QtWidgets.QMessageBox.StandardButton.Yes
# log immediately
self.log(f"Searching for: {name} in: {search_dir}")
QtWidgets.QApplication.processEvents()
# busy UI hints
QtWidgets.QApplication.setOverrideCursor(QtCore.Qt.CursorShape.WaitCursor)
self.statusBar().showMessage("Searching Person Images…")
# spin up worker
archive_root = getattr(self.cfg, "persons_images_found", "") or "" # optional
self._sthread = QtCore.QThread(self)
self._sworker = PersonSearchWorker(
person_name=name,
folder_path=search_dir,
kb_path=kb_path,
reid_model=self.cfg.reid_model,
face_threshold=float(self.cfg.face_tol),
body_threshold=float(self.cfg.body_tol),
valid_exts=tuple(self.cfg.valid_exts),
recursive=rec,
archive_root=archive_root,
)
self._sworker.moveToThread(self._sthread)
# wire signals
self._sthread.started.connect(self._sworker.run)
self._sworker.progress.connect(self.progress.setValue)
self._sworker.log.connect(self.log)
def _done(msg: str):
QtWidgets.QApplication.restoreOverrideCursor()
self.statusBar().showMessage("Ready")
QMessageBox.information(self, "Search", msg)
self._sworker.finished.connect(_done)
# cleanup
self._sworker.finished.connect(self._sthread.quit)
self._sthread.finished.connect(lambda: setattr(self, "_sthread", None))
self._sthread.start()
Process Optimization
To speed up the actual comparison we once again make use of the smart helper methods for banking and matrix comparison mentioned already above. We address the purpose and workings of this optimization in handling the KB below.
# refactored to use unified helpers & params
class PersonSearchWorker(QtCore.QObject):
progress = pyqtSignal(int)
log = pyqtSignal(str)
finished = pyqtSignal(str)
def __init__(self, person_name: str, folder_path: str | Path, kb_path: str | Path, *,
reid_model: str, face_threshold: float, body_threshold: float,
valid_exts: Tuple[str, ...], recursive: bool = False,
face_gap: float = DEFAULTS["face_gap"], face_relax: float = DEFAULTS["face_relax"],
face_ratio_max: float = DEFAULTS["face_ratio_max"],
body_gap: float = DEFAULTS["body_gap"], body_relax: float = DEFAULTS["body_relax"],
body_ratio_min: float = DEFAULTS["body_ratio_min"],
archive_root: str = "", parent=None):
super().__init__(parent)
self.name = person_name
self.folder = str(folder_path)
self.kb_path = Path(kb_path)
self.reid_model = reid_model
self.face_thr = float(face_threshold)
self.body_thr = float(body_threshold)
self.valid_exts = _norm_exts(valid_exts)
self.recursive = bool(recursive)
self.face_gap, self.face_relax, self.face_ratio_max = face_gap, face_relax, face_ratio_max
self.body_gap, self.body_relax, self.body_ratio_min = body_gap, body_relax, body_ratio_min
self.archive_root = archive_root
@QtCore.pyqtSlot()
def run(self):
try:
# Collect files (recursive or single folder)
if not os.path.isdir(self.folder):
self.finished.emit("Selected path is not a folder."); return
if self.recursive:
file_paths = []
for root, _, fns in os.walk(self.folder):
for fn in fns:
if fn.lower().endswith(self.valid_exts):
file_paths.append(os.path.join(root, fn))
else:
file_paths = [os.path.join(self.folder, fn)
for fn in os.listdir(self.folder)
if fn.lower().endswith(self.valid_exts)]
total = len(file_paths)
if not total:
self.finished.emit("No images found in selected folder."); return
# Load KB + banks
db = kb_load(self.kb_path)
F, F_names = build_face_bank(db)
B, B_names, body_dim = build_body_bank(db)
B_unit = _normalize_rows(B) if B.size else B
tgt_has_face = np.any(F_names == self.name)
tgt_has_body = np.any(B_names == self.name)
if not (tgt_has_face or tgt_has_body):
self.finished.emit(f"No face or body data found for '{self.name}'."); return
# Per-run results folder at the *selected* root. We’ll preserve the
# subfolder structure under this root when recursive=True.
results_root = os.path.join(self.folder, f"results_{self.name}")
os.makedirs(results_root, exist_ok=True)
ts = datetime.datetime.now().strftime("%Y-%m-%d_%H-%M")
arch_dir = ""
if self.archive_root:
os.makedirs(self.archive_root, exist_ok=True)
arch_dir = os.path.join(self.archive_root, f"{self.name}_{ts}")
os.makedirs(arch_dir, exist_ok=True)
# ReID once (if needed)
reid = body_extractor(self.reid_model) if (np.any(B_names == self.name) and body_dim > 0 and B.shape[0] > 0) else None
matched = 0
# invariants (tiny micro-opts)
has_target_face = bool(F.size) and np.any(F_names == self.name)
has_body_bank = bool(B.size) and body_dim > 0
for i, path in enumerate(file_paths, 1):
fn = os.path.basename(path)
rel = os.path.relpath(path, self.folder)
# open safely
try:
with Image.open(path) as im:
img = im.convert("RGB")
except Exception as e:
self.log.emit(f"[Open] {fn}: {e}")
self.progress.emit(int(i * 100 / total))
continue
face_match = False
body_match = False
# ----- FACE (distance: lower is better)
if has_target_face:
try:
qf = face_embed(img)
if qf is not None:
_, d_all, order, _ = topk_face(F, F_names, qf, k=3)
ok, best_t, imp, (gap, ratio) = pass_face(
d_all, F_names, self.name,
self.face_thr, self.face_gap, self.face_relax, self.face_ratio_max
)
face_match = bool(ok)
# debug top3 (use order from topk_face)
tops = ", ".join(f"{F_names[j]}:{d_all[j]:.3f}" for j in (order[:3] if order is not None else []))
self.log.emit(f"[FaceScore] {rel}: "
f"best_t={best_t:.4f}, imp_best={imp:.4f}, gap={gap:.4f}, ratio={ratio:.3f} | top3: {tops}")
except Exception as e:
self.log.emit(f"[Face] {fn}: {e}")
# ----- BODY (cosine: higher is better)
if reid is not None and has_body_bank:
try:
qv = np.asarray(reid(img), dtype=np.float32).reshape(-1)
if qv.shape[0] == body_dim and np.isfinite(qv).all():
body_result, sims, order, body_top1 = topk_body_dot(B_unit, B_names, qv, k=3)
ok, best_t, imp, (gap, ratio) = pass_body(
sims, B_names, self.name,
self.body_thr, self.body_gap, self.body_relax, self.body_ratio_min
)
body_match = bool(ok)
tops = ", ".join(f"{B_names[j]}:{sims[j]:.3f}" for j in (order[:3] if order is not None else []))
self.log.emit(
f"[BodyScore] {rel}: "
f"best_t={best_t:.4f}, imp_best={imp:.4f}, gap={gap:.4f}, ratio={ratio:.3f} | top3: {tops}"
)
except Exception as e:
self.log.emit(f"[Body] {fn}: {e}")
if face_match or body_match:
matched += 1
self.log.emit(f"[Match] {rel} → {self.name}")
# Preserve subfolders when recursive; in flat mode rel==filename
dst = os.path.join(results_root, rel)
os.makedirs(os.path.dirname(dst), exist_ok=True)
try:
shutil.copy2(path, dst)
except Exception as e:
self.log.emit(f"[Copy] {rel}: {e}")
self.progress.emit(int(i * 100 / total))
mode = "recursive" if self.recursive else "single folder"
summary = f"Search for '{self.name}' ({mode}): {matched} of {total} images matched."
self.log.emit(summary)
self.finished.emit(summary)
except Exception as e:
self.finished.emit(f"ERROR: {e}")
Housekeeping KB.py
After implementing these two new search & compare functions with all the helper methods involved on top of those already there, kb.py could use a little housekeeping.
We focused first on deduplication of functions, unifying similar methods for data loading & saving and ensuring proper use of context managers. Thus we went from three loading and two saving methods to one pair of canonical data handling methods.
Some code duplication between the methods add_person_from_folder and reencode_person was also solved with a small helper method.
Enhancing KB usage
Considering code efficiency and performance enhancement, we decided to add caching, streamline the normalization process of embeddings and switching from lists to numpy arrays with memory mapping to optimize memory usage. Looping over encodings in Python is slow for a simple reason. The expensive part isn’t the math, it’s the Python overhead per comparison (function calls, attribute lookups, branching, list indexing). Once you move the bank into a single numpy array, you can let optimized the C/BLAS code from the library chew through thousands of comparisons in one go.
Consider the following code examples.
# Face (distance: lower is better)
# Loop style (bad for performance):
best = 1e9
for v in face_bank: # Python loop = overhead per vector
d = np.linalg.norm(v - q) # called N times
best = min(best, d)
# Vectorized (fast):
d = np.linalg.norm(face_bank - q[None, :], axis=1) # one call only
best = d.min()
#Body (cosine similarity: higher is better)
# If you pre-normalize the bank once (B_unit) and normalize each query once (q_unit),
# cosine similarity becomes a dot product:
sims = B_unit @ q_unit # (N,)
best = sims.max()
# That dot product in the library again hits highly optimized matrix–vector code (BLAS).
# It can be shockingly fast even on CPU.
Normalization & Top-k argpartition
Why normalization matters (especially for body). Without pre-normalization, cosine similarity needs two norms per vector: cos(q,b) = dot(q,b) / (||q||·||b||).
If you normalize bank rows once, you remove the ||b|| part from every future comparison. Then each query only pays one ||q|| normalization, one dot product with the whole bank.
That turns per-vector work into a one-time preprocessing step plus a cheap per-query dot product.
Practical effect being that in a recursive search you might do hundreds (or even thousands) of comparisons per run – where standard Python loops will crawl.
Top-k argpartition also is much faster than argsort. When you only need the best few matches, say 3, sorting the full array is wasted work. An argsort sorts all N items → O(N log N) while argpartition finds the 3 best without fully sorting.
What doesn’t speed up alas, is the encoding step, this is still the heavyweight. Face embedding (dlib/face_recognition) and ReID inference (TorchReID) dominate per-image time. However, vectorized comparison speeds up the matching step, which becomes more significant when your bank grows and you search lots of images.
Settings and Constants
Part of the housekeeping was a decision to make stricter use of configuration constants. We now manage constants in one place and then reference these constants everywhere. Instead of having valid extensions for image files appear as literals in multiple functions. Also threshold/gap/relax defaults were spread across functions. We fixed this by defining module-level constants for kb.py (that can be overridden with the settings from settings.json at runtime).
The configuration file settings.json now hold the following constants:
{
"dataset_dir": "G:/Coding/recognize/posts/app/persons_dataset",
"processed_dir": "G:/Coding/recognize/posts/app/persons_processed",
"process_dir": "G:/Coding/recognize/posts/app/persons_unknown",
"valid_exts": [".jpg", ".jpeg", ".png", ".webp"],
"reid_model": "osnet_ain_x1_0",
"kb_batch_size": 16,
"encodings_filename": "encodings.pkl",
"resize_max": 800,
"face_tol": 0.40,
"face_gap": 0.06,
"face_relax": 0.03,
"face_ratio_max": 0.92,
"body_tol": 0.80,
"body_gap": 0.05,
"body_relax": 0.03,
"body_ratio_min": 1.03
}
We adjusted both config.py and the Preferences dialog accordingly.
New Downloads Available
In the last post we presented a package of source files for the application with instructions for setting things up. Since that previous post, the following files have changed:
- ui.py, menu bar, menu handler methods for identification and folder search;
- kb.py, actual searching & comparison plus housekeeping;
- config.py: settings handling
We’ve therefore created an updated download package containing only these modified files, which you can use to replace the earlier versions.
If you like this post and the provided code, consider making a donation to help cover the hosting expenses.