Marble Diagrams with RxPY

This is a fantastic feature to produce and visualize streams and to verify how various operators work on them.

Have also a look at rxmarbles for interactive visualisations.

ONE DASH IS 100 MILLISECONDS!



In [8]:

    
%run startup.py

Create Streams from Strings: from_marbles



In [9]:

    
rst(O.from_marbles)
ts = time.time()
# producing a stream
s = O.from_marbles('1-2-3|')
# mapping into real time:
s2 = s.to_blocking()
# adding times
s3 = s2.map(lambda x: 'val: %s, dt: %s' % (x, time.time()-ts))
# subscribing to it:
d = s3.subscribe(print)









    




========== from_marbles ==========

module rx.testing.marbles
@extensionclassmethod(Observable, alias="from_string")
def from_marbles(cls, string, scheduler=None):
    Convert a marble diagram string to an observable sequence, using
    an optional scheduler to enumerate the events.

    Special characters:
    - = Timespan of 100 ms
    x = on_error()
    | = on_completed()

    All other characters are treated as an on_next() event at the given
    moment they are found on the string.

    Examples:
    1 - res = rx.Observable.from_string("1-2-3-|")
    2 - res = rx.Observable.from_string("1-(42)-3-|")
    3 - res = rx.Observable.from_string("1-2-3-x", rx.Scheduler.timeout)

    Keyword arguments:
    string -- String with marble diagram
    scheduler -- [Optional] Scheduler to run the the input sequence on.

    Returns the observable sequence whose elements are pulled from the
    given marble diagram string.
--------------------------------------------------------------------------------
val: 1, dt: 0.0117020606995
val: 2, dt: 0.125038146973
val: 3, dt: 0.234502077103

Visualize Streams as Marble Strings: to_marbles



In [11]:

    
rst(rx.core.blockingobservable.BlockingObservable.to_marbles)
s1 = O.from_marbles('1---2-3|')
s2 = O.from_marbles('-a-b-c-|')
print(s1.merge(s2).to_blocking().to_marbles())









    




========== to_marbles ==========

module rx.testing.marbles
@extensionmethod(BlockingObservable, alias="to_string")  # noqa
def to_marbles(self, scheduler=None):
    Convert an observable sequence into a marble diagram string

    Keyword arguments:
    scheduler -- [Optional] The scheduler used to run the the input
        sequence on.

    Returns marble string.
--------------------------------------------------------------------------------
1-a-b-2c-3|



In [ ]: