GitHub

clock

PreviousNext

API reference for clock.

Time. Clock is a reference, so effects can sleep without declaring it, and tests can swap in pyfect.testing.TestClock.

Wall time (now) can jump, backwards too; monotonic time never does, so measure elapsed time with it. Each comes in float seconds and in whole nanoseconds (Effect's currentTimeNanos / monotonicTimeNanos).

Classes

ClockServiceLiveClockClock

Functions

sleep

sleep(duration) -> Effect[None]
ParameterTypeDescription
durationDuration

Returns Effect[None]

def sleep(duration: Duration) -> Effect[None]:
    seconds = to_seconds(duration)
    return Clock().flat_map(lambda c: c.sleep(seconds))

now

now() -> Effect[float]

Wall-clock time, seconds since the epoch.

Returns Effect[float]

def now() -> Effect[float]:
    """Wall-clock time, seconds since the epoch."""
    return Clock().map(lambda c: c.unsafe_now())

monotonic

monotonic() -> Effect[float]

Monotonic time in seconds.

Returns Effect[float]

def monotonic() -> Effect[float]:
    """Monotonic time in seconds."""
    return Clock().map(lambda c: c.unsafe_monotonic())

now_nanos

now_nanos() -> Effect[int]

Wall-clock time, whole nanoseconds since the epoch.

Returns Effect[int]

def now_nanos() -> Effect[int]:
    """Wall-clock time, whole nanoseconds since the epoch."""
    return Clock().map(lambda c: c.unsafe_now_nanos())

monotonic_nanos

monotonic_nanos() -> Effect[int]

Monotonic time in whole nanoseconds.

Returns Effect[int]

def monotonic_nanos() -> Effect[int]:
    """Monotonic time in whole nanoseconds."""
    return Clock().map(lambda c: c.unsafe_monotonic_nanos())

clock_with

clock_with(f) -> Effect[A, E, R]

Run f with the current clock.

ParameterTypeDescription
fCallable[[ClockService], Effect[A, E, R]]

Returns Effect[A, E, R]

def clock_with[A, E = Never, R = Never](
    f: Callable[[ClockService], Effect[A, E, R]],
) -> Effect[A, E, R]:
    """Run ``f`` with the current clock."""
    return Clock().flat_map(f)

timed

timed(eff) -> Effect[tuple[float, A], E, R]

Pair eff's value with how long it took in seconds, measured in whole monotonic nanoseconds.

ParameterTypeDescription
effEffect[A, E, R]

Returns Effect[tuple[float, A], E, R]

def timed[A, E, R](eff: Effect[A, E, R]) -> Effect[tuple[float, A], E, R]:
    """Pair ``eff``'s value with how long it took in seconds, measured in
    whole monotonic nanoseconds."""
    require_effect(eff, "eff")

    def body():
        start = yield from monotonic_nanos()
        value = yield from eff
        end = yield from monotonic_nanos()
        return ((end - start) / _NANOS, value)

    return _Gen(body)