1634 lines
47 KiB
Python
1634 lines
47 KiB
Python
from __future__ import annotations
|
|
|
|
import base64
|
|
import json
|
|
import math
|
|
import os
|
|
import urllib.error
|
|
import urllib.request
|
|
import zlib
|
|
from pathlib import Path
|
|
from typing import Iterable
|
|
from urllib.parse import unquote, urljoin
|
|
from xml.etree import ElementTree as ET
|
|
|
|
from fastapi import FastAPI, HTTPException, Request, Response
|
|
from fastapi.responses import HTMLResponse
|
|
from pydantic import BaseModel
|
|
|
|
|
|
app = FastAPI(title="System Simulation App")
|
|
INDEX_TEMPLATE = Path(__file__).parent / "static" / "index.html"
|
|
DRAWIO_UPSTREAM_URL = (
|
|
os.getenv("DRAWIO_UPSTREAM_URL")
|
|
or os.getenv("DRAWIO_BASE_URL")
|
|
or "http://127.0.0.1:8081/"
|
|
)
|
|
DRAWIO_APP_SCRIPT_PATH = "js/app.min.js"
|
|
GROUP_PROTECTION_STYLE = (
|
|
"editable=0;resizable=0;rotatable=0;movable=1;deletable=1;connectable=0;"
|
|
)
|
|
CHILD_PROTECTION_STYLE = (
|
|
"part=1;pointerEvents=0;editable=0;movable=0;resizable=0;rotatable=0;"
|
|
"deletable=0;connectable=0;"
|
|
)
|
|
PORT_PROTECTION_STYLE = (
|
|
"editable=0;movable=0;resizable=0;rotatable=0;deletable=0;connectable=1;"
|
|
)
|
|
|
|
|
|
def library_cell(
|
|
cell_id: str,
|
|
style: str,
|
|
x: float,
|
|
y: float,
|
|
width: float,
|
|
height: float,
|
|
value: str = "",
|
|
) -> dict[str, str]:
|
|
return {
|
|
"id": cell_id,
|
|
"value": value,
|
|
"style": style,
|
|
"x": f"{x:g}",
|
|
"y": f"{y:g}",
|
|
"width": f"{width:g}",
|
|
"height": f"{height:g}",
|
|
}
|
|
|
|
|
|
def component_library_entry(
|
|
*,
|
|
title: str,
|
|
width: int,
|
|
height: int,
|
|
component_id: str,
|
|
label: str,
|
|
component_type: str,
|
|
model_type: str,
|
|
ports: str,
|
|
parameters: dict[str, object],
|
|
cells: list[dict[str, str]],
|
|
) -> dict[str, object]:
|
|
attributes = {
|
|
"id": component_id,
|
|
"label": label,
|
|
"componentType": component_type,
|
|
"modelType": model_type,
|
|
"ports": ports,
|
|
}
|
|
attributes.update({key: str(value) for key, value in parameters.items()})
|
|
|
|
graph = ET.Element("mxGraphModel")
|
|
root = ET.SubElement(graph, "root")
|
|
ET.SubElement(root, "mxCell", {"id": "0"})
|
|
ET.SubElement(root, "mxCell", {"id": "1", "parent": "0"})
|
|
|
|
component = ET.SubElement(root, "object", attributes)
|
|
group_cell = ET.SubElement(
|
|
component,
|
|
"mxCell",
|
|
{
|
|
"style": (
|
|
"group;html=1;whiteSpace=wrap;container=1;recursiveResize=0;"
|
|
"collapsible=0;labelPosition=center;verticalLabelPosition=bottom;"
|
|
"align=center;verticalAlign=top;spacingTop=4;fontSize=12;"
|
|
f"{GROUP_PROTECTION_STYLE}"
|
|
),
|
|
"vertex": "1",
|
|
"connectable": "0",
|
|
"parent": "1",
|
|
},
|
|
)
|
|
ET.SubElement(
|
|
group_cell,
|
|
"mxGeometry",
|
|
{"width": str(width), "height": str(height), "as": "geometry"},
|
|
)
|
|
|
|
for cell in cells:
|
|
cell_id = f"{component_id}_{cell['id']}"
|
|
is_port = cell["id"].startswith("port_")
|
|
|
|
if is_port:
|
|
cell_parent = ET.SubElement(
|
|
root,
|
|
"object",
|
|
{
|
|
"id": cell_id,
|
|
"label": cell["value"],
|
|
"portName": cell["id"],
|
|
"portType": "fluid",
|
|
},
|
|
)
|
|
style = f"{cell['style']}{PORT_PROTECTION_STYLE}"
|
|
connectable = "1"
|
|
else:
|
|
cell_parent = root
|
|
style = f"{cell['style']}{CHILD_PROTECTION_STYLE}"
|
|
connectable = "0"
|
|
|
|
cell_attributes = {
|
|
"value": cell["value"],
|
|
"style": style,
|
|
"vertex": "1",
|
|
"connectable": connectable,
|
|
"parent": component_id,
|
|
}
|
|
if not is_port:
|
|
cell_attributes["id"] = cell_id
|
|
|
|
child = ET.SubElement(cell_parent, "mxCell", cell_attributes)
|
|
ET.SubElement(
|
|
child,
|
|
"mxGeometry",
|
|
{
|
|
"x": cell["x"],
|
|
"y": cell["y"],
|
|
"width": cell["width"],
|
|
"height": cell["height"],
|
|
"as": "geometry",
|
|
},
|
|
)
|
|
|
|
return {
|
|
"title": title,
|
|
"w": width,
|
|
"h": height,
|
|
"xml": ET.tostring(graph, encoding="unicode"),
|
|
}
|
|
|
|
|
|
STROKE_DARK = "strokeColor=#3d4752;strokeWidth=2;"
|
|
PORT_STYLE = "ellipse;whiteSpace=wrap;html=1;fillColor=#ffffff;strokeColor=#3d4752;strokeWidth=2;"
|
|
LABEL_STYLE = (
|
|
"text;html=1;strokeColor=none;fillColor=none;align=center;"
|
|
"verticalAlign=middle;whiteSpace=wrap;fontSize=12;fontStyle=1;"
|
|
)
|
|
|
|
COMPONENT_SCHEMAS = {
|
|
"cylinder": {
|
|
"label": "\u6c14\u74f6",
|
|
"parameters": {
|
|
"volume": {
|
|
"label": "\u5bb9\u79ef",
|
|
"type": "float",
|
|
"unit": "m3",
|
|
"default": 0.01,
|
|
"min": 0,
|
|
},
|
|
"p0": {
|
|
"label": "\u521d\u59cb\u538b\u529b",
|
|
"type": "float",
|
|
"unit": "Pa",
|
|
"default": 35000000,
|
|
"min": 0,
|
|
},
|
|
"T0": {
|
|
"label": "\u521d\u59cb\u6e29\u5ea6",
|
|
"type": "float",
|
|
"unit": "K",
|
|
"default": 300,
|
|
"min": 0,
|
|
},
|
|
},
|
|
},
|
|
"tank": {
|
|
"label": "\u8d2e\u7bb1",
|
|
"parameters": {
|
|
"volume": {
|
|
"label": "\u5bb9\u79ef",
|
|
"type": "float",
|
|
"unit": "m3",
|
|
"default": 0.1,
|
|
"min": 0,
|
|
},
|
|
"p0": {
|
|
"label": "\u521d\u59cb\u538b\u529b",
|
|
"type": "float",
|
|
"unit": "Pa",
|
|
"default": 100000,
|
|
"min": 0,
|
|
},
|
|
"T0": {
|
|
"label": "\u521d\u59cb\u6e29\u5ea6",
|
|
"type": "float",
|
|
"unit": "K",
|
|
"default": 300,
|
|
"min": 0,
|
|
},
|
|
},
|
|
},
|
|
"pipe": {
|
|
"label": "\u7ba1\u6bb5",
|
|
"parameters": {
|
|
"length": {
|
|
"label": "\u957f\u5ea6",
|
|
"type": "float",
|
|
"unit": "m",
|
|
"default": 5.0,
|
|
"min": 0,
|
|
},
|
|
"diameter": {
|
|
"label": "\u76f4\u5f84",
|
|
"type": "float",
|
|
"unit": "m",
|
|
"default": 0.02,
|
|
"min": 0,
|
|
},
|
|
"lambda_darcy": {
|
|
"label": "\u6469\u963b\u7cfb\u6570",
|
|
"type": "float",
|
|
"default": 0.02,
|
|
"min": 0,
|
|
},
|
|
"p0": {
|
|
"label": "\u521d\u59cb\u538b\u529b",
|
|
"type": "float",
|
|
"unit": "Pa",
|
|
"default": 100000,
|
|
"min": 0,
|
|
},
|
|
"T0": {
|
|
"label": "\u521d\u59cb\u6e29\u5ea6",
|
|
"type": "float",
|
|
"unit": "K",
|
|
"default": 300,
|
|
"min": 0,
|
|
},
|
|
},
|
|
},
|
|
"orifice": {
|
|
"label": "\u5b54\u677f/\u9600\u95e8",
|
|
"parameters": {
|
|
"K": {
|
|
"label": "\u6d41\u91cf\u7cfb\u6570",
|
|
"type": "float",
|
|
"default": 0.00001,
|
|
"min": 0,
|
|
},
|
|
"opening": {
|
|
"label": "\u5f00\u5ea6",
|
|
"type": "float",
|
|
"default": 1.0,
|
|
"min": 0,
|
|
"max": 1,
|
|
},
|
|
},
|
|
},
|
|
"tee": {
|
|
"label": "\u4e09\u901a",
|
|
"parameters": {},
|
|
},
|
|
}
|
|
|
|
|
|
COMPONENT_LIBRARY = [
|
|
component_library_entry(
|
|
title="\u6c14\u74f6",
|
|
width=90,
|
|
height=120,
|
|
component_id="cylinder",
|
|
label="\u6c14\u74f6",
|
|
component_type="cylinder",
|
|
model_type="cylinder",
|
|
ports="port_b",
|
|
parameters={"volume": 0.01, "p0": 35000000, "T0": 300},
|
|
cells=[
|
|
library_cell(
|
|
"cap",
|
|
"rounded=1;whiteSpace=wrap;html=1;arcSize=35;fillColor=#e5e7eb;"
|
|
f"{STROKE_DARK}",
|
|
31,
|
|
2,
|
|
28,
|
|
10,
|
|
),
|
|
library_cell(
|
|
"neck",
|
|
f"rounded=0;whiteSpace=wrap;html=1;fillColor=#9ca3af;{STROKE_DARK}",
|
|
37,
|
|
11,
|
|
16,
|
|
10,
|
|
),
|
|
library_cell(
|
|
"body",
|
|
"rounded=1;whiteSpace=wrap;html=1;arcSize=28;fillColor=#dae8fc;"
|
|
"strokeColor=#2f5597;strokeWidth=2;",
|
|
24,
|
|
18,
|
|
42,
|
|
80,
|
|
),
|
|
library_cell(
|
|
"shoulder",
|
|
"ellipse;whiteSpace=wrap;html=1;fillColor=none;"
|
|
"strokeColor=#2f5597;strokeWidth=2;",
|
|
29,
|
|
18,
|
|
32,
|
|
16,
|
|
),
|
|
library_cell(
|
|
"foot",
|
|
f"rounded=0;whiteSpace=wrap;html=1;fillColor=#9ca3af;{STROKE_DARK}",
|
|
28,
|
|
98,
|
|
34,
|
|
6,
|
|
),
|
|
library_cell("port_b", PORT_STYLE, 67, 50, 10, 10),
|
|
library_cell("label", LABEL_STYLE, 14, 104, 62, 16, "\u6c14\u74f6"),
|
|
],
|
|
),
|
|
component_library_entry(
|
|
title="\u8d2e\u7bb1",
|
|
width=100,
|
|
height=112,
|
|
component_id="tank",
|
|
label="\u8d2e\u7bb1",
|
|
component_type="tank",
|
|
model_type="tank",
|
|
ports="port_a",
|
|
parameters={"volume": 0.1, "p0": 100000, "T0": 300},
|
|
cells=[
|
|
library_cell(
|
|
"vessel",
|
|
"shape=cylinder;whiteSpace=wrap;html=1;boundedLbl=1;"
|
|
"backgroundOutline=1;size=14;fillColor=#d5e8d4;"
|
|
"strokeColor=#4f7f45;strokeWidth=2;",
|
|
18,
|
|
8,
|
|
64,
|
|
72,
|
|
),
|
|
library_cell(
|
|
"level",
|
|
"rounded=0;whiteSpace=wrap;html=1;fillColor=#ffffff;"
|
|
"strokeColor=#4f7f45;strokeWidth=1;dashed=1;",
|
|
27,
|
|
43,
|
|
46,
|
|
1,
|
|
),
|
|
library_cell(
|
|
"leg_left",
|
|
f"rounded=0;whiteSpace=wrap;html=1;fillColor=#9ca3af;{STROKE_DARK}",
|
|
30,
|
|
80,
|
|
8,
|
|
16,
|
|
),
|
|
library_cell(
|
|
"leg_right",
|
|
f"rounded=0;whiteSpace=wrap;html=1;fillColor=#9ca3af;{STROKE_DARK}",
|
|
62,
|
|
80,
|
|
8,
|
|
16,
|
|
),
|
|
library_cell("port_a", PORT_STYLE, 84, 39, 10, 10),
|
|
library_cell("label", LABEL_STYLE, 18, 94, 64, 16, "\u8d2e\u7bb1"),
|
|
],
|
|
),
|
|
component_library_entry(
|
|
title="\u7ba1\u6bb5",
|
|
width=122,
|
|
height=52,
|
|
component_id="pipe",
|
|
label="\u7ba1\u6bb5",
|
|
component_type="pipe",
|
|
model_type="pipe",
|
|
ports="port_a,port_b",
|
|
parameters={
|
|
"length": 5.0,
|
|
"diameter": 0.02,
|
|
"lambda_darcy": 0.02,
|
|
"p0": 100000,
|
|
"T0": 300,
|
|
},
|
|
cells=[
|
|
library_cell(
|
|
"body",
|
|
"rounded=1;whiteSpace=wrap;html=1;arcSize=12;fillColor=#f5f5f5;"
|
|
"strokeColor=#6b7280;strokeWidth=2;",
|
|
14,
|
|
17,
|
|
94,
|
|
16,
|
|
),
|
|
library_cell(
|
|
"centerline",
|
|
"rounded=0;whiteSpace=wrap;html=1;fillColor=#6b7280;strokeColor=none;",
|
|
20,
|
|
24,
|
|
82,
|
|
2,
|
|
),
|
|
library_cell(
|
|
"flange_a",
|
|
f"rounded=0;whiteSpace=wrap;html=1;fillColor=#d1d5db;{STROKE_DARK}",
|
|
7,
|
|
12,
|
|
8,
|
|
26,
|
|
),
|
|
library_cell(
|
|
"flange_b",
|
|
f"rounded=0;whiteSpace=wrap;html=1;fillColor=#d1d5db;{STROKE_DARK}",
|
|
107,
|
|
12,
|
|
8,
|
|
26,
|
|
),
|
|
library_cell("port_a", PORT_STYLE, 0, 20, 10, 10),
|
|
library_cell("port_b", PORT_STYLE, 112, 20, 10, 10),
|
|
library_cell("label", LABEL_STYLE, 29, 36, 64, 14, "\u7ba1\u6bb5"),
|
|
],
|
|
),
|
|
component_library_entry(
|
|
title="\u5b54\u677f/\u9600\u95e8",
|
|
width=96,
|
|
height=76,
|
|
component_id="orifice",
|
|
label="\u5b54\u677f/\u9600\u95e8",
|
|
component_type="valve",
|
|
model_type="orifice",
|
|
ports="port_a,port_b",
|
|
parameters={"K": 0.00001, "opening": 1.0},
|
|
cells=[
|
|
library_cell(
|
|
"line_a",
|
|
"rounded=0;whiteSpace=wrap;html=1;fillColor=#6b7280;strokeColor=none;",
|
|
0,
|
|
35,
|
|
14,
|
|
4,
|
|
),
|
|
library_cell(
|
|
"tri_left",
|
|
"shape=triangle;direction=east;whiteSpace=wrap;html=1;"
|
|
"fillColor=#ffe6cc;strokeColor=#d79b00;strokeWidth=2;",
|
|
13,
|
|
22,
|
|
32,
|
|
30,
|
|
),
|
|
library_cell(
|
|
"tri_right",
|
|
"shape=triangle;direction=west;whiteSpace=wrap;html=1;"
|
|
"fillColor=#ffe6cc;strokeColor=#d79b00;strokeWidth=2;",
|
|
51,
|
|
22,
|
|
32,
|
|
30,
|
|
),
|
|
library_cell(
|
|
"plate",
|
|
"rounded=0;whiteSpace=wrap;html=1;fillColor=#d79b00;"
|
|
"strokeColor=#a86f00;strokeWidth=1;",
|
|
46,
|
|
16,
|
|
4,
|
|
42,
|
|
),
|
|
library_cell(
|
|
"stem",
|
|
"rounded=0;whiteSpace=wrap;html=1;fillColor=#6b7280;strokeColor=none;",
|
|
47,
|
|
7,
|
|
2,
|
|
10,
|
|
),
|
|
library_cell(
|
|
"handle",
|
|
f"rounded=1;whiteSpace=wrap;html=1;arcSize=30;fillColor=#f3f4f6;{STROKE_DARK}",
|
|
38,
|
|
0,
|
|
20,
|
|
8,
|
|
),
|
|
library_cell(
|
|
"line_b",
|
|
"rounded=0;whiteSpace=wrap;html=1;fillColor=#6b7280;strokeColor=none;",
|
|
82,
|
|
35,
|
|
14,
|
|
4,
|
|
),
|
|
library_cell("port_a", PORT_STYLE, 0, 32, 10, 10),
|
|
library_cell("port_b", PORT_STYLE, 86, 32, 10, 10),
|
|
library_cell(
|
|
"label",
|
|
LABEL_STYLE,
|
|
18,
|
|
58,
|
|
60,
|
|
16,
|
|
"\u5b54\u677f/\u9600\u95e8",
|
|
),
|
|
],
|
|
),
|
|
component_library_entry(
|
|
title="\u4e09\u901a",
|
|
width=100,
|
|
height=96,
|
|
component_id="tee",
|
|
label="\u4e09\u901a",
|
|
component_type="tee",
|
|
model_type="tee",
|
|
ports="port_in,port_out1,port_out2",
|
|
parameters={},
|
|
cells=[
|
|
library_cell(
|
|
"horizontal",
|
|
"rounded=1;whiteSpace=wrap;html=1;arcSize=12;fillColor=#f5f5f5;"
|
|
"strokeColor=#6b7280;strokeWidth=2;",
|
|
13,
|
|
40,
|
|
74,
|
|
16,
|
|
),
|
|
library_cell(
|
|
"vertical",
|
|
"rounded=1;whiteSpace=wrap;html=1;arcSize=12;fillColor=#f5f5f5;"
|
|
"strokeColor=#6b7280;strokeWidth=2;",
|
|
42,
|
|
10,
|
|
16,
|
|
62,
|
|
),
|
|
library_cell(
|
|
"junction",
|
|
"ellipse;whiteSpace=wrap;html=1;fillColor=#ffffff;"
|
|
"strokeColor=#6b7280;strokeWidth=2;",
|
|
38,
|
|
36,
|
|
24,
|
|
24,
|
|
),
|
|
library_cell("port_out1", PORT_STYLE, 90, 43, 10, 10),
|
|
library_cell("port_out2", PORT_STYLE, 0, 43, 10, 10),
|
|
library_cell("port_in", PORT_STYLE, 45, 0, 10, 10),
|
|
library_cell("label", LABEL_STYLE, 24, 74, 52, 16, "\u4e09\u901a"),
|
|
],
|
|
),
|
|
]
|
|
|
|
|
|
DRAWIO_SELECTION_PLUGIN = r"""
|
|
(function() {
|
|
function postPluginStatus(status, extra) {
|
|
try {
|
|
var payload = {
|
|
event: "systemSimulationPluginStatus",
|
|
status: status,
|
|
};
|
|
|
|
if (extra) {
|
|
for (var key in extra) {
|
|
payload[key] = extra[key];
|
|
}
|
|
}
|
|
|
|
window.parent.postMessage(JSON.stringify(payload), "*");
|
|
} catch (error) {
|
|
if (window.console) {
|
|
console.warn("System simulation plugin status failed", error);
|
|
}
|
|
}
|
|
}
|
|
|
|
function bridgeCellId(cell) {
|
|
if (!cell) {
|
|
return null;
|
|
}
|
|
if (cell.id != null) {
|
|
return String(cell.id);
|
|
}
|
|
if (typeof cell.getId === "function") {
|
|
return String(cell.getId());
|
|
}
|
|
return null;
|
|
}
|
|
|
|
function bridgeParentOf(cell) {
|
|
if (!cell) {
|
|
return null;
|
|
}
|
|
if (typeof cell.getParent === "function") {
|
|
return cell.getParent();
|
|
}
|
|
return cell.parent || null;
|
|
}
|
|
|
|
function bridgeCellAndAncestorIds(cell) {
|
|
var ids = [];
|
|
var seen = {};
|
|
var current = cell;
|
|
var depth = 0;
|
|
|
|
while (current && depth < 10) {
|
|
var id = bridgeCellId(current);
|
|
if (id != null && !seen[id]) {
|
|
seen[id] = true;
|
|
ids.push(id);
|
|
}
|
|
current = bridgeParentOf(current);
|
|
depth += 1;
|
|
}
|
|
|
|
return ids;
|
|
}
|
|
|
|
function postSelectionForGraph(graph, source) {
|
|
try {
|
|
if (!graph || typeof graph.getSelectionCells !== "function") {
|
|
return;
|
|
}
|
|
|
|
var cells = graph.getSelectionCells() || [];
|
|
var ids = [];
|
|
var details = [];
|
|
|
|
for (var i = 0; i < cells.length; i++) {
|
|
var cell = cells[i];
|
|
var id = bridgeCellId(cell);
|
|
var parent = bridgeParentOf(cell);
|
|
var parentId = bridgeCellId(parent);
|
|
var ancestorIds = bridgeCellAndAncestorIds(cell);
|
|
|
|
for (var j = 0; j < ancestorIds.length; j++) {
|
|
ids.push(ancestorIds[j]);
|
|
}
|
|
details.push({
|
|
id: id,
|
|
parentId: parentId,
|
|
ancestorIds: ancestorIds,
|
|
});
|
|
}
|
|
|
|
window.parent.postMessage(JSON.stringify({
|
|
event: "systemSimulationSelection",
|
|
source: source || "mxgraph-prototype",
|
|
ids: ids,
|
|
cells: details,
|
|
}), "*");
|
|
} catch (error) {
|
|
if (window.console) {
|
|
console.warn("System simulation prototype bridge failed", error);
|
|
}
|
|
}
|
|
}
|
|
|
|
function scheduleSelectionForGraph(graph, source) {
|
|
window.setTimeout(function() {
|
|
postSelectionForGraph(graph, source);
|
|
}, 0);
|
|
}
|
|
|
|
function wrapPrototypeMethod(prototype, methodName, graphFromContext) {
|
|
var original = prototype && prototype[methodName];
|
|
if (
|
|
typeof original !== "function" ||
|
|
original.__systemSimulationSelectionWrapped
|
|
) {
|
|
return false;
|
|
}
|
|
|
|
var wrapped = function() {
|
|
var result = original.apply(this, arguments);
|
|
var graph = graphFromContext ? graphFromContext(this) : this;
|
|
scheduleSelectionForGraph(graph, "prototype:" + methodName);
|
|
return result;
|
|
};
|
|
wrapped.__systemSimulationSelectionWrapped = true;
|
|
prototype[methodName] = wrapped;
|
|
return true;
|
|
}
|
|
|
|
function installPrototypeSelectionBridge() {
|
|
if (window.__systemSimulationPrototypeSelectionBridgeInstalled) {
|
|
return true;
|
|
}
|
|
|
|
var patched = false;
|
|
var graphPrototype =
|
|
window.mxGraph && window.mxGraph.prototype ? window.mxGraph.prototype : null;
|
|
if (graphPrototype) {
|
|
[
|
|
"setSelectionCell",
|
|
"setSelectionCells",
|
|
"addSelectionCell",
|
|
"addSelectionCells",
|
|
"removeSelectionCell",
|
|
"removeSelectionCells",
|
|
"clearSelection",
|
|
].forEach(function(methodName) {
|
|
patched = wrapPrototypeMethod(graphPrototype, methodName) || patched;
|
|
});
|
|
}
|
|
|
|
var selectionPrototype =
|
|
window.mxGraphSelectionModel && window.mxGraphSelectionModel.prototype
|
|
? window.mxGraphSelectionModel.prototype
|
|
: null;
|
|
if (selectionPrototype) {
|
|
patched =
|
|
wrapPrototypeMethod(selectionPrototype, "changeSelection", function(model) {
|
|
return model.graph || null;
|
|
}) || patched;
|
|
}
|
|
|
|
if (patched) {
|
|
window.__systemSimulationPrototypeSelectionBridgeInstalled = true;
|
|
postPluginStatus("prototype-ready");
|
|
return true;
|
|
}
|
|
|
|
return false;
|
|
}
|
|
|
|
function registerSelectionPlugin() {
|
|
if (window.__systemSimulationSelectionPluginRegistered) {
|
|
return true;
|
|
}
|
|
|
|
if (!window.Draw || typeof window.Draw.loadPlugin !== "function") {
|
|
return false;
|
|
}
|
|
|
|
window.__systemSimulationSelectionPluginRegistered = true;
|
|
window.Draw.loadPlugin(function(ui) {
|
|
var graph = ui && ui.editor ? ui.editor.graph : null;
|
|
if (!graph) {
|
|
postPluginStatus("graph-missing", {
|
|
hasUi: !!ui,
|
|
hasEditor: !!(ui && ui.editor),
|
|
});
|
|
return;
|
|
}
|
|
|
|
function cellId(cell) {
|
|
if (!cell) {
|
|
return null;
|
|
}
|
|
if (cell.id != null) {
|
|
return String(cell.id);
|
|
}
|
|
if (typeof cell.getId === "function") {
|
|
return String(cell.getId());
|
|
}
|
|
return null;
|
|
}
|
|
|
|
function parentOf(cell) {
|
|
if (!cell) {
|
|
return null;
|
|
}
|
|
if (typeof cell.getParent === "function") {
|
|
return cell.getParent();
|
|
}
|
|
return cell.parent || null;
|
|
}
|
|
|
|
function eventName(mxName, fallback) {
|
|
var mx = window.mxEvent || (typeof mxEvent !== "undefined" ? mxEvent : null);
|
|
return mx && mx[mxName] ? mx[mxName] : fallback;
|
|
}
|
|
|
|
function collectCellAndAncestorIds(cell) {
|
|
var ids = [];
|
|
var seen = {};
|
|
var current = cell;
|
|
var depth = 0;
|
|
|
|
while (current && depth < 10) {
|
|
var id = cellId(current);
|
|
if (id != null && !seen[id]) {
|
|
seen[id] = true;
|
|
ids.push(id);
|
|
}
|
|
current = parentOf(current);
|
|
depth += 1;
|
|
}
|
|
|
|
return ids;
|
|
}
|
|
|
|
function postSelection() {
|
|
try {
|
|
var cells = graph.getSelectionCells ? graph.getSelectionCells() : [];
|
|
var ids = [];
|
|
var details = [];
|
|
|
|
for (var i = 0; i < cells.length; i++) {
|
|
var cell = cells[i];
|
|
var id = cellId(cell);
|
|
var parent = parentOf(cell);
|
|
var parentId = cellId(parent);
|
|
var ancestorIds = collectCellAndAncestorIds(cell);
|
|
|
|
for (var j = 0; j < ancestorIds.length; j++) {
|
|
ids.push(ancestorIds[j]);
|
|
}
|
|
details.push({
|
|
id: id,
|
|
parentId: parentId,
|
|
ancestorIds: ancestorIds,
|
|
});
|
|
}
|
|
|
|
window.parent.postMessage(JSON.stringify({
|
|
event: "systemSimulationSelection",
|
|
ids: ids,
|
|
cells: details,
|
|
}), "*");
|
|
} catch (error) {
|
|
if (window.console) {
|
|
console.warn("System simulation selection plugin failed", error);
|
|
}
|
|
}
|
|
}
|
|
|
|
function schedulePostSelection() {
|
|
window.setTimeout(postSelection, 0);
|
|
}
|
|
|
|
function wrapSelectionMethod(methodName) {
|
|
var original = graph[methodName];
|
|
if (
|
|
typeof original !== "function" ||
|
|
original.__systemSimulationSelectionWrapped
|
|
) {
|
|
return;
|
|
}
|
|
|
|
var wrapped = function() {
|
|
var result = original.apply(this, arguments);
|
|
schedulePostSelection();
|
|
return result;
|
|
};
|
|
wrapped.__systemSimulationSelectionWrapped = true;
|
|
graph[methodName] = wrapped;
|
|
}
|
|
|
|
var selectionModel = graph.getSelectionModel ? graph.getSelectionModel() : null;
|
|
if (selectionModel && typeof selectionModel.addListener === "function") {
|
|
selectionModel.addListener(eventName("CHANGE", "change"), function() {
|
|
schedulePostSelection();
|
|
});
|
|
}
|
|
|
|
if (typeof graph.addListener === "function") {
|
|
graph.addListener(eventName("CLICK", "click"), function() {
|
|
schedulePostSelection();
|
|
});
|
|
graph.addListener(eventName("DOUBLE_CLICK", "doubleClick"), function() {
|
|
schedulePostSelection();
|
|
});
|
|
}
|
|
|
|
[
|
|
"setSelectionCell",
|
|
"setSelectionCells",
|
|
"addSelectionCell",
|
|
"addSelectionCells",
|
|
"removeSelectionCell",
|
|
"removeSelectionCells",
|
|
"clearSelection",
|
|
].forEach(wrapSelectionMethod);
|
|
|
|
if (graph.container && typeof graph.container.addEventListener === "function") {
|
|
graph.container.addEventListener("mouseup", schedulePostSelection, true);
|
|
graph.container.addEventListener("pointerup", schedulePostSelection, true);
|
|
graph.container.addEventListener("keyup", schedulePostSelection, true);
|
|
}
|
|
|
|
window.parent.postMessage(JSON.stringify({
|
|
event: "systemSimulationPluginReady",
|
|
}), "*");
|
|
postPluginStatus("ready");
|
|
schedulePostSelection();
|
|
});
|
|
|
|
return true;
|
|
}
|
|
|
|
if (!installPrototypeSelectionBridge() && !registerSelectionPlugin()) {
|
|
var attempts = 0;
|
|
postPluginStatus("waiting-for-draw-loadplugin", { attempts: attempts });
|
|
var timer = window.setInterval(function() {
|
|
attempts += 1;
|
|
var prototypeReady = installPrototypeSelectionBridge();
|
|
var pluginReady = registerSelectionPlugin();
|
|
|
|
if (pluginReady) {
|
|
postPluginStatus("registered", { attempts: attempts });
|
|
}
|
|
|
|
if (prototypeReady || pluginReady) {
|
|
window.clearInterval(timer);
|
|
} else if (
|
|
attempts === 10 ||
|
|
attempts === 50 ||
|
|
attempts === 100 ||
|
|
attempts === 300 ||
|
|
attempts % 600 === 0
|
|
) {
|
|
postPluginStatus("waiting-for-draw-loadplugin", { attempts: attempts });
|
|
}
|
|
}, 100);
|
|
}
|
|
})();
|
|
"""
|
|
|
|
|
|
class DiagramPayload(BaseModel):
|
|
xml: str
|
|
|
|
|
|
class UpdateComponentParametersPayload(BaseModel):
|
|
xml: str
|
|
component_id: str
|
|
name: str | None = None
|
|
parameters: dict[str, object]
|
|
|
|
|
|
@app.get("/")
|
|
def index(request: Request) -> HTMLResponse:
|
|
html = INDEX_TEMPLATE.read_text(encoding="utf-8")
|
|
drawio_base_url = default_drawio_base_url(request)
|
|
html = html.replace("__DRAWIO_BASE_URL__", json.dumps(drawio_base_url))
|
|
return HTMLResponse(html, headers={"Cache-Control": "no-store"})
|
|
|
|
|
|
@app.get("/api/component-library.drawiolib")
|
|
def component_library() -> Response:
|
|
library = ET.Element("mxlibrary", {"title": "System Components"})
|
|
library.text = json.dumps(COMPONENT_LIBRARY, ensure_ascii=False)
|
|
return Response(
|
|
content=ET.tostring(library, encoding="utf-8"),
|
|
media_type="application/xml",
|
|
headers={
|
|
"Access-Control-Allow-Origin": "*",
|
|
"Cache-Control": "no-store",
|
|
},
|
|
)
|
|
|
|
|
|
@app.get("/api/component-schemas")
|
|
def component_schemas() -> dict[str, object]:
|
|
return COMPONENT_SCHEMAS
|
|
|
|
|
|
@app.get("/api/drawio-selection-plugin.js")
|
|
def drawio_selection_plugin() -> Response:
|
|
return Response(
|
|
content=DRAWIO_SELECTION_PLUGIN,
|
|
media_type="application/javascript",
|
|
headers={"Cache-Control": "no-store"},
|
|
)
|
|
|
|
|
|
@app.post("/api/diagram-components")
|
|
def diagram_components(payload: DiagramPayload) -> dict[str, object]:
|
|
try:
|
|
model = read_graph_model(payload.xml)
|
|
root = get_graph_root(model)
|
|
components = list(read_components(root))
|
|
except ValueError as exc:
|
|
raise HTTPException(status_code=400, detail=str(exc)) from exc
|
|
|
|
return {"components": components}
|
|
|
|
|
|
@app.api_route("/drawio", methods=["GET", "POST", "HEAD", "OPTIONS"])
|
|
@app.api_route("/drawio/{path:path}", methods=["GET", "POST", "HEAD", "OPTIONS"])
|
|
async def drawio_proxy(request: Request, path: str = "") -> Response:
|
|
upstream_url = build_drawio_upstream_url(path, request.url.query)
|
|
body = await request.body()
|
|
normalized_path = normalize_drawio_proxy_path(path)
|
|
headers = {
|
|
key: value
|
|
for key, value in request.headers.items()
|
|
if key.lower()
|
|
not in {
|
|
"host",
|
|
"connection",
|
|
"content-length",
|
|
"accept-encoding",
|
|
"if-modified-since",
|
|
"if-none-match",
|
|
"if-range",
|
|
}
|
|
}
|
|
|
|
upstream_request = urllib.request.Request(
|
|
upstream_url,
|
|
data=body if body else None,
|
|
headers=headers,
|
|
method=request.method,
|
|
)
|
|
|
|
try:
|
|
with urllib.request.urlopen(upstream_request, timeout=30) as upstream_response:
|
|
content = upstream_response.read()
|
|
content = inject_drawio_selection_plugin(
|
|
content,
|
|
upstream_response.headers,
|
|
request.url.query,
|
|
normalized_path,
|
|
)
|
|
response_headers = filter_proxy_headers(upstream_response.headers)
|
|
if is_drawio_injection_target(
|
|
normalized_path,
|
|
upstream_response.headers,
|
|
request.url.query,
|
|
):
|
|
response_headers["Cache-Control"] = "no-store"
|
|
response_headers.pop("ETag", None)
|
|
response_headers.pop("Last-Modified", None)
|
|
return Response(
|
|
content=content,
|
|
status_code=upstream_response.status,
|
|
headers=response_headers,
|
|
)
|
|
except urllib.error.HTTPError as exc:
|
|
return Response(
|
|
content=exc.read(),
|
|
status_code=exc.code,
|
|
headers=filter_proxy_headers(exc.headers),
|
|
)
|
|
except urllib.error.URLError as exc:
|
|
raise HTTPException(
|
|
status_code=502,
|
|
detail=f"Unable to reach draw.io upstream: {exc.reason}",
|
|
) from exc
|
|
|
|
|
|
@app.post("/api/system-xml")
|
|
def export_system_xml(payload: DiagramPayload) -> Response:
|
|
try:
|
|
model = read_graph_model(payload.xml)
|
|
system_xml = build_system_xml(model)
|
|
except ValueError as exc:
|
|
raise HTTPException(status_code=400, detail=str(exc)) from exc
|
|
|
|
return Response(content=system_xml, media_type="application/xml")
|
|
|
|
|
|
@app.post("/api/update-component-parameters")
|
|
def update_component_parameters_endpoint(
|
|
payload: UpdateComponentParametersPayload,
|
|
) -> Response:
|
|
try:
|
|
model = read_graph_model(payload.xml)
|
|
update_component_parameters(
|
|
model,
|
|
payload.component_id,
|
|
payload.name,
|
|
payload.parameters,
|
|
)
|
|
except ValueError as exc:
|
|
raise HTTPException(status_code=400, detail=str(exc)) from exc
|
|
|
|
return Response(
|
|
content=ET.tostring(model, encoding="utf-8"),
|
|
media_type="application/xml",
|
|
)
|
|
|
|
|
|
def default_drawio_base_url(request: Request) -> str:
|
|
hostname = request.url.hostname or "127.0.0.1"
|
|
if ":" in hostname and not hostname.startswith("["):
|
|
hostname = f"[{hostname}]"
|
|
|
|
port = f":{request.url.port}" if request.url.port else ""
|
|
return f"{request.url.scheme}://{hostname}{port}/drawio/"
|
|
|
|
|
|
def build_drawio_upstream_url(path: str, query: str) -> str:
|
|
target = urljoin(DRAWIO_UPSTREAM_URL, path)
|
|
if query:
|
|
target = f"{target}?{query}"
|
|
return target
|
|
|
|
|
|
def normalize_drawio_proxy_path(path: str) -> str:
|
|
return path.replace("\\", "/").lstrip("/")
|
|
|
|
|
|
def filter_proxy_headers(headers) -> dict[str, str]:
|
|
excluded = {
|
|
"connection",
|
|
"content-encoding",
|
|
"content-length",
|
|
"keep-alive",
|
|
"proxy-authenticate",
|
|
"proxy-authorization",
|
|
"te",
|
|
"trailer",
|
|
"transfer-encoding",
|
|
"upgrade",
|
|
}
|
|
return {
|
|
key: value
|
|
for key, value in headers.items()
|
|
if key.lower() not in excluded
|
|
}
|
|
|
|
|
|
def is_drawio_injection_target(path: str, headers, query: str) -> bool:
|
|
content_type = headers.get("Content-Type", "")
|
|
if "text/html" in content_type and "embed=1" in query:
|
|
return True
|
|
return path == DRAWIO_APP_SCRIPT_PATH and is_javascript_content(content_type)
|
|
|
|
|
|
def is_javascript_content(content_type: str) -> bool:
|
|
lowered = content_type.lower()
|
|
return "javascript" in lowered or "ecmascript" in lowered
|
|
|
|
|
|
def inject_drawio_selection_plugin(
|
|
content: bytes,
|
|
headers,
|
|
query: str,
|
|
path: str = "",
|
|
) -> bytes:
|
|
content_type = headers.get("Content-Type", "")
|
|
|
|
if path == DRAWIO_APP_SCRIPT_PATH and is_javascript_content(content_type):
|
|
return append_drawio_selection_plugin_to_app_script(content)
|
|
|
|
if "text/html" not in content_type or "embed=1" not in query:
|
|
return content
|
|
|
|
try:
|
|
html = content.decode("utf-8")
|
|
except UnicodeDecodeError:
|
|
return content
|
|
|
|
plugin_tag = '<script src="/api/drawio-selection-plugin.js"></script>'
|
|
if plugin_tag in html:
|
|
return content
|
|
|
|
main_script = '<script src="js/main.js"></script>'
|
|
if main_script in html:
|
|
html = html.replace(main_script, f"{plugin_tag}\n{main_script}", 1)
|
|
elif "</body>" in html:
|
|
html = html.replace("</body>", f"{plugin_tag}\n</body>", 1)
|
|
else:
|
|
return content
|
|
|
|
return html.encode("utf-8")
|
|
|
|
|
|
def append_drawio_selection_plugin_to_app_script(content: bytes) -> bytes:
|
|
marker = b"systemSimulationSelectionPluginRegistered"
|
|
if marker in content:
|
|
return content
|
|
|
|
return (
|
|
content
|
|
+ b"\n;\n/* SystemSimulationApp selection bridge */\n"
|
|
+ DRAWIO_SELECTION_PLUGIN.encode("utf-8")
|
|
)
|
|
|
|
|
|
def read_graph_model(xml_text: str) -> ET.Element:
|
|
xml_text = xml_text.strip()
|
|
if not xml_text:
|
|
raise ValueError("Diagram XML is empty.")
|
|
|
|
try:
|
|
root = ET.fromstring(xml_text)
|
|
except ET.ParseError as exc:
|
|
raise ValueError("Diagram XML is invalid.") from exc
|
|
|
|
if tag_name(root) == "mxGraphModel":
|
|
return root
|
|
|
|
if tag_name(root) != "mxfile":
|
|
raise ValueError("Expected mxGraphModel or mxfile XML.")
|
|
|
|
diagram = next((child for child in root if tag_name(child) == "diagram"), None)
|
|
if diagram is None:
|
|
raise ValueError("mxfile does not contain a diagram.")
|
|
|
|
embedded_model = next(
|
|
(child for child in diagram if tag_name(child) == "mxGraphModel"),
|
|
None,
|
|
)
|
|
if embedded_model is not None:
|
|
return embedded_model
|
|
|
|
if not diagram.text:
|
|
raise ValueError("diagram node is empty.")
|
|
|
|
return decode_diagram_payload(diagram.text.strip())
|
|
|
|
|
|
def decode_diagram_payload(payload: str) -> ET.Element:
|
|
if payload.startswith("<"):
|
|
return parse_graph_model(payload)
|
|
|
|
if payload.startswith("%3C"):
|
|
return parse_graph_model(unquote(payload))
|
|
|
|
try:
|
|
padded = payload + ("=" * (-len(payload) % 4))
|
|
compressed = base64.b64decode(padded)
|
|
try:
|
|
decoded = zlib.decompress(compressed, -15)
|
|
except zlib.error:
|
|
decoded = zlib.decompress(compressed)
|
|
return parse_graph_model(unquote(decoded.decode("utf-8")))
|
|
except Exception as exc: # noqa: BLE001 - keep API error small and stable.
|
|
raise ValueError("Unable to decode draw.io diagram payload.") from exc
|
|
|
|
|
|
def parse_graph_model(xml_text: str) -> ET.Element:
|
|
try:
|
|
model = ET.fromstring(xml_text)
|
|
except ET.ParseError as exc:
|
|
raise ValueError("Decoded diagram XML is invalid.") from exc
|
|
|
|
if tag_name(model) != "mxGraphModel":
|
|
raise ValueError("Decoded diagram is not an mxGraphModel.")
|
|
|
|
return model
|
|
|
|
|
|
def get_graph_root(model: ET.Element) -> ET.Element:
|
|
root = next((child for child in model if tag_name(child) == "root"), None)
|
|
if root is None:
|
|
raise ValueError("mxGraphModel does not contain a root node.")
|
|
return root
|
|
|
|
|
|
def build_system_xml(model: ET.Element) -> bytes:
|
|
root = get_graph_root(model)
|
|
system = ET.Element("System")
|
|
components_node = ET.SubElement(system, "Components")
|
|
connections_node = ET.SubElement(system, "Connections")
|
|
|
|
for component in read_components(root):
|
|
component_node = ET.SubElement(
|
|
components_node,
|
|
"Component",
|
|
{
|
|
"id": component["id"],
|
|
"type": component["type"],
|
|
"label": component["label"],
|
|
},
|
|
)
|
|
for name, value in component["properties"].items():
|
|
ET.SubElement(
|
|
component_node,
|
|
"Property",
|
|
{"name": name, "value": value},
|
|
)
|
|
|
|
for edge in read_connections(root):
|
|
ET.SubElement(connections_node, "Connection", edge)
|
|
|
|
ET.indent(system, space=" ")
|
|
return ET.tostring(system, encoding="utf-8", xml_declaration=True)
|
|
|
|
|
|
def update_component_parameters(
|
|
model: ET.Element,
|
|
component_id: str,
|
|
name: str | None,
|
|
parameters: dict[str, object],
|
|
) -> None:
|
|
root = get_graph_root(model)
|
|
component = find_component_element(root, component_id)
|
|
if component is None:
|
|
raise ValueError(f"Component '{component_id}' was not found.")
|
|
|
|
if name is not None:
|
|
normalized_name = normalize_component_name(name)
|
|
component.set("name", normalized_name)
|
|
component.set("label", normalized_name)
|
|
update_component_label_cells(root, component_id, normalized_name)
|
|
|
|
schema = schema_for_component(component)
|
|
allowed_parameters = set(schema.get("parameters", {}))
|
|
if not allowed_parameters:
|
|
if parameters:
|
|
raise ValueError("This component does not have configurable parameters.")
|
|
return
|
|
|
|
for name, value in parameters.items():
|
|
if name not in allowed_parameters:
|
|
raise ValueError(f"Parameter '{name}' is not supported by this component.")
|
|
|
|
definition = schema["parameters"][name]
|
|
component.set(name, normalize_parameter_value(name, value, definition))
|
|
|
|
|
|
def normalize_component_name(name: str) -> str:
|
|
normalized = name.strip()
|
|
if not normalized:
|
|
raise ValueError("Component name cannot be empty.")
|
|
return normalized
|
|
|
|
|
|
def update_component_label_cells(
|
|
root: ET.Element,
|
|
component_id: str,
|
|
label: str,
|
|
) -> None:
|
|
for cell in root.iter():
|
|
if tag_name(cell) != "mxCell":
|
|
continue
|
|
if cell.attrib.get("parent") != component_id:
|
|
continue
|
|
style = cell.attrib.get("style", "")
|
|
if style.startswith("text") or "text;" in style:
|
|
cell.set("value", label)
|
|
|
|
|
|
def find_component_element(root: ET.Element, component_id: str) -> ET.Element | None:
|
|
for element in root:
|
|
if tag_name(element) != "object":
|
|
continue
|
|
if "portName" in element.attrib:
|
|
continue
|
|
if element.attrib.get("id") == component_id:
|
|
return element
|
|
return None
|
|
|
|
|
|
def schema_for_component(component: ET.Element) -> dict[str, object]:
|
|
model_type = component.attrib.get("modelType")
|
|
component_type = component.attrib.get("componentType")
|
|
if model_type in COMPONENT_SCHEMAS:
|
|
return COMPONENT_SCHEMAS[model_type]
|
|
if component_type in COMPONENT_SCHEMAS:
|
|
return COMPONENT_SCHEMAS[component_type]
|
|
return {"label": component_type or "component", "parameters": {}}
|
|
|
|
|
|
def normalize_parameter_value(
|
|
name: str,
|
|
value: object,
|
|
definition: dict[str, object],
|
|
) -> str:
|
|
value_type = definition.get("type", "string")
|
|
if value is None or value == "":
|
|
raise ValueError(f"Parameter '{name}' cannot be empty.")
|
|
|
|
if value_type == "float":
|
|
try:
|
|
number = float(value)
|
|
except (TypeError, ValueError) as exc:
|
|
raise ValueError(f"Parameter '{name}' must be a number.") from exc
|
|
if not math.isfinite(number):
|
|
raise ValueError(f"Parameter '{name}' must be finite.")
|
|
validate_number_range(name, number, definition)
|
|
return f"{number:g}"
|
|
|
|
if value_type == "int":
|
|
try:
|
|
number = int(float(value))
|
|
except (TypeError, ValueError) as exc:
|
|
raise ValueError(f"Parameter '{name}' must be an integer.") from exc
|
|
validate_number_range(name, number, definition)
|
|
return str(number)
|
|
|
|
if value_type == "bool":
|
|
if isinstance(value, bool):
|
|
return "true" if value else "false"
|
|
return "true" if str(value).lower() in {"1", "true", "yes"} else "false"
|
|
|
|
return str(value)
|
|
|
|
|
|
def validate_number_range(
|
|
name: str,
|
|
value: float,
|
|
definition: dict[str, object],
|
|
) -> None:
|
|
min_value = definition.get("min")
|
|
max_value = definition.get("max")
|
|
if min_value is not None and value < float(min_value):
|
|
raise ValueError(f"Parameter '{name}' must be >= {min_value}.")
|
|
if max_value is not None and value > float(max_value):
|
|
raise ValueError(f"Parameter '{name}' must be <= {max_value}.")
|
|
|
|
|
|
def read_components(root: ET.Element) -> Iterable[dict]:
|
|
for element in root:
|
|
if tag_name(element) != "object":
|
|
continue
|
|
if "portName" in element.attrib:
|
|
continue
|
|
|
|
cell = next((child for child in element if tag_name(child) == "mxCell"), None)
|
|
if cell is None or cell.attrib.get("vertex") != "1":
|
|
continue
|
|
|
|
component_id = element.attrib.get("id") or cell.attrib.get("id")
|
|
if not component_id:
|
|
continue
|
|
|
|
properties = {
|
|
key: value
|
|
for key, value in element.attrib.items()
|
|
if key not in {"id", "label", "componentType"}
|
|
}
|
|
|
|
yield {
|
|
"id": component_id,
|
|
"type": element.attrib.get("componentType", "component"),
|
|
"label": element.attrib.get("label", ""),
|
|
"properties": properties,
|
|
}
|
|
|
|
|
|
def read_connections(root: ET.Element) -> Iterable[dict[str, str]]:
|
|
endpoints = read_connection_endpoints(root)
|
|
|
|
for cell in root.iter():
|
|
if tag_name(cell) != "mxCell" or cell.attrib.get("edge") != "1":
|
|
continue
|
|
|
|
source_cell = cell.attrib.get("source")
|
|
target_cell = cell.attrib.get("target")
|
|
if not source_cell or not target_cell:
|
|
continue
|
|
|
|
source = endpoints.get(source_cell, {"component": source_cell})
|
|
target = endpoints.get(target_cell, {"component": target_cell})
|
|
edge = {
|
|
"source": source["component"],
|
|
"target": target["component"],
|
|
}
|
|
if "port" in source:
|
|
edge["sourcePort"] = source["port"]
|
|
edge["sourceCell"] = source_cell
|
|
if "port" in target:
|
|
edge["targetPort"] = target["port"]
|
|
edge["targetCell"] = target_cell
|
|
if "id" in cell.attrib:
|
|
edge["id"] = cell.attrib["id"]
|
|
yield edge
|
|
|
|
|
|
def read_connection_endpoints(root: ET.Element) -> dict[str, dict[str, str]]:
|
|
endpoints: dict[str, dict[str, str]] = {}
|
|
component_ports: dict[str, list[str]] = {}
|
|
inferred_port_cells: dict[str, list[dict[str, object]]] = {}
|
|
|
|
for element in root.iter():
|
|
if tag_name(element) != "object":
|
|
continue
|
|
|
|
component_id = element.attrib.get("id")
|
|
ports = element.attrib.get("ports")
|
|
if component_id and ports and "portName" not in element.attrib:
|
|
component_ports[component_id] = [
|
|
port.strip()
|
|
for port in ports.split(",")
|
|
if port.strip()
|
|
]
|
|
|
|
endpoint_id = element.attrib.get("id")
|
|
port_name = element.attrib.get("portName")
|
|
if not endpoint_id or not port_name:
|
|
continue
|
|
|
|
cell = next((child for child in element if tag_name(child) == "mxCell"), None)
|
|
if cell is None:
|
|
continue
|
|
|
|
component_id = cell.attrib.get("parent")
|
|
if not component_id:
|
|
continue
|
|
|
|
endpoint = {
|
|
"component": component_id,
|
|
"port": port_name,
|
|
}
|
|
endpoints[endpoint_id] = endpoint
|
|
|
|
cell_id = cell.attrib.get("id")
|
|
if cell_id:
|
|
endpoints[cell_id] = endpoint
|
|
|
|
for cell in root.iter():
|
|
if tag_name(cell) != "mxCell":
|
|
continue
|
|
if cell.attrib.get("vertex") != "1" or not is_connectable(cell):
|
|
continue
|
|
|
|
cell_id = cell.attrib.get("id")
|
|
component_id = cell.attrib.get("parent")
|
|
if not cell_id or not component_id or cell_id in endpoints:
|
|
continue
|
|
if component_id not in component_ports:
|
|
continue
|
|
|
|
geometry = next(
|
|
(child for child in cell if tag_name(child) == "mxGeometry"),
|
|
None,
|
|
)
|
|
inferred_port_cells.setdefault(component_id, []).append(
|
|
{
|
|
"id": cell_id,
|
|
"x": read_float_attribute(geometry, "x"),
|
|
"y": read_float_attribute(geometry, "y"),
|
|
"width": read_float_attribute(geometry, "width"),
|
|
"height": read_float_attribute(geometry, "height"),
|
|
}
|
|
)
|
|
|
|
for component_id, cells in inferred_port_cells.items():
|
|
for cell_id, port_name in infer_port_cell_names(
|
|
cells,
|
|
component_ports.get(component_id, []),
|
|
).items():
|
|
endpoints.setdefault(
|
|
cell_id,
|
|
{
|
|
"component": component_id,
|
|
"port": port_name,
|
|
},
|
|
)
|
|
|
|
return endpoints
|
|
|
|
|
|
def is_connectable(cell: ET.Element) -> bool:
|
|
connectable = cell.attrib.get("connectable")
|
|
if connectable is not None:
|
|
return connectable == "1"
|
|
|
|
style = cell.attrib.get("style", "")
|
|
style_flags = {
|
|
item.split("=", 1)[0]: item.split("=", 1)[1]
|
|
for item in style.split(";")
|
|
if "=" in item
|
|
}
|
|
return style_flags.get("connectable") == "1"
|
|
|
|
|
|
def infer_port_cell_names(
|
|
cells: list[dict[str, object]],
|
|
port_names: list[str],
|
|
) -> dict[str, str]:
|
|
if len(cells) != len(port_names):
|
|
return {}
|
|
|
|
if len(port_names) == 1:
|
|
return {str(cells[0]["id"]): port_names[0]}
|
|
|
|
if set(port_names) == {"port_a", "port_b"} and len(cells) == 2:
|
|
sorted_cells = sorted(cells, key=lambda cell: cell_center(cell)[0])
|
|
return {
|
|
str(sorted_cells[0]["id"]): "port_a",
|
|
str(sorted_cells[1]["id"]): "port_b",
|
|
}
|
|
|
|
if set(port_names) == {"port_in", "port_out1", "port_out2"} and len(cells) == 3:
|
|
sorted_by_y = sorted(cells, key=lambda cell: cell_center(cell)[1])
|
|
top_cell = sorted_by_y[0]
|
|
remaining = sorted(sorted_by_y[1:], key=lambda cell: cell_center(cell)[0])
|
|
return {
|
|
str(top_cell["id"]): "port_in",
|
|
str(remaining[0]["id"]): "port_out2",
|
|
str(remaining[1]["id"]): "port_out1",
|
|
}
|
|
|
|
sorted_cells = sorted(cells, key=lambda cell: (cell_center(cell)[1], cell_center(cell)[0]))
|
|
return {
|
|
str(cell["id"]): port_names[index]
|
|
for index, cell in enumerate(sorted_cells)
|
|
}
|
|
|
|
|
|
def cell_center(cell: dict[str, object]) -> tuple[float, float]:
|
|
return (
|
|
float(cell["x"]) + float(cell["width"]) / 2.0,
|
|
float(cell["y"]) + float(cell["height"]) / 2.0,
|
|
)
|
|
|
|
|
|
def read_float_attribute(element: ET.Element | None, name: str) -> float:
|
|
if element is None:
|
|
return 0.0
|
|
|
|
try:
|
|
return float(element.attrib.get(name, "0"))
|
|
except ValueError:
|
|
return 0.0
|
|
|
|
|
|
def tag_name(element: ET.Element) -> str:
|
|
return element.tag.rsplit("}", 1)[-1]
|