vayu.time_utils

time_utils

TZ module-attribute

TZ = ZoneInfo(getenv('TZ')) if getenv('TZ') else tzinfo

Default timezone used by module-level helpers.

Resolved once at import time from the TZ environment variable if set, otherwise from the system's local timezone.

TimeWindow dataclass

TimeWindow(
    start: Union[datetime, date, None] = None,
    end: Union[datetime, date, None] = None,
)

Bases: Interval

An Interval over datetimes/dates, with convenience constructors.

Inherits the full interval algebra (intersects, intersection, union, & / |, containment, shifting). Both endpoints must be the same type: both date or both datetime. For tz-aware datetimes, both must have tzinfo or both must be naive.

Attributes:

Name	Type	Description
start	Union[datetime, date, None]	Lower bound. Defaults to the Unix epoch when left unset.
end	Union[datetime, date, None]	Upper bound. Defaults to time_now() when left unset.

duration property

duration: timedelta

end - start as a timedelta.

start_ms property

start_ms: int

start as a Unix timestamp in milliseconds.

end_ms property

end_ms: int

end as a Unix timestamp in milliseconds.

ahead staticmethod

ahead(
    *,
    t: datetime = None,
    duration: timedelta = None,
    days=0,
    seconds=0,
    microseconds=0,
    milliseconds=0,
    minutes=0,
    hours=0,
    weeks=0
) -> TimeWindow

Return the window [t, t + duration] (looking forward from t).

Pass either duration directly or individual timedelta kwargs (hours=..., minutes=..., etc.). t defaults to time_now().

Source code in vayu/time_utils.py

@staticmethod
def ahead(
    *,
    t: dt.datetime = None,
    duration: dt.timedelta = None,
    days=0,
    seconds=0,
    microseconds=0,
    milliseconds=0,
    minutes=0,
    hours=0,
    weeks=0,
) -> "TimeWindow":
    """Return the window ``[t, t + duration]`` (looking forward from ``t``).

    Pass either ``duration`` directly or individual ``timedelta`` kwargs
    (``hours=...``, ``minutes=...``, etc.). ``t`` defaults to ``time_now()``.
    """
    if not duration:
        duration = dt.timedelta(
            days=days,
            seconds=seconds,
            microseconds=microseconds,
            milliseconds=milliseconds,
            minutes=minutes,
            hours=hours,
            weeks=weeks,
        )
    else:
        assert days == seconds == milliseconds == milliseconds == minutes == hours == weeks == 0
    t = t or time_now()
    return TimeWindow(t, t + duration)

behind staticmethod

behind(
    *,
    t: datetime = None,
    duration: timedelta = None,
    days=0,
    seconds=0,
    microseconds=0,
    milliseconds=0,
    minutes=0,
    hours=0,
    weeks=0
) -> TimeWindow

Return the window [t - duration, t] (looking back from t).

Pass either duration directly or individual timedelta kwargs. t defaults to time_now().

Source code in vayu/time_utils.py

@staticmethod
def behind(
    *,
    t: dt.datetime = None,
    duration: dt.timedelta = None,
    days=0,
    seconds=0,
    microseconds=0,
    milliseconds=0,
    minutes=0,
    hours=0,
    weeks=0,
) -> "TimeWindow":
    """Return the window ``[t - duration, t]`` (looking back from ``t``).

    Pass either ``duration`` directly or individual ``timedelta`` kwargs.
    ``t`` defaults to ``time_now()``.
    """
    if not duration:
        duration = dt.timedelta(
            days=days,
            seconds=seconds,
            microseconds=microseconds,
            milliseconds=milliseconds,
            minutes=minutes,
            hours=hours,
            weeks=weeks,
        )
    else:
        assert days == seconds == milliseconds == milliseconds == minutes == hours == weeks == 0

    t = t or time_now()
    return TimeWindow(t - duration, t)

around staticmethod

around(
    *,
    t: datetime = None,
    duration: timedelta = None,
    days=0,
    seconds=0,
    microseconds=0,
    milliseconds=0,
    minutes=0,
    hours=0,
    weeks=0
)

Return [t - duration, t + duration] (symmetric window around t).

Pass either duration directly or individual timedelta kwargs. t defaults to time_now().

Source code in vayu/time_utils.py

