Initial commit

simra.py

+#!/usr/bin/env python3
+import tkinter as TK
+import collections
+
+flatten = lambda l: [item for sublist in l for item in sublist]
+
+# TODO: History should also store UP
+# TODO: Store and enforce node dependencies
+
+class Mergeable_dict(dict):
+    """Like dict, except that foo += bar update all entries of foo with the
+corresponding values in bar. This requires that bar.keys() is a subset of
+foo.keys(), otherwise IndexError is raised."""
+    def __iadd__(self, other):
+        for k in other.keys():
+            if not k in self.keys(): raise IndexError()
+            self[k] = other[k]
+        return self
+
+class AN:
+    """
+    An automata network.
+
+    Fields:
+    nodes        -- a set node "names", that can be arbitrary (hashable) values
+    dependencies -- a dict mapping nodes to lists/sets of nodes, on which it
+                    depends for the update
+    update_funcs -- a dict mapping nodes to functions ; each function takes a
+                    dictionary (neighbour node→current value) as a unique
+                    parameter
+                    (it is possible to do my_an.update_funcs += dict to merge
+                    in a dictionary, i.e. to change the functions of only some
+                    nodes)
+    update_mode  -- a list of sets (or list of lists) interpreted as a
+                    periodic update mode
+    init_state   -- a dict from nodes to values, that tells which values should
+                    be used when the network is (re)initialized
+    state        -- a dict nodes→values holding the current state (value of each
+                    node) of the network
+    history      -- a circular buffer of the last self.history_len states
+                    traversed by the network
+    up           -- (Update Pointer) is an index for self.update_mode ; it
+                    “points” to the next update to do
+    """
+    def __init__(self, nodes, dependencies, update_funcs, update_mode, init_state=None, history_len=10):
+        self.nodes = set(nodes)
+        self.dependencies = dependencies
+        self.update_funcs = update_funcs
+        self.update_mode = update_mode
+        self.init_state = init_state
+        self.reinit_state() # Sets self.state, self.up
+        self.history = collections.deque(maxlen=history_len)
+
+    # ################################################################
+
+    @property
+    def update_funcs(self):
+        return self._update_funcs
+
+    @update_funcs.setter
+    def update_funcs(self, new_update_funcs):
+        if not self.nodes.issubset(set(new_update_funcs.keys())):
+            raise ValueError("update function is missing for some nodes")
+        self._update_funcs = Mergeable_dict()
+        self.set_multiple_update_funcs(self, new_update_funcs)
+    
+    def set_multiple_update_funcs(self, new_update_funcs):
+        """
+        If update_funcs is a dictionary node->function, change the update
+        function of each node to the corresponding function. All nodes in
+        update_funcs.keys() need to exist; but not all nodes from the network
+        need to be changed by this operation.
+        """
+        for n in keys(new_update_funcs):
+            self.set_update_func(n, new_update_funcs[n])
+
+    def set_update_func(self, node, new_update_func):
+        """Set the update function of a node."""
+        if not node in self.nodes:
+            raise ValueError("trying to set update_func of nonexisting node")
+        self._update_funcs[node] = new_update_func
+
+    @property
+    def dependencies(self):
+        return self._dependencies
+
+    @dependencies.setter
+    def dependencies(self, new_dependencies):
+        if not self.nodes == set(new_dependencies.keys()):
+            raise ValueError("""trying to set a dependency for a nonexistent
+ node, or missing a dependency list for a node.""")
+        self._dependencies = new_dependencies
+
+    @property
+    def update_mode(self):
+        return self._update_mode
+
+    @update_mode.setter
+    def update_mode(self, update_mode):
+        """
+        Change the update mode.
+        update_mode needs to be a list of sets (or list of lists) and is
+        interpreted as a periodic update mode.
+        Changing the update_mode clears the state history to ensure that
+        an.unstep() followed an.step() is no-op (or raises an exception).
+        """
+        if not set(flatten(update_mode)).issubset(self.nodes):
+            raise ValueError("update_mode tries to update a nonexistent node")
+        self._update_mode = update_mode
+        self.history.clear()
+
+    @property
+    def init_state(self):
+        return self._init_state
+    
+    @init_state.setter
+    def init_state(self, init_state):
+        """
+        Change initial state.
+        Initial state is used when reinitializing the network.
+        If init_state is None, `0` for each node is used; otherwise, init_state
+        must be a dictionary node->value, where init_state.keys() is equal to
+        the set of nodes of the network.
+        """
+        if init_state:
+            if set(init_state.keys()) != self.nodes:
+                raise ValueError("init_state either misses state for a node or has a sate for a nonexistent node")
+            else:
+                self._init_state = init_state.copy()
+        else:
+            self._init_state = {}
+            for n in self.nodes:
+                self._init_state[n] = 0
+    
+    @property
+    def up(self):
+        return self._up
+
+    @up.setter
+    def up(self, new_up):
+        self._up = new_up % len(self.history)
+
+    @property
+    def history_len(self):
+        return len(self.history)
+
+    @history_len.setter
+    def history_len(self, new_history_len):
+        new_history = collections.deque(new_history_len)
+        for i in range(min(history_len, self.history_len)):
+            new_history.append(self.history[-(i+1)])
+        self.history = new_history
+
+    # ################################################################
+    
+    def reinitialize(self):
+        """
+        Reset the current state to a copy of the initial state,
+        and the UP to 0.
+        """
+        self.history.append(self.state)
+        self.state = self.init_state.copy()
+        self.up = 0
+
+    def step(self):
+        """Execute one step of the AN."""
+        new_state = self.state.copy()
+        for n in self.update_mode[self.up]:
+            new_state[n] = (self.update_func[n])(self.state)
+        
+        self.history.append(self.state)
+        self.state = new_state
+        self.up += 1
+
+    def run(self, steps):
+        """Execute `steps` steps of the AN."""
+        for i in range(steps):
+            self.step()
+
+    def unstep(self):
+        """Roll back one step of the AN, within limits of history_len."""
+        self.state = self.history.pop()
+        self.up -= 1
+
+# ################################################################
+
+def button_press(event):
+    print("clicked at ({},{})".format(event.x, event.y))
+
+def button_release(event):
+    print("Released at ({},{})".format(event.x, event.y))
+
+def main():
+    window = Tk()
+    window.title("SIMulateur de Réseau d'Automates")
+
+    canvas = Canvas(window)
+    canvas.create_rectangle(128, 64, 32, 32,
+        outline="#ff0000", fill="#aaaaaa")
+    canvas.bind("<Button-1>", button_press)
+    canvas.bind("<ButtonRelease-1>", button_release)
+    canvas.pack(fill=BOTH, expand=1)
+
+    window.mainloop()
+
+if __name__ == '__main__':
+    main()
+