schedule

eff.retry(schedule.exponential(0.1).up_to(times=5))
eff.repeat(schedule.spaced(1.0))

duration(duration) -> Schedule[Seconds]

def duration(duration: Duration) -> Schedule[Seconds]:
    """Recur once, after ``duration``. Outputs the duration, then 0."""
    delay = to_seconds(duration)
    return _stateless(lambda m: Continue(delay, delay) if m.attempt == 1 else Stop(ZERO))

during(duration) -> Schedule[Seconds]

def during(duration: Duration) -> Schedule[Seconds]:
    """Recur without waiting until more than ``duration`` has passed since
    the first step. Outputs the time elapsed."""
    limit = to_seconds(duration)

    def step(m: InputMetadata[object]) -> Decision[Seconds]:
        elapsed = Seconds(m.elapsed)
        return Stop(elapsed) if m.elapsed > limit else Continue(elapsed, ZERO)

    return _stateless(step)

exponential(base, factor=2.0) -> Schedule[Seconds]

def exponential(base: Duration, factor: float = 2.0) -> Schedule[Seconds]:
    """Recur forever, waiting ``base``, ``base*factor``, ``base*factor**2``,
    ... Outputs the delay."""
    first = _nanos(base)

    def step(m: InputMetadata[object]) -> Decision[Seconds]:
        delay = _seconds(first * factor ** (m.attempt - 1))
        return Continue(delay, delay)

    return _stateless(step)

fibonacci(one) -> Schedule[Seconds]

def fibonacci(one: Duration) -> Schedule[Seconds]:
    """Recur forever, waiting ``one``, ``2*one``, ``3*one``, ``5*one``, ...:
    each delay the sum of the two before. Outputs the delay."""
    unit = _nanos(one)

    def make() -> StepFn[Seconds, object, Any, Any]:
        a, b = 0, unit

        def step() -> Decision[Seconds]:
            nonlocal a, b
            a, b = b, a + b
            return Continue(_seconds(b), _seconds(b))

        return lambda _now, _input: _Sync(step)

    return from_step(_Sync(make))

fixed(interval) -> Schedule[int]

def fixed(interval: Duration) -> Schedule[int]:
    """Recur on a fixed beat of ``interval``, counted from the first step.
    A run that overruns the beat is followed at once; missed beats are not
    replayed. Outputs the count, from 0."""
    window = _nanos(interval)

    def make() -> Callable[[InputMetadata[object]], Effect[Decision[int]]]:
        # Effect's arithmetic, in whole nanoseconds since the first step.
        last_run: float = 0

        def step(m: InputMetadata[object]) -> Decision[int]:
            nonlocal last_run
            if window == 0:
                return Continue(m.attempt - 1, ZERO)
            if m.attempt == 1:
                last_run = window
                return Continue(0, _seconds(window))
            now = to_nanos(m.elapsed)
            running_behind = now > last_run + window
            delay = 0 if running_behind else _to_boundary(now, window)
            last_run = now + delay
            return Continue(m.attempt - 1, _seconds(delay))

        return lambda m: _Sync(lambda: step(m))

    return from_step_with_metadata(_Sync(make))

identity() -> Schedule[A, A]

def identity[A]() -> Schedule[A, A]:
    """Recur forever without waiting, outputting each input. Name the input
    type with an annotation: ``s: Schedule[str, str] = identity()``."""
    return _IDENTITY

max(first, *rest) -> Schedule[Seconds, I, E, R]

def max[I, E, R](
    first: Schedule[object, I, E, R], *rest: Schedule[object, I, E, R]
) -> Schedule[Seconds, I, E, R]:
    """Recur while every schedule recurs, waiting the longest of their
    delays. Outputs the delay."""

    def longest(delays: list[Seconds | None]) -> Seconds | None:
        return None if None in delays else builtins.max(d for d in delays if d is not None)

    return _combine((first, *rest), longest)

min(first, *rest) -> Schedule[Seconds, I, E, R]

def min[I, E, R](
    first: Schedule[object, I, E, R], *rest: Schedule[object, I, E, R]
) -> Schedule[Seconds, I, E, R]:
    """Recur while any schedule recurs, waiting the shortest delay of those
    that recur. Outputs the delay."""

    def shortest(delays: list[Seconds | None]) -> Seconds | None:
        live = [d for d in delays if d is not None]
        return builtins.min(live) if live else None

    return _combine((first, *rest), shortest)

recurs(times) -> Schedule[int]

def recurs(times: int) -> Schedule[int]:
    """Recur ``times`` times without waiting. Outputs the count, from 0."""
    if times < 0:
        raise ValueError(f"recurs needs times >= 0, got {times}")
    return forever.while_(lambda m: m.attempt <= times)

spaced(duration) -> Schedule[int]

def spaced(duration: Duration) -> Schedule[int]:
    """Recur forever, waiting ``duration`` each time. Outputs the count of
    recurrences so far, from 0."""
    delay = to_seconds(duration)
    return _stateless(lambda m: Continue(m.attempt - 1, delay))

windowed(interval) -> Schedule[int]

def windowed(interval: Duration) -> Schedule[int]:
    """Recur at the end of each ``interval``-long window, counted from the
    first step. Outputs the count, from 0."""
    window = _nanos(interval)

    def step(m: InputMetadata[object]) -> Decision[int]:
        delay = 0 if window == 0 else _to_boundary(to_nanos(m.elapsed), window)
        return Continue(m.attempt - 1, _seconds(delay))

    return _stateless(step)

