"""Local web server for the Fjordling simulator, with real ships from AIS (user, 5 Nov 2026). It serves this folder (the simulator page and its tiles), as `python -m http.server` did. It also reads the Norwegian Coastal Administration's (Kystverket) open AIS stream and answers the page's question "which ships are near me?". Page and ships come from the same origin (localhost), so the browser needs no CORS and no key. Source: Kystverket open AIS, TCP 153.44.253.27:5631, raw NMEA AIVDM (IEC 62320-1). Norwegian Licence for Open Government Data (NLOD 2.0), no registration. It leaves out fishing vessels under 15 m and leisure craft under 45 m. Attribution: "Contains data under the Norwegian licence for Open Government data (NLOD) distributed by the Norwegian Coastal Administration". Endpoints: GET /ais/status -> {"ok", "connected", "ships", "msgs", "since", "source", "licence"} GET /ais/near?E=..&N=..&r=15000 -> {"t": now (s), "ships": [ ... ]} (UTM33 / EPSG:25833, metres) each ship: mmsi, E, N, sog (kn), cog (deg true or null), hdg (deg true or null), t (report time, epoch s), name, type (AIS ship type), L, B (m), A, C (antenna from bow / from port, m), draught (m) Usage: python ais_server.py [port] [--bind 127.0.0.1] Standard library only (it runs on the user's own Python). """ import http.server, json, math, os, socket, socketserver, sys, threading, time, urllib.parse HOST, PORT_AIS = "153.44.253.27", 5631 KEEP_S = 600 # forget a ship 10 min after its last position report (chosen) STATIC_KEEP_S = 6 * 3600 # ---- lat/lon (ETRS89 ~ WGS84 at this accuracy) -> UTM zone 33 (EPSG:25833), Krueger series (n^4) ---- _a, _f = 6378137.0, 1 / 298.257222101 _n = _f / (2 - _f) _A = _a / (1 + _n) * (1 + _n ** 2 / 4 + _n ** 4 / 64) _al = (_n / 2 - 2 * _n ** 2 / 3 + 5 * _n ** 3 / 16 + 41 * _n ** 4 / 180, 13 * _n ** 2 / 48 - 3 * _n ** 3 / 5 + 557 * _n ** 4 / 1440, 61 * _n ** 3 / 240 - 103 * _n ** 4 / 140, 49561 * _n ** 4 / 161280) _e = math.sqrt(_f * (2 - _f)) def utm33(lat, lon): k0, lon0, FE = 0.9996, math.radians(15.0), 500000.0 phi, lam = math.radians(lat), math.radians(lon) - lon0 t = math.sinh(math.atanh(math.sin(phi)) - _e * math.atanh(_e * math.sin(phi))) xi, eta = math.atan2(t, math.cos(lam)), math.atanh(math.sin(lam) / math.sqrt(1 + t * t)) x, y = eta, xi for j, a in enumerate(_al, 1): x += a * math.cos(2 * j * xi) * math.sinh(2 * j * eta) y += a * math.sin(2 * j * xi) * math.cosh(2 * j * eta) return FE + k0 * _A * x, k0 * _A * y # ---- minimal AIVDM decoder: position 1/2/3/18/19, static 5/24 ---- def _bits(payload, fill): b = "".join(format((ord(c) - 48) - (8 if ord(c) - 48 > 40 else 0), "06b") for c in payload) return b[:len(b) - fill] if fill else b def _u(b, s, n): return int(b[s:s + n], 2) if len(b) >= s + n else None def _i(b, s, n): v = _u(b, s, n) if v is None: return None return v - (1 << n) if v >= 1 << (n - 1) else v def _txt(b, s, n): o = "" for k in range(s, s + n, 6): v = _u(b, k, 6) if v is None: break o += chr(v + 64 if v < 32 else v) return o.replace("@", "").strip() LOCK = threading.Lock() POS, STATIC = {}, {} # mmsi -> dict INFO = {"connected": False, "msgs": 0, "since": None, "error": None} _frag = {} def _pos(mmsi, lon, lat, sog, cog, hdg): if lon is None or lat is None or abs(lat) > 90 or abs(lon) > 180 or (lat == 91 or lon == 181): return if not (55 < lat < 75 and -10 < lon < 40): return # Norwegian waters only E, N = utm33(lat, lon) sog = None if sog is None or sog >= 102.2 else sog cog = None if cog is None or cog >= 360 else cog hdg = None if hdg is None or hdg >= 360 else hdg with LOCK: POS[mmsi] = {"E": round(E, 1), "N": round(N, 1), "sog": sog, "cog": cog, "hdg": hdg, "t": round(time.time(), 1)} def handle_line(s): k = s.find("!") if k < 0: return f = s[k:].split("*")[0].split(",") if len(f) < 7 or not f[0].endswith("VDM"): return try: tot, num, fill = int(f[1]), int(f[2]), int(f[6] or 0) except ValueError: return sid, pay = f[3], f[5] if tot > 1: key = (sid, f[4]) d = _frag.setdefault(key, {"t": time.time()}); d[num] = (pay, fill) if len(d) - 1 < tot: return _frag.pop(key) try: pay = "".join(d[n][0] for n in range(1, tot + 1)); fill = d[tot][1] except KeyError: return b = _bits(pay, fill); t = _u(b, 0, 6); mmsi = _u(b, 8, 30) if t is None or mmsi is None: return INFO["msgs"] += 1 if t in (1, 2, 3): _pos(mmsi, (_i(b, 61, 28) or 0) / 600000, (_i(b, 89, 27) or 0) / 600000, (_u(b, 50, 10) or 0) / 10, (_u(b, 116, 12) or 3600) / 10, _u(b, 128, 9)) elif t in (18, 19): _pos(mmsi, (_i(b, 57, 28) or 0) / 600000, (_i(b, 85, 27) or 0) / 600000, (_u(b, 46, 10) or 0) / 10, (_u(b, 112, 12) or 3600) / 10, _u(b, 124, 9)) if t == 19: st = {"name": _txt(b, 143, 120), "type": _u(b, 263, 8), "A": _u(b, 271, 9), "B": _u(b, 280, 9), "C": _u(b, 289, 6), "D": _u(b, 295, 6), "ts": time.time()} with LOCK: STATIC.setdefault(mmsi, {}).update(st) elif t == 5: st = {"name": _txt(b, 112, 120), "type": _u(b, 232, 8), "A": _u(b, 240, 9), "B": _u(b, 249, 9), "C": _u(b, 258, 6), "D": _u(b, 264, 6), "draught": (_u(b, 294, 8) or 0) / 10, "call": _txt(b, 70, 42), "dest": _txt(b, 302, 120), "imo": _u(b, 40, 30) or 0, "ts": time.time()} with LOCK: STATIC.setdefault(mmsi, {}).update(st) elif t == 24: part = _u(b, 38, 2) with LOCK: st = STATIC.setdefault(mmsi, {}); st["ts"] = time.time() if part == 0: st["name"] = _txt(b, 40, 120) elif part == 1: st.update({"type": _u(b, 40, 8), "A": _u(b, 132, 9), "B": _u(b, 141, 9), "C": _u(b, 150, 6), "D": _u(b, 156, 6)}) def reader(): back = 5 while True: try: sock = socket.create_connection((HOST, PORT_AIS), timeout=20); sock.settimeout(60) INFO.update(connected=True, error=None, since=INFO["since"] or time.time()); back = 5 print("AIS: connected to Kystverket's open stream", flush=True) buf = b"" while True: chunk = sock.recv(65536) if not chunk: raise ConnectionError("stream closed") buf += chunk *lines, buf = buf.split(b"\n") for ln in lines: try: handle_line(ln.decode("ascii", "ignore").strip()) except Exception: pass except Exception as e: INFO.update(connected=False, error=str(e)) print("AIS: %s - retrying in %d s" % (e, back), flush=True) time.sleep(back); back = min(120, back * 2) def janitor(): while True: time.sleep(30); now = time.time() with LOCK: for m in [m for m, p in POS.items() if now - p["t"] > KEEP_S]: del POS[m] for m in [m for m, s in STATIC.items() if now - s.get("ts", 0) > STATIC_KEEP_S and m not in POS]: del STATIC[m] for k in [k for k, d in _frag.items() if now - d["t"] > 30]: _frag.pop(k, None) def near(E, N, r): out = [] with LOCK: for m, p in POS.items(): if (p["E"] - E) ** 2 + (p["N"] - N) ** 2 > r * r: continue s = STATIC.get(m, {}) A, B, C, D = (s.get(k) or 0 for k in "ABCD") out.append(dict(p, mmsi=m, name=s.get("name") or "", type=s.get("type") or 0, L=A + B, B=C + D, A=A, C=C, draught=s.get("draught") or 0, call=s.get("call") or "", dest=s.get("dest") or "", imo=s.get("imo") or 0)) return out PUBLIC = False # set by --public RATE = {} # client address -> [window start, count]: at most 120 /ais requests a minute each def _rate_ok(addr): now = time.time(); w = RATE.get(addr) if not w or now - w[0] > 60: RATE[addr] = [now, 1]; return True w[1] += 1; return w[1] <= 120 class Handler(http.server.SimpleHTTPRequestHandler): def log_message(self, fmt, *a): if not self.path.startswith("/ais/"): super().log_message(fmt, *a) # /ais/* may be read from a page served elsewhere (another port, the PC's own browser); read only, no cookies. # The Private Network Access header lets Chrome allow it from a public https page to this local address. def _cors(self): self.send_header("Access-Control-Allow-Origin", "*") self.send_header("Access-Control-Allow-Methods", "GET") self.send_header("Access-Control-Allow-Private-Network", "true") def do_OPTIONS(self): if not self.path.startswith("/ais/"): self.send_error(405); return self.send_response(204); self._cors(); self.send_header("Access-Control-Max-Age", "600"); self.end_headers() def _json(self, obj): b = json.dumps(obj, separators=(",", ":")).encode() self.send_response(200); self.send_header("Content-Type", "application/json"); self._cors() self.send_header("Cache-Control", "no-store"); self.send_header("Content-Length", str(len(b))) self.end_headers(); self.wfile.write(b) def _client(self): # behind a local tunnel/proxy (Tailscale Funnel, Cloudflare Tunnel, nginx) every request arrives from 127.0.0.1: # then the visitor is the first address the proxy forwards, so the rate limit stays per visitor a = self.client_address[0] if a in ("127.0.0.1", "::1"): f = self.headers.get("CF-Connecting-IP") or (self.headers.get("X-Forwarded-For") or "").split(",")[0].strip() if f: return f return a def do_GET(self): u = urllib.parse.urlparse(self.path) if PUBLIC and not u.path.startswith("/ais/"): # --public: the relay only, none of this PC's files self.send_error(404); return if u.path.startswith("/ais/") and not _rate_ok(self._client()): self.send_response(429); self._cors(); self.end_headers(); return if u.path == "/ais/status": with LOCK: n = len(POS) return self._json(dict(ok=True, ships=n, source="Kystverket open AIS (NLOD)", licence="Contains data under the Norwegian licence for Open Government data (NLOD) distributed by the Norwegian Coastal Administration", **INFO)) if u.path == "/ais/near": q = urllib.parse.parse_qs(u.query) try: E, N, r = float(q["E"][0]), float(q["N"][0]), min(60000.0, float(q.get("r", ["15000"])[0])) except (KeyError, ValueError): self.send_error(400, "E, N needed"); return return self._json({"t": round(time.time(), 2), "ships": near(E, N, r)}) return super().do_GET() class Server(socketserver.ThreadingMixIn, http.server.HTTPServer): daemon_threads = True allow_reuse_address = True if __name__ == "__main__": port = next((int(a) for a in sys.argv[1:] if a.isdigit()), 8790) lan = "--lan" in sys.argv # also reachable from other devices on this network (phone, another PC) PUBLIC = "--public" in sys.argv # for a tunnel to the internet: answer /ais/ only (see mobile/ais-relay-hjemme.md) bind = sys.argv[sys.argv.index("--bind") + 1] if "--bind" in sys.argv else ("0.0.0.0" if lan else "127.0.0.1") os.chdir(os.path.dirname(os.path.abspath(__file__))) threading.Thread(target=reader, daemon=True).start() threading.Thread(target=janitor, daemon=True).start() with Server((bind, port), Handler) as srv: print("Serving %s at http://localhost:%d/ (ships from AIS at /ais/near)" % (os.getcwd(), port), flush=True) if bind == "0.0.0.0": ips = set() try: for info in socket.getaddrinfo(socket.gethostname(), None, socket.AF_INET): ips.add(info[4][0]) except OSError: pass try: s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM); s.connect(("10.255.255.255", 1)); ips.add(s.getsockname()[0]); s.close() except OSError: pass print("Other devices on this network can open (the phone page; VR needs https, so not the Quest this way):", flush=True) for ip in sorted(i for i in ips if not i.startswith("127.")): print(" http://%s:%d/fjordling-sim.html" % (ip, port), flush=True) print("Windows may ask to allow Python through the firewall: allow it for this network only.", flush=True) try: srv.serve_forever() except KeyboardInterrupt: pass