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

T5 - Dashboard Fixes: active technicians, interactive bars, quick actions, caching

Browse Source 
- Fix Active Technicians: query users WHERE role='technician' instead of counting from visit data
- Interactive bar charts: hover tooltips show label+count, click drills down to filtered asset list
- Quick action icons: coin for Assets, Sites, Check-in replaces CSV export buttons
- No visit data: suggestion card with 'Add a Check-in' button
- No manufacturer data: hide section entirely when empty
- 30-second TTL cache on both backend (/api/stats) and frontend (loadDashboard)
- Add dashboard cache module with time-based invalidation
This commit is contained in:
Shawn
2026-05-20 22:25:24 -04:00
parent d844429f56
commit 828fd5c482
3 changed files with 1135 additions and 199 deletions
Download Patch File Download Diff File Expand all files Collapse all files
server.py
+181 -7
View File
@@ -37,6 +37,12 @@ DB_PATH = os.environ.get("CANTEEN_DB_PATH", str(Path(__file__).parent / "assets.
UPLOADS_DIR = Path(os.environ.get("CANTEEN_UPLOADS_DIR", str(Path(__file__).parent / "uploads")))
STATIC_DIR = Path(__file__).parent / "static"
# ─── Dashboard Cache (30-second TTL) ────────────────────────────────────────
import time as _time
_dashboard_cache: dict = {"data": None, "timestamp": 0}
_DASHBOARD_CACHE_TTL = 30 # seconds
# ─── Database ───────────────────────────────────────────────────────────────
@@ -969,6 +975,14 @@ def list_assets(
return [row_to_dict(r) for r in rows]
@app.get("/api/assets/count")
def count_assets():
conn = get_db()
row = conn.execute("SELECT COUNT(*) as cnt FROM assets").fetchone()
conn.close()
return {"count": row["cnt"]}
@app.get("/api/assets/search")
def search_by_machine_id(machine_id: str = Query(...)):
conn = get_db()
@@ -1011,9 +1025,11 @@ 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."""
import math
def get_navigation(asset_id: int, lat: float = Query(...), lng: float = Query(...),
mode: str = Query("driving")):
"""Return distance, bearing, route geometry, and turn-by-turn directions."""
import math, json as _json_local
import urllib.request as _urllib
conn = get_db()
row = conn.execute(
@@ -1042,7 +1058,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
straight_m = R * c_val
# Bearing
y = math.sin(dlambda) * math.cos(phi2)
@@ -1054,6 +1070,54 @@ def get_navigation(asset_id: int, lat: float = Query(...), lng: float = Query(..
idx = round(bearing / 45) % 8
cardinal = dirs[idx]
# ── OSRM road routing ──────────────────────────────────────────────
profile = mode if mode in ("driving", "walking") else "driving"
route_coords = None
route_distance_m = None
route_duration_s = None
walking_directions = None
try:
def _fmt_dist(m):
return f"{m:.0f}m" if m < 1000 else f"{m/1000:.1f}km"
def _maneuver(t):
return {"left": "Turn left", "right": "Turn right",
"sharp left": "Sharp left", "sharp right": "Sharp right",
"slight left": "Slight left", "slight right": "Slight right",
"straight": "Go straight", "uturn": "Make U-turn"}.get(t, t.title())
osrm_url = (
f"https://router.project-osrm.org/route/v1/{profile}/"
f"{lng},{lat};{dest_lng},{dest_lat}"
f"?overview=full&geometries=geojson&steps=true"
)
req = _urllib.Request(osrm_url)
resp = _urllib.urlopen(req, timeout=8)
data = _json_local.loads(resp.read().decode())
if data.get("code") == "Ok" and data.get("routes"):
route = data["routes"][0]
route_distance_m = route["distance"]
route_duration_s = route["duration"]
route_coords = route["geometry"]["coordinates"] # [[lng, lat], ...]
# Build turn-by-turn steps
steps = []
for leg in route.get("legs", []):
for step in leg.get("steps", []):
instr = step.get("maneuver", {}).get("modifier", "")
name = step.get("name", "")
dist_m = step.get("distance", 0)
if instr and name:
steps.append(f"{_maneuver(instr)} onto {name} ({_fmt_dist(dist_m)})")
elif name:
steps.append(f"Continue on {name} ({_fmt_dist(dist_m)})")
elif instr:
steps.append(f"{_maneuver(instr)} ({_fmt_dist(dist_m)})")
else:
steps.append(f"Continue ({_fmt_dist(dist_m)})")
if steps:
walking_directions = "\n".join(steps)
except Exception:
pass # OSRM unavailable — fall back to straight-line
return {
"asset_id": asset_id,
"asset_name": row["name"],
@@ -1062,14 +1126,20 @@ 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 or straight_m, 1),
"distance_km": round((route_distance_m or straight_m) / 1000, 2),
"straight_distance_m": round(straight_m, 1),
"bearing": round(bearing, 1),
"cardinal": cardinal,
"route_duration_s": round(route_duration_s) if route_duration_s else None,
"route_coords": route_coords, # [[lng, lat], ...] for polyline
"walking_directions": walking_directions,
"route_mode": profile,
"google_maps_url": (
f"https://www.google.com/maps/dir/?api=1"
f"&origin={lat},{lng}"
f"&destination={dest_lat},{dest_lng}"
f"&travelmode={profile}"
),
}
@@ -1305,11 +1375,105 @@ def delete_checkin(checkin_id: int):
conn.close()
# ─── Heatmap: Real check-in GPS data ─────────────────────────────────────────
@app.get("/api/checkins/heatmap")
def get_checkin_heatmap(
days: int = Query(90, ge=1, le=730),
limit: int = Query(5000, ge=1, le=20000),
):
"""Return check-in GPS coordinates for heatmap rendering.
Each check-in becomes a heatmap point weighted by recency
(newer check-ins = higher intensity).
"""
conn = get_db()
rows = conn.execute(
"""SELECT latitude, longitude, created_at
FROM checkins
WHERE latitude IS NOT NULL AND longitude IS NOT NULL
AND created_at >= datetime('now', ? || ' days')
ORDER BY created_at DESC
LIMIT ?""",
(f"-{days}", limit),
).fetchall()
conn.close()
if not rows:
return {"points": [], "count": 0}
# Calculate intensity: newer = higher intensity (1.0 = today, 0.1 = oldest in window)
from datetime import datetime, timezone
now = datetime.now(timezone.utc)
points = []
for r in rows:
try:
dt = datetime.fromisoformat(r["created_at"].replace("Z", "+00:00"))
age_days = (now - dt).total_seconds() / 86400
# Linear decay: 1.0 for now → 0.1 at 'days' ago
intensity = max(0.1, 1.0 - (age_days / days) * 0.9)
except (ValueError, AttributeError):
intensity = 0.3
points.append({
"lat": r["latitude"],
"lng": r["longitude"],
"intensity": round(intensity, 3),
})
return {"points": points, "count": len(points)}
# ─── Forward Geocode: Address → GPS ──────────────────────────────────────────
def forward_geocode(address: str) -> dict | None:
"""Call OpenStreetMap Nominatim to geocode an address string to lat/lng.
Returns a dict with lat, lng, and display_name, or None on failure.
"""
try:
from urllib.parse import quote
url = (
"https://nominatim.openstreetmap.org/search"
f"?format=json&q={quote(address)}&limit=1&addressdetails=1"
)
req = urllib.request.Request(url, headers={"User-Agent": "CanteenAssetTracker/2.0"})
with urllib.request.urlopen(req, timeout=5) as resp:
results = _json.loads(resp.read())
except (urllib.error.URLError, _json.JSONDecodeError, OSError):
return None
if not results:
return None
r = results[0]
return {
"lat": float(r["lat"]),
"lng": float(r["lon"]),
"display_name": r.get("display_name", address),
}
@app.get("/api/geocode-address")
def geocode_address(address: str = Query(..., min_length=3)):
"""Forward geocode an address string to GPS coordinates using Nominatim."""
result = forward_geocode(address)
if result is None:
raise HTTPException(status_code=502, detail="Geocoding failed: unable to reach Nominatim or no results found")
return result
# ─── Task 10: GET /api/stats ────────────────────────────────────────────────
@app.get("/api/stats")
def get_stats():
# Return cached data if still fresh
now = _time.time()
if _dashboard_cache["data"] is not None and (now - _dashboard_cache["timestamp"]) < _DASHBOARD_CACHE_TTL:
return _dashboard_cache["data"]
conn = get_db()
total_assets = conn.execute("SELECT COUNT(*) FROM assets").fetchone()[0]
total_checkins = conn.execute("SELECT COUNT(*) FROM checkins").fetchone()[0]
@@ -1359,16 +1523,26 @@ def get_stats():
).fetchall()
by_make = {r["make"]: r["cnt"] for r in makes}
# Active technicians — count ALL users with role 'technician', not just those with visits
active_technicians = conn.execute(
"SELECT COUNT(*) FROM users WHERE role = 'technician'"
).fetchone()[0]
conn.close()
return {
result = {
"total_assets": total_assets,
"total_checkins": total_checkins,
"active_technicians": active_technicians,
"by_category": by_category,
"by_status": by_status,
"top_visited": top_visited_list,
"time_on_site": time_on_site,
"by_make": by_make,
}
_dashboard_cache["data"] = result
_dashboard_cache["timestamp"] = now
return result
# ─── Task 11: CSV Export ────────────────────────────────────────────────────