cron(expression, tz=None) -> Schedule[Seconds, object, CronParseError]

def cron(
    expression: str | Cron, tz: tzinfo | str | None = None
) -> Schedule[Seconds, object, CronParseError]:
    """Recur at each time matching ``expression``, waiting until it. Outputs
    the delay. A string is parsed with ``cron.parse(expression, tz)`` when
    the schedule runs, failing it with ``CronParseError`` if invalid; a
    ``Cron`` brings its own zone. A cron without a zone runs at the local
    offset of the current time."""
    parsed = Ok(expression) if isinstance(expression, Cron) else cron_module.parse(expression, tz)

    def step_for(c: Cron):
        def step(now: float, _input: object) -> Effect[Decision[Seconds]]:
            if math.isinf(now):
                return _Succeed(Stop(ZERO))

            def decide() -> Decision[Seconds]:
                start = datetime.fromtimestamp(now, UTC).astimezone()
                delay = Seconds(_since(cron_module.next(c, start).timestamp(), now))
                return Continue(delay, delay)

            return _Sync(decide)

        return step

    return from_step(from_result(parsed).map(step_for))

from_step(make_step) -> Schedule[Out, In, E, R]

def from_step[Out, In, E = Never, R = Never](
    make_step: Effect[StepFn[Out, In, E, R], E, R],
) -> Schedule[Out, In, E, R]:
    """A schedule from an effect that makes its step function. The effect
    runs once per run of the schedule, so the step can keep state."""
    require_effect(make_step, "make_step")
    return Schedule(make_step)

from_step_with_metadata(make_step) -> Schedule[Out, In, E, R]

def from_step_with_metadata[Out, In, E = Never, R = Never](
    make_step: Effect[Callable[[InputMetadata[In]], Effect[Decision[Out], E, R]], E, R],
) -> Schedule[Out, In, E, R]:
    """``from_step``, but the step gets ``InputMetadata`` (attempt, start,
    elapsed time) in place of ``(now, input)``."""
    require_effect(make_step, "make_step")

    def with_counter(
        f: Callable[[InputMetadata[In]], Effect[Decision[Out], E, R]],
    ) -> StepFn[Out, In, E, R]:
        meta = _MetadataCounter()
        return lambda now, input: f(meta(now, input))

    return Schedule(make_step.map(with_counter))

set_input_type(input_type) -> SetInputType[I]

schedule.recurs(3).pipe(set_input_type(int)).passthrough()  # Schedule[int, int]

def set_input_type[I](input_type: type[I], /) -> SetInputType[I]:
    """Type a schedule to take ``input_type`` inputs, so the metadata its
    combinators see carries that type::

        schedule.recurs(3).pipe(set_input_type(int)).passthrough()  # Schedule[int, int]

    Curried for ``.pipe``: as a method, pyright would fix the input type
    from the schedule before seeing ``input_type``."""
    return SetInputType()

to_step(schedule) -> Effect[StepFn[Out, In, E, R], Never, R]

def to_step[Out, In, E, R](
    schedule: Schedule[Out, In, E, R],
) -> Effect[StepFn[Out, In, E, R], Never, R]:
    """Make a fresh step function. If making it fails, every call of the
    step fails with that cause."""

    def failing(cause: Any) -> Effect[StepFn[Out, In, E, R]]:
        return _Succeed(lambda _now, _input: _FailCause(cause))

    return schedule._make_step.catch_cause(failing)

to_step_with_metadata(schedule) -> Effect[Callable[[In], Effect[Metadata[Out, In] | Stop[Out], E, R]], Never, R]

def to_step_with_metadata[Out, In, E, R](
    schedule: Schedule[Out, In, E, R],
) -> Effect[Callable[[In], Effect[Metadata[Out, In] | Stop[Out], E, R]], Never, R]:
    """A step that reads the clock, decides, and sleeps the delay before
    returning the step's ``Metadata``; or returns ``Stop`` without sleeping."""

    def make() -> Any:
        step = yield from to_step(schedule)
        clock = yield from Clock()
        meta = _MetadataCounter()

        def run(input: In) -> Effect[Metadata[Out, In] | Stop[Out], E, R]:
            def body() -> Any:
                now = clock.unsafe_now()
                decision = yield from step(now, input)
                if isinstance(decision, Stop):
                    return decision
                m = meta.full(now, input, decision.output, decision.delay)
                yield from sleep(decision.delay)
                return m

            return _Gen(body)

        return run

    return _Gen(make)

to_step_with_sleep(schedule) -> Effect[Callable[[In], Effect[Decision[Out], E, R]], Never, R]

def to_step_with_sleep[Out, In, E, R](
    schedule: Schedule[Out, In, E, R],
) -> Effect[Callable[[In], Effect[Decision[Out], E, R]], Never, R]:
    """``to_step_with_metadata``, returning ``Continue(output, delay)``
    after the sleep in place of the metadata."""

    def keep_decision(m: Metadata[Out, In] | Stop[Out]) -> Decision[Out]:
        return m if isinstance(m, Stop) else Continue(m.output, m.duration)

    return to_step_with_metadata(schedule).map(
        lambda step: lambda input: step(input).map(keep_decision)
    )