Merge branch 'main' into 191-implement-neighbourhood-working-for-any-…

…geodataframe-with-regions

srai/regionizers/h3_regionizer.py

@@ -84,11 +84,11 @@ def transform(self, gdf: gpd.GeoDataFrame) -> gpd.GeoDataFrame:
         gdf_h3 = self._gdf_from_h3_indexes(h3_indexes)
 
         # there may be too many cells because of too big buffer
-        gdf_h3_clipped = (
-            gdf_h3.sjoin(gdf_exploded[["geometry"]]).drop(columns="index_right").drop_duplicates()
-            if self.buffer
-            else gdf_h3
-        )
+        if self.buffer:
+            gdf_h3_clipped = gdf_h3.sjoin(gdf_exploded[["geometry"]]).drop(columns="index_right")
+            gdf_h3_clipped = gdf_h3_clipped[~gdf_h3_clipped.index.duplicated(keep="first")]
+        else:
+            gdf_h3_clipped = gdf_h3
 
         gdf_h3_clipped.index.name = REGIONS_INDEX
 

tests/regionizers/test_administrative_boundary_regionizer.py

@@ -4,6 +4,8 @@
 
 import geopandas as gpd
 import pytest
+from overpass import API
+from pytest_mock import MockerFixture
 from shapely.geometry import Point, box
 
 from srai.regionizers import AdministrativeBoundaryRegionizer
@@ -36,7 +38,6 @@
 def test_admin_level(
     admin_level: int,
     expectation: Any,
-    request: Any,
 ) -> None:
     """Test checks if illegal admin_level is disallowed."""
     with expectation:
@@ -57,6 +58,17 @@ def test_no_crs_gdf_value_error(gdf_no_crs) -> None:  # type: ignore
         abr.transform(gdf=gdf_no_crs)
 
 
+@pytest.fixture  # type: ignore
+def mock_overpass_api(mocker: MockerFixture) -> None:
+    """Mock overpass API."""
+    mocker.patch.object(API, "get", return_value={"elements": [{"type": "relation", "id": 2137}]})
+
+    geocoded_gdf = gpd.GeoDataFrame(
+        {"geometry": [box(minx=0, miny=0, maxx=1, maxy=1)]}, crs=WGS84_CRS
+    )
+    mocker.patch("osmnx.geocode_to_gdf", return_value=geocoded_gdf)
+
+
 @pytest.mark.parametrize(  # type: ignore
     "toposimplify",
     [
@@ -68,24 +80,17 @@ def test_no_crs_gdf_value_error(gdf_no_crs) -> None:  # type: ignore
         (0),
     ],
 )
-def test_single_points(toposimplify: Union[bool, float]) -> None:
-    """Test checks if single points work."""
-    country_points_gdf = gpd.GeoDataFrame(
-        {
-            "geometry": [
-                Point(19.24530, 52.21614),  # Poland
-                Point(10.48674, 51.38001),  # Germany
-                Point(14.74938, 47.69628),  # Austria
-                Point(15.00989, 49.79905),  # Czechia
-            ]
-        },
-        crs=WGS84_CRS,
-    )
+def test_empty_region_full_bounding_box(toposimplify: Union[bool, float], request: Any) -> None:
+    """Test checks if empty region fills required bounding box."""
+    request.getfixturevalue("mock_overpass_api")
+    request_bbox = box(minx=0, miny=0, maxx=2, maxy=2)
+    request_bbox_gdf = gpd.GeoDataFrame({"geometry": [request_bbox]}, crs=WGS84_CRS)
     abr = AdministrativeBoundaryRegionizer(
-        admin_level=2, return_empty_region=False, clip_regions=False, toposimplify=toposimplify
+        admin_level=4, return_empty_region=True, toposimplify=toposimplify
     )
