~nch/glue

0442bda3360a2f35a630f576f29c3c344e1bfc60 — nc 1 year, 4 months ago e5a1dfa
initial arrowlets
3 files changed, 104 insertions(+), 1 deletions(-)

A arrowlets.py
A arrowlets_test.py
M glue.py
A arrowlets.py => arrowlets.py +40 -0
@@ 0,0 1,40 @@
from typing import Callable, TypeVar, Any
from typing_extensions import Protocol

class _CpsA:
    @staticmethod
    def lift(f: Callable) -> '_CpsA':
        return _CpsA(lambda x, k: k(f(x)))

    def __init__(self, cps: Callable[[Any, Callable], '_CpsA']):
        self.cps = cps

    def next(self, g):
        g = CpsA(g)
        # Basically cons-ing onto the next function functiong
        # (we build the stack backwards so when we pass `x` in during the run it gets
        # sent through the entire stack)
        # In other words, CPS can be viewed as turning a sequence of funcalls into a linked list
        return _CpsA(lambda x, k: self.cps(x, lambda y: g.cps(y, k)))

    def run(self, x):
        return self.cps(x, lambda y: y)

def CpsA(f):
    if isinstance(f, _CpsA): # identity
        return f
    if callable(f): # function
        return _CpsA.lift(f) # lifted into CpsA
    assert False, f


class SimpleEventA(_CpsA):
    def __init__(self, eventname):
        self.eventname = eventname
        def _f(target, k):
            def handler(event):
                target.unbind(eventname, handler)
                k((target, event))
            target.bind(eventname, handler)
        super().__init__(_f)


A arrowlets_test.py => arrowlets_test.py +59 -0
@@ 0,0 1,59 @@
import unittest
from arrowlets import *

class Dummy:
    def __init__(self):
        self.event_bindings = {}

    def bind(self, event, f):
        if event in self.event_bindings:
            self.event_bindings[event].add(f)
        else:
            self.event_bindings[event] = {f}

    def unbind(self, event, f):
        self.event_bindings[event].remove(f)

    def trigger(self, event):
        assert event in self.event_bindings, f'"{event}" not found'
        # we construct a list to force a copy of this to be created
        # so each function gets triggered, but we may remove it in the
        # middle of iteration
        for f in list(self.event_bindings[event]):
            f(event)

class TestDummy(unittest.TestCase):
    def test_dummy(self):
        d = Dummy()
        flag = [False]
        def f(_):
            flag[0] = True
        d.bind('event', f)
        d.trigger('event')
        self.assertTrue(flag[0])
        flag[0] = False
        d.unbind('event', f)
        d.trigger('event')
        self.assertFalse(flag[0])

class ArrowletsTest(unittest.TestCase):
    def test_CpsA(self):
        add1 = lambda x: x + 1
        add2 = CpsA(add1).next(CpsA(add1))
        self.assertEqual(add2.run(1), 3)

    def test_SimpleEventA(self):
        nclicks = [0]
        def clickTargetA(x):
            target, event = x
            nclicks[0] += 1
            return target

        d = Dummy()
        SimpleEventA('click').next(clickTargetA).run(d)
        d.trigger('click')
        self.assertEqual(nclicks[0], 1)


if __name__ == '__main__':
    unittest.main()

M glue.py => glue.py +5 -1
@@ 60,6 60,9 @@ class Node:
    def move(self, x, y):
        self.canvas.coords(self.id, x, y)

    def dnd_accept(self, source, event):
        return self

class NodeCanvas:
    def __init__(self, root):
        self.canvas = tk.Canvas(root, width=800, height=600, background='#444')


@@ 67,7 70,8 @@ class NodeCanvas:
        self.canvas.dnd_accept = self.dnd_accept

    def dnd_accept(self, source, event):
        return self
        pass
        #return self

    def dnd_motion(self, source, event):
        x, y = _cursor_pos(self.canvas, event)