api/compat.py

"""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:
            # Identity evidence may be empty (e.g. the second photo is an excluded copy); geometry is still shown.
            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,
            # CarPrint extras
            "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]  # synthetic, disclosed sample (AI-generated studio image)
        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