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"""
Process Name
Documentation...
"""
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.")