data-meta-model/governance/generate_processbook_md.py

#!/usr/bin/env python3
"""
Generate the Markdown processbook, with Mermaid views, from processes.yaml
(EV-008). The BPMN files remain the authoritative form; the views here are
derived from the same source and cannot drift from it.

USAGE
    python3 generate_processbook_md.py [output.md]
"""
import os
import sys

import yaml

HERE = os.path.dirname(os.path.abspath(__file__))
SRC = os.path.join(HERE, "processes.yaml")
OUT = sys.argv[1] if len(sys.argv) > 1 else os.path.join(HERE, "PR_TBox_Processbook.md")

SHAPE = {
    "start": ("([", "])"),
    "end": ("([", "])"),
    "userTask": ("[", "]"),
    "scriptTask": ("[", "]"),
    "callActivity": ("[[", "]]"),
    "gateway": ("{", "}"),
}
KIND = {
    "userTask": "human",
    "scriptTask": "script",
    "callActivity": "calls",
    "gateway": "decision",
    "start": "start",
    "end": "end",
}


def flow(t):
    return " ".join((t or "").split())


def mermaid(proc):
    out = ["```mermaid", "flowchart TD"]
    for n in proc["flow"]:
        o, c = SHAPE[n["type"]]
        label = n["name"].replace('"', "'")
        if n["type"] == "callActivity":
            label = "%s: %s" % (n["calls"], label)
        out.append('    %s%s"%s"%s' % (n["id"], o, label, c))
    for f in proc["flows"]:
        if f.get("condition"):
            out.append('    %s -->|%s| %s' % (f["from"], f["condition"], f["to"]))
        else:
            out.append("    %s --> %s" % (f["from"], f["to"]))
    for n in proc["flow"]:
        if n["type"] == "userTask":
            out.append("    class %s human;" % n["id"])
        elif n["type"] == "gateway":
            out.append("    class %s decision;" % n["id"])
    out.append("    classDef human fill:#FFF4CE,stroke:#B08900;")
    out.append("    classDef decision fill:#E8F0F7,stroke:#3E6E96;")
    out.append("```")
    return out


def main():
    doc = yaml.safe_load(open(SRC, encoding="utf-8"))
    m, out = doc["meta"], []
    w = out.append

    w("# %s" % m["title"])
    w("")
    w("**Version %s** — %s — generated %s" % (m["version"], m["status"], m["date"]))
    w("")
    w("> Generated from `processes.yaml`. The BPMN files in `bpmn/` are the "
      "authoritative form; the views below are derived from the same source. "
      "Do not edit this document (EV-007, EV-008).")
    w("")
    w(flow(m["scope"]))
    w("")
    w("## Procedures")
    w("")
    w("| Id | Family | Procedure | Scope |")
    w("|---|---|---|---|")
    for p in doc["processes"]:
        w("| **%s** | %s | %s | %s |" % (p["id"], p["family"], p["name"], p.get("scope", "")))
    w("")
    w("## Reading a diagram")
    w("")
    w("- A **rounded** node is a start or end state.")
    w("- A **diamond** is a decision; every outgoing path is labelled.")
    w("- A **shaded** box is a human task: it cannot be automated, and the work "
      "stops there until someone acts.")
    w("- A **double-bordered** box calls another procedure by its identifier.")
    w("")

    for p in doc["processes"]:
        w("---")
        w("")
        w("# %s — %s" % (p["id"], p["name"]))
        w("")
        w("| | |")
        w("|---|---|")
        w("| Family | %s |" % p["family"])
        w("| Scope | %s |" % p.get("scope", ""))
        w("| Trigger | %s |" % p.get("trigger", ""))
        w("| Inputs | %s |" % ", ".join(p.get("inputs") or []))
        w("| Outputs | %s |" % ", ".join(p.get("outputs") or []))
        rules = sorted({r for n in p["flow"] for r in (n.get("rules") or [])})
        w("| Rules enforced | %s |" % (", ".join(rules) or "-"))
        calls = sorted({n["calls"] for n in p["flow"] if n["type"] == "callActivity"})
        w("| Calls | %s |" % (", ".join(calls) or "-"))
        w("")
        if p.get("parameter"):
            w("**Parameter.** %s" % flow(p["parameter"]))
            w("")
        if p.get("note"):
            w("**Note.** %s" % flow(p["note"]))
            w("")
        out.extend(mermaid(p))
        w("")
        w("| Step | Type | Rules |")
        w("|---|---|---|")
        for n in p["flow"]:
            if n["type"] in ("start", "end"):
                continue
            name = n["name"]
            if n["type"] == "callActivity":
                name = "%s (%s)" % (name, n["calls"])
            w("| %s | %s | %s |"
              % (name, KIND[n["type"]], ", ".join(n.get("rules") or []) or "-"))
        w("")

    w("---")
    w("")
    w("# Rule coverage")
    w("")
    w("Which procedures enforce each rule. Derived, not maintained: a rule cited "
      "in a diagram appears here automatically.")
    w("")
    cover = {}
    for p in doc["processes"]:
        for n in p["flow"]:
            for r in n.get("rules") or []:
                cover.setdefault(r, set()).add(p["id"])
    w("| Rule | Procedures |")
    w("|---|---|")
    for r in sorted(cover):
        w("| %s | %s |" % (r, ", ".join(sorted(cover[r]))))
    w("")

    open(OUT, "w", encoding="utf-8").write("\n".join(out))
    print("written: %s (%d lines, %d procedures, %d rules covered)"
          % (OUT, len(out), len(doc["processes"]), len(cover)))


if __name__ == "__main__":
    main()