# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements. See the NOTICE file
# distributed with this work for additional information
# regarding copyright ownership. The ASF licenses this file
# to you under the Apache License, Version 2.0 (the
# "License"); you may not use this file except in compliance
# with the License. You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing,
# software distributed under the License is distributed on an
# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
# KIND, either express or implied. See the License for the
# specific language governing permissions and limitations
# under the License.
import inspect
import sys
from functools import partial
from typing import Optional, Union
from pyspark.sql import Column
from sedona.spark.sql.dataframe_api import (
ColumnOrName,
call_sedona_function,
validate_argument_types,
)
_call_predicate_function = partial(call_sedona_function, "st_predicates")
[docs]@validate_argument_types
def ST_Contains(a: ColumnOrName, b: ColumnOrName) -> Column:
"""Check whether a contains b. Polymorphic over input type:
- Geometry / Geometry — topological containment via JTS.
- Geography / Geography — topological containment via S2.
- Box2D / Box2D — closed-interval bbox containment on both axes (PostGIS ``~`` on box2d).
- Box3D / Box3D — closed-interval bbox containment on all three axes.
:param a: Outer geometry or Box2D / Box3D column.
:type a: ColumnOrName
:param b: Inner geometry or Box2D / Box3D column.
:type b: ColumnOrName
:return: True if a contains b, false otherwise.
:rtype: Column
"""
return _call_predicate_function("ST_Contains", (a, b))
[docs]@validate_argument_types
def ST_Crosses(a: ColumnOrName, b: ColumnOrName) -> Column:
"""Check whether geometry a crosses geometry b.
:param a: Geometry to check crossing with.
:type a: ColumnOrName
:param b: Geometry to check crossing of.
:type b: ColumnOrName
:return: True if geometry a cross geometry b and False otherwise as a boolean column.
:rtype: Column
"""
return _call_predicate_function("ST_Crosses", (a, b))
[docs]@validate_argument_types
def ST_Disjoint(a: ColumnOrName, b: ColumnOrName) -> Column:
"""Check whether two geometries are disjoint.
:param a: One geometry column to check.
:type a: ColumnOrName
:param b: Other geometry column to check.
:type b: ColumnOrName
:return: True if a and b are disjoint and False otherwise as a boolean column.
:rtype: Column
"""
return _call_predicate_function("ST_Disjoint", (a, b))
[docs]@validate_argument_types
def ST_Equals(a: ColumnOrName, b: ColumnOrName) -> Column:
"""Check whether two geometries are equal regardless of vertex ordering.
:param a: One geometry column to check.
:type a: ColumnOrName
:param b: Other geometry column to check.
:type b: ColumnOrName
:return: True if a and b are equal and False otherwise, as a boolean column.
:rtype: Column
"""
return _call_predicate_function("ST_Equals", (a, b))
[docs]@validate_argument_types
def ST_EqualsExact(
a: ColumnOrName,
b: ColumnOrName,
tolerance: Union[ColumnOrName, float, int],
) -> Column:
"""Check whether two geometries have the same structure and coordinate ordering
within a tolerance.
The comparison uses x and y coordinates. Z and M coordinates are ignored.
:param a: One geometry column to check.
:type a: ColumnOrName
:param b: Other geometry column to check.
:type b: ColumnOrName
:param tolerance: Maximum distance allowed between corresponding coordinates.
:type tolerance: ColumnOrName or float or int
:return: True if a and b are exactly equal within tolerance, otherwise False.
:rtype: Column
"""
return _call_predicate_function("ST_EqualsExact", (a, b, tolerance))
[docs]@validate_argument_types
def ST_EqualsIdentical(a: ColumnOrName, b: ColumnOrName) -> Column:
"""Check whether two geometries are structurally identical in every dimension.
Geometry types, component ordering, vertex ordering, dimensionality, and all
coordinate values must match exactly. NaN ordinate values compare equal to
other NaN values.
:param a: One geometry column to check.
:type a: ColumnOrName
:param b: Other geometry column to check.
:type b: ColumnOrName
:return: True if a and b are identical, otherwise False.
:rtype: Column
"""
return _call_predicate_function("ST_EqualsIdentical", (a, b))
[docs]@validate_argument_types
def ST_Intersects(a: ColumnOrName, b: ColumnOrName) -> Column:
"""Check whether a and b intersect. Polymorphic over input type:
- Geometry / Geometry — topological intersection via JTS.
- Geography / Geography — topological intersection via S2.
- Box2D / Box2D — closed-interval bbox intersection on both axes (PostGIS ``&&`` on box2d).
- Box3D / Box3D — closed-interval bbox intersection on all three axes (PostGIS ``&&&``).
:param a: One geometry or Box2D / Box3D column.
:type a: ColumnOrName
:param b: Other geometry or Box2D / Box3D column.
:type b: ColumnOrName
:return: True if a and b intersect, false otherwise.
:rtype: Column
"""
return _call_predicate_function("ST_Intersects", (a, b))
[docs]@validate_argument_types
def ST_OrderingEquals(a: ColumnOrName, b: ColumnOrName) -> Column:
"""Check whether two geometries have identical vertices that are in the same order.
:param a: One geometry column to check.
:type a: ColumnOrName
:param b: Other geometry column to check.
:type b: ColumnOrName
:return: True if a and b are equal and False otherwise, as a boolean column.
:rtype: Column
"""
return _call_predicate_function("ST_OrderingEquals", (a, b))
[docs]@validate_argument_types
def ST_Overlaps(a: ColumnOrName, b: ColumnOrName) -> Column:
"""Check whether geometry a overlaps geometry b.
:param a: Geometry column to check overlapping for.
:type a: ColumnOrName
:param b: Geometry column to check overlapping of.
:type b: ColumnOrName
:return: True if a overlaps b and False otherwise, as a boolean column.
:rtype: Column
"""
return _call_predicate_function("ST_Overlaps", (a, b))
[docs]@validate_argument_types
def ST_Touches(a: ColumnOrName, b: ColumnOrName) -> Column:
"""Check whether two geometries touch.
:param a: One geometry column to check.
:type a: ColumnOrName
:param b: Other geometry column to check.
:type b: ColumnOrName
:return: True if a and b touch and False otherwise, as a boolean column.
:rtype: Column
"""
return _call_predicate_function("ST_Touches", (a, b))
[docs]@validate_argument_types
def ST_Relate(
a: ColumnOrName, b: ColumnOrName, intersectionMatrix: Optional[ColumnOrName] = None
) -> Column:
"""Check whether two geometries are related to each other.
:param a: One geometry column to check.
:type a: ColumnOrName
:param b: Other geometry column to check.
:type b: ColumnOrName
:param intersectionMatrix: intersectionMatrix column to check
:type intersectionMatrix: ColumnOrName
:return: True if a and b touch and False otherwise, as a boolean column.
:rtype: Column
"""
args = (a, b) if intersectionMatrix is None else (a, b, intersectionMatrix)
return _call_predicate_function("ST_Relate", args)
[docs]@validate_argument_types
def ST_RelateMatch(matrix1: ColumnOrName, matrix2: ColumnOrName) -> Column:
"""Check whether two DE-9IM are related to each other.
:param matrix1: One geometry column to check.
:type matrix1: ColumnOrName
:param matrix2: Other geometry column to check.
:type matrix2: ColumnOrName
:return: True if a and b touch and False otherwise, as a boolean column.
:rtype: Column
"""
return _call_predicate_function("ST_RelateMatch", (matrix1, matrix2))
[docs]@validate_argument_types
def ST_Within(a: ColumnOrName, b: ColumnOrName) -> Column:
"""Check if geometry a is within geometry b.
:param a: Geometry column to check.
:type a: ColumnOrName
:param b: Geometry column to check.
:type b: ColumnOrName
:return: True if a is within b and False otherwise, as a boolean column.
:rtype: Column
"""
return _call_predicate_function("ST_Within", (a, b))
[docs]@validate_argument_types
def ST_Covers(a: ColumnOrName, b: ColumnOrName) -> Column:
"""Check whether geometry a covers geometry b.
:param a: Geometry column to check.
:type a: ColumnOrName
:param b: Geometry column to check.
:type b: ColumnOrName
:return: True if a covers b and False otherwise, as a boolean column.
:rtype: Column
"""
return _call_predicate_function("ST_Covers", (a, b))
[docs]@validate_argument_types
def ST_CoveredBy(a: ColumnOrName, b: ColumnOrName) -> Column:
"""Check if geometry a is covered by geometry b.
:param a: Geometry column to check.
:type a: ColumnOrName
:param b: Geometry column to check.
:type b: ColumnOrName
:return: True if a is covered by b and False otherwise, as a boolean column.
:rtype: Column
"""
return _call_predicate_function("ST_CoveredBy", (a, b))
[docs]@validate_argument_types
def ST_DWithin(
a: ColumnOrName,
b: ColumnOrName,
distance: Union[ColumnOrName, float],
use_sphere: Optional[Union[ColumnOrName, bool]] = None,
):
"""
Check if geometry a is within 'distance' units of geometry b
:param a: Geometry column to check
:param b: Geometry column to check
:param distance: distance units to check the within predicate
:param use_sphere: whether to use spheroid distance or euclidean distance
:return: True if a is within distance units of Geometry b
"""
args = (
(a, b, distance, use_sphere)
if use_sphere is not None
else (
a,
b,
distance,
)
)
return _call_predicate_function("ST_DWithin", args)
[docs]@validate_argument_types
def ST_3DDWithin(
a: ColumnOrName,
b: ColumnOrName,
distance: Union[ColumnOrName, float],
) -> Column:
"""Check whether two geometries (or two Box3D values) are within ``distance`` units using 3D
Euclidean distance.
Mirrors PostGIS ``ST_3DDWithin``. Coordinates without a Z dimension are treated as
``z = 0``. For Box3D inputs the test is a closed-interval 3D distance between the two
cuboids; inverted bounds raise ``IllegalArgumentException``. NULL on null input.
:param a: First geometry or Box3D column.
:type a: ColumnOrName
:param b: Second geometry or Box3D column.
:type b: ColumnOrName
:param distance: Distance threshold.
:type distance: Union[ColumnOrName, float]
:return: True if the 3D distance between a and b is ≤ ``distance``.
:rtype: Column
"""
return _call_predicate_function("ST_3DDWithin", (a, b, distance))
# Automatically populate __all__
__all__ = [
name
for name, obj in inspect.getmembers(sys.modules[__name__])
if inspect.isfunction(obj)
]