Stream

__init__(channel) -> None

def __init__(self, channel: StreamChannel[A_co, E_co, R_co]) -> None:
    self._channel = channel

pipe(f) -> B

def pipe[B](self, f: Callable[[Self], B], /) -> B:
    """Apply ``f`` to this stream (one step; see ``Effect.pipe``)."""
    return f(self)

pipe_through_channel(channel) -> Stream[B, E2, R_co | R2]

def pipe_through_channel[B, E2 = Never, R2 = Never](
    self, channel: Channel[tuple[B, ...], E2, object, tuple[A_co, ...], E_co, None, R2]
) -> Stream[B, E2, R_co | R2]:
    """The stream of what ``channel`` writes when this stream's chunks
    are its input. The channel sees this stream's failure (and decides
    what to do with it); its end value is dropped."""
    from pyfect.stream import _text

    return _text.pipe_through_channel(self, channel)

pipe_through_channel_or_fail(channel) -> Stream[B, E_co | E2, R_co | R2]

def pipe_through_channel_or_fail[B, E2 = Never, R2 = Never](
    self, channel: Channel[tuple[B, ...], E2, object, tuple[A_co, ...], Never, None, R2]
) -> Stream[B, E_co | E2, R_co | R2]:
    """``pipe_through_channel`` for a channel that cannot see this
    stream's failures: they pass through it untouched and fail the
    stream."""
    from pyfect.stream import _text

    return _text.pipe_through_channel_or_fail(self, channel)

decode_text(encoding='utf-8') -> Stream[str, E, R]

def decode_text[E, R](
    self: ChannelMethods[bytes, E, R], encoding: str = "utf-8"
) -> Stream[str, E, R]:
    """Decode byte strings as text in ``encoding``, keeping a character
    split across byte strings whole; bad bytes become U+FFFD and a UTF-8
    byte-order mark is dropped (``channel.decode_text``)."""
    from pyfect.stream import _text

    return _text.decode_text(self, encoding)

encode_text() -> Stream[bytes, E, R]

def encode_text[E, R](self: ChannelMethods[str, E, R]) -> Stream[bytes, E, R]:
    """Encode each string as UTF-8."""
    from pyfect.stream import _text

    return _text.encode_text(self)

split_lines() -> Stream[str, E, R]

def split_lines[E, R](self: ChannelMethods[str, E, R]) -> Stream[str, E, R]:
    """Split text into lines at ``\\n``, ``\\r\\n`` and ``\\r``, wherever
    the chunks break, without the line breaks. The text after the last
    break is the last line; a trailing break adds no empty line."""
    from pyfect.stream import _text

    return _text.split_lines(self)

run_into_queue(queue) -> Effect[None, Never, R_co]

def run_into_queue(self, queue: Enqueue[A_co, E_co | Done]) -> Effect[None, Never, R_co]:
    """Offer every item to ``queue``, then end it, or fail it with the
    stream's cause. The queue is finished even if the run is
    interrupted."""
    return self._channel.run_into_queue_array(queue)

to_queue(*, capacity, strategy='suspend') -> Effect[Dequeue[A_co, E_co | Done], Never, R_co | Scope]

def to_queue(
    self, *, capacity: Capacity, strategy: Strategy = "suspend"
) -> Effect[Dequeue[A_co, E_co | Done], Never, R_co | Scope]:
    """Run the stream on a fiber in the current ``Scope``, offering its
    items to a new queue of ``capacity``; the queue ends (or fails) as
    the stream does and shuts down with the scope."""
    if capacity == "unbounded":
        return self._channel.to_queue_array(capacity=capacity)
    return self._channel.to_queue_array(capacity=capacity, strategy=strategy)

run_into_pubsub(pubsub, *, shutdown_on_end=False) -> Effect[None, E_co, R_co]

def run_into_pubsub(
    self, pubsub: PubSub[A_co], *, shutdown_on_end: bool = False
) -> Effect[None, E_co, R_co]:
    """Publish every item to ``pubsub``; with ``shutdown_on_end``, shut it
    down when the run ends."""
    return self._channel.run_into_pubsub_array(pubsub, shutdown_on_end=shutdown_on_end)

to_pubsub(*, capacity, strategy='suspend', replay=None, shutdown_on_end=True) -> Effect[PubSub[A_co], Never, R_co | Scope]

def to_pubsub(
    self,
    *,
    capacity: Capacity,
    strategy: Strategy = "suspend",
    replay: int | None = None,
    shutdown_on_end: bool = True,
) -> Effect[PubSub[A_co], Never, R_co | Scope]:
    """Run the stream on a fiber in the current ``Scope``, publishing its
    items to a new pubsub, shut down with the scope (and, unless
    ``shutdown_on_end`` is off, when the run ends)."""
    if capacity == "unbounded":
        return self._channel.to_pubsub_array(
            capacity=capacity, replay=replay, shutdown_on_end=shutdown_on_end
        )
    return self._channel.to_pubsub_array(
        capacity=capacity, strategy=strategy, replay=replay, shutdown_on_end=shutdown_on_end
    )

to_pubsub_take(*, capacity, strategy='suspend', replay=None) -> Effect[PubSub[Take[A_co, E_co]], Never, R_co | Scope]

def to_pubsub_take(
    self, *, capacity: Capacity, strategy: Strategy = "suspend", replay: int | None = None
) -> Effect[PubSub[Take[A_co, E_co]], Never, R_co | Scope]:
    """Run the stream on a fiber in the current ``Scope``, publishing
    each chunk as a ``Take``, then its exit, so every subscriber (one
    that subscribes late too) sees the end or the failure."""
    if capacity == "unbounded":
        return self._channel.to_pubsub_take(capacity=capacity, replay=replay)
    return self._channel.to_pubsub_take(capacity=capacity, strategy=strategy, replay=replay)

to_async_iterable() -> StreamAsyncIterable[A]

def to_async_iterable[A, E](self: BridgeMethods[A, E, Never]) -> StreamAsyncIterable[A]:
    """The stream as an ``async for`` source, for asyncio code. Each
    ``async for`` runs the stream afresh on the running loop; a failure
    raises the squashed error. Close an iterator left early with
    ``contextlib.aclosing`` (or ``aclose()``)."""
    from pyfect.context import Context
    from pyfect.stream._async_iterable import StreamAsyncIterable

    return StreamAsyncIterable(self._channel, Context.empty())

to_async_iterable_with(context) -> StreamAsyncIterable[A_co]

def to_async_iterable_with(self, context: Context[R_co]) -> StreamAsyncIterable[A_co]:
    """``to_async_iterable`` for a stream that needs the services in
    ``context``."""
    from pyfect.stream._async_iterable import StreamAsyncIterable

    return StreamAsyncIterable(self._channel, context)

to_async_iterable_effect() -> Effect[StreamAsyncIterable[A_co], Never, R_co]

def to_async_iterable_effect(self) -> Effect[StreamAsyncIterable[A_co], Never, R_co]:
    """``to_async_iterable_with`` the current fiber's services."""
    from pyfect._context_effects import context
    from pyfect.stream._async_iterable import StreamAsyncIterable

    return context().map(lambda ctx: StreamAsyncIterable(self._channel, ctx))

