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
  7from __future__ import annotations
  8
  9import json
 10from functools import partial
 11from typing import Optional
 12
 13import requests
 14from geopandas import GeoDataFrame, GeoSeries
 15from pandas import DataFrame, Series
 16from shapely import wkt, GEOSException, wkb
 17
 18from apb_extra_utils.postgres_pckg.psql_alchemy import EngPsqlAlchemy
 19from apb_extra_utils.utils_logging import get_base_logger
 20from apb_pandas_utils import df_from_url, df_from_pg_table
 21
 22logger = get_base_logger(__name__)
 23
 24
 25def gdf_to_geojson(gdf: GeoDataFrame, name: Optional[str] = None, with_crs: bool = True, show_bbox: bool = True,
 26                   drop_id: bool = False, path_file: str = None) -> dict:
 27    """
 28    Convierte un GeoDataFrame a diccionario geojson
 29
 30    Args:
 31        gdf (GeoDataFrame):
 32        name (str=None):
 33        with_crs (bool=True):
 34        show_bbox (bool=True):
 35        drop_id (bool=False):
 36        path_file (str=None): Si se indica se guarda el geojson en el path indicado
 37
 38    Returns:
 39        dict_geojson (dict)
 40    """
 41    dict_geojson = gdf.to_geo_dict(show_bbox=show_bbox, drop_id=drop_id)
 42    if name:
 43        dict_geojson["name"] = name
 44    if with_crs and gdf.crs is not None:
 45        auth = gdf.crs.to_authority()
 46        dict_geojson["crs"] = {"type": "name", "properties": {"name": f"urn:ogc:def:crs:{auth[0]}::{auth[1]}"}}
 47
 48    if path_file:
 49        geojson = json.dumps(dict_geojson, default=str, ensure_ascii=False)
 50        with open(path_file, 'w', encoding='utf-8') as f:
 51            f.write(geojson)
 52
 53    return dict_geojson
 54
 55
 56def gdf_to_df(gdf: GeoDataFrame, as_wkb=False) -> DataFrame:
 57    """
 58    Convert a GeoDataFrame to DataFrame converting the geometry columns to a str column in WKT format (WKB if as_wkb=True)
 59
 60    Args:
 61        gdf (GeoDataFrame):
 62        as_wkb (bool=False): If True, the geometry column is converted to WKB format
 63
 64    Returns:
 65        DataFrame
 66    """
 67    f_conv = 'to_wkb' if as_wkb else 'to_wkt'
 68
 69    # Convert all columns type geometry to WKT
 70    gdf_aux = gdf.copy()
 71    for col in df_geometry_columns(gdf_aux):
 72        gdf_aux[col] = getattr(gdf_aux[col], f_conv)()
 73    return DataFrame(gdf_aux)
 74
 75
 76def df_geometry_columns(df: GeoDataFrame | DataFrame) -> list:
 77    """
 78    Devuelve las columnas tipo geometría de un GeoDataFrame
 79
 80    Args:
 81        df (GeoDataFrame | DataFrame):
 82
 83    Returns:
 84        list
 85    """
 86    return df.select_dtypes(include=["geometry"]).columns.tolist()
 87
 88
 89def df_to_crs(df: GeoDataFrame | DataFrame, crs: str) -> GeoDataFrame | DataFrame:
 90    """
 91    Convierte todas las columnas tipo geometría de un GeoDataFrame o DataFrame al CRS indicado
 92
 93    Args:
 94        df (GeoDataFrame | DataFrame):
 95        crs (str): name CRS (EPSG) coord .sys. destino de las geometrías (e.g. 'EPSG:25831')
 96                    [Can be anything accepted by pyproj.CRS.from_user_input()]
 97
 98    Returns:
 99        GeoDataFrame | DataFrame
100    """
101    df_aux = df.copy()
102    for geom in df_geometry_columns(df_aux):
103        df_aux[geom] = df_aux[geom].to_crs(crs)
104
105    df_aux = df_aux.to_crs(crs)
106
107    return df_aux
108
109
110def gdf_from_df(df: DataFrame, geom_col: str, crs: str | None, cols_geom: list[str] = None) -> GeoDataFrame:
111    """
112    Crea un GeoDataFrame a partir de un DataFrame
113
114    Args:
115        df (DataFrame):
116        geom_col (str): Columna geometría con el que se creará el GeoDataFrame
117        crs (str | None): CRS (EPSG) coord .sys. origen de las geometrías (e.g. 'EPSG:25831')
118                    [Can be anything accepted by pyproj.CRS.from_user_input()]
119        cols_geom (list=None): Columnas con geometrías
120
121    Returns:
122        GeoDataFrame
123    """
124    if cols_geom is None:
125        cols_geom = []
126
127    cols_geom = set(cols_geom)
128    cols_geom.add(geom_col)
129
130    df_aux = df.copy()
131    idx_prev = df_aux.index
132    # We only deal with index when has names setted referred to possible columns
133    set_idx = None not in idx_prev.names
134    if set_idx:
135        df_aux.reset_index(inplace=True)
136
137    def convert_to_wkt(val_col):
138        return wkt.loads(val_col) if isinstance(val_col, str) else None
139
140    def convert_to_wkb(val_col):
141        return wkb.loads(val_col) if isinstance(val_col, str) else None
142
143    gdf = GeoDataFrame(df_aux)
144    for col in (col for col in gdf.columns if col in cols_geom):
145        ds_col = gdf[col]
146        if isinstance(ds_col, GeoSeries):
147            continue
148
149        if (dtype := ds_col.dtype.name) in ('str', 'object'):
150            try:
151                gdf[col] = gdf[col].apply(convert_to_wkt)
152            except GEOSException:
153                gdf[col] = gdf[col].apply(convert_to_wkb)
154
155        if crs:
156            gdf.set_geometry(col, inplace=True, crs=crs)
157        else:
158            gdf.set_geometry(col, inplace=True)
159
160    if set_idx:
161        gdf = gdf.set_index(idx_prev.names, drop=True)
162
163    if crs:
164        gdf.set_geometry(geom_col, crs=crs, inplace=True)
165    else:
166        gdf.set_geometry(geom_col, inplace=True)
167
168    return gdf
169
170
171def gdf_from_pg_table(table: str, geom_col: str | None = None, filter_sql: str | None = None,
172                      crs_gdf: str | None = None,
173                      add_goto_url: bool = False, user: str | None = None, psw: str | None = None,
174                      srvr_db: str = 'localhost', port_db: int = 5432, db: str = 'postgres',
175                      schemas: str | None = None, a_logger=None,
176                      url_conn_string: str | None = None) -> GeoDataFrame | DataFrame:
177    """
178    Carga una tabla/vista de PostgreSQL como GeoDataFrame cuando existe una columna geométrica usable.
179
180    Si no se informa ``geom_col``, intenta detectar la primera geometría con ``EngPsqlAlchemy.geoms_table``.
181    Si no hay geometrías hace fallback a ``df_from_pg_table``.
182    Si la columna indicada no existe/no es geométrica lanza exception ValueError
183
184    Args:
185        table (str): Nombre de tabla o vista.
186        geom_col (str | None): Columna geométrica a usar como activa.
187        filter_sql (str | None): Condición SQL opcional para cláusula WHERE. ATENCIÓN: en PG si hay campos en MAYÚSCULAS hay que indicarlos entre ""
188        crs_gdf (str | None): CRS destino opcional (e.g. ``'EPSG:4326'``).
189        add_goto_url (bool): Si True añade columna ``goto_url`` basada en centroides.
190        user (str | None): Usuario PostgreSQL.
191        psw (str | None): Password PostgreSQL.
192        srvr_db (str): Host del servidor.
193        port_db (int): Puerto del servidor.
194        db (str): Nombre de la base de datos.
195        schemas (str | None): Schemas separados por coma para ``search_path``.
196        a_logger (logging.Logger | None): Logger opcional.
197        url_conn_string (str | sqlalchemy.engine.urlURL | None): Connection string completa o URL de SQLAlchemy (ver apb_extra_utils.postgres_pckg.psql_alchemy.url_pg_string_connection). Si se proporciona, se ignoran los parámetros anteriores.
198
199    Returns:
200        GeoDataFrame | DataFrame
201    """
202    df_call = partial(
203        df_from_pg_table,
204        table=table,
205        filter_sql=filter_sql,
206        user=user,
207        psw=psw,
208        srvr_db=srvr_db,
209        port_db=port_db,
210        db=db,
211        schemas=schemas,
212        a_logger=a_logger,
213        url_conn_string=url_conn_string,
214    )
215
216    def _fallback_df(reason: str) -> DataFrame:
217        logger.warning(f"{reason} Fallback a df_from_pg_table para '{table}'.")
218        return df_call()
219
220    eng = EngPsqlAlchemy.get_cached(
221        user=user,
222        psw=psw,
223        srvr_db=srvr_db,
224        port_db=port_db,
225        db=db,
226        schemas=schemas,
227        a_logger=a_logger,
228        url_conn_string=url_conn_string,
229    )
230
231    try:
232        geoms_info = eng.geoms_table_view(table)
233    except Exception as exc:
234        return _fallback_df(f"No se pudo inspeccionar geometrías ({exc}).")
235
236    if not geoms_info:
237        return _fallback_df("La tabla/vista no tiene columnas geométricas.")
238
239    selected_geom_col = geom_col if geom_col else next(iter(geoms_info.keys()))
240    if selected_geom_col not in geoms_info:
241        raise ValueError(f"geom_col='{selected_geom_col}' no existe o no es geométrica.")
242
243    df = df_call()
244
245    if selected_geom_col not in df.columns:
246        return _fallback_df(f"La columna geométrica '{selected_geom_col}' no está en el resultado SQL.")
247
248    srid = geoms_info[selected_geom_col].get('srid') if isinstance(geoms_info[selected_geom_col], dict) else None
249    src_crs = f"EPSG:{srid}" if isinstance(srid, int) and srid > 0 else None
250
251    if not src_crs and crs_gdf:
252        logger.warning("No se ha podido determinar SRID de la geometría; no se reproyecta a crs_gdf.")
253
254    gdf = gdf_from_df(df=df, geom_col=selected_geom_col, crs=src_crs)
255
256    if add_goto_url:
257        add_goto_google_maps_url_to_gdf(gdf)
258
259    if crs_gdf and src_crs:
260        gdf = gdf.to_crs(crs_gdf)
261
262    return gdf
263
264
265def gdf_from_url(url_rest_api: str, api_params: dict | None = None, crs_api: str | None = None,
266                 headers: dict | None = None, crs_gdf: str | None = None, add_goto_url: bool = False,
267                 features_key: str = 'features', next_key: str = 'next', results_key: str = 'results',
268                 timeout: int | tuple[int, int] = (10, 30), max_retries: int = 3,
269                 session: requests.Session | None = None) -> GeoDataFrame | DataFrame | None:
270    """
271    Fetch paginated GeoJSON from a REST API and return a GeoPandas GeoDataFrame.
272
273    Delegates HTTP handling and pagination to :func:`apb_pandas_utils._fetch_pages`.
274    Supports FeatureCollection responses (``features`` key) and optional CRS reprojection.
275
276    Args:
277        url_rest_api (str): The base URL of the API endpoint.
278        api_params (dict, optional): Query parameters for the initial request.
279        crs_api (str, optional): CRS of the geometries returned by the API
280            (e.g. ``'EPSG:25831'``). Passed to :meth:`GeoDataFrame.from_features`.
281        headers (dict, optional): HTTP headers for the request.
282        crs_gdf (str, optional): Target CRS to reproject the GeoDataFrame to
283            after fetching (e.g. ``'EPSG:4326'``). No reprojection if ``None``.
284        add_goto_url (bool): If ``True``, adds a ``'goto_url'`` column with a
285            Google Maps link for each feature centroid. Defaults to ``False``.
286        features_key (str): Key in the JSON response dict that contains the
287            GeoJSON features list. Defaults to ``'features'``.
288        next_key (str): Key in the JSON response containing the next-page URL.
289            Defaults to ``'next'``.
290        results_key (str): Key in the JSON response containing the results:
291        timeout (int | tuple[int, int]): Request timeout ``(connect, read)`` in seconds.
292            Defaults to ``(10, 30)``.
293        max_retries (int): Retries on transient HTTP errors (429, 500-504).
294            Defaults to ``3``.
295        session (requests.Session, optional): Existing session to reuse.
296            If None, a new session with retry logic is created and closed after use.
297
298    Returns:
299        GeoDataFrame | Dataframe | None: A GeoDataFrame with all features, or ``None`` if empty.
300
301    Raises:
302        requests.HTTPError: If any HTTP request returns an error status.
303        requests.ConnectionError: If the connection fails after all retries.
304        ValueError: If a page response has an unexpected structure.
305    """
306    from apb_pandas_utils import _iter_fetch_pages
307
308    all_features: list = []
309    try_as_df: bool = False
310
311    for data in _iter_fetch_pages(url_rest_api, api_params, headers, next_key, timeout, max_retries, session):
312        page_features = None
313
314        if isinstance(data, list):
315            page_features = data
316
317        elif isinstance(data, dict):
318            # Standard GeoJSON FeatureCollection — look for features_key directly
319            # or nested inside a 'results' wrapper (e.g. DRF + djangorestframework-gis)
320            if features_key in data:
321                page_features = data[features_key]
322            elif results_key in data and isinstance(data[results_key], dict):
323                page_features = data[results_key].get(features_key, data)
324        else:
325            raise ValueError(
326                f"Unexpected JSON structure: expected list or dict, got {type(data).__name__}"
327            )
328
329        if page_features and len(page_features) > 0 and isinstance(page_features[0], dict) and page_features[0].get(
330                'type') == 'Feature':
331            all_features.extend(page_features)
332            logger.debug(f"Got {len(page_features)} features (total so far: {len(all_features)})")
333        else:
334            logger.warning(f"Data with unexpected structure for GeoDataframe! Using instead as Dataframe")
335            try_as_df = True
336            break
337
338    if try_as_df:
339        return df_from_url(
340            url_rest_api=url_rest_api,
341            api_params=api_params,
342            headers=headers,
343            results_key=results_key,
344            next_key=next_key,
345            max_retries=max_retries,
346            session=session)
347
348    if not all_features:
349        return None
350
351    gdf = GeoDataFrame.from_features(all_features, crs=crs_api)
352    logger.debug(f"GeoDataFrame created with {len(gdf)} rows and CRS={crs_api}")
353
354    if add_goto_url:
355        add_goto_google_maps_url_to_gdf(gdf)
356
357    if crs_gdf:
358        gdf = gdf.to_crs(crs_gdf)
359        logger.debug(f"Reprojected GeoDataFrame to CRS={crs_gdf}")
360
361    return gdf
362
363
364def add_goto_google_maps_url_to_gdf(gdf: GeoDataFrame):
365    """
366    Add goto URL to Google Maps URL to GeoDataFrame on column 'goto_url'.
367
368    Args:
369        gdf (GeoDataFrame): GeoDataFrame to add URL to.
370
371    Returns:
372        None: Modifies the GeoDataFrame in place.
373    """
374    centroids = gdf.geometry.centroid.to_crs('EPSG:4326')
375    mask = centroids.notna()
376    gdf['goto_url'] = Series([None] * len(gdf), index=gdf.index)
377    gdf.loc[mask, 'goto_url'] = centroids.loc[mask].apply(
378        lambda p: f"https://www.google.com/maps?q={p.y},{p.x}"
379    )