|
|
|
@ -1195,6 +1195,7 @@ def route_eplan_connections_from_payload(doc, payload, options=None, prepared_la
|
|
|
|
"collision_samples": [],
|
|
|
|
"collision_samples": [],
|
|
|
|
"errors": [],
|
|
|
|
"errors": [],
|
|
|
|
"error_samples": [],
|
|
|
|
"error_samples": [],
|
|
|
|
|
|
|
|
"route_status_counts": {},
|
|
|
|
"routes": [],
|
|
|
|
"routes": [],
|
|
|
|
}
|
|
|
|
}
|
|
|
|
if isinstance(prepared_layout, dict):
|
|
|
|
if isinstance(prepared_layout, dict):
|
|
|
|
@ -1202,9 +1203,14 @@ def route_eplan_connections_from_payload(doc, payload, options=None, prepared_la
|
|
|
|
missing_endpoint_uuids = set()
|
|
|
|
missing_endpoint_uuids = set()
|
|
|
|
lane_indexes_by_pair = {}
|
|
|
|
lane_indexes_by_pair = {}
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def add_status(status):
|
|
|
|
|
|
|
|
key = str(status or "").strip() or "Unknown"
|
|
|
|
|
|
|
|
report["route_status_counts"][key] = report["route_status_counts"].get(key, 0) + 1
|
|
|
|
|
|
|
|
|
|
|
|
for item in wires:
|
|
|
|
for item in wires:
|
|
|
|
if not isinstance(item, dict):
|
|
|
|
if not isinstance(item, dict):
|
|
|
|
report["skipped_invalid"] += 1
|
|
|
|
report["skipped_invalid"] += 1
|
|
|
|
|
|
|
|
add_status("Invalid")
|
|
|
|
continue
|
|
|
|
continue
|
|
|
|
start_uuid = _wire_item_value(item, "start_terminal_uuid")
|
|
|
|
start_uuid = _wire_item_value(item, "start_terminal_uuid")
|
|
|
|
end_uuid = _wire_item_value(item, "end_terminal_uuid")
|
|
|
|
end_uuid = _wire_item_value(item, "end_terminal_uuid")
|
|
|
|
@ -1212,6 +1218,7 @@ def route_eplan_connections_from_payload(doc, payload, options=None, prepared_la
|
|
|
|
end_terminal = terminals.get(end_uuid)
|
|
|
|
end_terminal = terminals.get(end_uuid)
|
|
|
|
if start_terminal is None or end_terminal is None:
|
|
|
|
if start_terminal is None or end_terminal is None:
|
|
|
|
report["skipped_missing_terminal"] += 1
|
|
|
|
report["skipped_missing_terminal"] += 1
|
|
|
|
|
|
|
|
add_status("MissingTerminal")
|
|
|
|
for terminal_uuid in (start_uuid, end_uuid):
|
|
|
|
for terminal_uuid in (start_uuid, end_uuid):
|
|
|
|
if terminal_uuid and terminal_uuid not in terminals:
|
|
|
|
if terminal_uuid and terminal_uuid not in terminals:
|
|
|
|
missing_endpoint_uuids.add(terminal_uuid)
|
|
|
|
missing_endpoint_uuids.add(terminal_uuid)
|
|
|
|
@ -1252,6 +1259,7 @@ def route_eplan_connections_from_payload(doc, payload, options=None, prepared_la
|
|
|
|
except Exception as exc:
|
|
|
|
except Exception as exc:
|
|
|
|
error_text = str(exc)
|
|
|
|
error_text = str(exc)
|
|
|
|
report["errors"].append(error_text)
|
|
|
|
report["errors"].append(error_text)
|
|
|
|
|
|
|
|
add_status("Error")
|
|
|
|
if len(report["error_samples"]) < 8:
|
|
|
|
if len(report["error_samples"]) < 8:
|
|
|
|
report["error_samples"].append(
|
|
|
|
report["error_samples"].append(
|
|
|
|
{
|
|
|
|
{
|
|
|
|
@ -1266,6 +1274,7 @@ def route_eplan_connections_from_payload(doc, payload, options=None, prepared_la
|
|
|
|
lane_indexes_by_pair[lane_key] = route_lane_index + 1
|
|
|
|
lane_indexes_by_pair[lane_key] = route_lane_index + 1
|
|
|
|
if result["route_status"] == "CollisionWarning":
|
|
|
|
if result["route_status"] == "CollisionWarning":
|
|
|
|
report["collision_warnings"] += 1
|
|
|
|
report["collision_warnings"] += 1
|
|
|
|
|
|
|
|
add_status(result["route_status"])
|
|
|
|
route_collision_samples = []
|
|
|
|
route_collision_samples = []
|
|
|
|
for collision in list(result.get("collisions", []) or [])[:3]:
|
|
|
|
for collision in list(result.get("collisions", []) or [])[:3]:
|
|
|
|
sample = dict(collision)
|
|
|
|
sample = dict(collision)
|
|
|
|
@ -1311,6 +1320,32 @@ def format_eplan_connection_route_report(report):
|
|
|
|
report.get("collision_warnings", 0),
|
|
|
|
report.get("collision_warnings", 0),
|
|
|
|
report.get("skipped_missing_terminal", 0),
|
|
|
|
report.get("skipped_missing_terminal", 0),
|
|
|
|
)
|
|
|
|
)
|
|
|
|
|
|
|
|
status_counts = report.get("route_status_counts", {})
|
|
|
|
|
|
|
|
if isinstance(status_counts, dict) and status_counts:
|
|
|
|
|
|
|
|
status_labels = {
|
|
|
|
|
|
|
|
"Routed": "正常",
|
|
|
|
|
|
|
|
"CollisionWarning": "碰撞告警",
|
|
|
|
|
|
|
|
"Error": "错误",
|
|
|
|
|
|
|
|
"MissingTerminal": "缺失端子",
|
|
|
|
|
|
|
|
"Invalid": "无效任务",
|
|
|
|
|
|
|
|
}
|
|
|
|
|
|
|
|
def status_count_value(value):
|
|
|
|
|
|
|
|
try:
|
|
|
|
|
|
|
|
return int(value or 0)
|
|
|
|
|
|
|
|
except Exception:
|
|
|
|
|
|
|
|
return 0
|
|
|
|
|
|
|
|
status_parts = []
|
|
|
|
|
|
|
|
for key in ("Routed", "CollisionWarning", "Error", "MissingTerminal", "Invalid"):
|
|
|
|
|
|
|
|
value = status_count_value(status_counts.get(key, 0))
|
|
|
|
|
|
|
|
if value > 0:
|
|
|
|
|
|
|
|
status_parts.append("{0} {1} 条".format(status_labels[key], value))
|
|
|
|
|
|
|
|
for key, value in sorted(status_counts.items()):
|
|
|
|
|
|
|
|
value = status_count_value(value)
|
|
|
|
|
|
|
|
if key in status_labels or value <= 0:
|
|
|
|
|
|
|
|
continue
|
|
|
|
|
|
|
|
status_parts.append("{0} {1} 条".format(key, value))
|
|
|
|
|
|
|
|
if status_parts:
|
|
|
|
|
|
|
|
message += "\n结果状态:{0}。".format(",".join(status_parts))
|
|
|
|
prepared_layout = report.get("prepared_layout")
|
|
|
|
prepared_layout = report.get("prepared_layout")
|
|
|
|
if isinstance(prepared_layout, dict):
|
|
|
|
if isinstance(prepared_layout, dict):
|
|
|
|
message += "\n布线布局空间:线槽路径 {0} 条,布线面 {1} 条,端子接入 {2} 条。".format(
|
|
|
|
message += "\n布线布局空间:线槽路径 {0} 条,布线面 {1} 条,端子接入 {2} 条。".format(
|
|
|
|
|