group_by(f, *, buffer_size=GROUP_BUFFER_SIZE, idle_time_to_live=None) -> Stream[tuple[K, Stream[V]], E_co | E2, R_co | R2]

def group_by[K: Hashable, V, E2 = Never, R2 = Never](
    self,
    f: Callable[[A_co], Effect[tuple[K, V], E2, R2]],
    *,
    buffer_size: int = GROUP_BUFFER_SIZE,
    idle_time_to_live: Duration | None = None,
) -> Stream[tuple[K, Stream[V]], E_co | E2, R_co | R2]:
    """Write ``(key, group)`` once per new key ``f(item)`` gives, where
    ``group`` is the stream of that key's values. Upstream runs on a
    fiber and waits while a group holds ``buffer_size`` values; a group
    read by nobody for ``idle_time_to_live`` ends, and the key's next
    value starts a new one. A group that stops early drops its values."""
    from pyfect.stream import _grouping

    return _grouping.group_by(self, f, buffer_size, idle_time_to_live)

group_by_key(f, *, buffer_size=GROUP_BUFFER_SIZE, idle_time_to_live=None) -> Stream[tuple[K, Stream[A_co]], E_co, R_co]

def group_by_key[K: Hashable](
    self,
    f: Callable[[A_co], K],
    *,
    buffer_size: int = GROUP_BUFFER_SIZE,
    idle_time_to_live: Duration | None = None,
) -> Stream[tuple[K, Stream[A_co]], E_co, R_co]:
    """``group_by`` with the items as values, keyed by ``f(item)``."""
    from pyfect.stream import _grouping

    return _grouping.group_by_key(self, f, buffer_size, idle_time_to_live)

group_adjacent_by(f) -> Stream[tuple[K, tuple[A_co, ...]], E_co, R_co]

def group_adjacent_by[K](
    self, f: Callable[[A_co], K]
) -> Stream[tuple[K, tuple[A_co, ...]], E_co, R_co]:
    """Write each run of neighbouring items with equal keys ``f(item)``
    as ``(key, items)``; a key that comes back later starts a new run."""
    from pyfect.stream import _grouping

    return _grouping.group_adjacent_by(self, f)

partition(f, *, capacity=16) -> Effect[tuple[Stream[P, E_co], Stream[F, E_co]], Never, R_co | Scope]

def partition[P, F](
    self, f: Callable[[A_co], Result[P, F]], *, capacity: Capacity = 16
) -> Effect[tuple[Stream[P, E_co], Stream[F, E_co]], Never, R_co | Scope]:
    """Split the items into the ``Ok`` values of ``f(item)`` and the
    ``Err`` ones. Upstream runs on a fiber in the current ``Scope``; the
    faster reader gets at most ``capacity`` items ahead of the slower."""
    from pyfect.stream import _fanout

    return _fanout.partition(self, f, capacity)

partition_effect(f, *, capacity=4096, concurrency=None) -> Effect[tuple[Stream[P, E_co | E2], Stream[F, E_co | E2]], Never, R_co | R2 | Scope]

def partition_effect[P, F, E2 = Never, R2 = Never](
    self,
    f: Callable[[A_co], Effect[Result[P, F], E2, R2]],
    *,
    capacity: Capacity = 4096,
    concurrency: Concurrency = None,
) -> Effect[tuple[Stream[P, E_co | E2], Stream[F, E_co | E2]], Never, R_co | R2 | Scope]:
    """``partition`` with an effectful ``f``, run on up to
    ``concurrency`` items at once (in order)."""
    from pyfect.stream import _fanout

    return _fanout.partition_effect(self, f, capacity, concurrency)

partition_queue(f, *, capacity=4096) -> Effect[tuple[Dequeue[P, E_co | Done], Dequeue[F, E_co | Done]], Never, R_co | Scope]

def partition_queue[P, F](
    self, f: Callable[[A_co], Result[P, F]], *, capacity: Capacity = 4096
) -> Effect[tuple[Dequeue[P, E_co | Done], Dequeue[F, E_co | Done]], Never, R_co | Scope]:
    """``partition`` into two queues, which end (or fail) as the stream
    does."""
    from pyfect.stream import _fanout

    return _fanout.partition_queue(self, f, capacity)

broadcast(*, capacity, strategy='suspend', replay=None) -> Effect[Stream[A_co, E_co], Never, R_co | Scope]

def broadcast(
    self, *, capacity: Capacity, strategy: Strategy = "suspend", replay: int | None = None
) -> Effect[Stream[A_co, E_co], Never, R_co | Scope]:
    """Run the stream on a fiber in the current ``Scope``, publishing its
    chunks to a pubsub of ``capacity`` chunks; each run of the returned
    stream reads what is published from then on (and the last
    ``replay`` chunks), then the end or failure."""
    from pyfect.stream import _fanout

    return _fanout.broadcast(self, capacity, strategy, replay)

broadcast_n(n, *, capacity, strategy='suspend', replay=None) -> Effect[tuple[Stream[A_co, E_co], ...], Never, R_co | Scope]

def broadcast_n(
    self,
    n: int,
    *,
    capacity: Capacity,
    strategy: Strategy = "suspend",
    replay: int | None = None,
) -> Effect[tuple[Stream[A_co, E_co], ...], Never, R_co | Scope]:
    """``n`` streams (none below 1) each reading every chunk. They
    subscribe now, so none misses the start; with ``"suspend"`` the
    upstream stays at most ``capacity`` chunks ahead of the slowest. A
    stream that stops unsubscribes."""
    from pyfect.stream import _fanout

    return _fanout.broadcast_n(self, n, capacity, strategy, replay)

share(*, capacity, strategy='suspend', replay=None, idle_time_to_live=None) -> Effect[Stream[A_co, E_co], Never, R_co | Scope]

def share(
    self,
    *,
    capacity: Capacity,
    strategy: Strategy = "suspend",
    replay: int | None = None,
    idle_time_to_live: Duration | None = None,
) -> Effect[Stream[A_co, E_co], Never, R_co | Scope]:
    """A stream that ``broadcast``s this one when its first reader
    starts and stops it when the last one stops (or ``idle_time_to_live``
    after, so a later reader continues where it left off)."""
    from pyfect.stream import _fanout

    return _fanout.share(self, capacity, strategy, replay, idle_time_to_live)

run(sink) -> Effect[B, E_co | E2, R_co | R2]

def run[B, L, E2 = Never, R2 = Never](
    self, sink: Sink[B, A_co, L, E2, R2]
) -> Effect[B, E_co | E2, R_co | R2]:
    """Run the stream into ``sink`` and succeed with the sink's value.
    The stream's failures pass by the sink's error handlers."""
    from pyfect.stream import _sinking

    _require_sink(sink)
    return _sinking.run(self, sink)

transduce(sink) -> Stream[Any, Any, Any]

