Skip to content

hyperion.dateutils

hyperion.dateutils

truncate_datetime 

truncate_datetime(base, unit)

Truncate datetime to the specified unit (set all smaller units to zero).

Source code in hyperion/dateutils.py 
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
def truncate_datetime(base: datetime.datetime | datetime.date, unit: TimeResolutionUnit) -> datetime.datetime:
    """Truncate datetime to the specified unit (set all smaller units to zero)."""
    if not isinstance(base, datetime.datetime):
        base = datetime.datetime(base.year, base.month, base.day, tzinfo=datetime.UTC)
    match unit:
        case "s":
            return base.replace(microsecond=0)
        case "m":
            return base.replace(second=0, microsecond=0)
        case "h":
            return base.replace(minute=0, second=0, microsecond=0)
        case "d":
            return base.replace(hour=0, minute=0, second=0, microsecond=0)
        case "w":
            return base.replace(hour=0, minute=0, second=0, microsecond=0) - datetime.timedelta(days=base.weekday())
        case "M":
            return base.replace(day=1, hour=0, minute=0, second=0, microsecond=0)
        case "y":
            return base.replace(month=1, day=1, hour=0, minute=0, second=0, microsecond=0)
    raise ValueError(f"Unknown time unit {unit!r}. Pick one of {', '.join(TIME_UNITS)}")

iter_dates_between 

iter_dates_between(start_date, end_date, granularity)

Iterate over datetimes between start_date and end_date with steps based on the given granularity. Includes the start_date and may include end_date (if end date is reachable from start date with given granularity).

:param start_date: The starting datetime. :param end_date: The ending datetime. :param granularity: The granularity for steps (e.g., "d" for days, "M" for months). :return: An iterator of datetime objects.

Source code in hyperion/dateutils.py 
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
def iter_dates_between(
    start_date: datetime.datetime | datetime.date,
    end_date: datetime.datetime | datetime.date,
    granularity: TimeResolutionUnit,
) -> Iterator[datetime.datetime]:
    """
    Iterate over datetimes between start_date and end_date with steps based on the given granularity.
    Includes the start_date and may include end_date (if end date is reachable from start date with given granularity).

    :param start_date: The starting datetime.
    :param end_date: The ending datetime.
    :param granularity: The granularity for steps (e.g., "d" for days, "M" for months).
    :return: An iterator of datetime objects.
    """
    start_date = assure_timezone(start_date)
    end_date = assure_timezone(end_date)
    logger.debug(
        "Generating dates between two points.", start_date=start_date.isoformat(), end_date=end_date.isoformat()
    )
    if start_date > end_date:
        raise ValueError("Start date cannot be later than end date.")

    current = start_date

    delta: datetime.timedelta | relativedelta

    match granularity:
        case "s":
            delta = datetime.timedelta(seconds=1)
        case "m":
            delta = datetime.timedelta(minutes=1)
        case "h":
            delta = datetime.timedelta(hours=1)
        case "d":
            delta = datetime.timedelta(days=1)
        case "w":
            delta = datetime.timedelta(weeks=1)
        case "M":
            delta = relativedelta(months=1)
        case "y":
            delta = relativedelta(years=1)
        case default:
            raise ValueError(f"Unsupported granularity {default!r}.")

    while current <= end_date:
        yield current
        current += delta

quantize_datetime 

quantize_datetime(base, resolution)

Quantize a datetime to the next interval based on the specified resolution.

This function aligns a given datetime to the next moment in time defined by the resolution. The resolution is expressed as a unit (seconds, minutes, hours, or days) and a value (e.g., 5 seconds, 15 minutes).

Important Notes: - The resolution is always calculated relative to the higher unit, which may lead to overlapping intervals for non-standard values. For example: - A 7-second resolution could result in intervals ending at 12:50:56 and 12:51:03, with another interval starting at 12:51:00, causing overlaps. - To avoid such overlaps, it is recommended to use resolutions that are divisors of the higher unit (e.g., 60 for seconds and minutes).

Parameters: - base (datetime.datetime): The datetime to quantize. - resolution (TimeResolution | str): The resolution for quantization. If a string is provided, it should follow the format "{value}{unit}" (e.g., "5s", "15m", "2h").

Returns: - datetime.datetime: The quantized datetime.

Raises: - ValueError: If the resolution unit is unsupported.

