shawn/canteen-asset-tracker
1
0
Fork 0
Code Issues 8 Pull Requests Actions Packages Projects Releases Wiki Activity

t_2f17b1b4: Add EXIF GPS extraction to upload/photo and OCR endpoints

Browse Source 
- _extract_gps_from_bytes helper reads GPSInfo IFD via PIL get_ifd()
- Converts DMS to decimal degrees, returns {lat, lng} or None
- /api/upload/photo: reads bytes first, extracts GPS before saving
- /api/ocr: extracts GPS from raw bytes before OCR processing
- exif_gps returned alongside path/machine_id in both endpoints
- Photos without EXIF GPS omit the exif_gps key entirely
- Raw file bytes preserved (no re-compression, no EXIF stripping)
This commit is contained in:
Shawn
2026-05-21 17:17:38 -04:00
parent 5d2d169885
commit 7cf566ad6d
2 changed files with 692 additions and 943 deletions
Download Patch File Download Diff File Expand all files Collapse all files
server.py
+166 -23
View File
@@ -2585,6 +2585,81 @@ ICON_ALLOWED_EXTS = {".png", ".jpg", ".jpeg", ".svg"}
PHOTO_ALLOWED_EXTS = {".png", ".jpg", ".jpeg"}
def _dms_to_decimal(dms_tuple: tuple, ref: str) -> float:
"""Convert GPS DMS (degrees, minutes, seconds) tuple to decimal degrees."""
try:
degrees = float(dms_tuple[0])
minutes = float(dms_tuple[1])
seconds = float(dms_tuple[2])
decimal = degrees + minutes / 60.0 + seconds / 3600.0
if ref in ("S", "W"):
decimal = -decimal
return round(decimal, 6)
except (TypeError, ValueError, IndexError):
return None
def _extract_gps_from_bytes(image_bytes: bytes) -> dict | None:
"""Extract GPS coordinates from image EXIF data.
Reads the raw image bytes with PIL, extracts the GPS IFD,
and converts DMS coordinates to decimal degrees.
Returns {lat, lng} or None if no GPS data found.
"""
try:
img = PILImage.open(io.BytesIO(image_bytes))
exif = img.getexif()
# GPSInfo IFD tag = 0x8825 = 34853
gps_ifd = exif.get_ifd(0x8825)
except Exception:
return None
if not gps_ifd:
return None
# GPSLatitudeRef = tag 1, GPSLatitude = tag 2
# GPSLongitudeRef = tag 3, GPSLongitude = tag 4
lat_ref = gps_ifd.get(1)
lat_dms = gps_ifd.get(2)
lng_ref = gps_ifd.get(3)
lng_dms = gps_ifd.get(4)
if lat_dms is None or lng_dms is None:
return None
lat = _dms_to_decimal(lat_dms, lat_ref or "N")
lng = _dms_to_decimal(lng_dms, lng_ref or "E")
if lat is None or lng is None:
return None
return {"lat": lat, "lng": lng}
def _save_upload_bytes(contents: bytes, filename: str | None, subdir: str, allowed_exts: set, max_size: int) -> str:
"""Save raw bytes to uploads/{subdir}/ with a UUID filename.
Like _save_upload but accepts pre-read bytes instead of an UploadFile,
so callers can extract EXIF before saving.
"""
ext = Path(filename or "").suffix.lower()
if not ext or ext not in allowed_exts:
allowed = ", ".join(sorted(allowed_exts))
raise HTTPException(400, f"Invalid file type. Allowed: {allowed}")
if len(contents) > max_size:
mb = max_size // (1024 * 1024)
raise HTTPException(413, f"File too large. Maximum size: {mb} MB")
dest_dir = UPLOADS_DIR / subdir
dest_dir.mkdir(parents=True, exist_ok=True)
fname = f"{uuid.uuid4().hex}{ext}"
(dest_dir / fname).write_bytes(contents)
return f"/uploads/{subdir}/{fname}"
def _save_upload(upload: UploadFile, subdir: str, allowed_exts: set, max_size: int) -> str:
"""Save uploaded file to uploads/{subdir}/ with a UUID filename.
@@ -2616,8 +2691,15 @@ async def upload_icon(file: UploadFile = File(...)):
@app.post("/api/upload/photo", status_code=201)
async def upload_photo(file: UploadFile = File(...)):
path = _save_upload(file, "photos", PHOTO_ALLOWED_EXTS, PHOTO_MAX_SIZE)
return {"path": path}
# Read bytes for EXIF extraction before saving
contents = await file.read()
exif_gps = _extract_gps_from_bytes(contents) if contents else None
# Save using the raw bytes
path = _save_upload_bytes(contents, file.filename, "photos", PHOTO_ALLOWED_EXTS, PHOTO_MAX_SIZE)
result = {"path": path}
if exif_gps:
result["exif_gps"] = exif_gps
return result
@app.post("/api/ocr", status_code=200)
@@ -2625,13 +2707,19 @@ async def ocr_sticker(file: UploadFile = File(...)):
"""OCR a sticker photo to extract machine_id from XXXXX-XXXXXX format.
Accepts an image upload, runs Tesseract OCR, and looks for a pattern like
'12345-678901'. Returns the extracted machine_id or an error.
'12345-678901'. Also extracts EXIF GPS data from the raw file bytes before
processing, returning both in one response.
Returns {machine_id, exif_gps (if found), raw_text, raw_match, confidence}.
"""
# Read file contents first (validate size before saving)
contents = await file.read()
if len(contents) > 20 * 1024 * 1024: # 20MB max
raise HTTPException(status_code=400, detail="Image too large (max 20MB)")
# Extract EXIF GPS from raw bytes before any processing
exif_gps = _extract_gps_from_bytes(contents) if contents else None
# Save temp file for OCR — try original extension, fallback to jpg
ext = file.filename.rsplit(".", 1)[-1].lower() if "." in (file.filename or "") else "jpg"
if ext not in {"png", "jpg", "jpeg", "webp", "bmp", "dng", "tiff", "tif"}:
@@ -2653,37 +2741,42 @@ async def ocr_sticker(file: UploadFile = File(...)):
# Don't keep temp file after processing
temp_path.unlink(missing_ok=True)
# Search for XXXXX-XXXXXX pattern (5 digits - 6 digits or more)
# Build response — search for XXXXX-XXXXXX pattern (5 digits - 6 digits or more)
match = re.search(r"(\d{5})[-\s]*(\d{6,})", text)
result: dict = {}
if match:
# Take only the last 5 digits of the matched ID (e.g., "05912-095330" → "95330")
full_match = match.group(0)
digits_only = re.sub(r"\D", "", full_match)
machine_id = digits_only[-5:]
return {
result = {
"machine_id": machine_id,
"raw_text": text.strip()[:500],
"raw_match": full_match,
"confidence": "high",
}
else:
# Try looser: any 5+ digit number, take the last 5 digits
loose = re.search(r"(\d{5,})", text)
if loose:
digits = loose.group(1)
machine_id = digits[-5:] if len(digits) > 5 else digits
result = {
"machine_id": machine_id,
"raw_text": text.strip()[:500],
"confidence": "low",
}
else:
result = {
"machine_id": None,
"raw_text": text.strip()[:500],
"confidence": "none",
"detail": "No machine ID pattern found in image. Try again with better lighting.",
}
# Try looser: any 5+ digit number, take the last 5 digits
loose = re.search(r"(\d{5,})", text)
if loose:
digits = loose.group(1)
machine_id = digits[-5:] if len(digits) > 5 else digits
return {
"machine_id": machine_id,
"raw_text": text.strip()[:500],
"confidence": "low",
}
return {
"machine_id": None,
"raw_text": text.strip()[:500],
"confidence": "none",
"detail": "No machine ID pattern found in image. Try again with better lighting.",
}
if exif_gps:
result["exif_gps"] = exif_gps
return result
# ─── T4: Connect Label — unified photo + OCR + GPS endpoint ─────────────────
@@ -2854,6 +2947,56 @@ def geocode(lat: float = Query(...), lng: float = Query(...)):
return result
# ─── Admin endpoints ────────────────────────────────────────────────────────
class AdminQuery(BaseModel):
sql: str
@app.post("/api/admin/reset", status_code=200)
def admin_reset():
"""Drop and recreate all tables, reseed defaults. Destructive."""
conn = get_db()
# Drop all tables (order matters for foreign keys)
tables = [
"sessions", "geofence_users", "geofences", "visits",
"asset_badges", "asset_keys", "checkins", "assets",
"models", "makes", "badge_types", "key_types", "key_names",
"categories", "rooms", "locations", "customer_contacts",
"customers", "activity_log", "users", "settings",
]
for table in tables:
conn.execute(f"DROP TABLE IF EXISTS {table}")
conn.commit()
_create_v2_tables(conn)
_seed_data(conn)
_ensure_unique_machine_id(conn)
conn.execute(
"CREATE INDEX IF NOT EXISTS idx_assets_category ON assets(category)"
)
conn.commit()
conn.close()
return {"message": "Database reset complete — all tables reinitialized with defaults."}
@app.post("/api/admin/query")
def admin_query(body: AdminQuery):
"""Run an arbitrary SQL query (SELECT only). Returns rows as JSON."""
sql = body.sql.strip().upper()
# Only allow SELECT queries
if not sql.startswith("SELECT"):
raise HTTPException(status_code=400, detail="Only SELECT queries are allowed")
conn = get_db()
try:
rows = conn.execute(body.sql).fetchall()
conn.close()
return [dict(r) for r in rows]
except Exception as e:
conn.close()
raise HTTPException(status_code=400, detail=str(e))
# ─── Static Files (mounted last to not shadow routes) ──────────────────────
app.mount("/uploads", StaticFiles(directory=str(UPLOADS_DIR)), name="uploads")