def transduce(self, sink: Sink[Any, Any, Any, Any, Any]) -> Stream[Any, Any, Any]:
    """Run ``sink`` again and again on the stream, writing each value it
    ends with; its leftovers start the next run. A stream failure ends
    the current run first."""
    from pyfect.stream import _sinking

    _require_sink(sink)
    return _sinking.transduce(self, sink)

aggregate(sink) -> Stream[Any, Any, Any]

def aggregate(self, sink: Sink[Any, Any, Any, Any, Any]) -> Stream[Any, Any, Any]:
    """``transduce`` while the stream keeps running: a fiber pulls the
    stream as ``sink`` asks for input, and each run of the sink ends
    when the sink is done."""
    from pyfect import schedule
    from pyfect.stream import _sinking

    _require_sink(sink)
    return _sinking.aggregate_within(self, sink, schedule.forever)

aggregate_within(sink, schedule) -> Stream[Any, Any, Any]

def aggregate_within(
    self,
    sink: Sink[Any, Any, Any, Any, Any],
    schedule: Schedule[Any, Any, Any, Any],
) -> Stream[Any, Any, Any]:
    """``aggregate``, also ending each run of ``sink`` when ``schedule``
    says so. The schedule steps once input arrives (with the last value
    written) and its clock leaves out the time between runs; a schedule
    that stops ends the stream once the sink has drained its
    leftovers."""
    from pyfect.stream import _sinking

    _require_sink(sink)
    return _sinking.aggregate_within(self, sink, schedule)

grouped_within(chunk_size, duration) -> Stream[list[A_co], E_co, R_co]

def grouped_within(self, chunk_size: int, duration: Duration) -> Stream[list[A_co], E_co, R_co]:
    """Group the items in lists of up to ``chunk_size`` (at least 1),
    ending a group early once ``duration`` has passed since its first
    item."""
    from pyfect import schedule
    from pyfect.sink import take
    from pyfect.stream import _sinking

    return _sinking.aggregate_within(self, take(max(chunk_size, 1)), schedule.spaced(duration))

pipe_through(sink) -> Stream[L, E_co | E2, R_co | R2]

def pipe_through[B, L, E2 = Never, R2 = Never](
    self, sink: Sink[B, A_co, L, E2, R2]
) -> Stream[L, E_co | E2, R_co | R2]:
    """Run the stream into ``sink`` and write the sink's leftovers. The
    stream's failures pass by the sink's error handlers."""
    from pyfect.stream import _sinking

    _require_sink(sink)
    return _sinking.pipe_through(self, sink)

tap_sink(sink) -> Stream[A_co, E_co | E2, R_co | R2]

def tap_sink[E2 = Never, R2 = Never](
    self, sink: Sink[object, A_co, object, E2, R2]
) -> Stream[A_co, E_co | E2, R_co | R2]:
    """Send each chunk to ``sink`` too, as it is written. The stream
    waits for the sink to take each chunk; once the sink is done it
    goes on alone, and a sink failure fails the stream."""
    from pyfect.stream import _sinking

    _require_sink(sink)
    return _sinking.tap_sink(self, sink)

peel(sink) -> Effect[Any, Any, Any]

def peel(self, sink: Sink[Any, Any, Any, Any, Any]) -> Effect[Any, Any, Any]:
    """Run ``sink`` on the start of the stream and succeed with its value
    and the rest of the stream (the sink's leftovers first). The rest
    reads the stream's pull, which lives in the current ``Scope``."""
    from pyfect.stream import _sinking

    _require_sink(sink)
    return _sinking.peel(self, sink)

run_collect() -> Effect[list[A_co], E_co, R_co]

def run_collect(self) -> Effect[list[A_co], E_co, R_co]:
    """Run the stream and collect its items."""
    from pyfect.stream import _run

    return _run.run_collect(self)

run_drain() -> Effect[None, E_co, R_co]

def run_drain(self) -> Effect[None, E_co, R_co]:
    """Run the stream, dropping its items."""
    return self._channel.run_drain()

run_count() -> Effect[int, E_co, R_co]

def run_count(self) -> Effect[int, E_co, R_co]:
    """Run the stream and count its items."""
    from pyfect.stream import _run

    return _run.run_count(self)

run_sum() -> Effect[N, E, R]

def run_sum[N: (int, float), E, R](self: RunMethods[N, E, R]) -> Effect[N, E, R]:
    """Run a stream of numbers and add them up (``0`` when empty)."""
    from pyfect.stream import _run

    return _run.run_sum(self)

run_fold(initial, f) -> Effect[Z, E_co, R_co]

def run_fold[Z](
    self, initial: Callable[[], Z], f: Callable[[Z, A_co], Z]
) -> Effect[Z, E_co, R_co]:
    """Fold the items into ``initial()`` with ``f``."""
    from pyfect.stream import _run

    return _run.run_fold(self, initial, f)

run_fold_effect(initial, f) -> Effect[Z, E_co | E2, R_co | R2]

def run_fold_effect[Z, E2 = Never, R2 = Never](
    self, initial: Callable[[], Z], f: Callable[[Z, A_co], Effect[Z, E2, R2]]
) -> Effect[Z, E_co | E2, R_co | R2]:
    """Fold the items into ``initial()`` with the effect ``f``, one item
    at a time."""
    from pyfect.stream import _run

    return _run.run_fold_effect(self, initial, f)

run_head() -> Effect[Option[A_co], E_co, R_co]

def run_head(self) -> Effect[Option[A_co], E_co, R_co]:
    """The first item (``Some``), or ``None`` for an empty stream. Stops
    the stream after its first chunk."""
    from pyfect.stream import _run

    return _run.run_head(self)

run_last() -> Effect[Option[A_co], E_co, R_co]

def run_last(self) -> Effect[Option[A_co], E_co, R_co]:
    """The last item (``Some``), or ``None`` for an empty stream."""
    from pyfect.stream import _run

    return _run.run_last(self)

run_for_each(f) -> Effect[None, E_co | E2, R_co | R2]

def run_for_each[E2 = Never, R2 = Never](
    self, f: Callable[[A_co], Effect[object, E2, R2]]
) -> Effect[None, E_co | E2, R_co | R2]:
    """Run ``f`` on each item, in order."""
    from pyfect.stream import _run

    return _run.run_for_each(self, f)

run_for_each_while(f) -> Effect[None, E_co | E2, R_co | R2]

def run_for_each_while[E2 = Never, R2 = Never](
    self, f: Callable[[A_co], Effect[bool, E2, R2]]
) -> Effect[None, E_co | E2, R_co | R2]:
    """Run ``f`` on each item until it succeeds with ``False``; the
    stream stops there."""
    from pyfect.stream import _run

    return _run.run_for_each_while(self, f)

run_for_each_array(f) -> Effect[None, E_co | E2, R_co | R2]

def run_for_each_array[E2 = Never, R2 = Never](
    self, f: Callable[[tuple[A_co, ...]], Effect[object, E2, R2]]
) -> Effect[None, E_co | E2, R_co | R2]:
    """Run ``f`` on each chunk (a non-empty tuple), in order."""
    return self._channel.run_for_each(f)

run_string() -> Effect[str, E, R]

