Bastien Gourdon
207 lines
7.9 KiB
Python
207 lines
7.9 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Generate strict, queryable BPMN 2.0 files from processes.yaml (EV-008).
|
|
|
|
USAGE
|
|
python3 generate_bpmn.py [output_dir]
|
|
|
|
One file per procedure, in bpmn/. Rule identifiers are carried in
|
|
extensionElements under the project namespace, so a rule can be traced to every
|
|
procedure that enforces it with a single XPath expression:
|
|
|
|
//bpmn:process[.//pr:rule='TN-002']/@id
|
|
|
|
Diagram interchange is emitted with a simple left-to-right layout so the files
|
|
open directly in any BPMN editor. Layout is cosmetic and never authoritative.
|
|
"""
|
|
import os
|
|
import sys
|
|
from xml.sax.saxutils import escape
|
|
|
|
import yaml
|
|
|
|
HERE = os.path.dirname(os.path.abspath(__file__))
|
|
SRC = os.path.join(HERE, "processes.yaml")
|
|
OUTDIR = sys.argv[1] if len(sys.argv) > 1 else os.path.join(HERE, "bpmn")
|
|
|
|
BPMN = "http://www.omg.org/spec/BPMN/20100524/MODEL"
|
|
BPMNDI = "http://www.omg.org/spec/BPMN/20100524/DI"
|
|
DI = "http://www.omg.org/spec/DD/20100524/DI"
|
|
DC = "http://www.omg.org/spec/DD/20100524/DC"
|
|
|
|
TAG = {
|
|
"start": "startEvent",
|
|
"end": "endEvent",
|
|
"userTask": "userTask",
|
|
"scriptTask": "scriptTask",
|
|
"callActivity": "callActivity",
|
|
"gateway": "exclusiveGateway",
|
|
}
|
|
W = {"start": 36, "end": 36, "gateway": 50}
|
|
H = {"start": 36, "end": 36, "gateway": 50}
|
|
DEFAULT_W, DEFAULT_H = 150, 70
|
|
COL, ROW = 200, 130
|
|
|
|
|
|
def selfcheck(doc, rule_ids):
|
|
"""A procedure must be a connected graph, and every rule it cites must exist."""
|
|
problems = []
|
|
ids = {p["id"] for p in doc["processes"]}
|
|
for p in doc["processes"]:
|
|
nodes = {n["id"] for n in p["flow"]}
|
|
for f in p["flows"]:
|
|
for side in ("from", "to"):
|
|
if f[side] not in nodes:
|
|
problems.append("%s: flow references unknown node %s" % (p["id"], f[side]))
|
|
targets = {f["to"] for f in p["flows"]}
|
|
sources = {f["from"] for f in p["flows"]}
|
|
for n in p["flow"]:
|
|
if n["type"] == "start" and n["id"] in targets:
|
|
problems.append("%s: start event %s has an incoming flow" % (p["id"], n["id"]))
|
|
if n["type"] == "end" and n["id"] in sources:
|
|
problems.append("%s: end event %s has an outgoing flow" % (p["id"], n["id"]))
|
|
if n["type"] not in ("start", "end") and n["id"] not in targets:
|
|
problems.append("%s: node %s is unreachable" % (p["id"], n["id"]))
|
|
if n["type"] != "end" and n["id"] not in sources:
|
|
problems.append("%s: node %s is a dead end" % (p["id"], n["id"]))
|
|
if n["type"] == "callActivity" and n.get("calls") not in ids:
|
|
problems.append("%s: %s calls unknown procedure %s"
|
|
% (p["id"], n["id"], n.get("calls")))
|
|
for r in n.get("rules") or []:
|
|
if rule_ids and r not in rule_ids:
|
|
problems.append("%s/%s: unknown rule %s" % (p["id"], n["id"], r))
|
|
return problems
|
|
|
|
|
|
def layout(proc):
|
|
"""Assign a column by longest path from start, a row to disambiguate siblings."""
|
|
succ = {}
|
|
for f in proc["flows"]:
|
|
succ.setdefault(f["from"], []).append(f["to"])
|
|
starts = [n["id"] for n in proc["flow"] if n["type"] == "start"]
|
|
depth = {}
|
|
for s in starts:
|
|
stack = [(s, 0)]
|
|
seen = set()
|
|
while stack:
|
|
node, d = stack.pop()
|
|
if (node, d) in seen:
|
|
continue
|
|
seen.add((node, d))
|
|
if depth.get(node, -1) < d:
|
|
depth[node] = d
|
|
for nxt in succ.get(node, []):
|
|
if d < 60:
|
|
stack.append((nxt, d + 1))
|
|
used = {}
|
|
pos = {}
|
|
for n in proc["flow"]:
|
|
c = depth.get(n["id"], 0)
|
|
r = used.get(c, 0)
|
|
used[c] = r + 1
|
|
pos[n["id"]] = (60 + c * COL, 80 + r * ROW)
|
|
return pos
|
|
|
|
|
|
def build(proc, ns):
|
|
pid = proc["id"]
|
|
o = []
|
|
a = o.append
|
|
a('<?xml version="1.0" encoding="UTF-8"?>')
|
|
a('<bpmn:definitions xmlns:bpmn="%s" xmlns:bpmndi="%s" xmlns:di="%s" '
|
|
'xmlns:dc="%s" xmlns:pr="%s" id="defs_%s" targetNamespace="%s" '
|
|
'exporter="generate_bpmn.py">' % (BPMN, BPMNDI, DI, DC, ns, pid, ns))
|
|
a(' <bpmn:process id="%s" name="%s" isExecutable="false">'
|
|
% (pid, escape(proc["name"])))
|
|
doc = "%s | Scope: %s | Trigger: %s" % (
|
|
proc["name"], proc.get("scope", ""), proc.get("trigger", ""))
|
|
a(' <bpmn:documentation>%s</bpmn:documentation>' % escape(doc))
|
|
|
|
incoming, outgoing = {}, {}
|
|
for i, f in enumerate(proc["flows"], start=1):
|
|
fid = "flow_%d" % i
|
|
f["_id"] = fid
|
|
outgoing.setdefault(f["from"], []).append(fid)
|
|
incoming.setdefault(f["to"], []).append(fid)
|
|
|
|
for n in proc["flow"]:
|
|
tag = TAG[n["type"]]
|
|
attrs = 'id="%s" name="%s"' % (n["id"], escape(n["name"]))
|
|
if n["type"] == "callActivity":
|
|
attrs += ' calledElement="%s"' % n["calls"]
|
|
a(' <bpmn:%s %s>' % (tag, attrs))
|
|
rules = n.get("rules") or []
|
|
if rules:
|
|
a(' <bpmn:extensionElements>')
|
|
a(' <pr:rules>')
|
|
for r in rules:
|
|
a(' <pr:rule>%s</pr:rule>' % r)
|
|
a(' </pr:rules>')
|
|
a(' </bpmn:extensionElements>')
|
|
for fid in incoming.get(n["id"], []):
|
|
a(' <bpmn:incoming>%s</bpmn:incoming>' % fid)
|
|
for fid in outgoing.get(n["id"], []):
|
|
a(' <bpmn:outgoing>%s</bpmn:outgoing>' % fid)
|
|
a(' </bpmn:%s>' % tag)
|
|
|
|
for f in proc["flows"]:
|
|
nm = ' name="%s"' % escape(f["condition"]) if f.get("condition") else ""
|
|
a(' <bpmn:sequenceFlow id="%s"%s sourceRef="%s" targetRef="%s" />'
|
|
% (f["_id"], nm, f["from"], f["to"]))
|
|
a(' </bpmn:process>')
|
|
|
|
pos = layout(proc)
|
|
a(' <bpmndi:BPMNDiagram id="diagram_%s">' % pid)
|
|
a(' <bpmndi:BPMNPlane id="plane_%s" bpmnElement="%s">' % (pid, pid))
|
|
for n in proc["flow"]:
|
|
x, y = pos[n["id"]]
|
|
w, h = W.get(n["type"], DEFAULT_W), H.get(n["type"], DEFAULT_H)
|
|
a(' <bpmndi:BPMNShape id="shape_%s" bpmnElement="%s">' % (n["id"], n["id"]))
|
|
a(' <dc:Bounds x="%d" y="%d" width="%d" height="%d" />' % (x, y, w, h))
|
|
a(' </bpmndi:BPMNShape>')
|
|
for f in proc["flows"]:
|
|
sx, sy = pos[f["from"]]
|
|
tx, ty = pos[f["to"]]
|
|
sw = W.get(next(n["type"] for n in proc["flow"] if n["id"] == f["from"]), DEFAULT_W)
|
|
sh = H.get(next(n["type"] for n in proc["flow"] if n["id"] == f["from"]), DEFAULT_H)
|
|
th = H.get(next(n["type"] for n in proc["flow"] if n["id"] == f["to"]), DEFAULT_H)
|
|
a(' <bpmndi:BPMNEdge id="edge_%s" bpmnElement="%s">' % (f["_id"], f["_id"]))
|
|
a(' <di:waypoint x="%d" y="%d" />' % (sx + sw, sy + sh // 2))
|
|
a(' <di:waypoint x="%d" y="%d" />' % (tx, ty + th // 2))
|
|
a(' </bpmndi:BPMNEdge>')
|
|
a(' </bpmndi:BPMNPlane>')
|
|
a(' </bpmndi:BPMNDiagram>')
|
|
a('</bpmn:definitions>')
|
|
return "\n".join(o) + "\n"
|
|
|
|
|
|
def main():
|
|
doc = yaml.safe_load(open(SRC, encoding="utf-8"))
|
|
rule_ids = set()
|
|
rb = os.path.join(HERE, "rules.yaml")
|
|
if os.path.exists(rb):
|
|
rule_ids = {r["id"] for r in yaml.safe_load(open(rb, encoding="utf-8"))["rules"]}
|
|
|
|
problems = selfcheck(doc, rule_ids)
|
|
if problems:
|
|
print("SELF-CHECK FAILED")
|
|
for p in problems:
|
|
print(" " + p)
|
|
sys.exit(1)
|
|
|
|
os.makedirs(OUTDIR, exist_ok=True)
|
|
ns = doc["meta"]["namespace"]
|
|
for proc in doc["processes"]:
|
|
path = os.path.join(OUTDIR, "%s.bpmn" % proc["id"])
|
|
open(path, "w", encoding="utf-8").write(build(proc, ns))
|
|
n_rules = len({r for n in proc["flow"] for r in (n.get("rules") or [])})
|
|
print(" %-4s %-46s %2d nodes, %2d flows, %2d rules"
|
|
% (proc["id"], proc["name"], len(proc["flow"]),
|
|
len(proc["flows"]), n_rules))
|
|
print("self-check passed: %d procedures, graphs connected, every rule known"
|
|
% len(doc["processes"]))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|