减少爆管定位代码中引入的不确定性

app/algorithms/burst_location/burst_location.py

@@ -220,6 +220,7 @@ def run_burst_location(
                 node_y=node_y,
                 pipe_start_node_all=pipe_start_node_all,
                 pipe_end_node_all=pipe_end_node_all,
+                pipe_diameter=pipe_diameter,
                 couple_node_length=couple_node_length,
                 node_pipe_dic=node_pipe_dic,
                 all_node_series=all_node_series,

app/algorithms/burst_location/burst_locator.py

@@ -345,6 +345,7 @@ def DN_search_multi_simple_add_flow_count_new(
     node_y,
     pipe_start_node_all,
     pipe_end_node_all,
+    pipe_diameter,
     couple_node_length,
     node_pipe_dic,
     all_node_series,
@@ -415,7 +416,12 @@ def DN_search_multi_simple_add_flow_count_new(
 
             # group 分组，得出候选漏损中心
             stage_start = perf_counter()
-            candidate_center_list, candidate_group_list, new_all_node = (
+            (
+                candidate_center_list,
+                candidate_group_list,
+                new_all_node,
+                candidate_center_candidates,
+            ) = (
                 metis_grouping_pipe_weight(
                     G0,
                     wn,
@@ -429,6 +435,7 @@ def DN_search_multi_simple_add_flow_count_new(
                     node_pipe_dic,
                     all_node_series,
                     couple_node_length,
+                    pipe_diameter,
                 )
             )
             _accumulate_stage(stage_timing, "group_partitioning", stage_start)
@@ -479,19 +486,30 @@ def DN_search_multi_simple_add_flow_count_new(
             add_center = []
             leak_center_dict = dict()
             for i in range(len(candidate_center_list)):
-                houxuan_center = []
+                primary_center = candidate_center_list[i]
+                houxuan_center = [
+                    center
+                    for center in candidate_center_candidates[i]
+                    if center != primary_center
+                ]
                 candidate_group_set = set(candidate_group_list[i])
                 for each_center in record_center_dataset:
                     if (
                         each_center in candidate_group_set
-                        and each_center != candidate_center_list[i]
+                        and each_center != primary_center
                     ):
                         houxuan_center.append(each_center)
                 add_center = add_center + houxuan_center
-                houxuan_center.append(candidate_center_list[i])
-                leak_center_dict[candidate_center_list[i]] = houxuan_center
+                leak_center_dict[primary_center] = _dedupe_preserve_order(
+                    houxuan_center + [primary_center]
+                )
             add_center = _dedupe_preserve_order(add_center)
-            for each_center in candidate_center_list:
+            for each_group_centers in candidate_center_candidates:
+                for each_center in each_group_centers:
+                    if each_center not in record_center_set:
+                        record_center_dataset.append(each_center)
+                        record_center_set.add(each_center)
+            for each_center in add_center:
                 if each_center not in record_center_set:
                     record_center_dataset.append(each_center)
                     record_center_set.add(each_center)

app/algorithms/burst_location/leak_simulator.py

@@ -474,6 +474,7 @@ def cal_signature_pipe_multi_pf(
                 float(leak_mag),
                 list(sensor_name),
                 option_values,
+                list(candidate_center),
             ),
         ) as pool:
             for i, (center_name, pressure_array) in enumerate(
@@ -482,6 +483,9 @@ def cal_signature_pipe_multi_pf(
                 pressure_leak.loc[(center_name, slice(None)), :] = pressure_array
                 sys.stdout.write("\r" + "已经完成计算" + str(i + 1) + "个特征中心")
     else:
+        # Pre-insert all mid-nodes so every simulation sees the same topology
+        for center in candidate_center:
+            ensure_mid_node(wn, center)
         for i in range(candidate_center_num):
             temp_prefix = _make_temp_prefix(f"sig_{i}")
             wn, pressure_output = leak_simulation_pipe_dd_multi_pf(
@@ -500,10 +504,17 @@ def cal_signature_pipe_multi_pf(
     return pressure_leak, candidate_center
 
 
-def _signature_worker_init(inp_path, leak_mag, sensor_name, option_values):
+def _signature_worker_init(
+    inp_path, leak_mag, sensor_name, option_values, candidate_centers=None
+):
     global _SIGNATURE_WORKER_DATA
     wn = wntr.network.WaterNetworkModel(inp_path)
     _apply_hydraulic_options(wn, option_values)
+    # Pre-insert ALL mid-nodes so every simulation runs on the same topology,
+    # regardless of which worker handles which task.
+    if candidate_centers is not None:
+        for center in candidate_centers:
+            ensure_mid_node(wn, center)
     _SIGNATURE_WORKER_DATA = {
         "wn": wn,
         "leak_mag": leak_mag,

app/algorithms/burst_location/network_partitioner.py

@@ -20,6 +20,17 @@ def _to_metis_edge_weight(edge_weight):
     return max(1, int(round(weight)))
 
 
+def _dedupe_preserve_order(items):
+    seen = set()
+    output = []
+    for item in items:
+        if item in seen:
+            continue
+        seen.add(item)
+        output.append(item)
+    return output
+
+
 def pick_center_pipe(node_x, node_y, candidate_pipe, pipe_start_node, pipe_end_node):
     candidate_pipe_list = list(candidate_pipe)
     start_nodes = pipe_start_node[candidate_pipe_list]
@@ -34,14 +45,52 @@ def pick_center_pipe(node_x, node_y, candidate_pipe, pipe_start_node, pipe_end_n
     return candidate_pipe_list[center_idx]
 
 
+def pick_max_diameter_pipe(candidate_pipe, pipe_diameter):
+    candidate_pipe_list = list(candidate_pipe)
+    diameters = pd.to_numeric(
+        pipe_diameter.reindex(candidate_pipe_list), errors="coerce"
+    ).dropna()
+    if len(diameters) != len(candidate_pipe_list):
+        missing = sorted(set(candidate_pipe_list) - set(diameters.index))
+        preview = ", ".join(map(str, missing[:10]))
+        raise ValueError(f"Missing or invalid diameter for pipes: {preview}")
+
+    max_diameter = float(diameters.max())
+    max_diameter_pipes = sorted(
+        [pipe for pipe, diameter in diameters.items() if float(diameter) == max_diameter],
+        key=str,
+    )
+    return max_diameter_pipes[0]
+
+
+def pick_dual_center_pipes(
+    node_x, node_y, candidate_pipe, pipe_start_node, pipe_end_node, pipe_diameter
+):
+    geometric_center = pick_center_pipe(
+        node_x, node_y, candidate_pipe, pipe_start_node, pipe_end_node
+    )
+    diameter_center = pick_max_diameter_pipe(candidate_pipe, pipe_diameter)
+    return _dedupe_preserve_order([geometric_center, diameter_center])
+
+
 def find_new_center_pipe(
-    node_x, node_y, candidate_pipe, pipe_start_node, pipe_end_node, record_center
+    node_x,
+    node_y,
+    candidate_pipe,
+    pipe_start_node,
+    pipe_end_node,
+    pipe_diameter,
+    record_center,
 ):
     new_candidate_pipe = sorted(set(candidate_pipe) - set(record_center))
     if new_candidate_pipe == []:
         new_candidate_pipe = candidate_pipe
     center_t = pick_center_pipe(
-        node_x, node_y, new_candidate_pipe, pipe_start_node, pipe_end_node
+        node_x,
+        node_y,
+        new_candidate_pipe,
+        pipe_start_node,
+        pipe_end_node,
     )
     return center_t
 
@@ -66,6 +115,7 @@ def metis_grouping_pipe_weight(
     node_pipe_dic,
     all_node_series,
     couple_node_length,
+    pipe_diameter,
 ):
     all_node_iter_series_new = all_node_series[all_node_iter]
     all_node_iter_series_new = all_node_iter_series_new.sort_values(ascending=True)
@@ -82,10 +132,12 @@ def metis_grouping_pipe_weight(
     correspond_dic = {}
     count_node = 0
     w = []
-    for node_name, neighbors in G1.adjacency():
+    for node_name in all_node_iter_new:
+        neighbors = G1[node_name]
         w_temp = []
         n_t = [node_dict[node_name]]
-        for neighbor_name, edge_data in neighbors.items():
+        for neighbor_name in sorted(neighbors.keys()):
+            edge_data = neighbors[neighbor_name]
             edge_key = f"{node_name},{neighbor_name}"
             reverse_edge_key = f"{neighbor_name},{node_name}"
             if edge_key in couple_node_length:
@@ -122,14 +174,14 @@ def metis_grouping_pipe_weight(
         w_f = w_f + w_new[i]
 
     # (edgecuts, parts) = pymetis.part_graph(nparts=group_num, adjacency=adjacency_list_new)
-    # metis_options = pymetis.Options()
-    # metis_options.seed = 42
+    metis_options = pymetis.Options()
+    metis_options.seed = 42
     (edgecuts, parts) = pymetis.part_graph(
         nparts=group_num,
         adjncy=final_adjacency_list,
         xadj=xadj,
         eweights=w_f,
-        # options=metis_options,
+        options=metis_options,
     )
     # (edgecuts, parts) = pymetis.part_graph(nparts=group_num, adjacency=adjacency_list_new)
     candidate_group_list = [[] * 1 for i in range(group_num)]
@@ -145,6 +197,7 @@ def metis_grouping_pipe_weight(
 
     new_center = []
     new_group = []
+    new_center_candidates = []
     new_all_node = []
     candidate_pipe_set = set(candidate_pipe_input)
     all_grouped_pipe = []
@@ -158,7 +211,7 @@ def metis_grouping_pipe_weight(
             # 求交集
             nodeset = G_sub.nodes()
             pipeset_set = set(cal_area_node_linked_pipe(nodeset, node_pipe_dic))
-            candidate_pipe = list(pipeset_set.intersection(candidate_pipe_set))
+            candidate_pipe = sorted(pipeset_set.intersection(candidate_pipe_set))
 
             # 判断集合是否保留
             if len(candidate_pipe) > 0:
@@ -170,21 +223,30 @@ def metis_grouping_pipe_weight(
                     pipe_start_node_all,
                     pipe_end_node_all,
                 )
+                center_candidates_t = pick_dual_center_pipes(
+                    node_x,
+                    node_y,
+                    candidate_pipe,
+                    pipe_start_node_all,
+                    pipe_end_node_all,
+                    pipe_diameter,
+                )
 
                 # 更新
                 new_center.append(center_t)
+                new_center_candidates.append(center_candidates_t)
                 new_group.append(candidate_pipe)
                 new_all_node.append(nodeset)
                 all_grouped_pipe = all_grouped_pipe + candidate_pipe
         else:
-            for c in sub_graphs:
+            for c in sorted(sub_graphs, key=lambda c: min(c)):
                 G_temp = G0.subgraph(c)
                 nodeset = G_temp.nodes()
                 pipeset = cal_area_node_linked_pipe(nodeset, node_pipe_dic)
                 pipeset_set = set(pipeset)
 
                 # 求交集
-                candidate_pipe = list(pipeset_set.intersection(candidate_pipe_set))
+                candidate_pipe = sorted(pipeset_set.intersection(candidate_pipe_set))
                 # print(len(candidate_node))
                 # 判断集合是否保留
                 if len(candidate_pipe) > 0:
@@ -196,8 +258,17 @@ def metis_grouping_pipe_weight(
                         pipe_start_node_all,
                         pipe_end_node_all,
                     )
+                    center_candidates_t = pick_dual_center_pipes(
+                        node_x,
+                        node_y,
+                        candidate_pipe,
+                        pipe_start_node_all,
+                        pipe_end_node_all,
+                        pipe_diameter,
+                    )
                     # 更新
                     new_center.append(center_t)
+                    new_center_candidates.append(center_candidates_t)
                     new_group.append(candidate_pipe)
                     new_all_node.append(nodeset)
                     all_grouped_pipe = all_grouped_pipe + candidate_pipe
@@ -217,8 +288,12 @@ def metis_grouping_pipe_weight(
                     each_group,
                     pipe_start_node_all,
                     pipe_end_node_all,
+                    pipe_diameter,
                     record_center,
                 )
+                new_center_candidates[c_g] = _dedupe_preserve_order(
+                    [new_center[c_g]] + list(new_center_candidates[c_g])
+                )
             record_center.append(new_center[c_g])
         c_g += 1
 
@@ -227,7 +302,7 @@ def metis_grouping_pipe_weight(
     #     node_x, node_y,
     #     pipe_start_node_all, pipe_end_node_all
     # )
-    return new_center, new_group, new_all_node
+    return new_center, new_group, new_all_node, new_center_candidates
 
 
 def visualize_metis_partition(

app/algorithms/burst_location/similarity_calculator.py

@@ -44,7 +44,7 @@ def cal_similarity_simple_return_dd(
             similarity_dis = 0
         else:"""
         none_flag = 0
-        sensor_for_cos = list(
+        sensor_for_cos = sorted(
             set(dpressure_s.index).intersection(set(act_dpressure.index))
         )
         """if len(dpressure_s) ==0 or len(dpressure) ==0:
@@ -74,7 +74,7 @@ def cal_similarity_simple_return_dd(
             none_flag = 1
         else:"""
         none_flag = 0
-        important_sensor = list(
+        important_sensor = sorted(
             set(important_sensor).intersection(set(act_dpressure.index))
         )
         part_dpressure = dpressure_s[important_sensor] - dpressure[important_sensor]
@@ -107,7 +107,7 @@ def cal_similarity_simple_return_dd(
 
     elif similarity_mode == "CAD_new_gy" or similarity_mode == "CDF":
         # cos
-        sensor_for_cos = list(
+        sensor_for_cos = sorted(
             set(dpressure_s.index).intersection(set(act_dpressure.index))
         )
         if len(sensor_for_cos) == 0 and len(dpressure_s) == 0:
@@ -141,7 +141,7 @@ def cal_similarity_simple_return_dd(
             similarity_cos = s1 / s2
 
         # DIS
-        important_sensor_new = list(
+        important_sensor_new = sorted(
             set(important_sensor).intersection(set(act_dpressure.index))
         )
         if len(important_sensor_new) == 0:
@@ -172,7 +172,7 @@ def cal_similarity_simple_return_dd(
         similarity_cos = 0
         none_flag = 0
         # DIS
-        important_sensor_new = list(
+        important_sensor_new = sorted(
             set(important_sensor).intersection(set(act_dpressure.index))
         )
         if len(important_sensor_new) == 0:
@@ -762,15 +762,24 @@ def update_similarity(leak_candidate_center, similarity, leak_center_dict):
     return similarity_new
 
 
-def extra_judge(similarity):
+def extra_judge(
+    similarity, min_candidates_to_prune: int = 200, std_relax_factor: float = 0.5
+):
+    if len(similarity.index) == 0:
+        return 1.0, similarity
+    if len(similarity.index) < int(min_candidates_to_prune):
+        return 1.0, similarity
+
     mean_similarity = float(similarity.mean())
-    record_sensor = []
-    record_value = []
-    for i in range(len(similarity.index)):
-        if similarity.iloc[i] >= mean_similarity - 1e-10:
-            record_value.append(similarity.iloc[i])
-            record_sensor.append(similarity.index[i])
-    out_put_similarity = pd.Series(record_value, index=record_sensor, dtype=float)
+    std_similarity = float(similarity.std())
+    if not math.isfinite(std_similarity):
+        std_similarity = 0.0
+
+    threshold = mean_similarity - float(std_relax_factor) * std_similarity
+    out_put_similarity = similarity[similarity >= threshold - 1e-10]
+    if len(out_put_similarity.index) == 0:
+        out_put_similarity = similarity.iloc[:1]
+
     cut_ratio = len(out_put_similarity.index) / len(similarity.index)
     return cut_ratio, out_put_similarity