def run_string[E, R](self: RunMethods[str, E, R]) -> Effect[str, E, R]:
    """Run a stream of strings and join them (Effect's ``mkString``)."""
    from pyfect.stream import _run

    return _run.run_string(self)

run_bytes() -> Effect[bytes, E, R]

def run_bytes[E, R](self: RunMethods[bytes, E, R]) -> Effect[bytes, E, R]:
    """Run a stream of byte strings and join them (Effect's
    ``mkUint8Array`` and ``mkArrayBuffer``)."""
    from pyfect.stream import _run

    return _run.run_bytes(self)

to_pull() -> Effect[Pull[tuple[A_co, ...], E_co], Never, R_co | Scope]

def to_pull(self) -> Effect[Pull[tuple[A_co, ...], E_co], Never, R_co | Scope]:
    """Build the stream's pull in the current ``Scope``, which holds its
    resources. Each pull succeeds with the next chunk or fails with
    ``Done()`` at the end; fibers pulling at once take turns."""
    return self._channel.to_pull()

update_service(tag, f) -> Stream[A_co, E_co, R_co | T]

def update_service[S, T](
    self, tag: TagLike[S, T], f: Callable[[S], S]
) -> Stream[A_co, E_co, R_co | T]:
    """Run with ``f`` applied to the service of ``tag``, which this
    stream then needs."""
    from pyfect.stream._stream import Stream

    return Stream(self._channel.update_service(tag, f))

with_span(name, **options) -> Stream[A_co, E_co, R_co]

def with_span(self, name: str, **options: Unpack[SpanOptions]) -> Stream[A_co, E_co, R_co]:
    """Run inside a new span, the parent of the spans made inside. It
    ends with the stream's exit."""
    from pyfect.stream._stream import Stream

    return Stream(self._channel.with_span(name, **options))

schedule(schedule) -> Stream[A_co, E_co | EE, R_co | RR]

def schedule[EE = Never, RR = Never](
    self, schedule: Schedule[object, A_co, EE, RR]
) -> Stream[A_co, E_co | EE, R_co | RR]:
    """Step ``schedule`` with each item, writing it (one per chunk) after
    the step's delay; end when the schedule stops, dropping the item
    that stopped it."""
    from pyfect.stream import _timing

    return _timing.schedule(self, schedule)

repeat(schedule) -> Stream[Any, Any, Any]

def repeat(
    self, schedule: Schedule[Any, Any, Any, Any] | Callable[[Any], Any], /
) -> Stream[Any, Any, Any]:
    """Run the stream again after each end, while ``schedule`` recurs.
    Each run sees its step's ``schedule.CurrentMetadata``."""
    from pyfect.stream._stream import Stream

    return Stream(self._channel.repeat(schedule))

repeat_elements(schedule) -> Stream[A_co, E_co | EE, R_co | RR]

def repeat_elements[EE = Never, RR = Never](
    self, schedule: Schedule[object, A_co, EE, RR]
) -> Stream[A_co, E_co | EE, R_co | RR]:
    """Write each item, then again each time ``schedule`` (stepped with
    it, fresh for each item) recurs."""
    from pyfect.stream import _timing

    return _timing.repeat_elements(self, schedule)

forever() -> Stream[A_co, E_co, R_co]

def forever(self) -> Stream[A_co, E_co, R_co]:
    """Run the stream again each time it ends, until it fails."""
    from pyfect.stream._stream import Stream

    forever: Any = self._channel.forever()
    return Stream(forever)

timeout(duration) -> Stream[A_co, E_co, R_co]

def timeout(self, duration: Duration) -> Stream[A_co, E_co, R_co]:
    """End the stream if a pull waits ``duration`` for its chunk."""
    from pyfect.stream import _timing

    return _timing.timeout_or_else(self, duration, None)

timeout_or_else(duration, or_else) -> Stream[A_co | B, E_co | E2, R_co | R2]

def timeout_or_else[B, E2 = Never, R2 = Never](
    self, duration: Duration, or_else: Callable[[], Stream[B, E2, R2]]
) -> Stream[A_co | B, E_co | E2, R_co | R2]:
    """Go on with ``or_else()`` if a pull waits ``duration`` for its
    chunk."""
    from pyfect.stream import _timing

    return _timing.timeout_or_else(self, duration, or_else)

debounce(duration) -> Stream[A_co, E_co, R_co]

def debounce(self, duration: Duration) -> Stream[A_co, E_co, R_co]:
    """Write the last item of a burst once ``duration`` passes without a
    new chunk; earlier items of the burst are dropped."""
    from pyfect.stream import _timing

    return _timing.debounce(self, duration)

throttle(*, cost, units, duration, burst=0, strategy='shape') -> Stream[A_co, E_co, R_co]

def throttle(
    self,
    *,
    cost: Callable[[tuple[A_co, ...]], float],
    units: float,
    duration: Duration,
    burst: float = 0,
    strategy: ThrottleStrategy = "shape",
) -> Stream[A_co, E_co, R_co]:
    """Limit the stream to ``units`` of ``cost`` per ``duration``, with a
    bucket of up to ``units + burst`` tokens: ``"shape"`` delays a chunk
    until its tokens are there, ``"enforce"`` drops it."""
    from pyfect.effect import succeed
    from pyfect.stream import _timing

    return _timing.throttle(
        self,
        lambda chunk: succeed(cost(chunk)),
        units,
        duration,
        burst=burst,
        strategy=strategy,
    )

throttle_effect(*, cost, units, duration, burst=0, strategy='shape') -> Stream[A_co, E_co | E2, R_co | R2]

def throttle_effect[E2 = Never, R2 = Never](
    self,
    *,
    cost: Callable[[tuple[A_co, ...]], Effect[float, E2, R2]],
    units: float,
    duration: Duration,
    burst: float = 0,
    strategy: ThrottleStrategy = "shape",
) -> Stream[A_co, E_co | E2, R_co | R2]:
    """``throttle`` with an effectful ``cost``."""
    from pyfect.stream import _timing

    return _timing.throttle(self, cost, units, duration, burst=burst, strategy=strategy)

buffer(*, capacity, strategy='suspend') -> Stream[A_co, E_co, R_co]

def buffer(
    self, *, capacity: int | Literal["unbounded"], strategy: Strategy = "suspend"
) -> Stream[A_co, E_co, R_co]:
    """Pull ahead on a fiber, keeping up to ``capacity`` items for the
    reader; a full buffer waits, or drops the new (``"dropping"``) or
    oldest (``"sliding"``) items."""
    from pyfect.stream._stream import Stream

    if capacity == "unbounded":
        return Stream(self._channel.buffer_array(capacity=capacity))
    return Stream(self._channel.buffer_array(capacity=capacity, strategy=strategy))

buffer_array(*, capacity, strategy='suspend') -> Stream[A_co, E_co, R_co]

