| """Compatibility layer for the web studio (web/src/App.tsx): POST /api/compare with `reference` + `current`.
|
|
|
| Returns the schema the studio already renders, computed by the CarPrint pipeline, plus extra CarPrint fields
|
| (congruent_cells, zones, cohort_size, integrity). Errors use a plain-string `detail`, as the studio expects.
|
| """
|
| from __future__ import annotations
|
|
|
| import base64
|
| import os
|
| import time
|
| from pathlib import Path
|
|
|
| import cv2
|
| import numpy as np
|
| from fastapi import APIRouter, File, Form, HTTPException, UploadFile
|
| from fastapi.responses import Response
|
|
|
| from carprint.core.image_io import load_image
|
| from carprint.identity.features import match
|
| from carprint.identity.geometry import fit_zones
|
| from carprint.integrity.hashing import DUPLICATE_MAX_DISTANCE, hamming, phash64, sha256_hex
|
|
|
| router = APIRouter()
|
| MAX_BYTES = 8 * 1024 * 1024
|
| WEB_ROOT = Path(os.environ.get("CARPRINT_WEB", Path(__file__).resolve().parents[1] / "web"))
|
|
|
| DECISION_EN = {
|
| "SAME": "Same vehicle: individual marks agree in {cells} independent areas",
|
| "DIFFERENT": "Different vehicle: the design aligns, but the reference's individual marks are absent",
|
| "REVIEW": "Not enough individual evidence for an automatic decision",
|
| "RECAPTURE_REQUIRED": "Photo quality is too low for a decision. Please retake",
|
| }
|
| QUALITY_EN = {
|
| "blurry": "A photo is blurred. Hold the phone still and retake.",
|
| "too_dark": "A photo is too dark. Use the flash or a torch.",
|
| "overexposed": "A photo is overexposed. Change the angle.",
|
| "glare": "Strong glare on the car. Move to remove the reflection.",
|
| "car_too_small": "The car is too small in the frame. Move closer.",
|
| "no_car": "No car was found in the frame.",
|
| }
|
| LIMITATIONS = ("CarPrint decides SAME only from individual marks (chips, curb rash, scratches) after discounting features "
|
| "shared by every car of the same model. Identical new cars without marks stay REVIEW and need a VIN check. "
|
| "Findings are evidence for a person, not an automatic charge or legal assessment.")
|
| ALGORITHM = ("ALIKED + LightGlue, MAGSAC++ planar zones with a-contrario (ORSA) gate, same-model cohort "
|
| "distinctiveness, congruent-cell voting")
|
|
|
|
|
| def _data_url(rgb: np.ndarray, max_side: int = 1400) -> str:
|
| h, w = rgb.shape[:2]
|
| s = min(1.0, max_side / max(h, w))
|
| if s < 1:
|
| rgb = cv2.resize(rgb, (round(w * s), round(h * s)), interpolation=cv2.INTER_AREA)
|
| ok, enc = cv2.imencode(".jpg", cv2.cvtColor(rgb, cv2.COLOR_RGB2BGR), [cv2.IMWRITE_JPEG_QUALITY, 85])
|
| return "data:image/jpeg;base64," + base64.b64encode(enc.tobytes()).decode()
|
|
|
|
|
| def _read(upload: UploadFile) -> bytes:
|
| data = upload.file.read(MAX_BYTES + 1)
|
| if not data or len(data) > MAX_BYTES:
|
| raise HTTPException(413, "Each image must be between 1 byte and 8 MB.")
|
| if not (data.startswith(b"\xff\xd8\xff") or data.startswith(b"\x89PNG\r\n\x1a\n")
|
| or (data.startswith(b"RIFF") and data[8:12] == b"WEBP")):
|
| raise HTTPException(415, "Supported image formats are JPG, PNG and WebP.")
|
| return data
|
|
|
|
|
| def build_router(main):
|
| """`main` is api.main (passed in to avoid a circular import)."""
|
|
|
| @router.get("/api/health")
|
| def health():
|
| h = main.health()
|
| return {"status": "ok", "algorithm": ALGORITHM, "version": main.app.version, **h}
|
|
|
| @router.post("/api/compare")
|
| def compare(reference: UploadFile = File(...), current: UploadFile = File(...), model: str = Form("")):
|
| t0 = time.perf_counter()
|
| blobs = [_read(reference), _read(current)]
|
| for b in blobs:
|
| try:
|
| img = load_image(b)
|
| except Exception:
|
| raise HTTPException(422, "Could not read the image. Please try a different file.")
|
| if min(img.rgb.shape[:2]) < 96:
|
| raise HTTPException(422, "The image is too small. Use at least 96 × 96 pixels.")
|
| ins = []
|
| for b in blobs:
|
| i = main.Inspection(f"ins_{os.urandom(8).hex()}", f"veh_{os.urandom(12).hex()}", "owner", "compare", "",
|
| main.now() + main.timedelta(hours=2), model.strip().lower())
|
| main.INSPECTIONS[i.inspection_id] = i
|
| main._ingest(i, b, "extra", "gallery")
|
| ins.append(i)
|
| res = main._verify(ins[0], ins[1])
|
| ref, cur = ins[0].frames[0], ins[1].frames[0]
|
| zones = [z for z in res["evidence"]]
|
| best = max(zones, key=lambda z: z["inliers"], default=None)
|
| verdict = res["decision"]
|
| cells = max((z.get("congruent_cells", 0) for z in zones), default=0)
|
| registered = best is not None and best["log10_nfa"] < main.REGISTRATION_NFA
|
|
|
| rh, rw = ref.frame.rgb.shape[:2]
|
| aligned, coverage, err, inliers, tentative = None, 0.0, None, 0, 0
|
| zone = None
|
| if registered:
|
| v_zone = next((z for z in main.LAST_ZONES.get(res["verification_id"], []) if z.zone.n_inliers == best["inliers"]), None)
|
| zone = v_zone.zone if v_zone is not None else None
|
| elif not zones:
|
|
|
| with main.compute_lock:
|
| m, _ = match(ref.frame.feats, cur.frame.feats)
|
| if len(m) >= 10:
|
| found = fit_zones(ref.frame.feats.keypoints[m[:, 0]], cur.frame.feats.keypoints[m[:, 1]], rw * rh)
|
| zone = max(found, key=lambda z: z.n_inliers, default=None)
|
| if zone is not None and zone.log10_nfa >= main.REGISTRATION_NFA:
|
| zone = None
|
| if zone is not None:
|
| aligned = cv2.warpPerspective(cur.frame.rgb, np.linalg.inv(zone.H), (rw, rh))
|
| x, y, w, h = zone.bbox0
|
| coverage = round(100.0 * (w * h) / (rw * rh), 1)
|
| err, inliers, tentative = round(zone.reproj_err_px, 2), zone.n_inliers, zone.n_tentative
|
| registered = True
|
|
|
| warnings = []
|
| for f in (ref, cur):
|
| warnings += [QUALITY_EN[i] for i in f.frame.quality.issues if i in QUALITY_EN]
|
| copy = blobs[0] == blobs[1] or hamming(phash64(ref.frame.rgb), phash64(cur.frame.rgb)) <= DUPLICATE_MAX_DISTANCE
|
| if copy:
|
| warnings.append("The second photo is a copy or a transformed copy of the first (perceptual hash): it is "
|
| "excluded from identity evidence. Alignment is shown for illustration only.")
|
| if verdict == "REVIEW":
|
| warnings.append("Photograph the same side of the car from a similar angle; include wheels and any marks.")
|
| warnings.append("Matching shape alone never proves identity: features shared by the model are discounted.")
|
|
|
| evidence_img = main.ARTIFACTS / res["evidence"][0]["match_viz_url"].rsplit("/", 1)[-1] if res["evidence"] else None
|
| evidence = cv2.cvtColor(cv2.imread(str(evidence_img)), cv2.COLOR_BGR2RGB) if evidence_img and evidence_img.exists() \
|
| else np.concatenate([cv2.resize(ref.frame.rgb, (480, 360)), cv2.resize(cur.frame.rgb, (480, 360))], axis=1)
|
| return {
|
| "decision": verdict,
|
| "geometry": "MATCHED" if registered else "INSUFFICIENT",
|
| "message": ("Copy detected: the second photo is not an independent capture" if copy
|
| else DECISION_EN.get(verdict, verdict).format(cells=cells)),
|
| "inliers": int(inliers),
|
| "candidate_matches": int(tentative),
|
| "reprojection_error_px": err,
|
| "coverage_percent": coverage,
|
| "elapsed_ms": round((time.perf_counter() - t0) * 1000),
|
| "warnings": list(dict.fromkeys(warnings)),
|
| "evidence_image": _data_url(evidence),
|
| "reference_image": _data_url(ref.frame.rgb),
|
| "aligned_image": _data_url(aligned) if aligned is not None else None,
|
| "algorithm": ALGORITHM,
|
| "algorithm_version": f"carprint-{main.app.version}",
|
| "sha256": [sha256_hex(b) for b in blobs],
|
| "limitations": LIMITATIONS,
|
|
|
| "congruent_cells": cells, "score_raw": res["score_raw"], "cohort_size": res["cohort_size"],
|
| "zones": zones, "integrity": res["integrity"], "copy_detected": copy,
|
| "damage_timeline": res["damage_timeline"],
|
| "cost_estimate_new": res.get("cost_estimate_new"), "cost_estimate_current": res.get("cost_estimate_current"),
|
| }
|
|
|
| @router.get("/api/example/{view}")
|
| def example(view: str):
|
| if view not in ("reference", "current"):
|
| raise HTTPException(404, "Photo not found.")
|
| image = cv2.imread(str(WEB_ROOT / "public" / "images" / "car-studio.png"))
|
| if image is None:
|
| raise HTTPException(503, "Sample images are unavailable.")
|
| image = image[300:850, 350:1500]
|
| if view == "current":
|
| h, w = image.shape[:2]
|
| image = cv2.warpAffine(image, cv2.getRotationMatrix2D((w / 2, h / 2), 3, 0.96), (w, h),
|
| borderMode=cv2.BORDER_REFLECT)
|
| image = cv2.convertScaleAbs(image, alpha=0.92, beta=8)
|
| ok, enc = cv2.imencode(".jpg", image, [cv2.IMWRITE_JPEG_QUALITY, 85])
|
| return Response(enc.tobytes(), media_type="image/jpeg", headers={"Cache-Control": "public, max-age=3600"})
|
|
|
| return router
|
|
|
