#NEWLINE# @ extensionclassmethod ( Observable , alias = 'from_string' ) #NEWLINE# def from_marbles( cls , string , scheduler = None ) : #NEWLINE# #INDENT# 'Convert a marble diagram string to an observable sequence, using\n    an optional scheduler to enumerate the events.\n\n    Special characters:\n    - = Timespan of 100 ms\n    x = on_error()\n    | = on_completed()\n\n    All other characters are treated as an on_next() event at the given\n    moment they are found on the string.\n\n    Examples:\n    1 - res = rx.Observable.from_string("1-2-3-|")\n    2 - res = rx.Observable.from_string("1-(42)-3-|")\n    3 - res = rx.Observable.from_string("1-2-3-x", rx.Scheduler.timeout)\n\n    Keyword arguments:\n    string -- String with marble diagram\n    scheduler -- [Optional] Scheduler to run the the input sequence on.\n\n    Returns the observable sequence whose elements are pulled from the\n    given marble diagram string.\n    ' #NEWLINE# scheduler = ( scheduler or timeout_scheduler ) #NEWLINE# completed = [ False ] #NEWLINE# messages = [ ] #NEWLINE# timespan = [ 0 ] #NEWLINE# def handle_timespan( value ) : #NEWLINE# #INDENT# timespan [ 0 ] += 100 #NEWLINE# #UNINDENT# def handle_on_next( value ) : #NEWLINE# #INDENT# messages . append ( on_next ( timespan [ 0 ] , value ) ) #NEWLINE# #UNINDENT# def handle_on_completed( value ) : #NEWLINE# #INDENT# messages . append ( on_completed ( timespan [ 0 ] ) ) #NEWLINE# completed [ 0 ] = True #NEWLINE# #UNINDENT# def handle_on_error( value ) : #NEWLINE# #INDENT# messages . append ( on_error ( timespan [ 0 ] , value ) ) #NEWLINE# completed [ 0 ] = True #NEWLINE# #UNINDENT# specials = { #NEWLINE# #INDENT# '-' : handle_timespan , #NEWLINE# 'x' : handle_on_error , #NEWLINE# 'X' : handle_on_error , #NEWLINE# '|' : handle_on_completed , #NEWLINE# #UNINDENT# } #NEWLINE# for groups in _tokens . findall ( string ) : #NEWLINE# #INDENT# for token in groups : #NEWLINE# #INDENT# if token : #NEWLINE# #INDENT# func = specials . get ( token , func ) #NEWLINE# func ( token ) #NEWLINE# #UNINDENT# #UNINDENT# #UNINDENT# if ( not completed [ 0 ] ) : #NEWLINE# #INDENT# messages . append ( on_completed ( timespan [ 0 ] ) ) #NEWLINE# #UNINDENT# return ColdObservable ( scheduler , messages )