640fea31f4
Keep both UDOT IBI 511 (this PR) and ODOT TripCheck (merged #40) plus existing MDOT parsers. Disjoint camera sources.
988 lines
33 KiB
Python
988 lines
33 KiB
Python
"""Unit tests for live map-layer mappers (aircraft, trains, AIS, WFIGS, Caltrans)."""
|
|
|
|
import json
|
|
|
|
from live_layers import (
|
|
MARKER_FIELDS,
|
|
bbox_center_radius_nm,
|
|
clip_fc_to_bbox,
|
|
filter_points_bbox,
|
|
parse_bbox,
|
|
quantize_bbox,
|
|
pick_sentinel_feature,
|
|
rainviewer_tile_url,
|
|
sign_cog_url,
|
|
sentinel1_tile_url,
|
|
slim_alert_properties,
|
|
to_marker,
|
|
transform_adsb_lol,
|
|
transform_ais_frame,
|
|
transform_amtraker,
|
|
transform_nhc_storms,
|
|
transform_wfigs_incidents,
|
|
SENTINEL1_ATTRIBUTION,
|
|
TITILER_COG_TILES,
|
|
_cache,
|
|
_ttl_get,
|
|
_wfigs_params,
|
|
)
|
|
|
|
from camera_scraper import parse_caltrans_json, parse_udot_ibi_page, parse_odot_json, parse_mdot_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()
|
|
|
|
|
|
# ── UDOT IBI 511 parser ──────────────────────────────────────────────────
|
|
|
|
def _udot_row(cam_id, lng, lat, **img_overrides):
|
|
img = {
|
|
"id": cam_id, "cameraSiteId": cam_id,
|
|
"imageUrl": f"/map/Cctv/{cam_id}", "disabled": False, "blocked": False,
|
|
}
|
|
img.update(img_overrides)
|
|
return {
|
|
"id": cam_id, "sourceId": "102771", "source": "ADX",
|
|
"roadway": "Unknown", "direction": "Unknown",
|
|
"location": "Freedom Blvd / 200 W @ 1100 N, PVO",
|
|
"latLng": {"geography": {
|
|
"coordinateSystemId": 4326,
|
|
"wellKnownText": f"POINT ({lng} {lat})"}},
|
|
"images": [img],
|
|
}
|
|
|
|
|
|
def _udot_page(rows):
|
|
import json
|
|
return json.dumps({"draw": 0, "recordsTotal": len(rows),
|
|
"recordsFiltered": len(rows), "data": rows})
|
|
|
|
|
|
def test_parse_udot_wkt_maps_lng_lat():
|
|
cams = parse_udot_ibi_page(_udot_page([_udot_row(112731, -111.66204, 40.24863)]))
|
|
assert len(cams) == 1
|
|
cam = cams[0]
|
|
# WKT is `POINT (lng lat)` — order must not be swapped.
|
|
assert cam["location_lat"] == 40.24863
|
|
assert cam["location_lon"] == -111.66204
|
|
assert cam["discovery_source"] == "udot"
|
|
assert cam["vendor"] == "UDOT"
|
|
assert cam["source_url"] == "https://prod-ut.ibi511.com/map/Cctv/112731"
|
|
assert cam["snapshot_url"] == cam["source_url"]
|
|
assert "rtsp://" not in cam["source_url"].lower()
|
|
assert cam["raw"]["udot_id"] == 112731
|
|
|
|
|
|
def test_parse_udot_skips_blocked_and_disabled():
|
|
rows = [
|
|
_udot_row(1, -111.0, 40.0),
|
|
_udot_row(2, -111.1, 40.1, blocked=True),
|
|
_udot_row(3, -111.2, 40.2, disabled=True),
|
|
]
|
|
rows.append(_udot_row(4, -111.3, 40.3))
|
|
rows[3]["images"] = [] # no images → drop
|
|
cams = parse_udot_ibi_page(_udot_page(rows))
|
|
assert [c["raw"]["udot_id"] for c in cams] == [1]
|
|
|
|
|
|
def test_parse_udot_drops_out_of_bbox():
|
|
rows = [
|
|
_udot_row(1, -111.0, 40.0), # inside Utah
|
|
_udot_row(2, -100.0, 40.0), # east of -108.9
|
|
_udot_row(3, -120.0, 40.0), # west of -114.2
|
|
_udot_row(4, -111.0, 44.0), # north of 42.1
|
|
_udot_row(5, -111.0, 30.0), # south of 36.9
|
|
]
|
|
cams = parse_udot_ibi_page(_udot_page(rows))
|
|
assert [c["raw"]["udot_id"] for c in cams] == [1]
|
|
|
|
|
|
def test_parse_udot_bad_payload_returns_empty():
|
|
import json
|
|
assert parse_udot_ibi_page("not json") == []
|
|
assert parse_udot_ibi_page(json.dumps({"data": None})) == []
|
|
assert parse_udot_ibi_page(json.dumps({"data": "nope"})) == []
|
|
|
|
|
|
def test_parse_udot_missing_wkt_skipped():
|
|
row = _udot_row(1, -111.0, 40.0)
|
|
row["latLng"] = {}
|
|
assert parse_udot_ibi_page(_udot_page([row])) == []
|
|
|
|
|
|
def test_parse_odot_tripcheck_keeps_valid_skips_missing_and_oob():
|
|
payload = """
|
|
{"features":[
|
|
{"attributes":{
|
|
"cameraId":277,"filename":"AstoriaUS101_pid392.jpg",
|
|
"latitude":46.18785,"longitude":-123.85347,
|
|
"route":"US101 ","title":"US101 at Astoria"
|
|
}},
|
|
{"attributes":{
|
|
"cameraId":200,"filename":"","latitude":45.0,"longitude":-122.0,
|
|
"route":"I-5","title":"missing filename"
|
|
}},
|
|
{"attributes":{
|
|
"cameraId":300,"filename":"nocal_pid1.jpg",
|
|
"latitude":40.0,"longitude":-122.0,
|
|
"route":"US97","title":"out of bbox"
|
|
}},
|
|
{"attributes":{
|
|
"cameraId":400,"filename":"badcoord_pid2.jpg",
|
|
"latitude":null,"longitude":-122.0,
|
|
"route":"OR22","title":"null coord"
|
|
}}
|
|
]}
|
|
"""
|
|
cams = parse_odot_json(payload, "www.tripcheck.com")
|
|
assert len(cams) == 1
|
|
cam = cams[0]
|
|
assert cam["discovery_source"] == "odot"
|
|
assert cam["snapshot_url"] == (
|
|
"https://tripcheck.com/RoadCams/cams/AstoriaUS101_pid392.jpg")
|
|
assert cam["source_url"] == cam["snapshot_url"]
|
|
assert cam["location_lat"] == 46.18785
|
|
assert cam["location_lon"] == -123.85347
|
|
assert "US101 at Astoria" in cam["location_name"]
|
|
assert cam["vendor"] == "ODOT"
|
|
assert cam["device_type"] == "http"
|
|
assert "rtsp://" not in cam["snapshot_url"].lower()
|
|
|
|
|
|
def test_parse_odot_tripcheck_handles_malformed():
|
|
assert parse_odot_json("not json", "www.tripcheck.com") == []
|
|
assert parse_odot_json('{"features":null}', "www.tripcheck.com") == []
|
|
|
|
|
|
def test_parse_mdot_extracts_html_fields_and_bbox_filters():
|
|
rows = [
|
|
# In-bbox, full fields.
|
|
{
|
|
"route": "11 Mile",
|
|
"county": 'Wayne County <a href="/MiDrive/map?cameras=true&lat=42.491304&lon=-83.04479&zoom=15&id=1129"target="_blank">Go to</a>',
|
|
"location": " @ Mound NB",
|
|
"direction": "Traffic closest to camera is traveling north.",
|
|
"image": '<img alt="x" class="cameraImageForActivePane" id="1129Img" src="https://micamerasimages.net/thumbs/semtoc_cam_253.flv.jpg?item=1" height="170" width="250" onerror="cameraImageBroken(this)">',
|
|
},
|
|
# Out of bbox (lat 50) → drop.
|
|
{
|
|
"route": "Far",
|
|
"county": 'Nowhere <a href="/MiDrive/map?lat=50.0&lon=-83.0&zoom=15&id=9999">Go to</a>',
|
|
"location": "",
|
|
"image": '<img src="https://micamerasimages.net/thumbs/x.jpg">',
|
|
},
|
|
# Missing coordinates → drop.
|
|
{
|
|
"route": "NoCoords",
|
|
"county": 'Somewhere <a href="/MiDrive/map?zoom=15&id=8888">Go to</a>',
|
|
"location": "",
|
|
"image": '<img src="https://micamerasimages.net/thumbs/y.jpg">',
|
|
},
|
|
# Missing image → drop.
|
|
{
|
|
"route": "NoImage",
|
|
"county": 'Kent <a href="/MiDrive/map?lat=42.8841&lon=-85.6646&zoom=15&id=2113">Go to</a>',
|
|
"location": " @ Division",
|
|
"image": "",
|
|
},
|
|
# RTSP image src → drop.
|
|
{
|
|
"route": "Rtsp",
|
|
"county": 'Wayne <a href="/MiDrive/map?lat=42.4&lon=-83.1&zoom=15&id=1234">Go to</a>',
|
|
"location": "",
|
|
"image": '<img src="rtsp://10.0.0.1/stream">',
|
|
},
|
|
]
|
|
cams = parse_mdot_json(json.dumps(rows), "mdotjboss.state.mi.us")
|
|
assert len(cams) == 1
|
|
cam = cams[0]
|
|
assert cam["discovery_source"] == "mdot"
|
|
assert cam["location_lat"] == 42.491304
|
|
assert cam["location_lon"] == -83.04479
|
|
assert cam["snapshot_url"] == "https://micamerasimages.net/thumbs/semtoc_cam_253.flv.jpg?item=1"
|
|
assert cam["source_url"] == "https://mdotjboss.state.mi.us/MiDrive/camera/1129"
|
|
assert cam["device_type"] == "http"
|
|
assert cam["vendor"] == "MDOT"
|
|
assert "11 Mile @ Mound NB" in cam["location_name"]
|
|
assert "Wayne County" in cam["location_name"]
|
|
|
|
|
|
def test_parse_mdot_handles_malformed_payload():
|
|
assert parse_mdot_json("not json", "mdot") == []
|
|
assert parse_mdot_json('{"not": "a list"}', "mdot") == []
|
|
assert parse_mdot_json("[]", "mdot") == []
|
|
|
|
|
|
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
|
|
|
|
|
|
def test_nws_alerts_does_not_send_bbox_param(monkeypatch):
|
|
"""api.weather.gov/alerts/active 400s on bbox — clip locally instead."""
|
|
import asyncio
|
|
|
|
from live_layers import fetch_weather_alerts, _cache
|
|
|
|
seen = []
|
|
|
|
async def fake_get(url, params=None):
|
|
seen.append((url, dict(params or {})))
|
|
if "weather.gov" in url:
|
|
return {
|
|
"type": "FeatureCollection",
|
|
"features": [{
|
|
"type": "Feature",
|
|
"properties": {"event": "Tornado Warning", "severity": "Extreme"},
|
|
"geometry": {"type": "Point", "coordinates": [-78.7, 35.8]},
|
|
}],
|
|
}
|
|
return {"type": "FeatureCollection", "features": []}
|
|
|
|
monkeypatch.setattr("live_layers._get_json", fake_get)
|
|
_cache.clear()
|
|
fc = asyncio.run(fetch_weather_alerts(None, "-79.0,35.5,-78.0,36.0"))
|
|
nws_calls = [p for u, p in seen if "weather.gov" in u]
|
|
assert nws_calls, "NWS should still be fetched"
|
|
assert "bbox" not in nws_calls[0]
|
|
assert fc.get("nws_ok") is True
|
|
assert len(fc["features"]) == 1
|
|
|
|
|
|
def test_nws_alerts_failure_is_flagged(monkeypatch):
|
|
import asyncio
|
|
|
|
from live_layers import fetch_weather_alerts, _cache
|
|
|
|
async def fake_get(url, params=None):
|
|
if "weather.gov" in url:
|
|
raise RuntimeError("400 Bad Request")
|
|
return {"type": "FeatureCollection", "features": []}
|
|
|
|
monkeypatch.setattr("live_layers._get_json", fake_get)
|
|
_cache.clear()
|
|
fc = asyncio.run(fetch_weather_alerts(None, None))
|
|
assert fc.get("nws_ok") is False
|
|
|
|
|
|
def test_fetch_aircraft_get_path_does_not_persist(monkeypatch):
|
|
"""GET /api/aircraft must serve last-known without track/geofence writes."""
|
|
import asyncio
|
|
|
|
from live_layers import (
|
|
aircraft_last_known, fetch_aircraft, persist_aircraft_snapshot, _cache,
|
|
)
|
|
|
|
aircraft_last_known.clear()
|
|
aircraft_last_known["abc"] = {
|
|
"id": "abc", "lat": 35.8, "lon": -78.7, "heading": 90, "speed": 400,
|
|
"label": "ABC", "extra": {},
|
|
}
|
|
writes = {"n": 0}
|
|
|
|
async def boom(*a, **k):
|
|
writes["n"] += 1
|
|
raise AssertionError("GET path must not persist")
|
|
|
|
monkeypatch.setattr("tracks.record_position", boom)
|
|
monkeypatch.setattr("geofence.record_and_notify", boom)
|
|
_cache.clear()
|
|
rows = asyncio.run(fetch_aircraft("-79,35,-78,36", persist=False))
|
|
assert writes["n"] == 0
|
|
assert any(r["id"] == "abc" for r in rows)
|
|
|
|
|
|
def test_persist_aircraft_snapshot_writes_tracks(monkeypatch):
|
|
import asyncio
|
|
|
|
from live_layers import persist_aircraft_snapshot
|
|
|
|
recorded = []
|
|
|
|
async def fake_record(kind, marker):
|
|
recorded.append((kind, marker["id"]))
|
|
return True
|
|
|
|
async def fake_gf(**kw):
|
|
return 0
|
|
|
|
monkeypatch.setattr("tracks.record_position", fake_record)
|
|
monkeypatch.setattr("geofence.record_and_notify", fake_gf)
|
|
monkeypatch.setattr("ws_manager.manager.has_clients", lambda: False)
|
|
|
|
rows = [{
|
|
"id": "abc", "lat": 35.8, "lon": -78.7, "heading": 90, "speed": 400,
|
|
"label": "ABC", "extra": {},
|
|
}]
|
|
asyncio.run(persist_aircraft_snapshot(rows))
|
|
assert recorded == [("aircraft", "abc")]
|
|
|
|
|
|
# ── Planespotters.net photo lookup ────────────────────────────────────────
|
|
|
|
def test_normalize_planespotter_photo_prefers_large_thumbnail():
|
|
from live_layers import _normalize_planespotter_photo
|
|
|
|
out = _normalize_planespotter_photo({
|
|
"id": "1053982",
|
|
"thumbnail": {"src": "https://t.plnspttrs.net/x_t.jpg", "size": {"width": 200, "height": 141}},
|
|
"thumbnail_large": {"src": "https://t.plnspttrs.net/x_280.jpg", "size": {"width": 395, "height": 280}},
|
|
"link": "https://www.planespotters.net/photo/1053982/foo",
|
|
"photographer": "Günther Feniuk",
|
|
})
|
|
assert out["id"] == "1053982"
|
|
assert out["src"] == "https://t.plnspttrs.net/x_280.jpg"
|
|
assert out["width"] == 395
|
|
assert out["height"] == 280
|
|
assert out["photographer"] == "Günther Feniuk"
|
|
assert "planespotters.net" in out["link"]
|
|
|
|
|
|
def test_normalize_planespotter_photo_empty_or_malformed_returns_none():
|
|
from live_layers import _normalize_planespotter_photo
|
|
|
|
assert _normalize_planespotter_photo({}) is None
|
|
assert _normalize_planespotter_photo({"thumbnail": {}}) is None
|
|
assert _normalize_planespotter_photo(None) is None
|
|
assert _normalize_planespotter_photo("not-a-dict") is None
|
|
|
|
|
|
def test_fetch_planespotters_photo_hex_builds_url_and_normalizes(monkeypatch):
|
|
import asyncio
|
|
|
|
from live_layers import fetch_planespotters_photo, _cache
|
|
|
|
seen = []
|
|
|
|
async def fake_get(url, params=None, headers=None):
|
|
seen.append((url, (headers or {}).get("User-Agent", "")))
|
|
return {"photos": [{
|
|
"id": "1", "thumbnail": {"src": "https://t.plnspttrs.net/a_t.jpg"},
|
|
"thumbnail_large": {"src": "https://t.plnspttrs.net/a_280.jpg"},
|
|
"link": "https://www.planespotters.net/photo/1/x", "photographer": "A",
|
|
}]}
|
|
|
|
monkeypatch.setattr("live_layers._get_json", fake_get)
|
|
_cache.clear()
|
|
out = asyncio.run(fetch_planespotters_photo(hex_code="e8027e"))
|
|
assert out["src"] == "https://t.plnspttrs.net/a_280.jpg"
|
|
assert seen[0][0] == "https://api.planespotters.net/pub/photos/hex/e8027e"
|
|
assert "@" in seen[0][1] or "http" in seen[0][1]
|
|
|
|
|
|
def test_fetch_planespotters_photo_reg_fallback_and_no_result(monkeypatch):
|
|
import asyncio
|
|
|
|
from live_layers import fetch_planespotters_photo, _cache
|
|
|
|
seen = []
|
|
|
|
async def fake_get(url, params=None, headers=None):
|
|
seen.append(url)
|
|
return {"photos": []}
|
|
|
|
monkeypatch.setattr("live_layers._get_json", fake_get)
|
|
_cache.clear()
|
|
assert asyncio.run(fetch_planespotters_photo(reg="D-ABCD")) is None
|
|
assert seen == ["https://api.planespotters.net/pub/photos/reg/D-ABCD"]
|
|
# no hex and no reg → no upstream call at all
|
|
assert asyncio.run(fetch_planespotters_photo()) is None
|
|
|
|
|
|
def test_planespotters_headers_add_contact_when_ua_is_generic(monkeypatch):
|
|
import live_layers
|
|
|
|
monkeypatch.setattr(live_layers, "OSINT_USER_AGENT", "osint-dashboard/1.0 (self-hosted)")
|
|
ua = live_layers._planespotters_headers()["User-Agent"]
|
|
assert "osint-dashboard" in ua
|
|
assert "@" in ua
|
|
|
|
|
|
# ── Sentinel-1 SAR (Planetary Computer STAC → signed COG template) ────────
|
|
|
|
def test_sign_cog_url_appends_token():
|
|
# PC returns the token pre-encoded as a query string; append verbatim.
|
|
assert sign_cog_url("https://blob.example/x.tif", "st=s&se=e&sig=x%3D") == \
|
|
"https://blob.example/x.tif?st=s&se=e&sig=x%3D"
|
|
# Existing query string → append with &
|
|
assert sign_cog_url("https://blob.example/x.tif?foo=1", "st=s&sig=x") == \
|
|
"https://blob.example/x.tif?foo=1&st=s&sig=x"
|
|
|
|
|
|
def test_sentinel1_tile_url_contains_titiler_rescale_and_cfastie():
|
|
signed = "https://blob.example/x.tif?token=secret"
|
|
url = sentinel1_tile_url(signed)
|
|
assert url.startswith(TITILER_COG_TILES + "?")
|
|
assert "WebMercatorQuad/{z}/{x}/{y}?" in url
|
|
assert "url=https%3A%2F%2Fblob.example%2Fx.tif%3Ftoken%3Dsecret" in url
|
|
assert "rescale=0%2C500" in url
|
|
assert "colormap_name=cfastie" in url
|
|
|
|
|
|
def test_sentinel1_tile_url_is_same_origin_relative():
|
|
# Self-hosted TiTiler: the browser must hit the Pi's nginx vhost, not
|
|
# titiler.xyz or a raw host:port. The template is a root-relative path.
|
|
url = sentinel1_tile_url("https://blob.example/x.tif")
|
|
assert url.startswith("/titiler/cog/tiles/WebMercatorQuad/")
|
|
assert "://" not in url
|
|
assert "titiler.xyz" not in url
|
|
|
|
|
|
def _stac_feature(assets: dict) -> dict:
|
|
return {
|
|
"type": "Feature",
|
|
"id": "S1A_IW_GRDH_1SDV_20240820T000000",
|
|
"properties": {"datetime": "2024-08-20T00:00:00Z"},
|
|
"assets": assets,
|
|
}
|
|
|
|
|
|
def test_fetch_sentinel1_vv_signed_tile_url(monkeypatch):
|
|
import asyncio
|
|
from live_layers import fetch_sentinel1, _cache
|
|
|
|
calls = []
|
|
|
|
async def fake_post(url, json=None, headers=None):
|
|
calls.append(("post", url, json))
|
|
return {"features": [_stac_feature({
|
|
"vv": {"href": "https://blob.example/grd-vv.tif"},
|
|
})]}
|
|
|
|
async def fake_get(url, params=None, headers=None):
|
|
calls.append(("get", url))
|
|
return {"token": "sig=abc123"}
|
|
|
|
monkeypatch.setattr("live_layers._post_json", fake_post)
|
|
monkeypatch.setattr("live_layers._get_json", fake_get)
|
|
_cache.clear()
|
|
|
|
out = asyncio.run(fetch_sentinel1("-80,35,-79,36"))
|
|
assert out["id"] == "sentinel-1-sar"
|
|
assert out["kind"] == "raster"
|
|
assert out["polarization"] == "vv"
|
|
assert out["opacity"] == 0.8
|
|
assert out["itemId"].startswith("S1A")
|
|
assert out["attribution"] == SENTINEL1_ATTRIBUTION
|
|
assert "WebMercatorQuad/{z}/{x}/{y}?" in out["tileUrl"]
|
|
assert "rescale=0%2C500" in out["tileUrl"]
|
|
assert "colormap_name=cfastie" in out["tileUrl"]
|
|
# SAS token "sig=abc123" is appended top-level, then the whole COG URL is
|
|
# percent-encoded again as a query param (=> sig%3Dabc123).
|
|
assert "sig%3Dabc123" in out["tileUrl"]
|
|
# STAC search payload shape
|
|
post_url, post_json = calls[0][1], calls[0][2]
|
|
assert post_url.endswith("/api/stac/v1/search")
|
|
assert post_json["collections"] == ["sentinel-1-grd"]
|
|
assert post_json["limit"] >= 1
|
|
assert post_json["sortby"][0]["direction"] == "desc"
|
|
assert "bbox" in out
|
|
|
|
|
|
def test_fetch_sentinel1_uses_hh_when_vv_missing(monkeypatch):
|
|
import asyncio
|
|
from live_layers import fetch_sentinel1, _cache
|
|
|
|
async def fake_post(url, json=None, headers=None):
|
|
return {"features": [_stac_feature({
|
|
"hh": {"href": "https://blob.example/grd-hh.tif"},
|
|
})]}
|
|
|
|
async def fake_get(url, params=None, headers=None):
|
|
return {"token": "tok"}
|
|
|
|
monkeypatch.setattr("live_layers._post_json", fake_post)
|
|
monkeypatch.setattr("live_layers._get_json", fake_get)
|
|
_cache.clear()
|
|
|
|
out = asyncio.run(fetch_sentinel1("-80,35,-79,36"))
|
|
assert out["polarization"] == "hh"
|
|
assert "url=https%3A%2F%2Fblob.example%2Fgrd-hh.tif" in out["tileUrl"]
|
|
|
|
|
|
def test_fetch_sentinel1_none_on_empty_features(monkeypatch):
|
|
import asyncio
|
|
from live_layers import fetch_sentinel1, _cache
|
|
|
|
async def fake_post(url, json=None, headers=None):
|
|
return {"features": []}
|
|
|
|
monkeypatch.setattr("live_layers._post_json", fake_post)
|
|
_cache.clear()
|
|
|
|
assert asyncio.run(fetch_sentinel1("-80,35,-79,36")) is None
|
|
|
|
|
|
def test_fetch_sentinel1_none_when_no_vv_or_hh(monkeypatch):
|
|
import asyncio
|
|
from live_layers import fetch_sentinel1, _cache
|
|
|
|
async def fake_post(url, json=None, headers=None):
|
|
return {"features": [_stac_feature({"thumbnail": {"href": "https://x"}})]}
|
|
|
|
monkeypatch.setattr("live_layers._post_json", fake_post)
|
|
_cache.clear()
|
|
|
|
assert asyncio.run(fetch_sentinel1("-80,35,-79,36")) is None
|
|
|
|
|
|
def test_pick_sentinel_feature_prefers_scene_covering_center():
|
|
features = [
|
|
{"id": "far", "bbox": [10.0, 10.0, 12.0, 12.0]},
|
|
{"id": "cover", "bbox": [-80.5, 34.5, -78.5, 36.5]},
|
|
{"id": "also-far", "bbox": [-10.0, 0.0, -8.0, 2.0]},
|
|
]
|
|
picked = pick_sentinel_feature(features, -79.5, 35.5)
|
|
assert picked["id"] == "cover"
|
|
|
|
|
|
def test_pick_sentinel_feature_falls_back_to_first_when_none_cover():
|
|
features = [
|
|
{"id": "a", "bbox": [10.0, 10.0, 12.0, 12.0]},
|
|
{"id": "b", "bbox": [20.0, 20.0, 22.0, 22.0]},
|
|
]
|
|
assert pick_sentinel_feature(features, -79.5, 35.5)["id"] == "a"
|
|
assert pick_sentinel_feature([], -79.5, 35.5) is None
|