Code reference

This page documents the code in the pydagoras package, and has been made with the help of mkdocstrings.

node.py

class Node

Source code in pydagoras/node.py

class Node:
    def __init__(self, node_id=None, calc=None, usedby=None, nodetype=None, display_name=None, tooltip='notset'):
        self.calc = calc
        self.node_id = node_id
        self.usedby = usedby
        #self.value = None
        self.dag = DAG()   
        self.dag.set_value(node_id, None)

        self.nodetype = nodetype
        if display_name:
            self.display_name = display_name
        else:
            self.display_name = node_id
        self.tooltip = tooltip
        self.orig_tooltip = tooltip

    def pp(self):
        print(f'NODE: {self.nodetype}, id:{self.node_id}, value:{self.get_value()}')
        print(f'      display_name:{self.display_name} tooltip:{self.tooltip} ')
        print(f'      calc: {self.calc} usedby:{self.usedby[0].node_id if self.usedby else None}')

    def set_tooltip(self, tooltip):
        self.tooltip = tooltip

    def set_value(self, value):
        self.dag.set_value(self.node_id, value)

    def get_value(self):
        return self.dag.get_value(self.node_id)

dag.py

class DAG

Source code in pydagoras/dag.py

class DAG: # implementation
    __shared_state = {}

    def __init__(self):
        self.__dict__ = self.__shared_state

        if hasattr(self,'values'):
            return

        self.values = {}

    def set_value(self, key, value):
        self.values[key] = value

    def get_value(self, key):
        return self.values.get(key, None)

dag_dot.py

class DAG_dot

Source code in pydagoras/dag_dot.py

class DAG_dot:

    def __init__(self, label):
        logger.info(f'creating {label=}')
        self.label = label
        self.dag = DAG()  # Initialize the DAG instance
        self.G=pgv.AGraph(directed=True, strict=True, rankdir='LR', label=label, labelloc="t")
        self.input_nodes = []
        self.internal_nodes = []
        self.output_node = None
        self.nodes = []  # List to keep track of all nodes


        @calc
        def calc_out(node=None):
            return self.n_out.get_value() 

        self.n_out = self.makeNode('out', calc=calc_out, usedby=[], nodetype='out') 


    def makeNode(self,label,calc,usedby=None, nodetype='internal', display_name=None, tooltip=''):
        if usedby is None:
            usedby = [self.n_out]
        n = Node(label,calc,usedby,nodetype,display_name,tooltip)
        if nodetype == 'in':
            self.input_nodes.append(n)
        elif nodetype == 'out':
            self.output_node = n
        else:
            self.internal_nodes.append(n)
        self.nodes.append(n)  # Add the node to the list of all nodes

        self.drawNode(n,usedby, nodetype, tooltip)
        return n


    def drawNode(self,node,usedby,nodetype,tooltip):
        doc = node.display_name
        if nodetype == 'in':
            self.G.add_node(doc, shape="square", tooltip=tooltip)
            for n in usedby:
                self.draw_edge(doc,n.display_name)
        elif nodetype == 'internal':
            self.G.add_node(doc, tooltip=tooltip)
            for n in usedby:
                self.draw_edge(doc,n.display_name)
        elif nodetype == 'out':
            self.G.add_node(doc, color="white")


    def draw_edge(self,node1,node2):
        self.G.add_edge(node1,node2,label='Undefined', fontname="Courier")


    def update_node(self,node1,node2,value,tooltip='not set XXX'):
        fontcolor, color = self.get_colors(value)
        self.G.add_node(node1,color=color,fontcolor=fontcolor,tooltip=tooltip)
        self.G.add_edge(node1,node2, label=value,fontcolor=fontcolor,color=color, fontname="Courier")

    # special cases
    @classmethod
    def get_colors(cls, value):
        if value in ( 'e',):
            return 'red', 'red'
        return 'blue', 'green'

    def set_input(self,node_id,value): # set values in all node
        for node in self.nodes:
            if node.node_id == node_id:
                for usedby in node.usedby:
                    self.update_node(node.display_name,usedby.node_id, value=value, tooltip=node.tooltip)
                    self.set_input(usedby.node_id, value)  # recursion 
                    #self.setValue(node, value)
                self.setValue(node, value)


    def setValue(self,n,v):
        if v == self.dag.get_value(n):
            return

        # build the DAG
        n.set_value(v) 
        x = self.dag.get_value(n.node_id)

        for u in n.usedby:
            if u.calc == None:
                continue
            try:
                #new_value = u.calc(node=n)
                #new_value = u.calc(self, node=n)
                new_value = u.calc(self, node=n)  # Call the calc function with self and node
            except Exception as e:
                print('Error in setValue')
                print (str(e))
                new_value = 'ee' # ??


            self.dag.set_value(u.node_id, new_value)
            #self.set_value(u,new_value)
            #self.setValue(u,new_value)

        if not n.usedby:
            return

        if n.usedby[0].usedby == []: # output node
            # not shown!
            msg = 'update dag_dot.py %s %s' %(n.usedby[0].node_id, n.get_value())
            logger.info (msg)
            #print(msg)


    def ppInputs(self):
        for n in self.input_nodes:
            n.pp()

    def ppInternals(self):
        for n in self.internal_nodes:
            n.pp()

    def ppOutput(self):
        self.output_node.pp()

    def pp(self):
        print('==============') 
        print(f'DAG: {self.label}')
        print('Inputs:')
        self.ppInputs()
        print('Internals:')
        self.ppInternals()
        print('Output:')
        self.ppOutput()

    def ppValues(self):
        print(self.dag.values)  # Print the values of all nodes in the DAG

    def get_value(self):
         return self.dag.get_value(self.label)

    def set_value(self, node, value):
        self.dag.set_value(node.node_id, value)
        self.setValue(node,value)

function calc

Source code in pydagoras/dag_dot.py

def calc(f1): # decorator deffinition
    @wraps(f1)
    def f3(dag, *args, **kwargs):
        node=kwargs['node']

        u_node = node.usedby[0] if node.usedby else None
        #        self.update_node(u_node.node_id,o_node.node_id, value='-', tooltip=node.tooltip)

        try:
            rtn = f1(*args, **kwargs) # Call the original function
            u_node.set_value(rtn)  # Set the value of the node
            u_node.set_tooltip(u_node.orig_tooltip)  # Set the tooltip of the node

        except Exception as e:
            print ('Error in %s: %s' %(u_node.node_id,str(e)))
            rtn = 'e'
            u_node.set_value(rtn)  # Set the value of the node
            print('o'*20)
            u_node.set_tooltip(str(e))  # Set the tooltip of the node

        #print('checking nodes to update....')
        for u_node in node.usedby:
            #print(f'Updating node: {u_node.node_id} with value: {rtn}')
            dag.set_input(u_node.node_id, rtn)

        return rtn
    return f3