feat: classify aircraft/vessels as military vs civilian in popups #12

Merged
sirius merged 1 commit from feat/ac-vessel-intel into master 2026-08-28 20:58:51 -04:00
3 changed files with 435 additions and 25 deletions

View file

@ -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 AE0000AEFFFF.
_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(

View file

@ -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 @@
<label class="lp-name"><input type="checkbox" id="lp-ac-on" checked onchange="toggleAircraft()"> <span class="lp-dot flights"></span> Aircraft</label>
<span class="lp-count" id="lp-ac-count">0</span>
</div>
<div class="lp-note">Magenta = military · orange = firefighting · else altitude. Click a plane for type / squawk / operator.</div>
</div>
<div class="lp-layer">
<div class="lp-row">
@ -784,7 +795,7 @@
<label class="lp-name"><input type="checkbox" id="lp-vessels-on" onchange="toggleVessels()"> <span class="lp-dot vessels"></span> Vessels · AIS</label>
<span class="lp-count" id="lp-vessels-count">0</span>
</div>
<div class="lp-note">Needs AISSTREAM_API_KEY in Keys.</div>
<div class="lp-note">Needs AISSTREAM_API_KEY. Magenta = military · gold = law/gov · teal = underway. Click for type / dest / IMO.</div>
</div>
<div class="lp-layer">
<div class="lp-row">
@ -1855,9 +1866,13 @@ function applyLiveMarker(kind, p) {
}
}
function acColor(p) {
const t = String(((p.extra || {}).type || '')).toUpperCase();
if ((p.extra || {}).firefighter || firefighterHex.has(String(p.id)) || FF_ICAO.has(t)) return '#fb923c';
return altColor((p.extra || {}).alt_baro);
const extra = p.extra || {};
const t = String(extra.type || '').toUpperCase();
const em = String(extra.emergency || '').toLowerCase();
if (em && em !== 'none') return '#ff5d5d';
if (extra.firefighter || firefighterHex.has(String(p.id)) || FF_ICAO.has(t)) return '#fb923c';
if (extra.role === 'military') return '#f472b6';
return altColor(extra.alt_baro);
}
function connectLiveWs() {
if (liveWs && (liveWs.readyState === 0 || liveWs.readyState === 1)) return;
@ -2563,12 +2578,57 @@ function altColor(alt) {
if (a >= 1000) return '#fbbf24';
return '#fb923c';
}
function _popVal(v) {
if (v == null || v === '' || v === 'none' || v === 'undefined') return null;
return v;
}
function pointPopup(p) {
const extra = p.extra || {};
const rows = Object.keys(extra).filter(k => k !== 'stations' && extra[k] != null && extra[k] !== '')
.slice(0, 8)
.map(k => `<tr><td class="k">${esc(k)}</td><td>${esc(extra[k])}</td></tr>`).join('');
return `<div class="cam-pop"><b>${esc(p.label || p.id)}</b><table>${rows}</table></div>`;
const src = extra.src || '';
const role = extra.firefighter ? 'firefighter' : extra.role;
const badge = role
? ` <span class="role-badge ${esc(role)}">${esc(String(role).toUpperCase())}</span>`
: '';
const rows = [];
const add = (k, v) => {
const val = _popVal(v);
if (val == null) return;
rows.push(`<tr><td class="k">${esc(k)}</td><td>${esc(val)}</td></tr>`);
};
if (src === 'adsb.lol') {
add('type', extra.type);
add('aircraft', extra.desc);
add('operator', extra.ownOp);
add('reg', extra.reg);
const alt = extra.alt_baro;
add('alt', alt != null ? `${alt} ft` : null);
add('vs', extra.vs != null ? `${extra.vs} fpm` : null);
add('gs', p.speed != null ? `${p.speed} kt` : null);
add('hdg', p.heading != null ? `${Math.round(p.heading)}°` : null);
add('squawk', extra.squawk);
add('emergency', extra.emergency);
add('class', extra.emitter);
add('hex', extra.hex);
} else if (src === 'aisstream') {
add('kind', extra.kind);
add('flag', extra.country);
add('dest', extra.dest);
add('nav', extra.nav);
add('callsign', extra.callsign);
add('imo', extra.imo);
add('mmsi', extra.mmsi);
if (extra.length && extra.beam) add('size', `${extra.length} × ${extra.beam} m`);
else add('length', extra.length != null ? `${extra.length} m` : null);
add('draught', extra.draught != null ? `${extra.draught} m` : null);
add('sog', p.speed != null ? `${p.speed} kt` : null);
add('hdg', p.heading != null ? `${Math.round(p.heading)}°` : null);
add('eta', extra.eta);
} else {
Object.keys(extra).filter(k => k !== 'stations' && extra[k] != null && extra[k] !== '')
.slice(0, 8)
.forEach(k => add(k, extra[k]));
}
return `<div class="cam-pop"><b>${esc(p.label || p.id)}</b>${badge}<table>${rows.join('')}</table></div>`;
}
/* ── Heading-aware live-feed glyph markers ───────────────────────────────
@ -2617,15 +2677,19 @@ function feedIcon(feed, color, heading) {
function makePointMarker(p, colorFn, feed, renderer) {
if (p.lat == null || p.lon == null) return null;
const col = sanitizeColor(colorFn(p), '#35e0ff');
let m;
if (feed) {
const heading = Number(p.heading);
const icon = feedIcon(feed, col, Number.isNaN(heading) ? null : heading);
return L.marker([p.lat, p.lon], { icon }).bindPopup(() => pointPopup(p));
m = L.marker([p.lat, p.lon], { icon });
} else {
m = L.circleMarker([p.lat, p.lon], {
radius: 5, color: col, fillColor: col, fillOpacity: 0.9, weight: 1,
renderer,
});
}
return L.circleMarker([p.lat, p.lon], {
radius: 5, color: col, fillColor: col, fillOpacity: 0.9, weight: 1,
renderer,
}).bindPopup(() => pointPopup(p));
m._osintP = p;
return m.bindPopup(() => pointPopup(m._osintP || p));
}
function renderPoints(existing, points, colorFn, cluster, feed) {
const zoom = map.getZoom();
@ -2662,6 +2726,7 @@ function renderPoints(existing, points, colorFn, cluster, feed) {
const m = byId.get(id);
if (m) {
m.setLatLng([p.lat, p.lon]);
m._osintP = p;
if (feed) m.setIcon(feedIcon(feed, col, p.heading));
else if (m.setStyle) m.setStyle({ color: col, fillColor: col });
} else {
@ -2920,6 +2985,9 @@ async function loadVessels() {
const pts = await r.json();
if (req !== overlayReq.vessels) return;
vesselsGroup = renderPoints(vesselsGroup, Array.isArray(pts) ? pts : [], p => {
const extra = p.extra || {};
if (extra.role === 'military') return '#f472b6';
if (extra.role === 'government') return '#facc15';
const sog = Number((p.speed) || 0);
return sog > 0.5 ? '#2dd4bf' : '#64748b';
}, true, 'vessel');

View file

@ -362,3 +362,113 @@ def test_wfigs_params_requests_simplified_geometry():
# Envelope is the quantized cell, not the raw pan box.
geom = params["geometry"]
assert geom != "-84.5,33.8,-75.4,36.6"
def test_transform_adsb_lol_flags_military_from_dbflags():
payload = {
"ac": [
{
"hex": "ae01ab",
"flight": "RCH123 ",
"r": "04-1234",
"t": "C17",
"lat": 35.1,
"lon": -77.9,
"alt_baro": 24000,
"gs": 410,
"track": 90,
"squawk": "5101",
"emergency": "none",
"category": "A5",
"dbFlags": 1,
"baro_rate": 64,
"alt_geom": 24500,
"desc": "Boeing C-17A Globemaster III",
"ownOp": "USAF",
},
{
"hex": "a1b2c3",
"flight": "AAL123",
"r": "N123AA",
"t": "B738",
"lat": 35.88,
"lon": -78.79,
"alt_baro": 32000,
"gs": 430,
"track": 87,
"squawk": "1200",
"emergency": "none",
"category": "A3",
},
]
}
rows = {r["id"]: r for r in transform_adsb_lol(payload)}
mil = rows["ae01ab"]["extra"]
civ = rows["a1b2c3"]["extra"]
assert mil["role"] == "military"
assert mil["role_src"] == "dbFlags"
assert mil["emitter"] == "heavy"
assert mil["desc"] == "Boeing C-17A Globemaster III"
assert mil["ownOp"] == "USAF"
assert mil["vs"] == 64
assert mil["alt_geom"] == 24500
assert civ["role"] == "civilian"
assert civ["emitter"] == "large"
def test_transform_adsb_lol_military_from_icao_type_and_hex():
payload = {
"ac": [
{"hex": "3b76aa", "flight": "FAF123", "t": "F16", "lat": 1, "lon": 2, "category": "A1"},
{"hex": "ae1234", "flight": "BOXER1", "t": "C172", "lat": 1, "lon": 2, "category": "A1"},
]
}
rows = {r["id"]: r for r in transform_adsb_lol(payload)}
assert rows["3b76aa"]["extra"]["role"] == "military"
assert rows["3b76aa"]["extra"]["role_src"] == "type"
assert rows["ae1234"]["extra"]["role"] == "military"
assert rows["ae1234"]["extra"]["role_src"] == "hex"
def test_transform_ais_static_classifies_military_and_cargo():
mil = transform_ais_frame({
"MessageType": "ShipStaticData",
"MetaData": {"MMSI": 338123456, "ShipName": "USNS BOB", "Latitude": 32.7, "Longitude": -117.2},
"Message": {"ShipStaticData": {
"Type": 35, "CallSign": "NBXX", "ImoNumber": 0,
"Destination": "SAN DIEGO", "MaximumStaticDraught": 8.2,
"Dimension": {"A": 80, "B": 20, "C": 8, "D": 8},
"Eta": {"Month": 8, "Day": 29, "Hour": 14, "Minute": 0},
}},
})
cargo = transform_ais_frame({
"MessageType": "ShipStaticData",
"MetaData": {"MMSI": 477123456, "ShipName": "EVER GIVEN", "Latitude": 36.9, "Longitude": -76.3},
"Message": {"ShipStaticData": {
"Type": 70, "CallSign": "VRXX", "ImoNumber": 9811000,
"Destination": "NORFOLK", "MaximumStaticDraught": 14.5,
"Dimension": {"A": 200, "B": 150, "C": 20, "D": 20},
}},
})
assert mil is not None and cargo is not None
assert mil["extra"]["role"] == "military"
assert mil["extra"]["kind"] == "military"
assert mil["extra"]["callsign"] == "NBXX"
assert mil["extra"]["length"] == 100
assert mil["extra"]["beam"] == 16
assert mil["extra"]["dest"] == "SAN DIEGO"
assert mil["extra"]["country"] == "United States"
assert cargo["extra"]["role"] == "civilian"
assert cargo["extra"]["kind"] == "cargo"
assert cargo["extra"]["imo"] == 9811000
def test_transform_ais_position_decodes_navstat():
row = transform_ais_frame({
"MessageType": "PositionReport",
"MetaData": {"MMSI": 366912810, "ShipName": "EVER GIVEN", "latitude": 36.9, "longitude": -76.3},
"Message": {"PositionReport": {"Sog": 0.1, "Cog": 88.0, "TrueHeading": 90, "NavigationalStatus": 5}},
})
assert row is not None
assert row["extra"]["nav"] == "moored"
assert row["extra"]["navstat"] == 5