"""
Module qui contient les fonctions pour manipuler les GeoDataFrame des polygones de Voronoi.
Les fonctions permettent de créer les polygones de Voronoi, de les joindre avec les stations et de fusionner les attributs.
"""
from multiprocessing import cpu_count
from typing import Optional, Collection
import dask_geopandas as dgpd
import geopandas as gpd
import shapely
from loguru import logger
import i18n
from shapely import (
Geometry,
GeometryCollection,
)
from shapely.constructive import concave_hull
from .voronoi_algorithm import create_voronoi_polygons
from .voronoi_models import TimeSeriesProtocol, StationsHandlerProtocol
import schema
from schema import model_ids as schema_ids
LOGGER = logger.bind(name="CSB-Processing.Tide.Voronoi.Geodataframe")
WGS84: int = 4326
"""EPSG code pouur le système de coordonnées WGS84."""
[docs]
def from_shapely_object_to_geodataframe(
geometry: shapely.geometry, epsg: Optional[int] = WGS84
) -> gpd.GeoDataFrame:
"""
Fonction qui transforme un objet Shapely en GeoDataFrame.
:param geometry: La géométrie Shapely à transformer.
:type geometry: shapely.geometry
:param epsg: Le code EPSG du CRS à utiliser.
:type epsg: Optional[int]
:return: Le GeoDataFrame.
:rtype: gpd.GeoDataFrame
"""
LOGGER.debug(
i18n.t(
"tide.voronoi.voronoi_geodataframe.transforming_geometry",
type=type(geometry).__name__,
)
)
if isinstance(geometry, GeometryCollection):
geometries = list(geometry.geoms)
else:
geometries = [geometry]
return gpd.GeoDataFrame(geometry=geometries, crs=epsg)
[docs]
def create_voronoi_gdf(geometry: Geometry) -> gpd.GeoDataFrame:
"""
Crée un GeoDataFrame de polygone de Voronoi à partir d'une geometrie Shapely.
:param geometry: La géométrie Shapely à utiliser pour créer les polygones de Voronoi.
:type geometry: Geometry
:return: Le GeoDataFrame contenant les polygones de Voronoi.
:rtype: gpd.GeoDataFrame
"""
voronoi: GeometryCollection = create_voronoi_polygons(geometry=geometry)
return from_shapely_object_to_geodataframe(geometry=voronoi)
[docs]
def join_stations_voronoi(
gdf_stations: gpd.GeoDataFrame, gdf_voronoi: gpd.GeoDataFrame
) -> gpd.GeoDataFrame:
"""
Crée une jointure spatiale entre les stations et les polygones de Voronoi.
:param gdf_stations: Le GeoDataFrame des stations.
:type gdf_stations: gpd.GeoDataFrame
:param gdf_voronoi: Le GeoDataFrame des polygones de Voronoi.
:type gdf_voronoi: gpd.GeoDataFrame
:return: Le GeoDataFrame joint.
:rtype: gpd.GeoDataFrame
"""
LOGGER.debug(i18n.t("tide.voronoi.voronoi_geodataframe.joining_stations_voronoi"))
return gpd.sjoin(gdf_stations, gdf_voronoi, how="inner", predicate="within")
[docs]
def merge_attributes(
gdf_voronoi: gpd.GeoDataFrame, gdf_joined: gpd.GeoDataFrame
) -> gpd.GeoDataFrame:
"""
Fusionne les attributs des stations avec les polygones de Voronoi.
:param gdf_voronoi: Le GeoDataFrame des polygones de Voronoi.
:type gdf_voronoi: gpd.GeoDataFrame
:param gdf_joined: Le GeoDataFrame avec le geodataframe des stations et des polygones de Voronoi joins.
:type gdf_joined: gpd.GeoDataFrame
:return: Le GeoDataFrame avec les attributs fusionnés.
:rtype: gpd.GeoDataFrame
"""
LOGGER.debug(i18n.t("tide.voronoi.voronoi_geodataframe.merging_attributes"))
gdf_voronoi = (
gdf_voronoi.merge(
gdf_joined,
left_index=True,
right_on="index_right",
)
.rename(
columns={
"geometry_x": schema_ids.GEOMETRY,
"geometry_y": schema_ids.STATION_POSITION,
}
)
.drop(columns=["index_right"])
)
return gdf_voronoi
[docs]
@schema.validate_schemas(return_schema=schema.TideZoneStationSchema)
def get_voronoi_geodataframe(
stations_handler: StationsHandlerProtocol,
time_series: Collection[TimeSeriesProtocol],
excluded_stations: Optional[Collection[str] | None] = None,
water_level_stations: Optional[Collection[str] | None] = None,
**kwargs,
) -> gpd.GeoDataFrame:
"""
Récupère le GeoDataFrame des polygones de Voronoi.
:param stations_handler: Gestionnaire des stations.
:type stations_handler: StationsHandlerABC
:param time_series: Liste des séries temporelles pour filtrer les stations.
:type time_series: Collection[TimeSeriesProtocol]
:param excluded_stations: Liste des stations à exclure.
:type excluded_stations: Collection[str] | None
:param water_level_stations: Liste des stations de niveau d'eau à utiliser.
:type water_level_stations: Collection[str] | None
:return: Le GeoDataFrame des polygones de Voronoi.
:rtype: gpd.GeoDataFrame[schema.TideZoneStationSchema]
"""
gdf_stations: gpd.GeoDataFrame[schema.StationsSchema] = (
(
stations_handler.get_stations_geodataframe(
filter_time_series=time_series,
excluded_stations=excluded_stations,
**kwargs,
)
)
if not water_level_stations
else stations_handler.get_stations_geodataframe_from_codes(
water_level_stations, filter_time_series=time_series
)
)
gdf_voronoi: gpd.GeoDataFrame = create_voronoi_gdf(
geometry=gdf_stations.geometry.unary_union
)
gdf_joined: gpd.GeoDataFrame = join_stations_voronoi(
gdf_stations=gdf_stations, gdf_voronoi=gdf_voronoi
)
gdf_voronoi: gpd.GeoDataFrame[schema.TideZoneStationSchema] = merge_attributes(
gdf_voronoi=gdf_voronoi, gdf_joined=gdf_joined
)
return gdf_voronoi
[docs]
def get_polygon_by_station_id(
gdf_voronoi: gpd.GeoDataFrame, station_id: str
) -> gpd.GeoDataFrame:
"""
Récupère le polygone de Voronoi pour une station donnée.
:param gdf_voronoi: Le GeoDataFrame des polygones de Voronoi.
:type gdf_voronoi: gpd.GeoDataFrame[schema.TideZoneStationSchema]
:param station_id: L'identifiant de la station.
:type station_id: str
:return: Le polygone de Voronoi de la station.
:rtype: gpd.GeoDataFrame
"""
LOGGER.debug(
i18n.t(
"tide.voronoi.voronoi_geodataframe.getting_polygon", station_id=station_id
)
)
return gdf_voronoi.loc[gdf_voronoi[schema_ids.ID] == station_id]
[docs]
def get_time_series_by_station_id(
gdf_voronoi: gpd.GeoDataFrame, station_id: str
) -> list[str]:
"""
Récupère les séries temporelles pour une station donnée.
:param gdf_voronoi: Le GeoDataFrame des polygones de Voronoi.
:type gdf_voronoi: gpd.GeoDataFrame[schema.TideZoneStationSchema]
:param station_id: L'identifiant de la station.
:type station_id: str
:return: Les séries temporelles de la station.
:rtype: list[str]
"""
LOGGER.debug(
i18n.t(
"tide.voronoi.voronoi_geodataframe.getting_time_series",
station_id=station_id,
)
)
return gdf_voronoi.loc[gdf_voronoi[schema_ids.ID] == station_id][
schema_ids.TIME_SERIES
].values[0]
[docs]
def get_code_by_station_id(gdf_voronoi: gpd.GeoDataFrame, station_id: str) -> str:
"""
Récupère le code de la station.
:param gdf_voronoi: Le GeoDataFrame des polygones de Voronoi.
:type gdf_voronoi: gpd.GeoDataFrame[schema.TideZoneStationSchema]
:param station_id: L'identifiant de la station.
:type station_id: str
:return: Le code de la station.
:rtype: str
"""
LOGGER.debug(
i18n.t("tide.voronoi.voronoi_geodataframe.getting_code", station_id=station_id)
)
return gdf_voronoi.loc[gdf_voronoi[schema_ids.ID] == station_id][
schema_ids.CODE
].values[0]
[docs]
def get_name_by_station_id(gdf_voronoi: gpd.GeoDataFrame, station_id: str) -> str:
"""
Récupère le nom de la station.
:param gdf_voronoi: Le GeoDataFrame des polygones de Voronoi.
:type gdf_voronoi: gpd.GeoDataFrame[schema.TideZoneStationSchema]
:param station_id: L'identifiant de la station.
:type station_id: str
:return: Le nom de la station.
:rtype: str
"""
LOGGER.debug(
i18n.t("tide.voronoi.voronoi_geodataframe.getting_name", station_id=station_id)
)
return gdf_voronoi.loc[gdf_voronoi[schema_ids.ID] == station_id][
schema_ids.NAME
].values[0]
[docs]
def get_station_position_by_station_id(
gdf_voronoi: gpd.GeoDataFrame, station_id: str
) -> gpd.GeoSeries:
"""
Récupère la position de la station.
:param gdf_voronoi: Le GeoDataFrame des polygones de Voronoi.
:type gdf_voronoi: gpd.GeoDataFrame[schema.TideZoneStationSchema]
:param station_id: L'identifiant de la station.
:type station_id: str
:return: La position de la station.
:rtype: gpd.GeoSeries
"""
LOGGER.debug(
i18n.t(
"tide.voronoi.voronoi_geodataframe.getting_position", station_id=station_id
)
)
return gdf_voronoi.loc[gdf_voronoi[schema_ids.ID] == station_id][
schema_ids.STATION_POSITION
]
[docs]
def get_polygon_by_geometry(
gdf_voronoi: gpd.GeoDataFrame,
geometry: gpd.GeoDataFrame,
) -> gpd.GeoDataFrame:
"""
Récupère les polygones de Voronoi qui intersectent les géométries données.
:param gdf_voronoi: Le GeoDataFrame des polygones de Voronoi.
:type gdf_voronoi: gpd.GeoDataFrame[schema.TideZoneStationSchema]
:param geometry: La géométrie a utilisé pour l'intersection.
:type geometry: gpd.GeoDataFrame
:return: Le GeoDataFrame des polygones de Voronoi qui intersectent les géométries.
:rtype: gpd.GeoDataFrame[schema.TideZoneSchema]
"""
LOGGER.debug(
i18n.t(
"tide.voronoi.voronoi_geodataframe.getting_polygon_by_geometry",
)
)
dask_gdf_voronoi: dgpd.GeoDataFrame[schema.TideZoneStationSchema] = (
dgpd.from_geopandas(gdf_voronoi, npartitions=cpu_count())
)
dask_geometry: dgpd.GeoDataFrame = dgpd.from_geopandas(geometry, npartitions=12)
result: dgpd.GeoDataFrame = dgpd.sjoin(
dask_gdf_voronoi, dask_geometry, how="inner", predicate="intersects"
)
result_computed: gpd.GeoDataFrame = result.compute()
result_unique: gpd.GeoDataFrame[schema.TideZoneStationSchema] = (
result_computed.drop_duplicates(subset=[schema_ids.ID]).reset_index(drop=True)
)
return result_unique[gdf_voronoi.columns]
[docs]
def get_concave_hull(
geometry: Geometry, ratio: float = 0.5, allow_holes: bool = True
) -> Geometry:
"""
Récupère l'enveloppe concave des polygones de Voronoi.
:param geometry: La géométrie a utilisé pour l'enveloppe concave.
:type geometry: Geometry
:param ratio: Le ratio de l'enveloppe concave.
:type ratio: float
:param allow_holes: Autorise les trous dans l'enveloppe concave.
:type allow_holes: bool
:return: L'enveloppe concave des polygones.
:rtype: Geometry
"""
LOGGER.debug(
i18n.t(
"tide.voronoi.voronoi_geodataframe.creating_concave_hull",
ratio=ratio,
)
)
return concave_hull(geometry, ratio=ratio, allow_holes=allow_holes)
[docs]
def get_station_title(gdf_voronoi: gpd.GeoDataFrame, station_id: str) -> str:
"""
Retourne le titre formaté d'une station marégraphique (nom + code).
:param gdf_voronoi: GeoDataFrame des polygones de Voronoi.
:type gdf_voronoi: gpd.GeoDataFrame[schema.TideZoneStationSchema]
:param station_id: Identifiant de la station.
:type station_id: str
:return: Titre de la station au format ``"Nom (code)"``.
:rtype: str
"""
return (
f"{get_name_by_station_id(gdf_voronoi=gdf_voronoi, station_id=station_id)} "
f"({get_code_by_station_id(gdf_voronoi=gdf_voronoi, station_id=station_id)})"
)