T4: Connect Label upload workflow — unified photo + OCR + GPS

- Added 'Connect' mode toggle on Add Asset tab with Camera/Gallery options
- processConnectLabelPhoto(file): parallel OCR + GPS extraction
- Shows results card with editable machine_id, lat/lng, map preview
- POST /api/connect-label endpoint with photo upload + GPS + reverse geocode
- Connect Label section in Assets list (localStorage-backed recent scans)
- Handles OCR failure (manual entry) and missing EXIF GPS gracefully
This commit is contained in:
2026-05-21 11:50:09 -04:00
parent f895bf99de
commit 622996624a
2 changed files with 561 additions and 2 deletions
+102 -1
View File
@@ -22,7 +22,7 @@ from pathlib import Path
import pytesseract
from PIL import Image as PILImage
from fastapi import FastAPI, HTTPException, Query, Request, UploadFile, File
from fastapi import FastAPI, HTTPException, Query, Request, UploadFile, File, Form
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import JSONResponse, StreamingResponse
from fastapi.staticfiles import StaticFiles
@@ -2687,6 +2687,107 @@ async def ocr_sticker(file: UploadFile = File(...)):
}
# ─── T4: Connect Label — unified photo + OCR + GPS endpoint ─────────────────
class ConnectLabelRequest(BaseModel):
"""Request body for the connect-label endpoint."""
machine_id: str
name: str = ""
latitude: Optional[float] = None
longitude: Optional[float] = None
category: Optional[str] = "Other"
photo_path: Optional[str] = None
@app.post("/api/connect-label", status_code=201)
async def connect_label(
photo: Optional[UploadFile] = File(None),
machine_id: str = Form(""),
name: str = Form(""),
latitude: Optional[float] = Form(None),
longitude: Optional[float] = Form(None),
category: str = Form("Other"),
photo_path: str = Form(""),
):
"""Unified endpoint for Connect Label workflow.
Accepts multipart form data with optional photo upload.
Creates an asset with the provided machine_id, name, GPS coords.
If photo is uploaded, it saves the file and sets photo_path.
Validates machine_id format (XXXXX-XXXXXX → last 5 digits).
"""
# If machine_id/name come as form fields, use those; otherwise try query params
machine_id = _sanitize_machine_id(machine_id)
name = _sanitize_name(name or "Scanned Asset")
_validate_status("active")
# Save photo if provided
saved_photo_path = photo_path or ""
if photo and photo.filename:
saved_photo_path = _save_upload(photo, "connect_labels", PHOTO_ALLOWED_EXTS, PHOTO_MAX_SIZE)
conn = get_db()
_validate_enum_table(conn, "categories", category or "Other", "category")
# Build insert
columns: list[str] = ["machine_id", "name", "category", "status"]
values: list = [machine_id, name, category or "Other", "active"]
if saved_photo_path:
columns.append("photo_path")
values.append(saved_photo_path)
if latitude is not None:
columns.append("latitude")
values.append(latitude)
if longitude is not None:
columns.append("longitude")
values.append(longitude)
# Auto-populate address from GPS
if latitude is not None and longitude is not None:
geo = reverse_geocode(latitude, longitude)
if geo:
if "address" not in columns:
columns.append("address")
values.append(geo.get("address", geo.get("formatted", "")))
if "building_name" not in columns and geo.get("building_name"):
columns.append("building_name")
values.append(geo.get("building_name", ""))
if "building_number" not in columns and geo.get("building_number"):
columns.append("building_number")
values.append(geo.get("building_number", ""))
placeholders = ", ".join(["?"] * len(columns))
col_names = ", ".join(columns)
try:
cursor = conn.execute(
f"INSERT INTO assets ({col_names}) VALUES ({placeholders})",
values,
)
except sqlite3.IntegrityError:
conn.close()
raise HTTPException(
status_code=409,
detail=f"Asset with machine_id '{machine_id}' already exists",
)
raw_id = cursor.lastrowid
assert raw_id is not None, "INSERT did not return lastrowid"
asset_id: int = raw_id
_log_activity(conn, "created", "asset", asset_id,
f"Asset '{name}' (machine_id: {machine_id}) created via Connect Label")
conn.commit()
row = conn.execute("SELECT * FROM assets WHERE id = ?", (asset_id,)).fetchone()
result = row_to_dict(row)
result["keys"] = _get_asset_keys(conn, asset_id)
result["badges"] = _get_asset_badges(conn, asset_id)
conn.close()
return result
# ─── Reverse Geocode (Nominatim) ────────────────────────────────────────────
def reverse_geocode(lat: float, lng: float) -> dict | None: