"""PlaybookAsBpmn — a domain service: one playbook VERSION → a BPMN 2.0 definitions document (XML), so the customer,
their auditors, any BPMN tool (Camunda Modeler, Signavio, bpmn.io) and an agent can see exactly what the configured
flow does, with its inputs and outputs. It reads the playbook and changes nothing; the same playbook and setting
always give the same bytes (no clock, no random ids). Every id is derived from a step's KEY, never its position,
so inserting or removing a step changes no other element's id: Task_<key>, Timer_<key>, Gateway_<key>__if,
Gateway_<key>__fork, Task_<key>__draft, Flow_<source>__<target>… — what a scenario path, a signed change item and
a process-mining join can rely on.
The mapping (README "Your process as BPMN 2.0"):
process the playbook: name, direction (onboarding | offboarding | conversion | change), version
lane a responsible group (hr, it, finance, manager, buddy, starter, director…), the row's values
userTask a task, an approval, the starter's intake form, and "Approve: <draft>" after a draft
serviceTask an action or document step calling a capability (capabilities.yaml), named by the capability;
with no provider wired it becomes a MANUAL task for the same group carrying the same input
(rule 8) — said in its documentation. A fan-out step is a parallel multi-instance task.
sequenceFlow depends_on: start → the steps with no dependency, the steps nobody depends on → end; parallel
gateways where a step forks to several or waits on several
exclusiveGateway a step's `if` condition (worker_type_is, via_company, rate_outside_band, client_size_is), with
the condition in words; and each `requires` gate (pack_requires): recorded → on, otherwise
"Record the …" first (the step is blocked until the record exists)
timer catch event a step's timing rule before it, e.g. "3 working days before start_date, 09:00 Europe/London"
messageFlow to a collapsed participant per outside system, named by capability family, never a vendor
dataObject the starter's details (intake fields), each draft, each contract — with data input / output
associations where a step reads or writes them; every service task's ioSpecification lists
its inputs (the step's input_template fields, else the capability's input) and its outputs
(the capability's output)
BPMNDiagram a deterministic layout: columns by dependency depth, rows by lane
dodil: attributes (namespace DODIL_NS; BPMN's anyAttribute ##other, so the document stays XSD-valid)
dodil:step = the step key on its task; dodil:pcf = the APQC PCF id of the process and of a step
that maps to one (domain/standards.py); dodil:deviation = variation | warning on a step that
differs from the pack's reference playbook (domain/deviations.py), with a documentation note
Plain Python and the standard library's xml only: no I/O, no framework.
"""
import re
import xml.etree.ElementTree as ET
from dataclasses import dataclass, field
from typing import Optional
from brick_hr.domain import standards
from brick_hr.domain.playbook import Playbook, PlaybookStep
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",
}
DODIL_NS = NS["dodil"]
TARGET_NAMESPACE = "https://git.dodil.io/dodil/brick-hr/playbooks"
FAMILIES = {"identity": "Identity (work accounts)", "groups": "Groups", "chat": "Chat spaces", "drive": "Files (drive)",
"mail": "Mail", "calendar": "Calendar", "contracts": "Contracts (signature)",
"drafting": "Drafting (Models)", "starter": "The starter (public form)"}
RECORDS_IN = "hr" # who records a compliance record (record_compliance)
_PLACE = re.compile(r"\{([a-z_][a-z0-9_.]*)\}")
# Layout (pixels). Columns by dependency depth, rows by lane; a lane grows a row for each node in one column.
X0, Y0, POOL_HEAD, LANE_HEAD, PAD, COL_W, ROW_H = 150, 80, 30, 30, 20, 150, 120
SIZE = {"userTask": (100, 80), "serviceTask": (100, 80), "startEvent": (36, 36), "endEvent": (36, 36),
"intermediateCatchEvent": (36, 36), "exclusiveGateway": (50, 50), "parallelGateway": (50, 50),
"dataObjectReference": (36, 50)}
PARTICIPANT_W, PARTICIPANT_H = 220, 60
@dataclass(frozen=True)
class Capability:
"""What capabilities.yaml says about one capability — the part a reader of the process needs."""
name: str
summary: str = ""
inputs: tuple = ()
outputs: tuple = ()
destructive: bool = False
read: bool = False
@dataclass(frozen=True)
class Setting:
"""The configuration around a playbook that the picture needs: the direction its events start, the org's zone,
the labels of groups, drafts and documents, the intake fields, and the capability contract."""
direction: str
timezone: str = "UTC"
groups: dict = field(default_factory=dict) # group_code → label
drafts: dict = field(default_factory=dict) # draft_code → (label, fields)
documents: dict = field(default_factory=dict) # document_type → label
intake_fields: tuple = () # (field, label, worker_types) — names only, never values
capabilities: dict = field(default_factory=dict) # name → Capability
exporter_version: str = ""
chase: dict = field(default_factory=dict) # step kind (task | approval | intake) → cadence in words
signer: str = "" # who signs a contract for the company, in words
@dataclass
class _Node:
id: str
tag: str
name: str
lane: str
doc: str = ""
timer: Optional[str] = None
inputs: list = field(default_factory=list) # (field, where it comes from)
outputs: list = field(default_factory=list) # field names
loop: Optional[str] = None # a parallel multi-instance, in words
default: Optional[str] = None # a gateway's default flow
attrs: dict = field(default_factory=dict) # dodil: extension attributes
@dataclass
class _Flow:
id: str
source: str
target: str
name: str = ""
condition: Optional[str] = None
@dataclass
class _Data:
id: str # the dataObjectReference's id
name: str
doc: str
lane: str = ""
writer: Optional[str] = None
@dataclass
class _Assoc:
id: str
kind: str # "in" (data object → a task's input) | "out" (a task's output → data object)
node: str
data: str
ports: list # the task's dataInput (one) / dataOutput fields
@dataclass
class _Message:
id: str
node: str
family: str
inbound: bool = False # from the participant to the task (the starter's form arriving)
name: str = ""
# ── words ─────────────────────────────────────────────────────────────────────────────────────────────────────
def _words(code: str) -> str:
return (code or "").replace("_", " ")
def _one_of(args: list) -> str:
args = [str(a) for a in args or []]
if len(args) == 1:
return f"is {args[0]}"
return "is one of " + ", ".join(args)
def condition_text(spec: dict) -> str:
"""One named specification (playbook SKIP_SPECS), in words."""
s, args = spec.get("spec"), spec.get("args") or []
if s == "worker_type_is":
who = "the previous worker type" if spec.get("of") == "previous" else "the worker type"
return f"{who} {_one_of(args)}"
if s == "via_company":
return "they work through their own company"
if s == "rate_outside_band":
return "the agreed rate is outside the pay band"
if s == "client_size_is":
return "our size as the client " + _one_of(args)
if s == "reason_is":
return "the reason " + _one_of(args)
return _words(str(s))
def conditions_text(specs: list) -> str:
return " and ".join(condition_text(s) for s in specs or [])
def timing_text(step: PlaybookStep, timezone: str, names: Optional[dict] = None) -> Optional[str]:
"""The step's timing rule in words, or None when it is due as soon as it is reached. `names` (step number →
title) names the step a step:<n>.done anchor refers to."""
t = step.timing or {}
anchor = t.get("anchor") or "run.created"
offset, unit, at = int(t.get("offset") or 0), t.get("unit") or "days", t.get("time")
if anchor == "run.created" and not offset and not at:
return None
if anchor.startswith("step:"):
ref = anchor[5:].split(".")[0]
what = f"'{names[int(ref)]}' is done" if names and ref.isdigit() and int(ref) in names else f"step {ref} is done"
elif anchor == "run.created":
what = "the run is created"
else:
what = anchor
if not offset:
when = (f"when {what}" if anchor.startswith("step:") or anchor == "run.created" else f"on {what}")
else:
n = abs(offset)
u = ("working day" if unit == "working_days" else "day") + ("" if n == 1 else "s")
when = f"{n} {u} {'before' if offset < 0 else 'after'} {what}"
return f"{when}, {at} {timezone}" if at else f"{when} ({timezone})"
def _show(v) -> str:
if isinstance(v, str):
return v
import json
return json.dumps(v, sort_keys=True, default=str)
def _ncname(s: str) -> str:
out = re.sub(r"[^A-Za-z0-9_.-]", "_", s or "")
return out if re.match(r"[A-Za-z_]", out) else "_" + out
# ── the model: steps → nodes, flows, data, messages ───────────────────────────────────────────────────────────
class _Model:
def __init__(self, pb: Playbook, setting: Setting, marks: Optional[dict] = None):
self.pb, self.setting, self.marks = pb, setting, marks or {}
self.nodes: dict = {}
self.flows: list = []
self.data: dict = {}
self.assocs: list = []
self.messages: list = []
self.lanes: list = []
self._ids: set = set()
# bookkeeping
def lane(self, code: str) -> str:
if code not in self.lanes:
self.lanes.append(code)
return code
def node(self, id: str, tag: str, name: str, lane: str, **kw) -> _Node:
assert id not in self.nodes, id
n = _Node(id, tag, name, self.lane(lane), **kw)
self.nodes[id] = n
return n
def flow(self, source: str, target: str, name: str = "", condition: Optional[str] = None) -> _Flow:
fid, i = f"Flow_{source}__{target}", 1
while fid in self._ids:
i += 1
fid = f"Flow_{source}__{target}_{i}"
self._ids.add(fid)
f = _Flow(fid, source, target, name, condition)
self.flows.append(f)
return f
def data_object(self, key: str, name: str, doc: str) -> _Data:
d = self.data.get(key)
if d is None:
d = self.data[key] = _Data(f"DataObjectReference_{_ncname(key)}", name, doc)
return d
def write(self, node: _Node, port: str, d: _Data) -> None:
"""One output association per (task, data object), carrying every output field that lands in it."""
if d.writer is None:
d.writer, d.lane = node.id, node.lane
aid = f"DataOutputAssociation_{node.id}_{d.id[len('DataObjectReference_'):]}"
for a in self.assocs:
if a.id == aid:
a.ports.append(port)
return
self.assocs.append(_Assoc(aid, "out", node.id, d.id, [port]))
def read(self, node: _Node, port: str, d: _Data) -> None:
aid = f"DataInputAssociation_{node.id}_{_ncname(port)}_{d.id[len('DataObjectReference_'):]}"
if all(a.id != aid for a in self.assocs):
self.assocs.append(_Assoc(aid, "in", node.id, d.id, [port]))
def message(self, node: _Node, family: str, inbound: bool = False, name: str = "") -> None:
self.messages.append(_Message(f"MessageFlow_{node.id}", node.id, family, inbound, name))
# the steps
def build(self) -> None:
pb = self.pb
steps = sorted(pb.steps, key=lambda s: s.step_no)
for s in steps: # lanes in the order the checklist first names them
self.lane(self._lane_of(s))
by = {s.step_no: s for s in steps}
deps = {s.step_no: sorted(d for d in set(s.depends_on) if d in by) for s in steps}
dependents = {n: sorted(m for m in by if n in deps[m]) for n in by}
first = self.lanes[0] if self.lanes else RECORDS_IN
self.node("StartEvent", "startEvent", self._start_name(), first,
doc=f"A {self._event_words()} is recorded: the run is expanded from this playbook version "
f"(open runs keep the version they started with).")
self.node("EndEvent", "endEvent", "Run complete", first,
doc="Every step is done, skipped (its condition did not hold) or cancelled.")
self.key = {s.step_no: s.key for s in steps}
self.names = {s.step_no: s.title for s in steps}
chains = {s.step_no: self._step(s) for s in steps}
for s in steps: # the step's own task carries its key, PCF id, deviation
node = self.nodes[f"Task_{s.key}"]
node.attrs["dodil:step"] = s.key
pcf = standards.step_pcf(s.key, s.kind, s.capability)
if pcf:
node.attrs["dodil:pcf"] = pcf[0]
if s.key in self.marks:
cls, why = self.marks[s.key]
node.attrs["dodil:deviation"] = cls
node.doc = (node.doc + " " if node.doc else "") + \
f"Deviation from the reference ({cls}): {why}."
roots = [n for n in by if not deps[n]]
leaves = [n for n in by if not dependents[n]]
if len(roots) > 1:
self.node("Gateway_start", "parallelGateway", "", first, doc="Every step with no dependency starts "
"at once; each then waits for its own timing rule.")
self.flow("StartEvent", "Gateway_start")
for n in roots:
self.flow("Gateway_start", chains[n][0])
else:
self.flow("StartEvent", chains[roots[0]][0])
fork, join = (lambda n: f"Gateway_{self.key[n]}__fork"), (lambda n: f"Gateway_{self.key[n]}__join")
titles = lambda ns: ", ".join(repr(self.names[m]) for m in ns) # noqa: E731
for n in by:
if len(dependents[n]) > 1:
self.node(fork(n), "parallelGateway", "", self._lane_of(by[n]),
doc=f"'{self.names[n]}' done: {titles(dependents[n])} may go.")
self.flow(chains[n][1], fork(n))
if len(deps[n]) > 1:
self.node(join(n), "parallelGateway", "", self._lane_of(by[n]),
doc=f"'{self.names[n]}' waits for {titles(deps[n])} (done or skipped).")
self.flow(join(n), chains[n][0])
for n in by:
for d in deps[n]:
src = fork(d) if len(dependents[d]) > 1 else chains[d][1]
tgt = join(n) if len(deps[n]) > 1 else chains[n][0]
self.flow(src, tgt)
if len(leaves) > 1:
self.node("Gateway_end", "parallelGateway", "", first, doc="The run is complete when every branch is.")
for n in leaves:
self.flow(chains[n][1], "Gateway_end")
self.flow("Gateway_end", "EndEvent")
else:
self.flow(chains[leaves[0]][1], "EndEvent")
self._reads()
def _lane_of(self, s: PlaybookStep) -> str:
return (s.approval_group or s.responsible_group) if s.kind == "approval" else s.responsible_group
def _event_words(self) -> str:
return " or ".join(_words(e) for e in self.pb.event_types) or "personnel event"
def _start_name(self) -> str:
return f"{self._event_words().capitalize()} recorded"
def _step(self, s: PlaybookStep) -> tuple:
"""The nodes one step becomes, chained; returns (entry id, exit id)."""
k, lane, cond = s.key, self._lane_of(s), s.condition or {}
items: list = [] # ("node", id) | ("gate", gate id, record id, yes text)
text = timing_text(s, self.setting.timezone, self.names)
if text:
self.node(f"Timer_{k}", "intermediateCatchEvent", text.split(",")[0].split(" (")[0], lane, timer=text,
doc=f"'{s.title}' is due {text}. Working days skip weekends and the bank holidays of the "
"person's calendar; moving the anchor date re-times only steps not yet done.")
items.append(("node", f"Timer_{k}"))
for gate in cond.get("requires") or []:
record = gate.get("record") or ""
only = conditions_text(gate.get("if") or [])
gid, rid = f"Gateway_{k}__requires_{_ncname(record)}", f"Task_{k}__record_{_ncname(record)}"
yes = f"a {_words(record)} is recorded" + (f", or it is not required (it is only when {only})"
if only else "")
self.node(gid, "exclusiveGateway", f"{_words(record).capitalize()} recorded?", lane,
doc=f"pack_requires {record}" + (f" when {only}" if only else "") +
f": '{s.title}' is BLOCKED until the record exists.")
rec_lane = RECORDS_IN if RECORDS_IN in self.setting.groups or RECORDS_IN in self.lanes else lane
self.node(rid, "userTask", f"Record the {_words(record)}", rec_lane,
doc=f"record_compliance ({record}): what was checked, when and by whom. The localisation "
"pack records evidence; it makes no legal judgement.")
items.append(("gate", gid, rid, yes))
items += [("node", x) for x in self._work(s, lane)]
def link(a, entry: str) -> None:
if a[0] == "node":
self.flow(a[1], entry)
else:
self.flow(a[1], entry, "recorded", a[3])
f = self.flow(a[1], a[2], "not yet")
self.nodes[a[1]].default = f.id
self.flow(a[2], entry)
for a, b in zip(items, items[1:]):
link(a, b[1])
entry, exit_ = items[0][1], items[-1][1]
ifs = cond.get("if") or []
if ifs:
words = conditions_text(ifs)
gif, gend = f"Gateway_{k}__if", f"Gateway_{k}__if_end"
self.node(gif, "exclusiveGateway", f"{words[0].upper()}{words[1:]}?", lane,
doc=f"'{s.title}' applies only when {words}; otherwise it is skipped (a skipped step counts as "
"done for the steps after it).")
self.node(gend, "exclusiveGateway", "", lane)
self.flow(gif, entry, "yes", words)
skip = self.flow(gif, gend, "no: skipped")
self.nodes[gif].default = skip.id
self.flow(exit_, gend)
entry, exit_ = gif, gend
return entry, exit_
def _work(self, s: PlaybookStep, lane: str) -> list:
"""The task(s) that do the step itself."""
st = self.setting
tid = f"Task_{s.key}"
group = st.groups.get(s.responsible_group) or s.responsible_group
chase = st.chase.get(s.kind)
chased = f" {chase}" if chase else ""
answer = (f" Completing it records the outcome: {' or '.join(s.outcomes)} (output.outcome)."
if s.outcomes else "")
if s.kind == "task":
self.node(tid, "userTask", s.title, lane, doc=f"A task for {group}; done when someone in the group marks it "
f"done (complete_step).{answer}{chased}")
return [tid]
if s.kind == "approval":
self.node(tid, "userTask", s.title, lane,
doc=f"An approval for {st.groups.get(lane) or lane}: granted or refused (approve_step). A "
"refusal cancels the destructive steps that depend on it; nothing destructive runs on a "
f"timer alone.{chased}")
return [tid]
if s.kind == "intake":
node = self.node(tid, "userTask", s.title, lane,
doc="The starter fills in their details through the public link (one form, hashed "
"token, expiring). Values go to the sensitive record and are never shown back; "
f"{group} sees how many fields are done.{chased}")
fields = self._intake_fields()
node.outputs = [f for f, _ in fields]
if fields:
d = self.data_object("intake", "Starter's details",
"The intake fields (names only, values are sensitive): " +
"; ".join(f"{f} ({label})" for f, label in fields))
for f, _ in fields:
self.write(node, f, d)
self.message(node, "starter", inbound=True, name="the form, submitted")
return [tid]
if s.kind == "draft":
label, fields = st.drafts.get(s.draft_instruction) or (_words(s.draft_instruction), ())
did = f"Task_{s.key}__draft"
draft = self.node(did, "serviceTask", f"Draft: {label}", lane,
doc=f"Models drafts the {label.lower()} from the instruction {s.draft_instruction!r}, given "
"only non-sensitive fields. If Models fails, the step is a task: write it by hand.")
draft.inputs = [(f, "") for f in fields]
draft.outputs = ["draft_text"]
d = self.data_object(f"draft_{s.draft_instruction}", f"Draft: {label}",
f"The {label.lower()} as approved (approve_draft); what later steps send.")
self.write(draft, "draft_text", d)
self.message(draft, "drafting", name="draft")
self.node(tid, "userTask", f"Approve: {s.title}", lane,
doc=f"{group} reads the draft, edits it if needed and approves it (approve_draft). Nothing is "
"sent before it is approved.")
return [did, tid]
# action | document: a capability
cap = st.capabilities.get(s.capability) or Capability(s.capability or "")
node = self.node(tid, "serviceTask", s.title, lane)
doc = [f"Capability {cap.name}" + (f": {cap.summary}" if cap.summary else ".")]
if s.kind == "document":
kind = st.documents.get(s.template_kind) or _words(s.template_kind)
doc.append(f"Drafts the {kind} from the approved template (contracts.draft), sends it for signature "
"(contracts.send), waits on contracts.status and files the signed copy on the person "
"(contracts.fetch_executed)." + (f" Signed for the company by {st.signer}." if st.signer else ""))
doc.append(f"With no provider wired it becomes a manual task for {group} carrying the same input and the plan "
"in words; wiring a provider later makes the same step automatic." + answer)
if cap.destructive:
doc.append("Destructive: it runs only after an approval it depends on was GRANTED.")
if s.capability == "identity.delete_user":
doc.append("It also waits until drive.transfer_status says the ownership transfer completed.")
node.doc = " ".join(doc)
node.inputs = self._inputs(s, cap)
node.outputs = list(cap.outputs)
if s.fan_out:
what = _words(s.fan_out.split(":", 1)[1])
node.loop = f"One per {what} row of the person's unit (unit_access), in parallel."
if s.kind == "document" and cap.outputs:
kind = st.documents.get(s.template_kind) or _words(s.template_kind)
d = self.data_object(f"contract_{s.key}", f"Contract: {kind}",
f"The {kind} drafted, signed and filed on the person ({s.template_kind}).")
self.write(node, cap.outputs[0], d)
self.message(node, (cap.name.split(".")[0] or "outside"), name=cap.name)
return [tid]
def _inputs(self, s: PlaybookStep, cap: Capability) -> list:
if not s.input_template:
return [(f, "") for f in cap.inputs]
t = s.input_template
order = [f for f in cap.inputs if f in t] + sorted(k for k in t if k not in cap.inputs)
return [(k, _show(t[k])) for k in order]
def _intake_fields(self) -> list:
wts = set(self.pb.worker_types)
out = []
for f, label, only in self.setting.intake_fields:
if not only or not wts or wts & set(only):
out.append((f, label))
return out
def _reads(self) -> None:
"""A step whose input names a draft or an intake field reads that data object."""
intake = {f for f, _ in self._intake_fields()}
for node in list(self.nodes.values()):
for port, src in node.inputs:
for ref in _PLACE.findall(src or ""):
if ref.startswith("draft.") and f"draft_{ref[6:]}" in self.data:
self.read(node, port, self.data[f"draft_{ref[6:]}"])
elif ref in intake and "intake" in self.data:
self.read(node, port, self.data["intake"])
# ── layout ────────────────────────────────────────────────────────────────────────────────────────────────────
class _Layout:
def __init__(self, m: _Model):
self.m = m
order = list(m.nodes)
succ = {i: [] for i in order}
indeg = {i: 0 for i in order}
for f in m.flows:
succ[f.source].append(f.target)
indeg[f.target] += 1
col = {i: 0 for i in order}
ready = [i for i in order if indeg[i] == 0]
rank = {i: k for k, i in enumerate(order)}
while ready: # longest path from the start: columns by depth
ready.sort(key=rank.get)
i = ready.pop(0)
for t in succ[i]:
col[t] = max(col[t], col[i] + 1)
indeg[t] -= 1
if indeg[t] == 0:
ready.append(t)
self.col = col
cells: dict = {}
self.row = {}
for i in order:
key = (m.nodes[i].lane, col[i])
self.row[i] = len(cells.setdefault(key, []))
cells[key].append(i)
for d in m.data.values(): # a data object sits under the step that writes it
if d.writer is None:
continue
key = (d.lane, col[d.writer])
self.col[d.id] = col[d.writer]
self.row[d.id] = len(cells.setdefault(key, []))
cells[key].append(d.id)
self.lane_rows = {ln: max([len(v) for (l2, _), v in cells.items() if l2 == ln] or [1]) for ln in m.lanes}
self.lane_y, y = {}, Y0
for ln in m.lanes:
self.lane_y[ln] = y
y += self.lane_rows[ln] * ROW_H
self.pool_h = y - Y0
self.ncols = max(self.col.values()) + 1
self.pool_w = POOL_HEAD + LANE_HEAD + PAD + self.ncols * COL_W + PAD
self.x0 = X0 + POOL_HEAD + LANE_HEAD + PAD
self.bounds = {}
for i, node in m.nodes.items():
self.bounds[i] = self._box(node.lane, i, node.tag)
for d in m.data.values():
if d.writer is not None:
self.bounds[d.id] = self._box(d.lane, d.id, "dataObjectReference")
self._participants()
def _box(self, lane: str, i: str, tag: str) -> tuple:
w, h = SIZE[tag]
cx = self.x0 + self.col[i] * COL_W + COL_W // 2
cy = self.lane_y[lane] + self.row[i] * ROW_H + ROW_H // 2
return (cx - w // 2, cy - h // 2, w, h)
def _participants(self) -> None:
firsts: dict = {}
for msg in self.m.messages:
x = self.bounds[msg.node][0] + self.bounds[msg.node][2] // 2
firsts[msg.family] = min(firsts.get(msg.family, x), x)
self.families = sorted(firsts, key=lambda f: (firsts[f], f))
y, right = Y0 + self.pool_h + 100, X0
self.pbounds = {}
for fam in self.families:
x = max(firsts[fam] - PARTICIPANT_W // 2, right)
self.pbounds[fam] = (x, y, PARTICIPANT_W, PARTICIPANT_H)
right = x + PARTICIPANT_W + 30
# waypoints
def cell_bottom(self, i: str) -> int:
return self.lane_y[self.m.nodes[i].lane] + (self.row[i] + 1) * ROW_H
def flow_points(self, f: _Flow) -> list:
sx, sy, sw, sh = self.bounds[f.source]
tx, ty, tw, th = self.bounds[f.target]
scy, tcy = sy + sh // 2, ty + th // 2
if self.col[f.target] - self.col[f.source] == 1:
if scy == tcy:
return [(sx + sw, scy), (tx, tcy)]
mx = (sx + sw + tx) // 2
return [(sx + sw, scy), (mx, scy), (mx, tcy), (tx, tcy)]
# longer than one column: under the source's row, then in the gap just before the target
yb = self.cell_bottom(f.source) - 8
cell_left = self.x0 + self.col[f.target] * COL_W
mx = tx - (tx - cell_left) // 2
scx = sx + sw // 2
return [(scx, sy + sh), (scx, yb), (mx, yb), (mx, tcy), (tx, tcy)]
def message_points(self, msg: _Message) -> list:
x, y, w, h = self.bounds[msg.node]
px, py, pw, _ = self.pbounds[msg.family]
cx = x + w // 2 + 30 # off the column's centre line, so it passes beside the events and gateways
tx = min(max(cx, px + 15), px + pw - 15)
pts = [(cx, y + h), (tx, py)] if tx == cx else [(cx, y + h), (cx, py - 30), (tx, py - 30), (tx, py)]
return list(reversed(pts)) if msg.inbound else pts
def assoc_points(self, a: _Assoc) -> list:
nx, ny, nw, nh = self.bounds[a.node]
if a.data not in self.bounds:
return []
dx, dy, dw, dh = self.bounds[a.data]
if a.kind == "out":
if dy >= ny + nh:
return [(nx + nw // 2, ny + nh), (dx + dw // 2, dy)]
return [(nx + nw, ny + nh // 2), (dx, dy + dh // 2)]
below = ny + nh // 2 > dy + dh // 2 # the reader is below the data object: enter at its top
return [(dx + dw, dy + dh // 2), (nx + nw // 2, ny if below else ny + nh)]
# ── XML ───────────────────────────────────────────────────────────────────────────────────────────────────────
def _el(parent, tag: str, attrs: Optional[dict] = None, text: Optional[str] = None):
e = ET.SubElement(parent, tag, {k: str(v) for k, v in (attrs or {}).items() if v is not None})
if text is not None:
e.text = text
return e
def _doc(parent, text: str) -> None:
if text:
_el(parent, "bpmn:documentation", text=text)
def to_bpmn(pb: Playbook, setting: Setting, deviations: tuple = ()) -> str:
"""The playbook version as a BPMN 2.0 definitions document (UTF-8 XML text), the same for the same input.
`deviations` (domain/deviations.compare) mark the steps that differ from the pack's reference playbook, and
the process documentation lists the reference steps this one does not have."""
from brick_hr.domain import deviations as dv
m = _Model(pb, setting, dv.marks(list(deviations)))
m.build()
lay = _Layout(m)
pid = _ncname(pb.playbook_id)
process_id, collab_id = f"Process_{pid}_v{pb.version}", f"Collaboration_{pid}_v{pb.version}"
root = ET.Element("bpmn:definitions", {
**{f"xmlns:{k}": v for k, v in NS.items()},
"id": f"Definitions_{pid}_v{pb.version}", "targetNamespace": TARGET_NAMESPACE,
"exporter": "brick-hr", "exporterVersion": setting.exporter_version or "0"})
collab = _el(root, "bpmn:collaboration", {"id": collab_id})
_el(collab, "bpmn:participant", {"id": f"Participant_{pid}", "name": f"{pb.name} (v{pb.version})",
"processRef": process_id})
for fam in lay.families:
p = _el(collab, "bpmn:participant", {"id": f"Participant_{_ncname(fam)}", "name": FAMILIES.get(fam, fam)})
_doc(p, "The starter, outside the company, through the public link." if fam == "starter" else
"Drafting through DODIL Models (an OpenAI-compatible endpoint)." if fam == "drafting" else
f"Whatever provider the composition wires for {fam}.* (capabilities.yaml). With none, each call is a "
"manual task carrying the same input.")
for msg in m.messages:
part = f"Participant_{_ncname(msg.family)}"
src, tgt = (part, msg.node) if msg.inbound else (msg.node, part)
_el(collab, "bpmn:messageFlow", {"id": msg.id, "name": msg.name or None, "sourceRef": src, "targetRef": tgt})
pcf = standards.process_pcf(setting.direction)
proc = _el(root, "bpmn:process", {"id": process_id, "name": pb.name, "isExecutable": "false",
"dodil:pcf": pcf[0] if pcf else None})
scope = "; ".join(x for x in (
"event types " + ", ".join(pb.event_types),
("worker types " + ", ".join(pb.worker_types)) if pb.worker_types else "every worker type",
("reasons " + ", ".join(pb.reasons)) if pb.reasons else "",
f"org unit {pb.org_unit_id}" if pb.org_unit_id else "") if x)
_doc(proc, f"{pb.name}: the {setting.direction} playbook {pb.playbook_id}, version {pb.version} ({pb.status}), "
f"for {scope}. Times are in {setting.timezone}. Generated by brick-hr from the playbook rows, which "
"are the source: change them with keep_playbook or change_playbook (a new version), not this file." +
(f" APQC PCF {pcf[0]} {pcf[1]}." if pcf else "") +
"".join(f" Not in this playbook, from the reference: {d.before} ({d.cls}: {d.reason})."
for d in deviations if d.change == "removed") +
"".join(f" Setting ({d.cls}): {d.reason}." for d in deviations if d.change == "setting"))
lane_set = _el(proc, "bpmn:laneSet", {"id": f"LaneSet_{pid}"})
for ln in m.lanes:
label = setting.groups.get(ln)
lane = _el(lane_set, "bpmn:lane", {"id": f"Lane_{_ncname(ln)}", "name": f"{label} ({ln})" if label else ln})
for i, node in m.nodes.items():
if node.lane == ln:
_el(lane, "bpmn:flowNodeRef", text=i)
incoming, outgoing = {}, {}
for f in m.flows:
outgoing.setdefault(f.source, []).append(f.id)
incoming.setdefault(f.target, []).append(f.id)
assocs = {}
for a in m.assocs:
assocs.setdefault(a.node, []).append(a)
for i, node in m.nodes.items():
attrs = {"id": i, "name": node.name or None, **node.attrs}
if node.tag == "serviceTask":
attrs["implementation"] = "##unspecified"
if node.default:
attrs["default"] = node.default
e = _el(proc, f"bpmn:{node.tag}", attrs)
_doc(e, node.doc)
for fid in incoming.get(i, []):
_el(e, "bpmn:incoming", text=fid)
for fid in outgoing.get(i, []):
_el(e, "bpmn:outgoing", text=fid)
if node.timer:
ted = _el(e, "bpmn:timerEventDefinition", {"id": f"TimerEventDefinition_{i}"})
_el(ted, "bpmn:timeDate", {"xsi:type": "bpmn:tFormalExpression"}, node.timer)
if node.tag in ("userTask", "serviceTask"):
ins = [(f"{i}_in_{_ncname(p)}", p, src) for p, src in node.inputs]
outs = [(f"{i}_out_{_ncname(p)}", p) for p in node.outputs]
if ins or outs:
io = _el(e, "bpmn:ioSpecification", {"id": f"InputOutputSpecification_{i}"})
for did, p, src in ins:
di = _el(io, "bpmn:dataInput", {"id": did, "name": p})
_doc(di, f"from {src}" if src else "")
for did, p in outs:
_el(io, "bpmn:dataOutput", {"id": did, "name": p})
iset = _el(io, "bpmn:inputSet", {"id": f"InputSet_{i}"})
for did, _, _ in ins:
_el(iset, "bpmn:dataInputRefs", text=did)
oset = _el(io, "bpmn:outputSet", {"id": f"OutputSet_{i}"})
for did, _ in outs:
_el(oset, "bpmn:dataOutputRefs", text=did)
for a in assocs.get(i, []):
if a.kind == "in":
da = _el(e, "bpmn:dataInputAssociation", {"id": a.id})
_el(da, "bpmn:sourceRef", text=a.data)
_el(da, "bpmn:targetRef", text=f"{i}_in_{_ncname(a.ports[0])}")
for a in assocs.get(i, []):
if a.kind == "out":
da = _el(e, "bpmn:dataOutputAssociation", {"id": a.id})
for port in a.ports:
_el(da, "bpmn:sourceRef", text=f"{i}_out_{_ncname(port)}")
_el(da, "bpmn:targetRef", text=a.data)
if node.loop:
lc = _el(e, "bpmn:multiInstanceLoopCharacteristics", {"id": f"MultiInstance_{i}"})
_doc(lc, node.loop)
for f in m.flows:
e = _el(proc, "bpmn:sequenceFlow", {"id": f.id, "name": f.name or None, "sourceRef": f.source,
"targetRef": f.target})
if f.condition:
_el(e, "bpmn:conditionExpression", {"xsi:type": "bpmn:tFormalExpression"}, f.condition)
for d in m.data.values():
obj = f"DataObject_{d.id[len('DataObjectReference_'):]}"
o = _el(proc, "bpmn:dataObject", {"id": obj, "name": d.name})
_doc(o, d.doc)
_el(proc, "bpmn:dataObjectReference", {"id": d.id, "name": d.name, "dataObjectRef": obj})
# the diagram
dia = _el(root, "bpmndi:BPMNDiagram", {"id": f"BPMNDiagram_{pid}_v{pb.version}", "name": pb.name})
plane = _el(dia, "bpmndi:BPMNPlane", {"id": f"BPMNPlane_{pid}_v{pb.version}", "bpmnElement": collab_id})
def shape(element: str, b: tuple, **kw) -> None:
s = _el(plane, "bpmndi:BPMNShape", {"id": f"{element}_di", "bpmnElement": element, **kw})
_el(s, "dc:Bounds", {"x": b[0], "y": b[1], "width": b[2], "height": b[3]})
def edge(element: str, pts: list) -> None:
if not pts:
return
s = _el(plane, "bpmndi:BPMNEdge", {"id": f"{element}_di", "bpmnElement": element})
for x, y in pts:
_el(s, "di:waypoint", {"x": x, "y": y})
shape(f"Participant_{pid}", (X0, Y0, lay.pool_w, lay.pool_h), isHorizontal="true", isExpanded="true")
for ln in m.lanes:
shape(f"Lane_{_ncname(ln)}", (X0 + POOL_HEAD, lay.lane_y[ln], lay.pool_w - POOL_HEAD,
lay.lane_rows[ln] * ROW_H), isHorizontal="true")
for i in m.nodes:
shape(i, lay.bounds[i])
for d in m.data.values():
if d.id in lay.bounds:
shape(d.id, lay.bounds[d.id])
for fam in lay.families:
shape(f"Participant_{_ncname(fam)}", lay.pbounds[fam], isHorizontal="true", isExpanded="false")
for f in m.flows:
edge(f.id, lay.flow_points(f))
for msg in m.messages:
edge(msg.id, lay.message_points(msg))
for a in m.assocs:
edge(a.id, lay.assoc_points(a))
ET.indent(root, space=" ")
return '<?xml version="1.0" encoding="UTF-8"?>\n' + ET.tostring(root, encoding="unicode") + "\n"
__all__ = ["to_bpmn", "Setting", "Capability", "timing_text", "condition_text", "conditions_text", "NS",
"TARGET_NAMESPACE", "DODIL_NS"]