backend/app/api/routes/alpr.py

209 lines · python

"""ALPR recognition API: turn images (uploads or public snapshot URLs) into plate reads.
 
  POST /alpr/recognize       multipart image  -> detected plates (no DB write; demo/test)
  POST /alpr/recognize-url   {url}            -> fetch public image, detect plates
  POST /alpr/ingest          image + camera   -> detect plates AND create real reads via the pipeline
 
Recognized plates from real images are real reads and flow through normalize/dedup/watchlist
exactly like a file import or live stream."""
from __future__ import annotations
 
from datetime import datetime, timezone
 
import httpx
from fastapi import APIRouter, Depends, File, Form, HTTPException, UploadFile
from pydantic import BaseModel
from sqlalchemy import select
from sqlalchemy.orm import Session
 
from app.api.deps import Perm, get_db, require
from app.core.config import settings
from app.models import AuditLog, Camera, Location, User
from app.services.ingest import ingest_live_reads
 
router = APIRouter(prefix="/alpr", tags=["alpr"])
 
 
def _to_dicts(dets):
    return [{"plate": d.plate, "plate_norm": d.plate_norm,
             "detection_confidence": round(d.detection_confidence, 3),
             "ocr_confidence": round(d.ocr_confidence, 3), "bbox": d.bbox} for d in dets]
 
 
@router.post("/recognize")
async def recognize(file: UploadFile = File(...), min_conf: float = Form(default=0.4),
                    user: User = Depends(require(Perm.READ))):
    from app.services.alpr import recognize_bytes
    data = await file.read()
    if not data:
        raise HTTPException(400, "empty image")
    try:
        dets = recognize_bytes(data, min_conf=min_conf)
    except Exception as e:
        raise HTTPException(422, f"recognition failed: {e}")
    return {"plates": _to_dicts(dets), "count": len(dets)}
 
 
class UrlReq(BaseModel):
    url: str
    min_conf: float = 0.4
 
 
@router.post("/recognize-url")
def recognize_url(req: UrlReq, user: User = Depends(require(Perm.READ))):
    from app.services.alpr import recognize_bytes
    try:
        r = httpx.get(req.url, timeout=30, follow_redirects=True,
                      headers={"User-Agent": "Sightline-ALPR/1.0"})
        r.raise_for_status()
    except Exception as e:
        raise HTTPException(502, f"could not fetch image: {e}")
    try:
        dets = recognize_bytes(r.content, min_conf=req.min_conf)
    except Exception as e:
        raise HTTPException(422, f"recognition failed: {e}")
    return {"url": req.url, "bytes": len(r.content), "plates": _to_dicts(dets), "count": len(dets)}
 
 
@router.post("/ingest")
async def ingest_image(
    file: UploadFile = File(...),
    camera_id: str | None = Form(default=None),
    camera_external_id: str | None = Form(default=None),
    state: str | None = Form(default=None),
    min_conf: float = Form(default=0.4),
    db: Session = Depends(get_db),
    user: User = Depends(require(Perm.INGEST)),
):
    """Recognize plate(s) in an authorized image and create real reads attached to a camera."""
    from app.services.alpr import recognize_bytes
    data = await file.read()
    if not data:
        raise HTTPException(400, "empty image")
 
    cam = None
    if camera_id:
        cam = db.get(Camera, camera_id)
    elif camera_external_id:
        cam = db.scalar(select(Camera).where(Camera.external_camera_id == camera_external_id))
    if (camera_id or camera_external_id) and not cam:
        raise HTTPException(404, "camera not found")
 
    try:
        dets = recognize_bytes(data, min_conf=min_conf)
    except Exception as e:
        raise HTTPException(422, f"recognition failed: {e}")
    if not dets:
        return {"recognized": 0, "inserted": 0, "plates": []}
 
    loc = db.get(Location, cam.location_id) if cam and cam.location_id else None
    now = datetime.now(timezone.utc)
    reads = [{"plate": d.plate_norm, "state": (state or (cam.state if cam else None)),
              "timestamp": now.isoformat(),
              "camera": cam.external_camera_id if cam else "alpr-upload",
              "latitude": loc.lat if loc else None, "longitude": loc.lon if loc else None,
              "confidence": d.ocr_confidence} for d in dets]
    summary = ingest_live_reads(db, cam.source_id if cam else None, reads,
                                label="alpr-recognition", created_by=user.id)
    db.add(AuditLog(actor_user_id=user.id, action="alpr.ingest",
                    entity_type="import", entity_id=summary.get("import_id"),
                    meta={"recognized": len(dets), "camera": camera_external_id or camera_id}))
    db.commit()
    return {"recognized": len(dets), **summary, "plates": _to_dicts(dets)}
 
 
