1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
|
from typing import Callable, Type, TypeVar, Union, cast
from reactivex import Observable, abc, defer
from reactivex import operators as ops
from reactivex.internal.utils import NotSet
from reactivex.typing import Accumulator
_T = TypeVar("_T")
_TState = TypeVar("_TState")
def scan_(
accumulator: Accumulator[_TState, _T], seed: Union[_TState, Type[NotSet]] = NotSet
) -> Callable[[Observable[_T]], Observable[_TState]]:
has_seed = seed is not NotSet
def scan(source: Observable[_T]) -> Observable[_TState]:
"""Partially applied scan operator.
Applies an accumulator function over an observable sequence and
returns each intermediate result.
Examples:
>>> scanned = scan(source)
Args:
source: The observable source to scan.
Returns:
An observable sequence containing the accumulated values.
"""
def factory(scheduler: abc.SchedulerBase) -> Observable[_TState]:
has_accumulation = False
accumulation: _TState = cast(_TState, None)
def projection(x: _T) -> _TState:
nonlocal has_accumulation
nonlocal accumulation
if has_accumulation:
accumulation = accumulator(accumulation, x)
else:
accumulation = (
accumulator(cast(_TState, seed), x)
if has_seed
else cast(_TState, x)
)
has_accumulation = True
return accumulation
return source.pipe(ops.map(projection))
return defer(factory)
return scan
__all__ = ["scan_"]
|