def buffer_array(
    self, *, capacity: int | Literal["unbounded"], strategy: Strategy = "suspend"
) -> Stream[A_co, E_co, R_co]:
    """``buffer`` counting chunks rather than items."""
    from pyfect.stream._stream import Stream

    if capacity == "unbounded":
        return Stream(self._channel.buffer(capacity=capacity))
    return Stream(self._channel.buffer(capacity=capacity, strategy=strategy))

halt_when(eff) -> Stream[A_co, E_co | E2, R_co | R2]

def halt_when[E2 = Never, R2 = Never](
    self, eff: Effect[object, E2, R2]
) -> Stream[A_co, E_co | E2, R_co | R2]:
    """Run ``eff`` alongside; once it has finished, the next pull ends
    the stream (or fails with ``eff``'s failure) instead of pulling."""
    from pyfect._effect_args import require_effect
    from pyfect.stream._stream import Stream

    require_effect(eff, "eff")
    halted: Any = self._channel.halt_when(eff)
    return Stream(halted)

interrupt_when(eff) -> Stream[A_co, E_co | E2, R_co | R2]

def interrupt_when[E2 = Never, R2 = Never](
    self, eff: Effect[object, E2, R2]
) -> Stream[A_co, E_co | E2, R_co | R2]:
    """Run ``eff`` alongside; when it succeeds, interrupt the stream and
    end; when it fails, fail."""
    from pyfect._effect_args import require_effect
    from pyfect.stream._stream import Stream

    require_effect(eff, "eff")
    interrupted: Any = self._channel.interrupt_when(eff)
    return Stream(interrupted)

on_start(eff) -> Stream[A_co, E_co | E2, R_co | R2]

def on_start[E2 = Never, R2 = Never](
    self, eff: Effect[object, E2, R2]
) -> Stream[A_co, E_co | E2, R_co | R2]:
    """Run ``eff`` on the first pull, before the stream starts."""
    from pyfect._effect_args import require_effect
    from pyfect.stream._stream import Stream

    require_effect(eff, "eff")
    return Stream(self._channel.on_start(eff))

on_first(f) -> Stream[A_co, E_co | E2, R_co | R2]

def on_first[E2 = Never, R2 = Never](
    self, f: Callable[[A_co], Effect[object, E2, R2]]
) -> Stream[A_co, E_co | E2, R_co | R2]:
    """Run ``f`` on the first item, before writing it."""
    from pyfect.stream._stream import Stream

    return Stream(self._channel.on_first(lambda chunk: f(chunk[0])))

on_end(eff) -> Stream[A_co, E_co | E2, R_co | R2]

def on_end[E2 = Never, R2 = Never](
    self, eff: Effect[object, E2, R2]
) -> Stream[A_co, E_co | E2, R_co | R2]:
    """Run ``eff`` when the stream ends (not when it fails)."""
    from pyfect._effect_args import require_effect
    from pyfect.stream._stream import Stream

    require_effect(eff, "eff")
    return Stream(self._channel.on_end(eff))

on_exit(f) -> Stream[A_co, E_co, R_co | R2]

def on_exit[R2 = Never](
    self, f: Callable[[Exit[None, E_co]], Effect[object, Never, R2]]
) -> Stream[A_co, E_co, R_co | R2]:
    """Run ``f(exit)`` as soon as the stream ends or fails, or when its
    run stops early."""
    from pyfect.stream._stream import Stream

    return Stream(self._channel.on_exit(f))

on_error(f) -> Stream[A_co, E_co, R_co | R2]

def on_error[R2 = Never](
    self, f: Callable[[Cause[E_co]], Effect[object, Never, R2]]
) -> Stream[A_co, E_co, R_co | R2]:
    """``on_exit`` for a failure (or an early stop that interrupts)."""
    from pyfect.stream._stream import Stream

    return Stream(self._channel.on_error(f))

ensuring(finalizer) -> Stream[A_co, E_co, R_co | R2]

def ensuring[R2 = Never](
    self, finalizer: Effect[object, Never, R2]
) -> Stream[A_co, E_co, R_co | R2]:
    """Run ``finalizer`` however the stream stops."""
    from pyfect._effect_args import require_effect
    from pyfect.stream._stream import Stream

    require_effect(finalizer, "finalizer")
    return Stream(self._channel.ensuring(finalizer))

catch_cause(f) -> Stream[A_co | B, E2, R_co | R2]

def catch_cause[B, E2 = Never, R2 = Never](
    self, f: Callable[[Cause[E_co]], Stream[B, E2, R2]]
) -> Stream[A_co | B, E2, R_co | R2]:
    """On any failure, defects included, release this stream's resources
    and go on with ``f(cause)``."""
    from pyfect.stream._stream import Stream

    return Stream(self._channel.catch_cause(lambda cause: f(cause)._channel))

catch_cause_if(predicate, f) -> Stream[A_co | B, E_co | E2, R_co | R2]

def catch_cause_if[B, E2 = Never, R2 = Never](
    self,
    predicate: Callable[[Cause[E_co]], bool],
    f: Callable[[Cause[E_co]], Stream[B, E2, R2]],
) -> Stream[A_co | B, E_co | E2, R_co | R2]:
    """``catch_cause`` for the causes that pass ``predicate``."""
    from pyfect.stream._stream import Stream

    return Stream(self._channel.catch_cause_if(predicate, lambda cause: f(cause)._channel))

catch_all(f) -> Stream[A_co | B, E2, R_co | R2]

def catch_all[B, E2 = Never, R2 = Never](
    self, f: Callable[[E_co], Stream[B, E2, R2]]
) -> Stream[A_co | B, E2, R_co | R2]:
    """Go on with ``f(error)`` after a typed failure (Effect's
    ``Stream.catch``). Defects and interruptions pass through."""
    from pyfect.stream._stream import Stream

    return Stream(self._channel.catch_all(lambda error: f(error)._channel))

catch_if(predicate, f, *, or_else=None) -> Stream[Any, Any, Any]

def catch_if(
    self,
    predicate: Callable[[Any], bool],
    f: Callable[[Any], Stream[Any, Any, Any]],
    *,
    or_else: Callable[[Any], Stream[Any, Any, Any]] | None = None,
) -> Stream[Any, Any, Any]:
    """Go on with ``f(error)`` after a typed failure that passes
    ``predicate``; ``or_else``, if given, handles the rest. To remove the
    caught errors from the type, use ``pipe(stream.catch_if(...))`` with
    a ``TypeIs`` refinement."""
    from pyfect.stream import _catch

    return _catch.catch_if(self, predicate, f, or_else)

catch_filter(filter, f, *, or_else=None) -> Stream[Any, Any, Any]

def catch_filter(
    self,
    filter: Callable[[Any], Result[Any, Any]],
    f: Callable[[Any], Stream[Any, Any, Any]],
    *,
    or_else: Callable[[Any], Stream[Any, Any, Any]] | None = None,
) -> Stream[Any, Any, Any]:
    """Go on with ``f(value)`` after a typed failure ``filter`` maps to
    ``Ok(value)``. One it maps to ``Err(x)`` goes to ``or_else(x)``, or
    fails with ``x``."""
    from pyfect.stream._constructors import fail
    from pyfect.stream._stream import Stream

    def handle(error: Any) -> Any:
        match filter(error):
            case Ok(value):
                return f(value)._channel
            case Err(rest):
                return (fail(rest) if or_else is None else or_else(rest))._channel

    return Stream(self._channel.catch_all(handle))

