apb_pandas_utils.geopandas_utils

  1#  coding=utf-8
  2#
  3#  Author: Ernesto Arredondo Martinez (ernestone@gmail.com)
  4#  Created: 7/6/19 18:23
  5#  Last modified: 7/6/19 18:21
  6#  Copyright (c) 2019
  7import json
  8from typing import Optional
  9
 10from geopandas import GeoDataFrame, GeoSeries
 11from pandas import DataFrame
 12from shapely import wkt
 13
 14
 15def gdf_to_geojson(gdf: GeoDataFrame, name: Optional[str] = None, with_crs: bool = True, show_bbox: bool = True,
 16                   drop_id: bool = False, path_file: str = None) -> dict:
 17    """
 18    Convierte un GeoDataFrame a diccionario geojson
 19
 20    Args:
 21        gdf (GeoDataFrame):
 22        name (str=None):
 23        with_crs (bool=True):
 24        show_bbox (bool=True):
 25        drop_id (bool=False):
 26        path_file (str=None): Si se indica se guarda el geojson en el path indicado
 27
 28    Returns:
 29        dict_geojson (dict)
 30    """
 31    dict_geojson = gdf.to_geo_dict(show_bbox=show_bbox, drop_id=drop_id)
 32    if name:
 33        dict_geojson["name"] = name
 34    if with_crs and gdf.crs is not None:
 35        auth = gdf.crs.to_authority()
 36        dict_geojson["crs"] = {"type": "name", "properties": {"name": f"urn:ogc:def:crs:{auth[0]}::{auth[1]}"}}
 37
 38    if path_file:
 39        geojson = json.dumps(dict_geojson, default=str, ensure_ascii=False)
 40        with open(path_file, 'w', encoding='utf-8') as f:
 41            f.write(geojson)
 42
 43    return dict_geojson
 44
 45
 46def gdf_to_df(gdf: GeoDataFrame, as_wkb=False) -> DataFrame:
 47    """
 48    Convert a GeoDataFrame to DataFrame converting the geometry columns to a str column in WKT format (WKB if as_wkb=True)
 49
 50    Args:
 51        gdf (GeoDataFrame):
 52        as_wkb (bool=False): If True, the geometry column is converted to WKB format
 53
 54    Returns:
 55        DataFrame
 56    """
 57    f_conv = 'to_wkb' if as_wkb else 'to_wkt'
 58
 59    # Convert all columns type geometry to WKT
 60    gdf_aux = gdf.copy()
 61    for col in df_geometry_columns(gdf_aux):
 62        gdf_aux[col] = getattr(gdf_aux[col], f_conv)()
 63    return DataFrame(gdf_aux)
 64
 65
 66def df_geometry_columns(df: GeoDataFrame | DataFrame) -> list:
 67    """
 68    Devuelve las columnas tipo geometría de un GeoDataFrame
 69
 70    Args:
 71        df (GeoDataFrame | DataFrame):
 72
 73    Returns:
 74        list
 75    """
 76    return df.select_dtypes(include=["geometry"]).columns.tolist()
 77
 78
 79def df_to_crs(df: GeoDataFrame | DataFrame, crs: str) -> GeoDataFrame | DataFrame:
 80    """
 81    Convierte todas las columnas tipo geometría de un GeoDataFrame o DataFrame al CRS indicado
 82
 83    Args:
 84        df (GeoDataFrame | DataFrame):
 85        crs (str): name CRS (EPSG) coord .sys. destino de las geometrías (e.g. 'EPSG:25831')
 86                    [Can be anything accepted by pyproj.CRS.from_user_input()]
 87
 88    Returns:
 89        GeoDataFrame | DataFrame
 90    """
 91    df_aux = df.copy()
 92    for geom in df_geometry_columns(df_aux):
 93        df_aux[geom] = df_aux[geom].to_crs(crs)
 94
 95    df_aux = df_aux.to_crs(crs)
 96
 97    return df_aux
 98
 99
100def gdf_from_df(df: DataFrame, geom_col: str, crs: str, cols_geom: list[str] = None) -> GeoDataFrame:
101    """
102    Crea un GeoDataFrame a partir de un DataFrame
103
104    Args:
105        df (DataFrame):
106        geom_col (str): Columna geometría con el que se creará el GeoDataFrame
107        crs (str): CRS (EPSG) coord .sys. origen de las geometrías (e.g. 'EPSG:25831')
108                    [Can be anything accepted by pyproj.CRS.from_user_input()]
109        cols_geom (list=None): Columnas con geometrías
110
111    Returns:
112        GeoDataFrame
113    """
114    if cols_geom is None:
115        cols_geom = []
116
117    cols_geom = set(cols_geom)
118    cols_geom.add(geom_col)
119
120    df_aux = df.copy()
121    idx_prev = df_aux.index
122    # We only deal with index when has names setted referred to possible columns
123    set_idx = None not in idx_prev.names
124    if set_idx:
125        df_aux.reset_index(inplace=True)
126
127    def convert_to_wkt(val_col):
128        return wkt.loads(val_col) if isinstance(val_col, str) else None
129
130    gdf = GeoDataFrame(df_aux)
131    for col in (col for col in gdf.columns if col in cols_geom):
132        ds_col = gdf[col]
133        if isinstance(ds_col, GeoSeries):
134            continue
135
136        if (dtype := ds_col.dtype.name) == 'object':
137            gdf[col] = gdf[col].apply(convert_to_wkt)
138
139        gdf.set_geometry(col, inplace=True, crs=crs)
140
141    if set_idx:
142        gdf = gdf.set_index(idx_prev.names, drop=True)
143
144    gdf.set_geometry(geom_col, crs=crs, inplace=True)
145
146    return gdf