"""Normalisation des lignes CSV FIRMS -> objets du contrat /api/fires.""" from sources import ( _confidence_modis, _confidence_viirs, normalize_firms_rows, ) VIIRS_ROW = { "latitude": "43.30", "longitude": "5.40", "bright_ti4": "312.5", "scan": "0.4", "track": "0.4", "acq_date": "2026-07-27", "acq_time": "0137", "satellite": "N20", "confidence": "nominal", "version": "2.0NRT", "bright_ti5": "280.1", "frp": "12.3", "daynight": "N", } MODIS_ROW = { "latitude": "44.85", "longitude": "-0.57", "brightness": "305.2", "scan": "1.0", "track": "1.0", "acq_date": "2026-07-27", "acq_time": "1045", "satellite": "Aqua", "instrument": "MODIS", "confidence": "85", "version": "6.1NRT", "bright_t31": "290.0", "frp": "8.1", "daynight": "D", } def test_viirs_normalisation(): out = normalize_firms_rows([VIIRS_ROW], "NASA FIRMS VIIRS NOAA-20 (Europe 24 h)") assert len(out) == 1 f = out[0] assert f["lat"] == 43.30 and f["lon"] == 5.40 assert f["confidence"] == "nominal" assert f["frp"] == 12.3 assert f["brightness"] == 312.5 assert f["acquired_at"] == "2026-07-27T01:37:00Z" assert f["daynight"] == "N" assert f["type"] == "thermal_anomaly" assert f["instrument"] == "VIIRS" assert f["id"] and isinstance(f["id"], str) def test_modis_normalisation(): out = normalize_firms_rows([MODIS_ROW], "NASA FIRMS MODIS (Europe 24 h)") assert len(out) == 1 f = out[0] assert f["instrument"] == "MODIS" assert f["confidence"] == "high" # 85 >= 80 assert f["brightness"] == 305.2 assert f["acquired_at"] == "2026-07-27T10:45:00Z" def test_confidence_mappings(): assert _confidence_viirs("l") == "low" assert _confidence_viirs("low") == "low" assert _confidence_viirs("N") == "nominal" assert _confidence_viirs("High") == "high" assert _confidence_viirs("") is None assert _confidence_modis("10") == "low" assert _confidence_modis("50") == "nominal" assert _confidence_modis("95") == "high" assert _confidence_modis("abc") is None def test_lignes_invalides_ignorees(): base = dict(VIIRS_ROW) mauvaises = [ {**base, "latitude": "abc"}, {**base, "longitude": ""}, {**base, "latitude": "95.0"}, {**base, "longitude": "200.0"}, {**base, "acq_date": "pas-une-date"}, {**base, "acq_time": "99999"}, ] assert normalize_firms_rows(mauvaises, "test") == [] def test_hors_france_filtre(): madrid = {**VIIRS_ROW, "latitude": "40.42", "longitude": "-3.70"} assert normalize_firms_rows([madrid], "test") == [] def test_corse_conservee(): ajaccio = {**VIIRS_ROW, "latitude": "41.93", "longitude": "8.74"} assert len(normalize_firms_rows([ajaccio], "test")) == 1 def test_frp_manquant(): row = {**VIIRS_ROW, "frp": ""} out = normalize_firms_rows([row], "test") assert out[0]["frp"] is None