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

Major app overhaul: MSFS data import, lookup-only Find tab, OSRM nav, cleanup

Browse Source
This commit is contained in:
Shawn
2026-05-28 21:57:58 -04:00
parent a9f2f88604
commit f3d176116a
3 changed files with 848 additions and 976 deletions
Download Patch File Download Diff File Expand all files Collapse all files
server.py
+97 -53
View File
@@ -21,7 +21,12 @@ from contextlib import asynccontextmanager
from datetime import date
from pathlib import Path
import pytesseract
try:
import pytesseract
_HAS_TESSERACT = True
except ImportError:
pytesseract = None
_HAS_TESSERACT = False
import piexif
from PIL import Image as PILImage
@@ -1123,13 +1128,23 @@ def get_asset(asset_id: int):
# ─── Navigation endpoint ────────────────────────────────────────────────────
@app.get("/api/assets/{asset_id}/navigation")
def get_navigation(asset_id: int, lat: float = Query(...), lng: float = Query(...)):
"""Return distance, bearing, and coordinates for navigation from GPS to asset."""
def get_navigation(
asset_id: int,
lat: float = Query(...),
lng: float = Query(...),
mode: str = Query("driving", pattern="^(driving|walking)$"),
):
"""Return navigation info from GPS to asset using OSRM (falls back to Haversine).
Accepts optional mode: 'driving' (default) or 'walking'.
On success returns route_coords, route_duration_s, walking_directions, route_mode
plus the standard distance/bearing fields.
"""
import math
conn = get_db()
row = conn.execute(
"SELECT id, name, latitude, longitude, machine_id, address FROM assets WHERE id = ?",
"SELECT id, name, latitude, longitude, machine_id, address, walking_directions FROM assets WHERE id = ?",
(asset_id,),
).fetchone()
conn.close()
@@ -1145,7 +1160,47 @@ def get_navigation(asset_id: int, lat: float = Query(...), lng: float = Query(..
dest_lat, dest_lng = row["latitude"], row["longitude"]
# Haversine distance
# ── Try OSRM first ──────────────────────────────────────────────────
osrm_profile = "driving" if mode == "driving" else "foot"
route_coords = None
route_duration_s = None
route_distance_m = None
walking_directions = None
try:
url = (
f"https://router.project-osrm.org/route/v1/{osrm_profile}/{lng},{lat};{dest_lng},{dest_lat}"
"?overview=full&geometries=geojson&steps=true&alternatives=false"
)
req = urllib.request.Request(url, headers={"User-Agent": "CanteenAssetTracker/1.0"})
with urllib.request.urlopen(req, timeout=10) as resp:
data = _json.loads(resp.read().decode())
if data.get("code") == "Ok" and data.get("routes"):
route = data["routes"][0]
coords_raw = route.get("geometry", {}).get("coordinates", [])
route_coords = [{"lat": c[1], "lng": c[0]} for c in coords_raw]
route_duration_s = round(route.get("duration", 0))
route_distance_m = round(route.get("distance", 0))
# Build step-by-step walking directions (only for walking mode)
walking_directions = None
if mode == "walking":
steps = []
for leg in route.get("legs", []):
for step in leg.get("steps", []):
steps.append({
"instruction": step.get("maneuver", {}).get("instruction", step.get("name", "")),
"distance_m": round(step.get("distance", 0)),
"duration_s": round(step.get("duration", 0)),
})
if steps:
walking_directions = steps
except Exception:
# OSRM failed — fall through to Haversine below
pass
# ── Haversine distance (always computed for fallback / reference) ────
R = 6371000 # Earth radius in meters
phi1 = math.radians(lat)
phi2 = math.radians(dest_lat)
@@ -1154,7 +1209,7 @@ def get_navigation(asset_id: int, lat: float = Query(...), lng: float = Query(..
a_val = math.sin(dphi / 2) ** 2 + math.cos(phi1) * math.cos(phi2) * math.sin(dlambda / 2) ** 2
c_val = 2 * math.atan2(math.sqrt(a_val), math.sqrt(1 - a_val))
distance_m = R * c_val
haversine_distance_m = R * c_val
# Bearing
y = math.sin(dlambda) * math.cos(phi2)
@@ -1166,7 +1221,7 @@ def get_navigation(asset_id: int, lat: float = Query(...), lng: float = Query(..
idx = round(bearing / 45) % 8
cardinal = dirs[idx]
return {
result = {
"asset_id": asset_id,
"asset_name": row["name"],
"asset_lat": dest_lat,
@@ -1174,10 +1229,11 @@ def get_navigation(asset_id: int, lat: float = Query(...), lng: float = Query(..
"asset_address": row["address"] or None,
"origin_lat": lat,
"origin_lng": lng,
"distance_meters": round(distance_m, 1),
"distance_km": round(distance_m / 1000, 2),
"distance_meters": round(route_distance_m if route_distance_m else haversine_distance_m, 1),
"distance_km": round((route_distance_m if route_distance_m else haversine_distance_m) / 1000, 2),
"bearing": round(bearing, 1),
"cardinal": cardinal,
"route_mode": mode,
"google_maps_url": (
f"https://www.google.com/maps/dir/?api=1"
f"&origin={lat},{lng}"
@@ -1185,6 +1241,21 @@ def get_navigation(asset_id: int, lat: float = Query(...), lng: float = Query(..
),
}
if route_coords:
result["route_coords"] = route_coords
result["route_duration_s"] = route_duration_s
result["route_distance_m"] = route_distance_m
result["route_mode"] = mode
result["osrm_source"] = "osrm"
if mode == "walking" and walking_directions:
result["walking_directions"] = walking_directions
else:
result["osrm_source"] = "haversine"
result["walking_directions"] = row["walking_directions"] or None
return result
# ─── Task 6: PUT / DELETE /api/assets/{id} ─────────────────────────────────
@@ -1468,11 +1539,6 @@ class LoginRequest(BaseModel):
password: str
remember_me: bool = False
class GeofencePointCheck(BaseModel):
lat: float
lng: float
# ─── Phase C: Auth API ───────────────────────────────────────────────────────
@@ -1554,26 +1620,7 @@ def logout(request: Request):
# ─── Phase 0: Proximity & Geofence Check ─────────────────────────────────────
def _point_in_polygon(lat: float, lng: float, polygon: list) -> bool:
"""Ray-casting algorithm for point-in-polygon test."""
inside = False
n = len(polygon)
if n < 3:
return False
j = n - 1
for i in range(n):
yi = polygon[i]["lat"] if isinstance(polygon[i], dict) else polygon[i][0]
xi = polygon[i]["lng"] if isinstance(polygon[i], dict) else polygon[i][1]
yj = polygon[j]["lat"] if isinstance(polygon[j], dict) else polygon[j][0]
xj = polygon[j]["lng"] if isinstance(polygon[j], dict) else polygon[j][1]
if ((yi > lat) != (yj > lat)) and (lng < (xj - xi) * (lat - yi) / (yj - yi) + xi):
inside = not inside
j = i
return inside
# ─── Phase 0: Proximity Check ─────────────────────────────────────────────────
@app.get("/api/proximity")
@@ -1608,24 +1655,6 @@ def proximity_check(
return results
@app.post("/api/geofences/check")
def check_geofence_point(body: GeofencePointCheck):
"""
Check if a GPS point falls inside any geofence polygon.
Returns list of matching geofences.
"""
conn = get_db()
rows = conn.execute("SELECT * FROM geofences ORDER BY name").fetchall()
conn.close()
matches = []
for row in rows:
points = _json.loads(row["points"])
if _point_in_polygon(body.lat, body.lng, points):
matches.append(row_to_dict(row))
return matches
# ─── Phase C: Visits API & Auto-visit Logging ────────────────────────────────
@@ -2050,11 +2079,26 @@ async def ocr_sticker(file: UploadFile = File(...), exif_data: str = Form(None))
img = PILImage.open(ocr_path)
# Preprocess: convert to grayscale and increase contrast for better OCR
img_gray = img.convert("L")
if not _HAS_TESSERACT:
raise RuntimeError(
"Tesseract OCR library (pytesseract) is not installed. "
"Run: pip install pytesseract && apt-get install -y tesseract-ocr"
)
text = pytesseract.image_to_string(img_gray, config="--psm 6")
except (RuntimeError, ImportError) as e:
if not saved_path:
ocr_path.unlink(missing_ok=True)
raise HTTPException(
status_code=503,
detail=f"OCR service unavailable: {str(e)}",
)
except Exception as e:
if not saved_path:
ocr_path.unlink(missing_ok=True)
raise HTTPException(status_code=500, detail=f"OCR processing failed: {str(e)}")
raise HTTPException(
status_code=500,
detail=f"OCR processing error: {str(e)}. Tesseract binary may not be installed system-wide. Run: apt-get install -y tesseract-ocr",
)
# Clean up temp file (but keep permanent photos)
if not saved_path: