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.

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.

def updateLabel(value):
    label.value = str(100 * slider.value)
            
slider.observe(updateLabel)

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

from ipywidgets import *
from IPython.display import display
sliderA = IntSlider()
sliderB = IntSlider()
labelSum = Label("0")
display(sliderA)
display(sliderB)
display(labelSum)

def updateSum(value):
    labelSum.value = str(sliderA.value + sliderB.value)
    
sliderA.observe(updateSum)
sliderB.observe(updateSum)

I dislike two things about this:

• 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

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.

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.

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.

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

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.

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:

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 :-)