Master/data-meta-model
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

tbox: v1.6 ...

Browse Source
This commit is contained in:
Bastien Gourdon
2026-08-05 09:14:34 +02:00
parent 1c8ab77003
commit fbc63e55d8
9 changed files with 2669 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
governance/generate_bpmn.py
+206
View File
@@ -0,0 +1,206 @@
#!/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()
governance/generate_processbook_md.py
+157
View File
@@ -0,0 +1,157 @@
#!/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()
governance/generate_rulebook_md.py
+154
View File
@@ -0,0 +1,154 @@
#!/usr/bin/env python3
"""
Generate the Markdown rulebook from rules.yaml (EV-008).
USAGE
python3 generate_rulebook_md.py [output.md]
rules.yaml is the single source of truth. This script never edits it.
The output must never be edited by hand (EV-007).
"""
import os
import sys
import yaml
HERE = os.path.dirname(os.path.abspath(__file__))
SRC = os.path.join(HERE, "rules.yaml")
OUT = sys.argv[1] if len(sys.argv) > 1 else os.path.join(HERE, "PR_TBox_Rulebook.md")
def selfcheck(doc):
"""EV-011 and EV-012 applied to the rulebook itself."""
problems = []
for r in doc["rules"]:
ctl = r.get("control") or {}
if not ctl.get("tier"):
problems.append("%s: no control tier (EV-012)" % r["id"])
if r.get("severity") == "BLOCKING" and not ctl.get("executor"):
problems.append("%s: BLOCKING without an executor (EV-011)" % r["id"])
if r.get("category") not in doc["categories"]:
problems.append("%s: unknown category" % r["id"])
return problems
def flow(text):
"""Collapse a YAML folded scalar into a single paragraph."""
return " ".join((text or "").split())
def joined(v):
return ", ".join(v) if isinstance(v, list) else (v or "-")
def main():
doc = yaml.safe_load(open(SRC, encoding="utf-8"))
problems = selfcheck(doc)
if problems:
print("SELF-CHECK FAILED")
for p in problems:
print(" " + p)
sys.exit(1)
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 `rules.yaml`. Do not edit this document: edit the source "
"and regenerate (EV-007, EV-008).")
w("")
w("## Scope")
w("")
w(flow(m["scope"]))
w("")
w(flow(m["audience"]))
w("")
w("## Categories")
w("")
w("| Category | Subject | Rules |")
w("|---|---|---|")
for k, v in doc["categories"].items():
n = sum(1 for r in doc["rules"] if r["category"] == k)
w("| **%s** | %s | %d |" % (k, v["title"], n))
w("")
w("## Severity")
w("")
for k, v in doc["severity_model"].items():
w("- **%s** — %s" % (k, flow(v)))
w("")
for cat, info in doc["categories"].items():
w("---")
w("")
w("# %s — %s" % (cat, info["title"]))
w("")
w(flow(info["intent"]))
w("")
for r in [x for x in doc["rules"] if x["category"] == cat]:
ctl = r.get("control") or {}
w("## %s — %s" % (r["id"], r["title"]))
w("")
w("**Statement.** %s" % flow(r["statement"]))
w("")
w("| | |")
w("|---|---|")
w("| Severity | **%s** |" % r["severity"])
w("| Applies to | %s |" % joined(r.get("scope")))
w("| Enforced at | %s |" % joined(ctl.get("tier")))
w("| Executor | `%s` |" % (ctl.get("executor") or "-"))
w("| Procedure | %s |" % (ctl.get("procedure") or "_pending_"))
if r.get("filiation"):
w("| Related | %s |" % r["filiation"])
w("")
if r.get("rationale"):
w("**Why.** %s" % flow(r["rationale"]))
w("")
if r.get("examples"):
w("| From | To | Note |")
w("|---|---|---|")
for e in r["examples"]:
w("| %s | %s | %s |" % (e["from"], e["to"], e.get("note") or ""))
w("")
ab = doc["abstractness"]
w("---")
w("")
w("# Annex A — Abstractness")
w("")
w("Applies TN-006 and TN-007 to the model. %d classes, of which **%d abstract** "
"and **%d concrete**. Maintained with the model, not after it."
% (len(ab), sum(1 for a in ab if a["is_abstract"]),
sum(1 for a in ab if not a["is_abstract"])))
w("")
w("| Class | isAbstract | Note |")
w("|---|---|---|")
for a in ab:
w("| `%s` | %s | %s |"
% (a["term"], "**true**" if a["is_abstract"] else "false", a.get("note") or ""))
w("")
w("---")
w("")
w("# Annex B — Display")
w("")
w("Applies TN-011 and TN-022. The short label is what screens display and what "
"relation names reuse; the acronym is a search key, never an identity.")
w("")
w("| IRI | rdfs:label | shortLabel | acronym | Note |")
w("|---|---|---|---|---|")
for d in doc["display"]:
w("| `%s` | %s | %s | %s | %s |"
% (d["iri"], d["label"], d.get("short_label") or "-",
d.get("acronym") or "-", d.get("note") or ""))
w("")
open(OUT, "w", encoding="utf-8").write("\n".join(out))
print("self-check passed: %d rules, every tier assigned, no blocking rule "
"without an executor" % len(doc["rules"]))
print("written: %s (%d lines)" % (OUT, len(out)))
if __name__ == "__main__":
main()
governance/generate_rulebook_xlsx.py
+218
View File
@@ -0,0 +1,218 @@
#!/usr/bin/env python3
"""
Generate the Excel rulebook from rules.yaml (EV-008).
USAGE
python3 generate_rulebook_xlsx.py [output.xlsx]
rules.yaml is the single source of truth. This script never edits it.
The output must never be edited by hand (EV-007).
"""
import os
import sys
import yaml
from openpyxl import Workbook
from openpyxl.styles import Alignment, Border, Font, PatternFill, Side
from openpyxl.utils import get_column_letter
HERE = os.path.dirname(os.path.abspath(__file__))
SRC = os.path.join(HERE, "rules.yaml")
OUT = sys.argv[1] if len(sys.argv) > 1 else os.path.join(HERE, "PR_TBox_Rulebook.xlsx")
FONT = "Arial"
INK = "1F2933"
HEAD_FILL = PatternFill("solid", fgColor="1F2933")
BAND = {
"Identifier": "E8F0F7",
"Label": "EAF3EC",
"Declaration": "FBF0E4",
"Evolution": "F2EAF5",
}
MAJOR_FILL = PatternFill("solid", fgColor="FFF4CE")
THIN = Side(style="thin", color="C9CFD6")
BORDER = Border(left=THIN, right=THIN, top=THIN, bottom=THIN)
def selfcheck(doc):
"""EV-011 and EV-012 applied to the rulebook itself."""
problems = []
for r in doc["rules"]:
ctl = r.get("control") or {}
if not ctl.get("tier"):
problems.append("%s: no control tier (EV-012)" % r["id"])
if r.get("severity") == "BLOCKING" and not ctl.get("executor"):
problems.append("%s: BLOCKING without an executor (EV-011)" % r["id"])
if r.get("category") not in doc["categories"]:
problems.append("%s: unknown category" % r["id"])
return problems
def flow(t):
return " ".join((t or "").split())
def joined(v):
return ", ".join(v) if isinstance(v, list) else (v if v else "")
def header(ws, cols):
for i, (title, width) in enumerate(cols, start=1):
c = ws.cell(row=1, column=i, value=title)
c.font = Font(name=FONT, size=10, bold=True, color="FFFFFF")
c.fill = HEAD_FILL
c.alignment = Alignment(vertical="center", horizontal="left", wrap_text=True)
c.border = BORDER
ws.column_dimensions[get_column_letter(i)].width = width
ws.row_dimensions[1].height = 28
ws.freeze_panes = ws.cell(row=2, column=1)
ws.auto_filter.ref = "A1:%s1" % get_column_letter(len(cols))
def put(ws, r, c, value, wrap=False, bold=False, fill=None, size=10):
cell = ws.cell(row=r, column=c, value=value)
cell.font = Font(name=FONT, size=size, bold=bold, color=INK)
cell.alignment = Alignment(vertical="top", wrap_text=wrap)
cell.border = BORDER
if fill:
cell.fill = fill
return cell
def sheet_cover(wb, doc):
m = doc["meta"]
ws = wb.create_sheet("Cover")
ws.column_dimensions["A"].width = 24
ws.column_dimensions["B"].width = 100
put(ws, 1, 1, m["title"], bold=True, size=14)
rows = [
("Version", "%s — %s" % (m["version"], m["status"])),
("Generated", "%s from rules.yaml" % m["date"]),
("Scope", flow(m["scope"])),
("Audience", flow(m["audience"])),
("Generation", "Generated from rules.yaml by generate_rulebook_xlsx.py. "
"Do not edit this workbook: edit the source and regenerate "
"(EV-007, EV-008)."),
]
r = 3
for k, v in rows:
put(ws, r, 1, k, bold=True)
put(ws, r, 2, v, wrap=True)
ws.row_dimensions[r].height = 15 * (len(v) // 95 + 1)
r += 1
r += 1
put(ws, r, 1, "Categories", bold=True, size=12)
r += 1
for k, v in doc["categories"].items():
put(ws, r, 1, k, bold=True, fill=PatternFill("solid", fgColor=BAND[k]))
put(ws, r, 2, "%s — %s" % (v["title"], flow(v["intent"])), wrap=True)
ws.row_dimensions[r].height = 30
r += 1
r += 1
put(ws, r, 1, "Counts", bold=True, size=12)
r += 1
for label, formula in [
("Rules", "=COUNTA(Rules!A2:A200)"),
("of which blocking", '=COUNTIF(Rules!D2:D200,"BLOCKING")'),
("Examples", "=COUNTA(Examples!A2:A400)"),
("Classes", "=COUNTA(Abstractness!A2:A200)"),
("of which abstract", "=COUNTIF(Abstractness!B2:B200,TRUE)"),
("Display entries", "=COUNTA(Display!A2:A200)"),
]:
put(ws, r, 1, label)
put(ws, r, 2, formula)
r += 1
def sheet_rules(wb, doc):
ws = wb.create_sheet("Rules")
header(ws, [("Rule", 10), ("Category", 14), ("Title", 40), ("Severity", 11),
("Statement", 70), ("Applies to", 24), ("Enforced at", 14),
("Executor", 26), ("Procedure", 16), ("Related", 16),
("Why", 90)])
r = 2
for x in doc["rules"]:
ctl = x.get("control") or {}
band = PatternFill("solid", fgColor=BAND[x["category"]])
put(ws, r, 1, x["id"], bold=True, fill=band)
put(ws, r, 2, x["category"], fill=band)
put(ws, r, 3, x["title"], wrap=True, bold=True)
put(ws, r, 4, x["severity"],
fill=MAJOR_FILL if x["severity"] != "BLOCKING" else None)
put(ws, r, 5, flow(x["statement"]), wrap=True)
put(ws, r, 6, joined(x.get("scope")), wrap=True)
put(ws, r, 7, joined(ctl.get("tier")))
put(ws, r, 8, ctl.get("executor") or "")
put(ws, r, 9, ctl.get("procedure") or "pending")
put(ws, r, 10, x.get("filiation") or "")
put(ws, r, 11, flow(x.get("rationale")), wrap=True)
ws.row_dimensions[r].height = 78
r += 1
def sheet_examples(wb, doc):
ws = wb.create_sheet("Examples")
header(ws, [("Rule", 10), ("Category", 14), ("From", 48), ("To", 48), ("Note", 62)])
r = 2
for x in doc["rules"]:
for e in x.get("examples") or []:
band = PatternFill("solid", fgColor=BAND[x["category"]])
put(ws, r, 1, x["id"], bold=True, fill=band)
put(ws, r, 2, x["category"], fill=band)
put(ws, r, 3, str(e["from"]), wrap=True)
put(ws, r, 4, str(e["to"]), wrap=True)
put(ws, r, 5, e.get("note") or "", wrap=True)
r += 1
def sheet_abstractness(wb, doc):
ws = wb.create_sheet("Abstractness")
header(ws, [("Class", 36), ("isAbstract", 12), ("Note", 96)])
r = 2
for a in doc["abstractness"]:
put(ws, r, 1, a["term"], bold=a["is_abstract"])
put(ws, r, 2, a["is_abstract"])
put(ws, r, 3, a.get("note") or "", wrap=True)
r += 1
def sheet_display(wb, doc):
ws = wb.create_sheet("Display")
header(ws, [("IRI", 36), ("rdfs:label", 36), ("shortLabel", 22),
("acronym", 10), ("Note", 80)])
r = 2
for d in doc["display"]:
put(ws, r, 1, d["iri"], bold=True)
put(ws, r, 2, d["label"])
put(ws, r, 3, d.get("short_label") or "")
put(ws, r, 4, d.get("acronym") or "")
put(ws, r, 5, d.get("note") or "", wrap=True)
r += 1
def main():
doc = yaml.safe_load(open(SRC, encoding="utf-8"))
problems = selfcheck(doc)
if problems:
print("SELF-CHECK FAILED")
for p in problems:
print(" " + p)
sys.exit(1)
wb = Workbook()
wb.remove(wb.active)
sheet_cover(wb, doc)
sheet_rules(wb, doc)
sheet_examples(wb, doc)
sheet_abstractness(wb, doc)
sheet_display(wb, doc)
wb.save(OUT)
print("self-check passed: %d rules" % len(doc["rules"]))
print("written: %s (%d sheets)" % (OUT, len(wb.sheetnames)))
if __name__ == "__main__":
main()
governance/migration/migrate_tbox_v2_0.py
+357
View File
@@ -0,0 +1,357 @@
#!/usr/bin/env python3
"""
Migrate the Pernod Ricard Data MetaModel from v1.6 to v2.0.
USAGE
python3 migrate_tbox_v2_0.py # dry run, writes nothing
python3 migrate_tbox_v2_0.py --apply # writes in place
python3 migrate_tbox_v2_0.py --apply --log-dir logs/
--ontology path to the ontology TTL (default ../ontology/pr_metamodel.ttl)
--instances path to an instance TTL (repeatable)
--shapes path to a shapes TTL (repeatable)
DESIGN
EV-004 dry run is the default; nothing is written without --apply
EV-005 every edit goes through rdflib, never a regex on the text
EV-006 guards test the target state, so a replay reports zero change
EV-015 an execution log is written for every attempt
Run on GrosseBertha, inside the pinned venv:
. venv/bin/activate && python3 migrate_tbox_v2_0.py
"""
import argparse
import datetime
import hashlib
import json
import os
import re
import sys
from collections import OrderedDict
import yaml
from rdflib import Graph, Literal, Namespace, RDF, RDFS, OWL, URIRef, XSD
HERE = os.path.dirname(os.path.abspath(__file__))
SPEC = os.path.join(HERE, "renames.yaml")
RULES = os.path.normpath(os.path.join(HERE, "..", "rules.yaml"))
DCTERMS = Namespace("http://purl.org/dc/terms/")
# ----------------------------------------------------------------- utilities
class Report(object):
"""Counts every step, so that idempotence is provable and not merely hoped."""
def __init__(self):
self.steps = OrderedDict()
self.notes = []
def add(self, step, n, detail=None):
self.steps[step] = self.steps.get(step, 0) + n
if detail:
self.notes.append("%s: %s" % (step, detail))
@property
def total(self):
return sum(self.steps.values())
def md5(path):
h = hashlib.md5()
with open(path, "rb") as f:
for chunk in iter(lambda: f.read(65536), b""):
h.update(chunk)
return h.hexdigest()
def local(uri, ns):
s = str(uri)
return s[len(ns):] if s.startswith(ns) else None
def decamelise(name, is_class):
"""TN-018. Consecutive capitals are kept together: they carry an acronym."""
spaced = re.sub(r"(?<=[a-z0-9])(?=[A-Z])|(?<=[A-Z])(?=[A-Z][a-z])", " ", name)
return spaced if is_class else spaced[0].lower() + spaced[1:]
# ------------------------------------------------------------------ the work
def rename(graph, pr, old, new, report, step):
"""EV-001. Rewrite every triple naming the identifier, in all three positions."""
src, dst = pr[old], pr[new]
if (dst, None, None) in graph and (src, None, None) not in graph:
return 0 # EV-006: already done
n = 0
for s, p, o in list(graph):
ns, np_, no = s, p, o
if s == src:
ns = dst
if p == src:
np_ = dst
if o == src:
no = dst
if (ns, np_, no) != (s, p, o):
graph.remove((s, p, o))
graph.add((ns, np_, no))
n += 1
report.add(step, n)
return n
def merge(graph, pr, src_name, dst_name, report):
"""The source disappears into an existing target, which keeps its declaration."""
src, dst = pr[src_name], pr[dst_name]
n = 0
for s, p, o in list(graph.triples((None, src, None))):
graph.remove((s, p, o))
graph.add((s, dst, o))
n += 1
for s, p, o in list(graph.triples((src, None, None))):
graph.remove((s, p, o)) # drop the source declaration
n += 1
report.add("merge", n)
return n
def drop_subject(graph, subject, report, step):
"""EV-003. Remove the block AND every reference naming it, or inference rebuilds it."""
n = 0
for t in list(graph.triples((subject, None, None))):
graph.remove(t)
n += 1
for t in list(graph.triples((None, None, subject))):
graph.remove(t)
n += 1
for t in list(graph.triples((None, subject, None))):
graph.remove(t)
n += 1
report.add(step, n)
return n
def count_instances(graphs, term):
"""EV-002. The proof required before any permanent withdrawal."""
n = 0
for g in graphs:
n += len(list(g.triples((None, RDF.type, term))))
n += len(list(g.triples((None, term, None))))
n += len(list(g.triples((None, None, term))))
return n
def migrate(onto, others, spec, rules, report):
ns = spec["meta"]["namespace"]
pr = Namespace(ns)
all_graphs = [onto] + others
# 1 — deprecated terms are withdrawn outright (phase clause), proof first
if spec["structural"].get("drop_deprecated"):
for subj in list(onto.subjects(OWL.deprecated, Literal(True))):
name = local(subj, ns) or str(subj)
used = sum(count_instances([g], subj) for g in others)
if used:
report.add("deprecated_kept", 1, "%s still used %d times" % (name, used))
continue
for g in all_graphs:
drop_subject(g, subj, report, "deprecated_dropped")
# 2 — renames, in dependency order
for order in (1, 2, 3):
for r in [x for x in spec["renames"] if x["order"] == order]:
for g in all_graphs:
rename(g, pr, r["from"], r["to"], report, "rename_order_%d" % order)
# 3 — merges
for m in spec.get("merges") or []:
for g in all_graphs:
merge(g, pr, m["from"], m["into"], report)
# 4 — reclassify display properties (TN-003)
for rc in spec["structural"]["reclassify"]:
term = pr[rc["term"]]
if (term, RDF.type, OWL.AnnotationProperty) not in onto:
onto.remove((term, RDF.type, getattr(OWL, rc["from"])))
onto.add((term, RDF.type, OWL.AnnotationProperty))
report.add("reclassified", 1, rc["term"])
# 5 — create the governance layer root (TN-027)
for c in spec["structural"].get("create_classes") or []:
term = pr[c["term"]]
if (term, RDF.type, OWL.Class) not in onto:
onto.add((term, RDF.type, OWL.Class))
onto.add((term, RDFS.subClassOf, pr[c["parent"]]))
onto.add((term, RDFS.label, Literal(c["label"])))
onto.add((term, RDFS.comment, Literal(" ".join(c["comment"].split()))))
onto.add((term, pr.isAbstract, Literal(True)))
report.add("classes_created", 1, c["term"])
# 6 — reparent (TN-027)
for rp in spec["structural"].get("reparent") or []:
term, parent = pr[rp["term"]], pr[rp["parent"]]
if (term, RDFS.subClassOf, parent) not in onto:
onto.add((term, RDFS.subClassOf, parent))
report.add("reparented", 1, "%s -> %s" % (rp["term"], rp["parent"]))
# 7 — sub-properties (TN-028)
for sp in spec["structural"].get("subproperties") or []:
term, parent = pr[sp["term"]], pr[sp["parent"]]
if (term, RDFS.subPropertyOf, parent) not in onto:
onto.add((term, RDFS.subPropertyOf, parent))
report.add("subproperties", 1, "%s -> %s" % (sp["term"], sp["parent"]))
# 8 — controlled values become literals (TN-016, TN-017)
for conv in spec["structural"].get("to_literal") or []:
prop = pr[conv["property"]]
if (prop, RDF.type, OWL.DatatypeProperty) not in onto:
onto.remove((prop, RDF.type, OWL.ObjectProperty))
onto.add((prop, RDF.type, OWL.DatatypeProperty))
onto.remove((prop, RDFS.range, None))
onto.add((prop, RDFS.range, XSD.string))
onto.add((prop, RDFS.domain, pr[conv["domain"]]))
report.add("to_literal_property", 1, conv["property"])
for g in all_graphs: # rewrite the asserted values
for s, p, o in list(g.triples((None, prop, None))):
name = local(o, ns)
if name and name in conv["value_map"]:
g.remove((s, p, o))
g.add((s, p, Literal(conv["value_map"][name])))
report.add("to_literal_values", 1)
for ind in conv["drop_individuals"] + [conv["drop_class"]]:
subj = pr[ind]
if (subj, None, None) in onto:
for g in all_graphs:
drop_subject(g, subj, report, "to_literal_dropped")
# 9 — abstractness, from the rulebook annex (TN-006, TN-007)
for a in rules["abstractness"]:
term = pr[a["term"]]
if (term, RDF.type, OWL.Class) not in onto:
report.add("abstract_missing_class", 1, a["term"])
continue
want = Literal(bool(a["is_abstract"]))
if (term, pr.isAbstract, want) not in onto:
onto.remove((term, pr.isAbstract, None))
onto.add((term, pr.isAbstract, want))
report.add("isAbstract_declared", 1)
# 10 — display annotations, from the rulebook annex (TN-022)
for d in rules["display"]:
term = pr[d["iri"]]
for prop, value in ((pr.shortLabel, d.get("short_label")),
(pr.acronym, d.get("acronym"))):
if not value:
continue
if (term, prop, Literal(value)) not in onto:
onto.remove((term, prop, None))
onto.add((term, prop, Literal(value)))
report.add("display_set", 1)
# 11 — regenerate every label by derivation (TN-018, TN-023)
kinds = (OWL.Class, OWL.ObjectProperty, OWL.DatatypeProperty, OWL.AnnotationProperty)
for kind in kinds:
for term in set(onto.subjects(RDF.type, kind)):
name = local(term, ns)
if not name:
continue
want = Literal(decamelise(name, kind == OWL.Class))
if (term, RDFS.label, want) not in onto:
onto.remove((term, RDFS.label, None))
onto.add((term, RDFS.label, want))
report.add("labels_regenerated", 1)
# 12 — bump the ontology version (EV-014: the namespace itself never moves)
target = URIRef(ns.rstrip("/") + "/" + spec["meta"]["to_version"])
for onto_iri in set(onto.subjects(RDF.type, OWL.Ontology)):
if (onto_iri, OWL.versionIRI, target) not in onto:
onto.remove((onto_iri, OWL.versionIRI, None))
onto.add((onto_iri, OWL.versionIRI, target))
report.add("version_bumped", 1, str(target))
# ----------------------------------------------------------------------- main
def main():
ap = argparse.ArgumentParser()
ap.add_argument("--apply", action="store_true", help="write the files (default: dry run)")
ap.add_argument("--ontology", default=os.path.join(HERE, "..", "ontology", "pr_metamodel.ttl"))
ap.add_argument("--instances", action="append", default=[])
ap.add_argument("--shapes", action="append", default=[])
ap.add_argument("--log-dir", default=os.path.join(HERE, "logs"))
args = ap.parse_args()
spec = yaml.safe_load(open(SPEC, encoding="utf-8"))
rules = yaml.safe_load(open(RULES, encoding="utf-8"))
paths = [args.ontology] + args.instances + args.shapes
missing = [p for p in paths if not os.path.exists(p)]
if missing:
print("MISSING INPUT")
for p in missing:
print(" " + p)
sys.exit(1)
inputs = OrderedDict((p, md5(p)) for p in paths)
onto = Graph()
onto.parse(args.ontology, format="turtle")
others, other_paths = [], args.instances + args.shapes
for p in other_paths:
g = Graph()
g.parse(p, format="turtle")
others.append(g)
before = [len(onto)] + [len(g) for g in others]
report = Report()
migrate(onto, others, spec, rules, report)
after = [len(onto)] + [len(g) for g in others]
print("MIGRATION %s -> %s %s"
% (spec["meta"]["from_version"], spec["meta"]["to_version"],
"APPLY" if args.apply else "DRY RUN"))
print()
for step, n in report.steps.items():
print(" %-28s %6d" % (step, n))
print(" %-28s %6d" % ("total changes", report.total))
print()
for p, b, a in zip(paths, before, after):
print(" %-46s %6d -> %6d triples" % (os.path.basename(p), b, a))
if report.notes:
print()
for n in report.notes:
print(" note: " + n)
log = {
"attempt_timestamp": datetime.datetime.now().isoformat(timespec="seconds"),
"mode": "apply" if args.apply else "dry-run",
"from_version": spec["meta"]["from_version"],
"to_version": spec["meta"]["to_version"],
"input_checksums": inputs,
"steps": report.steps,
"total_changes": report.total,
"triples_before": dict(zip(paths, before)),
"triples_after": dict(zip(paths, after)),
"notes": report.notes,
}
os.makedirs(args.log_dir, exist_ok=True)
stamp = datetime.datetime.now().strftime("%Y%m%dT%H%M%S")
log_path = os.path.join(args.log_dir, "migration_%s.json" % stamp)
json.dump(log, open(log_path, "w", encoding="utf-8"), indent=2)
print("\n log: %s" % log_path)
if not args.apply:
print("\n DRY RUN — nothing written. Re-run with --apply when the counts "
"above are what you expect.")
return
onto.serialize(destination=args.ontology, format="turtle")
for p, g in zip(other_paths, others):
g.serialize(destination=p, format="turtle")
print("\n written. Replay this script now: a second run must report zero "
"changes (EV-006).")
if __name__ == "__main__":
main()
governance/migration/renames.yaml
+154
View File
@@ -0,0 +1,154 @@
# T-BOX MIGRATION TO v2.0 — RENAME MAP AND STRUCTURAL OPERATIONS
# =============================================================================
# Data consumed by migrate_tbox_v2_0.py. Kept separate from the script so that
# the transformation is reviewable without reading code (EV-004).
#
# Not listed here and derived from other sources at run time:
# - isAbstract declarations -> rules.yaml, annex `abstractness`
# - shortLabel and acronym -> rules.yaml, annex `display`
# - deprecated terms to drop -> found in the graph by owl:deprecated true
# Deriving them keeps a single source of truth for each fact (EV-008).
# =============================================================================
meta:
from_version: "1.6"
to_version: "2.0"
namespace: "https://ontology.pernod-ricard.com/metamodel/"
# -----------------------------------------------------------------------------
# RENAMES — 48. Applied everywhere the identifier appears: subject, predicate,
# object. Nothing cascades in RDF (EV-001).
#
# `order` is a migration constraint, not a preference:
# 1 ordinary terms
# 2 terms named in SHACL shapes or SPARQL constraints
# 3 the four attributes borne by MetaModelObject, therefore by every subject
# -----------------------------------------------------------------------------
renames:
# --- classes
- {from: SubDomain, to: DataSubDomain, kind: class, order: 1}
- {from: SubDomainOwner, to: DataSubDomainOwner, kind: class, order: 1}
- {from: KPI, to: KeyPerformanceIndicator, kind: class, order: 1}
- {from: BIWorkspace, to: BusinessIntelligenceWorkspace, kind: class, order: 1}
- {from: BIDataSource, to: BusinessIntelligenceDataSource, kind: class, order: 1}
- {from: BIField, to: BusinessIntelligenceField, kind: class, order: 1}
- {from: BIReport, to: BusinessIntelligenceReport, kind: class, order: 1}
# --- individual
- {from: BITool, to: BusinessIntelligenceTool, kind: individual, order: 1}
# --- relations
- {from: hasDGL, to: hasGovernanceLead, kind: relation, order: 1}
- {from: aboutConcept, to: isAbout, kind: relation, order: 1}
- {from: primaryLocation, to: primarilyStoredIn, kind: relation, order: 1}
- {from: inDatabase, to: isInDatabase, kind: relation, order: 1}
- {from: inSchema, to: isInSchema, kind: relation, order: 1}
- {from: inWorkspace, to: isInBIWorkspace, kind: relation, order: 1}
- {from: inBIDataSource, to: isInBIDataSource, kind: relation, order: 1}
- {from: hasBIField, to: containsBIField, kind: relation, order: 1}
- {from: hasBIOwner, to: hasPublisher, kind: relation, order: 1}
- {from: fromField, to: referencesSourceField, kind: relation, order: 1}
- {from: toField, to: referencesTargetField, kind: relation, order: 1}
- {from: biDerivedFrom, to: derivedFromSourceField, kind: relation, order: 1}
- {from: owningDomain, to: ownedByDomain, kind: relation, order: 2}
# --- attributes
- {from: hasBusinessDefinition, to: businessDefinition, kind: attribute, order: 1}
- {from: hasTechnicalDefinition, to: technicalDefinition, kind: attribute, order: 1}
- {from: hasSynonym, to: synonym, kind: attribute, order: 1}
- {from: hasFormula, to: formula, kind: attribute, order: 1}
- {from: hasUnit, to: unit, kind: attribute, order: 1}
- {from: hasTimeAggregation, to: timeAggregation, kind: attribute, order: 1}
- {from: hasFormat, to: logicalFormat, kind: attribute, order: 1}
- {from: hasPhysicalDataType, to: physicalDataType, kind: attribute, order: 1}
- {from: hasViewDefinition, to: viewDefinition, kind: attribute, order: 1}
- {from: hasExpression, to: expression, kind: attribute, order: 1}
- {from: hasBusinessRule, to: businessRule, kind: attribute, order: 1}
- {from: hasExampleValue, to: exampleValue, kind: attribute, order: 1}
- {from: createdOn, to: creationDate, kind: attribute, order: 1}
- {from: lastReviewedOn, to: lastReviewDate, kind: attribute, order: 1}
- {from: harvestedOn, to: harvestDate, kind: attribute, order: 1}
- {from: lastQueriedOn, to: lastQueryDate, kind: attribute, order: 1}
- {from: definedBy, to: definitionAddress, kind: attribute, order: 1}
- {from: queryCount30d, to: queryCount, kind: attribute, order: 1}
- {from: hasShortLabel, to: shortLabel, kind: annotation, order: 2}
- {from: hasAcronym, to: acronym, kind: annotation, order: 2}
- {from: hasIdentifier, to: identifier, kind: attribute, order: 3}
- {from: hasName, to: canonicalName, kind: attribute, order: 3}
- {from: hasStatus, to: status, kind: attribute, order: 3}
- {from: hasVersion, to: version, kind: attribute, order: 3}
# --- relations that change nature (TN-016): object property -> datatype property
- {from: hasActivationStatus, to: activationStatus, kind: relation_to_attribute, order: 1}
- {from: deployedIn, to: environment, kind: relation_to_attribute, order: 1}
# -----------------------------------------------------------------------------
# MERGES — the source identifier disappears into an existing target.
# Unlike a rename, the target already exists and keeps its own declaration.
# -----------------------------------------------------------------------------
merges:
- from: biConsumes
into: consumes
widen_domain: true
reason: >
Same verb, same target, identical meaning. The domain widens and scope is
controlled per class by SHACL. Ranges are compatible, so no RDFS retyping
is introduced (TN-028).
# -----------------------------------------------------------------------------
# STRUCTURAL OPERATIONS
# -----------------------------------------------------------------------------
structural:
reclassify:
- {term: shortLabel, from: DatatypeProperty, to: AnnotationProperty, rule: TN-003}
- {term: acronym, from: DatatypeProperty, to: AnnotationProperty, rule: TN-003}
create_classes:
- term: GovernanceLayerObject
parent: MetaModelObject
is_abstract: true
label: Governance Layer Object
comment: >
Root of the governance layer. Holds the actors and governance objects that
are not data objects and therefore sit outside the data layers. Created by
insertion above Actor rather than by renaming it, so that every range
pointing at Actor keeps stating the nature of its target rather than its
position in the model.
rule: TN-027
reparent:
- {term: Actor, parent: GovernanceLayerObject, rule: TN-027}
- {term: PhysicalLayerObject, parent: MetaModelObject, rule: TN-027}
subproperties:
- {term: primarilyStoredIn, parent: storedIn, rule: TN-028}
# TN-016: closed governance states become literals. The classes and their
# individuals are withdrawn, and the properties become datatype properties.
to_literal:
- property: activationStatus
domain: DataDomain
drop_class: ActivationStatus
drop_individuals: [NotActivated, LightActivation, FullActivation]
value_map:
NotActivated: NOT_ACTIVATED
LightActivation: LIGHT_ACTIVATION
FullActivation: FULL_ACTIVATION
- property: environment
domain: CapturedObject
drop_class: Environment
drop_individuals: [Development, UserAcceptance, Production]
value_map:
Development: DEVELOPMENT
UserAcceptance: USER_ACCEPTANCE
Production: PRODUCTION
# Phase clause: before v2.0 is published the project is in design, so
# deprecated terms are removed outright rather than kept as stubs. EV-002
# still applies — the count must be zero.
drop_deprecated: true
governance/processes.yaml
+375
View File
@@ -0,0 +1,375 @@
# PERNOD RICARD DATA METAMODEL — T-BOX PROCESSBOOK
# =============================================================================
# SINGLE SOURCE OF TRUTH for the procedures. The BPMN 2.0 files, the Mermaid
# views and the Markdown document are GENERATED from this file (EV-008).
#
# python3 generate_bpmn.py -> bpmn/<id>.bpmn (strict, queryable)
# python3 generate_processbook_md.py -> PR_TBox_Processbook.md (Mermaid views)
#
# NODE TYPES
# start | end events
# userTask a human decides or writes; cannot be automated
# scriptTask fully automated
# callActivity invokes another procedure by its id
# gateway exclusive decision; every outgoing flow is guarded
#
# Every node may carry `rules`, the identifiers of the rulebook rules it
# enforces. That list is what makes the BPMN queryable and what feeds the
# control.procedure field back into the rulebook.
# =============================================================================
meta:
title: Pernod Ricard Data MetaModel — T-Box Processbook
version: "0.1"
status: Draft for review
date: "2026-08-03"
scope: >
Procedures for changing the vocabulary of the model. Each procedure names
the rules it enforces, the steps a person must perform, and the points at
which the work stops rather than continues.
namespace: "https://ontology.pernod-ricard.com/process/"
families:
Create: Bringing a new element into the vocabulary.
Verify: Establishing that what exists conforms.
Update: Changing an element that already exists.
Delete: Removing an element from the vocabulary.
Release: Propagating a change and publishing a version.
processes:
# ---------------------------------------------------------------------- X1
- id: X1
family: Release
name: Propagate a change and publish a version
scope: Any validated modification of the vocabulary.
trigger: A change to the vocabulary has been agreed and is ready to apply.
inputs: [the agreed change, the target version, the list of affected artifacts]
outputs: [a merged branch, a tagged version, an execution log]
note: >
The most frequently invoked procedure of the processbook: every other
procedure ends by calling it. Two of its steps are gateways rather than
checks, because an idempotence failure and a validation failure must stop
the work rather than be noted in passing.
flow:
- {id: start, type: start, name: Change agreed}
- {id: branch, type: scriptTask, name: Create the migration branch, rules: [EV-009]}
- {id: apply, type: scriptTask, name: Apply the change through an RDF parser, rules: [EV-005, EV-006]}
- {id: labels, type: scriptTask, name: Regenerate labels by derivation, rules: [TN-023]}
- {id: abox, type: scriptTask, name: Propagate to the instances, rules: [EV-009]}
- {id: shapes, type: scriptTask, name: Update the shapes and align conformsTo, rules: [EV-010]}
- {id: artifacts, type: scriptTask, name: Regenerate every derived artifact, rules: [EV-008]}
- {id: replay, type: scriptTask, name: Replay the chain a second time, rules: [EV-006, EV-007]}
- {id: idem, type: gateway, name: "Second run reports zero change?"}
- {id: verify_voc, type: callActivity, calls: R1, name: Verify the vocabulary}
- {id: verify_abox, type: callActivity, calls: R2, name: Validate the instances}
- {id: log, type: scriptTask, name: Write the execution log for this attempt, rules: [EV-015]}
- {id: clean, type: gateway, name: "Any violation?"}
- {id: decide, type: userTask, name: Correct or abandon, rules: [EV-004, EV-007]}
- {id: drop, type: scriptTask, name: Destroy the branch and restart from the published version, rules: [EV-007]}
- {id: review, type: userTask, name: Review the merge, rules: [EV-004, EV-009]}
- {id: merge, type: scriptTask, name: Merge as a block, tag, push the tag separately, rules: [EV-009]}
- {id: bump, type: scriptTask, name: Increment the ontology version, rules: [EV-014]}
- {id: end, type: end, name: Version published}
- {id: aborted, type: end, name: Change abandoned}
flows:
- {from: start, to: branch}
- {from: branch, to: apply}
- {from: apply, to: labels}
- {from: labels, to: abox}
- {from: abox, to: shapes}
- {from: shapes, to: artifacts}
- {from: artifacts, to: replay}
- {from: replay, to: idem}
- {from: idem, to: verify_voc, condition: "zero change"}
- {from: idem, to: decide, condition: "the chain is not replayable"}
- {from: verify_voc, to: verify_abox}
- {from: verify_abox, to: log}
- {from: log, to: clean}
- {from: clean, to: review, condition: "none"}
- {from: clean, to: decide, condition: "at least one"}
- {from: decide, to: apply, condition: correct}
- {from: decide, to: drop, condition: abandon}
- {from: drop, to: aborted}
- {from: review, to: merge}
- {from: merge, to: bump}
- {from: bump, to: end}
# ---------------------------------------------------------------------- C1
- id: C1
family: Create
name: Create a term
scope: class | relation | attribute | annotation
parameter: >
The nature of the term selects the naming rules applied at the naming step:
a class takes TN-001, TN-002, TN-004, TN-005, TN-006, TN-007 and TN-027; a
relation takes TN-001, TN-002, TN-009, TN-010, TN-011 and TN-028; an
attribute takes TN-001, TN-002, TN-004, TN-012, TN-013, TN-014 and TN-015;
an annotation takes TN-001, TN-002 and TN-003.
trigger: A missing concept, edge or field has been identified.
inputs: [the intended meaning, the nature, the target layer, the provenance, the parent or the domain and range]
outputs: [a declared term, a published version]
note: >
The first step has no tool today. Checking that no existing term already
covers the need is the semantic uniqueness control that SHACL cannot
perform, and it stays a human task until R3 exists.
flow:
- {id: start, type: start, name: Need identified}
- {id: unique, type: userTask, name: Check that no existing term covers the need}
- {id: exists, type: gateway, name: "A term already covers it?"}
- {id: reuse, type: end, name: Reuse the existing term}
- {id: name, type: scriptTask, name: Name the term according to its nature, rules: [TN-001, TN-002, TN-003, TN-004, TN-005, TN-006, TN-009, TN-010, TN-011, TN-012, TN-013, TN-014, TN-015, TN-028]}
- {id: label, type: scriptTask, name: Derive the label, rules: [TN-018, TN-019, TN-020, TN-021, TN-023]}
- {id: short, type: userTask, name: Decide the short label and the acronym, rules: [TN-022]}
- {id: comment, type: userTask, name: Write the comment, rules: [TN-024]}
- {id: declare, type: scriptTask, name: Declare typing, provenance and attachment, rules: [TN-007, TN-025, TN-026, TN-027]}
- {id: validate, type: userTask, name: Validate before applying, rules: [EV-004]}
- {id: check, type: callActivity, calls: R1, name: Verify against the rulebook}
- {id: release, type: callActivity, calls: X1, name: Propagate and publish}
- {id: end, type: end, name: Term available}
flows:
- {from: start, to: unique}
- {from: unique, to: exists}
- {from: exists, to: reuse, condition: "yes"}
- {from: exists, to: name, condition: "no"}
- {from: name, to: label}
- {from: label, to: short}
- {from: short, to: comment}
- {from: comment, to: declare}
- {from: declare, to: validate}
- {from: validate, to: check}
- {from: check, to: release}
- {from: release, to: end}
# ---------------------------------------------------------------------- C2
- id: C2
family: Create
name: Create a controlled value
scope: typed individual | literal
parameter: >
The form is not a style choice but the outcome of the first decision. The
two branches have different consequences: the literal branch edits the
shapes and therefore the vocabulary, so it falls under the version freeze;
the individual branch touches nothing else.
trigger: A new value is needed in a controlled set.
inputs: [the value, the set it belongs to, whether a domain may add others]
outputs: [a declared value, a published version]
flow:
- {id: start, type: start, name: New value needed}
- {id: decide, type: userTask, name: "Must it be defined, owned or extended by a domain?", rules: [TN-016]}
- {id: form, type: gateway, name: "Which form?"}
- {id: indiv, type: scriptTask, name: Create the typed individual and derive its label, rules: [TN-001, TN-018, TN-022]}
- {id: literal, type: scriptTask, name: Write the literal in upper snake case, rules: [TN-017]}
- {id: shape, type: callActivity, calls: C3, name: Extend the closed list in the shapes}
- {id: validate, type: userTask, name: Validate before applying, rules: [EV-004]}
- {id: check, type: callActivity, calls: R1, name: Verify against the rulebook}
- {id: release, type: callActivity, calls: X1, name: Propagate and publish}
- {id: end, type: end, name: Value available}
flows:
- {from: start, to: decide}
- {from: decide, to: form}
- {from: form, to: indiv, condition: "extensible by a domain"}
- {from: form, to: literal, condition: "closed governance state"}
- {from: indiv, to: validate}
- {from: literal, to: shape}
- {from: shape, to: validate}
- {from: validate, to: check}
- {from: check, to: release}
- {from: release, to: end}
# ---------------------------------------------------------------------- C3
- id: C3
family: Create
name: Create a shape
scope: Any SHACL constraint added to the validation set.
trigger: A rule needs an executor, or a new class enters the validation perimeter.
inputs: [the rule to enforce, the target class or property]
outputs: [a shape, a rulebook entry naming it as executor]
note: >
The step that is forgotten is the last one before release: recording the
shape as the executor of its rule. Without it a rule stays declared
blocking with nothing enforcing it, which EV-011 forbids.
flow:
- {id: start, type: start, name: Constraint needed}
- {id: write, type: userTask, name: Write the constraint}
- {id: pattern, type: scriptTask, name: Check that every pattern is expressed positively, rules: [EV-013]}
- {id: portable, type: gateway, name: "Expressible within the specification?"}
- {id: demote, type: userTask, name: Move the rule to the script tier, rules: [EV-012]}
- {id: conforms, type: scriptTask, name: Align the declared ontology version, rules: [EV-010]}
- {id: test, type: callActivity, calls: R2, name: Test against real instances}
- {id: record, type: userTask, name: Record the shape as the executor of its rule, rules: [EV-011, EV-012]}
- {id: release, type: callActivity, calls: X1, name: Propagate and publish}
- {id: end, type: end, name: Shape in force}
flows:
- {from: start, to: write}
- {from: write, to: pattern}
- {from: pattern, to: portable}
- {from: portable, to: conforms, condition: "yes"}
- {from: portable, to: demote, condition: "no"}
- {from: demote, to: record}
- {from: conforms, to: test}
- {from: test, to: record}
- {from: record, to: release}
- {from: release, to: end}
# ---------------------------------------------------------------------- D1
- id: D1
family: Delete
name: Deprecate a term
scope: class | relation | attribute | annotation | individual
trigger: A term is superseded or no longer needed.
inputs: [the term, its replacement if any]
outputs: [a deprecated stub, a published version]
note: >
The count at step two is informative, not a condition: a term may be
deprecated whether or not it is instantiated. It becomes a condition only
in D2.
flow:
- {id: start, type: start, name: Term superseded}
- {id: replacement, type: userTask, name: Identify the replacement, or record that there is none}
- {id: count, type: scriptTask, name: Count the instances for information, rules: [EV-015]}
- {id: mark, type: scriptTask, name: Mark deprecated and declare the replacement, rules: [EV-001]}
- {id: strip, type: scriptTask, name: Strip the stub of every edge, rules: [EV-001]}
- {id: label, type: scriptTask, name: Remove any mention of state from the label, rules: [TN-020]}
- {id: release, type: callActivity, calls: X1, name: Propagate and publish}
- {id: end, type: end, name: Term deprecated}
flows:
- {from: start, to: replacement}
- {from: replacement, to: count}
- {from: count, to: mark}
- {from: mark, to: strip}
- {from: strip, to: label}
- {from: label, to: release}
- {from: release, to: end}
# ---------------------------------------------------------------------- D2
- id: D2
family: Delete
name: Withdraw a term permanently
scope: class | relation | attribute | annotation | individual
trigger: A deprecated term is to be removed from the vocabulary.
inputs: [the deprecated term]
outputs: [a vocabulary without the term, a proof of non-instantiation, a published version]
note: >
The reference cleaning step is what prevents phantom nodes: a subject
removed while its identifier is still cited elsewhere is reconstructed by
inference, present in traversals and absent from every control. It is a step
of the procedure, not a check at the end of a script.
flow:
- {id: start, type: start, name: Withdrawal requested}
- {id: count, type: scriptTask, name: Run the counting query, rules: [EV-002, EV-015]}
- {id: instantiated, type: gateway, name: "Any instance found?"}
- {id: keep, type: end, name: Kept deprecated}
- {id: proof, type: scriptTask, name: Attach the proof to the commit, rules: [EV-002, EV-015]}
- {id: remove, type: scriptTask, name: Remove the subject block}
- {id: refs, type: scriptTask, name: Remove every reference naming the subject, rules: [EV-003]}
- {id: check_voc, type: callActivity, calls: R1, name: Verify the vocabulary}
- {id: check_abox, type: callActivity, calls: R2, name: Validate the instances}
- {id: release, type: callActivity, calls: X1, name: Propagate and publish}
- {id: end, type: end, name: Term withdrawn}
flows:
- {from: start, to: count}
- {from: count, to: instantiated}
- {from: instantiated, to: keep, condition: "yes"}
- {from: instantiated, to: proof, condition: "no"}
- {from: proof, to: remove}
- {from: remove, to: refs}
- {from: refs, to: check_voc}
- {from: check_voc, to: check_abox}
- {from: check_abox, to: release}
- {from: release, to: end}
# ---------------------------------------------------------------------- R1
- id: R1
family: Verify
name: Verify the vocabulary against the rulebook
scope: The whole vocabulary, or the subset touched by a change.
trigger: A term has been created, changed or withdrawn, or a release is prepared.
inputs: [the vocabulary, the rulebook source]
outputs: [a conformance report]
note: >
Two tiers run in sequence, not in parallel: the script settles everything
mechanisable, and a person answers only for the rules no script can judge.
Reversing the order wastes review time on findings the script would have
caught.
flow:
- {id: start, type: start, name: Verification requested}
- {id: script, type: scriptTask, name: "Run the naming and declaration checks", rules: [TN-001, TN-002, TN-003, TN-004, TN-005, TN-006, TN-007, TN-009, TN-010, TN-011, TN-012, TN-013, TN-014, TN-015, TN-017, TN-018, TN-019, TN-020, TN-021, TN-023, TN-025, TN-026, TN-027, TN-028, EV-014]}
- {id: mechanised, type: gateway, name: "Any mechanised violation?"}
- {id: report_fail, type: end, name: Report returned with violations}
- {id: human, type: userTask, name: "Review the rules no script can judge", rules: [TN-008, TN-016, TN-022, TN-024]}
- {id: judged, type: gateway, name: "Reviewer raises an issue?"}
- {id: report_ok, type: end, name: Vocabulary conforms}
flows:
- {from: start, to: script}
- {from: script, to: mechanised}
- {from: mechanised, to: report_fail, condition: "at least one"}
- {from: mechanised, to: human, condition: none}
- {from: human, to: judged}
- {from: judged, to: report_fail, condition: "yes"}
- {from: judged, to: report_ok, condition: "no"}
# ---------------------------------------------------------------------- R2
- id: R2
family: Verify
name: Validate instances against the shapes
scope: Any instance graph in the validation perimeter.
trigger: Instances have changed, shapes have changed, or a release is prepared.
inputs: [the instance graph, the shapes, the ontology]
outputs: [a validation report]
note: >
The version check comes first and aborts rather than warns. Validating
against shapes that target another version of the ontology does not fail
loudly: it returns a long list of violations that reads exactly like a
regression of the model.
flow:
- {id: start, type: start, name: Validation requested}
- {id: conforms, type: scriptTask, name: "Compare the declared target version with the ontology version", rules: [EV-010]}
- {id: match, type: gateway, name: "Versions match?"}
- {id: abort, type: end, name: Aborted on version mismatch}
- {id: run, type: scriptTask, name: Run the shape validation}
- {id: violations, type: gateway, name: "Any violation?"}
- {id: fail, type: end, name: Report returned with violations}
- {id: ok, type: end, name: Instances conform}
flows:
- {from: start, to: conforms}
- {from: conforms, to: match}
- {from: match, to: abort, condition: "no"}
- {from: match, to: run, condition: "yes"}
- {from: run, to: violations}
- {from: violations, to: fail, condition: "at least one"}
- {from: violations, to: ok, condition: none}
# ---------------------------------------------------------------------- R3
- id: R3
family: Verify
name: Audit what the shapes cannot see
scope: The whole vocabulary and its source file.
trigger: Periodic audit, or before a version is published.
inputs: [the vocabulary, its source file]
outputs: [a shortlist of suspected duplicates, a list of duplicated blocks]
note: >
Shape validation reads a graph, not a file and not meaning. Two identifiers
standing for the same notion produce two individually valid graphs; a
duplicated block of text produces identical triples and no complaint. This
procedure is the tier those rules fall to, and its output is a shortlist for
a person rather than a verdict.
flow:
- {id: start, type: start, name: Audit requested}
- {id: index, type: scriptTask, name: "Build the normalised label index", rules: [EV-012]}
- {id: hash, type: scriptTask, name: "Hash every normalised subject block", rules: [EV-012]}
- {id: shortlist, type: gateway, name: "Any candidate found?"}
- {id: clean, type: end, name: Nothing to arbitrate}
- {id: review, type: userTask, name: Arbitrate each candidate}
- {id: act, type: gateway, name: "Duplication confirmed?"}
- {id: merge, type: end, name: Referred to the merge procedure}
- {id: dismissed, type: end, name: Candidates dismissed}
flows:
- {from: start, to: index}
- {from: index, to: hash}
- {from: hash, to: shortlist}
- {from: shortlist, to: clean, condition: none}
- {from: shortlist, to: review, condition: "at least one"}
- {from: review, to: act}
- {from: act, to: merge, condition: "yes"}
- {from: act, to: dismissed, condition: "no"}
governance/requirements.txt
+3
View File
@@ -0,0 +1,3 @@
PyYAML>=6.0,<7
openpyxl>=3.1,<3.2
governance/rules.yaml
+1045
View File
File diff suppressed because it is too large Load Diff