-    countries_result_gdf = abr.transform(gdf=country_points_gdf)
-    assert list(countries_result_gdf.index) == ["Poland", "Germany", "Austria", "Czechia"]
+    result_gdf = abr.transform(gdf=request_bbox_gdf)
+    assert _merge_disjointed_gdf_geometries(result_gdf).difference(request_bbox).is_empty
+    assert "EMPTY" in result_gdf.index
 
 
 @pytest.mark.parametrize(  # type: ignore
@@ -99,20 +104,41 @@ def test_single_points(toposimplify: Union[bool, float]) -> None:
         (0),
     ],
 )
-def test_empty_region_full_bounding_box(toposimplify: Union[bool, float]) -> None:
-    """Test checks if empty region fills required bounding box."""
-    madagascar_bbox = box(
-        minx=43.2541870461, miny=-25.6014344215, maxx=50.4765368996, maxy=-12.0405567359
-    )
-    madagascar_bbox_gdf = gpd.GeoDataFrame({"geometry": [madagascar_bbox]}, crs=WGS84_CRS)
+def test_no_empty_region_full_bounding_box(toposimplify: Union[bool, float], request: Any) -> None:
+    """Test checks if no empty region is generated when not needed."""
+    request.getfixturevalue("mock_overpass_api")
+    request_bbox = box(minx=0, miny=0, maxx=1, maxy=1)
+    request_bbox_gdf = gpd.GeoDataFrame({"geometry": [request_bbox]}, crs=WGS84_CRS)
     abr = AdministrativeBoundaryRegionizer(
-        admin_level=4, return_empty_region=True, toposimplify=toposimplify
+        admin_level=2, return_empty_region=True, toposimplify=toposimplify
     )
-    madagascar_result_gdf = abr.transform(gdf=madagascar_bbox_gdf)
-    assert (
-        _merge_disjointed_gdf_geometries(madagascar_result_gdf).difference(madagascar_bbox).is_empty
+    result_gdf = abr.transform(gdf=request_bbox_gdf)
+    assert _merge_disjointed_gdf_geometries(result_gdf).difference(request_bbox).is_empty
+    assert "EMPTY" not in result_gdf.index
+
+
+@pytest.mark.parametrize(  # type: ignore
+    "toposimplify",
+    [
+        (True),
+        (0.0001),
+        (0.001),
+        (0.01),
+        (False),
+        (0),
+    ],
+)
+def test_points_in_result(toposimplify: Union[bool, float], request: Any) -> None:
+    """Test checks case when points are in a requested region."""
+    request.getfixturevalue("mock_overpass_api")
+    request_gdf = gpd.GeoDataFrame({"geometry": [Point(0.5, 0.5)]}, crs=WGS84_CRS)
+
+    abr = AdministrativeBoundaryRegionizer(
+        admin_level=2, return_empty_region=False, clip_regions=False, toposimplify=toposimplify
     )
-    assert "EMPTY" in madagascar_result_gdf.index
+
+    result_gdf = abr.transform(gdf=request_gdf)
+    assert request_gdf.geometry[0].within(result_gdf.geometry[0])
 
 
 @pytest.mark.parametrize(  # type: ignore
@@ -126,18 +152,17 @@ def test_empty_region_full_bounding_box(toposimplify: Union[bool, float]) -> Non
         (0),
     ],
 )
