Copyright (C) 2019-2022 HERE Global B.V. and its affiliate(s).

All rights reserved.

This software and other materials contain proprietary information

controlled by HERE and are protected by applicable copyright legislation.

Any use and utilization of this software and other materials and

disclosure to any third parties is conditional upon having a separate

agreement with HERE for the access, use, utilization or disclosure of this

software. In the absence of such agreement, the use of the software is not

allowed.

"""
Module to operate with HERETile tiling scheme using (Geo)DataFrames.
"""

from typing import Optional

import geopandas as gpd
import here.geotiles.heretile as ht
import numpy as np
import pandas as pd

"""
The CRS of all the GeoSeries and GeoDataFrame accepted or returned.
"""
crs = "EPSG:4326"

def _verify_integer_series(tile_ids: pd.Series):
"""
Verify that the series is a series of integers. This is an important preliminary check
to make sure we're dealing with potentially-valid tile ids and make sure the user
hasn't converted, by mistake, tiles ids to floats or other types. On tile ids
we perform bit-level operations so floats are not ok.

This checks the type of the series, not that the integers they contain are valid ids.

NA values are supported by checking if the type
is uppercase Int or UInt, types that support NA.

:param tile_ids: series containing tile ids
:raises ValueError: in case the series doesn't contain integers
"""

TODO: try with is_integer_dtype

error_msg: str = "Series data type different than integer and string: ".format(
tile_ids.dtype
)
if tile_ids.dtype not in (
np.int8,
np.int16,
np.int32,
np.int64,
np.uint8,
np.uint16,
np.uint32,
np.uint64,
np.str_,
str,
) and not isinstance(
tile_ids.dtype,
(
pd.Int8Dtype,
pd.Int16Dtype,
pd.Int32Dtype,
pd.Int64Dtype,
pd.UInt8Dtype,
pd.UInt16Dtype,
pd.UInt32Dtype,
pd.UInt64Dtype,
pd.StringDtype,
),
):
raise ValueError(error_msg)

def _verify_crs(series: gpd.GeoSeries):
if series.crs is not None and series.crs != crs:
raise ValueError("GeoSeries with incompatible CRS: ".format(series.crs))

def to_geoseries(tile_ids: pd.Series, to_geo) -> gpd.GeoSeries: verify_integer_series(tile_ids)

def fn(t):
if not pd.isna(t) and ht.is_valid(t):
return to_geo(t)
else:
return None

return gpd.GeoSeries(map(fn, tile_ids), tile_ids.index, crs=crs)

def _to_bool(tile_ids: pd.Series, other: pd.Series, pred) -> pd.Series:
df = pd.DataFrame({"tile_ids": tile_ids.astype("Int64"), "other": other})

def fn(s):
t, o = s
if not pd.isna(t) and ht.is_valid(t) and not pd.isna(o):
return pred(t, o)
else:
return pd.NA

Check if there's a better type for nullable boolean, or simply return false

return pd.Series(map(fn, df.itertuples(index=False)), index=df.index, dtype="object")

def to_tile_list(tile_ids: pd.Series, to_list) -> pd.Series: verify_integer_series(tile_ids)

def fn(t):
if not pd.isna(t) and ht.is_valid(t):
ids = to_list(t)
return ids if ids else None
else:
return None

return tile_ids.map(fn)

[docs]
def is_valid(tile_ids: pd.Series) -> pd.Series: # bool
"""
Check if integers are a valid tile IDs.

Index and NaN values are retained.

:param tile_ids: a Series of integers
:return: a Series of booleans with the result of the test
:raises ValueError: in case the series doesn't contain integers # noqa

Example:

is_valid(pd.Series([0, 1, 2, 3, 4]))
0 False
1 True
2 False
3 False
4 True
dtype: bool
"""
_verify_integer_series(tile_ids)
return tile_ids.map(ht.is_valid, na_action="ignore")

[docs]
def from_x_y_level(xs: pd.Series, ys: pd.Series, levels: pd.Series) -> pd.Series: # tile id
"""
Return the tiles given X and Y components and levels.

Series must be indexed the same. NaN values or series missing a row
for some index value determine the resulting tile ID to be NaN.
This is also the case if not tile ID corresponds to a (x, y, level) tuple.

Index is retained.

:param xs: the Series with the tile X component
:param ys: the Series with the tile Y component
:param levels: the Series with tile levels
:return: a Series tile IDs, NaN when no tile ID can be calculated
:raises ValueError: in case the series doesn't contain integers # noqa
"""
verify_integer_series(xs) verifyinteger_series(ys) verify_integer_series(levels)
df = pd.DataFrame({"x": xs, "y": ys, "level": levels}, dtype="Int64")

