import math from collections.abc import Sequence import numpy as np import pandas as pd import shapely from shapely.geometry import GeometryCollection, Point, Polygon, box import geopandas import geopandas._compat as compat from geopandas import ( GeoDataFrame, GeoSeries, points_from_xy, read_file, sjoin, sjoin_nearest, ) import pytest from geopandas.testing import assert_geodataframe_equal, assert_geoseries_equal from pandas.testing import assert_frame_equal, assert_index_equal, assert_series_equal @pytest.fixture() def dfs(request): polys1 = GeoSeries( [ Polygon([(0, 0), (5, 0), (5, 5), (0, 5)]), Polygon([(5, 5), (6, 5), (6, 6), (5, 6)]), Polygon([(6, 0), (9, 0), (9, 3), (6, 3)]), ] ) polys2 = GeoSeries( [ Polygon([(1, 1), (4, 1), (4, 4), (1, 4)]), Polygon([(4, 4), (7, 4), (7, 7), (4, 7)]), Polygon([(7, 7), (10, 7), (10, 10), (7, 10)]), ] ) df1 = GeoDataFrame({"geometry": polys1, "df1": [0, 1, 2]}) df2 = GeoDataFrame({"geometry": polys2, "df2": [3, 4, 5]}) if request.param == "string-index": df1.index = ["a", "b", "c"] df2.index = ["d", "e", "f"] if request.param == "named-index": df1.index.name = "df1_ix" df2.index.name = "df2_ix" if request.param == "multi-index": i1 = ["a", "b", "c"] i2 = ["d", "e", "f"] df1 = df1.set_index([i1, i2]) df2 = df2.set_index([i2, i1]) if request.param == "named-multi-index": i1 = ["a", "b", "c"] i2 = ["d", "e", "f"] df1 = df1.set_index([i1, i2]) df2 = df2.set_index([i2, i1]) df1.index.names = ["df1_ix1", "df1_ix2"] df2.index.names = ["df2_ix1", "df2_ix2"] # construction expected frames expected = {} part1 = df1.copy().reset_index().rename(columns={"index": "index_left"}) part2 = ( df2.copy() .iloc[[0, 1, 1, 2]] .reset_index() .rename(columns={"index": "index_right"}) ) part1["_merge"] = [0, 1, 2] part2["_merge"] = [0, 0, 1, 3] exp = pd.merge(part1, part2, on="_merge", how="outer") expected["intersects"] = exp.drop("_merge", axis=1).copy() part1 = df1.copy().reset_index().rename(columns={"index": "index_left"}) part2 = df2.copy().reset_index().rename(columns={"index": "index_right"}) part1["_merge"] = [0, 1, 2] part2["_merge"] = [0, 3, 3] exp = pd.merge(part1, part2, on="_merge", how="outer") expected["contains"] = exp.drop("_merge", axis=1).copy() part1["_merge"] = [0, 1, 2] part2["_merge"] = [3, 1, 3] exp = pd.merge(part1, part2, on="_merge", how="outer") expected["within"] = exp.drop("_merge", axis=1).copy() return [request.param, df1, df2, expected] @pytest.fixture() def dfs_shared_attribute(): geo_left = [ Point(0, 0), Point(1, 1), Point(2, 2), Point(3, 3), Point(4, 4), Point(5, 5), Point(6, 6), Point(7, 7), ] geo_right = [ Point(0, 0), Point(1, 1), Point(2, 2), Point(3, 3), Point(4, 4), Point(5, 5), Point(6, 6), Point(7, 7), ] attr_tracker = ["A", "B", "C", "D", "E", "F", "G", "H"] left_gdf = geopandas.GeoDataFrame( { "geometry": geo_left, "attr_tracker": attr_tracker, "duplicate_column": [0, 1, 2, 3, 4, 5, 6, 7], "attr1": [True, True, True, True, True, True, True, True], "attr2": [True, True, True, True, True, True, True, True], } ) right_gdf = geopandas.GeoDataFrame( { "geometry": geo_right, "duplicate_column": [0, 1, 2, 3, 4, 5, 6, 7], "attr1": [True, True, False, False, True, True, False, False], "attr2": [True, True, False, False, False, False, False, False], } ) return left_gdf, right_gdf class TestSpatialJoin: @pytest.mark.parametrize( "how, lsuffix, rsuffix, expected_cols", [ ("left", "left", "right", {"col_left", "col_right", "index_right"}), ("inner", "left", "right", {"col_left", "col_right", "index_right"}), ("right", "left", "right", {"col_left", "col_right", "index_left"}), ("left", "lft", "rgt", {"col_lft", "col_rgt", "index_rgt"}), ("inner", "lft", "rgt", {"col_lft", "col_rgt", "index_rgt"}), ("right", "lft", "rgt", {"col_lft", "col_rgt", "index_lft"}), ], ) def test_suffixes(self, how: str, lsuffix: str, rsuffix: str, expected_cols): left = GeoDataFrame({"col": [1], "geometry": [Point(0, 0)]}) right = GeoDataFrame({"col": [1], "geometry": [Point(0, 0)]}) joined = sjoin(left, right, how=how, lsuffix=lsuffix, rsuffix=rsuffix) assert set(joined.columns) == expected_cols | {"geometry"} @pytest.mark.skipif(not compat.HAS_PYPROJ, reason="pyproj not available") @pytest.mark.parametrize("dfs", ["default-index", "string-index"], indirect=True) def test_crs_mismatch(self, dfs): index, df1, df2, expected = dfs df1.crs = "epsg:4326" with pytest.warns(UserWarning, match="CRS mismatch between the CRS"): sjoin(df1, df2) @pytest.mark.parametrize("dfs", ["default-index"], indirect=True) def test_unknown_kwargs(self, dfs): _, df1, df2, _ = dfs with pytest.raises( TypeError, match=r"sjoin\(\) got an unexpected keyword argument 'extra_param'", ): sjoin(df1, df2, extra_param="test") @pytest.mark.parametrize( "dfs", [ "default-index", "string-index", "named-index", "multi-index", "named-multi-index", ], indirect=True, ) @pytest.mark.parametrize("predicate", ["intersects", "contains", "within"]) def test_inner(self, predicate, dfs): index, df1, df2, expected = dfs res = sjoin(df1, df2, how="inner", predicate=predicate) exp = expected[predicate].dropna().copy() exp = exp.drop("geometry_y", axis=1).rename(columns={"geometry_x": "geometry"}) exp[["df1", "df2"]] = exp[["df1", "df2"]].astype("int64") if index == "default-index": exp[["index_left", "index_right"]] = exp[ ["index_left", "index_right"] ].astype("int64") if index == "named-index": exp[["df1_ix", "df2_ix"]] = exp[["df1_ix", "df2_ix"]].astype("int64") exp = exp.set_index("df1_ix") if index in ["default-index", "string-index"]: exp = exp.set_index("index_left") exp.index.name = None if index == "multi-index": exp = exp.set_index(["level_0_x", "level_1_x"]).rename( columns={"level_0_y": "index_right0", "level_1_y": "index_right1"} ) exp.index.names = df1.index.names if index == "named-multi-index": exp = exp.set_index(["df1_ix1", "df1_ix2"]) assert_frame_equal(res, exp) @pytest.mark.parametrize( "dfs", [ "default-index", "string-index", "named-index", "multi-index", "named-multi-index", ], indirect=True, ) @pytest.mark.parametrize("predicate", ["intersects", "contains", "within"]) def test_left(self, predicate, dfs): index, df1, df2, expected = dfs res = sjoin(df1, df2, how="left", predicate=predicate) if index in ["default-index", "string-index"]: exp = expected[predicate].dropna(subset=["index_left"]).copy() elif index == "named-index": exp = expected[predicate].dropna(subset=["df1_ix"]).copy() elif index == "multi-index": exp = expected[predicate].dropna(subset=["level_0_x"]).copy() elif index == "named-multi-index": exp = expected[predicate].dropna(subset=["df1_ix1"]).copy() exp = exp.drop("geometry_y", axis=1).rename(columns={"geometry_x": "geometry"}) exp["df1"] = exp["df1"].astype("int64") if index == "default-index": exp["index_left"] = exp["index_left"].astype("int64") # TODO: in result the dtype is object res["index_right"] = res["index_right"].astype(float) elif index == "named-index": exp[["df1_ix"]] = exp[["df1_ix"]].astype("int64") exp = exp.set_index("df1_ix") if index in ["default-index", "string-index"]: exp = exp.set_index("index_left") exp.index.name = None if index == "multi-index": exp = exp.set_index(["level_0_x", "level_1_x"]).rename( columns={"level_0_y": "index_right0", "level_1_y": "index_right1"} ) exp.index.names = df1.index.names if index == "named-multi-index": exp = exp.set_index(["df1_ix1", "df1_ix2"]) assert_frame_equal(res, exp) def test_empty_join(self): # Check joins resulting in empty gdfs. polygons = geopandas.GeoDataFrame( { "col2": [1, 2], "geometry": [ Polygon([(0, 0), (1, 0), (1, 1), (0, 1)]), Polygon([(1, 0), (2, 0), (2, 1), (1, 1)]), ], } ) not_in = geopandas.GeoDataFrame({"col1": [1], "geometry": [Point(-0.5, 0.5)]}) empty = sjoin(not_in, polygons, how="left", predicate="intersects") assert empty.index_right.isnull().all() empty = sjoin(not_in, polygons, how="right", predicate="intersects") assert empty.index_left.isnull().all() empty = sjoin(not_in, polygons, how="inner", predicate="intersects") assert empty.empty @pytest.mark.parametrize( "predicate", [ "contains", "contains_properly", "covered_by", "covers", "crosses", "intersects", "touches", "within", ], ) @pytest.mark.parametrize( "empty", [ GeoDataFrame(geometry=[GeometryCollection(), GeometryCollection()]), GeoDataFrame(geometry=GeoSeries()), ], ) def test_join_with_empty(self, predicate, empty): # Check joins with empty geometry columns/dataframes. polygons = geopandas.GeoDataFrame( { "col2": [1, 2], "geometry": [ Polygon([(0, 0), (1, 0), (1, 1), (0, 1)]), Polygon([(1, 0), (2, 0), (2, 1), (1, 1)]), ], } ) result = sjoin(empty, polygons, how="left", predicate=predicate) assert result.index_right.isnull().all() result = sjoin(empty, polygons, how="right", predicate=predicate) assert result.index_left.isnull().all() result = sjoin(empty, polygons, how="inner", predicate=predicate) assert result.empty @pytest.mark.parametrize("dfs", ["default-index", "string-index"], indirect=True) def test_sjoin_invalid_args(self, dfs): index, df1, df2, expected = dfs with pytest.raises(ValueError, match="'left_df' should be GeoDataFrame"): sjoin(df1.geometry, df2) with pytest.raises(ValueError, match="'right_df' should be GeoDataFrame"): sjoin(df1, df2.geometry) @pytest.mark.parametrize( "dfs", [ "default-index", "string-index", "named-index", "multi-index", "named-multi-index", ], indirect=True, ) @pytest.mark.parametrize("predicate", ["intersects", "contains", "within"]) def test_right(self, predicate, dfs): index, df1, df2, expected = dfs res = sjoin(df1, df2, how="right", predicate=predicate) if index in ["default-index", "string-index"]: exp = expected[predicate].dropna(subset=["index_right"]).copy() elif index == "named-index": exp = expected[predicate].dropna(subset=["df2_ix"]).copy() elif index == "multi-index": exp = expected[predicate].dropna(subset=["level_0_y"]).copy() elif index == "named-multi-index": exp = expected[predicate].dropna(subset=["df2_ix1"]).copy() exp = exp.drop("geometry_x", axis=1).rename(columns={"geometry_y": "geometry"}) exp["df2"] = exp["df2"].astype("int64") if index == "default-index": exp["index_right"] = exp["index_right"].astype("int64") res["index_left"] = res["index_left"].astype(float) elif index == "named-index": exp[["df2_ix"]] = exp[["df2_ix"]].astype("int64") exp = exp.set_index("df2_ix") if index in ["default-index", "string-index"]: exp = exp.set_index("index_right") exp = exp.reindex(columns=res.columns) exp.index.name = None if index == "multi-index": exp = exp.set_index(["level_0_y", "level_1_y"]).rename( columns={"level_0_x": "index_left0", "level_1_x": "index_left1"} ) exp.index.names = df2.index.names if index == "named-multi-index": exp = exp.set_index(["df2_ix1", "df2_ix2"]) if predicate == "within": exp = exp.sort_index() assert_frame_equal(res, exp, check_index_type=False) @pytest.mark.skipif(not compat.GEOS_GE_310, reason="`dwithin` requires GEOS 3.10") @pytest.mark.parametrize("how", ["inner"]) @pytest.mark.parametrize( "geo_left, geo_right, expected_left, expected_right, distance", [ ( # Distance is number, 2x1 [Point(0, 0), Point(1, 1)], [Point(1, 1)], [0, 1], [0, 0], math.sqrt(2), ), # Distance is number, 2x2 ( [Point(0, 0), Point(1, 1)], [Point(0, 0), Point(1, 1)], [0, 1, 0, 1], [0, 0, 1, 1], math.sqrt(2), ), # Distance is array, matches len(left) ( [Point(0, 0), Point(0, 0), Point(-1, -1)], [Point(1, 1)], [1, 2], [0, 0], [0, math.sqrt(2), math.sqrt(8)], ), # Distance is np.array, matches len(left), # inner join sorts the right GeoDataFrame ( [Point(0, 0), Point(0, 0), Point(-1, -1)], [Point(1, 1), Point(0.5, 0.5)], [1, 2, 1, 2], [1, 1, 0, 0], np.array([0, math.sqrt(2), math.sqrt(8)]), ), ], ) def test_sjoin_dwithin( self, geo_left, geo_right, expected_left: Sequence[int], expected_right: Sequence[int], distance, how, ): left = geopandas.GeoDataFrame({"geometry": geo_left}) right = geopandas.GeoDataFrame({"geometry": geo_right}) expected_gdf = left.iloc[expected_left].copy() expected_gdf["index_right"] = expected_right joined = sjoin(left, right, how=how, predicate="dwithin", distance=distance) assert_frame_equal(expected_gdf.sort_index(), joined.sort_index()) # GH3239 @pytest.mark.parametrize( "predicate", [ "contains", "contains_properly", "covered_by", "covers", "crosses", "intersects", "touches", "within", ], ) def test_sjoin_left_order(self, predicate): # a set of points in random order -> that order should be preserved # with a left join pts = GeoDataFrame( geometry=points_from_xy([0.1, 0.4, 0.3, 0.7], [0.8, 0.6, 0.9, 0.1]) ) polys = GeoDataFrame( {"id": [1, 2, 3, 4]}, geometry=[ box(0, 0, 0.5, 0.5), box(0, 0.5, 0.5, 1), box(0.5, 0, 1, 0.5), box(0.5, 0.5, 1, 1), ], ) joined = sjoin(pts, polys, predicate=predicate, how="left") assert_index_equal(joined.index, pts.index) def test_sjoin_shared_attribute(self, naturalearth_lowres, naturalearth_cities): countries = read_file(naturalearth_lowres) cities = read_file(naturalearth_cities) countries = countries[["geometry", "name"]].rename(columns={"name": "country"}) # Add first letter of country/city as an attribute column to be compared countries["firstLetter"] = countries["country"].astype(str).str[0] cities["firstLetter"] = cities["name"].astype(str).str[0] result = sjoin(cities, countries, on_attribute="firstLetter") assert ( result["country"].astype(str).str[0] == result["name"].astype(str).str[0] ).all() assert result.shape == (23, 5) @pytest.mark.parametrize( "attr1_key_change_dict, attr2_key_change_dict", [ pytest.param( {True: "merge", False: "no_merge"}, {True: "merge", False: "no_merge"}, id="merge on string attributes", ), pytest.param( {True: 2, False: 1}, {True: 2, False: 1}, id="merge on integer attributes", ), pytest.param( {True: True, False: False}, {True: True, False: False}, id="merge on boolean attributes", ), pytest.param( {True: True, False: False}, {True: "merge", False: "no_merge"}, id="merge on mixed attributes", ), ], ) def test_sjoin_multiple_attributes_datatypes( self, dfs_shared_attribute, attr1_key_change_dict, attr2_key_change_dict ): left_gdf, right_gdf = dfs_shared_attribute left_gdf["attr1"] = left_gdf["attr1"].map(attr1_key_change_dict) left_gdf["attr2"] = left_gdf["attr2"].map(attr2_key_change_dict) right_gdf["attr1"] = right_gdf["attr1"].map(attr1_key_change_dict) right_gdf["attr2"] = right_gdf["attr2"].map(attr2_key_change_dict) joined = sjoin(left_gdf, right_gdf, on_attribute=("attr1", "attr2")) assert (["A", "B"] == joined["attr_tracker"].values).all() def test_sjoin_multiple_attributes_check_header(self, dfs_shared_attribute): left_gdf, right_gdf = dfs_shared_attribute joined = sjoin(left_gdf, right_gdf, on_attribute=["attr1"]) assert (["A", "B", "E", "F"] == joined["attr_tracker"].values).all() assert {"attr2_left", "attr2_right", "attr1"}.issubset(joined.columns) assert "attr1_left" not in joined def test_sjoin_error_column_does_not_exist(self, dfs_shared_attribute): left_gdf, right_gdf = dfs_shared_attribute right_gdf_dropped_attr = right_gdf.drop("attr1", axis=1) left_gdf_dropped_attr = left_gdf.drop("attr1", axis=1) with pytest.raises( ValueError, match="Expected column attr1 is missing from the right dataframe.", ): sjoin(left_gdf, right_gdf_dropped_attr, on_attribute="attr1") with pytest.raises( ValueError, match="Expected column attr1 is missing from the left dataframe.", ): sjoin(left_gdf_dropped_attr, right_gdf, on_attribute="attr1") with pytest.raises( ValueError, match="Expected column attr1 is missing from both of the dataframes.", ): sjoin(left_gdf_dropped_attr, right_gdf_dropped_attr, on_attribute="attr1") def test_sjoin_error_use_geometry_column(self, dfs_shared_attribute): left_gdf, right_gdf = dfs_shared_attribute with pytest.raises( ValueError, match="Active geometry column cannot be used as an input for " "on_attribute parameter.", ): sjoin(left_gdf, right_gdf, on_attribute="geometry") with pytest.raises( ValueError, match="Active geometry column cannot be used as an input for " "on_attribute parameter.", ): sjoin(left_gdf, right_gdf, on_attribute=["attr1", "geometry"]) class TestIndexNames: @pytest.mark.parametrize("how", ["inner", "left", "right"]) def test_preserve_index_names(self, how): # preserve names of both left and right index geoms = [Point(1, 1), Point(2, 2)] df1 = GeoDataFrame({"geometry": geoms}, index=pd.Index([1, 2], name="myidx1")) df2 = GeoDataFrame( {"geometry": geoms}, index=pd.Index(["a", "b"], name="myidx2") ) result = sjoin(df1, df2, how=how) if how in ("inner", "left"): expected = GeoDataFrame( {"myidx1": [1, 2], "geometry": geoms, "myidx2": ["a", "b"]} ).set_index("myidx1") else: # right join expected = GeoDataFrame( {"myidx2": ["a", "b"], "myidx1": [1, 2], "geometry": geoms}, ).set_index("myidx2") assert_geodataframe_equal(result, expected) # but also add suffixes if both left and right have the same index df1.index.name = "myidx" df2.index.name = "myidx" result = sjoin(df1, df2, how=how) if how in ("inner", "left"): expected = GeoDataFrame( {"myidx_left": [1, 2], "geometry": geoms, "myidx_right": ["a", "b"]} ).set_index("myidx_left") else: # right join expected = GeoDataFrame( {"myidx_right": ["a", "b"], "myidx_left": [1, 2], "geometry": geoms}, ).set_index("myidx_right") assert_geodataframe_equal(result, expected) @pytest.mark.parametrize("how", ["inner", "left", "right"]) def test_preserve_index_names_multiindex(self, how): # preserve names of both left and right index geoms = [Point(1, 1), Point(2, 2)] df1 = GeoDataFrame( {"geometry": geoms}, index=pd.MultiIndex.from_tuples( [("a", 1), ("b", 2)], names=["myidx1", "level2"] ), ) df2 = GeoDataFrame( {"geometry": geoms}, index=pd.MultiIndex.from_tuples( [("c", 3), ("d", 4)], names=["myidx2", None] ), ) result = sjoin(df1, df2, how=how) expected_base = GeoDataFrame( { "myidx1": ["a", "b"], "level2": [1, 2], "geometry": geoms, "myidx2": ["c", "d"], "index_right1": [3, 4], } ) if how in ("inner", "left"): expected = expected_base.set_index(["myidx1", "level2"]) else: # right join expected = expected_base.set_index(["myidx2", "index_right1"]) # if it was originally None, that is preserved expected.index.names = ["myidx2", None] assert_geodataframe_equal(result, expected) # but also add suffixes if both left and right have the same index df1.index.names = ["myidx", "level2"] df2.index.names = ["myidx", None] result = sjoin(df1, df2, how=how) expected_base = GeoDataFrame( { "myidx_left": ["a", "b"], "level2": [1, 2], "geometry": geoms, "myidx_right": ["c", "d"], "index_right1": [3, 4], } ) if how in ("inner", "left"): expected = expected_base.set_index(["myidx_left", "level2"]) else: # right join expected = expected_base.set_index(["myidx_right", "index_right1"]) # if it was originally None, that is preserved expected.index.names = ["myidx_right", None] assert_geodataframe_equal(result, expected) @pytest.mark.parametrize("how", ["inner", "left", "right"]) def test_duplicate_column_index_name(self, how): # case where a left column and the right index have the same name or the # other way around -> correctly add suffix or preserve index name geoms = [Point(1, 1), Point(2, 2)] df1 = GeoDataFrame({"myidx": [1, 2], "geometry": geoms}) df2 = GeoDataFrame( {"geometry": geoms}, index=pd.Index(["a", "b"], name="myidx") ) result = sjoin(df1, df2, how=how) if how in ("inner", "left"): expected = GeoDataFrame( {"myidx_left": [1, 2], "geometry": geoms, "myidx_right": ["a", "b"]} ) else: # right join expected = GeoDataFrame( {"index_left": [0, 1], "myidx_left": [1, 2], "geometry": geoms}, index=pd.Index(["a", "b"], name="myidx_right"), ) assert_geodataframe_equal(result, expected) result = sjoin(df2, df1, how=how) if how in ("inner", "left"): expected = GeoDataFrame( {"geometry": geoms, "index_right": [0, 1], "myidx_right": [1, 2]}, index=pd.Index(["a", "b"], name="myidx_left"), ) else: # right join expected = GeoDataFrame( {"myidx_left": ["a", "b"], "myidx_right": [1, 2], "geometry": geoms}, ) assert_geodataframe_equal(result, expected) @pytest.mark.parametrize("how", ["inner", "left", "right"]) def test_duplicate_column_index_name_multiindex(self, how): # case where a left column and the right index have the same name or the # other way around -> correctly add suffix or preserve index name geoms = [Point(1, 1), Point(2, 2)] df1 = GeoDataFrame({"myidx": [1, 2], "geometry": geoms}) df2 = GeoDataFrame( {"geometry": geoms}, index=pd.MultiIndex.from_tuples( [("a", 1), ("b", 2)], names=["myidx", "level2"] ), ) result = sjoin(df1, df2, how=how) if how in ("inner", "left"): expected = GeoDataFrame( { "myidx_left": [1, 2], "geometry": geoms, "myidx_right": ["a", "b"], "level2": [1, 2], } ) else: # right join expected = GeoDataFrame( {"index_left": [0, 1], "myidx_left": [1, 2], "geometry": geoms}, index=pd.MultiIndex.from_tuples( [("a", 1), ("b", 2)], names=["myidx_right", "level2"] ), ) assert_geodataframe_equal(result, expected) result = sjoin(df2, df1, how=how) if how in ("inner", "left"): expected = GeoDataFrame( {"geometry": geoms, "index_right": [0, 1], "myidx_right": [1, 2]}, index=pd.MultiIndex.from_tuples( [("a", 1), ("b", 2)], names=["myidx_left", "level2"] ), ) else: # right join expected = GeoDataFrame( { "myidx_left": ["a", "b"], "level2": [1, 2], "myidx_right": [1, 2], "geometry": geoms, }, ) assert_geodataframe_equal(result, expected) @pytest.mark.parametrize("how", ["inner", "left", "right"]) def test_conflicting_column_index_name(self, how): # test case where the auto-generated index name conflicts geoms = [Point(1, 1), Point(2, 2)] df1 = GeoDataFrame({"index_right": [1, 2], "geometry": geoms}) df2 = GeoDataFrame({"geometry": geoms}) with pytest.raises(ValueError, match="'index_right' cannot be a column name"): sjoin(df1, df2, how=how) @pytest.mark.parametrize("how", ["inner", "left", "right"]) def test_conflicting_column_with_suffix(self, how): # test case where the auto-generated index name conflicts geoms = [Point(1, 1), Point(2, 2)] df1 = GeoDataFrame( {"column": [1, 2], "column_right": ["a", "b"], "geometry": geoms} ) df2 = GeoDataFrame({"column": [0.1, 0.2], "geometry": geoms}) result = sjoin(df1, df2, how=how) if how in ("inner", "left"): expected = GeoDataFrame( {1: [1, 2], 2: ["a", "b"], 3: geoms, 4: [0, 1], 5: [0.1, 0.2]} ) expected.columns = [ "column_left", "column_right", "geometry", "index_right", "column_right", ] else: # right join expected = GeoDataFrame( {1: [0, 1], 2: [1, 2], 3: ["a", "b"], 4: [0.1, 0.2], 5: geoms} ) expected.columns = [ "index_left", "column_left", "column_right", "column_right", "geometry", ] expected = expected.set_geometry("geometry") assert_geodataframe_equal(result, expected) @pytest.mark.usefixtures("_setup_class_nybb_filename") class TestSpatialJoinNYBB: def setup_method(self): self.polydf = read_file(self.nybb_filename) self.crs = self.polydf.crs N = 20 b = [int(x) for x in self.polydf.total_bounds] self.pointdf = GeoDataFrame( [ {"geometry": Point(x, y), "pointattr1": x + y, "pointattr2": x - y} for x, y in zip( range(b[0], b[2], int((b[2] - b[0]) / N)), range(b[1], b[3], int((b[3] - b[1]) / N)), ) ], crs=self.crs, ) def test_geometry_name(self): # test sjoin is working with other geometry name polydf_original_geom_name = self.polydf.geometry.name self.polydf = self.polydf.rename(columns={"geometry": "new_geom"}).set_geometry( "new_geom" ) assert polydf_original_geom_name != self.polydf.geometry.name res = sjoin(self.polydf, self.pointdf, how="left") assert self.polydf.geometry.name == res.geometry.name def test_sjoin_left(self): df = sjoin(self.pointdf, self.polydf, how="left") assert df.shape == (21, 8) for i, row in df.iterrows(): assert row.geometry.geom_type == "Point" assert "pointattr1" in df.columns assert "BoroCode" in df.columns def test_sjoin_right(self): # the inverse of left df = sjoin(self.pointdf, self.polydf, how="right") df2 = sjoin(self.polydf, self.pointdf, how="left") assert df.shape == (12, 8) assert df.shape == df2.shape for i, row in df.iterrows(): assert row.geometry.geom_type == "MultiPolygon" for i, row in df2.iterrows(): assert row.geometry.geom_type == "MultiPolygon" def test_sjoin_inner(self): df = sjoin(self.pointdf, self.polydf, how="inner") assert df.shape == (11, 8) def test_sjoin_predicate(self): # points within polygons df = sjoin(self.pointdf, self.polydf, how="left", predicate="within") assert df.shape == (21, 8) assert df.loc[1]["BoroName"] == "Staten Island" # points contain polygons? never happens so we should have nulls df = sjoin(self.pointdf, self.polydf, how="left", predicate="contains") assert df.shape == (21, 8) assert np.isnan(df.loc[1]["Shape_Area"]) def test_sjoin_bad_predicate(self): # AttributeError: 'Point' object has no attribute 'spandex' with pytest.raises(ValueError): sjoin(self.pointdf, self.polydf, how="left", predicate="spandex") def test_sjoin_duplicate_column_name(self): pointdf2 = self.pointdf.rename(columns={"pointattr1": "Shape_Area"}) df = sjoin(pointdf2, self.polydf, how="left") assert "Shape_Area_left" in df.columns assert "Shape_Area_right" in df.columns @pytest.mark.parametrize("how", ["left", "right", "inner"]) def test_sjoin_named_index(self, how): # original index names should be unchanged pointdf2 = self.pointdf.copy() pointdf2.index.name = "pointid" polydf = self.polydf.copy() polydf.index.name = "polyid" res = sjoin(pointdf2, polydf, how=how) assert pointdf2.index.name == "pointid" assert polydf.index.name == "polyid" # original index name should pass through to result if how == "right": assert res.index.name == "polyid" else: # how == "left", how == "inner" assert res.index.name == "pointid" def test_sjoin_values(self): # GH190 self.polydf.index = [1, 3, 4, 5, 6] df = sjoin(self.pointdf, self.polydf, how="left") assert df.shape == (21, 8) df = sjoin(self.polydf, self.pointdf, how="left") assert df.shape == (12, 8) @pytest.mark.xfail def test_no_overlapping_geometry(self): # Note: these tests are for correctly returning GeoDataFrame # when result of the join is empty df_inner = sjoin(self.pointdf.iloc[17:], self.polydf, how="inner") df_left = sjoin(self.pointdf.iloc[17:], self.polydf, how="left") df_right = sjoin(self.pointdf.iloc[17:], self.polydf, how="right") expected_inner_df = pd.concat( [ self.pointdf.iloc[:0], pd.Series(name="index_right", dtype="int64"), self.polydf.drop("geometry", axis=1).iloc[:0], ], axis=1, ) expected_inner = GeoDataFrame(expected_inner_df) expected_right_df = pd.concat( [ self.pointdf.drop("geometry", axis=1).iloc[:0], pd.concat( [ pd.Series(name="index_left", dtype="int64"), pd.Series(name="index_right", dtype="int64"), ], axis=1, ), self.polydf, ], axis=1, ) expected_right = GeoDataFrame(expected_right_df).set_index("index_right") expected_left_df = pd.concat( [ self.pointdf.iloc[17:], pd.Series(name="index_right", dtype="int64"), self.polydf.iloc[:0].drop("geometry", axis=1), ], axis=1, ) expected_left = GeoDataFrame(expected_left_df) assert expected_inner.equals(df_inner) assert expected_right.equals(df_right) assert expected_left.equals(df_left) @pytest.mark.skip("Not implemented") def test_sjoin_outer(self): df = sjoin(self.pointdf, self.polydf, how="outer") assert df.shape == (21, 8) def test_sjoin_empty_geometries(self): # https://github.com/geopandas/geopandas/issues/944 empty = GeoDataFrame(geometry=[GeometryCollection()] * 3, crs=self.crs) df = sjoin(pd.concat([self.pointdf, empty]), self.polydf, how="left") assert df.shape == (24, 8) df2 = sjoin(self.pointdf, pd.concat([self.polydf, empty]), how="left") assert df2.shape == (21, 8) @pytest.mark.parametrize("predicate", ["intersects", "within", "contains"]) def test_sjoin_no_valid_geoms(self, predicate): """Tests a completely empty GeoDataFrame.""" empty = GeoDataFrame(geometry=[], crs=self.pointdf.crs) assert sjoin(self.pointdf, empty, how="inner", predicate=predicate).empty assert sjoin(self.pointdf, empty, how="right", predicate=predicate).empty assert sjoin(empty, self.pointdf, how="inner", predicate=predicate).empty assert sjoin(empty, self.pointdf, how="left", predicate=predicate).empty def test_empty_sjoin_return_duplicated_columns(self, nybb_filename): nybb = geopandas.read_file(nybb_filename) nybb2 = nybb.copy() nybb2.geometry = nybb2.translate(200000) # to get non-overlapping result = geopandas.sjoin(nybb, nybb2) assert "BoroCode_right" in result.columns assert "BoroCode_left" in result.columns @pytest.fixture def world(naturalearth_lowres): return read_file(naturalearth_lowres) @pytest.fixture def cities(naturalearth_cities): return read_file(naturalearth_cities) def test_sjoin_inner(world, cities): # GH637 countries = world[["geometry", "name"]] countries = countries.rename(columns={"name": "country"}) cities_with_country = sjoin(cities, countries, how="inner", predicate="intersects") assert cities_with_country.shape == (213, 4) class TestNearest: @pytest.mark.parametrize( "how_kwargs", ({}, {"how": "inner"}, {"how": "left"}, {"how": "right"}) ) def test_allowed_hows(self, how_kwargs): left = geopandas.GeoDataFrame({"geometry": []}) right = geopandas.GeoDataFrame({"geometry": []}) sjoin_nearest(left, right, **how_kwargs) # no error @pytest.mark.parametrize("how", ("outer", "abcde")) def test_invalid_hows(self, how: str): left = geopandas.GeoDataFrame({"geometry": []}) right = geopandas.GeoDataFrame({"geometry": []}) with pytest.raises(ValueError, match="`how` was"): sjoin_nearest(left, right, how=how) @pytest.mark.parametrize("distance_col", (None, "distance")) def test_empty_right_df_how_left(self, distance_col: str): # all records from left and no results from right left = geopandas.GeoDataFrame({"geometry": [Point(0, 0), Point(1, 1)]}) right = geopandas.GeoDataFrame({"geometry": []}) joined = sjoin_nearest( left, right, how="left", distance_col=distance_col, ) assert_geoseries_equal(joined["geometry"], left["geometry"]) assert joined["index_right"].isna().all() if distance_col is not None: assert joined[distance_col].isna().all() @pytest.mark.parametrize("distance_col", (None, "distance")) def test_empty_right_df_how_right(self, distance_col: str): # no records in joined left = geopandas.GeoDataFrame({"geometry": [Point(0, 0), Point(1, 1)]}) right = geopandas.GeoDataFrame({"geometry": []}) joined = sjoin_nearest( left, right, how="right", distance_col=distance_col, ) assert joined.empty if distance_col is not None: assert distance_col in joined @pytest.mark.parametrize("how", ["inner", "left"]) @pytest.mark.parametrize("distance_col", (None, "distance")) def test_empty_left_df(self, how, distance_col: str): right = geopandas.GeoDataFrame({"geometry": [Point(0, 0), Point(1, 1)]}) left = geopandas.GeoDataFrame({"geometry": []}) joined = sjoin_nearest(left, right, how=how, distance_col=distance_col) assert joined.empty if distance_col is not None: assert distance_col in joined @pytest.mark.parametrize("distance_col", (None, "distance")) def test_empty_left_df_how_right(self, distance_col: str): right = geopandas.GeoDataFrame({"geometry": [Point(0, 0), Point(1, 1)]}) left = geopandas.GeoDataFrame({"geometry": []}) joined = sjoin_nearest( left, right, how="right", distance_col=distance_col, ) assert_geoseries_equal(joined["geometry"], right["geometry"]) assert joined["index_left"].isna().all() if distance_col is not None: assert joined[distance_col].isna().all() @pytest.mark.parametrize("how", ["inner", "left"]) def test_empty_join_due_to_max_distance(self, how): # after applying max_distance the join comes back empty # (as in NaN in the joined columns) left = geopandas.GeoDataFrame({"geometry": [Point(0, 0)]}) right = geopandas.GeoDataFrame({"geometry": [Point(1, 1), Point(2, 2)]}) joined = sjoin_nearest( left, right, how=how, max_distance=1, distance_col="distances", ) expected = left.copy() expected["index_right"] = [np.nan] expected["distances"] = [np.nan] if how == "inner": expected = expected.dropna() expected["index_right"] = expected["index_right"].astype("int64") assert_geodataframe_equal(joined, expected) def test_empty_join_due_to_max_distance_how_right(self): # after applying max_distance the join comes back empty # (as in NaN in the joined columns) left = geopandas.GeoDataFrame({"geometry": [Point(0, 0), Point(1, 1)]}) right = geopandas.GeoDataFrame({"geometry": [Point(2, 2)]}) joined = sjoin_nearest( left, right, how="right", max_distance=1, distance_col="distances", ) expected = right.copy() expected["index_left"] = [np.nan] expected["distances"] = [np.nan] expected = expected[["index_left", "geometry", "distances"]] assert_geodataframe_equal(joined, expected) @pytest.mark.parametrize("how", ["inner", "left"]) def test_max_distance(self, how): left = geopandas.GeoDataFrame({"geometry": [Point(0, 0), Point(1, 1)]}) right = geopandas.GeoDataFrame({"geometry": [Point(1, 1), Point(2, 2)]}) joined = sjoin_nearest( left, right, how=how, max_distance=1, distance_col="distances", ) expected = left.copy() expected["index_right"] = [np.nan, 0] expected["distances"] = [np.nan, 0] if how == "inner": expected = expected.dropna() expected["index_right"] = expected["index_right"].astype("int64") assert_geodataframe_equal(joined, expected) def test_max_distance_how_right(self): left = geopandas.GeoDataFrame({"geometry": [Point(1, 1), Point(2, 2)]}) right = geopandas.GeoDataFrame({"geometry": [Point(0, 0), Point(1, 1)]}) joined = sjoin_nearest( left, right, how="right", max_distance=1, distance_col="distances", ) expected = right.copy() expected["index_left"] = [np.nan, 0] expected["distances"] = [np.nan, 0] expected = expected[["index_left", "geometry", "distances"]] assert_geodataframe_equal(joined, expected) @pytest.mark.parametrize("how", ["inner", "left"]) @pytest.mark.parametrize( "geo_left, geo_right, expected_left, expected_right, distances", [ ( [Point(0, 0), Point(1, 1)], [Point(1, 1)], [0, 1], [0, 0], [math.sqrt(2), 0], ), ( [Point(0, 0), Point(1, 1)], [Point(1, 1), Point(0, 0)], [0, 1], [1, 0], [0, 0], ), ( [Point(0, 0), Point(1, 1)], [Point(1, 1), Point(0, 0), Point(0, 0)], [0, 0, 1], [1, 2, 0], [0, 0, 0], ), ( [Point(0, 0), Point(1, 1)], [Point(1, 1), Point(0, 0), Point(2, 2)], [0, 1], [1, 0], [0, 0], ), ( [Point(0, 0), Point(1, 1)], [Point(1, 1), Point(0.25, 1)], [0, 1], [1, 0], [math.sqrt(0.25**2 + 1), 0], ), ( [Point(0, 0), Point(1, 1)], [Point(-10, -10), Point(100, 100)], [0, 1], [0, 0], [math.sqrt(10**2 + 10**2), math.sqrt(11**2 + 11**2)], ), ( [Point(0, 0), Point(1, 1)], [Point(x, y) for x, y in zip(np.arange(10), np.arange(10))], [0, 1], [0, 1], [0, 0], ), ( [Point(0, 0), Point(1, 1), Point(0, 0)], [Point(1.1, 1.1), Point(0, 0)], [0, 1, 2], [1, 0, 1], [0, np.sqrt(0.1**2 + 0.1**2), 0], ), ], ) def test_sjoin_nearest_left( self, geo_left, geo_right, expected_left: Sequence[int], expected_right: Sequence[int], distances: Sequence[float], how, ): left = geopandas.GeoDataFrame({"geometry": geo_left}) right = geopandas.GeoDataFrame({"geometry": geo_right}) expected_gdf = left.iloc[expected_left].copy() expected_gdf["index_right"] = expected_right # without distance col joined = sjoin_nearest(left, right, how=how) # inner / left join give a different row order check_like = how == "inner" assert_geodataframe_equal(expected_gdf, joined, check_like=check_like) # with distance col expected_gdf["distance_col"] = np.array(distances, dtype=float) joined = sjoin_nearest(left, right, how=how, distance_col="distance_col") assert_geodataframe_equal(expected_gdf, joined, check_like=check_like) @pytest.mark.parametrize( "geo_left, geo_right, expected_left, expected_right, distances", [ ([Point(0, 0), Point(1, 1)], [Point(1, 1)], [1], [0], [0]), ( [Point(0, 0), Point(1, 1)], [Point(1, 1), Point(0, 0)], [1, 0], [0, 1], [0, 0], ), ( [Point(0, 0), Point(1, 1)], [Point(1, 1), Point(0, 0), Point(0, 0)], [1, 0, 0], [0, 1, 2], [0, 0, 0], ), ( [Point(0, 0), Point(1, 1)], [Point(1, 1), Point(0, 0), Point(2, 2)], [1, 0, 1], [0, 1, 2], [0, 0, math.sqrt(2)], ), ( [Point(0, 0), Point(1, 1)], [Point(1, 1), Point(0.25, 1)], [1, 1], [0, 1], [0, 0.75], ), ( [Point(0, 0), Point(1, 1)], [Point(-10, -10), Point(100, 100)], [0, 1], [0, 1], [math.sqrt(10**2 + 10**2), math.sqrt(99**2 + 99**2)], ), ( [Point(0, 0), Point(1, 1)], [Point(x, y) for x, y in zip(np.arange(10), np.arange(10))], [0, 1] + [1] * 8, list(range(10)), [0, 0] + [np.sqrt(x**2 + x**2) for x in np.arange(1, 9)], ), ( [Point(0, 0), Point(1, 1), Point(0, 0)], [Point(1.1, 1.1), Point(0, 0)], [1, 0, 2], [0, 1, 1], [np.sqrt(0.1**2 + 0.1**2), 0, 0], ), ], ) def test_sjoin_nearest_right( self, geo_left, geo_right, expected_left: Sequence[int], expected_right: Sequence[int], distances: Sequence[float], ): left = geopandas.GeoDataFrame({"geometry": geo_left}) right = geopandas.GeoDataFrame({"geometry": geo_right}) expected_gdf = right.iloc[expected_right].copy() expected_gdf["index_left"] = expected_left expected_gdf = expected_gdf[["index_left", "geometry"]] # without distance col joined = sjoin_nearest(left, right, how="right") assert_geodataframe_equal(expected_gdf, joined) # with distance col expected_gdf["distance_col"] = np.array(distances, dtype=float) joined = sjoin_nearest(left, right, how="right", distance_col="distance_col") assert_geodataframe_equal(expected_gdf, joined) @pytest.mark.filterwarnings("ignore:Geometry is in a geographic CRS") def test_sjoin_nearest_inner(self, naturalearth_lowres, naturalearth_cities): # check equivalency of left and inner join countries = read_file(naturalearth_lowres) cities = read_file(naturalearth_cities) countries = countries[["geometry", "name"]].rename(columns={"name": "country"}) # default: inner and left give the same result result1 = sjoin_nearest(cities, countries, distance_col="dist") assert result1.shape[0] == cities.shape[0] result2 = sjoin_nearest(cities, countries, distance_col="dist", how="inner") assert_geodataframe_equal(result2, result1) result3 = sjoin_nearest(cities, countries, distance_col="dist", how="left") assert_geodataframe_equal(result3, result1, check_like=True) # with max_distance: rows that go above are dropped in case of inner result4 = sjoin_nearest(cities, countries, distance_col="dist", max_distance=1) assert_geodataframe_equal( result4, result1[result1["dist"] < 1], check_like=True ) result5 = sjoin_nearest( cities, countries, distance_col="dist", max_distance=1, how="left" ) assert result5.shape[0] == cities.shape[0] result5 = result5.dropna() result5["index_right"] = result5["index_right"].astype("int64") assert_geodataframe_equal(result5, result4, check_like=True) @pytest.mark.parametrize( "max_distance,expected", [(None, [1, 3, 3, 1, 2]), (1.1, [3, 3, 1, 2])] ) def test_sjoin_nearest_exclusive(self, max_distance, expected): geoms = shapely.points(np.arange(3), np.arange(3)) geoms = np.append(geoms, [Point(1, 2)]) df = geopandas.GeoDataFrame({"geometry": geoms}) result = df.sjoin_nearest( df, max_distance=max_distance, distance_col="dist", exclusive=True ) assert_series_equal( result["index_right"].reset_index(drop=True), pd.Series(expected), check_names=False, ) if max_distance: assert result["dist"].max() <= max_distance