update leakage detection

Author: zyrrron

AlgorithmComparison.py

@@ -115,11 +115,11 @@ def astar_search(g, position, ur):
 
 # One row in the table means one graph, but some rows are different but have same edges to remove because some different nodes may locate in the same
 # edge, so the length of final graph list may longer than the table
-def VeSpAGraphGenerator(table, nodes_group_list, vcofe, g):
+def VespaGraphGenerator(table, nodes_group_list, vcofe, g):
     g_list = []
     edge_remove_listall = []
     for row in table:
-        g_VeSpA = g.copy()
+        g_Vespa = g.copy()
         edge_remove_list = []
         # cut the edges when there are valves onside be set to 0 in a row
         for i in range(len(row)):
@@ -153,10 +153,10 @@ def VeSpAGraphGenerator(table, nodes_group_list, vcofe, g):
             continue
         # remove edges involved
         for e in edge_remove_list:
-            for edge in g_VeSpA.edges():
+            for edge in g_Vespa.edges():
                 if Counter(e) == Counter(edge):
-                    g_VeSpA.remove_edge(e[0], e[1])
-        g_list.append(g_VeSpA)
+                    g_Vespa.remove_edge(e[0], e[1])
+        g_list.append(g_Vespa)
         edge_remove_listall.append(edge_remove_list)
     return g_list
 
@@ -166,55 +166,82 @@ def VeSpAGraphGenerator(table, nodes_group_list, vcofe, g):
 # All node[0] node[1] here will be totally same or different at all. If they have intersection node, they will be
 # connected by the findallConnectedNodes() function. So, we can transfer the current d to a list of tuples consists
 # of two node groups, all nodes in the group should be open or close together.
-def enumerateVeSpAgraphs(g, d, vcofe, listlen):
-    conflict = 0
+def enumerateVespagraphs(g, d, vcofe, listlen):
+
     # creat node group constraint list
     conflict, ConstraintNGList, nodes_group_list = Node_NG_Constraint_translater(d)
     if conflict == 1:
         return 1, []
     # create a truth table including all node groups according to the ConstraintGroupList
-    conflict, table, fp = NodeGroupTruthTableBuilder(nodes_group_list, ConstraintNGList, listlen)
-    I_best = len(table)
+    conflict, table, NotAllGraph = NodeGroupTruthTableBuilder(nodes_group_list, ConstraintNGList, listlen)
+    NumOfGraph = len(table)
     # Use the table to generate graphs which lose some edges compared with the original graph.
-    g_list = VeSpAGraphGenerator(table, nodes_group_list, vcofe, g)
-    return conflict, g_list, fp, I_best
+    g_list = VespaGraphGenerator(table, nodes_group_list, vcofe, g)
+    return conflict, g_list, NotAllGraph, NumOfGraph
+
 
+def get_ports (G):
+    ports = []
+    for node in G:
+        if 'o' not in node:
+            ports.append(node)
+    return ports
 
-def VeSpA_search(g, g_c, position, ConstraintList, VCO2FEdictionary, ur, listlen):
+
+def Vespa_search(g, g_c, position, ConstraintList, VCO2FEdictionary, ur, listlen):
     global pos
     pos = position
     start = time.time()
+    ports = get_ports(g)
+    leakage_fail = 0
 
     # Create a constraint dictionary list represents the constraint equation:
     # Data structure: {'Type': 2, 'TruthTable': [[0, 0]], 'Nodes': [['co1'...], ['co3'...]]}
-    Conflict, NGConstraintDict = NodeGroupConstraintDictBuilder(ConstraintList, g, g_c)
+    Conflict, NGConstraintDict = NodeGroupConstraintDictBuilder(ConstraintList, g_c)
 
     # update graph with removing the edges in constraint type 1
     g, NGConstraintDictNew = updateGraphByNGConstraint(VCO2FEdictionary, g, NGConstraintDict)
     # generate all graphs satisfy the constraint type 2 as a list
-    Conflict, g_list, flagFalseNegative, I_best = enumerateVeSpAgraphs(g, NGConstraintDictNew, VCO2FEdictionary, listlen)
+    Conflict, g_list, NotAllGraph, NumOfGraph = enumerateVespagraphs(g, NGConstraintDictNew, VCO2FEdictionary, listlen)
     if Conflict == 1:
-        print("Constraint conflict 2!", ConstraintList)
+        print("Constraint conflict!", ConstraintList)
         end = time.time()
-        return end - start, [], -2, -1, I_best
-    VeSpAPathMin = []
-    VeSpALengthMin = 0
+        return end - start, [], -2, -1, NumOfGraph, leakage_fail
+    VespaPathMin = []
+    VespaLengthMin = 0
     # If the list is too big, we can randomly choose 1000 graphs from the big list to speedup the procedure. (random way)
     # Here we choose graphs in order from truth table elements are all 1 to all 0. (Our way)
     # if len(g_list) > listlen:
     #     flagFalseNegative = 1
     #     g_list = random.sample(g_list, listlen)
     for gb in g_list:
-        _, VeSpAPath, VeSpALength = netxsp_search(gb, ur)
-        if len(VeSpAPathMin) == 0 or VeSpALengthMin > VeSpALength > 0:
-            VeSpAPathMin = VeSpAPath
-            VeSpALengthMin = VeSpALength
+        _, VespaPath, VespaLength = netxsp_search(gb, ur)
+        if len(VespaPathMin) == 0 or VespaLengthMin > VespaLength > 0:
+            if VespaLength > 0:
+                flag_leakage = False
+                for p in ports:
+                    if p in ur[0]:
+                        continue
+                    ur_new = [ur[0], p]
+                    _, _, l = netxsp_search(gb, ur_new)
+
+                    # leakage issue arise
+                    if l > 0:
+                        if p not in ur[1]:
+                            flag_leakage = True
+                            break
+                # if current graph have leakage issue, skip it
+                if flag_leakage:
+                    leakage_fail += 1
+                    continue
+            VespaPathMin = VespaPath
+            VespaLengthMin = VespaLength
     end = time.time()
-    if not VeSpAPathMin:
-        # print(f"No such a path from {ur[0]} to {ur[1]} using VeSpA I={listlen} searching algorithm")
-        return end - start, [], -1, flagFalseNegative, I_best
-    VeSpATime = end - start
-    return VeSpATime, VeSpAPathMin, VeSpALengthMin, flagFalseNegative, I_best
+    if not VespaPathMin:
+        # print(f"No such a path from {ur[0]} to {ur[1]} using Vespa I={listlen} searching algorithm")
+        return end - start, [], -1, NotAllGraph, NumOfGraph, leakage_fail
+    VespaTime = end - start
+    return VespaTime, VespaPathMin, VespaLengthMin, NotAllGraph, NumOfGraph, leakage_fail
 
 
 def control_search(path, dictionary):
@@ -275,8 +302,8 @@ def simulateControlPathway(time, path, filepath, j, graph, GraphListInfo):
     pass
 
 
-def simulate(NodeInfo, NetxSPTime, NetxSPPath, NCP, DijkstraTime, DijkstraPath, DCP, AstarTime, AstarPath, ACP, VeSpATime,
-             VeSpAPath, BCP, i, j, g, g_c, gli):
+def simulate(NodeInfo, NetxSPTime, NetxSPPath, NCP, DijkstraTime, DijkstraPath, DCP, AstarTime, AstarPath, ACP, VespaTime,
+             VespaPath, BCP, i, j, g, g_c, gli):
     # NetxSP simulate flow pathway
     outputfolderpath = f"TestCaseFiles/NetxSP_result/Section_{i}/{NodeInfo}"
     simulateFlowPathway(NetxSPTime, NetxSPPath, outputfolderpath, j, g, gli)
