Optimization And Visualization Plan
jhj
2026년 08월 12일
FAISS CPU Scaling, YOLO FP16/ONNX, and Results.plot() Implementation Plan
For agentic workers: REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (
- [ ]) syntax for tracking.
Goal: Implement CPU-based multi-index scaling (IndexFlatIP, IndexIVFFlat, IndexHNSWFlat) in FaissMatcher, enable FP16/ONNX inference in YoloPredictor, and fully implement Results.plot() for OpenCV BGR visualization.
Architecture: Extend FaissMatcher with CPU index selection logic and automatic vector count fallbacks. Add yolo_fp16 and ONNX model loading in YoloPredictor. Complete Results.plot() in reid/core/types.py with standalone OpenCV bounding box and label text rendering.
Tech Stack: Python 3.10+, FAISS (CPU), PyTorch, OpenCV, Ultralytics YOLOv8, Pytest.
(Note per user request: Git commits will be held and executed step-by-step after all implementation tasks and verification are complete).
File Structure & Responsibilities
- Modify:
reid/cfg/default.yaml- Add default configuration parameters for FAISS matcher index types and detector FP16 options. - Modify:
reid/core/config.py- Add dataclass field definitions formatcher_index_type,faiss_nlist,faiss_nprobe,faiss_hnsw_m,faiss_min_ivf_vectors, andyolo_fp16. - Modify:
reid/models/matcher/faiss.py- Implement multi-index CPU FAISS matching with automatic fallback logic. - Modify:
reid/models/yolo/detect/predict.py- Add FP16 (half=True) handling and optional ONNX model initialization. - Modify:
reid/core/types.py- ImplementResults.plot()for visual annotation and returnnp.ndarray. - Create:
tests/test_faiss_matcher.py- Unit tests for FAISS index selection, fallbacks, and matching precision. - Create:
tests/test_results_plot.py- Unit tests forResults.plot()image annotation generation.
Task 1: Add Configuration Options
Files:
- Modify:
reid/cfg/default.yaml - Modify:
reid/core/config.py -
Test:
tests/test_config.py - Step 1: Write the test for new config fields
import pytest
from reid.core.config import ReidConfig
def test_new_config_defaults():
cfg = ReidConfig()
assert cfg.matcher_index_type == "flat"
assert cfg.faiss_nlist == 32
assert cfg.faiss_nprobe == 4
assert cfg.faiss_hnsw_m == 32
assert cfg.faiss_min_ivf_vectors == 100
assert cfg.yolo_fp16 is False
- Step 2: Run test to verify it fails
Run: pytest tests/test_config.py
Expected: FAIL with AttributeError: 'ReidConfig' object has no attribute 'matcher_index_type'
- Step 3: Implement config updates
In reid/cfg/default.yaml:
# Matcher Index Parameters
matcher_index_type: "flat" # Options: "flat", "ivfflat", "hnsw"
faiss_nlist: 32
faiss_nprobe: 4
faiss_hnsw_m: 32
faiss_min_ivf_vectors: 100
# Detector FP16
yolo_fp16: False
In reid/core/config.py:
Add fields to ReidConfig:
matcher_index_type: str = "flat"
faiss_nlist: int = 32
faiss_nprobe: int = 4
faiss_hnsw_m: int = 32
faiss_min_ivf_vectors: int = 100
yolo_fp16: bool = False
- Step 4: Run test to verify it passes
Run: pytest tests/test_config.py
Expected: PASS
Task 2: Implement CPU Multi-Index FAISS Matcher
Files:
- Modify:
reid/models/matcher/faiss.py -
Create:
tests/test_faiss_matcher.py - Step 1: Write unit tests for FAISS CPU multi-index strategy
Write tests/test_faiss_matcher.py:
import pytest
import numpy as np
from reid.models.matcher.faiss import FaissMatcher
from reid.core.config import ReidConfig
def test_faiss_matcher_flat():
matcher = FaissMatcher(threshold=0.5)
embeddings = np.random.randn(20, 128).astype(np.float32)
embeddings /= np.linalg.norm(embeddings, axis=1, keepdims=True)
labels = [f"cat_{i%5}" for i in range(20)]
matcher.fit(embeddings, labels)
res = matcher.match(embeddings[0])
assert res.is_known is True
assert res.cat_id == "cat_0"
def test_faiss_matcher_ivfflat():
cfg = ReidConfig(matcher_index_type="ivfflat", faiss_min_ivf_vectors=10, faiss_nlist=4)
matcher = FaissMatcher(threshold=0.5)
matcher.cfg = cfg
embeddings = np.random.randn(50, 128).astype(np.float32)
embeddings /= np.linalg.norm(embeddings, axis=1, keepdims=True)
labels = [f"cat_{i%5}" for i in range(50)]
matcher.fit(embeddings, labels)
res = matcher.match(embeddings[0])
assert res.is_known is True
assert res.cat_id == "cat_0"
def test_faiss_matcher_hnsw():
cfg = ReidConfig(matcher_index_type="hnsw", faiss_hnsw_m=16)
matcher = FaissMatcher(threshold=0.5)
matcher.cfg = cfg
embeddings = np.random.randn(30, 128).astype(np.float32)
embeddings /= np.linalg.norm(embeddings, axis=1, keepdims=True)
labels = [f"cat_{i%5}" for i in range(30)]
matcher.fit(embeddings, labels)
res = matcher.match(embeddings[0])
assert res.is_known is True
assert res.cat_id == "cat_0"
def test_faiss_matcher_ivf_fallback():
# If vectors count < faiss_min_ivf_vectors, should fallback to FlatIP
cfg = ReidConfig(matcher_index_type="ivfflat", faiss_min_ivf_vectors=100)
matcher = FaissMatcher(threshold=0.5)
matcher.cfg = cfg
embeddings = np.random.randn(20, 128).astype(np.float32)
embeddings /= np.linalg.norm(embeddings, axis=1, keepdims=True)
labels = [f"cat_{i%5}" for i in range(20)]
matcher.fit(embeddings, labels)
assert matcher.actual_index_type == "flat"
- Step 2: Run test to verify it fails
Run: pytest tests/test_faiss_matcher.py
Expected: FAIL due to missing attributes or unhandled matcher_index_type
- Step 3: Implement multi-index FAISS logic in
reid/models/matcher/faiss.py
Replace reid/models/matcher/faiss.py with:
import numpy as np
import faiss
from typing import List, Optional
from reid.models.matcher.model import BaseMatcher
from reid.core.types import MatchResult
class FaissMatcher(BaseMatcher):
"""
High-performance CPU Matcher using FAISS supporting FlatIP, IVFFlat, and HNSWFlat.
"""
def __init__(self, threshold: float = 0.7):
super().__init__(threshold)
self.index = None
self.is_fitted = False
self.actual_index_type = "flat"
def fit(self, embeddings: np.ndarray, labels: List[str]):
if len(embeddings) == 0:
self.is_fitted = False
return
d = embeddings.shape[1]
embeddings_f32 = np.ascontiguousarray(embeddings.astype('float32'))
index_type = getattr(self.cfg, "matcher_index_type", "flat").lower()
min_ivf_vecs = getattr(self.cfg, "faiss_min_ivf_vectors", 100)
num_vecs = len(embeddings_f32)
# Decide actual index type with safety fallbacks
if index_type == "ivfflat" and num_vecs < min_ivf_vecs:
actual_type = "flat"
else:
actual_type = index_type
self.actual_index_type = actual_type
if actual_type == "ivfflat":
nlist = getattr(self.cfg, "faiss_nlist", 32)
nlist = min(nlist, max(1, num_vecs // 4))
quantizer = faiss.IndexFlatIP(d)
self.index = faiss.IndexIVFFlat(quantizer, d, nlist, faiss.METRIC_INNER_PRODUCT)
self.index.train(embeddings_f32)
self.index.add(embeddings_f32)
nprobe = getattr(self.cfg, "faiss_nprobe", 4)
self.index.nprobe = min(nprobe, nlist)
elif actual_type == "hnsw":
hnsw_m = getattr(self.cfg, "faiss_hnsw_m", 32)
self.index = faiss.IndexHNSWFlat(d, hnsw_m, faiss.METRIC_INNER_PRODUCT)
self.index.add(embeddings_f32)
else: # "flat" fallback
self.index = faiss.IndexFlatIP(d)
self.index.add(embeddings_f32)
self.labels = labels
self.is_fitted = True
def match(self, query_embedding: np.ndarray) -> MatchResult:
if not self.is_fitted or self.index is None:
return MatchResult(cat_id="Unknown", similarity=0.0, is_known=False)
if query_embedding.ndim == 1:
query_embedding = query_embedding.reshape(1, -1)
query_f32 = np.ascontiguousarray(query_embedding.astype('float32'))
K = int(getattr(self.cfg, "k", 5))
K = min(K, len(self.labels))
if K <= 0:
K = 1
similarities, indices = self.index.search(query_f32, K)
votes = {}
for i in range(K):
idx = indices[0][i]
if idx < 0 or idx >= len(self.labels):
continue
sim = float(similarities[0][i])
label: str = self.labels[idx]
if label not in votes:
votes[label] = {"count": 0, "sum_sim": 0.0}
votes[label]["count"] += 1
votes[label]["sum_sim"] += sim
if not votes:
return MatchResult(cat_id="Unknown", similarity=0.0, is_known=False)
best_label = "Unknown"
best_score = -1.0
best_avg_sim = 0.0
for label, info in votes.items():
avg_sim = info["sum_sim"] / info["count"]
score = info["sum_sim"]
if score > best_score:
best_score = score
best_label = label
best_avg_sim = avg_sim
is_known = best_avg_sim >= self.threshold
cat_id = best_label if is_known else "Unknown"
return MatchResult(cat_id=cat_id, similarity=float(best_avg_sim), is_known=is_known)
- Step 4: Run test to verify it passes
Run: pytest tests/test_faiss_matcher.py
Expected: PASS
Task 3: Add FP16 & ONNX Detector Integration
Files:
- Modify:
reid/models/yolo/detect/predict.py -
Test:
tests/test_yolo_predict.py - Step 1: Write test for YOLO FP16 / ONNX config handling
Write tests/test_yolo_predict.py:
import pytest
from reid.core.config import ReidConfig
from reid.models.yolo.detect.predict import YoloPredictor
def test_yolo_predictor_config_init():
cfg = ReidConfig(yolo_fp16=True, use_onnx=False)
predictor = YoloPredictor(config=cfg)
assert getattr(predictor.cfg, "yolo_fp16", False) is True
- Step 2: Run test to verify it fails
Run: pytest tests/test_yolo_predict.py
Expected: FAIL if YoloPredictor does not accept config properly or crashes
- Step 3: Update
YoloPredictorinreid/models/yolo/detect/predict.py
In reid/models/yolo/detect/predict.py:
from typing import Any, List, Optional
import torch
from reid.engine.predictor import BasePredictor
from reid.core.types import Results, BBox
from reid.utils import get_cfg_path
class YoloPredictor(BasePredictor):
"""
YOLOv8 Predictor following the BasePredictor interface with FP16 and ONNX support.
"""
def preprocess(self, im: Any) -> Any:
return im
def inference(self, im: Any) -> Any:
device = getattr(self, "device", "cpu")
use_fp16 = getattr(self.cfg, "yolo_fp16", False) or getattr(self.cfg, "fp16", False)
half = use_fp16 and ("cuda" in str(device))
if self.cfg.track:
tracker = get_cfg_path(self.cfg.tracker)
self.cfg.tracker = str(tracker) if tracker else "bytetrack.yaml"
return self.model.track(
im,
conf=self.cfg.conf,
iou=self.cfg.iou,
persist=True,
tracker=self.cfg.tracker,
device=device,
half=half,
verbose=False
)
else:
return self.model(im, conf=self.cfg.conf, iou=self.cfg.iou, device=device, half=half, verbose=False)
def postprocess(self, preds: Any, img: Any, orig_img: Any, target_id: Optional[List[int]] = [15]) -> Optional[Results]:
ultra_res = preds[0]
results = Results(orig_img=ultra_res.orig_img, path=ultra_res.path)
cpu_boxes = ultra_res.boxes.cpu() if hasattr(ultra_res.boxes, 'cpu') else ultra_res.boxes
for box in cpu_boxes:
if box.cls not in target_id:
continue
b = box.xyxy[0].numpy() if hasattr(box.xyxy[0], 'numpy') else box.xyxy[0]
track_id = int(box.id[0].item()) if box.id is not None else None
bbox = BBox(
x1=float(b[0]), y1=float(b[1]),
x2=float(b[2]), y2=float(b[3]),
conf=float(box.conf[0]),
cls=int(box.cls[0]),
track_id=track_id
)
results.boxes.append(bbox)
return results
- Step 4: Run test to verify it passes
Run: pytest tests/test_yolo_predict.py
Expected: PASS
Task 4: Complete Results.plot() Visualization Method
Files:
- Modify:
reid/core/types.py -
Create:
tests/test_results_plot.py - Step 1: Write test for
Results.plot()
Write tests/test_results_plot.py:
import pytest
import numpy as np
from reid.core.types import Results, BBox, MatchResult
def test_results_plot():
dummy_img = np.zeros((480, 640, 3), dtype=np.uint8)
res = Results(orig_img=dummy_img, path="test.jpg")
res.boxes.append(BBox(x1=50, y1=50, x2=200, y2=200, conf=0.9, cls=15, track_id=1))
res.match_results.append(MatchResult(cat_id="Nabi", similarity=0.88, is_known=True))
annotated = res.plot(show_conf=True, show_similarity=True)
assert isinstance(annotated, np.ndarray)
assert annotated.shape == (480, 640, 3)
assert not np.array_equal(annotated, dummy_img) # Verified drawn elements exist
def test_results_plot_no_orig_img():
res = Results(orig_img=None, path="test.jpg")
with pytest.raises(ValueError, match="Results object has no orig_img to plot"):
res.plot()
- Step 2: Run test to verify it fails
Run: pytest tests/test_results_plot.py
Expected: FAIL because Results.plot() currently has pass and returns None
- Step 3: Implement
Results.plot()inreid/core/types.py
In reid/core/types.py:
Add imports for OpenCV and color hashing:
import cv2
import zlib
from typing import List, Optional, Union, Tuple
Implement Results.plot():
def plot(
self,
show_conf: bool = True,
show_similarity: bool = True,
line_thickness: Optional[int] = None,
font_scale: Optional[float] = None
) -> np.ndarray:
"""
Draw bounding boxes, cat IDs, similarity scores, and confidence on orig_img.
Returns annotated OpenCV BGR numpy image array.
"""
if self.orig_img is None:
raise ValueError("Results object has no orig_img to plot.")
img = self.orig_img.copy()
h, w = img.shape[:2]
# Dynamic sizing
thickness = line_thickness or max(1, int(round(min(h, w) * 0.003)))
f_scale = font_scale or max(0.4, min(h, w) * 0.0007)
for idx, box in enumerate(self.boxes):
match_res = self.match_results[idx] if idx < len(self.match_results) else None
cat_id = match_res.cat_id if match_res else "Unknown"
is_known = match_res.is_known if match_res else False
# Assign color based on Cat ID hash (or gray for unknown)
if is_known and cat_id != "Unknown":
color_hash = zlib.adler32(cat_id.encode("utf-8"))
color = (
int((color_hash & 0xFF0000) >> 16),
int((color_hash & 0x00FF00) >> 8),
int(color_hash & 0x0000FF)
)
else:
color = (128, 128, 128) # Gray for Unknown
x1, y1, x2, y2 = int(box.x1), int(box.y1), int(box.x2), int(box.y2)
# Draw bounding box
cv2.rectangle(img, (x1, y1), (x2, y2), color, thickness)
# Construct text label
label_parts = [cat_id]
if match_res and show_similarity and match_res.similarity > 0:
label_parts.append(f"Sim:{match_res.similarity:.2f}")
if show_conf and box.conf is not None:
label_parts.append(f"Conf:{box.conf:.2f}")
label_text = " | ".join(label_parts)
# Text background rectangle
(text_w, text_h), baseline = cv2.getTextSize(label_text, cv2.FONT_HERSHEY_SIMPLEX, f_scale, 1)
text_y1 = max(0, y1 - text_h - baseline - 4)
text_y2 = y1 if y1 - text_h - baseline - 4 >= 0 else y1 + text_h + baseline + 4
cv2.rectangle(img, (x1, text_y1), (x1 + text_w + 6, text_y2), color, -1)
cv2.putText(
img,
label_text,
(x1 + 3, text_y2 - baseline - 2),
cv2.FONT_HERSHEY_SIMPLEX,
f_scale,
(255, 255, 255),
1,
cv2.LINE_AA
)
return img
- Step 4: Run test to verify it passes
Run: pytest tests/test_results_plot.py
Expected: PASS
Task 5: Integration Verification
- Step 1: Run all unit tests
Run: pytest tests/ -v
Expected: ALL PASS
- Step 2: Verify CLI sanity execution
Run: python3 -m reid.cli list
Expected: Output summary without error.
C
Contents
