ddp_utils.timeutils¶

import ddp_utils.timeutils

Provide pauses, date formatting, timing, throttling, and work schedules.

The module combines best-effort date parsing, monotonic elapsed-time helpers, a thread-safe sliding-window throttle, and optional timezone-aware business hours.

Examples

Measure one operation:

from ddp_utils.timeutils import Stopwatch

with Stopwatch("fetch") as stopwatch:
    fetch_data()
print(stopwatch.elapsed())

Block for a fixed or randomly selected number of seconds.

Parameters

Name	Type	Description
s	float \| None \| Tuple[float, float] \| str	Fixed seconds, `(minimum, maximum)` tuple, or `"min-max"` string. When provided, it supplies or replaces the explicit bounds.
min_s	float \| None	Lower random bound. Without `max_s`, the upper bound is five.
max_s	float \| None	Upper random bound. Without `min_s`, the lower bound is one.

Raises

Exception	Description
ValueError	`s` has an unsupported type or string syntax.

Examples

Pause for a human-like random interval:

sleep(min_s=0.5, max_s=1.5)

ddp_utils.timeutils.safe_sleep(seconds: float, *, return_on_event: bool = False, max_events: int | None = None, watchers: Iterable[str] | None = None, browser: Any | None = None, cancel_event: Any | None = None, require_browser: bool = False, poll_interval: float = 0.1) → list[Any]¶

Wait while safely delivering browser watcher callbacks.

An explicit browser wins over the browser activated by the nearest with BrowserFactory().open(config) context. Without either browser, the function behaves as an interruptible monotonic sleep unless require_browser is true. New threads do not inherit the active browser; pass browser=browser when calling this function from a new thread.

Parameters

Name	Type	Description
seconds	float	Maximum non-negative wait duration in seconds.
return_on_event	bool	Return after the first delivered watcher batch.
max_events	int \| None	Optional positive maximum number of delivered events.
watchers	Iterable[str] \| None	Optional watcher names or identifiers to deliver.
browser	Any \| None	Explicit browser facade, overriding the active context.
cancel_event	Any \| None	Optional object exposing `is_set()` and `wait()`.
require_browser	bool	Raise instead of falling back to ordinary sleep.
poll_interval	float	Positive maximum seconds between browser safe points.

Returns

Type	Description
list[Any]	Watcher events whose callbacks completed during the wait.

Raises

Exception	Description
ValueError	A duration is invalid or `max_events` is not positive.
RuntimeError	A usable browser is required but unavailable.
Exception	A watcher callback fails. Callback failures deliberately propagate to business logic.

Examples

Process watchers from the nearest browser context:

with BrowserFactory().open(config) as browser:
    safe_sleep(300)

Return as soon as any watcher callback completes:

events = safe_sleep(300, return_on_event=True)

Deliver at most three selected watcher events:

events = safe_sleep(
    300,
    max_events=3,
    watchers={"captcha", "confirmation"},
)

Supply a browser explicitly outside its context manager:

safe_sleep(300, browser=browser, require_browser=True)

Cancel the wait from another thread:

safe_sleep(300, cancel_event=stop_event)

ddp_utils.timeutils.now_time_stamp(stamp: str = '%H:%M:%S') → str¶

Format the current local date and time.

Parameters

Name	Type	Description
stamp	str	Format accepted by `datetime.strftime()`.

Returns

Type	Description
str	Formatted current local time.

Examples

Produce an hour-minute-second timestamp:

timestamp = now_time_stamp()

ddp_utils.timeutils.dateformat(date: str, to: str = '%Y%m%d', fuzzy: bool = False) → str¶

Parse a date string and format it as naive local date data.

Parameters

Name	Type	Description
date	str	Source date text. Surrounding whitespace is removed.
to	str	Target `datetime.strftime()` format.
fuzzy	bool	Allow unrelated text around recognized date components.

Returns

Type	Description
str	Formatted date, or `""` when parsing or input conversion fails.

Examples

Normalize a date for an identifier:

normalized = dateformat("January 5, 2026")
assert normalized == "20260105"

ddp_utils.timeutils.date_abc(date_str: str, fuzzy: bool = False) → tuple[str | None, str | None, str | None]¶

Heuristically extract year, month, and day strings from date text.

Parameters

Name	Type	Description
date_str	str	Source date text.
fuzzy	bool	Allow unrelated text around recognized date components.

Returns

Type	Description
tuple[str \| None, str \| None, str \| None]	`(year, month, day)` strings. Failed parsing yields three `None` values. The month is always returned after successful parsing; year and day presence are inferred from source-text shape.

Examples

Extract components from an ISO-like date:

year, month, day = date_abc("2026-10-03")

ddp_utils.timeutils.time_now() → float¶

Return the current Unix timestamp.

Returns

Type	Description
float	Seconds since the Unix epoch from `time.time()`.

Examples

Timestamp an event:

event_time = time_now()

ddp_utils.timeutils.datetime_now(format: str | None = None) → datetime | str¶

Return the current naive local datetime or a formatted string.

Parameters

Name	Type	Description
format	str \| None	Optional `datetime.strftime()` format.

Returns

Type	Description
datetime \| str	A `datetime` when `format` is `None`; otherwise formatted text.

Raises

Exception	Description
ValueError	The platform rejects the supplied format.

Examples

Produce an ISO-style local timestamp:

timestamp = datetime_now("%Y-%m-%dT%H:%M:%S")

ddp_utils.timeutils.datetime_now_utc() → datetime¶

Return the current timezone-aware UTC datetime.

Returns

Type	Description
datetime	Result of `datetime.now(timezone.utc)`.

Examples

Create an aware audit timestamp:

created_at = datetime_now_utc()

ddp_utils.timeutils.timer_start() → float¶

Return a monotonic start marker for timer_stop().

Returns

Type	Description
float	Current `timeit.default_timer()` value.

Examples

Start a lightweight elapsed-time measurement:

started = timer_start()

ddp_utils.timeutils.timer_stop(start: float | None = None, prefix: str = 'Running time', result: str = '{hh}:{mm}:{ss}') → str | None¶

Format whole elapsed seconds since a timer_start() marker.

Parameters

Name	Type	Description
start	float \| None	Marker returned by `timer_start()`; `None` disables output.
prefix	str	Text placed before the formatted duration.
result	str	Template supporting `{hh}`, `{mm}`, `{ss}`, `{d}`, `{m}`, and `{y}` tokens. Zero-valued components are omitted.

Returns

Type	Description
str \| None	Prefixed duration text, or `None` when no marker is supplied.

Examples

Format a completed measurement:

message = timer_stop(started, prefix="Fetch")

ddp_utils.timeutils.date_delta(delta: dict, from_date: str | datetime | None = None, fmt: str = '%Y-%m-%d %H:%M:%S', direction: str = 'future') → tuple[str, str]¶

Return formatted start and shifted dates for a calendar delta.

Parameters

Name	Type	Description
delta	dict	Numeric `seconds`, `minutes`, `hours`, `days`, `months`, and `years` values. Missing keys mean zero.
from_date	str \| datetime \| None	Starting datetime or parseable text. `None` and unparseable text fall back to the current local datetime.
fmt	str	Output `datetime.strftime()` format.
direction	str	`"past"`, `"back"`, `"prev"`, or `"-"` shifts backward; every other value shifts forward.

Returns

Type	Description
tuple[str, str]	Pair containing the formatted start and shifted datetime.

Examples

Calculate a two-month future range:

start, end = date_delta(
    {"months": 2},
    from_date="2026-01-31",
    fmt="%Y-%m-%d",
)

class ddp_utils.timeutils.Stopwatch(name: str = '')¶

Bases: object

Measure total monotonic time and named interval checkpoints.

Examples

Measure two pipeline stages:

with Stopwatch("pipeline") as stopwatch:
    fetch()
    stopwatch.lap("fetch")
    process()
    stopwatch.lap("process")
stopwatch.report()

Initialize an idle stopwatch with no recorded laps.

Parameters

Name	Type	Description
name	str	Optional title used by `report()`.

Examples

Name a pipeline measurement:

stopwatch = Stopwatch("pipeline")

start() → Stopwatch¶

Reset and start the stopwatch, clearing recorded laps.

Returns

Type	Description
Stopwatch	This stopwatch for fluent use.

Examples

Restart a reusable stopwatch:

stopwatch.start()

lap(label: str = '') → float¶

Record a checkpoint and return its interval duration.

Parameters

Name	Type	Description
label	str	Checkpoint name. An omitted label becomes `lapN`.

Returns

Type	Description
float	Monotonic seconds since the previous lap or start. Calling this on an idle stopwatch starts it first.

Examples

Record a parsing stage:

parsing_seconds = stopwatch.lap("parse")

stop() → float¶

Stop measurement and return total elapsed seconds.

Returns

Type	Description
float	Total duration. An unstarted stopwatch returns `0.0`.