catch_cause_filter(filter, f) -> Stream[A_co | B, X | E2, R_co | R2]

def catch_cause_filter[EB, X, B, E2 = Never, R2 = Never](
    self,
    filter: Callable[[Cause[E_co]], Result[EB, Cause[X]]],
    f: Callable[[EB, Cause[E_co]], Stream[B, E2, R2]],
) -> Stream[A_co | B, X | E2, R_co | R2]:
    """Go on with ``f(value, cause)`` after a failure ``filter`` maps to
    ``Ok(value)``; one it maps to ``Err(rest)`` fails with ``rest``."""
    from pyfect.stream._constructors import fail_cause
    from pyfect.stream._stream import Stream

    def handle(cause: Cause[E_co]) -> Any:
        match filter(cause):
            case Ok(value):
                return f(value, cause)._channel
            case Err(rest):
                return fail_cause(rest)._channel

    return Stream(self._channel.catch_cause(handle))

catch_defect(f) -> Stream[A_co | B, E_co | E2, R_co | R2]

def catch_defect[B, E2 = Never, R2 = Never](
    self, f: Callable[[BaseException], Stream[B, E2, R2]]
) -> Stream[A_co | B, E_co | E2, R_co | R2]:
    """Go on with ``f(defect)`` after a failure with a defect. Typed
    errors pass through."""
    from pyfect.stream._stream import Stream

    return Stream(self._channel.catch_defect(lambda defect: f(defect)._channel))

or_else_if_empty(or_else) -> Stream[A_co | B, E_co | E2, R_co | R2]

def or_else_if_empty[B, E2 = Never, R2 = Never](
    self, or_else: Callable[[], Stream[B, E2, R2]]
) -> Stream[A_co | B, E_co | E2, R_co | R2]:
    """Go on with ``or_else()`` if this stream ends without writing."""
    from pyfect.stream._stream import Stream

    return Stream(self._channel.or_else_if_empty(lambda _: or_else()._channel))

or_else_succeed(f) -> Stream[A_co | B, Never, R_co]

def or_else_succeed[B](self, f: Callable[[E_co], B]) -> Stream[A_co | B, Never, R_co]:
    """Write ``f(error)`` and end instead of failing with a typed error."""
    from pyfect.stream._constructors import succeed

    return self.catch_all(lambda error: succeed(f(error)))

tap_cause(f) -> Stream[A_co, E_co | E2, R_co | R2]

def tap_cause[E2 = Never, R2 = Never](
    self, f: Callable[[Cause[E_co]], Effect[object, E2, R2]]
) -> Stream[A_co, E_co | E2, R_co | R2]:
    """Run ``f(cause)`` when the stream fails, then fail the same way."""
    from pyfect.stream._stream import Stream

    return Stream(self._channel.tap_cause(f))

tap_error(f) -> Stream[A_co, E_co | E2, R_co | R2]

def tap_error[E2 = Never, R2 = Never](
    self, f: Callable[[E_co], Effect[object, E2, R2]]
) -> Stream[A_co, E_co | E2, R_co | R2]:
    """Run ``f(error)`` when the stream fails with a typed error."""
    from pyfect.stream._stream import Stream

    return Stream(self._channel.tap_error(f))

tap_error_tag(tag, f) -> Stream[Any, Any, Any]

def tap_error_tag(
    self,
    tag: type[Any] | tuple[type[Any], ...],
    f: Callable[[Any], Effect[object, Any, Any]],
) -> Stream[Any, Any, Any]:
    """``tap_error`` for the errors that are instances of ``tag`` (a
    class or a tuple of classes)."""
    from pyfect.effect import void

    return self.tap_error(lambda e: f(e) if isinstance(e, tag) else void)

tap_defect(f) -> Stream[A_co, E_co | E2, R_co | R2]

def tap_defect[E2 = Never, R2 = Never](
    self, f: Callable[[BaseException], Effect[object, E2, R2]]
) -> Stream[A_co, E_co | E2, R_co | R2]:
    """Run ``f(defect)`` when the stream fails with a defect."""
    from pyfect.effect import void

    def tap(cause: Cause[E_co]) -> Effect[object, E2, R2]:
        defect = cause.find_defect()
        return void if defect is None else f(defect)

    return self.tap_cause(tap)

map_error(f) -> Stream[A_co, E2, R_co]

def map_error[E2](self, f: Callable[[E_co], E2]) -> Stream[A_co, E2, R_co]:
    """Apply ``f`` to the typed error. Defects and interruptions stay."""
    from pyfect.stream._stream import Stream

    return Stream(self._channel.map_error(f))

or_die() -> Stream[A_co, Never, R_co]

def or_die(self) -> Stream[A_co, Never, R_co]:
    """Die with the typed error instead of failing with it."""
    from pyfect.stream._stream import Stream

    return Stream(self._channel.or_die())

ignore(*, log=False) -> Stream[A_co, Never, R_co]

def ignore(self, *, log: bool | Severity = False) -> Stream[A_co, Never, R_co]:
    """End instead of failing with a typed error. With ``log``, log the
    cause first, at the current log level (``True``) or the level
    given."""
    from pyfect.stream._stream import Stream

    ignored: Any = self._channel.ignore(log=log)
    return Stream(ignored)

ignore_cause(*, log=False) -> Stream[A_co, Never, R_co]

def ignore_cause(self, *, log: bool | Severity = False) -> Stream[A_co, Never, R_co]:
    """End instead of failing in any way. ``log`` as for ``ignore``."""
    from pyfect.stream._stream import Stream

    ignored: Any = self._channel.ignore_cause(log=log)
    return Stream(ignored)

retry(policy) -> Stream[Any, Any, Any]

def retry(
    self, policy: Schedule[Any, Any, Any, Any] | Callable[[Any], Any], /
) -> Stream[Any, Any, Any]:
    """Run the stream again from the start after each typed failure,
    while ``policy`` recurs; then fail with the last error. Items
    already written stay written. A run that writes an item resets the
    policy; each run sees its step's ``schedule.CurrentMetadata``."""
    from pyfect.stream._stream import Stream

    return Stream(self._channel.retry(policy))

zip(that) -> Stream[tuple[A_co, B], E_co | E2, R_co | R2]

def zip[B, E2 = Never, R2 = Never](
    self, that: Stream[B, E2, R2]
) -> Stream[tuple[A_co, B], E_co | E2, R_co | R2]:
    """Pair the items of both streams in order, pulling both at once;
    end when either ends."""
    require_stream(that, "that")
    return self.zip_with(that, _pair)

zip_with(that, f) -> Stream[C, E_co | E2, R_co | R2]

def zip_with[B, C, E2 = Never, R2 = Never](
    self, that: Stream[B, E2, R2], f: Callable[[A_co, B], C]
) -> Stream[C, E_co | E2, R_co | R2]:
    """``zip``, writing ``f(left, right)`` for each pair."""
    from pyfect.stream import _zipping

    require_stream(that, "that")
    return _zipping.zip_with(self, that, f)

