#!/usr/bin/env python3
"""
PR Data Meta Model - T-Box viewer generator
===========================================
Reads the T-Box Turtle file and emits a self-contained, zero-dependency HTML
viewer. The TTL is the source of truth; the viewer is a build artefact and must
never be edited by hand.
USAGE
python generate_tbox_viewer.py [tbox.ttl] [out.html]
Defaults: pr_datametamodel_v1_0.ttl -> pr_metamodel_viewer_v1_0.html
Uses rdflib when available; otherwise falls back to a restricted Turtle reader
that handles the subset of syntax used by this T-Box. Run it in CI on every
merge so the viewer can never drift from the model.
"""
import json
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:curationMode": "curationMode",
}
# --------------------------------------------------------------------------
# Turtle reading
# --------------------------------------------------------------------------
def strip_comments(text):
"""Remove # comments that sit outside string literals."""
out, i, n = [], 0, len(text)
while i < n:
c = text[i]
if text.startswith('"""', i):
j = text.find('"""', i + 3)
j = n if j == -1 else j + 3
out.append(text[i:j]); i = j; continue
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_top(text, sep):
"""Split on `sep` at nesting depth 0, ignoring string literals."""
parts, buf, depth, i, n = [], [], 0, 0, len(text)
while i < n:
c = text[i]
if text.startswith('"""', i):
j = text.find('"""', i + 3)
j = n if j == -1 else j + 3
buf.append(text[i:j]); i = j; continue
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 in "([":
depth += 1
elif c in ")]":
depth -= 1
if c == sep and depth == 0:
parts.append("".join(buf)); buf = []; i += 1; continue
buf.append(c); i += 1
if "".join(buf).strip():
parts.append("".join(buf))
return parts
def tokens(text):
"""Split an object list into individual terms."""
return [t.strip() for t in split_top(text, ",") if t.strip()]
def unquote(tok):
tok = tok.strip()
if tok.startswith('"""'):
return tok[3:tok.rindex('"""')] if tok.count('"""') > 1 else tok[3:]
if tok.startswith('"'):
end = tok.rindex('"')
return tok[1:end] if end > 0 else tok[1:]
return tok
def parse_ttl_fallback(text):
text = strip_comments(text)
subjects = OrderedDict()
for stmt in split_top(text, "."):
stmt = stmt.strip()
if not stmt:
continue
chunks = split_top(stmt, ";")
head = chunks[0].strip()
m = re.match(r"^(\S+)\s+(.*)$", head, re.S)
if not m:
continue
subj, rest = m.group(1), m.group(2)
if not subj.startswith("pr:"):
continue
rec = subjects.setdefault(subj, {})
for chunk in [rest] + [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
pred, objs = pm.group(1), pm.group(2)
key = WANTED.get(pred)
if not key:
continue
rec.setdefault(key, []).extend(unquote(t) for t in tokens(objs))
return subjects
def parse_ttl_rdflib(path):
from rdflib import Graph, RDF, RDFS, OWL, URIRef, Namespace
DCT = Namespace("http://purl.org/dc/terms/")
g = Graph(); g.parse(path, format="turtle")
inv = {
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",
DCT.isReplacedBy: "isReplacedBy",
URIRef(PR + "harvestSource"): "harvestSource",
URIRef(PR + "curationMode"): "curationMode",
}
def short(term):
s = str(term)
for pfx, ns in (("pr:", PR),
("owl:", "http://www.w3.org/2002/07/owl#"),
("rdfs:", "http://www.w3.org/2000/01/rdf-schema#"),
("xsd:", "http://www.w3.org/2001/XMLSchema#"),
("prov:", "http://www.w3.org/ns/prov#")):
if s.startswith(ns):
return pfx + s[len(ns):]
return s
subjects = OrderedDict()
for s, p, o in g:
key = inv.get(p)
if key is None or not str(s).startswith(PR):
continue
rec = subjects.setdefault(short(s), {})
rec.setdefault(key, []).append(str(o) if key in ("label", "comment", "harvestSource", "curationMode", "deprecated") else short(o))
return subjects
def read_tbox(path):
try:
return parse_ttl_rdflib(path), "rdflib"
except ImportError:
with open(path, encoding="utf-8") as fh:
return parse_ttl_fallback(fh.read()), "built-in reader"
# --------------------------------------------------------------------------
# Model assembly
# --------------------------------------------------------------------------
BANDS = [
("roots", "Model roots", "How a term got here", "#0A1F44"),
("ownership", "Ownership & Categorization", "Who is accountable", "#B8935A"),
("business", "Business", "What the business means", "#8B3A3A"),
("logical", "Logical", "How the data is structured", "#7A4E8B"),
("delivery", "Delivery", "What we promise, to whom", "#B8935A"),
("physical", "Physical", "What actually runs", "#1E4B9B"),
("consumption", "Consumption", "What people open in the morning", "#1E7A5F"),
("context", "Actors & vocabularies", "Outside the object stack", "#6B6B6B"),
]
LAYER_ROOT = {
"pr:OwnershipLayerObject": "ownership",
"pr:BusinessLayerObject": "business",
"pr:LogicalLayerObject": "logical",
"pr:DeliveryLayerObject": "delivery",
"pr:PhysicalLayerObject": "physical",
"pr:ConsumptionLayerObject": "consumption",
}
def first(rec, key, default=""):
v = rec.get(key)
return v[0] if v else default
def build(subjects):
classes, objprops, dataprops, annprops, individuals = {}, {}, {}, {}, {}
for term, rec in subjects.items():
types = rec.get("type", [])
if "owl:Class" in types:
classes[term] = rec
elif "owl:ObjectProperty" in types:
objprops[term] = rec
elif "owl:DatatypeProperty" in types:
dataprops[term] = rec
elif "owl:AnnotationProperty" in types:
annprops[term] = rec
elif types and types[0].startswith("pr:"):
individuals.setdefault(types[0], []).append(term)
# layer + provenance by walking subClassOf upward
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
nodes = []
for term, rec in classes.items():
anc = ancestry(term)
band = next((LAYER_ROOT[a] for a in anc if a in LAYER_ROOT), None)
if band is None:
if "pr:Actor" in anc or term in ("pr:ActivationStatus", "pr:Environment", "pr:SystemType"):
band = "context"
else:
band = "roots"
provenance = ("observed" if "pr:ObservedObject" in anc
else "curated" if "pr:CuratedObject" in anc else "neutral")
nodes.append({
"id": term,
"label": first(rec, "label", term.split(":")[1]),
"band": band,
"prov": provenance,
"parent": first(rec, "subClassOf"),
"comment": first(rec, "comment"),
"harvest": first(rec, "harvestSource"),
"curation": first(rec, "curationMode"),
"deprecated": first(rec, "deprecated") in ("true", "True"),
"replacedBy": first(rec, "isReplacedBy"),
"members": [
{"id": m, "label": first(subjects.get(m, {}), "label", m.split(":")[1])}
for m in individuals.get(term, [])
],
})
known = {n["id"] for n in nodes}
edges, universal = [], []
for term, rec in objprops.items():
dom, rng = first(rec, "domain"), first(rec, "range")
item = {
"id": term,
"label": first(rec, "label", term.split(":")[1]),
"s": dom, "t": rng,
"comment": first(rec, "comment"),
"harvest": first(rec, "harvestSource"),
"curation": first(rec, "curationMode"),
"inverseOf": first(rec, "inverseOf"),
"subPropertyOf": first(rec, "subPropertyOf"),
"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"),
}
if dom in known and rng in known:
edges.append(item)
else:
item["s"] = dom or "pr:MetaModelObject"
universal.append(item)
datatypes = [{
"id": t, "label": first(r, "label", t.split(":")[1]),
"range": first(r, "range"), "comment": first(r, "comment"),
"harvest": first(r, "harvestSource"),
"functional": "owl:FunctionalProperty" in r.get("type", []),
"deprecated": first(r, "deprecated") in ("true", "True"),
"replacedBy": first(r, "isReplacedBy"),
} for t, r in dataprops.items()]
bridges = [{
"id": t, "label": first(r, "label", t.split(":")[1]),
"comment": first(r, "comment"),
} for t, r in annprops.items() if t.startswith("pr:denotes")]
onto = subjects.get("", {})
return {
"bands": [{"id": b[0], "title": b[1], "tag": b[2], "color": b[3]} for b in BANDS],
"nodes": nodes, "edges": edges, "universal": universal,
"datatypes": datatypes, "bridges": bridges,
}
# --------------------------------------------------------------------------
# HTML emission
# --------------------------------------------------------------------------
def emit(model, version, src_name, out_path):
payload = json.dumps(model, ensure_ascii=False, separators=(",", ":"))
html = (TEMPLATE.replace("__MODEL__", payload)
.replace("__SRC__", src_name)
.replace("__VERSION__", version))
with open(out_path, "w", encoding="utf-8") as fh:
fh.write(html)
TEMPLATE = r"""
PR Data Meta Model · T-Box __VERSION__
PR Data Meta Model · T-Box
Five layers, crossed with a provenance axis: what governance asserts, what harvesters observe
0Classes
0Relations
0Attributes
0Seam edges
drag to pan · scroll to zoom · click a class
"""
def main():
src = sys.argv[1] if len(sys.argv) > 1 else "pr_datametamodel_v1_0.ttl"
out = sys.argv[2] if len(sys.argv) > 2 else "pr_metamodel_viewer_v1_0.html"
subjects, engine = read_tbox(src)
model = build(subjects)
import os
emit(model, "v1.0", os.path.basename(src), out)
print("PR META MODEL - T-Box viewer")
print(" source : %s (parsed with %s)" % (src, engine))
print(" output : %s" % out)
print(" classes %d | object properties %d | datatype properties %d | bridge annotations %d"
% (len(model["nodes"]),
len(model["edges"]) + len(model["universal"]),
len(model["datatypes"]), len(model["bridges"])))
for b in model["bands"]:
c = len([n for n in model["nodes"] if n["band"] == b["id"]])
if c:
print(" %-26s %d" % (b["title"], c))
if __name__ == "__main__":
main()