From 53d89ea2f14a0e164e1b31173474dc47146dd0d5 Mon Sep 17 00:00:00 2001
From: George Marchment <georgemarchment@yahoo.fr>
Date: Fri, 25 Apr 2025 11:50:45 +0200
Subject: [PATCH] Updated the representation of the conditions in the workflow
-> the condition now flow:))
---
src/graph.py | 92 +++++++++++++++++++++++++++++++++++++++++-----------
1 file changed, 73 insertions(+), 19 deletions(-)
diff --git a/src/graph.py b/src/graph.py
index 5e22b22..1593040 100644
--- a/src/graph.py
+++ b/src/graph.py
@@ -7,6 +7,9 @@ import re
import ctypes
from .outils_graph import *
+from .bioflowinsighterror import BioFlowInsightError
+from . import constant
+
def get_object(address):
address = int(re.findall(r"\dx\w+", address)[0], base=16)
@@ -177,38 +180,67 @@ class Graph():
#This is a dictionnary which links every node to it's connected process
node_2_processes = copy.deepcopy(self.link_dico)
already_searched = {}
- for node in node_2_processes:
- already_searched[node] = [node]
+ for nodeA in node_2_processes:
+ already_searched[nodeA] = [nodeA]
changed = True
while(changed):
changed = False
- for node in node_2_processes:
- temp = node_2_processes[node].copy()
- for give in node_2_processes[node]:
+ for nodeA in node_2_processes:
+ temp = node_2_processes[nodeA].copy()
+ for give in node_2_processes[nodeA]:
if(is_operation(give)):
temp.remove(give)
- if(node!=give and give not in already_searched[node]):
- already_searched[node] += give
+ if(nodeA!=give and give not in already_searched[nodeA]):
+ already_searched[nodeA] += give
temp_temp = node_2_processes[give]
- for node_temp in already_searched[node]:
+ for node_temp in already_searched[nodeA]:
try:
temp_temp.remove(node_temp)
except:
None
temp+=temp_temp
changed = True
- node_2_processes[node] = list(set(temp))
+ node_2_processes[nodeA] = list(set(temp))
+
+ #Getting the dico of paths in the workflow
+ path_from_process_to_other_processes = {}
+ searching = True
+ timeout = 0
+ flat_dico = get_flatten_dico(dico.copy())
+ process_ids = []
+ for node in flat_dico["nodes"]:
+ process_ids.append(node['id'])
+ while(searching and timeout<constant.WHILE_UPPER_BOUND):
+ searching = False
+ for nodeA in process_ids:
+ try:
+ tmp = path_from_process_to_other_processes[nodeA]
+ except:
+ path_from_process_to_other_processes[nodeA] = node_2_processes[nodeA].copy()
+ searching = True
+
+ for connectedA in path_from_process_to_other_processes[nodeA]:
+ for nodeB in path_from_process_to_other_processes:
+ if(connectedA==nodeB):
+ for connectedB in path_from_process_to_other_processes[nodeB]:
+ if(connectedB not in path_from_process_to_other_processes[nodeA]):
+ path_from_process_to_other_processes[nodeA].append(connectedB)
+ searching = True
+ timeout+=1
+ if(timeout>=constant.WHILE_UPPER_BOUND):
+ raise BioFlowInsightError(f"The WHILE_UPPER_BOUND was exceeded. BioFlow-Insight was unable to create the dico of paths.", type="Unable to create the dico of paths")
+
-
#%2x%2x%2x
colours = ["#ffbe00", "#0055c8", "#6e6e00", "#a0006e", "#ff5a00", "#82dc73", "#ff82b4", "#d282be", "#d2d200", "#dc9600", "#6e491e", "#00643c", "#82c8e6", "#640082"]
- links_added, links_added__with_conditions= [], []
- def add_edges(dico, condition, checking_conditions, index = 0):
+ links_added, links_added__with_conditions, links_added_with_index_OG= [], [], []
+ def add_edges(dico, condition, checking_conditions, index = 0, added_in_condition = {}):
for node in dico['nodes']:
edges = node_2_processes[node['id']]
for B in edges:
link = f"{node['id']} -> {B}"
link_with_condition = f"{node['id']} -> {B} ({condition})"
+ link_with_index = f"{node['id']} -> {B} ({index})"
if(link_with_condition not in links_added__with_conditions):
if(checking_conditions):
p1, p2 = get_object(node['id']), get_object(B)
@@ -220,32 +252,54 @@ class Graph():
# if(c2.get_value()==condition):
if(link not in links_added):
dico['edges'].append({'A': node['id'], 'B': B, 'label': '', "colour": colours[index], "condition": condition})
+ added_in_condition[node['id']] = ''
links_added.append(link)
- links_added.append(link_with_condition)
+ links_added__with_conditions.append(link_with_condition)
+ links_added_with_index_OG.append(link_with_index)
else:
checking_conditions = False
else:
if(link not in links_added):
dico['edges'].append({'A': node['id'], 'B': B, 'label': ''})
links_added.append(link)
+
for sub in dico['subworkflows']:
- add_edges(dico["subworkflows"][sub], condition, checking_conditions, index=index)
+ add_edges(dico["subworkflows"][sub], condition, checking_conditions, index=index, added_in_condition = added_in_condition)
return checking_conditions
+
+ links_added_with_index = []
+ def add_edges_flow_edges(dico, added_in_condition):
+ for index in added_in_condition:
+ nodes_with_condition = added_in_condition[index]
+ for node in dico['nodes']:
+ edges = node_2_processes[node['id']]
+ for B in edges:
+ link_with_index = f"{node['id']} -> {B} ({index})"
+ for node_with_condition in nodes_with_condition:
+ if(link_with_index not in links_added_with_index and link_with_index not in links_added_with_index_OG):
+ if(node['id'] in path_from_process_to_other_processes[node_with_condition] and B in path_from_process_to_other_processes[node_with_condition]):
+ dico['edges'].append({'A': node['id'], 'B': B, 'label': '', "colour": colours[index]})
+ links_added_with_index.append(link_with_index)
+
+ for sub in dico['subworkflows']:
+ add_edges_flow_edges(dico["subworkflows"][sub], added_in_condition)
- #if(self.workflow.get_duplicate_status()):
- if(False):#Right now not generating the colored edges
+ if(self.workflow.get_duplicate_status()):
+ #if(False):#Right now not generating the colored edges
checking_conditions = True
most_influential_conditions = self.workflow.get_most_influential_conditions()
list_most_influential_conditions = list(most_influential_conditions.keys())
index = 0
+ added_in_condition = {}
while(checking_conditions and\
- index<len(list_most_influential_conditions) and \
- index<5):
+ index<len(list_most_influential_conditions) and index<5):
+ added_in_condition[index] = {}
condition = list_most_influential_conditions[index]
- checking_conditions = add_edges(dico, condition, checking_conditions, index=index)
+ checking_conditions = add_edges(dico, condition, checking_conditions, index=index, added_in_condition = added_in_condition[index])
index+=1
add_edges(dico, condition="", checking_conditions=False)
+ add_edges_flow_edges(dico, added_in_condition)
else:
add_edges(dico, condition="", checking_conditions=False)
--
GitLab