zip_with_array(that, f) -> Stream[C, E_co | E2, R_co | R2]

def zip_with_array[B, C, E2 = Never, R2 = Never](
    self,
    that: Stream[B, E2, R2],
    f: Callable[
        [tuple[A_co, ...], tuple[B, ...]],
        tuple[tuple[C, ...], tuple[A_co, ...], tuple[B, ...]],
    ],
) -> Stream[C, E_co | E2, R_co | R2]:
    """Zip chunk by chunk: ``f(left, right)`` gives the chunk to write
    and the left and right items it left over, which the next call gets
    before anything new is pulled from that side."""
    from pyfect.stream import _zipping

    require_stream(that, "that")
    return _zipping.zip_with_array(self, that, f)

zip_left(that) -> Stream[A_co, E_co | E2, R_co | R2]

def zip_left[E2 = Never, R2 = Never](
    self, that: Stream[object, E2, R2]
) -> Stream[A_co, E_co | E2, R_co | R2]:
    """``zip``, keeping the left items."""
    require_stream(that, "that")
    return self.zip_with(that, _left)

zip_right(that) -> Stream[B, E_co | E2, R_co | R2]

def zip_right[B, E2 = Never, R2 = Never](
    self, that: Stream[B, E2, R2]
) -> Stream[B, E_co | E2, R_co | R2]:
    """``zip``, keeping the right items."""
    require_stream(that, "that")
    return self.zip_with(that, _right)

zip_flatten(that) -> Stream[tuple[*Ts, B], E_co | E2, R_co | R2]

def zip_flatten[*Ts, B, E2 = Never, R2 = Never](
    self: ZipMethods[tuple[*Ts], E_co, R_co], that: Stream[B, E2, R2]
) -> Stream[tuple[*Ts, B], E_co | E2, R_co | R2]:
    """``zip`` for a stream of tuples: append each right item to its
    left tuple."""
    require_stream(that, "that")
    return self.zip_with(that, lambda left, right: (*left, right))

zip_latest(that) -> Stream[tuple[A_co, B], E_co | E2, R_co | R2]

def zip_latest[B, E2 = Never, R2 = Never](
    self, that: Stream[B, E2, R2]
) -> Stream[tuple[A_co, B], E_co | E2, R_co | R2]:
    """Once both streams have written, pair each new item with the
    other stream's latest; end when either ends."""
    from pyfect.stream import _zipping

    require_stream(that, "that")
    return _zipping.zip_latest_all(self, that)

zip_latest_with(that, f) -> Stream[C, E_co | E2, R_co | R2]

def zip_latest_with[B, C, E2 = Never, R2 = Never](
    self, that: Stream[B, E2, R2], f: Callable[[A_co, B], C]
) -> Stream[C, E_co | E2, R_co | R2]:
    """``zip_latest``, writing ``f(left, right)`` for each pair."""
    require_stream(that, "that")
    return self.zip_latest(that).map(lambda pair: f(*pair))

cross(that) -> Stream[tuple[A_co, B], E_co | E2, R_co | R2]

def cross[B, E2 = Never, R2 = Never](
    self, that: Stream[B, E2, R2]
) -> Stream[tuple[A_co, B], E_co | E2, R_co | R2]:
    """Every pair of a left and a right item: ``that`` runs again for
    each left item."""
    require_stream(that, "that")
    return self.cross_with(that, _pair)

cross_with(that, f) -> Stream[C, E_co | E2, R_co | R2]

def cross_with[B, C, E2 = Never, R2 = Never](
    self, that: Stream[B, E2, R2], f: Callable[[A_co, B], C]
) -> Stream[C, E_co | E2, R_co | R2]:
    """``cross``, writing ``f(left, right)`` for each pair."""
    from pyfect.stream._base import as_stream

    require_stream(that, "that")
    return as_stream(self).flat_map(lambda left: that.map(lambda right: f(left, right)))

merge(that, *, halt_strategy='both') -> Stream[A_co | B, E_co | E2, R_co | R2]

def merge[B, E2 = Never, R2 = Never](
    self, that: Stream[B, E2, R2], *, halt_strategy: HaltStrategy = "both"
) -> Stream[A_co | B, E_co | E2, R_co | R2]:
    """Run both streams at once, writing items as either writes them.
    The merge ends when both have ended, or just the ``"left"`` (this)
    or ``"right"`` one, or ``"either"``; a failure of either fails it."""
    from pyfect.stream._stream import Stream

    require_stream(that, "that")
    return Stream(self._channel.merge(that._channel, halt_strategy=halt_strategy))

merge_effect(eff) -> Stream[A_co, E_co | E2, R_co | R2]

def merge_effect[E2 = Never, R2 = Never](
    self, eff: Effect[object, E2, R2]
) -> Stream[A_co, E_co | E2, R_co | R2]:
    """Run ``eff`` alongside the stream: its failure fails the stream;
    it is interrupted when the stream ends."""
    from pyfect._effect_args import require_effect
    from pyfect.stream._stream import Stream

    require_effect(eff, "eff")
    return Stream(self._channel.merge_effect(eff))

merge_result(that) -> Stream[Result[A_co, B], E_co | E2, R_co | R2]

def merge_result[B, E2 = Never, R2 = Never](
    self, that: Stream[B, E2, R2]
) -> Stream[Result[A_co, B], E_co | E2, R_co | R2]:
    """``merge``, writing this stream's items as ``Ok`` and ``that``'s as
    ``Err``."""
    from pyfect.result import Err, Ok
    from pyfect.stream._base import as_stream

    require_stream(that, "that")
    lefts: Stream[Result[A_co, B], E_co, R_co] = as_stream(self).map(Ok)
    return lefts.merge(that.map(Err))

merge_left(that) -> Stream[A_co, E_co | E2, R_co | R2]

def merge_left[E2 = Never, R2 = Never](
    self, that: Stream[object, E2, R2]
) -> Stream[A_co, E_co | E2, R_co | R2]:
    """Run ``that`` alongside for its effects, writing this stream's
    items."""
    require_stream(that, "that")
    return self.merge_effect(that.run_drain())

merge_right(that) -> Stream[B, E_co | E2, R_co | R2]

def merge_right[B, E2 = Never, R2 = Never](
    self, that: Stream[B, E2, R2]
) -> Stream[B, E_co | E2, R_co | R2]:
    """Run this stream alongside for its effects, writing ``that``'s
    items."""
    from pyfect.stream._base import as_stream

    require_stream(that, "that")
    return that.merge_effect(as_stream(self).run_drain())

combine(that, initial, f) -> Stream[Any, Any, Any]

