ca15885e7f
Aircraft: readsb dbFlags bit0, then ICAO combat types / US-DoD hex / callsign. Popup shows type, operator, alt, vs, squawk, emitter class. Vessels: ITU ship type 35 = military, 50/51/55 = government; cargo/ tanker/passenger/etc otherwise. Popup shows kind, flag, dest, nav, IMO, size. Magenta glyphs = military, gold = law/gov.
474 lines
15 KiB
Python
474 lines
15 KiB
Python
"""Unit tests for live map-layer mappers (aircraft, trains, AIS, WFIGS, Caltrans)."""
|
|
|
|
from live_layers import (
|
|
MARKER_FIELDS,
|
|
bbox_center_radius_nm,
|
|
clip_fc_to_bbox,
|
|
filter_points_bbox,
|
|
parse_bbox,
|
|
quantize_bbox,
|
|
rainviewer_tile_url,
|
|
slim_alert_properties,
|
|
to_marker,
|
|
transform_adsb_lol,
|
|
transform_ais_frame,
|
|
transform_amtraker,
|
|
transform_nhc_storms,
|
|
transform_wfigs_incidents,
|
|
_cache,
|
|
_ttl_get,
|
|
_wfigs_params,
|
|
)
|
|
|
|
from camera_scraper import parse_caltrans_json
|
|
|
|
|
|
def test_parse_bbox_and_radius_clamps_to_150_nm():
|
|
minlon, minlat, maxlon, maxlat = parse_bbox("-84.5,33.8,-75.4,36.6")
|
|
assert (minlon, minlat, maxlon, maxlat) == (-84.5, 33.8, -75.4, 36.6)
|
|
lat, lon, radius = bbox_center_radius_nm(minlon, minlat, maxlon, maxlat)
|
|
assert 35.0 < lat < 35.4
|
|
assert -80.1 < lon < -79.8
|
|
assert 1 <= radius <= 150
|
|
|
|
|
|
def test_parse_bbox_rejects_malformed():
|
|
try:
|
|
parse_bbox("1,2,3")
|
|
assert False, "expected ValueError"
|
|
except ValueError:
|
|
pass
|
|
|
|
|
|
def test_transform_adsb_lol_maps_shared_marker_contract():
|
|
payload = {
|
|
"ac": [
|
|
{
|
|
"hex": "a1b2c3",
|
|
"flight": "AAL123 ",
|
|
"r": "N123AA",
|
|
"t": "B738",
|
|
"lat": 35.88,
|
|
"lon": -78.79,
|
|
"alt_baro": 32000,
|
|
"gs": 430.2,
|
|
"track": 87.5,
|
|
"squawk": "1200",
|
|
"emergency": "none",
|
|
"category": "A3",
|
|
"seen_pos": 0.4,
|
|
},
|
|
{"hex": "dead00", "flight": "NOFIX"}, # no coords → drop
|
|
]
|
|
}
|
|
rows = transform_adsb_lol(payload)
|
|
assert len(rows) == 1
|
|
m = rows[0]
|
|
assert set(MARKER_FIELDS).issubset(m)
|
|
assert m["id"] == "a1b2c3"
|
|
assert m["lat"] == 35.88
|
|
assert m["lon"] == -78.79
|
|
assert m["heading"] == 87.5
|
|
assert m["speed"] == 430.2
|
|
assert m["label"] == "AAL123"
|
|
assert m["extra"]["squawk"] == "1200"
|
|
assert m["extra"]["alt_baro"] == 32000
|
|
|
|
|
|
def test_transform_amtraker_flattens_train_numbers():
|
|
payload = {
|
|
"1": [
|
|
{
|
|
"trainID": "1-9",
|
|
"trainNum": "1",
|
|
"routeName": "Sunset Limited",
|
|
"lat": 29.76,
|
|
"lon": -95.36,
|
|
"heading": 90,
|
|
"velocity": 45.0,
|
|
"late": 12,
|
|
"iconColor": "#ee3a43",
|
|
"stations": [{"name": "Houston", "status": "enroute"}],
|
|
}
|
|
],
|
|
"5": [
|
|
{
|
|
"trainID": "5-12",
|
|
"trainNum": "5",
|
|
"routeName": "California Zephyr",
|
|
"lat": 40.0,
|
|
"lon": -105.0,
|
|
"heading": "W",
|
|
"lateMin": 5,
|
|
"iconColor": "#005eb8",
|
|
}
|
|
],
|
|
}
|
|
rows = transform_amtraker(payload)
|
|
assert {r["id"] for r in rows} == {"1-9", "5-12"}
|
|
sunset = next(r for r in rows if r["id"] == "1-9")
|
|
assert sunset["lat"] == 29.76
|
|
assert sunset["label"] == "Sunset Limited #1"
|
|
assert sunset["speed"] == 45.0
|
|
assert sunset["extra"]["late_min"] == 12
|
|
assert sunset["extra"]["iconColor"] == "#ee3a43"
|
|
|
|
|
|
def test_transform_ais_position_report():
|
|
frame = {
|
|
"MessageType": "PositionReport",
|
|
"MetaData": {
|
|
"MMSI": 366912810,
|
|
"ShipName": "EVER GIVEN",
|
|
"latitude": 36.9,
|
|
"longitude": -76.3,
|
|
},
|
|
"Message": {
|
|
"PositionReport": {
|
|
"Sog": 12.4,
|
|
"Cog": 88.0,
|
|
"TrueHeading": 90,
|
|
"NavigationalStatus": 0,
|
|
}
|
|
},
|
|
}
|
|
row = transform_ais_frame(frame)
|
|
assert row is not None
|
|
assert row["id"] == "366912810"
|
|
assert row["lat"] == 36.9
|
|
assert row["lon"] == -76.3
|
|
assert row["label"] == "EVER GIVEN"
|
|
assert row["speed"] == 12.4
|
|
assert row["heading"] == 90
|
|
assert row["extra"]["navstat"] == 0
|
|
|
|
|
|
def test_transform_ais_ignores_non_position():
|
|
assert transform_ais_frame({"MessageType": "Unknown"}) is None
|
|
|
|
|
|
def test_transform_wfigs_incidents_geojson():
|
|
fc = {
|
|
"type": "FeatureCollection",
|
|
"features": [
|
|
{
|
|
"type": "Feature",
|
|
"geometry": {"type": "Point", "coordinates": [-81.3, 28.5]},
|
|
"properties": {
|
|
"IncidentName": "Foster Bridge",
|
|
"IncidentSize": 675,
|
|
"PercentContained": 100,
|
|
"POOState": "US-FL",
|
|
"IncidentTypeCategory": "WF",
|
|
"FireCause": "Human",
|
|
"FireDiscoveryDateTime": 1750000000000,
|
|
},
|
|
}
|
|
],
|
|
}
|
|
rows = transform_wfigs_incidents(fc)
|
|
assert len(rows) == 1
|
|
assert rows[0]["label"] == "Foster Bridge"
|
|
assert rows[0]["lat"] == 28.5
|
|
assert rows[0]["lon"] == -81.3
|
|
assert rows[0]["extra"]["acres"] == 675
|
|
assert rows[0]["extra"]["contained"] == 100
|
|
|
|
|
|
def test_filter_points_bbox():
|
|
pts = [
|
|
to_marker("a", 35.0, -78.0, label="in"),
|
|
to_marker("b", 10.0, 20.0, label="out"),
|
|
]
|
|
kept = filter_points_bbox(pts, -80, 33, -75, 37)
|
|
assert [p["id"] for p in kept] == ["a"]
|
|
|
|
|
|
def test_rainviewer_tile_url():
|
|
url = rainviewer_tile_url(
|
|
host="https://tilecache.rainviewer.com",
|
|
path="/v2/radar/cb581daa2c0f",
|
|
)
|
|
assert url == (
|
|
"https://tilecache.rainviewer.com/v2/radar/cb581daa2c0f/256/{z}/{x}/{y}/2/1_1.png"
|
|
)
|
|
|
|
|
|
def test_transform_nhc_storms():
|
|
payload = {
|
|
"activeStorms": [
|
|
{
|
|
"id": "al042026",
|
|
"name": "Dolly",
|
|
"classification": "TS",
|
|
"latitudeNumeric": 13.6,
|
|
"longitudeNumeric": -38.7,
|
|
"movementDir": 280,
|
|
"movementSpeed": 12,
|
|
"intensity": 35,
|
|
}
|
|
]
|
|
}
|
|
rows = transform_nhc_storms(payload)
|
|
assert len(rows) == 1
|
|
assert rows[0]["id"] == "al042026"
|
|
assert rows[0]["label"] == "Tropical Storm Dolly"
|
|
assert rows[0]["lat"] == 13.6
|
|
|
|
|
|
def test_parse_caltrans_skips_oos_and_maps_jpeg_hls():
|
|
payload = """
|
|
{"data":[
|
|
{"cctv":{
|
|
"location":{
|
|
"latitude":"37.8","longitude":"-122.4",
|
|
"locationName":"I-80 WB","nearbyPlace":"SF",
|
|
"district":"4","route":"80","county":"SF","direction":"W"
|
|
},
|
|
"inService":"true",
|
|
"imageData":{
|
|
"static":{"currentImageURL":"https://cwwp2.dot.ca.gov/data/d4/cctv/image/cam.jpg"},
|
|
"streamingVideoURL":"https://wzmedia.dot.ca.gov/D4/cam.stream/playlist.m3u8"
|
|
}
|
|
}},
|
|
{"cctv":{
|
|
"location":{"latitude":"1","longitude":"2","locationName":"down"},
|
|
"inService":"false",
|
|
"imageData":{"static":{"currentImageURL":"https://example.com/x.jpg"}}
|
|
}}
|
|
]}
|
|
"""
|
|
cams = parse_caltrans_json(payload, "caltrans")
|
|
assert len(cams) == 1
|
|
cam = cams[0]
|
|
assert cam["discovery_source"] == "caltrans"
|
|
assert cam["snapshot_url"].endswith("cam.jpg")
|
|
assert cam["source_url"].endswith("playlist.m3u8")
|
|
assert cam["device_type"] == "hls"
|
|
assert cam["location_lat"] == 37.8
|
|
assert cam["location_lon"] == -122.4
|
|
assert "I-80" in cam["location_name"]
|
|
assert "rtsp://" not in cam["source_url"].lower()
|
|
assert "rtsp://" not in cam["snapshot_url"].lower()
|
|
|
|
|
|
def test_quantize_bbox_stable_under_jitter():
|
|
a = quantize_bbox(*parse_bbox("-78.7912,35.7711,-78.6101,35.9102"))
|
|
b = quantize_bbox(*parse_bbox("-78.7900,35.7700,-78.6110,35.9090"))
|
|
assert a == b
|
|
minlon, minlat, maxlon, maxlat = a
|
|
assert minlon <= -78.7912
|
|
assert minlat <= 35.7700
|
|
assert maxlon >= -78.6101
|
|
assert maxlat >= 35.9102
|
|
|
|
|
|
def test_ttl_get_does_not_block_other_keys():
|
|
import asyncio
|
|
|
|
_cache.clear()
|
|
order = []
|
|
|
|
async def slow():
|
|
order.append("slow-start")
|
|
await asyncio.sleep(0.2)
|
|
order.append("slow-end")
|
|
return "S"
|
|
|
|
async def fast():
|
|
order.append("fast")
|
|
return "F"
|
|
|
|
async def run():
|
|
t1 = asyncio.create_task(_ttl_get("slow", 5, slow))
|
|
await asyncio.sleep(0.01)
|
|
t2 = asyncio.create_task(_ttl_get("fast", 5, fast))
|
|
await asyncio.gather(t1, t2)
|
|
|
|
asyncio.run(run())
|
|
assert order.index("fast") < order.index("slow-end")
|
|
assert _cache["slow"][1] == "S"
|
|
assert _cache["fast"][1] == "F"
|
|
_cache.clear()
|
|
|
|
|
|
def test_clip_fc_to_bbox_drops_far_features_and_empty_geometry():
|
|
fc = {
|
|
"type": "FeatureCollection",
|
|
"features": [
|
|
{
|
|
"type": "Feature",
|
|
"properties": {"event": "near"},
|
|
"geometry": {"type": "Point", "coordinates": [-78.7, 35.8]},
|
|
},
|
|
{
|
|
"type": "Feature",
|
|
"properties": {"event": "far"},
|
|
"geometry": {"type": "Point", "coordinates": [-120.0, 45.0]},
|
|
},
|
|
{
|
|
"type": "Feature",
|
|
"properties": {"event": "nogeom"},
|
|
"geometry": None,
|
|
},
|
|
{
|
|
"type": "Feature",
|
|
"properties": {"event": "poly-overlap"},
|
|
"geometry": {
|
|
"type": "Polygon",
|
|
"coordinates": [[
|
|
[-79.0, 35.0], [-78.0, 35.0], [-78.0, 36.0],
|
|
[-79.0, 36.0], [-79.0, 35.0],
|
|
]],
|
|
},
|
|
},
|
|
],
|
|
}
|
|
clipped = clip_fc_to_bbox(fc, -79.0, 35.5, -78.0, 36.0)
|
|
events = [f["properties"]["event"] for f in clipped["features"]]
|
|
assert events == ["near", "poly-overlap"]
|
|
|
|
|
|
def test_slim_alert_properties_keeps_popup_fields_only():
|
|
fat = {
|
|
"event": "Tornado Warning",
|
|
"severity": "Extreme",
|
|
"headline": "TORNADO WARNING",
|
|
"areaDesc": "Wake",
|
|
"wfo": "RAH",
|
|
"source": "nws",
|
|
"parameters": {"WIND": ["70"]},
|
|
"description": "A long narrative " * 40,
|
|
"instruction": "Take shelter.",
|
|
"geocode": {"SAME": ["037183"]},
|
|
}
|
|
slim = slim_alert_properties(fat)
|
|
assert slim == {
|
|
"event": "Tornado Warning",
|
|
"severity": "Extreme",
|
|
"headline": "TORNADO WARNING",
|
|
"areaDesc": "Wake",
|
|
"wfo": "RAH",
|
|
"source": "nws",
|
|
}
|
|
|
|
|
|
def test_wfigs_params_requests_simplified_geometry():
|
|
params = _wfigs_params("-84.5,33.8,-75.4,36.6")
|
|
assert "maxAllowableOffset" in params
|
|
assert float(params["maxAllowableOffset"]) > 0
|
|
assert params["geometryPrecision"] == 5
|
|
assert int(params["resultRecordCount"]) <= 500
|
|
# 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
|