""" Mauritius Bus Route Planner — tiny stdlib HTTP server. Serves the static frontend and a small JSON API backed by busapi.py (journey planning via mauritius-buses.com) and geocode.py (stop -> coordinates via OpenStreetMap / Nominatim). python3 server.py # http://localhost:8000 PORT=9000 python3 server.py Endpoints: GET /api/stops -> [{id, name}, ...] GET /api/plan?day=&from=&to= -> itinerary with coords attached """ import hashlib import json import math import os import sys import threading import time import traceback import warnings import gzip as _gzip from email.utils import formatdate from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from urllib.parse import urlparse, parse_qs # Text assets worth gzipping on the wire (the bundled network.js/json are ~2.3MB # uncompressed but ~5x smaller gzipped — a big win for first load and the Android # data refresh). Binary types (png/ico) are already compressed; skip them. GZIP_EXTS = {".html", ".js", ".css", ".json", ".svg"} # Silence the harmless LibreSSL/urllib3 notice on stock macOS Python. warnings.filterwarnings("ignore", message=".*OpenSSL.*") import busapi import events as events_mod import geocode import nearby import roads try: import psutil # process + system CPU/RAM metrics for /metrics _PROC = psutil.Process() _PROC.cpu_percent(interval=None) # prime the per-process CPU delta psutil.cpu_percent(interval=None) # prime the system CPU delta except Exception: # noqa: BLE001 — metrics are best-effort; never block startup psutil = None _PROC = None ROOT = os.path.dirname(os.path.abspath(__file__)) PUBLIC = os.path.join(ROOT, "public") PORT = int(os.environ.get("PORT", "8000")) # /api/health reports "degraded" (HTTP 503) below this many cached events, so a # monitor can catch a silently-broken scraper (e.g. a source site redesign). EVENTS_MIN_HEALTHY = int(os.environ.get("EVENTS_MIN_HEALTHY", "5")) # Client error beacon: clients POST a tiny JSON blob ({msg, where, platform}) to # /api/clienterror when a JS error / unhandled rejection fires, so real-world # breakage is visible in the Space logs. Privacy-first: NOTHING is stored or # forwarded — we only log to stderr, never persist, and carry no identifiers. # A coarse global rate limit stops a looping client from flooding the log. _CLIENT_ERR_MAX_BYTES = 4096 _CLIENT_ERR_WINDOW = 60.0 # seconds _CLIENT_ERR_MAX_PER_WINDOW = 30 _client_err_hits = [] # recent timestamps (trimmed each call) # Per-IP rate limit on /api/*. A token bucket smooths bursts while stopping any # single client from flooding the upstream proxies (Overpass, Nominatim, the # event scrape). Tuned so normal app use never trips it: a bucket of RATE_BURST # tokens, refilled RATE_RPS/sec. Tunable via env on the Space. RATE_RPS = float(os.environ.get("API_RATE_RPS", "5")) # sustained req/s per IP RATE_BURST = float(os.environ.get("API_RATE_BURST", "40")) # burst bucket capacity class _RateLimiter: """Token bucket keyed by client IP. Thread-safe (ThreadingHTTPServer), and self-pruning so a flood of unique IPs can't grow the map without bound.""" def __init__(self, rps, burst): self.rps = rps self.burst = burst self._buckets = {} # ip -> [tokens, last_monotonic] self._lock = threading.Lock() def allow(self, ip): now = time.monotonic() with self._lock: tokens, last = self._buckets.get(ip, (self.burst, now)) tokens = min(self.burst, tokens + (now - last) * self.rps) ok = tokens >= 1.0 self._buckets[ip] = (tokens - 1.0 if ok else tokens, now) if len(self._buckets) > 4096: # prune idle/refilled buckets self._buckets = { k: v for k, v in self._buckets.items() if v[0] < self.burst or (now - v[1]) < 120 } return ok RATE_LIMITER = _RateLimiter(RATE_RPS, RATE_BURST) # External services we proxy. Map each hostname to a stable, low-cardinality # label so the dashboard can chart per-upstream error rate and latency — the # usual root cause of a user-facing failure, otherwise hidden in route latency. _UPSTREAM_HOSTS = { "overpass-api.de": "overpass", "nominatim.openstreetmap.org": "nominatim", "router.project-osrm.org": "osrm", "routing.openstreetmap.de": "osrm", "partyapp.mu": "events_partyapp", "www.partyapp.mu": "events_partyapp", "otayo.com": "events_otayo", "www.otayo.com": "events_otayo", } def _upstream_label(url): try: return _UPSTREAM_HOSTS.get(urlparse(url).hostname or "", "other") except Exception: # noqa: BLE001 return "other" def _instrument_requests(): """Wrap requests.get/post once to time external calls and record per-upstream success/latency in METRICS — without touching the worker modules (they call requests.get/post by attribute, so this is picked up transparently). A timeout/connection error counts as an error and is re-raised unchanged.""" try: import requests as _rq except Exception: # noqa: BLE001 return for name in ("get", "post"): orig = getattr(_rq, name, None) if orig is None or getattr(orig, "_mobaz_wrapped", False): continue def make(orig): def wrapper(url, *a, **kw): dep = _upstream_label(url if isinstance(url, str) else "") t0 = time.perf_counter() ok = False try: resp = orig(url, *a, **kw) ok = getattr(resp, "ok", True) return resp finally: METRICS.observe_upstream(dep, time.perf_counter() - t0, ok) wrapper._mobaz_wrapped = True return wrapper setattr(_rq, name, make(orig)) # --------------------------------------------------------------------------- # Metrics — a tiny, dependency-free Prometheus exposition. Prometheus (running # in the same container) scrapes /metrics; Grafana then renders it. We keep the # label cardinality bounded by mapping each request to a fixed "route" label # (never the raw path/query) so the series count stays small and stable. # --------------------------------------------------------------------------- _LATENCY_BUCKETS = (0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 2.5, 5, 10) _KNOWN_ROUTES = { "/api/plan", "/api/stops", "/api/nearby", "/api/geocode", "/api/walk", "/api/events", "/api/health", "/api/clienterror", "/metrics", "/status", } def _route_label(path): """Collapse a request path to a low-cardinality route label.""" if path in _KNOWN_ROUTES: return path if path.startswith("/api/"): return "/api/other" if path.startswith("/data/"): return "/data" if path in ("/", "", "/index.html"): return "/" return "/static" class _Metrics: def __init__(self): self._lock = threading.Lock() self.start = time.time() self.requests = {} # (route, method, code) -> count self.bucket_counts = [0] * len(_LATENCY_BUCKETS) self.duration_sum = 0.0 self.duration_count = 0 self.client_errors = 0 self.beacons_dropped = 0 # rate-limited client beacons self.inflight = 0 # requests currently being served self.clients = [] # sliding window of (ts, client_ip) self.plan_results = {} # outcome -> count (success/no_route/...) self.upstream = {} # dep -> {ok, err, sum} for external calls def observe(self, route, method, code, dur): with self._lock: key = (route, method, int(code)) self.requests[key] = self.requests.get(key, 0) + 1 self.duration_sum += dur self.duration_count += 1 for i, le in enumerate(_LATENCY_BUCKETS): if dur <= le: self.bucket_counts[i] += 1 def observe_plan(self, outcome): """Count journey-plan outcomes so 'no route found' is visible (a data gap), distinct from a successful plan or a server error.""" with self._lock: self.plan_results[outcome] = self.plan_results.get(outcome, 0) + 1 def observe_upstream(self, dep, dur, ok): """Record one external dependency call (Overpass/Nominatim/OSRM/scrape): count, error count, and total time so the dashboard can show per-upstream error rate and average latency — the usual root cause of user-facing failures, otherwise buried inside route latency.""" with self._lock: s = self.upstream.get(dep) if s is None: s = self.upstream[dep] = {"ok": 0, "err": 0, "sum": 0.0} s["ok" if ok else "err"] += 1 s["sum"] += dur def inc_client_error(self, dropped=False): with self._lock: if dropped: self.beacons_dropped += 1 else: self.client_errors += 1 def inflight_inc(self): with self._lock: self.inflight += 1 def inflight_dec(self): with self._lock: self.inflight = max(0, self.inflight - 1) def record_client(self, ip): now = time.time() with self._lock: self.clients.append((now, ip)) if len(self.clients) > 5000: # hard cap so a flood can't grow it unbounded self.clients = self.clients[-5000:] def active_clients(self, window=300): """Unique client IPs seen in the last `window` seconds (a stateless API has no real 'connections', so this is the meaningful 'active clients').""" cutoff = time.time() - window with self._lock: self.clients = [(t, ip) for (t, ip) in self.clients if t >= cutoff] return len({ip for _, ip in self.clients}) def _system_metrics(self): """Process + host CPU/RAM via psutil (best-effort; empty if unavailable).""" out = {} if psutil is None or _PROC is None: return out try: out["mobaz_process_cpu_percent"] = float(_PROC.cpu_percent(interval=None)) out["mobaz_process_resident_memory_bytes"] = int(_PROC.memory_info().rss) out["mobaz_process_threads"] = int(_PROC.num_threads()) try: out["mobaz_process_open_fds"] = int(_PROC.num_fds()) except Exception: # noqa: BLE001 — not on every platform pass out["mobaz_system_cpu_percent"] = float(psutil.cpu_percent(interval=None)) vm = psutil.virtual_memory() out["mobaz_system_memory_used_percent"] = float(vm.percent) out["mobaz_system_memory_total_bytes"] = int(vm.total) try: out["mobaz_system_load1"] = float(os.getloadavg()[0]) except Exception: # noqa: BLE001 — no getloadavg on some platforms pass except Exception: # noqa: BLE001 pass return out def _gauges(self): """Best-effort point-in-time gauges (never raise into /metrics).""" out = {} try: out["mobaz_stops"] = len(busapi.get_stops()) except Exception: # noqa: BLE001 out["mobaz_stops"] = 0 try: _ev = events_mod.health() out["mobaz_events_cached"] = _ev.get("count", 0) # Seconds since the last successful scrape (-1 = never): a freshness # signal that catches a silently-broken scraper even while it serves # the last good list. _age = _ev.get("age_seconds") out["mobaz_events_age_seconds"] = int(_age) if _age is not None else -1 except Exception: # noqa: BLE001 out["mobaz_events_cached"] = 0 try: out["mobaz_geocode_cache_entries"] = len(geocode._load_coords_cache()) except Exception: # noqa: BLE001 out["mobaz_geocode_cache_entries"] = 0 try: with nearby._cache_lock: out["mobaz_nearby_cache_entries"] = len(nearby._cache) except Exception: # noqa: BLE001 out["mobaz_nearby_cache_entries"] = 0 return out def render(self): """Prometheus text exposition (version 0.0.4).""" with self._lock: requests = dict(self.requests) buckets = list(self.bucket_counts) dur_sum, dur_count = self.duration_sum, self.duration_count client_errors, dropped = self.client_errors, self.beacons_dropped inflight = self.inflight uptime = time.time() - self.start plan_results = dict(self.plan_results) upstream = {k: dict(v) for k, v in self.upstream.items()} L = [] L.append("# HELP mobaz_up Whether the MoBaz backend is serving (always 1).") L.append("# TYPE mobaz_up gauge") L.append("mobaz_up 1") L.append("# HELP mobaz_uptime_seconds Seconds since the backend started.") L.append("# TYPE mobaz_uptime_seconds gauge") L.append("mobaz_uptime_seconds %.0f" % uptime) L.append( "# HELP mobaz_http_requests_total HTTP requests by route, method and code." ) L.append("# TYPE mobaz_http_requests_total counter") for (route, method, code), n in sorted(requests.items()): L.append( 'mobaz_http_requests_total{route="%s",method="%s",code="%d"} %d' % (route, method, code, n) ) L.append( "# HELP mobaz_http_request_duration_seconds Request latency histogram." ) L.append("# TYPE mobaz_http_request_duration_seconds histogram") cumulative = 0 for i, le in enumerate(_LATENCY_BUCKETS): cumulative = buckets[i] L.append( 'mobaz_http_request_duration_seconds_bucket{le="%s"} %d' % (le, cumulative) ) L.append('mobaz_http_request_duration_seconds_bucket{le="+Inf"} %d' % dur_count) L.append("mobaz_http_request_duration_seconds_sum %.6f" % dur_sum) L.append("mobaz_http_request_duration_seconds_count %d" % dur_count) L.append( "# HELP mobaz_client_errors_total Client-side JS errors reported via the beacon." ) L.append("# TYPE mobaz_client_errors_total counter") L.append("mobaz_client_errors_total %d" % client_errors) L.append( "# HELP mobaz_client_error_beacons_dropped_total Beacons dropped by the rate limit." ) L.append("# TYPE mobaz_client_error_beacons_dropped_total counter") L.append("mobaz_client_error_beacons_dropped_total %d" % dropped) L.append("# HELP mobaz_inflight_requests Requests currently being served.") L.append("# TYPE mobaz_inflight_requests gauge") L.append("mobaz_inflight_requests %d" % inflight) L.append( "# HELP mobaz_active_clients Unique client IPs seen in the last 5 minutes." ) L.append("# TYPE mobaz_active_clients gauge") L.append("mobaz_active_clients %d" % self.active_clients()) # Journey-plan outcomes (success / no_route / error) — surfaces data gaps. L.append( "# HELP mobaz_plan_results_total Journey-plan outcomes by result type." ) L.append("# TYPE mobaz_plan_results_total counter") for outcome, n in sorted(plan_results.items()): L.append('mobaz_plan_results_total{result="%s"} %d' % (outcome, n)) # External dependency calls (Overpass/Nominatim/OSRM/scrape): count + time. L.append( "# HELP mobaz_upstream_requests_total External calls by dependency and outcome." ) L.append("# TYPE mobaz_upstream_requests_total counter") for dep, s in sorted(upstream.items()): L.append( 'mobaz_upstream_requests_total{dep="%s",outcome="ok"} %d' % (dep, s["ok"]) ) L.append( 'mobaz_upstream_requests_total{dep="%s",outcome="error"} %d' % (dep, s["err"]) ) L.append( "# HELP mobaz_upstream_request_duration_seconds_sum Total time in external calls." ) L.append("# TYPE mobaz_upstream_request_duration_seconds_sum counter") for dep, s in sorted(upstream.items()): L.append( 'mobaz_upstream_request_duration_seconds_sum{dep="%s"} %.6f' % (dep, s["sum"]) ) # Data/cache gauges + process & host CPU/RAM (psutil). allg = {} allg.update(self._gauges()) allg.update(self._system_metrics()) for name, val in allg.items(): L.append("# TYPE %s gauge" % name) L.append(("%s %d" if isinstance(val, int) else "%s %.4f") % (name, val)) return "\n".join(L) + "\n" METRICS = _Metrics() _instrument_requests() # time external Overpass/Nominatim/OSRM/scrape calls def _status_html(): """A dependency-free server-rendered metrics summary. Grafana (at /) is the rich view; this stays up even if Grafana is down and needs no datasource.""" with METRICS._lock: reqs = dict(METRICS.requests) uptime = time.time() - METRICS.start dur_sum, dur_count = METRICS.duration_sum, METRICS.duration_count client_errors = METRICS.client_errors gauges = METRICS._gauges() total = sum(reqs.values()) avg_ms = (dur_sum / dur_count * 1000) if dur_count else 0.0 active = METRICS.active_clients() inflight = METRICS.inflight # Memory only (reading RSS is side-effect-free; CPU% would reset the delta # that the Prometheus scrape relies on, so leave CPU to Grafana). rss_mb = 0 if _PROC is not None: try: rss_mb = _PROC.memory_info().rss / (1024 * 1024) except Exception: # noqa: BLE001 rss_mb = 0 # Requests grouped by route (sum across method/code). by_route = {} for (route, _m, _c), n in reqs.items(): by_route[route] = by_route.get(route, 0) + n rows = ( "".join( "
Live counters from the running Space. Full dashboards in " "Grafana.
" "raw /metrics · " "/api/health · open the app
" ) CONTENT_TYPES = { ".html": "text/html; charset=utf-8", ".js": "application/javascript; charset=utf-8", ".css": "text/css; charset=utf-8", ".json": "application/json; charset=utf-8", ".svg": "image/svg+xml", ".png": "image/png", ".ico": "image/x-icon", } # --------------------------------------------------------------------------- # # Geocoding sanity: drop mis-resolved stops. # # Many Mauritian places share a name (several "Belle Vue", "St. Martin", ...), # so a stop can resolve to a same-named village on the far side of the island. # Such a point forces a large *detour* relative to going straight from its # previous to its next neighbour, whereas a correctly placed stop sits roughly # on the line between them. We null out the worst detour-causing points until # the path is consistent. This leaves genuinely distant endpoints alone. # --------------------------------------------------------------------------- # DETOUR_KM = 9.0 # interior point forcing more extra distance than this -> drop ENDPOINT_KM = 13.0 # terminal stop jumping more than this to its neighbour -> drop def _km(a, b): return math.hypot((a[0] - b[0]) * 111.0, (a[1] - b[1]) * 105.0) def _despike(stops): """Remove (null) coordinates that are clearly mis-geocoded for this leg.""" # Iteratively drop the single worst off-corridor node. while True: idx = [i for i, s in enumerate(stops) if s.get("coord")] if len(idx) < 3: break # Group consecutive resolved stops at (almost) the same spot into one # node, so a block of identical bad points is handled together. nodes = [] # [[positions_in_idx], coord] for k, i in enumerate(idx): c = stops[i]["coord"] if nodes and _km(nodes[-1][1], c) < 0.4: nodes[-1][0].append(k) else: nodes.append([[k], c]) if len(nodes) < 3: break worst = None for n in range(1, len(nodes) - 1): p, c, q = nodes[n - 1][1], nodes[n][1], nodes[n + 1][1] detour = _km(p, c) + _km(c, q) - _km(p, q) if detour > DETOUR_KM and (worst is None or detour > worst[0]): worst = (detour, n) if not worst: break for k in nodes[worst[1]][0]: stops[idx[k]]["coord"] = None # One-sided endpoint check: a terminal stop jumping implausibly far to its # nearest surviving neighbour is almost certainly wrong. idx = [i for i, s in enumerate(stops) if s.get("coord")] if ( len(idx) >= 2 and _km(stops[idx[0]]["coord"], stops[idx[1]]["coord"]) > ENDPOINT_KM ): stops[idx[0]]["coord"] = None idx = [i for i, s in enumerate(stops) if s.get("coord")] if ( len(idx) >= 2 and _km(stops[idx[-1]]["coord"], stops[idx[-2]]["coord"]) > ENDPOINT_KM ): stops[idx[-1]]["coord"] = None def _attach_coords(result): """Resolve every stop name in the itinerary to [lat, lon] in place.""" names = [] for opt in result.get("options", []): for leg in opt["legs"]: if leg["type"] == "bus": names += [s["name"] for s in leg["stops"]] elif leg.get("to"): names.append(leg["to"]) coords = geocode.resolve_many(names) for opt in result.get("options", []): for leg in opt["legs"]: if leg["type"] == "bus": for s in leg["stops"]: s["coord"] = coords.get(s["name"]) _despike(leg["stops"]) pts = [s["coord"] for s in leg["stops"] if s["coord"]] leg["fromCoord"] = pts[0] if pts else None leg["toCoord"] = pts[-1] if pts else None # Snap to roads so the leg follows streets, not straight hops. leg["geometry"] = roads.snap_route(pts) else: leg["toCoord"] = coords.get(leg.get("to")) return result class Handler(BaseHTTPRequestHandler): server_version = "MauritiusBusPlanner/1.0" def log_message(self, fmt, *args): sys.stderr.write("%s - %s\n" % (self.address_string(), fmt % args)) # --- metrics instrumentation ------------------------------------------- # send_response is the single choke point every reply flows through (JSON, # static file, 204, error), so we record the request count + latency here # exactly once, using the route/start stamped by the dispatcher. def send_response(self, code, message=None): super().send_response(code, message) if not getattr(self, "_recorded", False): self._recorded = True t0 = getattr(self, "_t0", None) if t0 is not None: route = getattr(self, "_route", _route_label(urlparse(self.path).path)) METRICS.observe( route, self.command or "GET", code, time.perf_counter() - t0 ) METRICS.inflight_dec() def _begin(self, path): """Stamp route/timer, count the request as in-flight, and note the client IP (behind nginx the real IP is in X-Forwarded-For).""" self._t0 = time.perf_counter() self._route = _route_label(path) METRICS.inflight_inc() xff = self.headers.get("X-Forwarded-For") ip = ( xff.split(",")[0].strip() if xff else (self.client_address[0] if self.client_address else "?") ) self._client_ip = ip METRICS.record_client(ip) def _cors(self): # The packaged app calls this API from a file:// origin, so allow any. self.send_header("Access-Control-Allow-Origin", "*") self.send_header("Access-Control-Allow-Methods", "GET, OPTIONS") def do_OPTIONS(self): self.send_response(204) self._cors() self.send_header("Content-Length", "0") self.end_headers() def _send_json(self, obj, status=200): body = json.dumps(obj, ensure_ascii=False).encode("utf-8") gzipped = False if self._accepts_gzip() and len(body) > 1024: body = _gzip.compress(body, 6) gzipped = True self.send_response(status) self.send_header("Content-Type", "application/json; charset=utf-8") if gzipped: self.send_header("Content-Encoding", "gzip") self.send_header("Vary", "Accept-Encoding") self.send_header("Content-Length", str(len(body))) self.send_header("Cache-Control", "no-store") self._cors() self.end_headers() self.wfile.write(body) def _etag_matches(self, etag): """True if the client's If-None-Match lists `etag` (so we can 304).""" inm = self.headers.get("If-None-Match") return bool(inm) and etag in [t.strip() for t in inm.split(",")] def _send_304(self, etag): self.send_response(304) self.send_header("ETag", etag) self.send_header("Cache-Control", "no-cache") self._cors() self.end_headers() def _send_rate_limited(self): body = b'{"error":"rate limited, slow down"}' self.send_response(429) self.send_header("Content-Type", "application/json; charset=utf-8") self.send_header("Content-Length", str(len(body))) self.send_header("Retry-After", "2") self._cors() self.end_headers() self.wfile.write(body) def _send_json_cacheable(self, obj): """Like _send_json but with an ETag so an unchanged body costs the client only a 304 (no re-download). Used for /api/stops, which the apps fetch repeatedly but which rarely changes. `no-cache` = always revalidate.""" body = json.dumps(obj, ensure_ascii=False).encode("utf-8") etag = '"%s"' % hashlib.md5(body).hexdigest()[:16] if self._etag_matches(etag): return self._send_304(etag) gzipped = False if self._accepts_gzip() and len(body) > 1024: body = _gzip.compress(body, 6) gzipped = True self.send_response(200) self.send_header("Content-Type", "application/json; charset=utf-8") if gzipped: self.send_header("Content-Encoding", "gzip") self.send_header("Vary", "Accept-Encoding") self.send_header("Content-Length", str(len(body))) self.send_header("ETag", etag) self.send_header("Cache-Control", "no-cache") self._cors() self.end_headers() self.wfile.write(body) def _send_text(self, text, ctype="text/plain; charset=utf-8", status=200): body = text.encode("utf-8") gzipped = False if self._accepts_gzip() and len(body) > 1024: body = _gzip.compress(body, 6) gzipped = True self.send_response(status) self.send_header("Content-Type", ctype) if gzipped: self.send_header("Content-Encoding", "gzip") self.send_header("Vary", "Accept-Encoding") self.send_header("Content-Length", str(len(body))) self.send_header("Cache-Control", "no-store") self._cors() self.end_headers() self.wfile.write(body) def _accepts_gzip(self): return "gzip" in (self.headers.get("Accept-Encoding") or "").lower() def _send_file(self, path): ext = os.path.splitext(path)[1].lower() # ETag from mtime+size lets the client (and the Android data refresh) # revalidate cheaply: an unchanged file returns 304 with no body. This # is the big win for /data/network.json (~2.3 MB) on every app launch. st = os.stat(path) etag = '"%x-%x"' % (int(st.st_mtime), st.st_size) if self._etag_matches(etag): return self._send_304(etag) with open(path, "rb") as fh: body = fh.read() gzipped = False if ext in GZIP_EXTS and self._accepts_gzip() and len(body) > 1024: body = _gzip.compress(body, 6) gzipped = True self.send_response(200) self.send_header( "Content-Type", CONTENT_TYPES.get(ext, "application/octet-stream") ) if gzipped: self.send_header("Content-Encoding", "gzip") self.send_header("Vary", "Accept-Encoding") self.send_header("Content-Length", str(len(body))) self.send_header("ETag", etag) self.send_header("Last-Modified", formatdate(st.st_mtime, usegmt=True)) self.send_header("Cache-Control", "no-cache") self._cors() self.end_headers() self.wfile.write(body) def do_GET(self): parsed = urlparse(self.path) path = parsed.path self._begin(path) # Per-IP rate limit on the API (health is exempt — it's the monitor's # probe and must always answer). Returns 429 + Retry-After; the 429 is # captured in mobaz_http_requests_total{code="429"} via send_response. if path.startswith("/api/") and path != "/api/health": if not RATE_LIMITER.allow(getattr(self, "_client_ip", "?")): return self._send_rate_limited() try: if path == "/metrics": return self._send_text(METRICS.render(), "text/plain; version=0.0.4") if path == "/status": return self._send_text(_status_html(), "text/html; charset=utf-8") if path == "/api/stops": return self._send_json_cacheable(busapi.get_stops()) if path == "/api/plan": q = parse_qs(parsed.query) day = int((q.get("day") or ["0"])[0]) frm = (q.get("from") or [""])[0] to = (q.get("to") or [""])[0] if not frm or not to: return self._send_json({"error": "from and to are required"}, 400) if frm == to: METRICS.observe_plan("same_stop") return self._send_json( {"error": "Origin and destination are the same."}, 400 ) result = busapi.plan(day, frm, to) _attach_coords(result) # A plan that finds no itinerary is a data gap, not an error — # count it separately so the dashboard surfaces routable demand. METRICS.observe_plan("success" if result.get("options") else "no_route") return self._send_json(result) if path == "/api/walk": # Pedestrian route from the user's location to their boarding # stop. ?from=lat,lon&to=lat,lon -> {geometry, distance, duration}. q = parse_qs(parsed.query) try: fa = [float(x) for x in (q.get("from") or [""])[0].split(",")] ta = [float(x) for x in (q.get("to") or [""])[0].split(",")] except ValueError: fa = ta = [] if len(fa) != 2 or len(ta) != 2: return self._send_json( {"error": "from and to must be 'lat,lon'"}, 400 ) route = roads.walk_route(fa, ta) if not route: return self._send_json({"error": "no walking route"}, 502) return self._send_json(route) if path == "/api/geocode": # Free-text destination address -> OSM-corrected candidates. q = (parse_qs(parsed.query).get("q") or [""])[0] if not q.strip(): return self._send_json({"results": []}) return self._send_json({"results": geocode.geocode_search(q)}) if path == "/api/nearby": # Nearby activities/attractions around ?lat=&lon= (proxies OSM # Overpass — which has no CORS — and attaches photos). Returns # [{name, coord, cat, emoji, dist, photo}]. q = parse_qs(parsed.query) try: lat = float((q.get("lat") or [""])[0]) lon = float((q.get("lon") or [""])[0]) except ValueError: return self._send_json({"error": "lat and lon are required"}, 400) radius = int((q.get("radius") or ["7000"])[0]) cat = (q.get("cat") or [""])[0] if cat and cat in nearby.CATEGORIES: results = nearby.nearby_category(lat, lon, radius, cat) else: # "top"/unknown → rich attractions feed results = nearby.nearby_activities(lat, lon, radius) return self._send_json({"results": results}) if path == "/api/events": # Upcoming Mauritius events scraped from partyapp.mu (public site). return self._send_json({"events": events_mod.get_events()}) if path == "/api/health": # Machine-readable status for monitoring. Returns 200 when the # core data is present, 503 when degraded (no stops, or the # event scrape has produced nothing) so an external check can # alert. Never triggers a blocking scrape. ev = events_mod.health() try: stops = len(busapi.get_stops()) except Exception: # noqa: BLE001 stops = 0 try: geo_cache = len(geocode._load_coords_cache()) except Exception: # noqa: BLE001 geo_cache = 0 degraded = stops == 0 or ev["count"] < EVENTS_MIN_HEALTHY body = { "status": "degraded" if degraded else "ok", "stops": stops, "events": ev, "geocode_cache": geo_cache, } return self._send_json(body, 503 if degraded else 200) # Static files. rel = path.lstrip("/") or "index.html" full = os.path.normpath(os.path.join(PUBLIC, rel)) if not full.startswith(PUBLIC): return self._send_json({"error": "forbidden"}, 403) if os.path.isdir(full): full = os.path.join(full, "index.html") if os.path.isfile(full): return self._send_file(full) return self._send_json({"error": "not found"}, 404) except Exception as exc: # noqa: BLE001 # Log with the request path + a timestamp so a 500 in the wild is # traceable in the Space logs (which is the only visibility we have). ts = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()) print(f"[{ts}] ERROR {self.path}: {type(exc).__name__}: {exc}", flush=True) traceback.print_exc() return self._send_json({"error": f"{type(exc).__name__}: {exc}"}, 500) def do_POST(self): parsed = urlparse(self.path) self._begin(parsed.path) if parsed.path != "/api/clienterror": return self._send_json({"error": "not found"}, 404) # Read at most a few KB; oversized or malformed bodies are dropped quietly. try: n = min(int(self.headers.get("Content-Length") or 0), _CLIENT_ERR_MAX_BYTES) raw = self.rfile.read(n) if n > 0 else b"" payload = json.loads(raw.decode("utf-8")) if raw else {} except Exception: # noqa: BLE001 payload = {} # Coarse global rate limit so a looping client can't flood the logs. now = time.time() _client_err_hits[:] = [ t for t in _client_err_hits if now - t < _CLIENT_ERR_WINDOW ] if len(_client_err_hits) < _CLIENT_ERR_MAX_PER_WINDOW: _client_err_hits.append(now) msg = str(payload.get("msg", ""))[:300] where = str(payload.get("where", ""))[:120] platform = str(payload.get("platform", ""))[:60] ts = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()) # stderr only — never persisted, no client identity captured. print(f"[{ts}] CLIENT-ERR [{platform}] {where}: {msg}", flush=True) METRICS.inc_client_error() else: METRICS.inc_client_error(dropped=True) # Always 204: the beacon is fire-and-forget and must never surface to users. self.send_response(204) self._cors() self.send_header("Content-Length", "0") self.end_headers() def main(): # Warm the caches so the first request is fast and failures surface early. print("Loading bus stops...", flush=True) stops = busapi.get_stops() print(f" {len(stops)} stops cached.", flush=True) print("Building OpenStreetMap geocoding index...", flush=True) try: idx = geocode._build_osm_index() print(f" {len(idx)} named features indexed.", flush=True) except Exception as exc: # noqa: BLE001 print( f" WARN: OSM index unavailable ({exc}); will rely on Nominatim.", flush=True, ) # Warm the events cache in the background so the Events tab is instant on the # first request without delaying server start-up. print("Pre-warming events (background)...", flush=True) events_mod.prewarm() try: httpd = ThreadingHTTPServer(("0.0.0.0", PORT), Handler) except OSError as exc: if exc.errno in (48, 98): # EADDRINUSE (macOS / Linux) print( f"\nPort {PORT} is already in use — another copy may still be running.", file=sys.stderr, ) print(f" Free it: lsof -ti tcp:{PORT} | xargs kill", file=sys.stderr) print( f" Or pick another port: PORT={PORT + 1} ./run.sh", file=sys.stderr ) sys.exit(1) raise print( f"\nMauritius Bus Route Planner running at http://localhost:{PORT}", flush=True ) print("Press Ctrl+C to stop.", flush=True) try: httpd.serve_forever() except KeyboardInterrupt: print("\nShutting down.") httpd.server_close() if __name__ == "__main__": main()