221 lines
9.6 KiB
Python
221 lines
9.6 KiB
Python
"""Собирает .archimate (нативный формат Archi) из профиля контура и матрицы связей.
|
||
|
||
py build_archimate.py <профиль.json> <матрица.json> <выход.archimate>
|
||
|
||
Все имена, зависящие от конкретной инсталляции (кластер, узлы, домены, реестр),
|
||
приходят из профиля — в самом скрипте их нет. См. contour-profile.example.json.
|
||
"""
|
||
import json, pathlib, sys, io
|
||
from xml.sax.saxutils import escape, quoteattr
|
||
|
||
sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding="utf-8")
|
||
if len(sys.argv) != 4:
|
||
sys.exit(__doc__)
|
||
PROFILE = json.loads(pathlib.Path(sys.argv[1]).read_text(encoding="utf-8"))
|
||
DATA = json.loads(pathlib.Path(sys.argv[2]).read_text(encoding="utf-8"))
|
||
OUT = pathlib.Path(sys.argv[3])
|
||
|
||
elements, relations, _rc = [], [], [0]
|
||
REL_INDEX = {}
|
||
|
||
def el(folder, eid, etype, name, doc=""):
|
||
elements.append((folder, eid, etype, name, doc)); return eid
|
||
|
||
def R(rtype, src, tgt):
|
||
_rc[0] += 1
|
||
rid = f"rel-{_rc[0]:03d}"
|
||
relations.append((rid, rtype, src, tgt))
|
||
REL_INDEX[(rtype, src, tgt)] = rid
|
||
return rid
|
||
|
||
# ---------------------------------------------------------------- технологический слой
|
||
T = "technology"
|
||
nd_cluster = el(T, "nd-cluster", "Node", PROFILE["cluster"]["name"], PROFILE["cluster"].get("doc", ""))
|
||
|
||
for s in PROFILE["software"]:
|
||
el(T, s["id"], "SystemSoftware", s["name"], s.get("doc", ""))
|
||
R("Composition", nd_cluster, s["id"])
|
||
|
||
cam_ids = []
|
||
for n in PROFILE.get("camunda_sub", []):
|
||
cid = "ss-cam-" + n.lower().replace(" ", "-")
|
||
el(T, cid, "SystemSoftware", n)
|
||
R("Composition", "ss-camunda", cid); cam_ids.append(cid)
|
||
|
||
for x in PROFILE["external"]:
|
||
el(T, x["id"], "Node", x["name"], x.get("doc", ""))
|
||
for a in x.get("artifacts", []):
|
||
el(T, a["id"], "Artifact", a["name"])
|
||
R("Composition", x["id"], a["id"])
|
||
if x.get("serves") == "cluster":
|
||
R("Serving", x["id"], nd_cluster)
|
||
elif x.get("serves"):
|
||
R("Serving", x["id"], x["serves"])
|
||
|
||
fn = PROFILE["frontend_node"]
|
||
el(T, fn["id"], "Node", fn["name"], fn.get("doc", ""))
|
||
R("Assignment", fn["id"], fn["assigned_to"])
|
||
|
||
for ts in PROFILE["services"]:
|
||
el(T, ts["id"], "TechnologyService", ts["name"])
|
||
for r in ts["realizers"]:
|
||
R("Realization", r, ts["id"])
|
||
if ts.get("serves_software"):
|
||
R("Serving", ts["id"], ts["serves_software"])
|
||
|
||
# ---------------------------------------------------------------- прикладной слой
|
||
app_id = {a: "ac-" + a for a in DATA["apps"]}
|
||
for a in DATA["apps"]:
|
||
el("application", app_id[a], "ApplicationComponent", a, f"apps/{a}/<contour>/")
|
||
|
||
consumers = {}
|
||
for ts in PROFILE["services"]:
|
||
key = ts.get("key")
|
||
if not key:
|
||
consumers[ts["id"]] = []; continue
|
||
if key == "istio":
|
||
lst = [a for a in DATA["apps"] if a in DATA["exposed"]]
|
||
else:
|
||
lst = [a for a in DATA["apps"] if DATA["matrix"][a].get(key)]
|
||
consumers[ts["id"]] = lst
|
||
for a in lst:
|
||
R("Serving", ts["id"], app_id[a])
|
||
|
||
# ---------------------------------------------------------------- представления
|
||
views, _oc = [], [0]
|
||
|
||
def view(name, doc=""):
|
||
v = {"id": f"view-{len(views)+1}", "name": name, "doc": doc, "objects": []}
|
||
views.append(v); return v
|
||
|
||
def obj(v, eid, x, y, w, h, parent=None):
|
||
_oc[0] += 1
|
||
o = {"id": f"do-{_oc[0]:04d}", "el": eid, "b": (x, y, w, h), "children": [], "src": [], "tgt": []}
|
||
(parent["children"] if parent else v["objects"]).append(o)
|
||
return o
|
||
|
||
def conn(rtype, so, to):
|
||
key = (rtype, so["el"], to["el"])
|
||
if key not in REL_INDEX:
|
||
raise KeyError(f"нет отношения в модели: {key}")
|
||
_oc[0] += 1
|
||
cid = f"cn-{_oc[0]:04d}"
|
||
so["src"].append({"id": cid, "rel": REL_INDEX[key], "tgt": to["id"]})
|
||
to["tgt"].append(cid)
|
||
|
||
SERVICES = PROFILE["services"]
|
||
SW = {s["id"]: s for s in PROFILE["software"]}
|
||
|
||
v1 = view("T1. Технологический слой",
|
||
"Что развёрнуто в кластере и какие технологические сервисы это даёт. "
|
||
"Приложения-потребители вынесены в отдельные представления T2 и далее.")
|
||
o_ts = {ts["id"]: obj(v1, ts["id"], 20 + i * 165, 30, 150, 60) for i, ts in enumerate(SERVICES)}
|
||
|
||
o_cluster = obj(v1, nd_cluster, 20, 140, 1270, 400)
|
||
o_ss, ids = {}, [s["id"] for s in PROFILE["software"]]
|
||
for i, sid in enumerate(ids[:5]):
|
||
o_ss[sid] = obj(v1, sid, 20 + i * 240, 45, 220, 70, o_cluster)
|
||
for i, sid in enumerate(ids[5:9]):
|
||
o_ss[sid] = obj(v1, sid, 20 + i * 240, 145, 220, 70, o_cluster)
|
||
if len(ids) > 9:
|
||
o_ss[ids[9]] = obj(v1, ids[9], 20, 245, 1230, 130, o_cluster)
|
||
for i, cid in enumerate(cam_ids):
|
||
obj(v1, cid, 10 + i * 174, 50, 162, 60, o_ss[ids[9]])
|
||
|
||
o_ext = {}
|
||
for i, x in enumerate(PROFILE["external"]):
|
||
h = 130 if x.get("artifacts") else 70
|
||
o_ext[x["id"]] = obj(v1, x["id"], 1320, 140 + i * 90, 330, h)
|
||
for j, a in enumerate(x.get("artifacts", [])):
|
||
obj(v1, a["id"], 15 + j * 155, 50, 145, 60, o_ext[x["id"]])
|
||
o_front = obj(v1, fn["id"], 20, 570, 300, 60)
|
||
|
||
for ts in SERVICES:
|
||
for r in ts["realizers"]:
|
||
conn("Realization", o_ss.get(r) or o_ext[r], o_ts[ts["id"]])
|
||
if ts.get("serves_software"):
|
||
conn("Serving", o_ts[ts["id"]], o_ss[ts["serves_software"]])
|
||
conn("Assignment", o_front, o_ss[fn["assigned_to"]])
|
||
for x in PROFILE["external"]:
|
||
if x.get("serves") == "cluster":
|
||
conn("Serving", o_ext[x["id"]], o_cluster)
|
||
elif x.get("serves"):
|
||
conn("Serving", o_ext[x["id"]], o_ss[x["serves"]])
|
||
|
||
PER_ROW = 6
|
||
for ts in SERVICES:
|
||
lst = consumers[ts["id"]]
|
||
if not lst:
|
||
continue
|
||
v = view(f"T{len(views)+1}. Потребители: {ts['name']}",
|
||
f"{len(lst)} из {len(DATA['apps'])} приложений контура. "
|
||
f"Связи выведены из манифестов приложений.")
|
||
ots = obj(v, ts["id"], 20, 20, 260, 60)
|
||
for i, r in enumerate(ts["realizers"]):
|
||
conn("Realization", obj(v, r, 320 + i * 280, 20, 260, 60), ots)
|
||
for i, a in enumerate(lst):
|
||
conn("Serving", ots,
|
||
obj(v, app_id[a], 20 + (i % PER_ROW) * 190, 150 + (i // PER_ROW) * 80, 170, 55))
|
||
|
||
# ---------------------------------------------------------------- сериализация
|
||
FOLDERS = [("Strategy", "strategy"), ("Business", "business"), ("Application", "application"),
|
||
("Technology & Physical", "technology"), ("Motivation", "motivation"),
|
||
("Implementation & Migration", "implementation_migration"), ("Other", "other"),
|
||
("Relations", "relations"), ("Views", "diagrams")]
|
||
|
||
def emit_obj(o, ind):
|
||
# Имена тегов — как их пишет сам Archi: <child xsi:type="archimate:DiagramObject">,
|
||
# соединение через атрибут relationship. Имена EClass из метамодели
|
||
# (DiagramModelArchimateObject) в XML не используются.
|
||
p = " " * ind
|
||
tgt = f' targetConnections={quoteattr(" ".join(o["tgt"]))}' if o["tgt"] else ""
|
||
x, y, w, h = o["b"]
|
||
s = f'{p}<child xsi:type="archimate:DiagramObject" id="{o["id"]}" archimateElement="{o["el"]}"{tgt}>\n'
|
||
s += f'{p} <bounds x="{x}" y="{y}" width="{w}" height="{h}"/>\n'
|
||
for c in o["src"]:
|
||
s += (f'{p} <sourceConnection xsi:type="archimate:Connection" id="{c["id"]}" '
|
||
f'source="{o["id"]}" target="{c["tgt"]}" relationship="{c["rel"]}"/>\n')
|
||
for c in o["children"]:
|
||
s += emit_obj(c, ind + 2)
|
||
return s + f'{p}</child>\n'
|
||
|
||
out = ['<?xml version="1.0" encoding="UTF-8"?>',
|
||
'<archimate:model xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" '
|
||
'xmlns:archimate="http://www.archimatetool.com/archimate" '
|
||
f'name={quoteattr(PROFILE["model"]["name"])} id="model-contour" version="5.0.0">',
|
||
f' <purpose>{escape(PROFILE["model"]["purpose"])}</purpose>']
|
||
|
||
for fname, ftype in FOLDERS:
|
||
inner = ""
|
||
if ftype in ("technology", "application"):
|
||
for folder, eid, etype, name, doc in elements:
|
||
if folder != ftype:
|
||
continue
|
||
if doc:
|
||
inner += (f' <element xsi:type="archimate:{etype}" name={quoteattr(name)} id="{eid}">\n'
|
||
f' <documentation>{escape(doc)}</documentation>\n </element>\n')
|
||
else:
|
||
inner += f' <element xsi:type="archimate:{etype}" name={quoteattr(name)} id="{eid}"/>\n'
|
||
elif ftype == "relations":
|
||
for rid, rtype, src, tgt in relations:
|
||
inner += (f' <element xsi:type="archimate:{rtype}Relationship" id="{rid}" '
|
||
f'source="{src}" target="{tgt}"/>\n')
|
||
elif ftype == "diagrams":
|
||
for v in views:
|
||
inner += f' <element xsi:type="archimate:ArchimateDiagramModel" name={quoteattr(v["name"])} id="{v["id"]}">\n'
|
||
if v["doc"]:
|
||
inner += f' <documentation>{escape(v["doc"])}</documentation>\n'
|
||
for o in v["objects"]:
|
||
inner += emit_obj(o, 6)
|
||
inner += ' </element>\n'
|
||
out.append(f' <folder name={quoteattr(fname)} id="folder-{ftype}" type="{ftype}">\n{inner} </folder>'
|
||
if inner else f' <folder name={quoteattr(fname)} id="folder-{ftype}" type="{ftype}"/>')
|
||
|
||
out.append('</archimate:model>')
|
||
OUT.write_text("\n".join(out) + "\n", encoding="utf-8", newline="\n")
|
||
|
||
print(f"записано: {OUT}")
|
||
print(f"элементов: {len(elements)} отношений: {len(relations)} представлений: {len(views)}")
|
||
for v in views:
|
||
print(f" {v['name']}")
|