Schedule

__init__(make_step) -> None

def __init__(
    self, make_step: Effect[StepFn[Out_co, In_contra, E_co, R_co], E_co, R_co]
) -> None:
    self._make_step = make_step

map(f) -> Schedule[B, I, E_co, R_co]

def map[B, I](
    self: Schedule[Out_co, I, E_co, R_co], f: Callable[[Metadata[Out_co, I]], B]
) -> Schedule[B, I, E_co, R_co]:
    """Replace each output (including the final one) with ``f(metadata)``."""
    from pyfect.schedule._combinators import map_effect

    return map_effect(self, lambda m: _Succeed(f(m)))

map_effect(f) -> Schedule[B, I, E_co | E2, R_co | R2]

def map_effect[B, I, E2 = Never, R2 = Never](
    self: Schedule[Out_co, I, E_co, R_co], f: Callable[[Metadata[Out_co, I]], Effect[B, E2, R2]]
) -> Schedule[B, I, E_co | E2, R_co | R2]:
    """``map`` with an effectful ``f``."""
    from pyfect.schedule._combinators import map_effect

    return map_effect(self, f)

tap(f) -> Schedule[Out_co, I, E_co | E2, R_co | R2]

def tap[I, E2 = Never, R2 = Never](
    self: Schedule[Out_co, I, E_co, R_co],
    f: Callable[[Metadata[Out_co, I]], Effect[object, E2, R2]],
) -> Schedule[Out_co, I, E_co | E2, R_co | R2]:
    """Run ``f(metadata)`` after each decision to recur."""
    from pyfect.schedule._combinators import tap

    return tap(self, f)

while_(predicate) -> Schedule[Out_co, I, E_co, R_co]

def while_[I](
    self: Schedule[Out_co, I, E_co, R_co], predicate: Callable[[Metadata[Out_co, I]], bool]
) -> Schedule[Out_co, I, E_co, R_co]:
    """Recur only while ``predicate(metadata)`` holds (Effect's
    ``Schedule.while``); stop with the step's output once it fails."""
    from pyfect.schedule._combinators import while_effect

    return while_effect(self, lambda m: _Succeed(predicate(m)))

while_effect(predicate) -> Schedule[Out_co, I, E_co | E2, R_co | R2]

def while_effect[I, E2 = Never, R2 = Never](
    self: Schedule[Out_co, I, E_co, R_co],
    predicate: Callable[[Metadata[Out_co, I]], Effect[bool, E2, R2]],
) -> Schedule[Out_co, I, E_co | E2, R_co | R2]:
    """``while_`` with an effectful predicate."""
    from pyfect.schedule._combinators import while_effect

    return while_effect(self, predicate)

modify_delay(f) -> Schedule[Out_co, I, E_co | E2, R_co | R2]

def modify_delay[I, E2 = Never, R2 = Never](
    self: Schedule[Out_co, I, E_co, R_co],
    f: Callable[[Metadata[Out_co, I]], Effect[Duration, E2, R2]],
) -> Schedule[Out_co, I, E_co | E2, R_co | R2]:
    """Replace each delay with the one ``f(metadata)`` gives."""
    from pyfect.schedule._combinators import modify_delay

    return modify_delay(self, f)

add_delay(f) -> Schedule[Out_co, I, E_co | E2, R_co | R2]

def add_delay[I, E2 = Never, R2 = Never](
    self: Schedule[Out_co, I, E_co, R_co],
    f: Callable[[Metadata[Out_co, I]], Effect[Duration, E2, R2]],
) -> Schedule[Out_co, I, E_co | E2, R_co | R2]:
    """Add the delay ``f(metadata)`` gives to each delay."""
    from pyfect.schedule._combinators import modify_delay

    return modify_delay(
        self, lambda m: f(m).map(lambda extra: Seconds(to_seconds(extra) + m.duration))
    )

jittered() -> Schedule[Out_co, In_contra, E_co, R_co]

def jittered(self) -> Schedule[Out_co, In_contra, E_co, R_co]:
    """Scale each delay by a random factor between 0.8 and 1.2."""
    from pyfect.schedule._combinators import jittered

    return jittered(self)

pipe(f) -> B

def pipe[B](self, f: Callable[[Self], B], /) -> B:
    """Apply ``f`` to this schedule (one step per call, as ``Effect.pipe``)."""
    return f(self)

passthrough() -> Schedule[I, I, E_co, R_co]

def passthrough[I](self: Schedule[Out_co, I, E_co, R_co]) -> Schedule[I, I, E_co, R_co]:
    """Keep the decisions, but output each input instead."""
    from pyfect.schedule._combinators import passthrough

    return passthrough(self)

up_to(*, times=None, duration=None) -> Schedule[Out_co, In_contra, E_co, R_co]

def up_to(
    self, *, times: int | None = None, duration: Duration | None = None
) -> Schedule[Out_co, In_contra, E_co, R_co]:
    """Stop after ``times`` recurrences, or once more than ``duration``
    has passed since the first step, whichever comes first."""
    if times is not None and times < 0:
        raise ValueError(f"up_to needs times >= 0, got {times}")
    limit = None if duration is None else to_seconds(duration)

    def within(m: Metadata[Any, Any]) -> bool:
        return (times is None or m.attempt <= times) and (limit is None or m.elapsed <= limit)

    return self.while_(within)

concat_result(other) -> Schedule[Result[O2, Out_co], I, EE, RR]

def concat_result[O2, I, EE, RR](
    self: Schedule[Out_co, I, EE, RR], other: Schedule[O2, I, EE, RR]
) -> Schedule[Result[O2, Out_co], I, EE, RR]:
    """Run this schedule until it stops, then ``other``. Outputs are
    ``Err(output)`` from this schedule and ``Ok(output)`` from ``other``."""
    from pyfect.schedule._combinators import concat_result

    return concat_result(self, other)

concat(other) -> Schedule[Out_co | O2, I, EE, RR]

def concat[O2, I, EE, RR](
    self: Schedule[Out_co, I, EE, RR], other: Schedule[O2, I, EE, RR]
) -> Schedule[Out_co | O2, I, EE, RR]:
    """Run this schedule until it stops, then ``other``."""
    from pyfect.result import Ok

    return self.concat_result(other).map(
        lambda m: m.output.value if isinstance(m.output, Ok) else m.output.error
    )