diff --git a/app/live_layers.py b/app/live_layers.py index e5b9f0e..f3e15eb 100644 --- a/app/live_layers.py +++ b/app/live_layers.py @@ -309,6 +309,204 @@ def _heading(value: object) -> float | None: return num +def _s(value: object) -> str | None: + if value is None: + return None + text = str(value).strip() + return text or None + + +# ADS-B emitter category (DO-260B). A3 airliners, A5 heavies, A7 helicopters. +_EMITTER = { + "A0": "unknown", "A1": "light", "A2": "small", "A3": "large", + "A4": "high vortex", "A5": "heavy", "A6": "high performance", "A7": "rotorcraft", + "B0": "unknown", "B1": "glider", "B2": "airship", "B3": "parachute", + "B4": "ultralight", "B6": "UAV", "B7": "space", + "C0": "ground unknown", "C1": "emergency vehicle", "C2": "service vehicle", + "D0": "unknown", "D1": "emergency", +} + +# Combat / dedicated-military ICAO types. C-130/C-17 omitted — those also fly +# as fire tankers and civil contractors; dbFlags/hex catch the real mil ones. +_MIL_ICAO = frozenset({ + "F15", "F16", "F18", "FA18", "F22", "F35", "F117", "A10", "A10A", + "B1", "B1B", "B2", "B52", "AV8B", "F4", "F5", "F14", + "SU27", "SU30", "SU34", "SU35", "SU57", + "MG29", "MIG29", "MG31", "MIG31", "J10", "J11", "J15", "J16", "J20", + "EUFI", "RFAL", "TOR", "E3TF", "E3CF", "E6", "E8", "P8", + "MQ9", "MQ1", "RQ4", "V22", "AH64", "H64", +}) + +# US DoD Mode-S block AE0000–AEFFFF. +_US_DOD_HEX_LO, _US_DOD_HEX_HI = 0xAE0000, 0xAEFFFF + +_MIL_CS_PREFIX = ("RCH", "NAVY", "ARMY", "MARINE", "GOTOF", "REACH") + + +def classify_adsb(ac: dict) -> tuple[str, str]: + """Return (role, role_src). Prefer readsb dbFlags bit0, then type/hex/cs.""" + flags = ac.get("dbFlags") + try: + flags_i = int(flags) if flags is not None else 0 + except (TypeError, ValueError): + flags_i = 0 + if flags_i & 1: + return "military", "dbFlags" + icao = str(ac.get("t") or "").strip().upper() + if icao in _MIL_ICAO: + return "military", "type" + try: + hx = int(str(ac.get("hex") or "").strip(), 16) + except ValueError: + hx = -1 + if _US_DOD_HEX_LO <= hx <= _US_DOD_HEX_HI: + return "military", "hex" + cs = str(ac.get("flight") or "").strip().upper() + if cs.startswith(_MIL_CS_PREFIX): + return "military", "callsign" + return "civilian", "default" + + +def _adsb_extra(ac: dict, hex_id: str) -> dict[str, Any]: + role, src = classify_adsb(ac) + cat = str(ac.get("category") or "").strip().upper() + extra: dict[str, Any] = { + "hex": hex_id, + "reg": _s(ac.get("r")), + "type": _s(ac.get("t")), + "alt_baro": ac.get("alt_baro"), + "squawk": _s(ac.get("squawk")), + "emergency": _s(ac.get("emergency")), + "category": cat or None, + "emitter": _EMITTER.get(cat), + "seen_pos": ac.get("seen_pos"), + "role": role, + "role_src": src, + "src": "adsb.lol", + } + desc = _s(ac.get("desc")) + if desc: + extra["desc"] = desc + own = _s(ac.get("ownOp") or ac.get("ownOpName") or ac.get("ownop")) + if own: + extra["ownOp"] = own + if ac.get("alt_geom") is not None: + extra["alt_geom"] = ac.get("alt_geom") + vs = ac.get("baro_rate") + if vs is None: + vs = ac.get("geom_rate") + if vs is not None: + extra["vs"] = vs + try: + raw_flags = ac.get("dbFlags") + flags_i = int(raw_flags) if raw_flags is not None else 0 + except (TypeError, ValueError): + flags_i = 0 + if flags_i: + extra["dbFlags"] = flags_i + extra["interesting"] = bool(flags_i & 2) + return extra + + +_NAVSTAT = { + 0: "underway", + 1: "at anchor", + 2: "not under command", + 3: "restricted manoeuvre", + 4: "constrained by draught", + 5: "moored", + 6: "aground", + 7: "fishing", + 8: "sailing", + 14: "AIS-SART", + 15: "undefined", +} + +# Compact MID → country for the flags that actually show up on AIS. +_MID_COUNTRY = { + 211: "Germany", 218: "Germany", + 219: "Denmark", 220: "Denmark", + 224: "Spain", 225: "Spain", + 226: "France", 227: "France", 228: "France", + 232: "United Kingdom", 233: "United Kingdom", 234: "United Kingdom", 235: "United Kingdom", + 236: "Gibraltar", 237: "Greece", 239: "Greece", 240: "Greece", 241: "Greece", + 244: "Netherlands", 245: "Netherlands", 246: "Netherlands", + 247: "Italy", 249: "Malta", 250: "Ireland", 251: "Iceland", + 255: "Portugal", 256: "Malta", + 257: "Norway", 258: "Norway", 259: "Norway", + 261: "Poland", 263: "Portugal", 265: "Sweden", 266: "Sweden", + 271: "Turkey", 273: "Russia", 276: "Estonia", 277: "Lithuania", + 301: "Anguilla", 303: "United States", 310: "Bermuda", 316: "Canada", + 319: "Cayman Islands", 338: "United States", 339: "Jamaica", + 345: "Mexico", 352: "Panama", 353: "Panama", 354: "Panama", + 355: "Panama", 356: "Panama", 357: "Panama", + 366: "United States", 367: "United States", 368: "United States", 369: "United States", + 370: "Panama", 371: "Panama", 372: "Panama", 373: "Panama", 374: "Panama", + 375: "St Vincent", 376: "St Vincent", 377: "St Vincent", + 412: "China", 413: "China", 414: "China", 416: "Taiwan", + 419: "India", 431: "Japan", 432: "Japan", 440: "South Korea", 441: "South Korea", + 477: "Hong Kong", 503: "Australia", 525: "Indonesia", 533: "Malaysia", + 538: "Marshall Islands", 548: "Philippines", 563: "Singapore", + 564: "Singapore", 565: "Singapore", 566: "Singapore", 567: "Thailand", + 574: "Vietnam", 636: "Liberia", 637: "Liberia", + 710: "Brazil", 725: "Chile", 730: "Colombia", 760: "Peru", +} + + +def _mmsi_country(mmsi: object) -> str | None: + digits = "".join(ch for ch in str(mmsi or "") if ch.isdigit()) + if len(digits) < 3: + return None + try: + mid = int(digits[:3]) + except ValueError: + return None + return _MID_COUNTRY.get(mid) + + +def classify_ais_type(type_code: int | None) -> tuple[str, str]: + """Return (role, kind) from ITU-R M.1371 ship-and-cargo type.""" + if type_code is None: + return "civilian", "unknown" + t = int(type_code) + tens = t // 10 + if t == 35: + return "military", "military" + if t == 30: + return "civilian", "fishing" + if t in (31, 32, 52): + return "civilian", "tug" + if t == 33: + return "civilian", "dredger" + if t == 34: + return "civilian", "diving" + if t == 36: + return "civilian", "sailing" + if t == 37: + return "civilian", "pleasure" + if t == 50: + return "government", "pilot" + if t == 51: + return "government", "SAR" + if t == 55: + return "government", "law" + if t == 54: + return "government", "anti-pollution" + if t == 58: + return "government", "medical" + if tens == 4: + return "civilian", "HSC" + if tens == 6: + return "civilian", "passenger" + if tens == 7: + return "civilian", "cargo" + if tens == 8: + return "civilian", "tanker" + if tens in (5, 9) or t in (53, 56, 57, 59): + return "civilian", "special" + return "civilian", "other" + + def transform_adsb_lol(payload: dict | list | None) -> list[dict]: """Map ADSB.lol v2 aircraft list to shared markers. Dedup on hex.""" if payload is None: @@ -333,17 +531,7 @@ def transform_adsb_lol(payload: dict | list | None) -> list[dict]: heading=_heading(ac.get("track")), speed=_f(ac.get("gs")), label=flight, - extra={ - "hex": hex_id, - "reg": ac.get("r"), - "type": ac.get("t"), - "alt_baro": ac.get("alt_baro"), - "squawk": ac.get("squawk"), - "emergency": ac.get("emergency"), - "category": ac.get("category"), - "seen_pos": ac.get("seen_pos"), - "src": "adsb.lol", - }, + extra=_adsb_extra(ac, hex_id), )) return out @@ -405,6 +593,9 @@ def transform_ais_frame(frame: dict | None) -> dict | None: or {} ) extra: dict[str, Any] = {"src": "aisstream", "mmsi": mmsi} + country = _mmsi_country(mmsi) + if country: + extra["country"] = country if frame.get("MessageType") == "ShipStaticData": static = msg.get("ShipStaticData") or {} dest = str(static.get("Destination") or static.get("destination") or "").strip() @@ -412,6 +603,42 @@ def transform_ais_frame(frame: dict | None) -> dict | None: extra["static"] = True if not name: name = str(static.get("Name") or static.get("name") or "").strip() + cs = _s(static.get("CallSign") or static.get("callSign")) + if cs: + extra["callsign"] = cs + try: + imo = int(static.get("ImoNumber") or static.get("imoNumber") or 0) + except (TypeError, ValueError): + imo = 0 + if imo: + extra["imo"] = imo + type_code = static.get("Type") if "Type" in static else static.get("type") + try: + type_i = int(type_code) if type_code is not None else None + except (TypeError, ValueError): + type_i = None + if type_i is not None: + extra["type_code"] = type_i + role, kind = classify_ais_type(type_i) + extra["role"] = role + extra["kind"] = kind + dim = static.get("Dimension") or static.get("dimension") or {} + if isinstance(dim, dict): + a, b = _f(dim.get("A")), _f(dim.get("B")) + c, d = _f(dim.get("C")), _f(dim.get("D")) + if a is not None and b is not None: + extra["length"] = int(round(a + b)) + if c is not None and d is not None: + extra["beam"] = int(round(c + d)) + draught = _f(static.get("MaximumStaticDraught") or static.get("maximumStaticDraught")) + if draught is not None: + extra["draught"] = draught + eta = static.get("Eta") or static.get("eta") or {} + if isinstance(eta, dict) and eta.get("Month"): + extra["eta"] = ( + f"{int(eta.get('Month') or 0):02d}-{int(eta.get('Day') or 0):02d} " + f"{int(eta.get('Hour') or 0):02d}:{int(eta.get('Minute') or 0):02d}" + ) if lat is None or lon is None: # Static-only update: caller merges onto last-known by MMSI. return to_marker(str(mmsi), None, None, label=name or str(mmsi), extra=extra) @@ -421,8 +648,13 @@ def transform_ais_frame(frame: dict | None) -> dict | None: sog = pos.get("Sog") navstat = pos.get("NavigationalStatus") extra["navstat"] = navstat + try: + extra["nav"] = _NAVSTAT.get(int(navstat)) if navstat is not None else None + except (TypeError, ValueError): + extra["nav"] = None extra["cog"] = pos.get("Cog") - extra["dest"] = extra.get("dest") + if not extra.get("dest"): + extra.pop("dest", None) if lat is None or lon is None: return None return to_marker( diff --git a/app/static/index.html b/app/static/index.html index 25d35d5..0bb8963 100644 --- a/app/static/index.html +++ b/app/static/index.html @@ -356,6 +356,16 @@ .cam-pop td { padding: 0.12rem 0.2rem; vertical-align: top; } .cam-pop td.k { color: var(--muted); text-transform: uppercase; font-size: 0.6rem; letter-spacing: 0.05em; white-space: nowrap; width: 34%; font-family: 'Share Tech Mono', monospace; } .cam-pop a { color: var(--cyan); word-break: break-all; } + .role-badge { + display: inline-block; font-family: 'Share Tech Mono', monospace; + font-size: 0.58rem; font-weight: 700; letter-spacing: 0.12em; + padding: 0.08rem 0.4rem; border-radius: 3px; margin-left: 0.4rem; + vertical-align: middle; + } + .role-badge.military { color: #f472b6; border: 1px solid #f472b6; background: rgba(244,114,182,0.12); } + .role-badge.civilian { color: #7dd3fc; border: 1px solid #38bdf8; background: rgba(56,189,248,0.1); } + .role-badge.government { color: #facc15; border: 1px solid #facc15; background: rgba(250,204,21,0.12); } + .role-badge.firefighter { color: #fb923c; border: 1px solid #fb923c; background: rgba(251,146,60,0.12); } .blip-pop { min-width: 200px; max-width: 260px; } .blip-pop .blip-src { font-family: 'Share Tech Mono', monospace; font-size: 0.62rem; color: var(--cyan); text-transform: uppercase; letter-spacing: 0.08em; } .blip-pop .blip-time { font-size: 0.7rem; color: var(--muted); font-family: 'Share Tech Mono', monospace; margin: 0.2rem 0 0.3rem; } @@ -771,6 +781,7 @@ 0 +