"""bpmn — a module's workflows as BPMN 2.0, the diagram the customer sees and signs before anything is built.
erp-compose bpmn --from modules/ar/module.yaml [--workflow invoice] --out process.bpmn
erp-compose bpmn --from ERP_INVOICING.md --out processes/invoicing.bpmn
Two inputs, one model:
(a) a module spec's `workflows:` (module.yaml, or the served spec.yaml) — per-entity state machines
{entity, states, transitions[{from, to, by, gate, requires, effect}], note}, actors from `actors:`;
(b) the `## Workflows` section of a customer's ERP_<MODULE>.md, written during Discover in the format
modules/template.md documents (parse_md below): a `### <entity>` heading, a `States:` line and a table
| Key | From | To | Who | Gate / requires | Effect | Timer | Approval |, actors from `## Actors`.
A malformed table fails with the file's line number (WorkflowError).
so a module with NO brick still has a diagram to show: the reason this exists (seemo, 2026-10-18: a module was
designed, the markdown signed, and no diagram ever shown, because only brick-hr could draw one).
THE MAPPING (ported from the blog's lib/bpmn.js, which draws /modules/<id>/process.bpmn):
participant + process one per workflow (entity), in one collaboration
lane one per actor that makes a transition (the first `by`), named from the actors
start event before the transition from `none` (or the first listed state)
task one per transition; userTask for a person, serviceTask for `system`, task for both;
gate / requires / effect in its documentation; `dodil:step="<key>"`
approval a transition marked approval: a userTask with `dodil:approval="true"`
timer a transition with a timer: an intermediate timer catch event before its task
exclusive gateway where a state has several exits, and after a transition with several `to`
end event for a state nothing leaves (ERP_<MODULE>.md: it must be listed under `Terminal:`)
self-step From = To (or To empty): a task in its state's sequence, by row order, back to the same
state; with a gate it is OPTIONAL (Gateway_<key>__if → task → Gateway_<key>__if_end)
terminal with exits `Terminal:` / `terminal:` — a choice between ending there and its exits
external actor `(external)` after its name in `## Actors` / `external: true`: a receiveTask in its own
lane that the process waits on, with a message flow from a collapsed pool of its own
STABLE IDS (modules/pattern.md "Stable ids are designed, not derived"; the scheme brick-hr's show_playbook_bpmn
ships) — every id comes from keys, never positions, so adding a step adds its own elements and touches only the
flows (and the one choice gateway) at the states it joins:
Task_<key> a transition's key: `key:` when given (recommended, frozen once signed), else
derived as <entity>_<from>_to_<to> (lists joined with _or_)
Timer_<key> its timer
Gateway_<key>__if the outcome choice after a transition with several `to`
Gateway_<entity>_<state>__if the choice at a state with several exits
Start_<entity>, End_<entity>_<state>, Lane_<entity>_<actor>, Participant_<process>_<entity>
Flow_<source id>__<target id>
Process_<process>_<entity>_v<n>, Collaboration_<process>_v<n>, Definitions_<process>_v<n> (the only versioned ids)
The same input gives the same bytes (no clock, no random id, a deterministic layout: columns by distance from the
start, rows by lane). Standard library only, like signoff: the document is what `erp-compose signoff record`
hashes, and `signoff verify` diffs by these ids.
"""
import argparse
import pathlib
import re
import sys
import xml.etree.ElementTree as ET
from dataclasses import dataclass, field
NS = {
"bpmn": "http://www.omg.org/spec/BPMN/20100524/MODEL",
"bpmndi": "http://www.omg.org/spec/BPMN/20100524/DI",
"dc": "http://www.omg.org/spec/DD/20100524/DC",
"di": "http://www.omg.org/spec/DD/20100524/DI",
"xsi": "http://www.w3.org/2001/XMLSchema-instance",
"dodil": "https://git.dodil.io/dodil/erp-platform/bpmn/1", # the same namespace brick-hr writes
}
TARGET_NAMESPACE = "https://git.dodil.io/dodil/erp-platform/processes"
SYSTEM_ACTORS = {"system", "calling_module"}
KEY = re.compile(r"^[a-z][a-z0-9_]*$")
class WorkflowError(Exception):
"""A workflow description that cannot be drawn. The message names the file and line where there is one."""
# ── the model ─────────────────────────────────────────────────────────────────────────────
@dataclass
class Transition:
key: str
frm: list
to: list
by: list
name: str = ""
gate: str = ""
requires: str = ""
effect: str = ""
timer: str = ""
approval: bool = False
line: int = 0
@dataclass
class Workflow:
entity: str
states: list
transitions: list
note: str = ""
states_text: str = ""
terminal: list = field(default_factory=list) # states the process may END in, even with exits
@dataclass
class Spec:
process: str
version: int
name: str
workflows: list
actors: dict = field(default_factory=dict) # id → name
source: str = ""
external: set = field(default_factory=set) # actor ids outside the system (a client, a customer)
def _human(s) -> str:
return str(s).replace("_", " ").lower()
def _cap(s: str) -> str:
return s[:1].upper() + s[1:]
def _nc(s) -> str:
out = re.sub(r"[^A-Za-z0-9_\-]", "_", str(s).strip())
return out if re.match(r"^[A-Za-z_]", out) else "_" + out
def _slug(s) -> str:
return re.sub(r"[^a-z0-9]+", "_", str(s).strip().lower()).strip("_")
def _list(v) -> list:
if v is None or v == "":
return []
return [str(x).strip() for x in (v if isinstance(v, list) else [v]) if str(x).strip()]
def derived_key(entity: str, frm: list, to: list) -> str:
return _slug(f"{entity}_{'_or_'.join(frm)}_to_{'_or_'.join(to)}")
def _check(spec: Spec, where) -> Spec:
"""What both inputs must satisfy: unique keys, transitions that name states."""
seen = {}
for wf in spec.workflows:
if not wf.transitions:
raise WorkflowError(f"{where(0)}: workflow {wf.entity!r} has no transitions")
for t in wf.transitions:
if not t.frm or not t.to or not t.by:
raise WorkflowError(f"{where(t.line)}: {wf.entity} transition {t.key or '?'} needs from, to and by")
if not KEY.match(t.key):
raise WorkflowError(f"{where(t.line)}: key {t.key!r} is not a slug (a-z, 0-9, _; a letter first)")
if t.key in seen:
raise WorkflowError(f"{where(t.line)}: key {t.key!r} is used twice (first at {where(seen[t.key])}); "
"a key is unique in the document")
seen[t.key] = t.line
return spec
# ── input (a): a module spec ──────────────────────────────────────────────────────────────
def from_module(data: dict, source: str = "module.yaml") -> Spec:
if not isinstance(data, dict) or not isinstance(data.get("workflows"), list):
raise WorkflowError(f"{source}: no `workflows:` list")
actors, external = {}, set()
for a in data.get("actors") or []:
if isinstance(a, dict) and a.get("id"):
actors[str(a["id"])] = str(a.get("name") or a["id"])
if a.get("external"):
external.add(str(a["id"]))
wfs = []
for n, w in enumerate(data["workflows"], 1):
if not isinstance(w, dict) or not w.get("entity"):
raise WorkflowError(f"{source}: workflows[{n}] has no entity")
ts = []
for i, t in enumerate(w.get("transitions") or [], 1):
if not isinstance(t, dict):
raise WorkflowError(f"{source}: workflows[{n}] ({w['entity']}) transition {i} is not a mapping")
frm, to = _list(t.get("from")), _list(t.get("to"))
to = to or (frm if len(frm) == 1 else [])
ts.append(Transition(key=str(t.get("key") or derived_key(w["entity"], frm, to)), frm=frm, to=to,
by=_list(t.get("by")), name=str(t.get("name") or ""), gate=str(t.get("gate") or ""),
requires=str(t.get("requires") or ""), effect=str(t.get("effect") or ""),
timer=str(t.get("timer") or ""), approval=bool(t.get("approval")), line=i))
if not ts:
continue # a state list with no transitions is not a process
states = w.get("states")
wfs.append(Workflow(str(w["entity"]), [str(s) for s in states] if isinstance(states, list) else [], ts,
note=" ".join(str(w.get("note") or "").split()),
states_text="" if isinstance(states, list) else str(states or ""),
terminal=_list(w.get("terminal"))))
if not wfs:
raise WorkflowError(f"{source}: no workflow has transitions")
pid = str(data.get("id") or pathlib.Path(source).parent.name or "module")
return _check(Spec(pid, int(data.get("version") or 1), str(data.get("name") or pid), wfs, actors, source,
external),
lambda line: f"{source}")
# ── input (b): ERP_<MODULE>.md ────────────────────────────────────────────────────────────
COLUMNS = {"key": "key", "step": "name", "name": "name", "from": "from", "to": "to", "who": "by", "by": "by",
"gate / requires": "gate", "gate": "gate", "requires": "requires", "effect": "effect", "timer": "timer",
"approval": "approval"}
_SPLIT = re.compile(r"\s*(?:,|→|->|\||\bor\b)\s*")
_YES = {"yes", "y", "true", "x", "✓", "approval"}
def _cells(line: str) -> list:
s = line.strip()
if s.startswith("|"):
s = s[1:]
if s.endswith("|"):
s = s[:-1]
return [c.strip() for c in s.split("|")]
def _split(v: str) -> list:
return [x for x in (p.strip().strip("`") for p in _SPLIT.split(v or "")) if x]
def parse_md(text: str, source: str = "ERP_MODULE.md") -> Spec:
"""The `## Workflows` section of an ERP_<MODULE>.md (format: modules/template.md). Line numbers are the file's."""
lines = text.splitlines()
where = lambda n: f"{source}:{n}" # noqa: E731
sec = next((i for i, l in enumerate(lines) if re.match(r"^##\s+Workflows\s*$", l)), None)
if sec is None:
raise WorkflowError(f"{source}: no `## Workflows` section (modules/template.md shows the format)")
end = next((i for i in range(sec + 1, len(lines)) if re.match(r"^##\s", lines[i])), len(lines))
m = re.search(r"ERP_([A-Za-z0-9_-]+)\.md$", pathlib.Path(source).name)
process, version = (_slug(m.group(1)).replace("_", "-") if m else "module"), 1
actors, external = _md_actors(lines)
wfs, wf, header, i = [], None, None, sec + 1
while i < end:
raw, n = lines[i], i + 1
s = raw.strip()
i += 1
if not s:
header = None # a table ends at a blank line
continue
if s.startswith(">") or s.startswith("<!--"):
continue
pm = re.match(r"^Process:\s*([A-Za-z0-9_.-]+)(?:\s+v(\d+))?\s*$", s, re.I)
if pm:
process, version = pm.group(1), int(pm.group(2) or 1)
continue
hm = re.match(r"^###\s+(.+?)\s*$", s)
if hm:
ent = _slug(hm.group(1))
if not ent or ent.startswith("entity"):
raise WorkflowError(f"{where(n)}: `### {hm.group(1)}` — write the entity's name (e.g. ### invoice)")
wf = Workflow(ent, [], [])
wf.line = n
wfs.append(wf)
header = None
continue
tm = re.match(r"^Terminal:\s*(.*)$", s, re.I)
if tm:
if wf is None or not wf.states:
raise WorkflowError(f"{where(n)}: `Terminal:` goes under `States:` in a `### <entity>`")
wf.terminal = _split(tm.group(1))
bad = [x for x in wf.terminal if x not in wf.states]
if bad or not wf.terminal:
raise WorkflowError(f"{where(n)}: Terminal {', '.join(bad) or '(empty)'} not in States "
f"({', '.join(wf.states)})")
continue
sm = re.match(r"^States:\s*(.*)$", s, re.I)
if sm:
if wf is None:
raise WorkflowError(f"{where(n)}: `States:` before any `### <entity>` heading")
wf.states = _split(sm.group(1))
wf.states_line = n
if not wf.states:
raise WorkflowError(f"{where(n)}: `States:` lists no states")
continue
if s.startswith("|"):
if wf is None:
raise WorkflowError(f"{where(n)}: a table before any `### <entity>` heading")
cells = _cells(s)
if header is None:
header = []
for c in cells:
k = COLUMNS.get(c.lower())
if k is None:
raise WorkflowError(f"{where(n)}: unknown column {c!r}; the columns are Key, Step, From, To, "
"Who, Gate / requires, Effect, Timer, Approval")
header.append(k)
missing = {"from", "to", "by"} - set(header)
if missing:
raise WorkflowError(f"{where(n)}: the table needs columns From, To and Who")
if i < end and re.match(r"^\|?\s*:?-{2,}", lines[i].strip()):
i += 1
continue
if re.match(r"^\|?\s*:?-{2,}", s):
continue
if len(cells) != len(header):
raise WorkflowError(f"{where(n)}: {len(cells)} cells, the header has {len(header)} "
"(a `|` inside a cell must be written `/`)")
row = dict(zip(header, cells))
frm, to, by = _split(row.get("from")), _split(row.get("to")), [_slug(b) for b in _split(row.get("by"))]
to = to or (frm if len(frm) == 1 else []) # a step that changes no state: To empty = From
if not frm or not to or not by:
raise WorkflowError(f"{where(n)}: From and Who are required, and To unless the step stays in its "
"one From state")
known = set(wf.states) | {"none"}
if not wf.states:
raise WorkflowError(f"{where(n)}: write `States: a, b, c` under `### {wf.entity}` before the table")
bad = [x for x in frm + to if x not in known]
if bad:
raise WorkflowError(f"{where(n)}: {', '.join(bad)} not in States ({', '.join(wf.states)}; `none` "
"is before the first state)")
if actors:
unknown = [b for b in by if b not in actors and b not in SYSTEM_ACTORS]
if unknown:
raise WorkflowError(f"{where(n)}: Who {', '.join(unknown)} is not an actor in `## Actors` "
f"({', '.join(sorted(actors))}) nor `system`")
key = (row.get("key") or "").strip("` ") or derived_key(wf.entity, frm, to)
wf.transitions.append(Transition(
key=key, frm=frm, to=to, by=by, name=row.get("name", ""), gate=row.get("gate", ""),
requires=row.get("requires", ""), effect=row.get("effect", ""), timer=row.get("timer", ""),
approval=(row.get("approval", "").strip().lower() in _YES), line=n))
continue
# prose between tables is allowed (the customer's own explanation); kept out of the model
for w in wfs:
if not w.transitions:
raise WorkflowError(f"{where(w.line)}: `### {w.entity}` has no transition table")
exits = {f for t in w.transitions if not is_self(t) for f in t.frm}
for t in w.transitions:
if is_self(t):
continue
for x in t.to:
if x not in exits and x not in w.terminal:
raise WorkflowError(f"{where(t.line)}: {w.entity} state {x!r} has no exit and is not terminal: "
f"add it to `Terminal:` under States, or give it a row out of it")
if not wfs:
raise WorkflowError(f"{where(sec + 1)}: `## Workflows` has no `### <entity>` with a table")
spec = Spec(process, version, process.replace("-", " ").title(), wfs, actors, source, external)
return _check(spec, lambda line: where(line or sec + 1))
def _md_actors(lines: list) -> tuple:
"""`## Actors` table: the first column is the actor (its slug is the id `Who` names). An actor outside the
system — a client, a customer — is marked `(external)` after its name: `| Client contact (external) | … |`."""
sec = next((i for i, l in enumerate(lines) if re.match(r"^##\s+Actors\s*$", l)), None)
if sec is None:
return {}, set()
out, external, header = {}, set(), False
for l in lines[sec + 1:]:
s = l.strip()
if re.match(r"^##\s", s):
break
if not s.startswith("|"):
continue
if not header:
header = True
continue
if re.match(r"^\|?\s*:?-{2,}", s):
continue
name = _cells(s)[0].strip("*` ")
ext = re.search(r"\(\s*external\s*\)\s*$", name, re.I)
if ext:
name = name[:ext.start()].strip("*` ")
if name and not name.startswith("<"):
out[_slug(name)] = name
if ext:
external.add(_slug(name))
return out, external
def load(path, workflow=None) -> Spec:
p = pathlib.Path(path)
try:
text = p.read_text()
except FileNotFoundError:
raise WorkflowError(f"{path}: no such file") from None
if p.suffix.lower() in (".md", ".markdown"):
spec = parse_md(text, str(path))
else:
import yaml
try:
spec = from_module(yaml.safe_load(text), str(path))
except yaml.YAMLError as e:
raise WorkflowError(f"{path}: not YAML ({e})") from None
if workflow:
keep = [w for w in spec.workflows if w.entity == workflow]
if not keep:
raise WorkflowError(f"{path}: no workflow {workflow!r} (there are: "
f"{', '.join(w.entity for w in spec.workflows)})")
spec.workflows = keep
return spec
# ── the graph ─────────────────────────────────────────────────────────────────────────────
VERB = {
"approved": "Approve the {e}", "cancelled": "Cancel the {e}", "changes_requested": "Request changes to the {e}",
"declined": "Decline the {e}", "expired_unsigned": "Let the {e} expire unsigned",
"in_approval": "Send the {e} for approval", "out_for_signature": "Send the {e} for signature",
"retired": "Retire the {e}", "signed": "Record the {e} as signed",
"signed_elsewhere": "Record the {e} as signed elsewhere", "superseded": "Supersede the {e}",
"terminated": "Terminate the {e}", "accepted": "Accept the {e}", "allocated": "Allocate the {e}",
"awaiting_approval": "Submit the {e} for approval", "closed": "Close the {e}", "closed_lost": "Close the {e} as lost",
"closed_won": "Close the {e} as won", "closed_short": "Close the {e} short", "converted": "Convert the {e}",
"disputed": "Dispute the {e}", "disqualified": "Disqualify the {e}", "done": "Complete the {e}",
"ending": "Record the end of the {e}", "escalated": "Escalate the {e}", "expired": "Expire the {e}",
"issued": "Issue the {e}", "match_failed": "Fail the {e} match", "matched": "Match the {e}",
"needs_approval": "Send the {e} for approval", "ordered": "Order the {e}", "paid": "Settle the {e}",
"part_received": "Receive part of the {e}", "posted": "Post the {e}", "promised": "Record a promise on the {e}",
"qualified": "Qualify the {e}", "raised": "Raise the {e}", "received": "Receive the {e}",
"refused": "Refuse the {e}", "rejected": "Reject the {e}", "resolved": "Resolve the {e}",
"returned": "Return the {e}", "reversed": "Reverse the {e}", "revoked": "Revoke the {e}", "sent": "Send the {e}",
"skipped": "Skip the {e}", "submitted": "Submit the {e}", "void": "Void the {e}", "waiting": "Put the {e} on wait",
"working": "Start working the {e}", "draft": "Return the {e} to draft", "future": "Record the {e}",
"open": "Reopen the {e}", "out": "Start the {e}", "invoiced": "Invoice the {e}", "active": "Start the {e}",
"completed": "Complete the {e}", "on_hold": "Put the {e} on hold",
}
CREATE = {"draft": "Draft the {e}", "open": "Start the {e}", "future": "Record the {e}", "issued": "Issue the {e}",
"posted": "Post the {e}", "raised": "Raise the {e}", "out": "Start the {e}", "active": "Start the {e}",
"planned": "Plan the {e}", "proposed": "Propose the {e}"}
def task_name(t: Transition, entity: str) -> str:
if t.name:
return t.name
e = _human(entity)
if len(t.to) > 1:
return f"Decide on the {e}"
k = _slug(t.to[0])
if t.to[0] in t.frm:
return f"Update the {e}"
tpl = ("none" in t.frm and CREATE.get(k)) or VERB.get(k) or "Move the {e} to " + _human(t.to[0])
return _cap(tpl.replace("{e}", e))
@dataclass
class Node:
id: str
tag: str
name: str = ""
lane: str = ""
doc: str = ""
attrs: dict = field(default_factory=dict)
timer: str = ""
@dataclass
class Flow:
id: str
source: str
target: str
name: str = ""
@dataclass
class Graph:
wf: Workflow
E: str
nodes: dict # id → Node, in insertion order
flows: list
lanes: list # [(actor, lane id, name)]
doc: str
messages: list = field(default_factory=list) # [(external actor, task id)]: they send, the process waits
def is_self(t: Transition) -> bool:
"""A step that changes no state (From = To, one state): drawn in the state's sequence, by row order."""
return len(t.frm) == 1 and t.to == t.frm
def graph(wf: Workflow, spec: Spec) -> Graph:
E = _nc(wf.entity)
nodes, flows, lanes = {}, [], []
outs, selfs, entered, messages = {}, {}, set(), []
for t in wf.transitions:
if is_self(t):
selfs.setdefault(t.frm[0], []).append(t)
continue
for f in t.frm:
outs.setdefault(f, []).append(t)
entered.update(t.to)
has_exit = list(dict.fromkeys([*outs, *selfs]))
listed = wf.states or []
initial = "none" if "none" in has_exit else next((s for s in listed if s in has_exit), wf.transitions[0].frm[0])
derived = [s for s in listed if s not in entered and s != initial and s not in has_exit]
def lane(actor):
if actor not in [a for a, _, _ in lanes]:
name = spec.actors.get(actor) or _cap(_human(actor))
lanes.append((actor, f"Lane_{E}_{_nc(actor)}", f"{name} (external)" if actor in spec.external else name))
return actor
def flow(src, tgt, name=""):
fid = f"Flow_{src}__{tgt}"
if not any(f.id == fid for f in flows):
flows.append(Flow(fid, src, tgt, name))
entry, task = {}, {}
for t in wf.transitions:
who = [spec.actors.get(b, b) for b in t.by]
doc = " ".join(x for x in (
f"By: {' or '.join(who)}.", f"From {' / '.join(t.frm)} to {' / '.join(t.to)}.",
f"When: {t.gate}." if t.gate else "", f"Requires: {t.requires}." if t.requires else "",
f"Effect: {t.effect}." if t.effect else "", "An approval." if t.approval else "",
f"Waits: {t.timer}." if t.timer else "") if x)
system = [b in SYSTEM_ACTORS for b in t.by]
tag = "serviceTask" if all(system) else "task" if any(system) else "userTask"
tid = f"Task_{_nc(t.key)}"
attrs = {"dodil:step": t.key}
if t.approval:
attrs["dodil:approval"] = "true"
if t.by[0] in spec.external:
# done outside the system: the process WAITS for it (a receive task), and the party is a pool of its
# own whose message reaches the task — what they send is part of the signed process
tag = "receiveTask"
attrs["dodil:external"] = "true"
doc += " Done outside the system: the process waits for it."
messages.append((t.by[0], tid))
nodes[tid] = Node(tid, tag, task_name(t, wf.entity), lane(t.by[0]), doc, attrs)
task[t.key] = tid
entry[t.key] = tid
if t.timer:
mid = f"Timer_{_nc(t.key)}"
nodes[mid] = Node(mid, "intermediateCatchEvent", t.timer, t.by[0], timer=t.timer)
flow(mid, tid)
entry[t.key] = mid
point = {}
def point_of(state):
if state in point:
return point[state]
chain = selfs.get(state, [])
if chain: # the state's own steps first, by row order, then its exits
ends = []
for t in chain:
tid = task[t.key]
if t.gate: # optional: only when its gate holds; otherwise straight to the rejoin
gw, join = f"Gateway_{_nc(t.key)}__if", f"Gateway_{_nc(t.key)}__if_end"
nodes[gw] = Node(gw, "exclusiveGateway", f"{t.gate}?", nodes[tid].lane)
nodes[join] = Node(join, "exclusiveGateway", "", nodes[tid].lane)
flow(gw, entry[t.key], t.gate)
flow(gw, join, "otherwise")
flow(tid, join)
ends.append((gw, join))
else:
ends.append((entry[t.key], tid))
point[state] = ends[0][0]
for (_, out), (nxt, _) in zip(ends, ends[1:]):
flow(out, nxt)
flow(ends[-1][1], exit_of(state))
return point[state]
point[state] = exit_of(state)
return point[state]
def exit_of(state):
o = outs.get(state, [])
if not o:
nid = f"End_{E}_{_nc(state)}"
nodes[nid] = Node(nid, "endEvent", f"{_cap(_human(wf.entity))} {_human(state)}")
elif state in wf.terminal: # it may end here, or go on by one of its exits
nid = f"Gateway_{E}_{_nc(state)}__if"
nodes[nid] = Node(nid, "exclusiveGateway", f"{_cap(_human(wf.entity))} {_human(state)}: done, or go on?")
end = f"End_{E}_{_nc(state)}"
nodes[end] = Node(end, "endEvent", f"{_cap(_human(wf.entity))} {_human(state)}")
for t in o:
flow(nid, entry[t.key])
flow(nid, end, "done")
return nid
elif len(o) == 1:
nid = entry[o[0].key]
else:
nid = f"Gateway_{E}_{_nc(state)}__if"
nodes[nid] = Node(nid, "exclusiveGateway", f"New {_human(wf.entity)}?" if state == "none"
else f"{_cap(_human(wf.entity))} {_human(state)}: what next?")
if len(o) > 1:
for t in o:
flow(nid, entry[t.key])
return nid
start = f"Start_{E}"
start_node = Node(start, "startEvent", f"{_cap(_human(wf.entity))} needed" if initial == "none"
else f"{_cap(_human(wf.entity))} {_human(initial)}")
nodes = {start: start_node, **nodes}
flow(start, point_of(initial))
for t in wf.transitions:
if is_self(t):
continue # point_of(its state) sequences it
if len(t.to) == 1:
flow(task[t.key], point_of(t.to[0]))
continue
gid = f"Gateway_{_nc(t.key)}__if"
nodes[gid] = Node(gid, "exclusiveGateway", "Outcome?", nodes[task[t.key]].lane)
flow(task[t.key], gid)
for s in t.to:
flow(gid, point_of(s), _human(s))
# a state reachable only as a source (never entered, not initial) still needs its exits drawn
for s in has_exit:
if s not in point and s != initial:
point_of(s)
src = point[s]
if not any(f.target == src for f in flows):
sid = f"Start_{E}_{_nc(s)}"
nodes[sid] = Node(sid, "startEvent", f"{_cap(_human(wf.entity))} {_human(s)}")
flow(sid, src)
# events and gateways sit in the lane of a neighbouring task
for _ in range(4):
for n in nodes.values():
if n.lane:
continue
f = next((f for f in flows if f.source == n.id and nodes[f.target].lane), None) or \
next((f for f in flows if f.target == n.id and nodes[f.source].lane), None)
if f:
n.lane = nodes[f.target if f.source == n.id else f.source].lane
for n in nodes.values():
n.lane = n.lane or lanes[0][0]
states = ", ".join(listed) if listed else wf.states_text or ", ".join(dict.fromkeys(
s for t in wf.transitions for s in t.frm + t.to if s != "none"))
doc = " ".join(x for x in (
f"Lifecycle of the {_human(wf.entity)}, generated from {pathlib.Path(spec.source).name} by erp-compose bpmn; "
"change the workflow there, not this file.", f"States: {states}.",
f"Derived states (never set by a transition): {', '.join(derived)}." if derived else "",
f"Note: {wf.note}" if wf.note else "") if x)
return Graph(wf, E, nodes, flows, lanes, doc, messages)
# ── layout: columns by distance from the start, rows by lane ──────────────────────────────
SIZE = {"startEvent": (36, 36), "endEvent": (36, 36), "intermediateCatchEvent": (36, 36),
"exclusiveGateway": (50, 50)}
COL_W, SLOT_H, POOL_LABEL, LANE_LABEL, GAP = 170, 120, 30, 30, 40
def _columns(g: Graph) -> dict:
first = next(iter(g.nodes))
col = {first: 0}
queue = [first]
while queue:
nid = queue.pop(0)
for f in g.flows:
if f.source == nid and f.target not in col:
col[f.target] = col[nid] + 1
queue.append(f.target)
for nid in g.nodes: # other start events: before what they lead to
if nid not in col:
nxt = [col[f.target] for f in g.flows if f.source == nid and f.target in col]
col[nid] = max(0, min(nxt) - 1) if nxt else 0
return col
def _layout(g: Graph, y0: int):
col = _columns(g)
cells = {}
for n in g.nodes.values():
cells.setdefault((n.lane, col[n.id]), []).append(n)
lane_h = {a: SLOT_H for a, _, _ in g.lanes}
for (lane, _c), ns in cells.items():
lane_h[lane] = max(lane_h[lane], len(ns) * SLOT_H + 20)
lane_y, y = {}, y0
for a, _, _ in g.lanes:
lane_y[a] = y
y += lane_h[a]
x0 = POOL_LABEL + LANE_LABEL + 30
width = x0 + (max(col.values()) + 1) * COL_W + 20
bounds = {}
for (lane, c), ns in sorted(cells.items(), key=lambda kv: (kv[0][1], kv[0][0])):
cx, yy = x0 + c * COL_W + COL_W // 2, lane_y[lane] + 20
for n in ns:
w, h = SIZE.get(n.tag, (100, 80))
bounds[n.id] = (cx - w // 2, yy + (80 - h) // 2, w, h)
yy += SLOT_H
return bounds, lane_y, lane_h, width, y - y0
def _edge(s, t, back):
sx, sy, sw, sh = s
tx, ty, tw, th = t
scy, tcy = sy + sh // 2, ty + th // 2
if tx > sx + sw:
if scy == tcy:
return [(sx + sw, scy), (tx, tcy)]
mx = tx - 25
return [(sx + sw, scy), (mx, scy), (mx, tcy), (tx, tcy)]
below = max(sy + sh, ty + th) + 15 + (back % 4) * 6
return [(sx + sw // 2, sy + sh), (sx + sw // 2, below), (tx + tw // 2, below), (tx + tw // 2, ty + th)]
# ── XML ───────────────────────────────────────────────────────────────────────────────────
def _el(parent, tag, attrs=None, text=None):
e = ET.SubElement(parent, tag, {k: str(v) for k, v in (attrs or {}).items() if v not in (None, "")})
if text is not None:
e.text = text
return e
def to_bpmn(spec: Spec) -> str:
"""The spec's workflows as one BPMN 2.0 document (UTF-8 text): the same spec gives the same bytes."""
pid = _nc(spec.process)
v = spec.version
root = ET.Element("bpmn:definitions", {**{f"xmlns:{k}": u for k, u in NS.items()},
"id": f"Definitions_{pid}_v{v}", "name": f"{spec.name} processes",
"targetNamespace": TARGET_NAMESPACE,
"exporter": "erp-platform erp-compose bpmn", "exporterVersion": "1"})
graphs = [graph(w, spec) for w in spec.workflows]
collab = _el(root, "bpmn:collaboration", {"id": f"Collaboration_{pid}_v{v}"})
_el(collab, "bpmn:documentation", text=f"{spec.name}: one pool per entity lifecycle, generated from "
f"{pathlib.Path(spec.source).name}.")
procs = []
for g in graphs:
proc_id = f"Process_{pid}_{g.E}_v{v}"
_el(collab, "bpmn:participant", {"id": f"Participant_{pid}_{g.E}", "name": _cap(_human(g.wf.entity)),
"processRef": proc_id})
procs.append(proc_id)
outside = list(dict.fromkeys(a for g in graphs for a, _ in g.messages))
for a in outside: # a collapsed pool per party outside the system
p = _el(collab, "bpmn:participant", {"id": f"Participant_{_nc(a)}",
"name": f"{spec.actors.get(a) or _cap(_human(a))} (external)"})
_el(p, "bpmn:documentation", text="Outside the system: the steps it sends are what the process waits on.")
msgs = [(f"Flow_Participant_{_nc(a)}__{tid}", f"Participant_{_nc(a)}", tid) for g in graphs for a, tid in g.messages]
for fid, src, tgt in msgs:
_el(collab, "bpmn:messageFlow", {"id": fid, "sourceRef": src, "targetRef": tgt})
for g, proc_id in zip(graphs, procs):
proc = _el(root, "bpmn:process", {"id": proc_id, "name": _cap(_human(g.wf.entity)), "isExecutable": "false"})
_el(proc, "bpmn:documentation", text=g.doc)
ls = _el(proc, "bpmn:laneSet", {"id": f"LaneSet_{g.E}"})
for actor, lid, name in g.lanes:
lane = _el(ls, "bpmn:lane", {"id": lid, "name": name})
for n in g.nodes.values():
if n.lane == actor:
_el(lane, "bpmn:flowNodeRef", text=n.id)
inc, out = {}, {}
for f in g.flows:
out.setdefault(f.source, []).append(f.id)
inc.setdefault(f.target, []).append(f.id)
for n in g.nodes.values():
attrs = {"id": n.id, "name": n.name, **n.attrs}
if n.tag == "serviceTask":
attrs["implementation"] = "##unspecified"
e = _el(proc, f"bpmn:{n.tag}", attrs)
if n.doc:
_el(e, "bpmn:documentation", text=n.doc)
for fid in inc.get(n.id, []):
_el(e, "bpmn:incoming", text=fid)
for fid in out.get(n.id, []):
_el(e, "bpmn:outgoing", text=fid)
if n.tag == "intermediateCatchEvent":
td = _el(e, "bpmn:timerEventDefinition", {"id": f"{n.id}__definition"})
_el(td, "bpmn:timeDuration", {"xsi:type": "bpmn:tFormalExpression"}, n.timer)
for f in g.flows:
_el(proc, "bpmn:sequenceFlow", {"id": f.id, "name": f.name, "sourceRef": f.source, "targetRef": f.target})
diagram = _el(root, "bpmndi:BPMNDiagram", {"id": f"BPMNDiagram_{pid}_v{v}"})
plane = _el(diagram, "bpmndi:BPMNPlane", {"id": f"BPMNPlane_{pid}_v{v}", "bpmnElement": f"Collaboration_{pid}_v{v}"})
def shape(eid, b, horizontal=False):
s = _el(plane, "bpmndi:BPMNShape", {"id": f"{eid}_di", "bpmnElement": eid,
"isHorizontal": "true" if horizontal else None})
_el(s, "dc:Bounds", {"x": b[0], "y": b[1], "width": b[2], "height": b[3]})
y, placed, width_all = 0, {}, 0
for g in graphs:
bounds, lane_y, lane_h, width, height = _layout(g, y)
shape(f"Participant_{pid}_{g.E}", (0, y, width, height), True)
for actor, lid, _ in g.lanes:
shape(lid, (POOL_LABEL, lane_y[actor], width - POOL_LABEL, lane_h[actor]), True)
for nid in g.nodes:
shape(nid, bounds[nid])
placed.update(bounds)
width_all = max(width_all, width)
back = 0
for f in g.flows:
pts = _edge(bounds[f.source], bounds[f.target], back)
back += len(pts) == 4 and bounds[f.target][0] <= bounds[f.source][0] + bounds[f.source][2]
e = _el(plane, "bpmndi:BPMNEdge", {"id": f"{f.id}_di", "bpmnElement": f.id})
for px, py in pts:
_el(e, "di:waypoint", {"x": px, "y": py})
y += height + GAP
tops = {}
for a in outside:
tops[f"Participant_{_nc(a)}"] = y
shape(f"Participant_{_nc(a)}", (0, y, width_all, 60), True)
y += 60 + GAP
for fid, src, tgt in msgs:
bx, by_, bw, bh = placed[tgt]
e = _el(plane, "bpmndi:BPMNEdge", {"id": f"{fid}_di", "bpmnElement": fid})
cx = bx + bw // 2
for px, py in ((cx, tops[src]), (cx, by_ + bh)):
_el(e, "di:waypoint", {"x": px, "y": py})
ET.indent(root, space=" ")
return '<?xml version="1.0" encoding="UTF-8"?>\n' + ET.tostring(root, encoding="unicode") + "\n"
def generate(path, workflow=None) -> str:
return to_bpmn(load(path, workflow))
# ── the command ───────────────────────────────────────────────────────────────────────────
def main(argv=None) -> int:
ap = argparse.ArgumentParser(prog="erp-compose bpmn", description="A module's workflows as BPMN 2.0: the "
"diagram the customer sees and signs (erp-compose signoff record).")
ap.add_argument("--from", dest="src", required=True, help="module.yaml / spec.yaml, or ERP_<MODULE>.md")
ap.add_argument("--workflow", help="one workflow (its entity); default all")
ap.add_argument("--out", help="where to write the .bpmn (default: stdout)")
a = ap.parse_args(argv)
try:
xml = generate(a.src, a.workflow)
except WorkflowError as e:
print(f"bpmn: {e}", file=sys.stderr)
return 1
if not a.out:
sys.stdout.write(xml)
return 0
out = pathlib.Path(a.out)
out.parent.mkdir(parents=True, exist_ok=True)
out.write_text(xml)
from erp_platform.signoff import content_hash
print(f"bpmn: {a.src} → {out} (content sha256 {content_hash(xml)[:8]}: what `erp-compose signoff record` signs)")
return 0
if __name__ == "__main__":
raise SystemExit(main())