Files
lingniu-vehicle-ingest/vehicle-map/server.py
T

225 lines
8.2 KiB
Python

"""Vehicle Map static server and server-side proxy for the vehicle open platform."""
from concurrent.futures import ThreadPoolExecutor
import hashlib
from http.server import HTTPServer, SimpleHTTPRequestHandler
import json
import os
from pathlib import Path
import threading
import time
from socketserver import ThreadingMixIn
from urllib.error import HTTPError, URLError
from urllib.parse import urlparse
from urllib.request import Request, urlopen
ROOT = Path(__file__).resolve().parent
def _static_asset_version():
"""Content fingerprint used to bypass upstream static-asset caches on release."""
digest = hashlib.sha256()
for filename in ("app.js", "styles.css"):
digest.update((ROOT / filename).read_bytes())
return digest.hexdigest()[:16]
STATIC_ASSET_VERSION = _static_asset_version()
HOST = os.getenv("VEHICLE_MAP_HOST", "0.0.0.0")
PORT = int(os.getenv("VEHICLE_MAP_PORT", "20800"))
OPEN_PLATFORM_BASE_URL = os.getenv("OPEN_PLATFORM_BASE_URL", "https://open.d.lnoneos.com").rstrip("/")
OPEN_PLATFORM_APP_KEY = os.getenv("OPEN_PLATFORM_APP_KEY", "").strip()
UPSTREAM_TIMEOUT_SECONDS = float(os.getenv("UPSTREAM_TIMEOUT_SECONDS", "15"))
DASHBOARD_CACHE_SECONDS = int(os.getenv("DASHBOARD_CACHE_SECONDS", "5"))
STATION_CACHE_SECONDS = int(os.getenv("STATION_CACHE_SECONDS", "3600"))
_cache_lock = threading.Lock()
_cache = {}
class ThreadingHTTPServer(ThreadingMixIn, HTTPServer):
daemon_threads = True
allow_reuse_address = True
def _json_bytes(value):
return json.dumps(value, ensure_ascii=False, separators=(",", ":")).encode("utf-8")
def _post_open_platform(path, body):
if not OPEN_PLATFORM_APP_KEY:
raise RuntimeError("OPEN_PLATFORM_APP_KEY is not configured")
request = Request(
OPEN_PLATFORM_BASE_URL + path,
data=_json_bytes(body),
method="POST",
headers={
"Authorization": "Bearer " + OPEN_PLATFORM_APP_KEY,
"Content-Type": "application/json",
"Accept": "application/json",
"User-Agent": "lingniu-vehicle-map/1.0",
},
)
try:
with urlopen(request, timeout=UPSTREAM_TIMEOUT_SECONDS) as response:
payload = json.load(response)
except HTTPError as exc:
detail = exc.read(2048).decode("utf-8", errors="replace")
raise RuntimeError(f"open platform returned HTTP {exc.code}: {detail}") from exc
except URLError as exc:
raise RuntimeError(f"open platform unavailable: {exc.reason}") from exc
if payload.get("code") != "SUCCESS":
raise RuntimeError(f"open platform rejected request: {payload.get('code')} {payload.get('message')}")
return payload.get("data") or []
def _cached(key, ttl_seconds, loader):
now = time.time()
with _cache_lock:
cached = _cache.get(key)
if cached and now - cached[0] < ttl_seconds:
return cached[1]
value = loader()
with _cache_lock:
_cache[key] = (now, value)
return value
def _load_stations():
return _cached(
"stations",
STATION_CACHE_SECONDS,
lambda: _post_open_platform("/api/v1/hydrogen-stations/query", {}),
)
def _load_dashboard():
today = time.strftime("%Y-%m-%d", time.localtime())
with ThreadPoolExecutor(max_workers=3) as executor:
realtime_future = executor.submit(
_post_open_platform, "/api/v1/vehicles/realtime/query", {}
)
mileage_future = executor.submit(
_post_open_platform, "/api/v1/vehicles/mileage/query", {"date": today}
)
stations_future = executor.submit(_load_stations)
vehicles = realtime_future.result()
mileage_rows = mileage_future.result()
stations = stations_future.result()
mileage_by_vin = {
row.get("vin"): row
for row in mileage_rows
if row.get("vin") and row.get("status") == "NORMAL"
}
# The daily activity view places every vehicle that reported
# today is placed in exactly one bucket using its latest selected speed.
active_vehicles = [item for item in vehicles if item.get("activeToday")]
driving = sum(1 for item in active_vehicles if float(item.get("speedKmh") or 0) > 3)
idle = len(active_vehicles) - driving
offline = sum(1 for item in vehicles if item.get("motionStatus") == "offline")
active_today = len(active_vehicles)
daily_mileage = round(
sum(float(row.get("dailyMileageKm") or 0) for row in mileage_rows), 3
)
monthly_hydrogen = round(
sum(float(station.get("monthlyHydrogenKg") or 0) for station in stations), 3
)
for vehicle in vehicles:
mileage = mileage_by_vin.get(vehicle.get("vin"), {})
vehicle["dailyMileageKm"] = mileage.get("dailyMileageKm", 0)
return {
"status": "ok",
"asOf": time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()),
"date": today,
"summary": {
"totalVehicles": len(vehicles),
"onlineVehicles": active_today,
"drivingVehicles": driving,
"idleVehicles": idle,
"offlineVehicles": offline,
"todayMileageKm": daily_mileage,
"monthlyHydrogenKg": monthly_hydrogen,
"totalStations": len(stations),
"cooperativeStations": sum(1 for item in stations if item.get("cooperative")),
},
"vehicles": vehicles,
"stations": stations,
}
class VehicleMapHandler(SimpleHTTPRequestHandler):
server_version = "LingniuVehicleMap/1.0"
def do_GET(self):
path = urlparse(self.path).path
if path in {"/", "/index.html"}:
self._serve_index()
return
if path == "/api/health":
self._write_json(200, {
"status": "ok",
"service": "vehicle-map",
"openPlatformConfigured": bool(OPEN_PLATFORM_APP_KEY),
})
return
if path == "/api/dashboard":
try:
payload = _cached("dashboard", DASHBOARD_CACHE_SECONDS, _load_dashboard)
self._write_json(200, payload)
except Exception as exc: # keep upstream details server-side only
self.log_error("dashboard refresh failed: %s", exc)
self._write_json(502, {
"status": "error",
"code": "UPSTREAM_UNAVAILABLE",
"message": "地图数据暂时不可用,请稍后重试",
})
return
super().do_GET()
def _serve_index(self):
try:
template = (ROOT / "index.html").read_text(encoding="utf-8")
body = template.replace("__ASSET_VERSION__", STATIC_ASSET_VERSION).encode("utf-8")
except OSError as exc:
self.log_error("index template unavailable: %s", exc)
self.send_error(500, "Map application is temporarily unavailable")
return
self.send_response(200)
self.send_header("Content-Type", "text/html; charset=utf-8")
self.send_header("Content-Length", str(len(body)))
self.send_header("Cache-Control", "no-cache")
self.end_headers()
self.wfile.write(body)
def end_headers(self):
if self.path == "/" or self.path.split("?", 1)[0].endswith((".html", ".css", ".js", ".svg")):
self.send_header("Cache-Control", "no-cache")
self.send_header("X-Content-Type-Options", "nosniff")
self.send_header("Referrer-Policy", "strict-origin-when-cross-origin")
self.send_header("X-Frame-Options", "SAMEORIGIN")
super().end_headers()
def _write_json(self, status, payload, headers=None):
body = _json_bytes(payload)
self.send_response(status)
self.send_header("Content-Type", "application/json; charset=utf-8")
self.send_header("Cache-Control", "no-store")
self.send_header("Content-Length", str(len(body)))
for name, value in (headers or {}).items():
self.send_header(name, value)
self.end_headers()
self.wfile.write(body)
if __name__ == "__main__":
server = ThreadingHTTPServer((HOST, PORT), VehicleMapHandler)
print(f"Lingniu Vehicle Map listening on http://{HOST}:{PORT}")
try:
server.serve_forever()
except KeyboardInterrupt:
pass
finally:
server.server_close()