"""EXIF + OCR test backend — validate that GPS survives upload pipeline.""" import io, json, re, uuid from pathlib import Path from fastapi import FastAPI, File, Form, HTTPException, UploadFile from fastapi.staticfiles import StaticFiles from PIL import Image as PILImage try: import pytesseract HAS_TESSERACT = True except ImportError: HAS_TESSERACT = False UPLOADS = Path(__file__).parent / "uploads" UPLOADS.mkdir(exist_ok=True) app = FastAPI(title="EXIF Test") def _dms_to_decimal(dms, ref): """Convert EXIF DMS tuple to decimal degrees.""" try: deg, minutes, sec = float(dms[0]), float(dms[1]), float(dms[2]) decimal = deg + minutes / 60.0 + sec / 3600.0 if ref in ("S", "W"): decimal = -decimal return round(decimal, 7) except Exception: return None def extract_exif(image_bytes: bytes) -> dict: """Pull all useful EXIF fields + GPS from raw image bytes.""" result = {"has_exif": False, "tags": {}, "gps": None} try: img = PILImage.open(io.BytesIO(image_bytes)) exif = img.getexif() if not exif: return result # Standard EXIF tags for tag_id, value in exif.items(): tag_name = PILImage.ExifTags.TAGS.get(tag_id, f"0x{tag_id:04x}") # Skip binary/thumbnail blobs if tag_name in ("MakerNote", "UserComment", "PrintImageMatching"): continue result["tags"][tag_name] = str(value)[:300] if result["tags"]: result["has_exif"] = True # GPS IFD gps_ifd = exif.get_ifd(0x8825) if gps_ifd: lat_ref = gps_ifd.get(1, "N") lat_dms = gps_ifd.get(2) lng_ref = gps_ifd.get(3, "E") lng_dms = gps_ifd.get(4) if lat_dms and lng_dms: lat = _dms_to_decimal(lat_dms, lat_ref) lng = _dms_to_decimal(lng_dms, lng_ref) if lat is not None and lng is not None: result["gps"] = { "lat": lat, "lng": lng, "lat_ref": str(lat_ref), "lng_ref": str(lng_ref), "raw_lat_dms": [float(v) for v in lat_dms], "raw_lng_dms": [float(v) for v in lng_dms], } except Exception as e: result["error"] = str(e) return result def run_ocr(image_bytes: bytes) -> dict: """Run Tesseract OCR on the image.""" if not HAS_TESSERACT: return {"available": False, "text": "", "error": "pytesseract not installed"} tmp_path = UPLOADS / f"ocr_{uuid.uuid4().hex}.jpg" tmp_path.write_bytes(image_bytes) try: img = PILImage.open(tmp_path) img_gray = img.convert("L") text = pytesseract.image_to_string(img_gray, config="--psm 6") # Extract machine ID patterns match_5dash = re.search(r"(\d{5})[-\s]*(\d{6,})", text) match_5plus = re.search(r"(\d{5,})", text) return { "available": True, "raw_text": text.strip()[:500], "match_5dash6": match_5dash.group(0) if match_5dash else None, "match_5plus": match_5plus.group(0) if match_5plus else None, } finally: tmp_path.unlink(missing_ok=True) @app.post("/api/analyze") async def analyze_photo(file: UploadFile = File(...)): """Upload a photo, get back EXIF + OCR results.""" contents = await file.read() file_size = len(contents) # Save ext = Path(file.filename or "photo.jpg").suffix.lower() if ext not in {".jpg", ".jpeg", ".png", ".webp", ".bmp", ".tiff", ".tif", ".dng"}: ext = ".jpg" fname = f"{uuid.uuid4().hex}{ext}" (UPLOADS / fname).write_bytes(contents) exif_result = extract_exif(contents) ocr_result = run_ocr(contents) if HAS_TESSERACT else {"available": False, "text": ""} # Extract machine ID from OCR for auto-lookup machine_id = None if ocr_result.get("match_5dash6"): machine_id = re.sub(r"\D", "", ocr_result["match_5dash6"])[-5:] return { "filename": file.filename, "saved_as": fname, "file_size": file_size, "file_size_kb": round(file_size / 1024, 1), "exif": exif_result, "ocr": ocr_result, "machine_id": machine_id, } @app.get("/api/lookup") async def lookup_asset(machine_id: str = ""): """Look up an asset by machine_id in the canteen assets database.""" if not machine_id or not machine_id.strip(): return {"found": False, "reason": "No machine_id provided"} import sqlite3 from pathlib import Path as _Path db_path = _Path(__file__).parent.parent / "canteen-asset-tracker" / "assets.db" if not db_path.exists(): return {"found": False, "reason": f"Database not found at {db_path}"} try: conn = sqlite3.connect(str(db_path)) conn.row_factory = sqlite3.Row row = conn.execute( "SELECT id, machine_id, name, category, status, address, building_name, " "floor, room, latitude, longitude, make, model, description, photo_path " "FROM assets WHERE machine_id = ?", (machine_id.strip(),), ).fetchone() conn.close() if row: return { "found": True, "asset": { "id": row["id"], "machine_id": row["machine_id"], "name": row["name"], "category": row["category"], "status": row["status"], "address": row["address"], "building_name": row["building_name"], "floor": row["floor"], "room": row["room"], "latitude": row["latitude"], "longitude": row["longitude"], "make": row["make"], "model": row["model"], "description": row["description"], "photo_path": row["photo_path"], }, } return {"found": False, "reason": f"No asset with machine_id '{machine_id}'"} except Exception as e: return {"found": False, "reason": str(e)} @app.get("/api/uploads") async def list_uploads(): """List previously uploaded files.""" files = sorted(UPLOADS.glob("*"), key=lambda p: p.stat().st_mtime, reverse=True) return [ {"name": f.name, "size_kb": round(f.stat().st_size / 1024, 1)} for f in files[:20] ] # Mount static LAST app.mount("/", StaticFiles(directory="static", html=True), name="static") if __name__ == "__main__": import uvicorn uvicorn.run(app, host="0.0.0.0", port=8903)