osint-dashboard/tests/test_live_layers.py

"""Unit tests for live map-layer mappers (aircraft, trains, AIS, WFIGS, Caltrans)."""

from live_layers import (
    MARKER_FIELDS,
    bbox_center_radius_nm,
    filter_points_bbox,
    parse_bbox,
    rainviewer_tile_url,
    to_marker,
    transform_adsb_lol,
    transform_ais_frame,
    transform_amtraker,
    transform_nhc_storms,
    transform_wfigs_incidents,
)

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()