@staticmethod
def around(
    *,
    t: dt.datetime = None,
    duration: dt.timedelta = None,
    days=0,
    seconds=0,
    microseconds=0,
    milliseconds=0,
    minutes=0,
    hours=0,
    weeks=0,
):
    """Return ``[t - duration, t + duration]`` (symmetric window around ``t``).

    Pass either ``duration`` directly or individual ``timedelta`` kwargs.
    ``t`` defaults to ``time_now()``.
    """
    if not duration:
        duration = dt.timedelta(
            days=days,
            seconds=seconds,
            microseconds=microseconds,
            milliseconds=milliseconds,
            minutes=minutes,
            hours=hours,
            weeks=weeks,
        )
    else:
        assert days == seconds == milliseconds == milliseconds == minutes == hours == weeks == 0

    t = t or time_now()
    return TimeWindow(t - duration, t + duration)

from_date staticmethod

from_date(
    year: int, month: int, day: int, tz=None
) -> TimeWindow

Return the window spanning a full calendar day (midnight → end of day).

tz defaults to UTC.

Source code in vayu/time_utils.py

@staticmethod
def from_date(year: int, month: int, day: int, tz=None) -> "TimeWindow":
    """Return the window spanning a full calendar day (midnight → end of day).

    ``tz`` defaults to UTC.
    """
    tz = tz or ZoneInfo("UTC")
    date = dt.date(year, month, day)
    return TimeWindow(start=min_time(date, tz=tz), end=max_time(date, tz=tz))

from_timestamp staticmethod

from_timestamp(
    start_ts: [int, float], end_ts: [int, float], tz=None
) -> TimeWindow

Return the window built from two Unix timestamps.

Each timestamp may be seconds or milliseconds (see from_timestamp).

Source code in vayu/time_utils.py

@staticmethod
def from_timestamp(start_ts: [int, float], end_ts: [int, float], tz=None) -> "TimeWindow":
    """Return the window built from two Unix timestamps.

    Each timestamp may be seconds or milliseconds (see ``from_timestamp``).
    """
    start = from_timestamp(start_ts, tz=tz)
    end = from_timestamp(end_ts, tz=tz)
    return TimeWindow(start, end)

epoch_time

epoch_time() -> datetime

Return the Unix epoch (1970-01-01 00:00:00) as a tz-aware datetime.

Source code in vayu/time_utils.py

def epoch_time() -> dt.datetime:
    """Return the Unix epoch (1970-01-01 00:00:00) as a tz-aware datetime."""
    return from_timestamp(0)

time_now

time_now(
    with_ms: bool = False, local: bool = True
) -> datetime

Return the current time as a tz-aware datetime.

Parameters:

Name	Type	Description	Default
with_ms	bool	Keep microseconds. Default is to truncate to whole seconds.	False
local	bool	If True, use the module TZ. If False, return UTC.	True

Source code in vayu/time_utils.py

def time_now(with_ms: bool = False, local: bool = True) -> dt.datetime:
    """Return the current time as a tz-aware datetime.

    Args:
        with_ms: Keep microseconds. Default is to truncate to whole seconds.
        local: If True, use the module ``TZ``. If False, return UTC.
    """
    t = dt.datetime.now(tz=TZ if local else datetime.UTC)
    if not with_ms:
        t = t.replace(microsecond=0)

    return t

min_time

min_time(date: date, tz=None) -> datetime

Return the earliest datetime on the given date (00:00:00). UTC by default.

Source code in vayu/time_utils.py

def min_time(date: dt.date, tz=None) -> dt.datetime:
    """Return the earliest datetime on the given date (``00:00:00``). UTC by default."""
    return dt.datetime.combine(date, dt.time.min, tzinfo=tz or dt.UTC)

max_time

max_time(date: date, tz=None) -> datetime

Return the latest datetime on the given date (23:59:59.999999). UTC by default.

Source code in vayu/time_utils.py

def max_time(date: dt.date, tz=None) -> dt.datetime:
    """Return the latest datetime on the given date (``23:59:59.999999``). UTC by default."""
    return dt.datetime.combine(date, dt.time.max, tzinfo=tz or dt.UTC)

ts_ms

ts_ms(t: datetime) -> int

Return a Unix timestamp in milliseconds.

Source code in vayu/time_utils.py

def ts_ms(t: datetime.datetime) -> int:
    """Return a Unix timestamp in milliseconds."""
    return int(t.timestamp() * 1000)

utc_datetime

utc_datetime(
    year: int,
    month: int,
    day: int,
    hour: int = 0,
    minute: int = 0,
    second: int = 0,
    microsecond: int = 0,
) -> datetime

