219 lines
8.2 KiB
Python
Executable File
219 lines
8.2 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""Export FT360/Traccar device status to /var/www/ops/data for dashboard use.
|
|
|
|
Uses the same backend logic as the FT360 MCP so dashboard semantics match Hermes tools:
|
|
- last_contact_time: device contacted Traccar
|
|
- gps_fix_time: actual GPS fix time
|
|
- is_location_fresh: GPS fix age <= 10 minutes
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import importlib.util
|
|
import json
|
|
import site
|
|
import time
|
|
import urllib.parse
|
|
import urllib.request
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
|
|
# Cron runs scripts with system Python. The FT360 MCP uses the shared
|
|
# ops-portal venv for FastMCP, so add that venv's site-packages before
|
|
# importing the MCP server module.
|
|
site.addsitedir("/opt/ops-portal/venv/lib/python3.13/site-packages")
|
|
|
|
SERVER_PATH = Path("/root/docker/ft360-mcp/server.py")
|
|
OUTPUT = Path("/var/www/ops/data/ft360-devices.json")
|
|
GEOCODE_CACHE = Path("/var/www/ops/data/ft360-geocode-cache.json")
|
|
GEOCODE_TTL_SECONDS = 7 * 24 * 3600
|
|
|
|
|
|
def _load_geocode_cache() -> dict:
|
|
try:
|
|
return json.loads(GEOCODE_CACHE.read_text())
|
|
except Exception:
|
|
return {}
|
|
|
|
|
|
def _save_geocode_cache(cache: dict) -> None:
|
|
GEOCODE_CACHE.parent.mkdir(parents=True, exist_ok=True)
|
|
tmp = GEOCODE_CACHE.with_suffix(".tmp")
|
|
tmp.write_text(json.dumps(cache, indent=2, sort_keys=True))
|
|
tmp.replace(GEOCODE_CACHE)
|
|
|
|
|
|
def _reverse_geocode(lat: float, lon: float, cache: dict) -> dict:
|
|
"""Reverse geocode lat/lon to city and country code using Nominatim.
|
|
|
|
Cache by ~100m cells to avoid hammering OSM from the 1-minute exporter.
|
|
"""
|
|
key = f"{lat:.3f},{lon:.3f}"
|
|
now = time.time()
|
|
cached = cache.get(key)
|
|
if cached and now - cached.get("ts", 0) < GEOCODE_TTL_SECONDS:
|
|
return cached.get("data", {})
|
|
|
|
params = urllib.parse.urlencode({
|
|
"format": "jsonv2",
|
|
"lat": f"{lat:.7f}",
|
|
"lon": f"{lon:.7f}",
|
|
"zoom": "10",
|
|
"addressdetails": "1",
|
|
})
|
|
req = urllib.request.Request(
|
|
f"https://nominatim.openstreetmap.org/reverse?{params}",
|
|
headers={"User-Agent": "ITPP-FT360-Dashboard/1.0 (info@itpropartner.com)"},
|
|
)
|
|
try:
|
|
with urllib.request.urlopen(req, timeout=10) as resp:
|
|
raw = resp.read().decode("utf-8", "replace")
|
|
payload = json.loads(raw)
|
|
addr = payload.get("address", {}) or {}
|
|
city = (
|
|
addr.get("city") or addr.get("town") or addr.get("village") or
|
|
addr.get("municipality") or addr.get("county") or addr.get("state") or ""
|
|
)
|
|
data = {
|
|
"city": city,
|
|
"country": addr.get("country", ""),
|
|
"country_code": (addr.get("country_code", "") or "").upper(),
|
|
"state": addr.get("state", ""),
|
|
"display_name": payload.get("display_name", ""),
|
|
}
|
|
except Exception as exc:
|
|
data = {"city": "", "country": "", "country_code": "", "state": "", "display_name": "", "geocode_error": str(exc)}
|
|
|
|
cache[key] = {"ts": now, "data": data}
|
|
return data
|
|
|
|
|
|
# ─── Saved locations for ETA computation ─────────────────────────────────────
|
|
# Edit this dict to add/remove locations. Each entry must have name, icon,
|
|
# lat, lon. OSRM walking ETAs are computed automatically by _compute_etas().
|
|
SAVED_LOCATIONS = [
|
|
{"name": "Airbnb", "icon": "🏠", "lat": 10.3981, "lon": -75.5534},
|
|
{"name": "Old City", "icon": "🏛️", "lat": 10.4236, "lon": -75.5503},
|
|
{"name": "Getsemaní", "icon": "🎨", "lat": 10.4203, "lon": -75.5467},
|
|
{"name": "Bocagrande", "icon": "🏖️", "lat": 10.4053, "lon": -75.5578},
|
|
]
|
|
|
|
|
|
def _compute_etas(from_lat: float, from_lon: float) -> list[dict]:
|
|
"""Compute walking distance and ETA to each saved location using OSRM."""
|
|
results = []
|
|
for loc in SAVED_LOCATIONS:
|
|
# Straight-line distance as fallback
|
|
from math import radians, sin, cos, sqrt, atan2
|
|
R = 3958.8
|
|
dlat = radians(loc["lat"] - from_lat)
|
|
dlon = radians(loc["lon"] - from_lon)
|
|
a = sin(dlat/2)**2 + cos(radians(from_lat)) * cos(radians(loc["lat"])) * sin(dlon/2)**2
|
|
straight_mi = 2 * R * atan2(sqrt(a), sqrt(1-a))
|
|
|
|
# Try OSRM for actual road/path distance
|
|
try:
|
|
url = (
|
|
f"https://router.project-osrm.org/route/v1/foot/"
|
|
f"{from_lon},{from_lat};{loc['lon']},{loc['lat']}"
|
|
f"?overview=false"
|
|
)
|
|
req = urllib.request.Request(url, headers={"User-Agent": "ITPP-FT360/1.0"})
|
|
with urllib.request.urlopen(req, timeout=8) as resp:
|
|
route = json.loads(resp.read().decode())
|
|
if route.get("code") == "Ok" and route.get("routes"):
|
|
leg = route["routes"][0]["legs"][0]
|
|
meters = leg["distance"]
|
|
seconds = leg["duration"]
|
|
miles = round(meters * 0.000621371, 1)
|
|
minutes = max(1, round(seconds / 60))
|
|
results.append({
|
|
"name": loc["name"],
|
|
"icon": loc["icon"],
|
|
"time": f"{minutes} min",
|
|
"time_minutes": minutes,
|
|
"distance_mi": miles,
|
|
"distance_str": f"{miles} mi",
|
|
"source": "OSRM walking",
|
|
})
|
|
continue
|
|
except Exception:
|
|
pass
|
|
|
|
# Fallback: straight-line estimate (3 mph walking)
|
|
minutes_fallback = max(1, round((straight_mi / 3.0) * 60))
|
|
results.append({
|
|
"name": loc["name"],
|
|
"icon": loc["icon"],
|
|
"time": f"{minutes_fallback} min",
|
|
"time_minutes": minutes_fallback,
|
|
"distance_mi": round(straight_mi, 1),
|
|
"distance_str": f"{straight_mi:.1f} mi",
|
|
"source": "straight-line estimate",
|
|
})
|
|
return results
|
|
|
|
|
|
|
|
|
|
def main() -> int:
|
|
spec = importlib.util.spec_from_file_location("ft360_mcp_server", SERVER_PATH)
|
|
if spec is None or spec.loader is None:
|
|
raise RuntimeError(f"Could not load {SERVER_PATH}")
|
|
mod = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(mod)
|
|
devices = mod._get_devices_raw() # noqa: SLF001 - intentional shared backend
|
|
cache = _load_geocode_cache()
|
|
for device in devices:
|
|
lat = device.get("latitude")
|
|
lon = device.get("longitude")
|
|
if isinstance(lat, (int, float)) and isinstance(lon, (int, float)):
|
|
geo = _reverse_geocode(float(lat), float(lon), cache)
|
|
device["city"] = geo.get("city", "")
|
|
device["country"] = geo.get("country", "")
|
|
device["country_code"] = geo.get("country_code", "")
|
|
device["state"] = geo.get("state", "")
|
|
device["display_name"] = geo.get("display_name", "")
|
|
else:
|
|
device["city"] = ""
|
|
device["country"] = ""
|
|
device["country_code"] = ""
|
|
device["state"] = ""
|
|
device["display_name"] = ""
|
|
_save_geocode_cache(cache)
|
|
payload = {
|
|
"generated_at": datetime.now(timezone.utc).isoformat(),
|
|
"count": len(devices),
|
|
"devices": devices,
|
|
}
|
|
# Try to append daily stats from the companion script
|
|
try:
|
|
import subprocess
|
|
r = subprocess.run(
|
|
["python3", "/root/.hermes/scripts/ft360-daily-stats.py"],
|
|
capture_output=True, text=True, timeout=120)
|
|
if r.returncode == 0:
|
|
payload["daily_stats"] = json.loads(r.stdout)
|
|
except Exception:
|
|
pass
|
|
|
|
# Compute ETAs from current device position to saved locations via OSRM (free, no API key)
|
|
for device in devices:
|
|
lat = device.get("latitude")
|
|
lon = device.get("longitude")
|
|
if isinstance(lat, (int, float)) and isinstance(lon, (int, float)):
|
|
payload["etas"] = _compute_etas(float(lat), float(lon))
|
|
break
|
|
else:
|
|
payload["etas"] = []
|
|
|
|
OUTPUT.parent.mkdir(parents=True, exist_ok=True)
|
|
tmp = OUTPUT.with_suffix(".tmp")
|
|
tmp.write_text(json.dumps(payload, indent=2, default=str))
|
|
tmp.replace(OUTPUT)
|
|
print(json.dumps({"ok": True, "output": str(OUTPUT), "count": len(devices)}, indent=2))
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|