def fn(s):
x = s.x
y = s.y
level = s.level
if not pd.isna(x) and not pd.isna(y) and not pd.isna(level):
try:
return ht.from_x_y_level(x, y, level)
except ValueError:
return None
else:
return None

This syntax is preferred than apply(fn, axis=1) because it's possible

to set the dtype to Int64, otherwise None values determine a conversion to float64

return pd.Series(map(fn, df.itertuples(index=False)), index=df.index, dtype="Int64")

[docs]
def from_coordinates(lngs: pd.Series, lats: pd.Series, level: int) -> pd.Series: # tile id
"""
Return the tiles that contain points.

Series must be indexed the same. NaN values or series missing a row
for some index value determine the resulting tile ID to be NaN.
This is also the case if a point can't be mapped to a tile ID.

Index is retained.

:param lngs: the Series with longitudes
:param lats: the Series with latitudes
:param level: the tiling level
:return: a Series tile IDs, NaN when no tile ID can be calculated
"""
df = pd.DataFrame({"lng": lngs, "lat": lats}, dtype="float")

def fn(c):
lng = c.lng
lat = c.lat
if not pd.isna(lng) and not pd.isna(lat):
try:
return ht.from_coordinates(lng, lat, level)
except ValueError:
return None
else:
return None

return pd.Series(map(fn, df.itertuples(index=False)), index=df.index, dtype="Int64")

[docs]
def from_point(pts: gpd.GeoSeries, level: int) -> pd.Series: # tile id
"""
Return the tiles that contain points.

NaN values or rows not containing a Point determine
the resulting tile ID to be NaN.
This is also the case if a point can't be mapped to a tile ID.

Index is retained.

:param pts: the GeoSeries with the points
:param level: the tiling level
:return: a Series tile IDs, NaN when no tile ID can be calculated
"""
_verify_crs(pts)

def fn(pt):
if not pd.isna(pt):
try:
return ht.from_point(pt, level)
except ValueError:
return None
else:
return None

return pd.Series(map(fn, pts), index=pts.index, dtype="Int64")

[docs]
def in_geometry_bounds(
geometries: gpd.GeoSeries, level: int, fully_contained: bool = False
) -> pd.Series: # list of tile ids
"""
Return the tiles that are included in the bounds of geometries.

If fully_contained is false, some of the returned tiles may be partially
outside the bounds. Otherwise, return only the tiles wholly inside the bounds.

Index and NaN values are retained. NaN is also returned in place of empty lists.

Use pd.Series.explode method to turn the lists into multiple rows.

:param geometries: the GeoSeries with arbitrary geometries
:param level: the tiling level
:param fully_contained: return only tiles wholly inside the bounding box
:return: a Series with the list of tile IDs corresponding to each geometry,
in the order of West to East first, then South to North
"""

def fn(geo):
ids = list(ht.in_geometry_bounds(geo, level, fully_contained=fully_contained))
return ids if ids else None

return geometries.map(fn, na_action="ignore")

[docs]
def in_geometry(
geometries: gpd.GeoSeries, level: int, fully_contained: bool = False
) -> pd.Series: # list of tile ids
"""
Return the tiles that are included in the geometries.

If fully_contained is false, some of the returned tiles may be partially
outside the geometry. Otherwise, return only the tiles wholly inside the geometry.

Inclusion is modeled as the shapely relation intersects (fully_contained is false)
or contains (otherwise) between the geometry and the tiles.

Examples (fully_contained is false):

• if the geometry is a point, return one, two or four tiles
• if the geometry is a line string, it return all the tiles along the line
• if the geometry is a polygon, it returns all the tiles intersecting with the polygon
• for multi-geometries, return the union of the tiles in each geometry

Examples (fully_contained is true):

• if the geometry is a point, nothing is returned
• if the geometry is a line string, nothing is returned
• if the geometry is a polygon, it returns all the tiles contained in the polygon
• for multi-geometries, return the union of the tiles in each geometry

Index and NaN values are retained. NaN is also returned in place of empty lists.

Use pd.Series.explode method to turn the lists into multiple rows.

:param geometries: the GeoSeries with arbitrary geometries
:param level: the tiling level
:param fully_contained: return only tiles wholly inside the bounding box
:return: a Series with the list of tile IDs corresponding to each geometry,
in the order of West to East first, then South to North
"""

def fn(geo):
ids = list(ht.in_geometry(geo, level, fully_contained=fully_contained))
return ids if ids else None

