@extensionclassmethod(Observable, alias="from_string")
def from_marbles(cls, string, scheduler=None):

    scheduler = scheduler or timeout_scheduler

    completed = [False]
    messages = []
    timespan = [0]

    def handle_timespan(value):
        timespan[0] += 100

    def handle_on_next(value):
        messages.append(on_next(timespan[0], value))

    def handle_on_completed(value):
        messages.append(on_completed(timespan[0]))
        completed[0] = True

    def handle_on_error(value):
        messages.append(on_error(timespan[0], value))
        completed[0] = True

    specials = {
        '-': handle_timespan,
        'x': handle_on_error,
        'X': handle_on_error,
        '|': handle_on_completed
    }

    for groups in _tokens.findall(string):
        for token in groups:
            if token:
                func = specials.get(token, handle_on_next)
                func(token)

    if not completed[0]:
        messages.append(on_completed(timespan[0]))

    return ColdObservable(scheduler, messages)
Convert a marble diagram string to an observable sequence, using
dataset/ETHPy150Open ReactiveX/RxPY/rx/testing/marbles.py/from_marbles