meerschaum.utils.dtypes

   1#! /usr/bin/env python3
   2# -*- coding: utf-8 -*-
   3# vim:fenc=utf-8
   4
   5"""
   6Utility functions for working with data types.
   7"""
   8
   9import traceback
  10import json
  11import uuid
  12import time
  13import struct
  14from datetime import timezone, datetime, date, timedelta
  15from decimal import Decimal, Context, InvalidOperation, ROUND_HALF_UP
  16
  17import meerschaum as mrsm
  18from meerschaum.utils.typing import Dict, Union, Any, Optional, Tuple
  19from meerschaum.utils.warnings import warn
  20from meerschaum._internal.static import STATIC_CONFIG as _STATIC_CONFIG
  21
  22MRSM_ALIAS_DTYPES: Dict[str, str] = {
  23    'decimal': 'numeric',
  24    'Decimal': 'numeric',
  25    'number': 'numeric',
  26    'jsonl': 'json',
  27    'JSON': 'json',
  28    'binary': 'bytes',
  29    'blob': 'bytes',
  30    'varbinary': 'bytes',
  31    'bytea': 'bytes',
  32    'guid': 'uuid',
  33    'UUID': 'uuid',
  34    'geom': 'geometry',
  35    'geog': 'geography',
  36    'boolean': 'bool',
  37    'day': 'date',
  38}
  39MRSM_PD_DTYPES: Dict[Union[str, None], str] = {
  40    'json': 'object',
  41    'numeric': 'object',
  42    'geometry': 'object',
  43    'geography': 'object',
  44    'uuid': 'object',
  45    'date': 'date32[day][pyarrow]',
  46    'datetime': 'datetime64[us, UTC]',
  47    'bool': 'bool[pyarrow]',
  48    'int': 'int64[pyarrow]',
  49    'int8': 'int8[pyarrow]',
  50    'int16': 'int16[pyarrow]',
  51    'int32': 'int32[pyarrow]',
  52    'int64': 'int64[pyarrow]',
  53    'str': 'string',
  54    'bytes': 'binary[pyarrow]',
  55    None: 'object',
  56}
  57
  58MRSM_PRECISION_UNITS_SCALARS: Dict[str, Union[int, float]] = {
  59    'nanosecond': 1_000_000_000,
  60    'microsecond': 1_000_000,
  61    'millisecond': 1000,
  62    'second': 1,
  63    'minute': (1 / 60),
  64    'hour': (1 / 3600),
  65    'day': (1 / 86400),
  66}
  67
  68MRSM_PRECISION_UNITS_ALIASES: Dict[str, str] = {
  69    'ns': 'nanosecond',
  70    'us': 'microsecond',
  71    'ms': 'millisecond',
  72    's': 'second',
  73    'sec': 'second',
  74    'm': 'minute',
  75    'min': 'minute',
  76    'h': 'hour',
  77    'hr': 'hour',
  78    'd': 'day',
  79    'D': 'day',
  80}
  81MRSM_PRECISION_UNITS_ABBREVIATIONS: Dict[str, str] = {
  82    'nanosecond': 'ns',
  83    'microsecond': 'us',
  84    'millisecond': 'ms',
  85    'second': 's',
  86    'minute': 'min',
  87    'hour': 'hr',
  88    'day': 'D',
  89}
  90
  91
  92def to_pandas_dtype(dtype: str) -> str:
  93    """
  94    Cast a supported Meerschaum dtype to a Pandas dtype.
  95    """
  96    known_dtype = MRSM_PD_DTYPES.get(dtype, None)
  97    if known_dtype is not None:
  98        return known_dtype
  99
 100    alias_dtype = MRSM_ALIAS_DTYPES.get(dtype, None)
 101    if alias_dtype is not None:
 102        return MRSM_PD_DTYPES[alias_dtype]
 103
 104    if dtype.startswith('numeric'):
 105        return MRSM_PD_DTYPES['numeric']
 106
 107    if dtype.startswith('geometry'):
 108        return MRSM_PD_DTYPES['geometry']
 109
 110    if dtype.startswith('geography'):
 111        return MRSM_PD_DTYPES['geography']
 112
 113    ### NOTE: Kind of a hack, but if the first word of the given dtype is in all caps,
 114    ### treat it as a SQL db type.
 115    if dtype.split(' ')[0].isupper():
 116        from meerschaum.utils.dtypes.sql import get_pd_type_from_db_type
 117        return get_pd_type_from_db_type(dtype)
 118
 119    from meerschaum.utils.packages import attempt_import
 120    _ = attempt_import('pyarrow', lazy=False)
 121    pandas = attempt_import('pandas', lazy=False)
 122
 123    try:
 124        return str(pandas.api.types.pandas_dtype(dtype))
 125    except Exception:
 126        warn(
 127            f"Invalid dtype '{dtype}', will use 'object' instead:\n"
 128            + f"{traceback.format_exc()}",
 129            stack=False,
 130        )
 131    return 'object'
 132
 133
 134def are_dtypes_equal(
 135    ldtype: Union[str, Dict[str, str]],
 136    rdtype: Union[str, Dict[str, str]],
 137) -> bool:
 138    """
 139    Determine whether two dtype strings may be considered
 140    equivalent to avoid unnecessary conversions.
 141
 142    Parameters
 143    ----------
 144    ldtype: Union[str, Dict[str, str]]
 145        The left dtype to compare.
 146        May also provide a dtypes dictionary.
 147
 148    rdtype: Union[str, Dict[str, str]]
 149        The right dtype to compare.
 150        May also provide a dtypes dictionary.
 151
 152    Returns
 153    -------
 154    A `bool` indicating whether the two dtypes are to be considered equivalent.
 155    """
 156    if isinstance(ldtype, dict) and isinstance(rdtype, dict):
 157        lkeys = sorted([str(k) for k in ldtype.keys()])
 158        rkeys = sorted([str(k) for k in rdtype.keys()])
 159        for lkey, rkey in zip(lkeys, rkeys):
 160            if lkey != rkey:
 161                return False
 162            ltype = ldtype[lkey]
 163            rtype = rdtype[rkey]
 164            if not are_dtypes_equal(ltype, rtype):
 165                return False
 166        return True
 167
 168    try:
 169        if ldtype == rdtype:
 170            return True
 171    except Exception:
 172        warn(f"Exception when comparing dtypes, returning False:\n{traceback.format_exc()}")
 173        return False
 174
 175    ### Sometimes pandas dtype objects are passed.
 176    ldtype = str(ldtype).split('[', maxsplit=1)[0]
 177    rdtype = str(rdtype).split('[', maxsplit=1)[0]
 178
 179    if ldtype in MRSM_ALIAS_DTYPES:
 180        ldtype = MRSM_ALIAS_DTYPES[ldtype]
 181
 182    if rdtype in MRSM_ALIAS_DTYPES:
 183        rdtype = MRSM_ALIAS_DTYPES[rdtype]
 184
 185    json_dtypes = ('json', 'object')
 186    if ldtype in json_dtypes and rdtype in json_dtypes:
 187        return True
 188
 189    numeric_dtypes = ('numeric', 'decimal', 'object')
 190    if ldtype in numeric_dtypes and rdtype in numeric_dtypes:
 191        return True
 192
 193    uuid_dtypes = ('uuid', 'object')
 194    if ldtype in uuid_dtypes and rdtype in uuid_dtypes:
 195        return True
 196
 197    bytes_dtypes = ('bytes', 'object', 'binary')
 198    if ldtype in bytes_dtypes and rdtype in bytes_dtypes:
 199        return True
 200
 201    geometry_dtypes = ('geometry', 'object', 'geography')
 202    if ldtype in geometry_dtypes and rdtype in geometry_dtypes:
 203        return True
 204
 205    if ldtype.lower() == rdtype.lower():
 206        return True
 207
 208    datetime_dtypes = ('datetime', 'timestamp')
 209    ldtype_found_dt_prefix = False
 210    rdtype_found_dt_prefix = False
 211    for dt_prefix in datetime_dtypes:
 212        ldtype_found_dt_prefix = (dt_prefix in ldtype.lower()) or ldtype_found_dt_prefix
 213        rdtype_found_dt_prefix = (dt_prefix in rdtype.lower()) or rdtype_found_dt_prefix
 214    if ldtype_found_dt_prefix and rdtype_found_dt_prefix:
 215        return True
 216
 217    string_dtypes = ('str', 'string', 'object')
 218    if ldtype in string_dtypes and rdtype in string_dtypes:
 219        return True
 220
 221    int_dtypes = (
 222        'int', 'int64', 'int32', 'int16', 'int8',
 223        'uint', 'uint64', 'uint32', 'uint16', 'uint8',
 224    )
 225    int_substrings = ('int',)
 226    if ldtype.lower() in int_dtypes and rdtype.lower() in int_dtypes:
 227        return True
 228    for substring in int_substrings:
 229        if substring in ldtype.lower() and substring in rdtype.lower():
 230            return True
 231
 232    float_dtypes = ('float', 'float64', 'float32', 'float16', 'float128', 'double')
 233    if ldtype.lower() in float_dtypes and rdtype.lower() in float_dtypes:
 234        return True
 235
 236    bool_dtypes = ('bool', 'boolean')
 237    if ldtype in bool_dtypes and rdtype in bool_dtypes:
 238        return True
 239
 240    date_dtypes = (
 241        'date', 'date32', 'date32[pyarrow]', 'date32[day][pyarrow]',
 242        'date64', 'date64[pyarrow]', 'date64[ms][pyarrow]',
 243    )
 244    if ldtype in date_dtypes and rdtype in date_dtypes:
 245        return True
 246
 247    return False
 248
 249
 250def is_dtype_numeric(dtype: str) -> bool:
 251    """
 252    Determine whether a given `dtype` string
 253    should be considered compatible with the Meerschaum dtype `numeric`.
 254
 255    Parameters
 256    ----------
 257    dtype: str
 258        The pandas-like dtype string.
 259
 260    Returns
 261    -------
 262    A bool indicating the dtype is compatible with `numeric`.
 263    """
 264    dtype_lower = dtype.lower()
 265
 266    acceptable_substrings = ('numeric', 'float', 'double', 'int')
 267    for substring in acceptable_substrings:
 268        if substring in dtype_lower:
 269            return True
 270
 271    return False
 272
 273
 274def attempt_cast_to_numeric(
 275    value: Any,
 276    quantize: bool = False,
 277    precision: Optional[int] = None,
 278    scale: Optional[int] = None,
 279)-> Any:
 280    """
 281    Given a value, attempt to coerce it into a numeric (Decimal).
 282
 283    Parameters
 284    ----------
 285    value: Any
 286        The value to be cast to a Decimal.
 287
 288    quantize: bool, default False
 289        If `True`, quantize the decimal to the specified precision and scale.
 290
 291    precision: Optional[int], default None
 292        If `quantize` is `True`, use this precision.
 293
 294    scale: Optional[int], default None
 295        If `quantize` is `True`, use this scale.
 296
 297    Returns
 298    -------
 299    A `Decimal` if possible, or `value`.
 300    """
 301    if isinstance(value, Decimal):
 302        if quantize and precision and scale:
 303            return quantize_decimal(value, precision, scale)
 304        return value
 305    try:
 306        if value_is_null(value):
 307            return Decimal('NaN')
 308
 309        dec = Decimal(str(value))
 310        if not quantize or not precision or not scale:
 311            return dec
 312        return quantize_decimal(dec, precision, scale)
 313    except Exception:
 314        return value
 315
 316
 317def attempt_cast_to_uuid(value: Any) -> Any:
 318    """
 319    Given a value, attempt to coerce it into a UUID (`uuid4`).
 320    """
 321    if isinstance(value, uuid.UUID):
 322        return value
 323    try:
 324        return (
 325            uuid.UUID(str(value))
 326            if not value_is_null(value)
 327            else None
 328        )
 329    except Exception:
 330        return value
 331
 332
 333def attempt_cast_to_bytes(value: Any) -> Any:
 334    """
 335    Given a value, attempt to coerce it into a bytestring.
 336    """
 337    if isinstance(value, bytes):
 338        return value
 339    try:
 340        return (
 341            deserialize_bytes_string(str(value))
 342            if not value_is_null(value)
 343            else None
 344        )
 345    except Exception:
 346        return value
 347
 348
 349def attempt_cast_to_geometry(value: Any) -> Any:
 350    """
 351    Given a value, attempt to coerce it into a `shapely` (`geometry`) object.
 352    """
 353    typ = str(type(value))
 354    if 'pandas' in typ and 'Series' in typ:
 355        if 'GeoSeries' in typ:
 356            return value
 357
 358        gpd = mrsm.attempt_import('geopandas', lazy=False)
 359        if len(value) == 0:
 360            return gpd.GeoSeries([])
 361
 362        ix = value.first_valid_index()
 363        if ix is None:
 364            try:
 365                return gpd.GeoSeries(value)
 366            except Exception:
 367                traceback.print_exc()
 368                return gpd.GeoSeries(attempt_cast_to_geometry(val) for val in value)
 369
 370        sample_val = value[ix]
 371        sample_typ = str(type(sample_val))
 372        if 'shapely' in sample_typ:
 373            try:
 374                return gpd.GeoSeries(value)
 375            except Exception:
 376                traceback.print_exc()
 377                return gpd.GeoSeries(attempt_cast_to_geometry(val) for val in value)
 378
 379        sample_is_gpkg = geometry_is_gpkg(sample_val)
 380        if sample_is_gpkg:
 381            try:
 382                value = value.apply(lambda x: gpkg_wkb_to_wkb(x)[0])
 383            except Exception:
 384                traceback.print_exc()
 385                return gpd.GeoSeries(attempt_cast_to_geometry(val) for val in value)
 386
 387        sample_is_wkt = geometry_is_wkt(sample_val) if not sample_is_gpkg else False
 388        try:
 389            return (
 390                gpd.GeoSeries.from_wkt(value)
 391                if sample_is_wkt
 392                else gpd.GeoSeries.from_wkb(value)
 393            )
 394        except Exception:
 395            traceback.print_exc()
 396            return gpd.GeoSeries(attempt_cast_to_geometry(val) for val in value)
 397
 398    if 'shapely' in typ:
 399        return value
 400
 401    shapely, shapely_wkt, shapely_wkb = mrsm.attempt_import(
 402        'shapely',
 403        'shapely.wkt',
 404        'shapely.wkb',
 405        lazy=False,
 406    )
 407
 408    if isinstance(value, (dict, list)):
 409        try:
 410            return shapely.from_geojson(json.dumps(value))
 411        except Exception:
 412            return value
 413
 414    value_is_gpkg = geometry_is_gpkg(value)
 415    if value_is_gpkg:
 416        try:
 417            wkb_data, _, _ = gpkg_wkb_to_wkb(value)
 418            return shapely_wkb.loads(wkb_data)
 419        except Exception:
 420            return value
 421
 422    value_is_wkt = geometry_is_wkt(value)
 423    if value_is_wkt is None:
 424        return value
 425
 426    try:
 427        return (
 428            shapely_wkt.loads(value)
 429            if value_is_wkt
 430            else shapely_wkb.loads(value)
 431        )
 432    except Exception:
 433        pass
 434
 435    return value
 436
 437
 438def geometry_is_wkt(value: Union[str, bytes]) -> Union[bool, None]:
 439    """
 440    Determine whether an input value should be treated as WKT or WKB geometry data.
 441
 442    Parameters
 443    ----------
 444    value: Union[str, bytes]
 445        The input data to be parsed into geometry data.
 446
 447    Returns
 448    -------
 449    A `bool` (`True` if `value` is WKT and `False` if it should be treated as WKB).
 450    Return `None` if `value` should be parsed as neither.
 451    """
 452    import re
 453    if not isinstance(value, (str, bytes)):
 454        return None
 455
 456    if isinstance(value, bytes):
 457        return False
 458    
 459    wkt_pattern = r'^\s*(POINT|LINESTRING|POLYGON|MULTIPOINT|MULTILINESTRING|MULTIPOLYGON|GEOMETRYCOLLECTION)\s*\(.*\)\s*$'
 460    if re.match(wkt_pattern, value, re.IGNORECASE):
 461        return True
 462    
 463    if all(c in '0123456789ABCDEFabcdef' for c in value) and len(value) % 2 == 0:
 464        return False
 465    
 466    return None
 467
 468
 469def geometry_is_gpkg(value: bytes) -> bool:
 470    """
 471    Return whether the input `value` is formatted as GeoPackage WKB.
 472    """
 473    if not isinstance(value, bytes) or len(value) < 2:
 474        return False
 475
 476    return value[0:2] == b'GP'
 477
 478def gpkg_wkb_to_wkb(gpkg_wkb_bytes: bytes) -> Tuple[bytes, int, bytes]:
 479    """
 480    Converts GeoPackage WKB to standard WKB by removing the header.
 481    
 482    Parameters
 483    ----------
 484    gpkg_wkb_bytes: bytes
 485        The GeoPackage WKB byte string.
 486        
 487    Returns
 488    -------
 489    A tuple containing the standard WKB bytes, SRID, and flags.
 490    """
 491    magic_number = gpkg_wkb_bytes[0:2]
 492    if magic_number != b'GP':
 493        raise ValueError("Invalid GeoPackage WKB header: missing magic number.")
 494    
 495    try:
 496        header = gpkg_wkb_bytes[0:8]
 497        header_vals = struct.unpack('<ccBBi', header)
 498        flags = header_vals[-2]
 499        srid = header_vals[-1]
 500    except struct.error:
 501        header = gpkg_wkb_bytes[0:6]
 502        header_vals = struct.unpack('<ccBBh', header)
 503        flags = header_vals[-2]
 504        srid = header_vals[-1]
 505
 506    envelope_type = (flags >> 1) & 0x07
 507    envelope_sizes = {
 508        0: 0,
 509        1: 32,
 510        2: 48,
 511        3: 48,
 512        4: 64,
 513    }
 514    header_length = 8 + envelope_sizes.get(envelope_type, 0)
 515    standard_wkb_bytes = gpkg_wkb_bytes[header_length:]
 516    return standard_wkb_bytes, srid, flags
 517
 518
 519def value_is_null(value: Any) -> bool:
 520    """
 521    Determine if a value is a null-like string.
 522    """
 523    return str(value).lower() in ('none', 'nan', 'na', 'nat', 'natz', '', '<na>')
 524
 525
 526def none_if_null(value: Any) -> Any:
 527    """
 528    Return `None` if a value is a null-like string.
 529    """
 530    return (None if value_is_null(value) else value)
 531
 532
 533def quantize_decimal(x: Decimal, precision: int, scale: int) -> Decimal:
 534    """
 535    Quantize a given `Decimal` to a known scale and precision.
 536
 537    Parameters
 538    ----------
 539    x: Decimal
 540        The `Decimal` to be quantized.
 541
 542    precision: int
 543        The total number of significant digits.
 544
 545    scale: int
 546        The number of significant digits after the decimal point.
 547
 548    Returns
 549    -------
 550    A `Decimal` quantized to the specified scale and precision.
 551    """
 552    precision_decimal = Decimal(('1' * (precision - scale)) + '.' + ('1' * scale))
 553    try:
 554        return x.quantize(precision_decimal, context=Context(prec=precision), rounding=ROUND_HALF_UP)
 555    except InvalidOperation:
 556        pass
 557
 558    raise ValueError(f"Cannot quantize value '{x}' to {precision=}, {scale=}.")
 559
 560
 561def serialize_decimal(
 562    x: Any,
 563    quantize: bool = False,
 564    precision: Optional[int] = None,
 565    scale: Optional[int] = None,
 566) -> Any:
 567    """
 568    Return a quantized string of an input decimal.
 569
 570    Parameters
 571    ----------
 572    x: Any
 573        The potential decimal to be serialized.
 574
 575    quantize: bool, default False
 576        If `True`, quantize the incoming Decimal to the specified scale and precision
 577        before serialization.
 578
 579    precision: Optional[int], default None
 580        The precision of the decimal to be quantized.
 581
 582    scale: Optional[int], default None
 583        The scale of the decimal to be quantized.
 584
 585    Returns
 586    -------
 587    A string of the input decimal or the input if not a Decimal.
 588    """
 589    if not isinstance(x, Decimal):
 590        return x
 591
 592    if value_is_null(x):
 593        return None
 594
 595    if quantize and scale and precision:
 596        x = quantize_decimal(x, precision, scale)
 597
 598    return f"{x:f}"
 599
 600
 601def coerce_timezone(
 602    dt: Any,
 603    strip_utc: bool = False,
 604) -> Any:
 605    """
 606    Given a `datetime`, pandas `Timestamp` or `Series` of `Timestamp`,
 607    return a UTC timestamp (strip timezone if `strip_utc` is `True`.
 608    """
 609    if dt is None:
 610        return None
 611
 612    if isinstance(dt, int):
 613        return dt
 614
 615    if isinstance(dt, str):
 616        dateutil_parser = mrsm.attempt_import('dateutil.parser')
 617        try:
 618            dt = dateutil_parser.parse(dt)
 619        except Exception:
 620            return dt
 621
 622    dt_is_series = hasattr(dt, 'dtype') and hasattr(dt, '__module__')
 623    if dt_is_series:
 624        pandas = mrsm.attempt_import('pandas', lazy=False)
 625
 626        if (
 627            pandas.api.types.is_datetime64_any_dtype(dt) and (
 628                (dt.dt.tz is not None and not strip_utc)
 629                or
 630                (dt.dt.tz is None and strip_utc)
 631            )
 632        ):
 633            return dt
 634
 635        dt_series = to_datetime(dt, coerce_utc=False)
 636        if dt_series.dt.tz is None:
 637            dt_series = dt_series.dt.tz_localize(timezone.utc)
 638        if strip_utc:
 639            try:
 640                if dt_series.dt.tz is not None:
 641                    dt_series = dt_series.dt.tz_localize(None)
 642            except Exception:
 643                pass
 644
 645        return dt_series
 646
 647    if dt.tzinfo is None:
 648        if strip_utc:
 649            return dt
 650        return dt.replace(tzinfo=timezone.utc)
 651
 652    utc_dt = dt.astimezone(timezone.utc)
 653    if strip_utc:
 654        return utc_dt.replace(tzinfo=None)
 655    return utc_dt
 656
 657
 658def to_datetime(
 659    dt_val: Any,
 660    as_pydatetime: bool = False,
 661    coerce_utc: bool = True,
 662    precision_unit: Optional[str] = None,
 663) -> Any:
 664    """
 665    Wrap `pd.to_datetime()` and add support for out-of-bounds values.
 666
 667    Parameters
 668    ----------
 669    dt_val: Any
 670        The value to coerce to Pandas Timestamps.
 671
 672    as_pydatetime: bool, default False
 673        If `True`, return a Python datetime object.
 674
 675    coerce_utc: bool, default True
 676        If `True`, ensure the value has UTC tzinfo.
 677
 678    precision_unit: Optional[str], default None
 679        If provided, enforce the provided precision unit.
 680    """
 681    pandas, dateutil_parser = mrsm.attempt_import('pandas', 'dateutil.parser', lazy=False)
 682    is_dask = 'dask' in getattr(dt_val, '__module__', '')
 683    dd = mrsm.attempt_import('dask.dataframe') if is_dask else None
 684    dt_is_series = hasattr(dt_val, 'dtype') and hasattr(dt_val, '__module__')
 685    pd = pandas if dd is None else dd
 686    enforce_precision = precision_unit is not None
 687    precision_unit = precision_unit or 'microsecond'
 688    true_precision_unit = MRSM_PRECISION_UNITS_ALIASES.get(precision_unit, precision_unit)
 689    precision_abbreviation = MRSM_PRECISION_UNITS_ABBREVIATIONS.get(true_precision_unit, None)
 690    if not precision_abbreviation:
 691        raise ValueError(f"Invalid precision '{precision_unit}'.")
 692
 693    def parse(x: Any) -> Any:
 694        try:
 695            return dateutil_parser.parse(x)
 696        except Exception:
 697            return x
 698
 699    def check_dtype(dtype_to_check: str, with_utc: bool = True) -> bool:
 700        dtype_check_against = (
 701            f"datetime64[{precision_abbreviation}, UTC]"
 702            if with_utc
 703            else f"datetime64[{precision_abbreviation}]"
 704        )
 705        return (
 706            dtype_to_check == dtype_check_against
 707            if enforce_precision
 708            else (
 709                dtype_to_check.startswith('datetime64[')
 710                and (
 711                    ('utc' in dtype_to_check.lower())
 712                    if with_utc
 713                    else ('utc' not in dtype_to_check.lower())
 714                )
 715            )
 716        )
 717
 718    if isinstance(dt_val, pd.Timestamp):
 719        dt_val_to_return = dt_val if not as_pydatetime else dt_val.to_pydatetime()
 720        return (
 721            coerce_timezone(dt_val_to_return)
 722            if coerce_utc
 723            else dt_val_to_return
 724        )
 725
 726    if dt_is_series:
 727        changed_tz = False
 728        original_tz = None
 729        dtype = str(getattr(dt_val, 'dtype', 'object'))
 730        if (
 731            are_dtypes_equal(dtype, 'datetime')
 732            and 'utc' not in dtype.lower()
 733            and hasattr(dt_val, 'dt')
 734        ):
 735            original_tz = dt_val.dt.tz
 736            dt_val = dt_val.dt.tz_localize(timezone.utc)
 737            changed_tz = True
 738            dtype = str(getattr(dt_val, 'dtype', 'object'))
 739        try:
 740            new_dt_series = (
 741                dt_val
 742                if check_dtype(dtype, with_utc=True)
 743                else dt_val.astype(f"datetime64[{precision_abbreviation}, UTC]")
 744            )
 745        except pd.errors.OutOfBoundsDatetime:
 746            try:
 747                next_precision = get_next_precision_unit(true_precision_unit)
 748                next_precision_abbrevation = MRSM_PRECISION_UNITS_ABBREVIATIONS[next_precision]
 749                new_dt_series = dt_val.astype(f"datetime64[{next_precision_abbrevation}, UTC]")
 750            except Exception:
 751                new_dt_series = None
 752        except ValueError:
 753            new_dt_series = None
 754        except TypeError:
 755            try:
 756                new_dt_series = (
 757                    new_dt_series
 758                    if check_dtype(str(getattr(new_dt_series, 'dtype', None)), with_utc=False)
 759                    else dt_val.astype(f"datetime64[{precision_abbreviation}]")
 760                )
 761            except Exception:
 762                new_dt_series = None
 763
 764        if new_dt_series is None:
 765            new_dt_series = dt_val.apply(lambda x: parse(str(x)))
 766
 767        if coerce_utc:
 768            return coerce_timezone(new_dt_series)
 769
 770        if changed_tz:
 771            new_dt_series = new_dt_series.dt.tz_localize(original_tz)
 772        return new_dt_series
 773
 774    try:
 775        new_dt_val = pd.to_datetime(dt_val, utc=True, format='ISO8601')
 776        if new_dt_val.unit != precision_abbreviation:
 777            new_dt_val = new_dt_val.as_unit(precision_abbreviation)
 778        if as_pydatetime:
 779            return new_dt_val.to_pydatetime()
 780        return new_dt_val
 781    except (pd.errors.OutOfBoundsDatetime, ValueError):
 782        pass
 783
 784    new_dt_val = parse(dt_val)
 785    if not coerce_utc:
 786        return new_dt_val
 787    return coerce_timezone(new_dt_val)
 788
 789
 790def serialize_bytes(data: bytes) -> str:
 791    """
 792    Return the given bytes as a base64-encoded string.
 793    """
 794    import base64
 795    if not isinstance(data, bytes) and value_is_null(data):
 796        return data
 797    return base64.b64encode(data).decode('utf-8')
 798
 799
 800def serialize_geometry(
 801    geom: Any,
 802    geometry_format: str = 'wkb_hex',
 803    srid: Optional[int] = None,
 804) -> Union[str, Dict[str, Any], bytes, None]:
 805    """
 806    Serialize geometry data as WKB, WKB (hex), GPKG-WKB, WKT, or GeoJSON. 
 807
 808    Parameters
 809    ----------
 810    geom: Any
 811        The potential geometry data to be serialized.
 812
 813    geometry_format: str, default 'wkb_hex'
 814        The serialization format for geometry data.
 815        Accepted formats are `wkb`, `wkb_hex`, `wkt`, `geojson`, and `gpkg_wkb`.
 816
 817    srid: Optional[int], default None
 818        If provided, use this as the source CRS when serializing to GeoJSON.
 819
 820    Returns
 821    -------
 822    A string containing the geometry data, or bytes, or a dictionary, or None.
 823    """
 824    if value_is_null(geom):
 825        return None
 826
 827    shapely, shapely_ops, pyproj, np = mrsm.attempt_import(
 828        'shapely', 'shapely.ops', 'pyproj', 'numpy',
 829        lazy=False,
 830    )
 831    if geometry_format == 'geojson':
 832        if srid:
 833            transformer = pyproj.Transformer.from_crs(f"EPSG:{srid}", "EPSG:4326", always_xy=True)
 834            geom = shapely_ops.transform(transformer.transform, geom)
 835        geojson_str = shapely.to_geojson(geom)
 836        return json.loads(geojson_str)
 837
 838    if not hasattr(geom, 'wkb_hex'):
 839        return str(geom)
 840
 841    byte_order = 1 if np.little_endian else 0
 842
 843    if geometry_format.startswith("wkb"):
 844        return shapely.to_wkb(geom, hex=(geometry_format=="wkb_hex"), include_srid=True)
 845    elif geometry_format == 'gpkg_wkb':
 846        wkb_data = shapely.to_wkb(geom, hex=False, byte_order=byte_order)
 847        flags = (
 848            ((byte_order & 0x01) | (0x20))
 849            if geom.is_empty
 850            else (byte_order & 0x01)
 851        )
 852        srid_val = srid or -1
 853        header = struct.pack(
 854            '<ccBBi',
 855            b'G', b'P',
 856            0,
 857            flags,
 858            srid_val
 859        )
 860        return header + wkb_data
 861
 862    return shapely.to_wkt(geom)
 863
 864
 865def deserialize_geometry(geom_wkb: Union[str, bytes]):
 866    """
 867    Deserialize a WKB string into a shapely geometry object.
 868    """
 869    shapely = mrsm.attempt_import('shapely', lazy=False)
 870    return shapely.wkb.loads(geom_wkb)
 871
 872
 873def project_geometry(geom, srid: int, to_srid: int = 4326):
 874    """
 875    Project a shapely geometry object to a new CRS (SRID).
 876    """
 877    pyproj, shapely_ops = mrsm.attempt_import('pyproj', 'shapely.ops', lazy=False)
 878    transformer = pyproj.Transformer.from_crs(f"EPSG:{srid}", f"EPSG:{to_srid}", always_xy=True)
 879    return shapely_ops.transform(transformer.transform, geom)
 880
 881
 882def deserialize_bytes_string(data: Optional[str], force_hex: bool = False) -> Union[bytes, None]:
 883    """
 884    Given a serialized ASCII string of bytes data, return the original bytes.
 885    The input data may either be base64- or hex-encoded.
 886
 887    Parameters
 888    ----------
 889    data: Optional[str]
 890        The string to be deserialized into bytes.
 891        May be base64- or hex-encoded (prefixed with `'\\x'`).
 892
 893    force_hex: bool = False
 894        If `True`, treat the input string as hex-encoded.
 895        If `data` does not begin with the prefix `'\\x'`, set `force_hex` to `True`.
 896        This will still strip the leading `'\\x'` prefix if present.
 897
 898    Returns
 899    -------
 900    The original bytes used to produce the encoded string `data`.
 901    """
 902    if not isinstance(data, str) and value_is_null(data):
 903        return data
 904
 905    import binascii
 906    import base64
 907
 908    is_hex = force_hex or data.startswith('\\x')
 909
 910    if is_hex:
 911        if data.startswith('\\x'):
 912            data = data[2:]
 913        return binascii.unhexlify(data)
 914
 915    return base64.b64decode(data)
 916
 917
 918def deserialize_base64(data: str) -> bytes:
 919    """
 920    Return the original bytestring from the given base64-encoded string.
 921    """
 922    import base64
 923    return base64.b64decode(data)
 924
 925
 926def encode_bytes_for_bytea(data: bytes, with_prefix: bool = True) -> Union[str, None]:
 927    """
 928    Return the given bytes as a hex string for PostgreSQL's `BYTEA` type.
 929    """
 930    import binascii
 931    if not isinstance(data, bytes) and value_is_null(data):
 932        return data
 933    return ('\\x' if with_prefix else '') + binascii.hexlify(data).decode('utf-8')
 934
 935
 936def serialize_datetime(dt: datetime) -> Union[str, None]:
 937    """
 938    Serialize a datetime object into JSON (ISO format string).
 939
 940    Examples
 941    --------
 942    >>> import json
 943    >>> from datetime import datetime
 944    >>> json.dumps({'a': datetime(2022, 1, 1)}, default=json_serialize_datetime)
 945    '{"a": "2022-01-01T00:00:00Z"}'
 946
 947    """
 948    if not hasattr(dt, 'isoformat'):
 949        return None
 950
 951    tz_suffix = 'Z' if getattr(dt, 'tzinfo', None) is None else ''
 952    return dt.isoformat() + tz_suffix
 953
 954
 955def serialize_date(d: date) -> Union[str, None]:
 956    """
 957    Serialize a date object into its ISO representation.
 958    """
 959    return d.isoformat() if hasattr(d, 'isoformat') else None
 960
 961
 962def json_serialize_value(x: Any, default_to_str: bool = True) -> Union[str, None]:
 963    """
 964    Serialize the given value to a JSON value. Accounts for datetimes, bytes, decimals, etc.
 965
 966    Parameters
 967    ----------
 968    x: Any
 969        The value to serialize.
 970
 971    default_to_str: bool, default True
 972        If `True`, return a string of `x` if x is not a designated type.
 973        Otherwise return x.
 974
 975    Returns
 976    -------
 977    A serialized version of x, or x.
 978    """
 979    if isinstance(x, (mrsm.Pipe, mrsm.connectors.Connector)):
 980        return x.meta
 981
 982    if hasattr(x, 'tzinfo'):
 983        return serialize_datetime(x)
 984
 985    if hasattr(x, 'isoformat'):
 986        return serialize_date(x)
 987
 988    if isinstance(x, bytes):
 989        return serialize_bytes(x)
 990
 991    if isinstance(x, Decimal):
 992        return serialize_decimal(x)
 993
 994    if 'shapely' in str(type(x)):
 995        return serialize_geometry(x)
 996
 997    if value_is_null(x):
 998        return None
 999
1000    if isinstance(x, (dict, list, tuple)):
1001        return json.dumps(x, default=json_serialize_value, separators=(',', ':'))
1002
1003    return str(x) if default_to_str else x
1004
1005
1006def get_geometry_type_srid(
1007    dtype: str = 'geometry',
1008    default_type: str = 'geometry',
1009    default_srid: Union[int, str] = 0,
1010) -> Tuple[str, Union[int, str, None]]:
1011    """
1012    Given the specified geometry `dtype`, return a tuple in the form (type, SRID).
1013
1014    Parameters
1015    ----------
1016    dtype: Optional[str], default None
1017        Optionally provide a specific `geometry` syntax (e.g. `geometry[MultiLineString, 4326]`).
1018        You may specify a supported `shapely` geometry type and an SRID in the dtype modifier:
1019
1020        - `Point`
1021        - `LineString`
1022        - `LinearRing`
1023        - `Polygon`
1024        - `MultiPoint`
1025        - `MultiLineString`
1026        - `MultiPolygon`
1027        - `GeometryCollection`
1028
1029    Returns
1030    -------
1031    A tuple in the form (type, SRID).
1032    Defaults to `(default_type, default_srid)`.
1033
1034    Examples
1035    --------
1036    >>> from meerschaum.utils.dtypes import get_geometry_type_srid
1037    >>> get_geometry_type_srid()
1038    ('geometry', 4326)
1039    >>> get_geometry_type_srid('geometry[]')
1040    ('geometry', 4326)
1041    >>> get_geometry_type_srid('geometry[Point, 0]')
1042    ('Point', 0)
1043    >>> get_geometry_type_srid('geometry[0, Point]')
1044    ('Point', 0)
1045    >>> get_geometry_type_srid('geometry[0]')
1046    ('geometry', 0)
1047    >>> get_geometry_type_srid('geometry[MULTILINESTRING, 4326]')
1048    ('MultiLineString', 4326)
1049    >>> get_geometry_type_srid('geography')
1050    ('geometry', 0)
1051    >>> get_geometry_type_srid('geography[POINT]')
1052    ('Point', 0)
1053    >>> get_geometry_type_srid('geometry[POINT, ESRI:102003]')
1054    ('Point', 'ESRI:102003')
1055    """
1056    from meerschaum.utils.misc import is_int
1057    ### NOTE: PostGIS syntax must also be parsed.
1058    dtype = dtype.replace('(', '[').replace(')', ']')
1059    bare_dtype = dtype.split('[', maxsplit=1)[0]
1060    modifier = dtype.split(bare_dtype, maxsplit=1)[-1].lstrip('[').rstrip(']')
1061    if not modifier:
1062        return default_type, default_srid
1063
1064    parts = [
1065        part.split('=')[-1].strip()
1066        for part in modifier.split(',')
1067    ]
1068    parts_casted = [
1069        (
1070            int(part)
1071            if is_int(part)
1072            else part
1073        )
1074        for part in parts
1075    ]
1076
1077    srid = default_srid
1078    geometry_type = default_type
1079
1080    for part in parts_casted:
1081        if isinstance(part, int) or ':' in str(part):
1082            srid = part
1083            break
1084
1085    for part in parts_casted:
1086        if isinstance(part, str) and part != srid:
1087            geometry_type = part
1088            break
1089
1090    return geometry_type, srid
1091
1092
1093def get_current_timestamp(
1094    precision_unit: str = _STATIC_CONFIG['dtypes']['datetime']['default_precision_unit'],
1095    precision_interval: int = 1,
1096    round_to: str = 'down',
1097    as_pandas: bool = False,
1098    as_int: bool = False,
1099    _now: Union[datetime, int, None] = None,
1100) -> 'Union[datetime, pd.Timestamp, int]':
1101    """
1102    Return the current UTC timestamp to nanosecond precision.
1103
1104    Parameters
1105    ----------
1106    precision_unit: str, default 'us'
1107        The precision of the timestamp to be returned.
1108        Valid values are the following:
1109            - `ns` / `nanosecond`
1110            - `us` / `microsecond`
1111            - `ms` / `millisecond`
1112            - `s` / `sec` / `second`
1113            - `m` / `min` / `minute`
1114            - `h` / `hr` / `hour`
1115            - `d` / `day`
1116
1117    precision_interval: int, default 1
1118        Round the timestamp to the `precision_interval` units.
1119        For example, `precision='minute'` and `precision_interval=15` will round to 15-minute intervals.
1120        Note: `precision_interval` must be 1 when `precision='nanosecond'`.
1121
1122    round_to: str, default 'down'
1123        The direction to which to round the timestamp.
1124        Available options are `down`, `up`, and `closest`.
1125
1126    as_pandas: bool, default False
1127        If `True`, return a Pandas Timestamp.
1128        This is always true if `unit` is `nanosecond`.
1129
1130    as_int: bool, default False
1131        If `True`, return the timestamp to an integer.
1132        Overrides `as_pandas`.
1133
1134    Returns
1135    -------
1136    A Pandas Timestamp, datetime object, or integer with precision to the provided unit.
1137
1138    Examples
1139    --------
1140    >>> get_current_timestamp('ns')
1141    Timestamp('2025-07-17 17:59:16.423644369+0000', tz='UTC')
1142    >>> get_current_timestamp('ms')
1143    Timestamp('2025-07-17 17:59:16.424000+0000', tz='UTC')
1144    """
1145    true_precision_unit = MRSM_PRECISION_UNITS_ALIASES.get(precision_unit, precision_unit)
1146    if true_precision_unit not in MRSM_PRECISION_UNITS_SCALARS:
1147        from meerschaum.utils.misc import items_str
1148        raise ValueError(
1149            f"Unknown precision unit '{precision_unit}'. "
1150            "Accepted values are "
1151            f"{items_str(list(MRSM_PRECISION_UNITS_SCALARS) + list(MRSM_PRECISION_UNITS_ALIASES))}."
1152        )
1153
1154    if not as_int:
1155        as_pandas = as_pandas or true_precision_unit == 'nanosecond'
1156    pd = mrsm.attempt_import('pandas', lazy=False) if as_pandas else None
1157
1158    if true_precision_unit == 'nanosecond':
1159        if precision_interval != 1:
1160            warn("`precision_interval` must be 1 for nanosecond precision.")
1161        now_ts = time.time_ns() if not isinstance(_now, int) else _now
1162        if as_int:
1163            return now_ts
1164        return pd.to_datetime(now_ts, unit='ns', utc=True)
1165
1166    now = datetime.now(timezone.utc) if not isinstance(_now, datetime) else _now
1167    delta = timedelta(**{true_precision_unit + 's': precision_interval})
1168    rounded_now = round_time(now, delta, to=round_to)
1169
1170    if as_int:
1171        return int(rounded_now.timestamp() * MRSM_PRECISION_UNITS_SCALARS[true_precision_unit])
1172
1173    ts_val = (
1174        pd.to_datetime(rounded_now, utc=True)
1175        if as_pandas
1176        else rounded_now
1177    )
1178
1179    if not as_pandas:
1180        return ts_val
1181
1182    as_unit_precisions = ('microsecond', 'millisecond', 'second')
1183    if true_precision_unit not in as_unit_precisions:
1184        return ts_val
1185
1186    return ts_val.as_unit(MRSM_PRECISION_UNITS_ABBREVIATIONS[true_precision_unit])
1187
1188
1189def is_dtype_special(type_: str) -> bool:
1190    """
1191    Return whether a dtype should be treated as a special Meerschaum dtype.
1192    This is not the same as a Meerschaum alias.
1193    """
1194    true_type = MRSM_ALIAS_DTYPES.get(type_, type_)
1195    if true_type in (
1196        'uuid',
1197        'json',
1198        'bytes',
1199        'numeric',
1200        'datetime',
1201        'geometry',
1202        'geography',
1203        'date',
1204        'bool',
1205    ):
1206        return True
1207
1208    if are_dtypes_equal(true_type, 'datetime'):
1209        return True
1210
1211    if are_dtypes_equal(true_type, 'date'):
1212        return True
1213
1214    if true_type.startswith('numeric'):
1215        return True
1216
1217    if true_type.startswith('bool'):
1218        return True
1219
1220    if true_type.startswith('geometry'):
1221        return True
1222
1223    if true_type.startswith('geography'):
1224        return True
1225
1226    return False
1227
1228
1229def get_next_precision_unit(precision_unit: str, decrease: bool = True) -> str:
1230    """
1231    Get the next precision string in order of value.
1232
1233    Parameters
1234    ----------
1235    precision_unit: str
1236        The precision string (`'nanosecond'`, `'ms'`, etc.).
1237
1238    decrease: bool, defaul True
1239        If `True` return the precision unit which is lower (e.g. `nanosecond` -> `millisecond`).
1240        If `False`, return the precision unit which is higher.
1241
1242    Returns
1243    -------
1244    A `precision` string which is lower or higher than the given precision unit.
1245
1246    Examples
1247    --------
1248    >>> get_next_precision_unit('nanosecond')
1249    'microsecond'
1250    >>> get_next_precision_unit('ms')
1251    'second'
1252    >>> get_next_precision_unit('hour', decrease=False)
1253    'minute'
1254    """
1255    true_precision_unit = MRSM_PRECISION_UNITS_ALIASES.get(precision_unit, precision_unit)
1256    precision_scalar = MRSM_PRECISION_UNITS_SCALARS.get(true_precision_unit, None)
1257    if not precision_scalar:
1258        raise ValueError(f"Invalid precision unit '{precision_unit}'.")
1259
1260    precisions = sorted(
1261        list(MRSM_PRECISION_UNITS_SCALARS),
1262        key=lambda p: MRSM_PRECISION_UNITS_SCALARS[p]
1263    )
1264
1265    precision_index = precisions.index(true_precision_unit)
1266    new_precision_index = precision_index + (-1 if decrease else 1)
1267    if new_precision_index < 0 or new_precision_index >= len(precisions):
1268        raise ValueError(f"No precision {'below' if decrease else 'above'} '{precision_unit}'.")
1269
1270    return precisions[new_precision_index]
1271
1272
1273def round_time(
1274    dt: Optional[datetime] = None,
1275    date_delta: Optional[timedelta] = None,
1276    to: 'str' = 'down'
1277) -> datetime:
1278    """
1279    Round a datetime object to a multiple of a timedelta.
1280
1281    Parameters
1282    ----------
1283    dt: Optional[datetime], default None
1284        If `None`, grab the current UTC datetime.
1285
1286    date_delta: Optional[timedelta], default None
1287        If `None`, use a delta of 1 minute.
1288
1289    to: 'str', default 'down'
1290        Available options are `'up'`, `'down'`, and `'closest'`.
1291
1292    Returns
1293    -------
1294    A rounded `datetime` object.
1295
1296    Examples
1297    --------
1298    >>> round_time(datetime(2022, 1, 1, 12, 15, 57, 200))
1299    datetime.datetime(2022, 1, 1, 12, 15)
1300    >>> round_time(datetime(2022, 1, 1, 12, 15, 57, 200), to='up')
1301    datetime.datetime(2022, 1, 1, 12, 16)
1302    >>> round_time(datetime(2022, 1, 1, 12, 15, 57, 200), timedelta(hours=1))
1303    datetime.datetime(2022, 1, 1, 12, 0)
1304    >>> round_time(
1305    ...   datetime(2022, 1, 1, 12, 15, 57, 200),
1306    ...   timedelta(hours=1),
1307    ...   to = 'closest'
1308    ... )
1309    datetime.datetime(2022, 1, 1, 12, 0)
1310    >>> round_time(
1311    ...   datetime(2022, 1, 1, 12, 45, 57, 200),
1312    ...   datetime.timedelta(hours=1),
1313    ...   to = 'closest'
1314    ... )
1315    datetime.datetime(2022, 1, 1, 13, 0)
1316
1317    """
1318    from decimal import Decimal, ROUND_HALF_UP, ROUND_DOWN, ROUND_UP
1319    if date_delta is None:
1320        date_delta = timedelta(minutes=1)
1321
1322    if dt is None:
1323        dt = datetime.now(timezone.utc).replace(tzinfo=None)
1324
1325    def get_total_microseconds(td: timedelta) -> int:
1326        return (td.days * 86400 + td.seconds) * 1_000_000 + td.microseconds
1327
1328    round_to_microseconds = get_total_microseconds(date_delta)
1329    if round_to_microseconds == 0:
1330        return dt
1331
1332    dt_delta_from_min = dt.replace(tzinfo=None) - datetime.min
1333    dt_total_microseconds = get_total_microseconds(dt_delta_from_min)
1334
1335    dt_dec = Decimal(dt_total_microseconds)
1336    round_to_dec = Decimal(round_to_microseconds)
1337
1338    div = dt_dec / round_to_dec
1339    if to == 'down':
1340        num_intervals = div.to_integral_value(rounding=ROUND_DOWN)
1341    elif to == 'up':
1342        num_intervals = div.to_integral_value(rounding=ROUND_UP)
1343    else:
1344        num_intervals = div.to_integral_value(rounding=ROUND_HALF_UP)
1345
1346    rounded_dt_total_microseconds = num_intervals * round_to_dec
1347    adjustment_microseconds = int(rounded_dt_total_microseconds) - dt_total_microseconds
1348
1349    return dt + timedelta(microseconds=adjustment_microseconds)