return geometries.map(fn, na_action="ignore")

[docs]
def get_x(tile_ids: pd.Series) -> pd.Series: # int
"""
Return the X components of the tiles.

Index and NaN values are retained.
NaN are also returned in case of invalid tile IDs.

:param tile_ids: the tile IDs
:return: a Series of X components, NaN when not applicable
:raises ValueError: in case the series doesn't contain integers # noqa
"""
_verify_integer_series(tile_ids)

def fn(t):
if not pd.isna(t) and ht.is_valid(t):
return ht.get_x(t)
else:
return None

This syntax is preferred than apply(fn, axis=1) because it's possible

to set the dtype to Int32, otherwise None values determine a conversion to float64

return pd.Series(map(fn, tile_ids), index=tile_ids.index, dtype="Int32")

[docs]
def get_y(tile_ids: pd.Series) -> pd.Series: # int
"""
Return the Y components of the tiles.

Index and NaN values are retained.
NaN are also returned in case of invalid tile IDs.

:param tile_ids: the tile IDs
:return: a Series of Y components, NaN when not applicable
:raises ValueError: in case the series doesn't contain integers # noqa
"""
_verify_integer_series(tile_ids)

def fn(t):
if not pd.isna(t) and ht.is_valid(t):
return ht.get_y(t)
else:
return None

This syntax is preferred than apply(fn, axis=1) because it's possible

to set the dtype to Int32, otherwise None values determine a conversion to float64

return pd.Series(map(fn, tile_ids), index=tile_ids.index, dtype="Int32")

[docs]
def get_level(tile_ids: pd.Series) -> pd.Series: # int
"""
Return the levels of the tiles.

Index and NaN values are retained.
NaN are also returned in case of invalid tile IDs.

:param tile_ids: the tile IDs
:return: a Series tiling levels, NaN when not applicable
:raises ValueError: in case the series doesn't contain integers # noqa
"""
_verify_integer_series(tile_ids)

def fn(t):
if not pd.isna(t) and ht.is_valid(t):
return ht.get_level(t)
else:
return None

This syntax is preferred than apply(fn, axis=1) because it's possible

to set the dtype to Int8, otherwise None values determine a conversion to float64

return pd.Series(map(fn, tile_ids), index=tile_ids.index, dtype="Int8")

[docs]
def get_x_y_level(tile_ids: pd.Series) -> pd.DataFrame: # x,y,level int
"""
Return the X and Y components and the levels of the tiles.

Index and NaN values are retained.
NaN are also returned in case of invalid tile IDs.

Resulting NaN values are removed, unless drop_na is set to false.

:param tile_ids: the tile IDs
:return: a DataFrame with columns x, y, level, NaN when not applicable
:raises ValueError: in case the series doesn't contain integers # noqa
"""
return pd.DataFrame({"x": get_x(tile_ids), "y": get_y(tile_ids), "level": get_level(tile_ids)})

[docs]
def get_center_coordinates(tile_ids: pd.Series) -> pd.DataFrame: # lng,lat float
"""
Return the center points of the tiles.

Index and NaN values are retained.
NaN are also returned in case of invalid tile IDs.

:param tile_ids: the tile IDs
:return: a tuple of 2 Series containing longitude and latitude, NaN when not applicable
:raises ValueError: in case the series doesn't contain integers # noqa
"""
_verify_integer_series(tile_ids)

def fn(t):
if not pd.isna(t) and ht.is_valid(t):
return ht.get_center_coordinates(t)
else:
return None

return pd.DataFrame.from_records(map(fn, tile_ids), tile_ids.index, columns=["lng", "lat"])

[docs]
def get_center_point(tile_ids: pd.Series) -> gpd.GeoSeries: # Point
"""
Return the center points of the tiles.

Index and NaN values are retained by returning None.
None are also returned in case of invalid tile IDs.

:param tile ids: the tile IDs
:return: a GeoSeries of points, None when not applicable
:raises ValueError: in case the series doesn't contain integers # noqa
"""
returnto_geoseries(tile_ids, ht.get_center_point)

[docs]
def get_corners(tile_ids: pd.Series) -> gpd.GeoDataFrame: # sw,ne Point
"""
Return the southwestern and northeastern corners of the tiles.

Index and NaN values are retained by returning None.
None are also returned in case of invalid tile IDs.

:param tile_ids: the tile IDs
:return: a GeoDataFrame with columns southwest, northeast. None when not applicable # noqa
:raises ValueError: in case the series doesn't contain integers or strings # noqa
"""
_verify_integer_series(tile_ids)