-def test_no_empty_region_full_bounding_box(toposimplify: Union[bool, float]) -> None:
-    """Test checks if no empty region is generated when not needed."""
-    asia_bbox = box(
-        minx=69.73278412113555,
-        miny=24.988848422533074,
-        maxx=88.50230949587835,
-        maxy=34.846427760404225,
+def test_toposimplify_on_real_data(toposimplify: Union[float, bool]) -> None:
+    """Test if toposimplify usage covers an entire region."""
+    madagascar_bbox = box(
+        minx=43.2541870461, miny=-25.6014344215, maxx=50.4765368996, maxy=-12.0405567359
     )
-    asia_bbox_gdf = gpd.GeoDataFrame({"geometry": [asia_bbox]}, crs=WGS84_CRS)
+    madagascar_bbox_gdf = gpd.GeoDataFrame({"geometry": [madagascar_bbox]}, crs=WGS84_CRS)
+
     abr = AdministrativeBoundaryRegionizer(
-        admin_level=2, return_empty_region=True, toposimplify=toposimplify
+        admin_level=4, return_empty_region=True, toposimplify=toposimplify
+    )
+    madagascar_result_gdf = abr.transform(gdf=madagascar_bbox_gdf)
+    assert (
+        _merge_disjointed_gdf_geometries(madagascar_result_gdf).difference(madagascar_bbox).is_empty
     )
-    asia_result_gdf = abr.transform(gdf=asia_bbox_gdf)
-    assert _merge_disjointed_gdf_geometries(asia_result_gdf).difference(asia_bbox).is_empty
-    assert "EMPTY" not in asia_result_gdf.index

tests/regionizers/test_h3_regionizer.py

@@ -26,29 +26,39 @@ def expected_h3_indexes() -> List[str]:
     ]
 
 
+@pytest.fixture  # type: ignore
+def expected_unbuffered_h3_indexes() -> List[str]:
+    """Get expected h3 index for the unbuffered case."""
+    return [
+        "83754efffffffff",
+    ]
+
+
 @pytest.mark.parametrize(  # type: ignore
-    "gdf_fixture,expected_h3_indexes_fixture,resolution,expectation",
+    "gdf_fixture,expected_h3_indexes_fixture,resolution,buffer,expectation",
     [
-        ("gdf_polygons", "expected_h3_indexes", H3_RESOLUTION, does_not_raise()),
-        ("gdf_multipolygon", "expected_h3_indexes", H3_RESOLUTION, does_not_raise()),
-        ("gdf_empty", "expected_h3_indexes", H3_RESOLUTION, pytest.raises(AttributeError)),
-        ("gdf_polygons", "expected_h3_indexes", -1, pytest.raises(ValueError)),
-        ("gdf_polygons", "expected_h3_indexes", 16, pytest.raises(ValueError)),
-        ("gdf_no_crs", "expected_h3_indexes", H3_RESOLUTION, pytest.raises(ValueError)),
+        ("gdf_polygons", "expected_h3_indexes", H3_RESOLUTION, True, does_not_raise()),
+        ("gdf_polygons", "expected_unbuffered_h3_indexes", H3_RESOLUTION, False, does_not_raise()),
+        ("gdf_multipolygon", "expected_h3_indexes", H3_RESOLUTION, True, does_not_raise()),
+        ("gdf_empty", "expected_h3_indexes", H3_RESOLUTION, True, pytest.raises(AttributeError)),
+        ("gdf_polygons", "expected_h3_indexes", -1, True, pytest.raises(ValueError)),
+        ("gdf_polygons", "expected_h3_indexes", 16, True, pytest.raises(ValueError)),
+        ("gdf_no_crs", "expected_h3_indexes", H3_RESOLUTION, True, pytest.raises(ValueError)),
     ],
 )
 def test_transform(
     gdf_fixture: str,
     expected_h3_indexes_fixture: str,
     resolution: int,
+    buffer: bool,
     expectation: Any,
     request: Any,
 ) -> None:
     """Test transform of H3Regionizer."""
     gdf: gpd.GeoDataFrame = request.getfixturevalue(gdf_fixture)
     h3_indexes: List[str] = request.getfixturevalue(expected_h3_indexes_fixture)
     with expectation:
-        gdf_h3 = H3Regionizer(resolution).transform(gdf)
+        gdf_h3 = H3Regionizer(resolution, buffer=buffer).transform(gdf)
 
         ut.assertCountEqual(first=gdf_h3.index.to_list(), second=h3_indexes)
         assert "geometry" in gdf_h3