ProcessModelDemo / src /streamlit_app.py
Alef1986's picture
Upload streamlit_app.py
b129529 verified
Raw History Blame Contribute Delete
17 kB
import streamlit as st
import xml.etree.ElementTree as ET
import networkx as nx
import streamlit.components.v1 as components
import json
# Page config
st.set_page_config(page_title="ISO 15288 Analyzer", layout="wide")
st.title("ISO 15288: Architecture Analyzer")
def render_mermaid(mermaid_code, nodes_json):
"""Рендерит Mermaid с поддержкой Зума, Экспорта в PNG и боковой панели"""
safe_mermaid_code = mermaid_code.replace('`', '\\`').replace('$', '\\$')
html_code = f"""
<!DOCTYPE html>
<html>
<head>
<script src="https://cdn.jsdelivr.net/npm/svg-pan-zoom@3.6.1/dist/svg-pan-zoom.min.js"></script>
<style>
body {{ margin: 0; padding: 0; font-family: sans-serif; background: #ffffff; overflow: hidden; }}
#graphDiv {{ width: 100vw; height: 100vh; cursor: grab; }}
#graphDiv:active {{ cursor: grabbing; }}
/* Стили кнопки экспорта */
#toolbar {{
position: absolute;
bottom: 25px;
right: 25px;
z-index: 100;
}}
button.export-btn {{
background: #ffffff; border: 1px solid #ccc; padding: 10px 18px;
border-radius: 8px; cursor: pointer; font-size: 14px; font-weight: bold;
box-shadow: 0 4px 6px rgba(0,0,0,0.1); transition: all 0.2s;
}}
button.export-btn:hover {{
background: #f0f0f0;
transform: translateY(-2px);
box-shadow: 0 6px 12px rgba(0,0,0,0.15);
}}
/* Стили всплывающей боковой панели */
#infoPanel {{
display: none;
position: absolute;
top: 20px;
right: 20px;
width: 350px;
max-height: 80vh;
background: #ffffff;
border: 1px solid #e0e0e0;
border-radius: 8px;
box-shadow: 0 10px 25px rgba(0,0,0,0.2);
padding: 20px;
overflow-y: auto;
z-index: 1000;
flex-direction: column;
}}
#infoPanel h3 {{ margin: 0 0 15px 0; font-size: 18px; color: #333; }}
#infoPanel p {{ margin: 0; font-size: 14px; color: #555; line-height: 1.6; }}
#closeBtn {{
position: absolute;
top: 15px;
right: 15px;
cursor: pointer;
font-size: 24px;
color: #aaa;
background: none;
border: none;
padding: 0;
line-height: 1;
}}
#closeBtn:hover {{ color: #333; }}
/* Указатель для кликабельных узлов */
.node {{ cursor: pointer !important; }}
</style>
<script type="module">
import mermaid from 'https://cdn.jsdelivr.net/npm/mermaid@10/dist/mermaid.esm.min.mjs';
mermaid.initialize({{ startOnLoad: false, theme: 'default', securityLevel: 'loose' }});
// 1. Рендеринг и подключение Зума
window.renderDiagram = async function() {{
const container = document.getElementById('graphDiv');
const {{ svg }} = await mermaid.render('mermaidSvg', `{safe_mermaid_code}`);
container.innerHTML = svg;
const svgElement = container.querySelector('svg');
svgElement.style.width = "100%";
svgElement.style.height = "100%";
window.panZoomSetup = svgPanZoom(svgElement, {{
zoomEnabled: true,
controlIconsEnabled: true,
fit: true,
center: true,
minZoom: 0.1,
maxZoom: 10
}});
}};
window.addEventListener('load', window.renderDiagram);
</script>
<script>
// 2. Логика Информационной Панели
const nodeData = {nodes_json};
window.handleNodeClick = function(nodeId) {{
const data = nodeData[nodeId];
if (data) {{
document.getElementById('panelTitle').innerText = data.name;
document.getElementById('panelDoc').innerText = data.doc ? data.doc : "Description is missing in the model.";
document.getElementById('infoPanel').style.display = 'flex';
}}
}};
window.closePanel = function() {{
document.getElementById('infoPanel').style.display = 'none';
}};
// 3. Экспорт в PNG
window.downloadPng = function() {{
const svg = document.querySelector('#graphDiv svg');
if (!svg) return;
const oldZoom = window.panZoomSetup.getZoom();
const oldPan = window.panZoomSetup.getPan();
window.panZoomSetup.resetZoom();
window.panZoomSetup.resetPan();
setTimeout(() => {{
const svgData = new XMLSerializer().serializeToString(svg);
const canvas = document.createElement("canvas");
const bbox = svg.getBoundingClientRect();
canvas.width = bbox.width * 2;
canvas.height = bbox.height * 2;
const ctx = canvas.getContext("2d");
ctx.scale(2, 2);
ctx.fillStyle = "white";
ctx.fillRect(0, 0, canvas.width, canvas.height);
const img = new Image();
img.src = 'data:image/svg+xml;base64,' + btoa(unescape(encodeURIComponent(svgData)));
img.onload = function() {{
ctx.drawImage(img, 0, 0);
const a = document.createElement("a");
a.download = "Architecture_Diagram.png";
a.href = canvas.toDataURL("image/png");
a.click();
window.panZoomSetup.zoom(oldZoom);
window.panZoomSetup.pan(oldPan);
}};
}}, 100);
}};
</script>
</head>
<body>
<div id="toolbar">
<button class="export-btn" onclick="downloadPng()">📸 Export as PNG</button>
</div>
<div id="infoPanel">
<button id="closeBtn" onclick="closePanel()">&times;</button>
<h3 id="panelTitle">Process Name</h3>
<p id="panelDoc">Documentation...</p>
</div>
<div id="graphDiv"></div>
</body>
</html>
"""
components.html(html_code, height=800, scrolling=False)
@st.cache_data
def load_and_parse_xml(file_path):
tree = ET.parse(file_path)
root = tree.getroot()
G = nx.DiGraph()
processes = {}
for elem in root.findall('.//*'):
if elem.tag.endswith('element'):
n_id = elem.get('identifier')
n_type = next((v for k, v in elem.attrib.items() if k.endswith('type')), 'Unknown')
name = "Unnamed"
doc = ""
for child in elem:
if child.tag.endswith('name') and child.text:
name = child.text
elif child.tag.endswith('documentation') and child.text:
doc = child.text
G.add_node(n_id, name=name, type=n_type, doc=doc)
if 'BusinessFunction' in n_type:
processes[n_id] = name
for rel in root.findall('.//*'):
if rel.tag.endswith('relationship'):
src = rel.get('source')
tgt = rel.get('target')
r_type = next((v for k, v in rel.attrib.items() if k.endswith('type')), 'Association')
access = rel.get('accessType', '')
if 'Access' in r_type and access.lower() == 'read':
G.add_edge(tgt, src, type=r_type, access=access)
else:
G.add_edge(src, tgt, type=r_type, access=access)
return G, processes
def graph_to_mermaid_and_data(G, nodes_to_include, edges_to_include):
if not nodes_to_include:
return "", "{}"
lines = [
"flowchart LR",
"classDef process fill:#FFF2CC,stroke:#D6B656,stroke-width:2px,color:#000;",
"classDef object fill:#D5E8D4,stroke:#82B366,stroke-width:2px,color:#000;"
]
nodes_data = {}
for n_id in set(nodes_to_include):
data = G.nodes.get(n_id, {})
m_id = n_id.replace("-", "_")
name = data.get("name", "Unknown").replace('"', "'")
n_type = data.get("type", "")
doc = data.get("doc", "").replace('"', "'").replace("\n", " ")
if 'BusinessProcess' in n_type or 'BusinessFunction' in n_type:
lines.append(f' {m_id}("{name}"):::process')
else:
lines.append(f' {m_id}["{name}"]:::object')
lines.append(f' click {m_id} call handleNodeClick()')
nodes_data[m_id] = {"name": name, "doc": doc}
added_edges = set()
for u, v, data in edges_to_include:
edge_key = (u, v)
if edge_key not in added_edges:
u_id = u.replace("-", "_")
v_id = v.replace("-", "_")
e_type = data.get("type", "")
access = data.get("access", "")
label = access if access else e_type.replace("Relationship", "").replace("archimate:", "")
lines.append(f' {u_id} -->|{label}| {v_id}')
added_edges.add(edge_key)
return "\n".join(lines), json.dumps(nodes_data)
# Загрузка графов
try:
G, processes = load_and_parse_xml("ISO15288_Archi_Import.xml")
U = G.to_undirected()
sorted_process_names = sorted(list(processes.values()))
except FileNotFoundError:
st.error("File 'ISO15288_Archi_Import.xml' not found.")
st.stop()
# Интерфейс
mode = st.radio(
"Select analysis scenario:",
(
"Case 1: Interface exchange between selected processes",
"Case 2: Shortest DATA FLOW path (Directed)",
"Case 3: Closest STRUCTURAL connection (Undirected)",
"Case 4: All possible DATA FLOW paths (Directed)"
)
)
st.markdown("---")
if mode == "Case 1: Interface exchange between selected processes":
st.write("Select 2 to 5 processes. The script will find objects they exchange data through.")
selected_names = st.multiselect("Select processes:", sorted_process_names, max_selections=5)
if st.button("Generate Interface Diagram") and len(selected_names) >= 2:
selected_ids = [k for k, v in processes.items() if v in selected_names]
nodes_to_draw = set(selected_ids)
edges_to_draw = []
def safe_add_edge(u, v):
if G.has_edge(u, v): edges_to_draw.append((u, v, G[u][v]))
for i in range(len(selected_ids)):
for j in range(i + 1, len(selected_ids)):
p1, p2 = selected_ids[i], selected_ids[j]
safe_add_edge(p1, p2)
safe_add_edge(p2, p1)
common_neighbors = set(U.neighbors(p1)).intersection(set(U.neighbors(p2)))
for neighbor in common_neighbors:
nodes_to_draw.add(neighbor)
safe_add_edge(p1, neighbor)
safe_add_edge(neighbor, p1)
safe_add_edge(p2, neighbor)
safe_add_edge(neighbor, p2)
if len(nodes_to_draw) == len(selected_ids) and not edges_to_draw:
st.warning("No direct links or shared exchange objects found between these processes.")
else:
st.success("Exchange interfaces found!")
mermaid_code, json_data = graph_to_mermaid_and_data(G, nodes_to_draw, edges_to_draw)
render_mermaid(mermaid_code, json_data)
elif mode == "Case 2: Shortest DATA FLOW path (Directed)":
st.write("Finds the shortest valid data sequence (respecting Write -> Read directions).")
col1, col2 = st.columns(2)
with col1:
p1_name = st.selectbox("Process A (From):", sorted_process_names, key="c2_p1")
with col2:
p2_name = st.selectbox("Process B (To):", sorted_process_names, key="c2_p2")
if st.button("Find Directed Path") and p1_name != p2_name:
p1_id = [k for k, v in processes.items() if v == p1_name][0]
p2_id = [k for k, v in processes.items() if v == p2_name][0]
try:
path = nx.shortest_path(G, source=p1_id, target=p2_id)
edges_in_path = [(path[i], path[i+1], G[path[i]][path[i+1]]) for i in range(len(path)-1)]
st.success(f"Data flow path found! Length: {len(path)-1} step(s).")
mermaid_code, json_data = graph_to_mermaid_and_data(G, path, edges_in_path)
render_mermaid(mermaid_code, json_data)
except nx.NetworkXNoPath:
st.error("No valid DATA FLOW exists between these processes (direction matters).")
elif mode == "Case 3: Closest STRUCTURAL connection (Undirected)":
st.write("Finds any connection (e.g., sharing the same input document), ignoring direction.")
col1, col2 = st.columns(2)
with col1:
p1_name = st.selectbox("Process A:", sorted_process_names, key="c3_p1")
with col2:
p2_name = st.selectbox("Process B:", sorted_process_names, key="c3_p2")
if st.button("Find Closest Connection") and p1_name != p2_name:
p1_id = [k for k, v in processes.items() if v == p1_name][0]
p2_id = [k for k, v in processes.items() if v == p2_name][0]
try:
path = nx.shortest_path(U, source=p1_id, target=p2_id)
edges_in_path = []
for i in range(len(path)-1):
u, v = path[i], path[i+1]
if G.has_edge(u, v): edges_in_path.append((u, v, G[u][v]))
elif G.has_edge(v, u): edges_in_path.append((v, u, G[v][u]))
st.success(f"Structural connection found! Length: {len(path)-1} step(s).")
mermaid_code, json_data = graph_to_mermaid_and_data(G, path, edges_in_path)
render_mermaid(mermaid_code, json_data)
except nx.NetworkXNoPath:
st.error("These processes are completely disconnected in the model.")
elif mode == "Case 4: All possible DATA FLOW paths (Directed)":
st.write("Finds all valid data flow sequences (respecting directions).")
col1, col2 = st.columns(2)
with col1:
p1_name = st.selectbox("Process A (From):", sorted_process_names, key="c4_p1")
with col2:
p2_name = st.selectbox("Process B (To):", sorted_process_names, key="c4_p2")
if st.button("Find All Directed Paths") and p1_name != p2_name:
p1_id = [k for k, v in processes.items() if v == p1_name][0]
p2_id = [k for k, v in processes.items() if v == p2_name][0]
try:
paths = list(nx.all_simple_paths(G, source=p1_id, target=p2_id, cutoff=6))
if not paths:
st.error("No valid data flow paths found within 6 steps.")
else:
st.success(f"Found {len(paths)} unique data flow path(s).")
nodes_in_paths = set()
edges_in_paths = []
for path in paths:
nodes_in_paths.update(path)
for i in range(len(path)-1):
u, v = path[i], path[i+1]
edges_in_paths.append((u, v, G[u][v]))
mermaid_code, json_data = graph_to_mermaid_and_data(G, nodes_in_paths, edges_in_paths)
render_mermaid(mermaid_code, json_data)
except nx.NetworkXNoPath:
st.error("No valid DATA FLOW exists between these processes.")