def to_corners(t):
corners = ht.get_corners(t) if not pd.isna(t) and ht.is_valid(t) else (None, None)
return {"southwest": corners[0], "northeast": corners[1]}

corners_list = list(map(to_corners, tile_ids))
sw = gpd.GeoSeries([corner["southwest"] for corner in corners_list], index=tile_ids.index)
ne = gpd.GeoSeries([corner["northeast"] for corner in corners_list], index=tile_ids.index)
return gpd.GeoDataFrame({"southwest": sw, "northeast": ne})

[docs]
def get_bounds(tile_ids: pd.Series) -> pd.DataFrame: # west,south,east,north float
"""
Return the bounds of the tiles.

This function returns bounds following the same
convention of the bounds function of shapely.

Index and NaN values are retained.
NaN are also returned in case of invalid tile IDs.

:param tile_ids: the tile IDs
:return: a DataFrame with columns west, south,
east, north. NaN when not applicable
:raises ValueError: in case the series doesn't contain integers # noqa
"""
_verify_integer_series(tile_ids)

def fn(t):
if not pd.isna(t) and ht.is_valid(t):
return ht.get_bounds(t)
else:
return None

return pd.DataFrame.from_records(
map(fn, tile_ids), tile_ids.index, columns=["west", "south", "east", "north"]
)

[docs]
def get_boundary_ring(tile_ids: pd.Series) -> gpd.GeoSeries: # LinearRing
"""
Return the boundary of the tiles as linear rings.

Each ring describes the boundary counter-clockwise.
It contains 5 points, since a linear ring is always closed
by repeating its first point.

Index and NaN values are retained by returning None.
None are also returned in case of invalid tile IDs.

:param tile ids: the tile IDs
:return: a gpd.GeoSeries of linear rings
:raises ValueError: in case the series doesn't contain integers # noqa
"""
returnto_geoseries(tile_ids, ht.get_boundary_ring)

[docs]
def get_polygon(tile_ids: pd.Series) -> gpd.GeoSeries: # Polygon
"""
Return the tiles as polygons.

Index and NaN values are retained by returning None.
None are also returned in case of invalid tile IDs.

:param tile ids: the tile IDs
:return: a gpd.GeoSeries of polygons
:raises ValueError: in case the series doesn't contain integers # noqa
"""
returnto_geoseries(tile_ids, ht.get_polygon)

[docs]
def contains_coordinates(
tile_ids: pd.Series, lngs: pd.Series, lats: pd.Series
) -> pd.Series: # bool
"""
Return if the tiles contains the points.

The check includes the tile boundary, aligned with the
way shapely intersects operates.

NaN values or not all the series containing a value for
a given index determine the resulting tile ID to be NaN.

:param tile_ids: the tile IDs
:param lngs: the Series with longitudes
:param lats: the Series with latitudes
:return: a pd.Series containing if the points are inside the tiles
:raises ValueError: in case the series doesn't contain integers # noqa
"""
df = pd.DataFrame({"tile_ids": tile_ids.astype("Int64"), "lng": lngs, "lat": lats})

def fn(s):
t, lng, lat = s
if not pd.isna(t) and ht.is_valid(t) and not pd.isna(lng) and not pd.isna(lat):
return ht.contains_coordinates(t, lng, lat)
else:
return pd.NA

Check if there's a better type for nullable boolean, or simply return false

return pd.Series(map(fn, df.itertuples(index=False)), index=df.index, dtype="object")

[docs]
def contains_point(tile_ids: pd.Series, pts: gpd.GeoSeries) -> pd.Series: # bool
"""
Return if the tiles contains the points.

This method is more efficient than contains_geometry for points.

The check includes the tile boundary, aligned with the
way shapely intersects operates.

NaN values or not all the series containing a value for
a given index determine the resulting tile ID to be NaN.

:param tile ids: the tile IDs
:param pts: the gpd.GeoSeries` with points :return: a pd.Series`` of booleans, if the points are inside the tiles
:raises ValueError: in case the series doesn't contain integers # noqa
"""
returnto_bool(tile_ids, pts, ht.contains_point)

[docs]
def contains_geometry(
tile_ids: pd.Series, geometries: gpd.GeoSeries, fully_contained=False
) -> pd.Series: # bool
"""
Return if the tiles contain the geometries.

Inclusion is modeled as the shapely relation intersects (fully_contained is false)
or contains (otherwise) between the tile and the geometry.

NaN values or not all the series containing a value for
a given index determine the resulting tile ID to be NaN.

:param tile_ids: the tile IDs
:param geometries: a gpd.GeoSeries` with any geometric type :param fully_contained: check if each geometry is fully contained in each tile :return: a pd.Series`` of booleans, if the geometries are inside the tiles
:raises ValueError: in case the series doesn't contain integers # noqa
"""

