## Reactive programming with applicatives!¶

One interesting thing we could do with applicatives, is to experiment with gui programming. I have been playing around with excelent flare library in purescript and it heavily uses applicative style to achieve a style of code that is resembling working with spreadsheets.

In ipython, we have access to ipywidgets, that can be used to create simple forms.

In [12]:
from ipywidgets import *
from IPython.display import display
slider = IntSlider()
label = Label("0")
display(slider)
display(label)


Then we can listen on changes and set values of these forms-objects.

In [2]:
def updateLabel(value):
label.value = str(100 * slider.value)

slider.observe(updateLabel)


Now, lets we want to have two sliders and display the sum.

In [13]:
from ipywidgets import *
from IPython.display import display
sliderA = IntSlider()
sliderB = IntSlider()
labelSum = Label("0")
display(sliderA)
display(sliderB)
display(labelSum)

labelSum.value = str(sliderA.value + sliderB.value)



• the objects are hardcoded in the update function
• I need to wire the observables by hand

You could solve this in many diferent ways, but one that I would preffer looks like this:

@lift
def sum(a,b):
return str(a + b)

labelSum.subscribesTo(sum(sliderA,sliderB))

First, lets bring back the Applicative class and the accompanying machinery, we will need

• curry
• apply
• lift
In [4]:
from functools import partial

def curry(n, fn):
if n == 1:
return fn
if n == 2:
return lambda x:partial(fn,x)
else:
return lambda x:curry(n-1,partial(fn,x))

class Applicative:
def pure(self, val):
raise NotImplementedError();

def apply(self, fn, val):
raise NotImplementedError();

def lift(self, fn):
def lifted(arg0, *args):
result = self.apply(self.pure(curry(len(args)+1, fn)),arg0)
for a in args:
result = self.apply(result, a)
return result
return lifted


Now I will create a simple applicative, reactive object, that I would call Propagated.

In [5]:
class Propagated:
def __init__(self, last):
self.last = last
self.subscriptions = []

def getValue(self):
return self.last

def subscribesTo(self, cb):
self.subscriptions +=[cb]

def observe(self, prop):
prop.subscribesTo(lambda x: self.setValue(x))

def setValue(self, value):
self.last = value
for cb in self.subscriptions:
cb(value)


Now I can create the Applicative instance.

In [6]:
class ApplyPropagated(Applicative):
def pure(self, val):
return Propagated(val)

def apply(self, fn, val):
result = Propagated(fn.getValue()(val.getValue()))
fn.subscribesTo(lambda f: result.setValue(f(val.getValue())))
val.subscribesTo(lambda v: result.setValue(fn.getValue()(v)))
return result

applyPropagated = ApplyPropagated()


Now I create two wrappers for converting the ipywidgets to faucets and sinks.

In [7]:
def propagatedWidgetFaucet(widget):
display(widget)
propagated = Propagated(widget.value)
def update(x):
if x['name'] == 'value':
propagated.setValue(x['new'])
widget.observe(update)
return propagated

def propagatedWidgetSink(widget):
display(widget)
p = Propagated(widget.value)
def setVal(value):
widget.value = value
p.subscribesTo(setVal)
return p;


And with these, we can easily define the data-flow of our form.

• A and B are the inputs
• sump is just a normal function, that does what we need
• SUM is the output
In [14]:
A = propagatedWidgetFaucet(IntSlider())
B = propagatedWidgetFaucet(IntSlider())
SUM = propagatedWidgetSink(Label("0"))

@applyPropagated.lift
def sump(*a):
return str(sum(a))

SUM.observe(sump(A, B))


What if we want to add another slider, with one more veriable? That is not a problem. We could even factor out str out of our sump.

In [9]:
A = propagatedWidgetFaucet(IntSlider())
B = propagatedWidgetFaucet(IntSlider())
C = propagatedWidgetFaucet(IntSlider())
SUM = propagatedWidgetSink(Label("0"))

@applyPropagated.lift
def strp(a):
return str(a)

@applyPropagated.lift
def sump(*a):
return sum(a)

SUM.observe(strp(sump(A, B, C)))


If we want to process the values differently, we just have to:

• create a new WidgetSink
• create a different function to process the values
• combine it with any previously defined functions:
In [10]:
AVG = propagatedWidgetSink(Label("0"))
@applyPropagated.lift
def awgp(*a):
return sum(a)/len(a)
AVG.observe(strp(awgp(A, B, C)))


And this, in my mind showcases the elegance of applicatives.

• We have object that has fairly complex api, i.e. our Propagated.
• Conceptually, we know that the object just wraps some value.

Because we have implemented applicative, if user of our Propagated wants to combine these values

• they just create a function that know how to combine the values and nothing about how they are encapsulated
• then they lift the function and use it on wrapped values :-)