Initial commit: Canteen GPS Map - interactive map combining booking + photo EXIF GPS data
Features: - Dark-themed interactive Leaflet map with marker clustering - Combined GPS sources: booking (technician check-in) + photo EXIF + account locations - Collapsible filter bar with search, source toggles, and source type dropdown - Info popups on click showing machine details, work orders, serials, photos - Account location layer toggle (3,487 accounts) - Export filtered data as JSON - Mobile-responsive (40vh map on mobile, filter collapse) - FastAPI backend with in-memory caching - Systemd service for auto-start - NPMPlus reverse proxy at gps.ourpad.casa Stats: 94 unique machines (73 booking, 21 photo), 231 total GPS points, 3,487 account locations
This commit is contained in:
@@ -0,0 +1,145 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Canteen GPS Map — FastAPI backend serving combined GPS data + interactive map."""
|
||||
import json
|
||||
import os
|
||||
import time
|
||||
from fastapi import FastAPI, Query
|
||||
from fastapi.staticfiles import StaticFiles
|
||||
from fastapi.responses import JSONResponse
|
||||
|
||||
HOST = os.environ.get("SERVE_HOST", "127.0.0.1")
|
||||
PORT = int(os.environ.get("SERVE_PORT", "8912"))
|
||||
DATA_DIR = os.path.join(os.path.dirname(__file__), "static", "data")
|
||||
|
||||
app = FastAPI(title="Canteen GPS Map")
|
||||
|
||||
# Cache the GPS data in memory (reload every 5 min)
|
||||
_data_cache = None
|
||||
_data_cache_time = 0
|
||||
CACHE_TTL = 300 # 5 minutes
|
||||
|
||||
|
||||
def load_data():
|
||||
global _data_cache, _data_cache_time
|
||||
now = time.time()
|
||||
if _data_cache is not None and (now - _data_cache_time) < CACHE_TTL:
|
||||
return _data_cache
|
||||
|
||||
data_path = os.path.join(DATA_DIR, "gps_points.json")
|
||||
if not os.path.exists(data_path):
|
||||
return {"error": "No data file found. Run build_data.py first."}
|
||||
|
||||
with open(data_path) as f:
|
||||
_data_cache = json.load(f)
|
||||
_data_cache_time = now
|
||||
return _data_cache
|
||||
|
||||
|
||||
@app.get("/health")
|
||||
async def health():
|
||||
return {"status": "ok", "timestamp": time.time()}
|
||||
|
||||
|
||||
@app.get("/api/points")
|
||||
async def get_points(
|
||||
source: str = Query("all", description="Filter by source: booking, photo, exif_dev, account, all"),
|
||||
machine_id: str = Query("", description="Filter by machine ID (partial match)"),
|
||||
account: str = Query("", description="Filter by account name (partial match)"),
|
||||
search: str = Query("", description="Search across all text fields"),
|
||||
limit: int = Query(0, description="Limit results (0 = all)"),
|
||||
summary_only: bool = Query(False, description="Return machine summary instead of all points"),
|
||||
):
|
||||
data = load_data()
|
||||
|
||||
if summary_only:
|
||||
result = data.get("machine_summary", {})
|
||||
items = [
|
||||
{"machine_id": mid, **info}
|
||||
for mid, info in result.items()
|
||||
]
|
||||
|
||||
# Filter
|
||||
if source != "all":
|
||||
items = [i for i in items if i["source"] == source]
|
||||
if machine_id:
|
||||
items = [i for i in items if machine_id.lower() in str(i.get("machine_id", "")).lower()]
|
||||
if account:
|
||||
items = [
|
||||
i
|
||||
for i in items
|
||||
if account.lower() in str(i.get("canteen_company", "")).lower()
|
||||
or account.lower() in str(i.get("canteen_name", "")).lower()
|
||||
]
|
||||
if search:
|
||||
q = search.lower()
|
||||
items = [
|
||||
i
|
||||
for i in items
|
||||
if q in json.dumps(i).lower()
|
||||
]
|
||||
|
||||
return JSONResponse(
|
||||
{
|
||||
"count": len(items),
|
||||
"total": len(result),
|
||||
"items": items[:limit] if limit else items,
|
||||
"stats": data.get("stats", {}),
|
||||
}
|
||||
)
|
||||
|
||||
# Full points mode
|
||||
points = data.get("all_points", [])
|
||||
|
||||
if source != "all":
|
||||
points = [p for p in points if p["source"] == source]
|
||||
if machine_id:
|
||||
points = [p for p in points if machine_id.lower() in str(p.get("machine_id", "")).lower()]
|
||||
if account:
|
||||
points = [
|
||||
p
|
||||
for p in points
|
||||
if account.lower() in str(p.get("account_name", "")).lower()
|
||||
or account.lower() in str(p.get("canteen_company", "")).lower()
|
||||
]
|
||||
if search:
|
||||
q = search.lower()
|
||||
points = [p for p in points if q in json.dumps(p).lower()]
|
||||
|
||||
return JSONResponse(
|
||||
{
|
||||
"count": len(points),
|
||||
"total": len(data.get("all_points", [])),
|
||||
"points": points[:limit] if limit else points,
|
||||
"stats": data.get("stats", {}),
|
||||
}
|
||||
)
|
||||
|
||||
|
||||
@app.get("/api/accounts")
|
||||
async def get_accounts():
|
||||
"""Return just the account GPS points for the account layer."""
|
||||
data = load_data()
|
||||
return JSONResponse(
|
||||
{
|
||||
"count": len(data.get("account_points", [])),
|
||||
"accounts": data.get("account_points", []),
|
||||
}
|
||||
)
|
||||
|
||||
|
||||
@app.get("/api/stats")
|
||||
async def get_stats():
|
||||
"""Return data statistics."""
|
||||
data = load_data()
|
||||
return JSONResponse(data.get("stats", {}))
|
||||
|
||||
|
||||
# Mount static files AFTER API routes
|
||||
app.mount("/", StaticFiles(directory=os.path.join(os.path.dirname(__file__), "static"), html=True), name="static")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
import uvicorn
|
||||
|
||||
print(f"Starting Canteen GPS Map on {HOST}:{PORT}")
|
||||
uvicorn.run(app, host=HOST, port=PORT)
|
||||
Reference in New Issue
Block a user