import graphviz class MyGraph(graphviz.Digraph): def __init__(self, *args): super(MyGraph, self).__init__(*args) self.graph_attr['rankdir'] = 'LR' self.node_attr['shape'] = 'Mrecord' self.graph_attr['splines'] = 'ortho' self.graph_attr['nodesep'] = '0.8' self.edge_attr.update(penwidth='2') def add_event(self, name): super(MyGraph, self).node(name, shape="circle", label="") def add_and_gateway(self, *args): super(MyGraph, self).node(*args, shape="diamond", width=".6",height=".6", fixedsize="true", fontsize="40",label="+") def add_xor_gateway(self, *args, **kwargs): super(MyGraph, self).node(*args, shape="diamond", width=".6",height=".6", fixedsize="true", fontsize="35",label="×") def add_and_split_gateway(self, source, targets, *args): gateway = 'ANDs '+str(source)+'->'+str(targets) self.add_and_gateway(gateway,*args) super(MyGraph, self).edge(source, gateway) for target in targets: super(MyGraph, self).edge(gateway, target) def add_xor_split_gateway(self, source, targets, *args): gateway = 'XORs '+str(source)+'->'+str(targets) self.add_xor_gateway(gateway, *args) super(MyGraph, self).edge(source, gateway) for target in targets: super(MyGraph, self).edge(gateway, target) def add_and_merge_gateway(self, sources, target, *args): gateway = 'ANDm '+str(sources)+'->'+str(target) self.add_and_gateway(gateway,*args) super(MyGraph, self).edge(gateway,target) for source in sources: super(MyGraph, self).edge(source, gateway) def add_xor_merge_gateway(self, sources, target, *args): gateway = 'XORm '+str(sources)+'->'+str(target) self.add_xor_gateway(gateway, *args) super(MyGraph, self).edge(gateway,target) for source in sources: super(MyGraph, self).edge(source, gateway) G = MyGraph() # adding split gateways based on causality for event in causality: if len(causality[event]) > 1: if tuple(causality[event]) in parallel_events: G.add_and_split_gateway(event,causality[event]) else: G.add_xor_split_gateway(event,causality[event]) # adding merge gateways based on inverted causality for event in inv_causality: if len(inv_causality[event]) > 1: if tuple(inv_causality[event]) in parallel_events: G.add_and_merge_gateway(inv_causality[event],event) else: G.add_xor_merge_gateway(inv_causality[event],event) elif len(inv_causality[event]) == 1: source = list(inv_causality[event])[0] G.edge(source,event) # adding start event G.add_event("start") if len(start_set_events) > 1: if tuple(start_set_events) in parallel_events: G.add_and_split_gateway(event,start_set_events) else: G.add_xor_split_gateway(event,start_set_events) else: G.edge("start",list(start_set_events)[0]) # adding end event G.add_event("end") if len(end_set_events) > 1: if tuple(end_set_events) in parallel_events: G.add_and_merge_gateway(end_set_events,event) else: G.add_xor_merge_gateway(end_set_events,event) else: G.edge(list(end_set_events)[0],"end") G.render('simple_graphviz_graph') G.view('simple_graphviz_graph')