File size: 2,308 Bytes
0dc87cd | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 | """Full ShelfEye pipeline: detect → crop → embed → match → position → JSON output."""
import json
from datetime import datetime
from pathlib import Path
from detector import detect_products
from cropper import crop_detections
from matcher import match_detections
from position import assign_positions
from visualizer import draw_detections
def run_pipeline(
image_path: str,
output_dir: str = "outputs",
use_vlm: bool = True,
device: str = None,
) -> dict:
"""Run the full pipeline on a single shelf image.
Returns structured JSON result.
"""
dets = detect_products(image_path, device=device)
dets = crop_detections(image_path, dets)
dets = match_detections(dets, use_vlm=use_vlm, device=device)
dets = assign_positions(dets)
products = []
for d in dets:
products.append({
"sku_id": d.get("sku_id"),
"product_name": d.get("product_name", "unknown"),
"confidence": d.get("similarity", d.get("confidence", 0)),
"bbox": [int(c) for c in d["bbox"]],
"row": d["row"],
"column": d["column"],
"matched": d.get("matched", False),
"match_method": d.get("match_method", "none"),
})
matched_count = sum(1 for p in products if p["matched"])
result = {
"image": Path(image_path).name,
"timestamp": datetime.now().isoformat(timespec="seconds"),
"total_products": len(products),
"matched": matched_count,
"unmatched": len(products) - matched_count,
"products": products,
}
Path(output_dir).mkdir(parents=True, exist_ok=True)
out_path = Path(output_dir) / f"result_{Path(image_path).stem}.json"
with open(out_path, "w") as f:
json.dump(result, f, indent=2)
img_out = str(Path(output_dir) / f"annotated_{Path(image_path).stem}.jpg")
draw_detections(image_path, dets, img_out)
print(f"[pipeline] Saved result to {out_path}")
return result
if __name__ == "__main__":
import sys
images = sys.argv[1:] or ["shelf_images/shelf_01.jpg"]
for img in images:
result = run_pipeline(img)
print(f"\n{result['image']}: {result['matched']}/{result['total_products']} matched, "
f"{result['unmatched']} unknown")
|