Construct a UTC datetime. Convenience wrapper over datetime(..., tzinfo=UTC).

Source code in vayu/time_utils.py

def utc_datetime(
    year: int,
    month: int,
    day: int,
    hour: int = 0,
    minute: int = 0,
    second: int = 0,
    microsecond: int = 0,
) -> dt.datetime:
    """Construct a UTC datetime. Convenience wrapper over ``datetime(..., tzinfo=UTC)``."""
    return dt.datetime(
        year=year,
        month=month,
        day=day,
        hour=hour,
        minute=minute,
        second=second,
        microsecond=microsecond,
        tzinfo=datetime.UTC,
    )

local_datetime

local_datetime(
    year: int,
    month: int,
    day: int,
    hour: int = 0,
    minute: int = 0,
    second: int = 0,
    microsecond: int = 0,
    tz=TZ,
) -> datetime

Construct a datetime in the module TZ (or the given tz).

Source code in vayu/time_utils.py

def local_datetime(
    year: int,
    month: int,
    day: int,
    hour: int = 0,
    minute: int = 0,
    second: int = 0,
    microsecond: int = 0,
    tz=TZ,
) -> datetime.datetime:
    """Construct a datetime in the module ``TZ`` (or the given ``tz``)."""
    t = datetime.datetime(
        year=year,
        month=month,
        day=day,
        hour=hour,
        minute=minute,
        second=second,
        microsecond=microsecond,
    )
    return t.replace(tzinfo=tz)

from_timestamp

from_timestamp(ts: float, tz=TZ) -> datetime

Convert a Unix timestamp to a tz-aware datetime.

Timestamps greater than 1e12 are treated as milliseconds (divided by 1000); smaller values are treated as seconds. This lets you pass either unit without worrying about the difference.

Source code in vayu/time_utils.py

def from_timestamp(ts: float, tz=TZ) -> dt.datetime:
    """Convert a Unix timestamp to a tz-aware datetime.

    Timestamps greater than ``1e12`` are treated as milliseconds (divided by 1000);
    smaller values are treated as seconds. This lets you pass either unit without
    worrying about the difference.
    """
    if ts > 1e12:
        # ts is in ms.
        ts = ts / 1000
    return dt.datetime.fromtimestamp(ts, tz=tz or dt.UTC)

from_z_string

from_z_string(t: str) -> datetime

Parse an ISO-8601 string with a trailing Z (Zulu / UTC) suffix.

Source code in vayu/time_utils.py

def from_z_string(t: str) -> dt.datetime:
    """Parse an ISO-8601 string with a trailing ``Z`` (Zulu / UTC) suffix."""
    return dt.datetime.fromisoformat(t.upper().replace("Z", "+00:00"))

timeit

timeit(f)

Decorator that prints how long the wrapped function took.

Useful for quick scripts. For library-level instrumentation, wire your own logging around the call instead — this helper uses print.

Source code in vayu/time_utils.py

def timeit(f):
    """Decorator that prints how long the wrapped function took.

    Useful for quick scripts. For library-level instrumentation, wire your own
    logging around the call instead — this helper uses ``print``.
    """

    @wraps(f)
    def timed(*args, **kw):

        t = -time.time()
        result = f(*args, **kw)
        t += time.time()

        print(f"`{f.__name__}` took: {round(t, 3)}s")
        return result

    return timed

to_human_readable_time

to_human_readable_time(t: Union[timedelta, float])

Format a duration as a compact human string (e.g. 1h2m3s).

Accepts a timedelta or a number of seconds. Returns "0s" for zero and "invalid-time" for negative values.

Source code in vayu/time_utils.py

def to_human_readable_time(t: Union[dt.timedelta, float]):
    """Format a duration as a compact human string (e.g. ``1h2m3s``).

    Accepts a ``timedelta`` or a number of seconds. Returns ``"0s"`` for zero
    and ``"invalid-time"`` for negative values.
    """
    if isinstance(t, dt.timedelta):
        t = int(t.total_seconds())
    if t < 0:
        return "invalid-time"

    h, remainder = divmod(t, 3600)
    m, s = divmod(remainder, 60)

    parts = []
    if h > 0:
        parts.append(f"{h}h")
    if m > 0:
        parts.append(f"{m}m")
    if s > 0:
        parts.append(f"{s}s")

    return "".join(parts) if parts else "0s"