meerschaum.utils.dataframe

   1#! /usr/bin/env python3
   2# -*- coding: utf-8 -*-
   3# vim:fenc=utf-8
   4
   5"""
   6Utility functions for working with DataFrames.
   7"""
   8
   9from __future__ import annotations
  10
  11from datetime import datetime, date
  12from collections import defaultdict
  13
  14import meerschaum as mrsm
  15from meerschaum.utils.typing import (
  16    Optional, Dict, Any, List, Hashable, Generator,
  17    Iterator, Iterable, Union, TYPE_CHECKING, Tuple,
  18)
  19from meerschaum import _core
  20
  21if TYPE_CHECKING:
  22    pd, dask = mrsm.attempt_import('pandas', 'dask')
  23
  24
  25#: The DataFrame libraries Meerschaum returns frames in (`meerschaum:dataframe`).
  26DATAFRAME_LIBRARIES: Tuple[str, ...] = ('polars', 'pandas')
  27
  28#: The environment variable that overrides `meerschaum:dataframe` for one process (and the jobs
  29#: and workers it starts).
  30DATAFRAME_ENV_VAR: str = 'MRSM_DATAFRAME'
  31
  32#: The library of a root that does not say (a fresh root writes `dataframe: polars`).
  33DEFAULT_DATAFRAME_LIBRARY: str = 'polars'
  34
  35#: The extra that installs each library.
  36DATAFRAME_EXTRAS: Dict[str, str] = {'polars': 'polars', 'pandas': 'pandas'}
  37
  38
  39def _normalize_dataframe_library(value: Any, source: str) -> str:
  40    """Return `'polars'` or `'pandas'` for a configured value, or raise naming `source`."""
  41    if value is None:
  42        return DEFAULT_DATAFRAME_LIBRARY
  43    text = str(value).strip().lower()
  44    if not text:
  45        return DEFAULT_DATAFRAME_LIBRARY
  46    if text not in DATAFRAME_LIBRARIES:
  47        raise ValueError(
  48            f"Unknown DataFrame library '{value}' ({source}): use 'polars' or 'pandas'."
  49        )
  50    return text
  51
  52
  53def get_dataframe_library() -> str:
  54    """
  55    Return the DataFrame library Meerschaum returns frames in: `'polars'` or `'pandas'`.
  56
  57    The setting is `dataframe` in `meerschaum.yaml` (config path `meerschaum:dataframe`), and the
  58    `MRSM_DATAFRAME` environment variable overrides it for one process (and the jobs it starts).
  59    A fresh root writes `dataframe: polars`; a root created by Meerschaum 4 or by an earlier 5.0
  60    release candidate is given `dataframe: pandas` the first time 5.0.0rc15 or newer opens it.
  61
  62    Every API that returns a DataFrame follows it (`Pipe.get_data()`, `SQLConnector.read()`,
  63    `Pipe.fetch()`, `Plan.to_dataframe()`, the frames handed to plugin code, ...), and
  64    `as_polars=True` / `as_pandas=True` override it for one call.
  65
  66    Returns
  67    -------
  68    `'polars'` or `'pandas'`.
  69
  70    Examples
  71    --------
  72    >>> import os
  73    >>> os.environ['MRSM_DATAFRAME'] = 'pandas'
  74    >>> get_dataframe_library()
  75    'pandas'
  76    """
  77    import os
  78    value = os.environ.get(DATAFRAME_ENV_VAR, None)
  79    if value is not None and value.strip():
  80        return _normalize_dataframe_library(value, DATAFRAME_ENV_VAR)
  81    from meerschaum.config import get_config
  82    return _normalize_dataframe_library(
  83        get_config('meerschaum', 'dataframe', warn=False),
  84        'meerschaum:dataframe',
  85    )
  86
  87
  88def resolve_dataframe_library(
  89    as_polars: Optional[bool] = None,
  90    as_pandas: Optional[bool] = None,
  91) -> str:
  92    """
  93    Return the library one call returns its frames in: `'polars'` or `'pandas'`.
  94
  95    Parameters
  96    ----------
  97    as_polars: Optional[bool], default None
  98        `True` asks for Polars and `False` for pandas (Meerschaum 4's meaning of
  99        `as_polars=False`). `None` follows `get_dataframe_library()`.
 100
 101    as_pandas: Optional[bool], default None
 102        `True` asks for pandas and `False` for Polars. `None` follows the setting.
 103
 104    Returns
 105    -------
 106    `'polars'` or `'pandas'`. Asking for both (or for neither) raises a `ValueError`.
 107    """
 108    wants_polars = None if as_polars is None else bool(as_polars)
 109    wants_pandas = None if as_pandas is None else bool(as_pandas)
 110    if wants_polars is not None and wants_pandas is not None and wants_polars == wants_pandas:
 111        raise ValueError(
 112            "`as_polars` and `as_pandas` contradict each other: pass only one of them."
 113        )
 114    if wants_polars is not None:
 115        return 'polars' if wants_polars else 'pandas'
 116    if wants_pandas is not None:
 117        return 'pandas' if wants_pandas else 'polars'
 118    return get_dataframe_library()
 119
 120
 121def wants_polars(
 122    as_polars: Optional[bool] = None,
 123    as_pandas: Optional[bool] = None,
 124) -> bool:
 125    """Return whether one call returns Polars frames (see `resolve_dataframe_library()`)."""
 126    return resolve_dataframe_library(as_polars=as_polars, as_pandas=as_pandas) == 'polars'
 127
 128
 129def import_dataframe_library(library: Optional[str] = None) -> Any:
 130    """
 131    Import the `polars` or `pandas` module (the configured one by default).
 132
 133    A missing library is installed into the `mrsm` virtual environment when automatic installs are
 134    allowed (as for any other optional package); otherwise the `ImportError` names the extra to
 135    install (`pip install meerschaum[polars]` or `pip install meerschaum[pandas]`).
 136    """
 137    library = library or get_dataframe_library()
 138    from meerschaum.utils.packages import attempt_import, import_pandas
 139    module = None
 140    try:
 141        if library == 'pandas':
 142            module = import_pandas(warn=False)
 143        else:
 144            module = attempt_import(library, lazy=False, warn=False)
 145    except Exception:
 146        module = None
 147    if module is None:
 148        raise missing_dataframe_library_error(library)
 149    return module
 150
 151
 152def ensure_dataframe_library(library: Optional[str] = None) -> Any:
 153    """
 154    Return the `polars` or `pandas` module (the configured one by default), importing it on the
 155    first call (`import_dataframe_library()`: installed on first use when automatic installs
 156    are allowed, else an `ImportError` naming the extra). Cheap once the module is imported.
 157    """
 158    import sys
 159    library = library or get_dataframe_library()
 160    module = sys.modules.get(library, None)
 161    if module is not None and not getattr(getattr(module, '__spec__', None), '_initializing', False):
 162        if library != 'pandas' or 'pyarrow' in sys.modules:
 163            return module
 164    return import_dataframe_library(library)
 165
 166
 167def missing_dataframe_library_error(library: str) -> ImportError:
 168    """The `ImportError` for a DataFrame library that is not installed (names the extra)."""
 169    extra = DATAFRAME_EXTRAS.get(library, library)
 170    other = 'pandas' if library == 'polars' else 'polars'
 171    return ImportError(
 172        f"Meerschaum returns {library} DataFrames here, but {library} is not installed. "
 173        f"Install it with `pip install meerschaum[{extra}]`, "
 174        f"or set `dataframe: {other}` in meerschaum.yaml (`MRSM_DATAFRAME={other}`)."
 175    )
 176
 177
 178def is_polars_frame(df: Any) -> bool:
 179    """Whether `df` is a Polars DataFrame or LazyFrame."""
 180    return (type(df).__module__ or '').split('.')[0] == 'polars'
 181
 182
 183def is_pandas_frame(df: Any) -> bool:
 184    """Whether `df` is a pandas (or GeoPandas) DataFrame."""
 185    return (
 186        (type(df).__module__ or '').split('.')[0] in ('pandas', 'geopandas')
 187        and hasattr(df, 'iloc')
 188    )
 189
 190
 191def frame_library_of(df: Any) -> Optional[str]:
 192    """`'polars'`, `'pandas'` or `None`: the library of a DataFrame (or `None` for other data)."""
 193    if is_polars_frame(df):
 194        return 'polars'
 195    if is_pandas_frame(df):
 196        return 'pandas'
 197    return None
 198
 199
 200def output_library_for(
 201    df: Any,
 202    as_polars: Optional[bool] = None,
 203    as_pandas: Optional[bool] = None,
 204) -> str:
 205    """
 206    The library a function that transforms a frame returns: what the call asks for
 207    (`as_polars` / `as_pandas`), else the library `df` came in (a Polars frame stays Polars, a
 208    pandas frame pandas), else (documents, dictionaries, Arrow data) `get_dataframe_library()`.
 209    """
 210    if as_polars is not None or as_pandas is not None:
 211        return resolve_dataframe_library(as_polars=as_polars, as_pandas=as_pandas)
 212    return frame_library_of(df) or get_dataframe_library()
 213
 214
 215def frame_from_arrow(
 216    data: Any,
 217    *,
 218    as_polars: Optional[bool] = None,
 219    as_pandas: Optional[bool] = None,
 220    dtypes: Optional[Dict[str, str]] = None,
 221    debug: bool = False,
 222) -> Any:
 223    """
 224    Build the configured library's DataFrame from Arrow data (the one conversion every
 225    frame-returning API uses).
 226
 227    Parameters
 228    ----------
 229    data: Any
 230        Anything that implements the Arrow PyCapsule interface (`__arrow_c_stream__` or
 231        `__arrow_c_array__`): a `meerschaum._core.ArrowStream`, a `pyarrow.Table`,
 232        `RecordBatch` or `RecordBatchReader`, ... or a list (or tuple) of record batches with one
 233        schema. A pandas or Polars DataFrame is converted to the requested library.
 234        `None` returns `None`.
 235
 236    as_polars: Optional[bool], default None
 237        `True` returns Polars, `False` pandas; `None` follows `get_dataframe_library()`.
 238
 239    as_pandas: Optional[bool], default None
 240        `True` returns pandas, `False` Polars; `None` follows the setting.
 241
 242    dtypes: Optional[Dict[str, str]], default None
 243        Per-column dtypes: Meerschaum dtype strings (`'int'`, `'json'`, `'geometry[Point, 4326]'`,
 244        ...) or pandas dtype strings (`'Int64'`, `'int64[pyarrow]'`, ...).
 245        For pandas they restore the exact pandas dtype within the column's kind, as every read of
 246        the façade does (`int64` vs `Int64` vs `int64[pyarrow]`, the datetime unit); a dtype of
 247        another kind does not cast the column. For Polars, declared `json` and `geometry`
 248        columns become extension columns even when every value is null.
 249
 250    debug: bool, default False
 251        Verbosity toggle.
 252
 253    Returns
 254    -------
 255    A `polars.DataFrame` or a `pandas.DataFrame` (a `geopandas.GeoDataFrame` when there are
 256    geometry columns): JSON, UUID, numeric and geometry columns are restored from the Arrow
 257    field metadata the kernel writes (`ARROW:extension:name`, `meerschaum:dtype`). Polars frames
 258    carry geometry as GeoArrow WKB (binary) with the CRS in the field metadata.
 259
 260    Examples
 261    --------
 262    >>> import pyarrow as pa
 263    >>> frame_from_arrow(pa.table({'a': [1, None]}), as_pandas=True).dtypes
 264    a    Int64
 265    dtype: object
 266    >>> frame_from_arrow(pa.table({'a': [1, None]}), as_polars=True).schema
 267    Schema([('a', Int64)])
 268    """
 269    if data is None:
 270        return None
 271    library = resolve_dataframe_library(as_polars=as_polars, as_pandas=as_pandas)
 272    to_polars_frame = library == 'polars'
 273    current = frame_library_of(data)
 274    if current is not None:
 275        if current == library:
 276            return data
 277        return to_polars(data) if to_polars_frame else to_pandas(data)
 278    ensure_dataframe_library(library)
 279    if isinstance(data, (list, tuple)):
 280        data = list(data)
 281    _rust = _core.frame_from_arrow
 282    return _rust(
 283        data,
 284        as_polars=to_polars_frame,
 285        dtypes=({str(col): str(typ) for col, typ in dtypes.items()} if dtypes else None),
 286        debug=debug,
 287    )
 288
 289
 290def to_pandas(df: Any) -> Any:
 291    """Convert a Polars DataFrame or LazyFrame to Pandas; otherwise return ``df``."""
 292    if df.__class__.__module__.split('.')[0] != 'polars':
 293        return df
 294    if df.__class__.__name__ == 'LazyFrame':
 295        df = df.collect()
 296    json_cols = [
 297        col
 298        for col, typ in df.schema.items()
 299        if getattr(typ, 'ext_name', lambda: None)() == 'arrow.json'
 300    ]
 301    if json_cols:
 302        df = df.with_columns(df[col].ext.storage() for col in json_cols)
 303    pandas_df = df.to_pandas(use_pyarrow_extension_array=True)
 304    if json_cols:
 305        import json
 306        for col in json_cols:
 307            pandas_df[col] = pandas_df[col].apply(
 308                lambda value: json.loads(value) if isinstance(value, str) else None
 309            )
 310    return pandas_df
 311
 312
 313def to_polars(
 314    df: Any,
 315    geometry_cols_types_srids: Optional[Dict[str, Tuple[str, Any]]] = None,
 316    json_cols: Optional[List[str]] = None,
 317) -> Any:
 318    """Convert a Pandas DataFrame to Polars; otherwise return ``df``."""
 319    if df.__class__.__module__.split('.')[0] == 'polars':
 320        return df
 321    polars = mrsm.attempt_import('polars')
 322    geometry_cols_types_srids = (
 323        get_geometry_cols(df, with_types_srids=True)
 324        if geometry_cols_types_srids is None and get_geometry_cols(df)
 325        else (geometry_cols_types_srids or {})
 326    )
 327    geometry_cols_types_srids = {
 328        col: type_srid
 329        for col, type_srid in geometry_cols_types_srids.items()
 330        if col in df.columns
 331    }
 332    json_cols = [col for col in (json_cols or []) if col in df.columns]
 333    if geometry_cols_types_srids or json_cols:
 334        try:
 335            import json
 336            geometry_cols = list(geometry_cols_types_srids)
 337            special_cols = list(dict.fromkeys(geometry_cols + json_cols))
 338            polars_df = to_polars(
 339                df.drop(columns=special_cols),
 340                geometry_cols_types_srids={},
 341                json_cols=[],
 342            )
 343            if geometry_cols:
 344                from meerschaum.utils.dtypes import attempt_cast_to_geometry
 345                shapely = mrsm.attempt_import('shapely', lazy=False)
 346                ### Import inside the `mrsm` venv context: callers (e.g. `mrsm.entry()`)
 347                ### may have deactivated it, and a bare import would then silently
 348                ### skip registration so Arrow exports plain binary storage.
 349                try:
 350                    geoarrow_pyarrow = mrsm.attempt_import(
 351                        'geoarrow.types.type_pyarrow',
 352                        lazy=False,
 353                        warn=False,
 354                        install=False,
 355                    )
 356                    geoarrow_pyarrow.register_extension_types()
 357                except Exception:
 358                    pass
 359            for col, (_, srid) in geometry_cols_types_srids.items():
 360                crs = str(srid) if srid else None
 361                if crs and ':' not in crs:
 362                    crs = 'EPSG:' + crs
 363                metadata = json.dumps(
 364                    ({'crs': crs, 'crs_type': 'authority_code'} if crs else {}),
 365                    separators=(',', ':'),
 366                )
 367                polars_df = polars_df.with_columns(polars.Series(
 368                    col,
 369                    shapely.to_wkb(
 370                        attempt_cast_to_geometry(df[col]),
 371                        hex=False,
 372                        include_srid=True,
 373                    ).tolist(),
 374                    dtype=polars.Extension('geoarrow.wkb', polars.Binary, metadata),
 375                ))
 376            if json_cols:
 377                from meerschaum.utils.dtypes import json_serialize_value, value_is_null
 378                for col in json_cols:
 379                    values = []
 380                    for value in df[col].tolist():
 381                        if value_is_null(value):
 382                            values.append(None)
 383                            continue
 384                        if isinstance(value, str):
 385                            try:
 386                                value = json.loads(value)
 387                            except json.JSONDecodeError:
 388                                pass
 389                        values.append(json.dumps(
 390                            value,
 391                            default=json_serialize_value,
 392                            separators=(',', ':'),
 393                            allow_nan=False,
 394                        ))
 395                    polars_df = polars_df.with_columns(polars.Series(
 396                        col,
 397                        values,
 398                        dtype=polars.Extension('arrow.json', polars.String, ''),
 399                    ))
 400            return polars_df.select(list(df.columns))
 401        except Exception:
 402            # ponytail: Preserve compatibility if Polars changes its unstable extension API.
 403            pass
 404    if get_uuid_cols(df):
 405        return polars.DataFrame(df.to_dict(orient='list'), strict=False)
 406    try:
 407        return polars.from_pandas(df, include_index=False)
 408    except (TypeError, ValueError):
 409        # ponytail: Preserve unsupported Python objects; remove when Polars supports UUIDs.
 410        return polars.DataFrame(df.to_dict(orient='list'), strict=False)
 411
 412
 413def add_missing_cols_to_df(
 414    df: 'pd.DataFrame',
 415    dtypes: Dict[str, Any],
 416) -> 'pd.DataFrame':
 417    """
 418    Add columns from the dtypes dictionary as null columns to a new DataFrame.
 419
 420    Parameters
 421    ----------
 422    df: pd.DataFrame
 423        The dataframe we should copy and add null columns.
 424
 425    dtypes:
 426        The data types dictionary which may contain keys not present in `df.columns`.
 427
 428    Returns
 429    -------
 430    A new `DataFrame` with the keys from `dtypes` added as null columns.
 431    If `df.dtypes` is the same as `dtypes`, then return a reference to `df`.
 432    NOTE: This will not ensure that dtypes are enforced!
 433
 434    Examples
 435    --------
 436    >>> import pandas as pd
 437    >>> df = pd.DataFrame([{'a': 1}])
 438    >>> dtypes = {'b': 'Int64'}
 439    >>> add_missing_cols_to_df(df, dtypes)
 440          a  b
 441       0  1  <NA>
 442    >>> add_missing_cols_to_df(df, dtypes).dtypes
 443    a    int64
 444    b    Int64
 445    dtype: object
 446    >>> add_missing_cols_to_df(df, {'a': 'object'}).dtypes
 447    a    int64
 448    dtype: object
 449    >>> 
 450    """
 451    if set(df.columns) == set(dtypes):
 452        return df
 453
 454    from meerschaum.utils.packages import attempt_import
 455    from meerschaum.utils.dtypes import to_pandas_dtype
 456    pandas = attempt_import('pandas')
 457
 458    def build_series(dtype: str):
 459        return pandas.Series([], dtype=to_pandas_dtype(dtype))
 460
 461    assign_kwargs = {
 462        str(col): build_series(str(typ))
 463        for col, typ in dtypes.items()
 464        if col not in df.columns
 465    }
 466    df_with_cols = df.assign(**assign_kwargs)
 467    for col in assign_kwargs:
 468        df_with_cols[col] = df_with_cols[col].fillna(pandas.NA)
 469    return df_with_cols
 470
 471
 472def filter_unseen_df(
 473    old_df: 'pd.DataFrame',
 474    new_df: 'pd.DataFrame',
 475    safe_copy: bool = True,
 476    dtypes: Optional[Dict[str, Any]] = None,
 477    include_unchanged_columns: bool = False,
 478    coerce_mixed_numerics: bool = True,
 479    debug: bool = False,
 480) -> 'pd.DataFrame':
 481    """
 482    Left join two DataFrames to find the newest unseen data.
 483
 484    Parameters
 485    ----------
 486    old_df: 'pd.DataFrame'
 487        The original (target) dataframe. Acts as a filter on the `new_df`.
 488
 489    new_df: 'pd.DataFrame'
 490        The fetched (source) dataframe. Rows that are contained in `old_df` are removed.
 491
 492    safe_copy: bool, default True
 493        If `True`, create a copy before comparing and modifying the dataframes.
 494        Setting to `False` may mutate the DataFrames.
 495
 496    dtypes: Optional[Dict[str, Any]], default None
 497        Optionally specify the datatypes of the dataframe.
 498
 499    include_unchanged_columns: bool, default False
 500        If `True`, include columns which haven't changed on rows which have changed.
 501
 502    coerce_mixed_numerics: bool, default True
 503        If `True`, cast mixed integer and float columns between the old and new dataframes into
 504        numeric values (`decimal.Decimal`).
 505
 506    debug: bool, default False
 507        Verbosity toggle.
 508
 509    Returns
 510    -------
 511    A pandas dataframe of the new, unseen rows in `new_df`.
 512
 513    Examples
 514    --------
 515    ```python
 516    >>> import pandas as pd
 517    >>> df1 = pd.DataFrame({'a': [1,2]})
 518    >>> df2 = pd.DataFrame({'a': [2,3]})
 519    >>> filter_unseen_df(df1, df2)
 520       a
 521    0  3
 522
 523    ```
 524
 525    """
 526    _rust = _core.filter_unseen_df
 527    return _rust(
 528        old_df, new_df=new_df, safe_copy=safe_copy, dtypes=dtypes,
 529        include_unchanged_columns=include_unchanged_columns,
 530        coerce_mixed_numerics=coerce_mixed_numerics, debug=debug,
 531    )
 532
 533
 534def parse_df_datetimes(
 535    df: 'pd.DataFrame',
 536    ignore_cols: Optional[Iterable[str]] = None,
 537    strip_timezone: bool = False,
 538    chunksize: Optional[int] = None,
 539    dtype_backend: str = 'numpy_nullable',
 540    ignore_all: bool = False,
 541    precision_unit: Optional[str] = None,
 542    coerce_utc: bool = True,
 543    debug: bool = False,
 544) -> 'pd.DataFrame':
 545    """
 546    Parse a pandas DataFrame for datetime columns and cast as datetimes.
 547
 548    Parameters
 549    ----------
 550    df: pd.DataFrame
 551        The pandas DataFrame to parse.
 552
 553    ignore_cols: Optional[Iterable[str]], default None
 554        If provided, do not attempt to coerce these columns as datetimes.
 555
 556    strip_timezone: bool, default False
 557        If `True`, remove the UTC `tzinfo` property.
 558
 559    chunksize: Optional[int], default None
 560        If the pandas implementation is `'dask'`, use this chunksize for the distributed dataframe.
 561
 562    dtype_backend: str, default 'numpy_nullable'
 563        If `df` is not a DataFrame and new one needs to be constructed,
 564        use this as the datatypes backend.
 565        Accepted values are 'numpy_nullable' and 'pyarrow'.
 566
 567    ignore_all: bool, default False
 568        If `True`, do not attempt to cast any columns to datetimes.
 569
 570    precision_unit: Optional[str], default None
 571        If provided, enforce the given precision on the coerced datetime columns.
 572
 573    coerce_utc: bool, default True
 574        Coerce the datetime columns to UTC (see `meerschaum.utils.dtypes.to_datetime()`).
 575
 576    debug: bool, default False
 577        Verbosity toggle.
 578
 579    Returns
 580    -------
 581    A new pandas DataFrame with the determined datetime columns
 582    (usually ISO strings) cast as datetimes.
 583
 584    Examples
 585    --------
 586    ```python
 587    >>> import pandas as pd
 588    >>> df = pd.DataFrame({'a': ['2022-01-01 00:00:00']}) 
 589    >>> df.dtypes
 590    a    object
 591    dtype: object
 592    >>> df2 = parse_df_datetimes(df)
 593    >>> df2.dtypes
 594    a    datetime64[us, UTC]
 595    dtype: object
 596
 597    ```
 598
 599    """
 600    from meerschaum.utils.packages import import_pandas, attempt_import
 601    from meerschaum.utils.debug import dprint
 602    from meerschaum.utils.warnings import warn
 603    from meerschaum.utils.misc import items_str
 604    from meerschaum.utils.dtypes import to_datetime, MRSM_PD_DTYPES
 605    import traceback
 606
 607    pd = import_pandas()
 608    pandas = attempt_import('pandas')
 609    pd_name = pd.__name__
 610    using_dask = 'dask' in pd_name
 611    df_is_dask = (hasattr(df, '__module__') and 'dask' in df.__module__)
 612    dask_dataframe = None
 613    if using_dask or df_is_dask:
 614        npartitions = chunksize_to_npartitions(chunksize)
 615        dask_dataframe = attempt_import('dask.dataframe')
 616
 617    ### if df is a dict, build DataFrame
 618    if isinstance(df, pandas.DataFrame):
 619        pdf = df
 620    elif df_is_dask and isinstance(df, dask_dataframe.DataFrame):
 621        pdf = get_first_valid_dask_partition(df)
 622    else:
 623        if debug:
 624            dprint(f"df is of type '{type(df)}'. Building {pd.DataFrame}...")
 625
 626        if using_dask:
 627            if isinstance(df, list):
 628                keys = set()
 629                for doc in df:
 630                    for key in doc:
 631                        keys.add(key)
 632                df = pd.DataFrame.from_dict(
 633                    {
 634                        k: [
 635                            doc.get(k, None)
 636                            for doc in df
 637                        ] for k in keys
 638                    },
 639                    npartitions=npartitions,
 640                )
 641            elif isinstance(df, dict):
 642                df = pd.DataFrame.from_dict(df, npartitions=npartitions)
 643            elif 'pandas.core.frame.DataFrame' in str(type(df)):
 644                df = pd.from_pandas(df, npartitions=npartitions)
 645            else:
 646                raise Exception("Can only parse dictionaries or lists of dictionaries with Dask.")
 647            pandas = attempt_import('pandas')
 648            pdf = get_first_valid_dask_partition(df)
 649
 650        else:
 651            df = pd.DataFrame(df).convert_dtypes(dtype_backend=dtype_backend)
 652            pdf = df
 653
 654    ### skip parsing if DataFrame is empty
 655    if len(pdf) == 0:
 656        if debug:
 657            dprint("df is empty. Returning original DataFrame without casting datetime columns...")
 658        return df
 659
 660    ignore_cols = set(
 661        (ignore_cols or []) + [
 662            col
 663            for col, dtype in pdf.dtypes.items() 
 664            if 'datetime' in str(dtype)
 665        ]
 666    )
 667    cols_to_inspect = [
 668        col
 669        for col in pdf.columns
 670        if col not in ignore_cols
 671    ] if not ignore_all else []
 672
 673    if len(cols_to_inspect) == 0:
 674        if debug:
 675            dprint("All columns are ignored, skipping datetime detection...")
 676        return df.infer_objects().fillna(pandas.NA)
 677
 678    ### apply regex to columns to determine which are ISO datetimes
 679    iso_dt_regex = r'\d{4}-\d{2}-\d{2}.\d{2}\:\d{2}\:\d+'
 680    dt_mask = pdf[cols_to_inspect].astype(str).apply(
 681        lambda s: s.str.match(iso_dt_regex).all()
 682    )
 683
 684    ### list of datetime column names
 685    datetime_cols = [col for col in pdf[cols_to_inspect].loc[:, dt_mask]]
 686    if not datetime_cols:
 687        if debug:
 688            dprint("No columns detected as datetimes, returning...")
 689        return df.infer_objects().fillna(pandas.NA)
 690
 691    if debug:
 692        dprint("Converting columns to datetimes: " + str(datetime_cols))
 693
 694    def _parse_to_datetime(x):
 695        return to_datetime(x, precision_unit=precision_unit, coerce_utc=coerce_utc)
 696
 697    try:
 698        if not using_dask:
 699            df[datetime_cols] = df[datetime_cols].apply(_parse_to_datetime)
 700        else:
 701            df[datetime_cols] = df[datetime_cols].apply(
 702                _parse_to_datetime,
 703                utc=True,
 704                axis=1,
 705                meta={
 706                    col: MRSM_PD_DTYPES['datetime']
 707                    for col in datetime_cols
 708                }
 709            )
 710    except Exception:
 711        warn(
 712            f"Unable to apply `to_datetime()` to {items_str(datetime_cols)}:\n"
 713            + f"{traceback.format_exc()}"
 714        )
 715
 716    if strip_timezone:
 717        for dt in datetime_cols:
 718            try:
 719                df[dt] = df[dt].dt.tz_localize(None)
 720            except Exception:
 721                warn(
 722                    f"Unable to convert column '{dt}' to naive datetime:\n"
 723                    + f"{traceback.format_exc()}"
 724                )
 725
 726    return df.fillna(pandas.NA)
 727
 728
 729def get_unhashable_cols(df: 'pd.DataFrame') -> List[str]:
 730    """
 731    Get the columns which contain unhashable objects from a Pandas DataFrame.
 732
 733    Parameters
 734    ----------
 735    df: pd.DataFrame
 736        The DataFrame which may contain unhashable objects.
 737
 738    Returns
 739    -------
 740    A list of columns.
 741    """
 742    if df is None:
 743        return []
 744    if len(df) == 0:
 745        return []
 746
 747    is_dask = 'dask' in df.__module__
 748    if is_dask:
 749        from meerschaum.utils.packages import attempt_import
 750        pandas = attempt_import('pandas')
 751        df = pandas.DataFrame(get_first_valid_dask_partition(df))
 752    return [
 753        col for col, val in df.iloc[0].items()
 754        if not isinstance(val, Hashable)
 755    ]
 756
 757
 758def get_json_cols(df: 'pd.DataFrame') -> List[str]:
 759    """
 760    Get the columns which contain unhashable objects from a Pandas DataFrame.
 761
 762    Parameters
 763    ----------
 764    df: pd.DataFrame
 765        The DataFrame which may contain unhashable objects.
 766
 767    Returns
 768    -------
 769    A list of columns to be encoded as JSON.
 770    """
 771    _rust = _core.get_json_cols
 772    return _rust(df)
 773
 774
 775def get_numeric_cols(df: 'pd.DataFrame') -> List[str]:
 776    """
 777    Get the columns which contain `decimal.Decimal` objects from a Pandas DataFrame.
 778
 779    Parameters
 780    ----------
 781    df: pd.DataFrame
 782        The DataFrame which may contain decimal objects.
 783
 784    Returns
 785    -------
 786    A list of columns to treat as numerics.
 787    """
 788    _rust = _core.get_numeric_cols
 789    return _rust(df)
 790
 791
 792def get_bool_cols(df: 'pd.DataFrame') -> List[str]:
 793    """
 794    Get the columns which contain `bool` objects from a Pandas DataFrame.
 795
 796    Parameters
 797    ----------
 798    df: pd.DataFrame
 799        The DataFrame which may contain bools.
 800
 801    Returns
 802    -------
 803    A list of columns to treat as bools.
 804    """
 805    if df is None:
 806        return []
 807
 808    is_dask = 'dask' in df.__module__
 809    if is_dask:
 810        df = get_first_valid_dask_partition(df)
 811
 812    if len(df) == 0:
 813        return []
 814
 815    from meerschaum.utils.dtypes import are_dtypes_equal
 816
 817    return [
 818        col
 819        for col, typ in df.dtypes.items()
 820        if are_dtypes_equal(str(typ), 'bool')
 821    ]
 822
 823
 824def get_uuid_cols(df: 'pd.DataFrame') -> List[str]:
 825    """
 826    Get the columns which contain `uuid.UUID` objects from a Pandas DataFrame.
 827
 828    Parameters
 829    ----------
 830    df: pd.DataFrame
 831        The DataFrame which may contain UUID objects.
 832
 833    Returns
 834    -------
 835    A list of columns to treat as UUIDs.
 836    """
 837    _rust = _core.get_uuid_cols
 838    return _rust(df)
 839
 840
 841def get_datetime_cols(
 842    df: 'pd.DataFrame',
 843    timezone_aware: bool = True,
 844    timezone_naive: bool = True,
 845    with_tz_precision: bool = False,
 846) -> Union[List[str], Dict[str, Tuple[Union[str, None], str]]]:
 847    """
 848    Get the columns which contain `datetime` or `Timestamp` objects from a Pandas DataFrame.
 849
 850    Parameters
 851    ----------
 852    df: pd.DataFrame
 853        The DataFrame which may contain `datetime` or `Timestamp` objects.
 854
 855    timezone_aware: bool, default True
 856        If `True`, include timezone-aware datetime columns.
 857
 858    timezone_naive: bool, default True
 859        If `True`, include timezone-naive datetime columns.
 860
 861    with_tz_precision: bool, default False
 862        If `True`, return a dictionary mapping column names to tuples in the form
 863        `(timezone, precision)`.
 864
 865    Returns
 866    -------
 867    A list of columns to treat as datetimes, or a dictionary of columns to tz+precision tuples
 868    (if `with_tz_precision` is `True`).
 869    """
 870    _rust = _core.get_datetime_cols
 871    return _rust(
 872        df, timezone_aware=timezone_aware, timezone_naive=timezone_naive,
 873        with_tz_precision=with_tz_precision,
 874    )
 875
 876
 877def get_datetime_cols_types(df: 'pd.DataFrame') -> Dict[str, str]:
 878    """
 879    Return a dictionary mapping datetime columns to specific types strings.
 880
 881    Parameters
 882    ----------
 883    df: pd.DataFrame
 884        The DataFrame which may contain datetime columns.
 885
 886    Returns
 887    -------
 888    A dictionary mapping the datetime columns' names to dtype strings
 889    (containing timezone and precision metadata).
 890
 891    Examples
 892    --------
 893    >>> from datetime import datetime, timezone
 894    >>> import pandas as pd
 895    >>> df = pd.DataFrame({'dt_tz_aware': [datetime(2025, 1, 1, tzinfo=timezone.utc)]})
 896    >>> get_datetime_cols_types(df)
 897    {'dt_tz_aware': 'datetime64[us, UTC]'}
 898    >>> df = pd.DataFrame({'distant_dt': [datetime(1, 1, 1)]})
 899    >>> get_datetime_cols_types(df)
 900    {'distant_dt': 'datetime64[us]'}
 901    >>> df = pd.DataFrame({'dt_second': datetime(2025, 1, 1)})
 902    >>> df['dt_second'] = df['dt_second'].astype('datetime64[s]')
 903    >>> get_datetime_cols_types(df)
 904    {'dt_second': 'datetime64[s]'}
 905    """
 906    from meerschaum.utils.dtypes import MRSM_PRECISION_UNITS_ABBREVIATIONS
 907    dt_cols_tuples = get_datetime_cols(df, with_tz_precision=True)
 908    if not dt_cols_tuples:
 909        return {}
 910
 911    return {
 912        col: (
 913            f"datetime64[{MRSM_PRECISION_UNITS_ABBREVIATIONS[precision]}]"
 914            if tz is None
 915            else f"datetime64[{MRSM_PRECISION_UNITS_ABBREVIATIONS[precision]}, {tz}]"
 916        )
 917        for col, (tz, precision) in dt_cols_tuples.items()
 918    }
 919
 920
 921def get_date_cols(df: 'pd.DataFrame') -> List[str]:
 922    """
 923    Get the `date` columns from a Pandas DataFrame.
 924
 925    Parameters
 926    ----------
 927    df: pd.DataFrame
 928        The DataFrame which may contain dates.
 929
 930    Returns
 931    -------
 932    A list of columns to treat as dates.
 933    """
 934    from meerschaum.utils.dtypes import are_dtypes_equal
 935    if df is None:
 936        return []
 937
 938    is_dask = 'dask' in df.__module__
 939    if is_dask:
 940        df = get_first_valid_dask_partition(df)
 941
 942    known_date_cols = [
 943        col
 944        for col, typ in df.dtypes.items()
 945        if are_dtypes_equal(typ, 'date')
 946    ]
 947
 948    if len(df) == 0:
 949        return known_date_cols
 950
 951    cols_indices = {
 952        col: df[col].first_valid_index()
 953        for col in df.columns
 954        if col not in known_date_cols
 955    }
 956    object_date_cols = [
 957        col
 958        for col, ix in cols_indices.items()
 959        if (
 960            ix is not None
 961            and isinstance(df.loc[ix][col], date)
 962        )
 963    ]
 964
 965    all_date_cols = set(known_date_cols + object_date_cols)
 966
 967    return [
 968        col
 969        for col in df.columns
 970        if col in all_date_cols
 971    ]
 972
 973
 974def get_bytes_cols(df: 'pd.DataFrame') -> List[str]:
 975    """
 976    Get the columns which contain bytes strings from a Pandas DataFrame.
 977
 978    Parameters
 979    ----------
 980    df: pd.DataFrame
 981        The DataFrame which may contain bytes strings.
 982
 983    Returns
 984    -------
 985    A list of columns to treat as bytes.
 986    """
 987    _rust = _core.get_bytes_cols
 988    return _rust(df)
 989
 990
 991def get_geometry_cols(
 992    df: 'pd.DataFrame',
 993    with_types_srids: bool = False,
 994) -> Union[List[str], Dict[str, Any]]:
 995    """
 996    Get the columns which contain shapely objects from a Pandas DataFrame.
 997
 998    Parameters
 999    ----------
1000    df: pd.DataFrame
1001        The DataFrame which may contain bytes strings.
1002
1003    with_types_srids: bool, default False
1004        If `True`, return a dictionary mapping columns to geometry types and SRIDs.
1005
1006    Returns
1007    -------
1008    A list of columns to treat as `geometry`.
1009    If `with_types_srids`, return a dictionary mapping columns to tuples in the form (type, SRID).
1010    """
1011    _rust = _core.get_geometry_cols
1012    return _rust(df, with_types_srids=with_types_srids)
1013
1014
1015def get_geometry_cols_types(df: 'pd.DataFrame') -> Dict[str, str]:
1016    """
1017    Return a dtypes dictionary mapping columns to specific geometry types (type, srid).
1018    """
1019    geometry_cols_types_srids = get_geometry_cols(df, with_types_srids=True)
1020    new_cols_types = {}
1021    for col, (geometry_type, srid) in geometry_cols_types_srids.items():
1022        new_dtype = "geometry"
1023        modifier = ""
1024        if not srid and geometry_type.lower() == 'geometry':
1025            new_cols_types[col] = new_dtype
1026            continue
1027
1028        modifier = "["
1029        if geometry_type.lower() != 'geometry':
1030            modifier += f"{geometry_type}"
1031
1032        if srid:
1033            if modifier != '[':
1034                modifier += ", "
1035            modifier += f"{srid}"
1036        modifier += "]"
1037        new_cols_types[col] = f"{new_dtype}{modifier}"
1038    return new_cols_types
1039
1040
1041def get_special_cols(df: 'pd.DataFrame') -> Dict[str, str]:
1042    """
1043    Return a dtypes dictionary mapping special columns to their dtypes.
1044    """
1045    return {
1046        **{col: 'json' for col in get_json_cols(df)},
1047        **{col: 'uuid' for col in get_uuid_cols(df)},
1048        **{col: 'bytes' for col in get_bytes_cols(df)},
1049        **{col: 'bool' for col in get_bool_cols(df)},
1050        **{col: 'numeric' for col in get_numeric_cols(df)},
1051        **{col: 'date' for col in get_date_cols(df)},
1052        **get_datetime_cols_types(df),
1053        **get_geometry_cols_types(df),
1054    }
1055
1056
1057def enforce_dtypes(
1058    df: 'pd.DataFrame',
1059    dtypes: Dict[str, str],
1060    explicit_dtypes: Optional[Dict[str, str]] = None,
1061    safe_copy: bool = True,
1062    coerce_numeric: bool = False,
1063    coerce_timezone: bool = True,
1064    strip_timezone: bool = False,
1065    as_polars: Optional[bool] = None,
1066    debug: bool = False,
1067    as_pandas: Optional[bool] = None,
1068) -> 'pd.DataFrame':
1069    """
1070    Enforce the `dtypes` dictionary on a DataFrame.
1071
1072    Parameters
1073    ----------
1074    df: pd.DataFrame
1075        The DataFrame on which to enforce dtypes.
1076
1077    dtypes: Dict[str, str]
1078        The data types to attempt to enforce on the DataFrame.
1079
1080    explicit_dtypes: Optional[Dict[str, str]], default None
1081        If provided, automatic dtype coersion will respect explicitly configured
1082        dtypes (`int`, `float`, `numeric`).
1083
1084    safe_copy: bool, default True
1085        If `True`, create a copy before comparing and modifying the dataframes.
1086        Setting to `False` may mutate the DataFrames.
1087        See `meerschaum.utils.dataframe.filter_unseen_df`.
1088
1089    coerce_numeric: bool, default False
1090        If `True`, convert float and int collisions to numeric.
1091
1092    coerce_timezone: bool, default True
1093        If `True`, convert datetimes to UTC.
1094
1095    strip_timezone: bool, default False
1096        If `coerce_timezone` and `strip_timezone` are `True`,
1097        remove timezone information from datetimes.
1098
1099    as_polars: Optional[bool], default None
1100        `True` returns a Polars DataFrame, `False` a pandas one. By default the result is the
1101        library `df` came in (a Polars frame stays Polars, a pandas frame pandas).
1102
1103    debug: bool, default False
1104        Verbosity toggle.
1105
1106    as_pandas: Optional[bool], default None
1107        `True` returns a pandas DataFrame, `False` a Polars one (see `as_polars`).
1108
1109    Returns
1110    -------
1111    The DataFrame with the types enforced, of the library `df` came in (or the one asked for).
1112    """
1113    if df is None:
1114        return df
1115    library = output_library_for(df, as_polars=as_polars, as_pandas=as_pandas)
1116    _rust = _core.enforce_dtypes
1117    return _rust(
1118        df, dtypes=dtypes, explicit_dtypes=explicit_dtypes, safe_copy=safe_copy,
1119        coerce_numeric=coerce_numeric, coerce_timezone=coerce_timezone,
1120        strip_timezone=strip_timezone, as_polars=(library == 'polars'), debug=debug,
1121    )
1122
1123
1124def get_datetime_bound_from_df(
1125    df: Union['pd.DataFrame', Dict[str, List[Any]], List[Dict[str, Any]]],
1126    datetime_column: str,
1127    minimum: bool = True,
1128) -> Union[int, datetime, None]:
1129    """
1130    Return the minimum or maximum datetime (or integer) from a DataFrame.
1131
1132    Parameters
1133    ----------
1134    df: Union['pd.DataFrame', Dict[str, List[Any]], List[Dict[str, Any]]]
1135        The DataFrame, list, or dict which contains the range axis.
1136
1137    datetime_column: str
1138        The name of the datetime (or int) column.
1139
1140    minimum: bool
1141        Whether to return the minimum (default) or maximum value.
1142
1143    Returns
1144    -------
1145    The minimum or maximum datetime value in the dataframe, or `None`.
1146    """
1147    from meerschaum.utils.dtypes import to_datetime, value_is_null
1148
1149    if df is None:
1150        return None
1151    if not datetime_column:
1152        return None
1153
1154    def compare(a, b):
1155        if a is None:
1156            return b
1157        if b is None:
1158            return a
1159        if minimum:
1160            return a if a < b else b
1161        return a if a > b else b
1162
1163    if isinstance(df, list):
1164        if len(df) == 0:
1165            return None
1166        best_yet = df[0].get(datetime_column, None)
1167        for doc in df:
1168            val = doc.get(datetime_column, None)
1169            best_yet = compare(best_yet, val)
1170        return best_yet
1171
1172    if isinstance(df, dict):
1173        if datetime_column not in df:
1174            return None
1175        best_yet = df[datetime_column][0]
1176        for val in df[datetime_column]:
1177            best_yet = compare(best_yet, val)
1178        return best_yet
1179
1180    if 'DataFrame' in str(type(df)):
1181        from meerschaum.utils.dtypes import are_dtypes_equal
1182        pandas = mrsm.attempt_import('pandas')
1183        is_dask = 'dask' in df.__module__
1184
1185        if datetime_column not in df.columns:
1186            return None
1187
1188        try:
1189            dt_val = (
1190                df[datetime_column].min(skipna=True)
1191                if minimum
1192                else df[datetime_column].max(skipna=True)
1193            )
1194        except Exception:
1195            dt_val = pandas.NA
1196        if is_dask and dt_val is not None and dt_val is not pandas.NA:
1197            dt_val = dt_val.compute()
1198
1199        return (
1200            to_datetime(dt_val, as_pydatetime=True)
1201            if are_dtypes_equal(str(type(dt_val)), 'datetime')
1202            else (dt_val if not value_is_null(dt_val) else None)
1203        )
1204
1205    return None
1206
1207
1208def get_unique_index_values(
1209    df: Union['pd.DataFrame', Dict[str, List[Any]], List[Dict[str, Any]]],
1210    indices: List[str],
1211) -> Dict[str, List[Any]]:
1212    """
1213    Return a dictionary of the unique index values in a DataFrame.
1214
1215    Parameters
1216    ----------
1217    df: Union['pd.DataFrame', Dict[str, List[Any]], List[Dict[str, Any]]]
1218        The dataframe (or list or dict) which contains index values.
1219
1220    indices: List[str]
1221        The list of index columns.
1222
1223    Returns
1224    -------
1225    A dictionary mapping indices to unique values.
1226    """
1227    if df is None:
1228        return {}
1229    if 'dataframe' in str(type(df)).lower():
1230        pandas = mrsm.attempt_import('pandas')
1231        return {
1232            col: list({
1233                (val if val is not pandas.NA else None)
1234                for val in df[col].unique()
1235            })
1236            for col in indices
1237            if col in df.columns
1238        }
1239
1240    unique_indices = defaultdict(lambda: set())
1241    if isinstance(df, list):
1242        for doc in df:
1243            for index in indices:
1244                if index in doc:
1245                    unique_indices[index].add(doc[index])
1246
1247    elif isinstance(df, dict):
1248        for index in indices:
1249            if index in df:
1250                unique_indices[index] = unique_indices[index].union(set(df[index]))
1251
1252    return {key: list(val) for key, val in unique_indices.items()}
1253
1254
1255def df_is_chunk_generator(df: Any) -> bool:
1256    """
1257    Determine whether to treat `df` as a chunk generator.
1258
1259    Note this should only be used in a context where generators are expected,
1260    as it will return `True` for any iterable.
1261
1262    Parameters
1263    ----------
1264    The DataFrame or chunk generator to evaluate.
1265
1266    Returns
1267    -------
1268    A `bool` indicating whether to treat `df` as a generator.
1269    """
1270    return (
1271        not isinstance(df, (dict, list, str))
1272        and 'DataFrame' not in str(type(df))
1273        and isinstance(df, (Generator, Iterable, Iterator))
1274    )
1275
1276
1277def chunksize_to_npartitions(chunksize: Optional[int]) -> int:
1278    """
1279    Return the Dask `npartitions` value for a given `chunksize`.
1280    """
1281    if chunksize == -1:
1282        from meerschaum.config import get_config
1283        chunksize = get_config('system', 'connectors', 'sql', 'chunksize')
1284    if chunksize is None:
1285        return 1
1286    return -1 * chunksize
1287
1288
1289def df_from_literal(
1290    pipe: Optional[mrsm.Pipe] = None,
1291    literal: Optional[str] = None,
1292    debug: bool = False
1293) -> 'pd.DataFrame':
1294    """
1295    Construct a dataframe from a literal value, using the pipe's datetime and value column names.
1296
1297    Parameters
1298    ----------
1299    pipe: Optional['meerschaum.Pipe'], default None
1300        The pipe which will consume the literal value.
1301
1302    Returns
1303    -------
1304    A 1-row pandas DataFrame from with the current UTC timestamp as the datetime columns
1305    and the literal as the value.
1306    """
1307    from meerschaum.utils.packages import import_pandas
1308    from meerschaum.utils.warnings import error, warn
1309    from meerschaum.utils.debug import dprint
1310    from meerschaum.utils.dtypes import get_current_timestamp
1311
1312    if pipe is None or literal is None:
1313        error("Please provide a Pipe and a literal value")
1314
1315    dt_col = pipe.columns.get(
1316        'datetime',
1317        mrsm.get_config('pipes', 'autotime', 'column_name_if_datetime_missing')
1318    )
1319    val_col = pipe.get_val_column(debug=debug)
1320
1321    val = literal
1322    if isinstance(literal, str):
1323        if debug:
1324            dprint(f"Received literal string: '{literal}'")
1325        import ast
1326        try:
1327            val = ast.literal_eval(literal)
1328        except Exception:
1329            warn(
1330                "Failed to parse value from string:\n" + f"{literal}" +
1331                "\n\nWill cast as a string instead."\
1332            )
1333            val = literal
1334
1335    now = get_current_timestamp(pipe.precision)
1336    pd = import_pandas()
1337    return pd.DataFrame({dt_col: [now], val_col: [val]})
1338
1339
1340def get_first_valid_dask_partition(ddf: 'dask.dataframe.DataFrame') -> Union['pd.DataFrame', None]:
1341    """
1342    Return the first valid Dask DataFrame partition (if possible).
1343    """
1344    pdf = None
1345    for partition in ddf.partitions:
1346        try:
1347            pdf = partition.compute()
1348        except Exception:
1349            continue
1350        if len(pdf) > 0:
1351            return pdf
1352    _ = mrsm.attempt_import('partd', lazy=False)
1353    return ddf.compute()
1354
1355
1356def query_df(
1357    df: 'pd.DataFrame',
1358    params: Optional[Dict[str, Any]] = None,
1359    begin: Union[datetime, int, None] = None,
1360    end: Union[datetime, int, None] = None,
1361    datetime_column: Optional[str] = None,
1362    select_columns: Optional[List[str]] = None,
1363    omit_columns: Optional[List[str]] = None,
1364    inplace: bool = False,
1365    reset_index: bool = False,
1366    coerce_types: bool = False,
1367    debug: bool = False,
1368) -> 'pd.DataFrame':
1369    """
1370    Query the dataframe with the params dictionary.
1371
1372    Parameters
1373    ----------
1374    df: pd.DataFrame
1375        The DataFrame to query against.
1376
1377    params: Optional[Dict[str, Any]], default None
1378        The parameters dictionary to use for the query.
1379
1380    begin: Union[datetime, int, None], default None
1381        If `begin` and `datetime_column` are provided, only return rows with a timestamp
1382        greater than or equal to this value.
1383
1384    end: Union[datetime, int, None], default None
1385        If `begin` and `datetime_column` are provided, only return rows with a timestamp
1386        less than this value.
1387
1388    datetime_column: Optional[str], default None
1389        A `datetime_column` must be provided to use `begin` and `end`.
1390
1391    select_columns: Optional[List[str]], default None
1392        If provided, only return these columns.
1393
1394    omit_columns: Optional[List[str]], default None
1395        If provided, do not include these columns in the result.
1396
1397    inplace: bool, default False
1398        If `True`, modify the DataFrame inplace rather than creating a new DataFrame.
1399
1400    reset_index: bool, default False
1401        If `True`, reset the index in the resulting DataFrame.
1402
1403    coerce_types: bool, default False
1404        If `True`, cast the dataframe and parameters as strings before querying.
1405
1406    Returns
1407    -------
1408    A Pandas DataFrame query result.
1409    """
1410    _rust = _core.query_df
1411    return _rust(
1412        df, params=params, begin=begin, end=end, datetime_column=datetime_column,
1413        select_columns=select_columns, omit_columns=omit_columns, inplace=inplace,
1414        reset_index=reset_index, coerce_types=coerce_types, debug=debug,
1415    )
1416
1417
1418def to_json(
1419    df: 'pd.DataFrame',
1420    safe_copy: bool = True,
1421    orient: str = 'records',
1422    date_format: str = 'iso',
1423    date_unit: str = 'us',
1424    double_precision: int = 15,
1425    geometry_format: str = 'geojson',
1426    **kwargs: Any
1427) -> str:
1428    """
1429    Serialize the given dataframe as a JSON string.
1430
1431    Parameters
1432    ----------
1433    df: pd.DataFrame
1434        The DataFrame to be serialized.
1435
1436    safe_copy: bool, default True
1437        If `False`, modify the DataFrame inplace.
1438
1439    date_format: str, default 'iso'
1440        The default format for timestamps.
1441
1442    date_unit: str, default 'us'
1443        The precision of the timestamps.
1444
1445    double_precision: int, default 15
1446        The number of decimal places to use when encoding floating point values (maximum 15).
1447
1448    geometry_format: str, default 'geojson'
1449        The serialization format for geometry data.
1450        Accepted values are `geojson`, `wkb_hex`, and `wkt`.
1451
1452    Returns
1453    -------
1454    A JSON string.
1455    """
1456    import warnings
1457    from meerschaum.utils.packages import import_pandas
1458    from meerschaum.utils.dtypes import (
1459        serialize_bytes,
1460        serialize_decimal,
1461        serialize_geometry,
1462        serialize_date,
1463    )
1464    pd = import_pandas()
1465    uuid_cols = get_uuid_cols(df)
1466    bytes_cols = get_bytes_cols(df)
1467    numeric_cols = get_numeric_cols(df)
1468    date_cols = get_date_cols(df)
1469    geometry_cols = get_geometry_cols(df)
1470    geometry_cols_srids = {
1471        col: int((getattr(df[col].crs, 'srs', '') or '').split(':', maxsplit=1)[-1] or '0')
1472        for col in geometry_cols
1473    } if 'geodataframe' in str(type(df)).lower() else {}
1474    if safe_copy and bool(uuid_cols or bytes_cols or geometry_cols or numeric_cols):
1475        df = df.copy()
1476    if 'geodataframe' in str(type(df)).lower():
1477        geometry_data = {
1478            col: df[col]
1479            for col in geometry_cols
1480        }
1481        df = pd.DataFrame({
1482            col: df[col]
1483            for col in df.columns
1484            if col not in geometry_cols
1485        })
1486        for col in geometry_cols:
1487            df[col] = pd.Series(ob for ob in geometry_data[col])
1488    for col in uuid_cols:
1489        df[col] = df[col].astype(str)
1490    for col in bytes_cols:
1491        df[col] = df[col].apply(serialize_bytes)
1492    for col in numeric_cols:
1493        df[col] = df[col].apply(serialize_decimal)
1494    for col in date_cols:
1495        df[col] = df[col].apply(serialize_date)
1496    with warnings.catch_warnings():
1497        warnings.simplefilter("ignore")
1498        for col in geometry_cols:
1499            srid = geometry_cols_srids.get(col, None) or None
1500            df[col] = pd.Series(
1501                serialize_geometry(val, geometry_format=geometry_format, srid=srid)
1502                for val in df[col]
1503            )
1504    return df.infer_objects().fillna(pd.NA).to_json(
1505        date_format=date_format,
1506        date_unit=date_unit,
1507        double_precision=double_precision,
1508        orient=orient,
1509        **kwargs
1510    )
1511
1512
1513def to_simple_lines(df: 'pd.DataFrame') -> str:
1514    """
1515    Serialize a Pandas Dataframe as lines of simple dictionaries.
1516
1517    Parameters
1518    ----------
1519    df: pd.DataFrame
1520        The dataframe to serialize into simple lines text.
1521
1522    Returns
1523    -------
1524    A string of simple line dictionaries joined by newlines.
1525    """
1526    from meerschaum.utils.misc import to_simple_dict
1527    if df is None or len(df) == 0:
1528        return ''
1529
1530    docs = df.to_dict(orient='records')
1531    return '\n'.join(to_simple_dict(doc) for doc in docs)
1532
1533
1534def parse_simple_lines(data: str) -> 'pd.DataFrame':
1535    """
1536    Parse simple lines text into a DataFrame.
1537
1538    Parameters
1539    ----------
1540    data: str
1541        The simple lines text to parse into a DataFrame.
1542
1543    Returns
1544    -------
1545    A dataframe containing the rows serialized in `data`.
1546    """
1547    from meerschaum.utils.misc import string_to_dict
1548    from meerschaum.utils.packages import import_pandas
1549    pd = import_pandas()
1550    lines = data.splitlines()
1551    try:
1552        docs = [string_to_dict(line) for line in lines]
1553        df = pd.DataFrame(docs)
1554    except Exception:
1555        df = None
1556
1557    if df is None:
1558        raise ValueError("Cannot parse simple lines into a dataframe.")
1559
1560    return df