def predicate(tile_id, geo):
return ht.contains_geometry(tile_id, geo, fully_contained=fully_contained)

return _to_bool(tile_ids, geometries, predicate)

[docs]
def is_contained_in_geometry(
tile_ids: pd.Series, geometries: gpd.GeoSeries, fully_contained=False
) -> pd.Series: # bool
"""
Return if the tiles are contained in the geometries.

Inclusion is modeled as the shapely relation intersects (fully_contained is false)
or contains (otherwise) between the geometry and the tile.

:param tile_ids: the tile IDs
:param geometries: a gpd.GeoSeries` with any geometric type :param fully_contained: check if the tile is fully contained in the geometry :return: a pd.Series`` of booleans, if the tiles are contained in the geometries,
fully or partially
:raises ValueError: in case the series doesn't contain integers # noqa
"""

def predicate(tile_id, geo):
return ht.is_contained_in_geometry(tile_id, geo, fully_contained=fully_contained)

return _to_bool(tile_ids, geometries, predicate)

[docs]
def ancestor(tile_ids: pd.Series, level: Optional[int] = None) -> pd.Series: # tile id
"""
Return the ancestor tiles at the given level for the tiles.

Index and NaN values are retained. NaN is also returned in case
of invalid tile IDs or tiling level, or when the ancestor doesn't exist.

:param tile_ids: the tile IDs
:param level: the tiling level
:return: a pd.Series with the tile IDs of the ancestors or parents, when existing
:raises ValueError: in case the series doesn't contain integers # noqa
"""
_verify_integer_series(tile_ids)

def fn(t):
if not pd.isna(t) and ht.is_valid(t):
a = ht.ancestor(t, level)
return a if a else pd.NA
else:
return pd.NA

This syntax is preferred than apply(fn, axis=1) because it's possible

to set the dtype to Int64, otherwise None values determine a conversion to float64

return pd.Series(map(fn, tile_ids), index=tile_ids.index, dtype="Int64")

[docs]
def descendants(tile_ids: pd.Series, level: Optional[int] = None) -> pd.Series: # list of tile ids
"""
Return the tile IDs of all descendants at some level for each tile.

Index and NaN values are retained. NaN is also returned in place of
empty lists or in case of invalid tile IDs or tiling level.

Use pd.Series.explode method to turn the lists into multiple rows.

:param tile_ids: the tile IDs
:param level: the tiling level
:return: a pd.Series with the list of descendants tile IDs for each tile
:raises ValueError: in case the series doesn't contain integers # noqa
"""

def fn(t):
return ht.descendants(t, level)

return _to_tile_list(tile_ids, fn)

[docs]
def is_parent(tile_ids: pd.Series, of_tile_ids: pd.Series) -> pd.Series: # bool
"""
Check if tiles are the direct parents of other tiles.

Index and NaN values are retained.
Series should be indexed the same.

:param tile ids: the subject tile IDs
:param of_tile_ids: check the relationship of the subjects with these tiles
:return: a pd.Series containing if each tile is the parent of the other tile
:raises ValueError: in case the series doesn't contain integers # noqa
"""
returnto_bool(tile_ids, of_tile_ids, ht.is_parent)

[docs]
def is_descendant(tile_ids: pd.Series, of_tile_ids: pd.Series) -> pd.Series: # bool
"""
Check if tiles are the direct descendant of other tiles.

Index and NaN values are retained.
Series should be indexed the same.

:param tile ids: the subject tile IDs
:param of_tile_ids: check the relationship of the subjects with these tiles
:return: a pd.Series containing if each tile is the descendant of the other tile
:raises ValueError: in case the series doesn't contain integers # noqa
"""
returnto_bool(tile_ids, of_tile_ids, ht.is_descendant)

[docs]
def neighbors(tile_ids: pd.Series, level: Optional[int] = None) -> pd.Series: # list of tile ids
"""
Return tile IDs of all neighbors on given level for tiles.

Index and NaN values are retained. NaN is also returned in place of
empty lists or in case of invalid tile IDs or tiling level.

Use pd.Series.explode method to turn the lists into multiple rows.

:param tile_ids: the subject tile IDs
:param level: the tiling level
:return: a pd.Series with the list of neighboring tile IDs
:raises ValueError: in case the series doesn't contain integers # noqa
"""

def fn(t):
return ht.neighbors(t, level)

return _to_tile_list(tile_ids, fn)