#!/usr/bin/env python3
"""
PR Data Meta Model - T-Box viewer generator
============================================
Reads the T-Box Turtle file and emits a self-contained HTML view.
Build artefact: regenerate, never edit.
USAGE
python3 scripts/generate_tbox_viewer.py [in.ttl] [out.html]
defaults: ontology/pr_metamodel.ttl -> generated/pr_metamodel_viewer.html
Same visual language as the SODH viewer, with one substitution. Where SODH
lays its columns out by owning domain, the T-Box has no domains -- its
orthogonal axis is PROVENANCE: what governance asserts against what a
harvester observes. The two-colour code carries that instead, blue for
defined and dark gold for captured, and the seam is what crosses between them.
No dependency: parses the Turtle subset this file uses, so it runs on the DSM
system python without a venv.
"""
import json
import os
import re
import sys
from collections import OrderedDict
PR = "https://ontology.pernod-ricard.com/metamodel/"
WANTED = {
"a": "type", "rdf:type": "type",
"rdfs:label": "label", "rdfs:comment": "comment",
"rdfs:subClassOf": "subClassOf", "rdfs:subPropertyOf": "subPropertyOf",
"rdfs:domain": "domain", "rdfs:range": "range",
"owl:inverseOf": "inverseOf", "owl:deprecated": "deprecated",
"dcterms:isReplacedBy": "isReplacedBy",
"pr:harvestSource": "harvestSource", "pr:authoringMode": "authoringMode",
"pr:hasAcronym": "acronym", "pr:hasShortLabel": "shortLabel",
}
LAYER_ROOT = OrderedDict([
("pr:OwnershipLayerObject", ("ownership", "Ownership & Categorization",
"who is accountable")),
("pr:BusinessLayerObject", ("business", "Business", "what it means")),
("pr:LogicalLayerObject", ("logical", "Logical",
"how it is structured")),
("pr:DeliveryLayerObject", ("delivery", "Delivery",
"what we promise, to whom, until when")),
("pr:PhysicalLayerObject", ("physical", "Physical",
"what actually runs")),
("pr:ConsumptionLayerObject", ("consumption", "Consumption",
"what people open in the morning")),
])
# The only relations crossing between what is asserted and what is harvested.
SEAM = {"pr:materializedAs", "pr:storedIn", "pr:physicalizedIn"}
# ---------------------------------------------------------------- parsing
def strip_comments(text):
out, i, n = [], 0, len(text)
while i < n:
c = text[i]
if c == '"':
j = i + 1
while j < n and text[j] != '"':
j += 2 if text[j] == "\\" else 1
j = min(j + 1, n)
out.append(text[i:j]); i = j; continue
if c == "#":
j = text.find("\n", i)
i = n if j == -1 else j
continue
out.append(c); i += 1
return "".join(out)
def split_outside(text, sep):
parts, buf, i, n = [], [], 0, len(text)
while i < n:
c = text[i]
if c == '"':
j = i + 1
while j < n and text[j] != '"':
j += 2 if text[j] == "\\" else 1
j = min(j + 1, n)
buf.append(text[i:j]); i = j; continue
if c == sep:
parts.append("".join(buf)); buf = []; i += 1; continue
buf.append(c); i += 1
parts.append("".join(buf))
return parts
def unquote(tok):
tok = tok.strip()
if tok.startswith('"'):
end = tok.rindex('"')
return tok[1:end].replace('\\"', '"').replace("\\\\", "\\")
return tok
def parse(path):
"""Turtle subset reader: enough for an instance file, keeps ordering."""
text = strip_comments(open(path, encoding="utf-8").read())
subjects = OrderedDict()
for stmt in split_outside(text, "."):
stmt = stmt.strip()
if not stmt or not stmt.startswith("ex:"):
continue
clauses = split_outside(stmt, ";")
head = clauses[0].strip()
m = re.match(r"^(\S+)\s+(.*)$", head, re.S)
if not m:
continue
subj, rest = m.group(1), m.group(2)
rec = subjects.setdefault(subj, {"id": subj})
for chunk in [rest] + [c.strip() for c in clauses[1:]]:
chunk = chunk.strip()
if not chunk:
continue
pm = re.match(r"^(\S+)\s+(.*)$", chunk, re.S)
if not pm:
continue
pred, objs = pm.group(1), pm.group(2)
key = "type" if pred in ("a", "rdf:type") else pred.split(":")[-1]
vals = [unquote(t) for t in split_outside(objs, ",") if t.strip()]
if key in MULTI or key in rec:
rec.setdefault(key, [])
if not isinstance(rec[key], list):
rec[key] = [rec[key]]
rec[key] += vals
else:
rec[key] = vals[0] if len(vals) == 1 else vals
return subjects
def as_list(v):
if v is None:
return []
return v if isinstance(v, list) else [v]
# ---------------------------------------------------------------- model
def first(rec, key, default=""):
v = rec.get(key)
return v[0] if v else default
def parse_tbox(path):
"""Turtle subset reader for the T-Box: pr: subjects, WANTED predicates."""
text = strip_comments(open(path, encoding="utf-8").read())
subjects = OrderedDict()
for stmt in split_outside(text, "."):
stmt = stmt.strip()
if not stmt.startswith("pr:"):
continue
chunks = split_outside(stmt, ";")
m = re.match(r"^(\S+)\s+(.*)$", chunks[0].strip(), re.S)
if not m:
continue
rec = subjects.setdefault(m.group(1), {})
for chunk in [m.group(2)] + [c.strip() for c in chunks[1:]]:
chunk = chunk.strip()
if not chunk:
continue
pm = re.match(r"^(\S+)\s+(.*)$", chunk, re.S)
if not pm:
continue
key = WANTED.get(pm.group(1))
if key:
rec.setdefault(key, []).extend(
unquote(t) for t in split_outside(pm.group(2), ",") if t.strip())
return subjects
def build(subjects):
classes, objp, datp, annp, individuals = {}, {}, {}, {}, {}
for term, rec in subjects.items():
types = rec.get("type", [])
if "owl:Class" in types:
classes[term] = rec
elif "owl:ObjectProperty" in types:
objp[term] = rec
elif "owl:DatatypeProperty" in types:
datp[term] = rec
elif "owl:AnnotationProperty" in types:
annp[term] = rec
elif types and types[0].startswith("pr:"):
individuals.setdefault(types[0], []).append(term)
def ancestry(term, seen=None):
seen = seen or set()
if term in seen:
return []
seen.add(term)
out = [term]
for parent in classes.get(term, {}).get("subClassOf", []):
out += ancestry(parent, seen)
return out
parents = {c: first(classes[c], "subClassOf") for c in classes}
has_kids = set(parents.values())
nodes = {}
for term, rec in classes.items():
anc = ancestry(term)
layer = next((LAYER_ROOT[a][0] for a in anc if a in LAYER_ROOT), None)
if layer is None:
if "pr:Actor" in anc:
layer = "actors"
elif term in ("pr:ActivationStatus", "pr:Environment", "pr:SystemType"):
layer = "vocab"
else:
layer = "roots"
# provenance is INFERRED from the type-of chain, never asserted:
# a Sub-domain is Defined because it is a kind of Ownership Layer
# Object, which is a kind of Defined Object. No separate edge exists
# because none is needed -- the subclass chain already says it.
prov = ("captured" if "pr:CapturedObject" in anc
else "defined" if "pr:DefinedObject" in anc else "structural")
depth, p, guard = 0, parents.get(term), 0
while p and p in classes and guard < 8:
depth += 1; guard += 1; p = parents.get(p)
nodes[term] = {
"id": term, "kind": "Class",
"label": first(rec, "label", term.split(":")[1]),
"shortLabel": first(rec, "shortLabel"),
"layer": layer, "prov": prov, "depth": depth,
"parent": parents.get(term, ""),
"comment": first(rec, "comment"),
"harvest": first(rec, "harvestSource"),
"authoring": first(rec, "authoringMode"),
"acronym": first(rec, "acronym"),
"deprecated": first(rec, "deprecated") in ("true", "True"),
"replacedBy": first(rec, "isReplacedBy"),
"abstract": term in has_kids and term not in LAYER_ROOT
and term not in ("pr:MetaModelObject",),
"members": [{"id": m, "label": first(subjects.get(m, {}), "label",
m.split(":")[1])}
for m in individuals.get(term, [])],
}
def prop(term, rec, kind):
return {
"id": term, "kind": kind,
"label": first(rec, "label", term.split(":")[1]),
"shortLabel": first(rec, "shortLabel"),
"comment": first(rec, "comment"),
"domain": first(rec, "domain"), "range": first(rec, "range"),
"inverseOf": first(rec, "inverseOf"),
"subPropertyOf": first(rec, "subPropertyOf"),
"authoring": first(rec, "authoringMode"),
"harvest": first(rec, "harvestSource"),
"functional": "owl:FunctionalProperty" in rec.get("type", []),
"transitive": "owl:TransitiveProperty" in rec.get("type", []),
"deprecated": first(rec, "deprecated") in ("true", "True"),
"replacedBy": first(rec, "isReplacedBy"),
"seam": term in SEAM,
"prov": "", "layer": "",
}
layers = [{"id": v[0], "title": v[1], "tag": v[2]} for v in LAYER_ROOT.values()]
layers = ([{"id": "roots", "title": "Model roots", "tag": "how a term got here"}]
+ layers
+ [{"id": "actors", "title": "Actors", "tag": "people and teams"},
{"id": "vocab", "title": "Vocabularies", "tag": "controlled value lists"}])
return {"nodes": list(nodes.values()),
"relations": [prop(t, r, "Relation") for t, r in objp.items()],
"attributes": [prop(t, r, "Attribute") for t, r in datp.items()],
"bridges": [prop(t, r, "Bridge") for t, r in annp.items()
if t.startswith("pr:denotes")],
"layers": layers}
# ---------------------------------------------------------------- output
def main():
root = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
src = sys.argv[1] if len(sys.argv) > 1 else os.path.join(
root, "ontology", "pr_metamodel.ttl")
out = sys.argv[2] if len(sys.argv) > 2 else os.path.join(
root, "generated", "pr_metamodel_viewer.html")
model = build(parse_tbox(src))
payload = json.dumps(model, ensure_ascii=False, separators=(",", ":"))
html = TEMPLATE.replace("__MODEL__", payload).replace(
"__SRC__", os.path.basename(src))
os.makedirs(os.path.dirname(out), exist_ok=True)
open(out, "w", encoding="utf-8").write(html)
live_c = [n for n in model["nodes"] if not n["deprecated"]]
dep = ([n for n in model["nodes"] if n["deprecated"]]
+ [r for r in model["relations"] + model["attributes"] if r["deprecated"]])
print("PR META MODEL - T-Box viewer")
print(" source : %s" % src)
print(" output : %s" % out)
print(" classes %d | relations %d | attributes %d | bridges %d | deprecated %d"
% (len(live_c),
len([r for r in model["relations"] if not r["deprecated"]]),
len([a for a in model["attributes"] if not a["deprecated"]]),
len(model["bridges"]), len(dep)))
for l in model["layers"]:
c = len([n for n in live_c if n["layer"] == l["id"]])
if c:
print(" %-28s %d" % (l["title"], c))
for p in ("defined", "captured", "structural"):
c = len([n for n in live_c if n["prov"] == p])
print(" %-28s %d" % (p, c))
TEMPLATE = r"""
PR Data Meta Model · T-Box
PR Data Meta Model · T-Box
Six layers, crossed with what governance asserts and what a harvester observes