def combine(
    self,
    that: Stream[Any, Any, Any],
    initial: Callable[[], Any],
    f: Callable[[Any, Any, Any], Effect[tuple[Any, Any], Any, Any]],
) -> Stream[Any, Any, Any]:
    """Write what ``f(state, pull_left, pull_right)`` makes, threading
    its state; ``f`` pulls single items from either side as it likes and
    ends the stream by failing with ``Done`` (as a pull of an ended side
    does)."""
    from pyfect.stream._mapping import one
    from pyfect.stream._stream import Stream

    require_stream(that, "that")
    items = self._channel.flatten_array()
    combined = items.combine(that._channel.flatten_array(), initial, f)
    return Stream(combined.map(one))

combine_array(that, initial, f) -> Stream[Any, Any, Any]

def combine_array(
    self,
    that: Stream[Any, Any, Any],
    initial: Callable[[], Any],
    f: Callable[[Any, Any, Any], Effect[tuple[Any, Any], Any, Any]],
) -> Stream[Any, Any, Any]:
    """``combine`` a chunk at a time: the pulls give chunks, and ``f``
    makes a non-empty tuple of items to write."""
    from pyfect.stream._stream import Stream

    require_stream(that, "that")
    return Stream(self._channel.combine(that._channel, initial, f))

interleave(that) -> Stream[A_co | B, E_co | E2, R_co | R2]

def interleave[B, E2 = Never, R2 = Never](
    self, that: Stream[B, E2, R2]
) -> Stream[A_co | B, E_co | E2, R_co | R2]:
    """Alternate items from this stream and ``that``, starting here; when
    one ends, write the rest of the other."""
    from pyfect.stream import _zipping

    require_stream(that, "that")
    return _zipping.interleave_with(self, that, _zipping.alternating())

interleave_with(that, decider) -> Stream[A_co | B, E_co | EE, R_co | RR]

def interleave_with[B, EE = Never, RR = Never](
    self, that: Stream[B, EE, RR], decider: Stream[bool, EE, RR]
) -> Stream[A_co | B, E_co | EE, R_co | RR]:
    """Take the next item from this stream when ``decider`` writes
    ``True`` and from ``that`` when it writes ``False``, skipping a side
    that has ended; end when both have ended or ``decider`` does."""
    from pyfect.stream import _zipping

    require_stream(that, "that")
    require_stream(decider, "decider")
    return _zipping.interleave_with(self, that, decider)

race(that) -> Stream[A_co | B, E_co | E2, R_co | R2]

def race[B, E2 = Never, R2 = Never](
    self, that: Stream[B, E2, R2]
) -> Stream[A_co | B, E_co | E2, R_co | R2]:
    """The stream that writes first; the other is interrupted."""
    from pyfect.stream import _zipping

    require_stream(that, "that")
    return _zipping.race_all(self, that)

chunks() -> Stream[tuple[A_co, ...], E_co, R_co]

def chunks(self) -> Stream[tuple[A_co, ...], E_co, R_co]:
    """Write each chunk as one item."""
    from pyfect.stream._stream import Stream

    return Stream(self._channel.map(lambda chunk: (chunk,)))

collect() -> Stream[list[A_co], E_co, R_co]

def collect(self) -> Stream[list[A_co], E_co, R_co]:
    """Write every item as one list, once the stream ends."""
    from pyfect.stream._base import as_stream
    from pyfect.stream._constructors import from_effect

    return from_effect(as_stream(self).run_collect())

accumulate() -> Stream[tuple[A_co, ...], E_co, R_co]

def accumulate(self) -> Stream[tuple[A_co, ...], E_co, R_co]:
    """For each chunk, write a tuple of every item so far."""
    from pyfect.stream import _chunking

    return _chunking.accumulate(self)

rechunk(size) -> Stream[A_co, E_co, R_co]

def rechunk(self, size: int) -> Stream[A_co, E_co, R_co]:
    """Write the items in chunks of ``size`` (the last may be smaller)."""
    from pyfect.stream import _chunking

    return _chunking.rechunk(self, size)

grouped(size) -> Stream[tuple[A_co, ...], E_co, R_co]

def grouped(self, size: int) -> Stream[tuple[A_co, ...], E_co, R_co]:
    """Write the items in tuples of ``size`` (the last may be smaller)."""
    return self.rechunk(size).chunks()

sliding(size) -> Stream[tuple[A_co, ...], E_co, R_co]

def sliding(self, size: int) -> Stream[tuple[A_co, ...], E_co, R_co]:
    """Windows of ``size`` items, each one item on from the last."""
    from pyfect.stream import _chunking

    return _chunking.sliding_size(self, size, 1)

sliding_size(size, step) -> Stream[tuple[A_co, ...], E_co, R_co]

def sliding_size(self, size: int, step: int) -> Stream[tuple[A_co, ...], E_co, R_co]:
    """Windows of ``size`` items, each ``step`` items on from the last.
    When the stream ends or fails, the items not yet in a full window
    are written as one shorter window first."""
    from pyfect.stream import _chunking

    return _chunking.sliding_size(self, size, step)

split(predicate) -> Stream[tuple[A_co, ...], E_co, R_co]

def split(self, predicate: Callable[[A_co], bool]) -> Stream[tuple[A_co, ...], E_co, R_co]:
    """Write the runs of items between the items that pass ``predicate``
    (which are dropped); empty runs are skipped."""
    from pyfect.stream import _chunking

    return _chunking.split(self, predicate)

scan(initial, f) -> Stream[S, E_co, R_co]

def scan[S](self, initial: Callable[[], S], f: Callable[[S, A_co], S]) -> Stream[S, E_co, R_co]:
    """Write ``initial()``, then each state ``f`` folds the items into."""
    from pyfect.stream import _chunking

    return _chunking.scan(self, initial, f)

scan_effect(initial, f) -> Stream[S, E_co | E2, R_co | R2]

def scan_effect[S, E2 = Never, R2 = Never](
    self, initial: Callable[[], S], f: Callable[[S, A_co], Effect[S, E2, R2]]
) -> Stream[S, E_co | E2, R_co | R2]:
    """``scan`` with an effectful ``f``."""
    from pyfect.stream import _chunking

    return _chunking.scan_effect(self, initial, f)

map_accum(initial, f, *, on_halt=None) -> Stream[B, E_co, R_co]

def map_accum[S, B](
    self,
    initial: Callable[[], S],
    f: Callable[[S, A_co], tuple[S, Sequence[B]]],
    *,
    on_halt: Callable[[S], Sequence[B]] | None = None,
) -> Stream[B, E_co, R_co]:
    """Thread a state through the items: ``f(state, item)`` gives the
    next state and the items to write. When the stream ends or fails,
    ``on_halt(state)`` gives the last items to write."""
    from pyfect.stream import _chunking

    return _chunking.map_accum(self, initial, f, on_halt, array=False)

map_accum_array(initial, f, *, on_halt=None) -> Stream[B, E_co, R_co]

def map_accum_array[S, B](
    self,
    initial: Callable[[], S],
    f: Callable[[S, tuple[A_co, ...]], tuple[S, Sequence[B]]],
    *,
    on_halt: Callable[[S], Sequence[B]] | None = None,
) -> Stream[B, E_co, R_co]:
    """``map_accum`` over each chunk."""
    from pyfect.stream import _chunking

    return _chunking.map_accum(self, initial, f, on_halt, array=True)