@@ -289,9 +316,9 @@ def simulate(NodeInfo, NetxSPTime, NetxSPPath, NCP, DijkstraTime, DijkstraPath,
     outputfolderpath = f"TestCaseFiles/A*_result/Section_{i}/{NodeInfo}"
     simulateFlowPathway(AstarTime, AstarPath, outputfolderpath, j, g, gli)
 
-    # VeSpA simulate flow pathway
-    outputfolderpath = f"TestCaseFiles/VeSpA_result/Section_{i}/{NodeInfo}"
-    simulateFlowPathway(VeSpATime, VeSpAPath, outputfolderpath, j, g, gli)
+    # Vespa simulate flow pathway
+    outputfolderpath = f"TestCaseFiles/Vespa_result/Section_{i}/{NodeInfo}"
+    simulateFlowPathway(VespaTime, VespaPath, outputfolderpath, j, g, gli)
 
     # NetxSP simulate control layer pathway
     outputfolderpath = f"TestCaseFiles/NetxSP_result/Section_{i}/{NodeInfo}"
@@ -305,6 +332,6 @@ def simulate(NodeInfo, NetxSPTime, NetxSPPath, NCP, DijkstraTime, DijkstraPath,
     outputfolderpath = f"TestCaseFiles/A*_result/Section_{i}/{NodeInfo}"
     simulateControlPathway(AstarTime, AstarPath, outputfolderpath, j, g, gli)
 
-    # VeSpA simulate control pathway
-    outputfolderpath = f"TestCaseFiles/VeSpA_result/Section_{i}/{NodeInfo}"
-    simulateControlPathway(VeSpATime, VeSpAPath, outputfolderpath, j, g, gli)
+    # Vespa simulate control pathway
+    outputfolderpath = f"TestCaseFiles/Vespa_result/Section_{i}/{NodeInfo}"
+    simulateControlPathway(VespaTime, VespaPath, outputfolderpath, j, g, gli)

ConstraintFunctions.py

@@ -34,7 +34,7 @@ def OriginalConstraintTablebuilder(type, node, t):
 # "[vi, vj]": {Type: 2, Nodes: [[vi, ci, vj], [vj, cj, vi]], TruthTable: [[0,0]]}, vi cannot be opened and closed at the same time
 # If we cannot generate a constraint dictionary, (not type1 and type2, may some other type)
 # then that means the constraint is incorrect, show warning!!!!!
-def NodeGroupConstraintDictBuilder(ConstraintList, g, g_c):
+def NodeGroupConstraintDictBuilder(ConstraintList, g_c):
     d = []
     conflict = 0
     for c in ConstraintList:
@@ -164,8 +164,8 @@ def NodeGroupTruthTableBuilder(nl, cl, listlen):
         listlen = min(listlen, 2**table_col)
     else:
         listlen = 2**table_col
-    d, tab, flagFalseNegative = createTruthTable(d, keys, 0, [], cl, listlen, 0)
+    d, tab, NotAllGraph = createTruthTable(d, keys, 0, [], cl, listlen, 0)
     if len(tab) == 0 and len(cl) != 0:
         conflict = 1
-    return conflict, tab, flagFalseNegative
+    return conflict, tab, NotAllGraph
 

HeatMapBioDesignFixed.py

@@ -6,7 +6,7 @@
 
 from ConstraintMaker import createRandomConstraint
 from MetricsGenerator import calculate_false_pos
-from AlgorithmComparison import VeSpA_search, control_search, netxsp_search, astar_search, findall_control_path
+from AlgorithmComparison import Vespa_search, control_search, netxsp_search, astar_search, findall_control_path
 import random
 import csv
 import networkx as nx
@@ -142,21 +142,21 @@ def locateValveAndCOonFE(vco_path):
                 PathLength = [NetxSPLength, AstarLength]
                 CtrlNodeLists = [NetxSPControlNodeList, AstarControlNodeList]
 
-                # Update the flow edge info after we get RandomConstraintList and use it in VeSpA_search
+                # Update the flow edge info after we get RandomConstraintList and use it in Vespa_search
                 VespaTime = []
                 flagFN = []
                 for ii in a:
-                    g_VeSpA = g.copy()
-                    VeSpATimeii, VeSpAPathii, VeSpALengthii, flag, _ = VeSpA_search(g_VeSpA, g_c, pos, ConstraintList, VCO2FEdictionary, ur, ii)
+                    g_Vespa = g.copy()
+                    VespaTimeii, VespaPathii, VespaLengthii, flag, _, _ = Vespa_search(g_Vespa, g_c, pos, ConstraintList, VCO2FEdictionary, ur, ii)
                     # Find all valves and other control components which may be involved giving the searched path
-                    VeSpAVCOList = control_search(VeSpAPathii, FE2VCOdictionary)
-                    VeSpAControlNodeList, VeSpAControlEdgeList = findall_control_path(VeSpAVCOList, g_c)
+                    VespaVCOList = control_search(VespaPathii, FE2VCOdictionary)
+                    VespaControlNodeList, VespaControlEdgeList = findall_control_path(VespaVCOList, g_c)
 
                     # add middle products into list
-                    VespaTime.append(format(VeSpATimeii, '.5f'))
-                    PathLength.append(VeSpALengthii)
+                    VespaTime.append(format(VespaTimeii, '.5f'))
+                    PathLength.append(VespaLengthii)
                     flagFN.append(flag)
-                    CtrlNodeLists.append(VeSpAControlNodeList)
+                    CtrlNodeLists.append(VespaControlNodeList)
 
                 # Calculate the false positive for each algorithm
                 l, t, nodeslist = calculate_false_pos(PathLength, ConstraintList, CtrlNodeLists, g_c, flagFN)

MetricsGenerator.py

@@ -1,5 +1,6 @@
 from ConstraintFunctions import findallConnectedNodes
 from collections import Counter
+from AlgorithmComparison import get_ports, NodeGroupConstraintDictBuilder, updateGraphByNGConstraint
 
 
 def fp_flag(a, b):
@@ -25,22 +26,36 @@ def checkandappend(nodeslist, b):
     return nodeslist
 
 
-def VeSpA_checker(t, l, flag):
-    # true negative for VeSpA, because it scanned all the possible graphs
+def Vespa_checker(t, l, flag):
+    # true negative for Vespa, because it scanned all the possible graphs
     t = fp_flag(t, l)
     if flag == 1 and t == 2:
         t = 23
     return t, l
 
 
-def calculate_false_pos(p, cl, c, g_c, flag):
+# calculate_false_pos([NetxSPLength, AstarLength, VespaLength1, VespaLength2, VespaLength], ConstraintList,
+#                                                       [NetxSPControlNodeList, AstarControlNodeList, VespaControlNodeList1, VespaControlNodeList2,
+#                                                        VespaControlNodeList], g_c, [flagFalseNegative1, flagFalseNegative2, flagFalseNegative])
+def calculate_false_pos(p, cl, c, g_c, flag, g, ur):
     # l means the result of predict value, t means [0-false, 1-true] in the beginning and fp value in the end.
     l = [1] * len(p)
     t = [1] * len(p)
     for i in range(len(p)):
         if p[i] == -1:
             l[i] = 0
     nodeslist = []
+
+    # check leakage in two naive algorithms
+    # transfer control protocol list to constraint list by labeling all un-mentioned control ports as type 1 constraint.
+    # feed that to NodeGroupConstraintDictBuilder() and updateGraphByNGConstraint(), we can get a residual graph without any edges blocked by the
+    # control ports not in the control protocol list.
+    # remove all the edges if they are not in the control list, then search the other ports.
+    # If find an "other" port can be reached, assign t[i]=0
+    ports = get_ports(g)
+
+
+    # check the constraints again, make sure the target path follow all constraints.
     for cc in cl:
         if cc[0] == 1:
             nodes = findallConnectedNodes([cc[1]], g_c.edges())
@@ -59,13 +74,14 @@ def calculate_false_pos(p, cl, c, g_c, flag):
                             t[i] = 0
             nodeslist = checkandappend(nodeslist, nodes1)
             nodeslist = checkandappend(nodeslist, nodes2)
-    # true positive = 1, false positive = 0, true negative= = 2, false negative = 3
+
+    # true positive = 1, false positive = 0, true negative = 2, false negative = 3
     t[0] = fp_flag(t[0], l[0])
     t[1] = fp_flag(t[1], l[1])
     for i in range(2, len(p)):
-        t[i], l[i] = VeSpA_checker(t[i], l[i], flag[i-2])
+        t[i], l[i] = Vespa_checker(t[i], l[i], flag[i-2])
 
-    # other algorithm can find a path without breaking the constraint, that means VeSpA is false negative
+    # other algorithm can find a path without breaking the constraint, that means Vespa is false negative
     if t[0] == 1 or t[1] == 1:
         for i in range(2, len(p)):
             if t[i] in [23, 2]: