Skip to content

Commit

Permalink
chore: merge branch '191-implement-neighbourhood-working-for-any-geod…
Browse files Browse the repository at this point in the history
…ataframe-with-regions' of https://github.com/srai-lab/srai into 191-implement-neighbourhood-working-for-any-geodataframe-with-regions
  • Loading branch information
RaczeQ committed Mar 9, 2023
2 parents cb397cc + 5ad9053 commit 81b6325
Show file tree
Hide file tree
Showing 3 changed files with 88 additions and 53 deletions.
10 changes: 5 additions & 5 deletions srai/regionizers/h3_regionizer.py
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand Down
105 changes: 65 additions & 40 deletions tests/regionizers/test_administrative_boundary_regionizer.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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:
Expand All @@ -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",
[
Expand All @@ -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
Expand All @@ -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
Expand All @@ -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
26 changes: 18 additions & 8 deletions tests/regionizers/test_h3_regionizer.py
Original file line number Diff line number Diff line change
Expand Up @@ -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

0 comments on commit 81b6325

Please sign in to comment.