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