<SYSTEM_TASK:>
Given the following text description, write Python code to implement the functionality described below step by step
<END_TASK>
<USER_TASK:>
Description:
 Reactive Programming in Python
https
Step1:  Concepts
Rx is about processing streams of events. With Rx you
Step2:  Concepts
The subscribe method takes an observer, or one to three callbacks for handing on_next(), on_error(), and on_completed().
This is why we can use print directly as the observer in the example above, since it becomes the on_next() handler for an anonymous observer
Interactive Rx - Generating
Step3:  Reference
Python Library
Step4:  Filtering a Sequence
Step5:  Interactive Rx - Filtering
Step6:  Transforming a Sequence
Step7:  Interactive Rx - Transformations
Step8:  Merge
Step9:  Space-Time of rx
Marbles and Marble Diagrams
- = Timespan of 100 ms
x = on_error()
| = on_completed()
Step10:  Note
Note the -x and -4x. The xs and ys stream items may come in any order
Interactive Rx - Marbles
Step11:  Reference
https
<ASSISTANT_TASK:>
Python Code:
from ipywidgets import interact, interactive, fixed
import ipywidgets as widgets
import rx
rx
from rx import Observable, Observer
help(Observer)
dir(Observer)
help(Observable)
Explanation: Reactive Programming in Python
https://github.com/ReactiveX/RxPY/blob/develop/notebooks/Getting%20Started.ipynb
About
An API for asynchronous programming with observable streams
ReactiveX, or Rx for short, is an API for programming with observable event streams.
The Observer pattern done right.
ReactiveX is a combination of the best ideas from the 
Observer pattern, 
The Iterator pattern, and 
Functional programming
http://reactivex.io/
End of explanation
class MyObserver(Observer):
    def on_next(self, x):
        print("Got: %s" % x)
    
    def on_error(self, e):
        print("Got ERROR: %s" % x)
    
    def on_completed(self):
        print("Sequence completed")
subject = Observable.from_iterable(range(10))
subject
observer = MyObserver()
observer
subject.subscribe(observer)
Explanation: Concepts
Rx is about processing streams of events. With Rx you:
Tell what you want to process (Observable)
How you want to process it (A composition of operators)
What you want to do with the result (Observer)
It's important to understand that with Rx you describe what you want to do with events if and when they arrive. 
It's all a 
Declarative composition of operators
Will do some processing on the events when they arrive.
If nothing happens, then nothing is processed.
Thus the pattern is that you subscribe to an Observable using an Observer:
subscription = Observable.subscribe(observer)
NOTE: Observables are not active in themselves. They need to be subscribed to make something happen. Simply having an Observable lying around doesn't make anything happen.
Generating a Sequence
End of explanation
def interactiveObservable(count):
    print("-" * 30)
    
    sequence = range(count)
    subject = Observable.from_iterable(sequence)
    observer = MyObserver()
    subject.subscribe(observer)
countSlider = widgets.IntSlider(min=2, max=10, step=1, value=3)
interact(interactiveObservable, count=countSlider)
from functools import partial
def interactiveSequenceObservable(observer, count):
    print("-" * 30)
    sequence = range(count)
    subject = Observable.from_iterable(sequence)
    subject.subscribe(observer)
Explanation: Concepts
The subscribe method takes an observer, or one to three callbacks for handing on_next(), on_error(), and on_completed().
This is why we can use print directly as the observer in the example above, since it becomes the on_next() handler for an anonymous observer
Interactive Rx - Generating
End of explanation
countSlider = widgets.IntSlider(min=2, max=10, step=1, value=3)
callback = partial(interactiveSequenceObservable, MyObserver())
callback.__name__ = interactiveSequenceObservable.__name__
interact(callback, count=countSlider)
countSlider = widgets.IntSlider(min=2, max=10, step=1, value=3)
callback = partial(interactiveSequenceObservable, print)
callback.__name__ = interactiveSequenceObservable.__name__
interact(callback, count=countSlider)
Explanation: Reference
Python Library: functools
https://docs.python.org/3/library/functools.html
End of explanation
subject = Observable.from_iterable(range(10))
disposable = subject.filter(lambda x : x % 2).subscribe(print)
disposable
Explanation: Filtering a Sequence
End of explanation
def interactiveFilter(observer, count, divisor):
    sequence = range(1, count+1)
    subject = Observable.from_iterable(sequence)
    filtered = subject.filter(lambda x: x % divisor)
    filtered.subscribe(observer)
callback = partial(interactiveFilter, print)
callback.__name__ = interactiveFilter.__name__
countSlider = widgets.IntSlider(min=2, max=50, value=10)
divisorSlider = widgets.IntSlider(min=2, max=10, value=2)
interact(callback, count=countSlider, divisor=divisorSlider)
Explanation: Interactive Rx - Filtering
End of explanation
subject = Observable.from_iterable(range(10))
disposable = subject.map(lambda x : x * 2).subscribe(print)
disposable
Explanation: Transforming a Sequence
End of explanation
def interactiveTransformer(observer, count, code, debug=False):
    sequence = range(1, count+1)
    if debug:
        print(sequence, tuple(sequence))
    subject = Observable.from_iterable(sequence)
    callback = eval(code)
    filtered = subject.map(callback)
    filtered.subscribe(observer)
callback = partial(interactiveTransformer, print)
callback.__name__ = interactiveTransformer.__name__
countSlider = widgets.IntSlider(min=2, max=50, value=10)
codeText = widgets.Text(value="lambda x: x % 2")
debugCheckBox = widgets.Checkbox(value=False)
interact(callback, count=countSlider, code=codeText, debug=debugCheckBox)
codeText = widgets.Text(value="lambda x: x % 2")
interact(callback, count=countSlider, code=codeText, debug=debugCheckBox)
Explanation: Interactive Rx - Transformations
End of explanation
xs = Observable.from_iterable(range(1, 6))
ys = Observable.from_iterable('abcde')
xsMys = xs.merge(ys)
xsMys.subscribe(print)
Explanation: Merge
End of explanation
from rx.testing import marbles
xs = Observable.from_marbles('a-b-c-|')
xs.to_blocking().to_marbles()
xs = Observable.from_marbles('a-b-c-x-e')
ys = Observable.from_marbles('1-2-3-4-5')
xs.merge(ys).to_blocking().to_marbles()
xs = Observable.from_marbles('1-2-3-x-4')
ys = Observable.from_marbles('1-2-3-4-5')
xs.merge(ys).to_blocking().to_marbles()
Explanation: Space-Time of rx
Marbles and Marble Diagrams
- = Timespan of 100 ms
x = on_error()
| = on_completed()
End of explanation
def interactiveMarbles(stream1, stream2):
    xs = Observable.from_marbles(stream1)
    ys = Observable.from_marbles(stream2)
    print(xs.merge(ys).to_blocking().to_marbles())
stream1Marble = widgets.Text(value="1-2-3-4-5")
stream2Marble = widgets.Text(value="a-b-c-d-e")
interact(interactiveMarbles, stream1=stream1Marble, stream2=stream2Marble)
Explanation: Note
Note the -x and -4x. The xs and ys stream items may come in any order
Interactive Rx - Marbles
End of explanation
result = eval('lambda x: x * 2')
result
tuple(map(result, (1, 2, 3, 4)))
Explanation: Reference
https://martinfowler.com/articles/collection-pipeline/
Python Sandbox
End of explanation
<END_TASK>