def _resolve_cam(db, camera_id, camera_external_id):
    cam = None
    if camera_id:
        cam = db.get(Camera, camera_id)
    elif camera_external_id:
        cam = db.scalar(select(Camera).where(Camera.external_camera_id == camera_external_id))
    if (camera_id or camera_external_id) and not cam:
        raise HTTPException(404, "camera not found")
    return cam
 
 
def _ingest_tracks(db, user, cam, plates, state, label):
    """Turn aggregated video/stream plate tracks into reads through the pipeline."""
    loc = db.get(Location, cam.location_id) if cam and cam.location_id else None
    now = datetime.now(timezone.utc)
    reads = [{"plate": p["plate_norm"], "state": state or (cam.state if cam else None),
              "timestamp": now.isoformat(), "camera": cam.external_camera_id if cam else "alpr-video",
              "latitude": loc.lat if loc else None, "longitude": loc.lon if loc else None,
              "confidence": p["ocr_confidence"]} for p in plates]
    summary = ingest_live_reads(db, cam.source_id if cam else None, reads, label=label, created_by=user.id)
    db.add(AuditLog(actor_user_id=user.id, action="alpr.video_ingest",
                    entity_type="import", entity_id=summary.get("import_id"),
                    meta={"plates": len(plates), "label": label}))
    db.commit()
    return summary
 
 
@router.post("/video")
async def recognize_video(
    file: UploadFile = File(...),
    sample_fps: float = Form(default=2.0),
    max_frames: int = Form(default=200),
    min_conf: float = Form(default=0.4),
    ingest: bool = Form(default=False),
    camera_external_id: str | None = Form(default=None),
    state: str | None = Form(default=None),
    db: Session = Depends(get_db),
    user: User = Depends(require(Perm.READ)),
):
    """Process an uploaded video clip: sample frames, recognize plates, aggregate to one entry per
    plate. Set ingest=true (analyst role) to create reads."""
    from app.services.video_alpr import process_video_bytes
    data = await file.read()
    if not data:
        raise HTTPException(400, "empty video")
    try:
        result = process_video_bytes(data, sample_fps=sample_fps, max_frames=max_frames, min_conf=min_conf)
    except Exception as e:
        raise HTTPException(422, f"video processing failed: {e}")
    if ingest and result["plates"]:
        from app.core.permissions import Perm as _P, role_has
        if not role_has(user.role, _P.INGEST):
            raise HTTPException(403, "ingest requires analyst role")
        cam = _resolve_cam(db, None, camera_external_id)
        result["ingest"] = _ingest_tracks(db, user, cam, result["plates"], state, "alpr-video")
    return result
 
 
class RtspReq(BaseModel):
    url: str
    seconds: float = 15.0
    sample_fps: float = 2.0
    min_conf: float = 0.4
    ingest: bool = False
    camera_external_id: str | None = None
    state: str | None = None
 
 
@router.post("/rtsp/probe")
def rtsp_probe(req: RtspReq, user: User = Depends(require(Perm.READ))):
    """Verify an RTSP/HTTP stream is reachable and decodable (grabs one frame)."""
    from app.services.video_alpr import probe_stream
    try:
        return probe_stream(req.url)
    except Exception as e:
        raise HTTPException(502, f"stream probe failed: {e}")
 
 
@router.post("/rtsp/capture")
def rtsp_capture(req: RtspReq, db: Session = Depends(get_db), user: User = Depends(require(Perm.READ))):
    """Capture a bounded window from an RTSP/HTTP stream and recognize plates. Optionally ingest."""
    from app.services.video_alpr import process_stream
    if req.seconds > 120:
        raise HTTPException(400, "capture window capped at 120s")
    try:
        result = process_stream(req.url, seconds=req.seconds, sample_fps=req.sample_fps, min_conf=req.min_conf)
    except Exception as e:
        raise HTTPException(502, f"stream capture failed: {e}")
    if req.ingest and result["plates"]:
        from app.core.permissions import Perm as _P, role_has
        if not role_has(user.role, _P.INGEST):
            raise HTTPException(403, "ingest requires analyst role")
        cam = _resolve_cam(db, None, req.camera_external_id)
        result["ingest"] = _ingest_tracks(db, user, cam, result["plates"], req.state, "alpr-rtsp")
    return result

