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