✏️ Editing: _tomarbles.py Read Only

from typing import Any, List, Optional

from reactivex import Observable, abc
from reactivex.scheduler import NewThreadScheduler
from reactivex.typing import RelativeTime

new_thread_scheduler = NewThreadScheduler()

def to_marbles(
    timespan: RelativeTime = 0.1, scheduler: Optional[abc.SchedulerBase] = None
):
    def to_marbles(source: Observable[Any]) -> Observable[str]:
        """Convert an observable sequence into a marble diagram string.

Args:
            timespan: [Optional] duration of each character in second.
                If not specified, defaults to 0.1s.
            scheduler: [Optional] The scheduler used to run the the input
                sequence on.

Returns:
            Observable stream.
        """

def subscribe(
            observer: abc.ObserverBase[str],
            scheduler: Optional[abc.SchedulerBase] = None,
        ):
            scheduler = scheduler or new_thread_scheduler

result: List[str] = []
            last = scheduler.now

def add_timespan():
                nonlocal last

now = scheduler.now
                diff = now - last
                last = now
                secs = scheduler.to_seconds(diff)
                timespan_ = scheduler.to_seconds(timespan)
                dashes = "-" * int((secs + timespan_ / 2.0) * (1.0 / timespan_))
                result.append(dashes)

def on_next(value: Any) -> None:
                add_timespan()
                result.append(stringify(value))

def on_error(exception: Exception) -> None:
                add_timespan()
                result.append(stringify(exception))
                observer.on_next("".join(n for n in result))
                observer.on_completed()

def on_completed():
                add_timespan()
                result.append("|")
                observer.on_next("".join(n for n in result))
                observer.on_completed()

return source.subscribe(on_next, on_error, on_completed)

return Observable(subscribe)

return to_marbles

def stringify(value: Any) -> str:
    """Utility for stringifying an event."""
    string = str(value)
    if len(string) > 1:
        string = "(%s)" % string

return string

__all__ = ["stringify"]

🔐 Sid Gifari File Manager Pro