Fixed tests and small logic tweak

Author: mdales

aoh/validation/validate_occurences.py

@@ -56,6 +56,7 @@ def process_species(
     clipped_points = gpd.sjoin(points_gdf, species_range, predicate='within', how='inner')
 
     pixel_set = set()
+    tested = 0
     with yg.read_raster(aoh_files[0]) as aoh:
         results = []
         for _, row in clipped_points.iterrows():
@@ -69,22 +70,36 @@ def process_species(
 
             value = aoh.read_array(pixel_x, pixel_y, 1, 1)
             results.append(value[0][0] > 0.0)
+            tested += 1
 
     # From Dahal et al: "Finally, we excluded species which had fewer than 10 point localities after
     # all the filters were applied."
-    if len(results) < 10:
+    if tested < 10:
         raise ValueError("Not enough occurrences")
 
     matches = len([x for x in results if x])
     point_prevalence = matches / len(results)
     model_prevalence = aoh_data['prevalence']
+
+    # From Dahal et al: "If the point prevalence exceeded model prevalence at
+    # species level, the AOH maps performed better than random,
+    # otherwise they were no better than random."
+    #
+    # However, note that this means if you have a point prevalence of 1.0 (all
+    # points match) and a model prevalence of 1.0 (range and AOH match, which
+    # under the IUCN method is the preferred fallback if we have zero on either
+    # elevation filtering or habitat filtering), then that would still be marked
+    # as an outlier, (as 1.0 is not exceeding 1.0) which seems wrong, so I'm
+    # special casing that.
+    is_outlier = (point_prevalence != 1.0) and (point_prevalence < model_prevalence)
+
     return (
         taxon_id,
         len(results),
         matches,
         point_prevalence,
         model_prevalence,
-        point_prevalence <= model_prevalence
+        is_outlier,
     )
 
 def process_species_wrapper(
@@ -127,8 +142,8 @@ def validate_occurrences(
         "model prevalence",
         "outlier",
     ])
-    summary = summary[summary.outlier == True]
-    summary.to_csv(output_path, index=False)
+    outliers = summary[summary.outlier is True]
+    outliers.to_csv(output_path, index=False)
 
 def main() -> None:
     parser = argparse.ArgumentParser(description="Validate map prevalence.")

tests/test_occurences.py

@@ -1,4 +1,5 @@
 import json
+import math
 import tempfile
 from pathlib import Path
 
@@ -9,16 +10,22 @@
 
 from aoh.validation.validate_occurences import process_species
 
-def test_empty_species_list() -> None:
-    df = pd.DataFrame([], columns=['iucn_taxon_id', 'decimalLatitude', 'decimalLongitude'])
-    res = process_species(Path("/some/aohs"), Path("/some/aohs"), df)
-    assert res is None
-
-def generate_faux_aoh(filename: Path, shape: Polygon | None = None) -> None:
-    shapes = [
-        shape if shape is not None else Polygon([(0, 0), (0, 10), (10, 10), (10, 0)])
-    ]
-
+def generate_occurrence_cluster(
+    latitude: float,
+    longitude: float,
+    count: int,
+    radius: float,
+) -> list[tuple[float,float]]:
+    res = [(latitude, longitude)]
+    rotation = (math.pi * 2) / (count - 1)
+    for i in range(count - 1):
+        angle = i * rotation
+        x = radius * math.cos(angle)
+        y = radius * math.sin(angle)
+        res.append((latitude + y, longitude + x))
+    return res
+
+def geojson_of_shaps(shapes):
     features = []
     for geom in shapes:
         feature = {
@@ -32,59 +39,128 @@ def generate_faux_aoh(filename: Path, shape: Polygon | None = None) -> None:
         "type": "FeatureCollection",
         "features": features
     }
+    return geojson
+
+def generate_faux_aoh(filename: Path, aoh_radius:float=5.0, range_radius:float=10.0) -> None:
+    aoh_shapes = [
+        Polygon([
+            (-aoh_radius, aoh_radius),
+            (aoh_radius, aoh_radius),
+            (aoh_radius, -aoh_radius),
+            (-aoh_radius, -aoh_radius)
+        ])
+    ]
+    aoh_area = sum(x.area for x in aoh_shapes)
+
+    range_shapes = [
+        Polygon([
+            (-range_radius, range_radius),
+            (range_radius, range_radius),
+            (range_radius, -range_radius),
+            (-range_radius, -range_radius)
+        ])
+    ]
+    range_area = sum(x.area for x in range_shapes)
+
+    assert aoh_area <= range_area
 
     geojson_path = filename.with_suffix('.geojson')
     with open(geojson_path, 'w', encoding="UTF-8") as f:
-        json.dump(geojson, f, indent=2)
+        json.dump(geojson_of_shaps(range_shapes), f, indent=2)
 
     json_path = filename.with_suffix('.json')
     with open(json_path, 'w', encoding='utf-8') as f:
-        json.dump({'prevalence': 1.0}, f)
+        json.dump({'prevalence': aoh_area / range_area}, f)
 
-    with yg.read_shape(geojson_path, ("epsg:4326", (1.0, -1.0))) as shape_layer:
-        shape_layer.to_geotiff(filename)
-
-@pytest.mark.parametrize("taxon_id,latitude,longitude,expected",[
-    (42, 5.0, 5.0, True),
-    (42, 12.0, 12.0, False),
+    with tempfile.TemporaryDirectory() as tmpdir:
+        aoh_geojson = Path(tmpdir) / "test.geojson"
+        with open(aoh_geojson, 'w', encoding="UTF-8") as f:
+            json.dump(geojson_of_shaps(aoh_shapes), f, indent=2)
+        with yg.read_shape(aoh_geojson, ("epsg:4326", (0.1, -0.1))) as shape_layer:
+            shape_layer.to_geotiff(filename)
+
+@pytest.mark.parametrize("taxon_id,latitude,longitude,expected_reject,expected_outlier",[
+    (42, 0.0, 0.0, False, False), # all in AoH
+    (42, 0.0, 4.0, False, False), # Most in AOH, a few in range
+    (42, 0.0, 6.5, False, True), # Most in range, a few in AOH
+    (42, 0.0, 7.5, False, True),  # all in range but not AOH
+    (42, 0.0, 11.0, True, None),  # most out of range
+    (42, 0.0, 20.0, True, None), # all out of range
 ])
-def test_simple_match(taxon_id: int, latitude: float, longitude: float, expected: bool) -> None:
+def test_simple_match_in_out_range(
+    taxon_id: int,
+    latitude: float,
+    longitude: float,
+    expected_reject: bool,
+    expected_outlier: bool,
+) -> None:
     with tempfile.TemporaryDirectory() as tmpdir:
         tmpdir_path = Path(tmpdir)
 
         for test_id in [41, 42, 43]:
             aoh_path = tmpdir_path / f"{test_id}.tif"
             generate_faux_aoh(aoh_path)
 
+        occurences = generate_occurrence_cluster(latitude, longitude, 20, 2.0)
         df = pd.DataFrame(
-            [(taxon_id, latitude, longitude)],
+            [(taxon_id, lat, lng) for (lat, lng) in occurences],
             columns=['iucn_taxon_id', 'decimalLatitude', 'decimalLongitude']
         )
-        res = process_species(tmpdir_path, tmpdir_path, df)
 
-        id_no, results, matches, point_prev, model_prev, outlier = res
-        assert id_no == taxon_id
-        assert results == 1
-        assert matches == (1 if expected else 0)
-        assert model_prev == 1.0
-        assert outlier == expected
-
-def test_no_aoh_found(taxon_id: int, latitude: float, longitude: float, expected: bool) -> None:
+        if not expected_reject:
+            res = process_species(tmpdir_path, tmpdir_path, df)
+            id_no, results, _matches, _point_prev, _model_prev, outlier = res
+            print(res)
+            assert id_no == taxon_id
+            assert results == len(occurences)
+            assert outlier == expected_outlier
+        else:
+            with pytest.raises(ValueError):
+                _ = process_species(tmpdir_path, tmpdir_path, df)
+
+@pytest.mark.parametrize("taxon_id,latitude,longitude,expected_prev,expected_reject,expected_outlier",[
+    (42, 0.0, 0.0, 1.0, False, False), # all in AoH
+    # (42, 0.0, 4.0, False, False), # Most in AOH, a few in range
+    # (42, 0.0, 6.5, False, True), # Most in range, a few in AOH
+    # (42, 0.0, 7.5, False, True),  # all in range but not AOH
+    # (42, 0.0, 11.0, True, None),  # most out of range
+    (42, 0.0, 20.0, 0.0, True, None), # all out of range
+])
+def test_model_prevalence_of_one(
+    taxon_id: int,
+    latitude: float,
+    longitude: float,
+    expected_prev: float,
+    expected_reject: bool,
+    expected_outlier: bool,
+) -> None:
     with tempfile.TemporaryDirectory() as tmpdir:
         tmpdir_path = Path(tmpdir)
 
         for test_id in [41, 42, 43]:
             aoh_path = tmpdir_path / f"{test_id}.tif"
-            generate_faux_aoh(aoh_path)
+            generate_faux_aoh(aoh_path, aoh_radius=5.0, range_radius=5.0)
 
+        occurences = generate_occurrence_cluster(latitude, longitude, 20, 2.0)
         df = pd.DataFrame(
-            [(40, 5.0, 5.0)],
+            [(taxon_id, lat, lng) for (lat, lng) in occurences],
             columns=['iucn_taxon_id', 'decimalLatitude', 'decimalLongitude']
         )
-        with pytest.raises(FileNotFoundError):
-            _ = process_species(tmpdir_path, tmpdir_path, df)
 
-def test_multiple_match() -> None:
+        if not expected_reject:
+            res = process_species(tmpdir_path, tmpdir_path, df)
+            id_no, results, _matches, point_prev, model_prev, outlier = res
+            print(res)
+            assert id_no == taxon_id
+            assert results == len(occurences)
+            assert point_prev == expected_prev
+            assert model_prev == 1.0
+            assert outlier == expected_outlier
+        else:
+            with pytest.raises(ValueError):
+                _ = process_species(tmpdir_path, tmpdir_path, df)
+
+def test_no_aoh_found() -> None:
     with tempfile.TemporaryDirectory() as tmpdir:
         tmpdir_path = Path(tmpdir)
 
@@ -93,17 +169,11 @@ def test_multiple_match() -> None:
             generate_faux_aoh(aoh_path)
 
         df = pd.DataFrame(
-            [
-                (42, 5.0, 5.0, True),
-                (42, 12.0, 12.0, False),
-            ],
-            columns=['iucn_taxon_id', 'decimalLatitude', 'decimalLongitude', 'expected']
+            [(40, 5.0, 5.0)],
+            columns=['iucn_taxon_id', 'decimalLatitude', 'decimalLongitude']
         )
-
-        res = process_species(tmpdir_path, tmpdir_path, df)
-
-        assert len(res) == len(df)
-        assert (res.occurence == res.expected).all()
+        with pytest.raises(FileNotFoundError):
+            _ = process_species(tmpdir_path, tmpdir_path, df)
 
 def test_too_many_ids() -> None:
     df = pd.DataFrame(
@@ -118,31 +188,23 @@ def test_too_many_ids() -> None:
     with pytest.raises(ValueError):
         _ = process_species(Path("/some/aohs"), Path("/some/aohs"), df)
 
-@pytest.mark.parametrize("taxon_id,latitude,longitude,expected",[
-    (42, 5.0, 5.0, True),
-    (42, -5.0, -5.0, True),
-    (42, 5.0, -5.0, False),
-    (42, -5.0, 5.0, False),
-    (40, 5.0, 5.0, False),
-])
-def test_find_seasonal(taxon_id: int, latitude: float, longitude: float, expected: bool) -> None:
+def test_find_seasonal() -> None:
     with tempfile.TemporaryDirectory() as tmpdir:
         tmpdir_path = Path(tmpdir)
 
-        for season, shape in [
-            ('breeding', Polygon([(0, 0), (0, 10), (10, 10), (10, 0)])),
-            ('nonbreeding', Polygon([(0, 0), (0, -10), (-10, -10), (-10, 0)])),
-        ]:
+        for season in ['breeding', 'nonbreeding']:
             aoh_path = tmpdir_path / f"42_{season}.tif"
-            generate_faux_aoh(aoh_path, shape)
+            generate_faux_aoh(aoh_path)
 
         df = pd.DataFrame(
-            [(taxon_id, latitude, longitude)],
+            [(42, 5.0, 5.0)],
             columns=['iucn_taxon_id', 'decimalLatitude', 'decimalLongitude']
         )
 
-        res = process_species(tmpdir_path, tmpdir_path, df)
+        with pytest.raises(RuntimeError):
+            _ = process_species(tmpdir_path, tmpdir_path, df)
 
-        assert len(res) == len(df)
-        occurence = res.occurence[0]
-        assert occurence == expected
+def test_empty_species_list() -> None:
+    df = pd.DataFrame([], columns=['iucn_taxon_id', 'decimalLatitude', 'decimalLongitude'])
+    with pytest.raises(ValueError):
+        _ = process_species(Path("/some/aohs"), Path("/some/aohs"), df)