1	"""ALPR recognition API: turn images (uploads or public snapshot URLs) into plate reads.
2
3	POST /alpr/recognize multipart image -> detected plates (no DB write; demo/test)
4	POST /alpr/recognize-url {url} -> fetch public image, detect plates
5	POST /alpr/ingest image + camera -> detect plates AND create real reads via the pipeline
6
7	Recognized plates from real images are real reads and flow through normalize/dedup/watchlist
8	exactly like a file import or live stream."""
9	from __future__ import annotations
10
11	from datetime import datetime, timezone
12
13	import httpx
14	from fastapi import APIRouter, Depends, File, Form, HTTPException, UploadFile
15	from pydantic import BaseModel
16	from sqlalchemy import select
17	from sqlalchemy.orm import Session
18
19	from app.api.deps import Perm, get_db, require
20	from app.core.config import settings
21	from app.models import AuditLog, Camera, Location, User
22	from app.services.ingest import ingest_live_reads
23
24	router = APIRouter(prefix="/alpr", tags=["alpr"])
25
26
27	def _to_dicts(dets):
28	return [{"plate": d.plate, "plate_norm": d.plate_norm,
29	"detection_confidence": round(d.detection_confidence, 3),
30	"ocr_confidence": round(d.ocr_confidence, 3), "bbox": d.bbox} for d in dets]
31
32
33	@router.post("/recognize")
34	async def recognize(file: UploadFile = File(...), min_conf: float = Form(default=0.4),
35	user: User = Depends(require(Perm.READ))):
36	from app.services.alpr import recognize_bytes
37	data = await file.read()
38	if not data:
39	raise HTTPException(400, "empty image")
40	try:
41	dets = recognize_bytes(data, min_conf=min_conf)
42	except Exception as e:
43	raise HTTPException(422, f"recognition failed: {e}")
44	return {"plates": _to_dicts(dets), "count": len(dets)}
45
46
47	class UrlReq(BaseModel):
48	url: str
49	min_conf: float = 0.4
50
51
52	@router.post("/recognize-url")
53	def recognize_url(req: UrlReq, user: User = Depends(require(Perm.READ))):
54	from app.services.alpr import recognize_bytes
55	try:
56	r = httpx.get(req.url, timeout=30, follow_redirects=True,
57	headers={"User-Agent": "Sightline-ALPR/1.0"})
58	r.raise_for_status()
59	except Exception as e:
60	raise HTTPException(502, f"could not fetch image: {e}")
61	try:
62	dets = recognize_bytes(r.content, min_conf=req.min_conf)
63	except Exception as e:
64	raise HTTPException(422, f"recognition failed: {e}")
65	return {"url": req.url, "bytes": len(r.content), "plates": _to_dicts(dets), "count": len(dets)}
66
67
68	@router.post("/ingest")
69	async def ingest_image(
70	file: UploadFile = File(...),
71	camera_id: str \| None = Form(default=None),
72	camera_external_id: str \| None = Form(default=None),
73	state: str \| None = Form(default=None),
74	min_conf: float = Form(default=0.4),
75	db: Session = Depends(get_db),
76	user: User = Depends(require(Perm.INGEST)),
77	):
78	"""Recognize plate(s) in an authorized image and create real reads attached to a camera."""
79	from app.services.alpr import recognize_bytes
80	data = await file.read()
81	if not data:
82	raise HTTPException(400, "empty image")
83
84	cam = None
85	if camera_id:
86	cam = db.get(Camera, camera_id)
87	elif camera_external_id:
88	cam = db.scalar(select(Camera).where(Camera.external_camera_id == camera_external_id))
89	if (camera_id or camera_external_id) and not cam:
90	raise HTTPException(404, "camera not found")
91
92	try:
93	dets = recognize_bytes(data, min_conf=min_conf)
94	except Exception as e:
95	raise HTTPException(422, f"recognition failed: {e}")
96	if not dets:
97	return {"recognized": 0, "inserted": 0, "plates": []}
98
99	loc = db.get(Location, cam.location_id) if cam and cam.location_id else None
100	now = datetime.now(timezone.utc)
101	reads = [{"plate": d.plate_norm, "state": (state or (cam.state if cam else None)),
102	"timestamp": now.isoformat(),
103	"camera": cam.external_camera_id if cam else "alpr-upload",
104	"latitude": loc.lat if loc else None, "longitude": loc.lon if loc else None,
105	"confidence": d.ocr_confidence} for d in dets]
106	summary = ingest_live_reads(db, cam.source_id if cam else None, reads,
107	label="alpr-recognition", created_by=user.id)
108	db.add(AuditLog(actor_user_id=user.id, action="alpr.ingest",
109	entity_type="import", entity_id=summary.get("import_id"),
110	meta={"recognized": len(dets), "camera": camera_external_id or camera_id}))
111	db.commit()
112	return {"recognized": len(dets), **summary, "plates": _to_dicts(dets)}
113
114
115	def _resolve_cam(db, camera_id, camera_external_id):
116	cam = None
117	if camera_id:
118	cam = db.get(Camera, camera_id)
119	elif camera_external_id:
120	cam = db.scalar(select(Camera).where(Camera.external_camera_id == camera_external_id))
121	if (camera_id or camera_external_id) and not cam:
122	raise HTTPException(404, "camera not found")
123	return cam
124
125
126	def _ingest_tracks(db, user, cam, plates, state, label):
127	"""Turn aggregated video/stream plate tracks into reads through the pipeline."""
128	loc = db.get(Location, cam.location_id) if cam and cam.location_id else None
129	now = datetime.now(timezone.utc)
130	reads = [{"plate": p["plate_norm"], "state": state or (cam.state if cam else None),
131	"timestamp": now.isoformat(), "camera": cam.external_camera_id if cam else "alpr-video",
132	"latitude": loc.lat if loc else None, "longitude": loc.lon if loc else None,
133	"confidence": p["ocr_confidence"]} for p in plates]
134	summary = ingest_live_reads(db, cam.source_id if cam else None, reads, label=label, created_by=user.id)
135	db.add(AuditLog(actor_user_id=user.id, action="alpr.video_ingest",
136	entity_type="import", entity_id=summary.get("import_id"),
137	meta={"plates": len(plates), "label": label}))
138	db.commit()
139	return summary
140
141
142	@router.post("/video")
143	async def recognize_video(
144	file: UploadFile = File(...),
145	sample_fps: float = Form(default=2.0),
146	max_frames: int = Form(default=200),
147	min_conf: float = Form(default=0.4),
148	ingest: bool = Form(default=False),
149	camera_external_id: str \| None = Form(default=None),
150	state: str \| None = Form(default=None),
151	db: Session = Depends(get_db),
152	user: User = Depends(require(Perm.READ)),
153	):
154	"""Process an uploaded video clip: sample frames, recognize plates, aggregate to one entry per
155	plate. Set ingest=true (analyst role) to create reads."""
156	from app.services.video_alpr import process_video_bytes
157	data = await file.read()
158	if not data:
159	raise HTTPException(400, "empty video")
160	try:
161	result = process_video_bytes(data, sample_fps=sample_fps, max_frames=max_frames, min_conf=min_conf)
162	except Exception as e:
163	raise HTTPException(422, f"video processing failed: {e}")
164	if ingest and result["plates"]:
165	from app.core.permissions import Perm as _P, role_has
166	if not role_has(user.role, _P.INGEST):
167	raise HTTPException(403, "ingest requires analyst role")
168	cam = _resolve_cam(db, None, camera_external_id)
169	result["ingest"] = _ingest_tracks(db, user, cam, result["plates"], state, "alpr-video")
170	return result
171
172
173	class RtspReq(BaseModel):
174	url: str
175	seconds: float = 15.0
176	sample_fps: float = 2.0
177	min_conf: float = 0.4
178	ingest: bool = False
179	camera_external_id: str \| None = None
180	state: str \| None = None
181
182
183	@router.post("/rtsp/probe")
184	def rtsp_probe(req: RtspReq, user: User = Depends(require(Perm.READ))):
185	"""Verify an RTSP/HTTP stream is reachable and decodable (grabs one frame)."""
186	from app.services.video_alpr import probe_stream
187	try:
188	return probe_stream(req.url)
189	except Exception as e:
190	raise HTTPException(502, f"stream probe failed: {e}")
191
192
193	@router.post("/rtsp/capture")
194	def rtsp_capture(req: RtspReq, db: Session = Depends(get_db), user: User = Depends(require(Perm.READ))):
195	"""Capture a bounded window from an RTSP/HTTP stream and recognize plates. Optionally ingest."""
196	from app.services.video_alpr import process_stream
197	if req.seconds > 120:
198	raise HTTPException(400, "capture window capped at 120s")
199	try:
200	result = process_stream(req.url, seconds=req.seconds, sample_fps=req.sample_fps, min_conf=req.min_conf)
201	except Exception as e:
202	raise HTTPException(502, f"stream capture failed: {e}")
203	if req.ingest and result["plates"]:
204	from app.core.permissions import Perm as _P, role_has
205	if not role_has(user.role, _P.INGEST):
206	raise HTTPException(403, "ingest requires analyst role")
207	cam = _resolve_cam(db, None, req.camera_external_id)
208	result["ingest"] = _ingest_tracks(db, user, cam, result["plates"], req.state, "alpr-rtsp")
209	return result