Examples

Freeze a completed measurement:

total = stopwatch.stop()

elapsed() → float¶

Return total elapsed monotonic seconds.

Returns

Type	Description
float	Zero before start, live elapsed time while running, or the frozen duration after `stop()`.

Examples

Inspect a running measurement:

seconds = stopwatch.elapsed()

report(*, print_fn=<built-in function print>) → str¶

Render, emit, and return the checkpoint report.

Parameters

Name	Description
print_fn	Callable receiving the complete report string.

Returns

Type	Description
str	Text table containing every lap and total duration.

Examples

Capture the report instead of printing it:

messages = []
text = stopwatch.report(print_fn=messages.append)

ddp_utils.timeutils.humanize_duration(seconds: float) → str¶

Convert seconds to a compact human-readable duration.

Parameters

Name	Type	Description
seconds	float	Positive or negative duration.

Returns

Type	Description
str	Milliseconds below one second, one-decimal seconds below one minute, or integer day/hour/minute/second components for longer durations.

Examples

Format short and long durations:

assert humanize_duration(0.045) == "45ms"
assert humanize_duration(90) == "1m 30s"

class ddp_utils.timeutils.Throttle(calls: int = 10, period: float = 1.0)¶

Bases: object

Apply a thread-safe sliding-window call-rate limit.

The same instance shares timestamps across every decorated call and context manager entry.

Examples

Limit an API function to ten calls per minute:

@Throttle(calls=10, period=60)
def api_call():
    return request_data()

Initialize an empty sliding-window limiter.

Parameters

Name	Type	Description
calls	int	Maximum entries retained within one window.
period	float	Sliding-window length in seconds.

Examples

Allow five operations per second:

throttle = Throttle(calls=5, period=1.0)

class ddp_utils.timeutils.BusinessHours(start: int = 9, end: int = 18, workdays: tuple = (0, 1, 2, 3, 4), tz: str | None = None)¶

Bases: object

Evaluate and wait for recurring hour-based business schedules.

Weekdays follow datetime.weekday(), where Monday is zero. A named timezone is resolved through zoneinfo and then dateutil.tz; unresolved or omitted zones use UTC.

Examples

Define weekday business hours in Yerevan:

hours = BusinessHours(
    start=9,
    end=18,
    workdays=(0, 1, 2, 3, 4),
    tz="Asia/Yerevan",
)

Store the schedule and resolve its optional timezone.

Parameters

Name	Type	Description
start	int	Inclusive opening hour.
end	int	Exclusive closing hour.
workdays	tuple	Weekday numbers accepted as open days.
tz	str \| None	Optional IANA timezone name.

Examples

Use UTC defaults for weekday hours:

hours = BusinessHours(start=9, end=18)

is_open() → bool¶

Return whether the current schedule time is open.

Returns

Type	Description
bool	`True` on an allowed weekday from `start` inclusive to `end` exclusive.

Examples

Guard a business-only operation:

if hours.is_open():
    process_request()

next_open_time() → datetime¶

Return the nearest scheduled opening datetime.

Returns

Type	Description
datetime	Today’s opening when it is an allowed day and still before opening; otherwise the next allowed day’s opening, searched up to one week.

Examples

Schedule deferred work:

next_open = hours.next_open_time()

wait_for_open(check_interval: float = 60.0) → None¶

Block until the schedule becomes open.

Parameters

Name	Type	Description
check_interval	float	Maximum seconds per sleep while waiting. Shorter remaining intervals are used directly.

Examples

Wait with frequent cancellation opportunities in caller code:

hours.wait_for_open(check_interval=10)

seconds_until_open() → float¶

Return nonnegative seconds until the schedule opens.

Returns

Type	Description
float	`0.0` while open; otherwise the difference to `next_open_time()`.

Examples

Display a reopening countdown:

remaining = hours.seconds_until_open()

ddp_utils.timeutils.datetime_format_to_regex(date_format: str) → str | None¶

Convert supported datetime placeholders to an anchored regex string.

Parameters

Name	Type	Description
date_format	str	Format containing any of `%Y`, `%m`, `%d`, `%H`, `%M`, and `%S`.

Returns

Type	Description
str \| None	Regex text anchored with `^` and `$`. Non-placeholder characters are retained verbatim and are not regex-escaped.

Examples

Match an ISO-like calendar date:

pattern = datetime_format_to_regex("%Y-%m-%d")
assert pattern == r"^\d{4}-\d{2}-\d{2}$"