@extensionmethod(Observable, alias="to_string")
def to_marbles(self, scheduler=None):
    scheduler = scheduler or timeout_scheduler
    source = self

    def subscribe(observer):
        result = []
        previously = [scheduler.now()]

        def add_timespan():
            now = scheduler.now()
            diff = now - previously[0]
            previously[0] = now
            msecs = scheduler.to_relative(diff)
            dashes = "-" * int((msecs+50)/100)
            result.append(dashes)

        def on_next(value):
            add_timespan()
            result.append(value)

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

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

        return source.subscribe(on_next, on_error, on_completed)
    return AnonymousObservable(subscribe)
Convert an observable sequence into a marble diagram string
dataset/ETHPy150Open ReactiveX/RxPY/rx/testing/marbles.py/to_marbles