fdd59052c5
National storm-based warnings are fetched once (45s TTL) and clipped to the quantized bbox. Popup fields only — NWS descriptions stay off the wire.
353 lines
10 KiB
Python
353 lines
10 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",
|
|
}
|