"""
Module qui contient les fonctions utilitaires pour l'exportation des données.
Ce module contient les fonctions qui permettent de sauvegarder les données dans des fichiers
de différents formats.
"""
from pathlib import Path
from typing import Optional, Any
import geopandas as gpd
import i18n
import pandas as pd
from loguru import logger
from .crs import transform_geodataframe_crs
from .path import sanitize_path_name
from . import geotiff
LOGGER = logger.bind(name="CSB-Processing.Export")
WGS84 = 4326
def transform_additional_geometry_columns_to_wkt(
geodataframe: gpd.GeoDataFrame,
) -> gpd.GeoDataFrame:
"""
Transforme les colonnes de géométrie supplémentaires en WKT.
:param geodataframe: Le GeoDataFrame.
:type geodataframe: gpd.GeoDataFrame
:return: Le GeoDataFrame avec les colonnes de géométrie supplémentaires transformées en WKT.
:rtype: gpd.GeoDataFrame
"""
additional_geometry_columns: list[str] = [
col
for col in geodataframe.columns
if geodataframe[col].dtype == "geometry" and col != geodataframe.geometry.name
]
if additional_geometry_columns:
geodataframe: gpd.GeoDataFrame = geodataframe.copy()
for col in additional_geometry_columns:
geodataframe[col] = geodataframe[col].apply(
lambda geom: geom.wkt if geom else None
)
return geodataframe
def export_geodataframe(
geodataframe: gpd.GeoDataFrame,
driver: str,
output_path: Path,
to_epsg: Optional[int] = WGS84,
**kwargs,
) -> None:
"""
Sauvegarde le GeoDataFrame dans un fichier GeoJSON.
:param geodataframe: Le GeoDataFrame.
:type geodataframe: gpd.GeoDataFrame
:param output_path: Le chemin du fichier de sortie.
:type output_path: Path
:param driver: Le driver du fichier de sortie.
:type driver: str
:param to_epsg: Le code EPSG de la projection.
:type to_epsg: Optional[int]
"""
LOGGER.debug(
i18n.t(
"export.export_format.saving_geodataframe",
driver=driver,
output_path=output_path,
)
)
transform_geodataframe_crs(geodataframe=geodataframe, to_epsg=to_epsg)
geodataframe: gpd.GeoDataFrame = transform_additional_geometry_columns_to_wkt(
geodataframe
)
geodataframe.to_file(str(sanitize_path_name(output_path)), driver=driver, **kwargs)
[docs]
def export_geodataframe_to_geojson(
geodataframe: gpd.GeoDataFrame,
output_path: Path,
to_epsg: Optional[int] = WGS84,
**kwargs,
) -> None:
"""
Sauvegarde le GeoDataFrame dans un fichier GeoJSON.
:param geodataframe: Le GeoDataFrame.
:type geodataframe: gpd.GeoDataFrame
:param output_path: Le chemin du fichier de sortie.
:type output_path: Path
:param to_epsg: Le code EPSG de la projection.
:type to_epsg: Optional[int]
"""
export_geodataframe(
geodataframe, "GeoJSON", output_path, to_epsg=to_epsg
) # , RFC7946=True)
[docs]
def export_geodataframe_to_shapefile(
geodataframe: gpd.GeoDataFrame,
output_path: Path,
to_epsg: Optional[int] = WGS84,
**kwargs,
) -> None:
"""
Sauvegarde le GeoDataFrame dans un fichier Shapefile.
:param geodataframe: Le GeoDataFrame.
:type geodataframe: gpd.GeoDataFrame
:param output_path: Le chemin du fichier de sortie.
:type output_path: Path
:param to_epsg: Le code EPSG de la projection.
:type to_epsg: Optional[int]
"""
export_geodataframe(geodataframe, "ESRI Shapefile", output_path, to_epsg=to_epsg)
[docs]
def export_geodataframe_to_gpkg(
geodataframe: gpd.GeoDataFrame,
output_path: Path,
to_epsg: Optional[int] = WGS84,
**kwargs,
) -> None:
"""
Sauvegarde le GeoDataFrame dans un fichier GeoPackage.
:param geodataframe: Le GeoDataFrame.
:type geodataframe: gpd.GeoDataFrame
:param output_path: Le chemin du fichier de sortie.
:type output_path: Path
:param to_epsg: Le code EPSG de la projection.
:type to_epsg: Optional[int]
"""
export_geodataframe(geodataframe, "GPKG", output_path, to_epsg=to_epsg)
def export_geodataframe_to_csv(
geodataframe: gpd.GeoDataFrame,
output_path: Path,
to_epsg: Optional[int] = WGS84,
**kwargs,
) -> None:
"""
Sauvegarde le GeoDataFrame dans un fichier CSV.
:param geodataframe: Le GeoDataFrame.
:type geodataframe: gpd.GeoDataFrame
:param output_path: Le chemin du fichier de sortie.
:type output_path: Path
:param to_epsg: Le code EPSG de la projection.
:type to_epsg: Optional[int]
"""
LOGGER.debug(i18n.t("export.export_format.saving_csv", output_path=output_path))
# Transformer le système de coordonnées si nécessaire
transform_geodataframe_crs(geodataframe, to_epsg)
# Convertir le GeoDataFrame en DataFrame en supprimant la colonne de géométrie
df = geodataframe.drop(columns=geodataframe.geometry.name)
# Sauvegarder le DataFrame en CSV
export_dataframe_to_csv(df, output_path)
def export_geodataframe_to_parquet(
geodataframe: gpd.GeoDataFrame,
output_path: Path,
to_epsg: Optional[int] = WGS84,
**kwargs: Any,
) -> None:
"""
Sauvegarde le GeoDataFrame dans un fichier Parquet.
:param geodataframe: Le GeoDataFrame.
:type geodataframe: gpd.GeoDataFrame
:param output_path: Le chemin du fichier de sortie.
:type output_path: Path
:param to_epsg: Le code EPSG de la projection.
:type to_epsg: Optional[int]
"""
LOGGER.debug(i18n.t("export.export_format.saving_parquet", output_path=output_path))
transform_geodataframe_crs(geodataframe, to_epsg)
geodataframe.to_parquet(sanitize_path_name(output_path), index=False)
def export_geodataframe_to_feather(
geodataframe: gpd.GeoDataFrame,
output_path: Path,
to_epsg: Optional[int] = WGS84,
**kwargs: Any,
) -> None:
"""
Sauvegarde le GeoDataFrame dans un fichier Feather.
:param geodataframe: Le GeoDataFrame.
:type geodataframe: gpd.GeoDataFrame
:param output_path: Le chemin du fichier de sortie.
:type output_path: Path
:param to_epsg: Le code EPSG de la projection.
:type to_epsg: Optional[int]
"""
LOGGER.debug(i18n.t("export.export_format.saving_feather", output_path=output_path))
transform_geodataframe_crs(geodataframe, to_epsg)
geodataframe.to_feather(sanitize_path_name(output_path))
[docs]
def export_geodataframe_to_geotiff(
geodataframe: gpd.GeoDataFrame,
output_path: Path,
column: str = "Depth_processed_meter",
resolution: float = 0.00002,
to_epsg: Optional[int] = WGS84,
**kwargs: Any,
) -> None:
"""
Exporte un GeoDataFrame en fichier GeoTIFF en rasterisant les géométries
:param geodataframe: Le GeoDataFrame à exporter.
:type geodataframe: gpd.GeoDataFrame
:param output_path: Le chemin du fichier de sortie.
:type output_path: Path
:param column: Le nom de la colonne contenant les valeurs à rasteriser.
:type column: str
:param resolution: La résolution du raster en unités de la CRS
:type resolution: float
:param to_epsg: Le code EPSG de la CRS cible
:type to_epsg: Optional[int]
:raises ValueError: Si le GeoDataFrame n'a pas de CRS défini ou si les dimensions du raster sont invalides.
"""
geotiff.export_geodataframe_to_geotiff(
geodataframe=geodataframe,
output_path=output_path,
column=column,
resolution=resolution,
to_epsg=to_epsg,
**kwargs,
)
[docs]
def export_geodataframe_to_csar_api(
geo_dataframe: gpd.GeoDataFrame, output_path: Path, config_caris, **kwargs
) -> None:
"""
Sauvegarde le GeoDataFrame dans un fichier CSAR.
:param geo_dataframe: Le GeoDataFrame.
:type geo_dataframe: gpd.GeoDataFrame
:param output_path: Le chemin du fichier de sortie.
:type output_path: Path
:param config_caris: La configuration de l'API Caris.
:type config_caris: CarisAPIConfigProtocol
"""
# Importation au runtime pour éviter des problèmes de dépendances si Caris n'est pas installé
from caris_api import export_csar_api
LOGGER.debug(i18n.t("export.export_format.saving_csar", output_path=output_path))
export_csar_api.export_geodataframe_to_csar(
data=geo_dataframe,
output_path=sanitize_path_name(output_path),
config=config_caris,
)
def export_geodataframe_to_csar_batch(
geo_dataframe: gpd.GeoDataFrame,
output_path: Path,
config_caris,
**kwargs,
) -> None:
"""
Sauvegarde le GeoDataFrame dans un fichier CSAR.
:param geo_dataframe: Le GeoDataFrame.
:type geo_dataframe: gpd.GeoDataFrame
:param output_path: Le chemin du fichier de sortie.
:type output_path: Path
:param config_caris: La configuration de l'API Caris.
:type config_caris: CarisAPIConfigProtocol
"""
from caris_api import export_csar_batch
csv_path: Path = output_path.with_name(f"{output_path.stem}_temp").with_suffix(
".csv"
)
export_dataframe_to_csv(
dataframe=geo_dataframe.fillna("<NA>"), output_path=csv_path
)
LOGGER.debug(i18n.t("export.export_format.saving_csar", output_path=output_path))
export_csar_batch.export_geodataframe_to_csar(
data=csv_path,
output_path=sanitize_path_name(output_path),
config=config_caris,
)
csv_path.unlink()
[docs]
def export_dataframe_to_csv(
dataframe: pd.DataFrame, output_path: Path, **kwargs
) -> None:
"""
Sauvegarde le DataFrame dans un fichier CSV.
:param dataframe: Le DataFrame.
:type dataframe: pd.DataFrame
:param output_path: Le chemin du fichier de sortie.
:type output_path: Path
"""
LOGGER.debug(
i18n.t("export.export_format.saving_dataframe_csv", output_path=output_path)
)
dataframe.to_csv(sanitize_path_name(output_path), index=False)