core.time¶

def ts_utc_aware_get(ts: datetime) -> datetime:
    """Normalize *ts* to a timezone-aware UTC datetime.

    This is the canonical timestamp normalizer for all domain models.

    - Naive datetimes are assumed to represent UTC and tagged with
      ``timezone.utc``.
    - Aware non-UTC datetimes are converted to UTC.
    - Aware UTC datetimes are returned unchanged.
    """
    if ts.tzinfo is None:
        return ts.replace(tzinfo=timezone.utc)
    if ts.utcoffset() != timedelta(0):
        return ts.astimezone(timezone.utc)
    return ts

def to_naive_utc(ts: datetime) -> datetime:
    """Normalize a datetime to naive UTC.

    Aware datetimes are converted to UTC then stripped of tzinfo.
    Naive datetimes are returned as-is (assumed to already represent UTC).

    .. deprecated::
        Use :func:`ts_utc_aware_get` instead.  This helper exists only
        for transitional callers that still expect naive-UTC values.
    """
    if ts.tzinfo is not None:
        return ts.astimezone(timezone.utc).replace(tzinfo=None)
    return ts

def align_to_bucket(
    ts: datetime,
    bucket_seconds: int = DEFAULT_BUCKET_SECONDS,
) -> datetime:
    """Floor *ts* to the nearest bucket boundary.

    If *ts* is timezone-aware it is first converted to UTC.  Naive
    datetimes are assumed to represent UTC.  The returned datetime is
    always a **timezone-aware** UTC value (``tzinfo=timezone.utc``).

    Args:
        ts: Timestamp to align.
        bucket_seconds: Bucket width in seconds (default 60).

    Returns:
        A timezone-aware UTC datetime whose POSIX epoch is a multiple of
        *bucket_seconds*.
    """
    if ts.tzinfo is not None:
        ts = ts.astimezone(timezone.utc).replace(tzinfo=None)

    epoch = int(ts.replace(tzinfo=timezone.utc).timestamp())
    aligned_epoch = (epoch // bucket_seconds) * bucket_seconds
    return datetime.fromtimestamp(aligned_epoch, tz=timezone.utc)

def generate_bucket_range(
    start: datetime,
    end: datetime,
    bucket_seconds: int = DEFAULT_BUCKET_SECONDS,
) -> list[datetime]:
    """Enumerate bucket-start timestamps from *start* to *end* (exclusive).

    Both *start* and *end* are aligned first, so callers need not pre-align.

    Args:
        start: Earliest timestamp (inclusive after alignment).
        end: Latest timestamp (exclusive after alignment).
        bucket_seconds: Bucket width in seconds (default 60).

    Returns:
        Sorted list of timezone-aware UTC bucket-start datetimes.
    """
    cur = align_to_bucket(start, bucket_seconds)
    end_aligned = align_to_bucket(end, bucket_seconds)
    step = timedelta(seconds=bucket_seconds)

    buckets: list[datetime] = []
    while cur < end_aligned:
        buckets.append(cur)
        cur += step
    return buckets