Source code in hyperion/dateutils.py 
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
def quantize_datetime(base: datetime.datetime, resolution: TimeResolution | str) -> datetime.datetime:
    """
    Quantize a datetime to the next interval based on the specified resolution.

    This function aligns a given datetime to the next moment in time defined
    by the resolution. The resolution is expressed as a unit (seconds, minutes,
    hours, or days) and a value (e.g., 5 seconds, 15 minutes).

    **Important Notes:**
    - The resolution is always calculated relative to the higher unit, which may lead
      to overlapping intervals for non-standard values. For example:
        - A 7-second resolution could result in intervals ending at 12:50:56 and 12:51:03,
          with another interval starting at 12:51:00, causing overlaps.
    - To avoid such overlaps, it is recommended to use resolutions that are divisors of
      the higher unit (e.g., 60 for seconds and minutes).

    Parameters:
    - base (datetime.datetime): The datetime to quantize.
    - resolution (TimeResolution | str): The resolution for quantization.
      If a string is provided, it should follow the format "{value}{unit}"
      (e.g., "5s", "15m", "2h").

    Returns:
    - datetime.datetime: The quantized datetime.

    Raises:
    - ValueError: If the resolution unit is unsupported.
    """
    resolution = resolution if isinstance(resolution, TimeResolution) else TimeResolution.from_str(resolution)
    base_truncated = truncate_datetime(base, resolution.unit)

    def _get_shift(value: int) -> int:
        return resolution.value - (value % resolution.value)

    match resolution.unit:
        case "s":
            seconds_shift = _get_shift(base.second)
            return base_truncated + datetime.timedelta(seconds=seconds_shift)
        case "m":
            minutes_shift = _get_shift(base.minute)
            return base_truncated + datetime.timedelta(minutes=minutes_shift)
        case "h":
            hours_shift = _get_shift(base.hour)
            return base_truncated + datetime.timedelta(hours=hours_shift)
        case "d":
            days_shift = _get_shift(base.day)
            return base_truncated + datetime.timedelta(days=days_shift)
        case "w":
            days_shift = resolution.value * 7 - (base.weekday() % resolution.value)
            return base_truncated + datetime.timedelta(days=days_shift)
        case "M":
            months_shift = _get_shift(base.month)
            return base_truncated + relativedelta(months=months_shift)
        case "y":
            years_shift = _get_shift(base.year)
            return base_truncated + relativedelta(years=years_shift)
    raise ValueError(f"Unsupported resolution unit {resolution.unit!r} for quantization.")  # pragma: no cover

assure_timezone 

assure_timezone(base, tz=UTC)

Assure datetime has a datetime and return it timezone-aware if not.

Source code in hyperion/dateutils.py 
191
192
193
194
195
196
197
198
199
200
201
202
def assure_timezone(
    base: datetime.datetime | datetime.date, tz: datetime.timezone = datetime.UTC
) -> datetime.datetime:
    """Assure datetime has a datetime and return it timezone-aware if not."""
    if not isinstance(base, datetime.datetime):
        return datetime.datetime(base.year, base.month, base.day, tzinfo=tz)
    if base.tzinfo is not None:
        if base.tzinfo == tz:
            return base
        return base.astimezone(tz)
    logger.warning(f"A timezone-unaware timestamp was given, assuming {tz!r}.")
    return base.replace(tzinfo=tz)

get_date_pattern 

get_date_pattern(date, unit)

Get a date pattern string up to the specified unit level.

Examples:

>>> get_date_pattern(datetime.datetime(2025, 1, 12, 12), "h")
"2025-01-12T12"
>>> get_date_pattern(datetime.datetime(2025, 1, 12, 12), "d")
"2025-01-12"
>>> get_date_pattern(datetime.datetime(2025, 1, 12, 12), "M")
"2025-01"
>>> get_date_pattern(datetime.datetime(2025, 1, 12, 12), "y")
"2025"
Source code in hyperion/dateutils.py 
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
def get_date_pattern(date: datetime.datetime, unit: TimeResolutionUnit) -> str:
    """Get a date pattern string up to the specified unit level.

    Examples:
        >>> get_date_pattern(datetime.datetime(2025, 1, 12, 12), "h")
        "2025-01-12T12"
        >>> get_date_pattern(datetime.datetime(2025, 1, 12, 12), "d")
        "2025-01-12"
        >>> get_date_pattern(datetime.datetime(2025, 1, 12, 12), "M")
        "2025-01"
        >>> get_date_pattern(datetime.datetime(2025, 1, 12, 12), "y")
        "2025"
    """
    truncated = truncate_datetime(date, unit)
    match unit:
        case "s":
            return truncated.strftime("%Y-%m-%dT%H:%M:%S")
        case "m":
            return truncated.strftime("%Y-%m-%dT%H:%M")
        case "h":
            return truncated.strftime("%Y-%m-%dT%H")
        case "d":
            return truncated.strftime("%Y-%m-%d")
        case "w":
            return truncated.strftime("%Y-%m-%d")  # Week is special, keep the day
        case "M":
            return truncated.strftime("%Y-%m")
        case "y":
            return truncated.strftime("%Y")
    raise ValueError(f"Unsupported time unit {unit!r}.")

iter_intervals 

iter_intervals(start_date, end_date, granularity)

Iter tuples of (interval start, interval end).

:param start_date: The starting datetime. :param end_date: The ending datetime. :param granularity: The granularity for steps (e.g., "d" for days, "M" for months). :return: An iterator of intervals (tuples of (start, end)).

Source code in hyperion/dateutils.py 
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
def iter_intervals(
    start_date: datetime.datetime, end_date: datetime.datetime, granularity: TimeResolutionUnit
) -> Iterable[tuple[datetime.datetime, datetime.datetime]]:
    """Iter tuples of (interval start, interval end).

    :param start_date: The starting datetime.
    :param end_date: The ending datetime.
    :param granularity: The granularity for steps (e.g., "d" for days, "M" for months).
    :return: An iterator of intervals (tuples of (start, end)).
    """
    interval_starts = list(iter_dates_between(start_date, end_date, granularity))
    for i, interval_start in enumerate(interval_starts):
        interval_end = interval_starts[i + 1] if i + 1 < len(interval_starts) else end_date
        if interval_start == interval_end:
            continue
        yield interval_start, interval_end