#!/usr/bin/env python3
"""
SODH instance viewer generator
===============================
Reads instances/sodh.ttl and emits a self-contained HTML view of the Sell Out
Data Hub. Build artefact: regenerate, never edit.
USAGE
python3 scripts/generate_sodh_viewer.py [in.ttl] [out.html]
defaults: instances/sodh.ttl -> generated/sodh_viewer.html
WHAT IT IS FOR
Making one thing obvious to a business audience: SODH is owned by Sales
Performance but it cannot exist without five other domains. The default
view is the scope, not the graph -- the point is federation, not topology.
Objects flagged TO_ARBITRATE are shown, not hidden. A declared gap is the
argument: it says which decisions are pending and who owes them.
"""
import json
import os
import re
import sys
from collections import OrderedDict
PR = "https://ontology.pernod-ricard.com/metamodel/"
DOMAIN_STYLE = OrderedDict([
("DD_06", ("#0A1F44", "Sales Performance", "owner")),
("DD_10", ("#1E4B9B", "Product & Material", "contributor")),
("DD_04", ("#1E7A5F", "Customer & Channels", "contributor")),
("DD_16", ("#8B3A3A", "Financial Performance", "contributor")),
("DD_21", ("#7A4E8B", "Data Governance", "contributor")),
("DD_05", ("#B8935A", "Sales Activation", "contributor")),
])
MULTI = {"composedOf", "usesConcept", "hasElement", "hasMetric", "measures",
"computedBy", "references", "hasGrainElement", "monitoredBy", "hasSynonym",
"hasUnit", "hasSource", "packages", "exposes"}
SUBJECT_DOMAIN = "DD_06" # the hub owner: everything else is a dependency
# Every user-facing label in one place. Hyphenated throughout, and never the
# raw class name: "Physical Relation" and "Physicalized in" told a business
# reader nothing.
LABELS = {
"DataDomain": "Domain", "SubDomain": "Sub-domain",
"BusinessObject": "Business object", "BusinessConcept": "Business concept",
"Metric": "Metric", "DataObject": "Data object", "DataElement": "Data element",
"BaseTable": "Base table", "Field": "Field",
"DataProduct": "Data product", "DataContract": "Data contract",
"DataInterface": "Data interface",
}
# Which ownership roles a card shows. The Steward section is gone: one section,
# different roles per object, so the reader always looks in the same place.
ROLES = {
"DataDomain": [("Domain owner", "owner"), ("Data-governance lead", "dgl")],
"SubDomain": [("Sub-domain owner", "subOwner"), ("Data-governance lead", "dgl_up")],
"BusinessObject": [("Steward", "steward")],
"BusinessConcept": [("Steward", "steward")],
"Metric": [("Steward", "steward")],
"DataObject": [("Product owner", "productOwner")],
"DataElement": [("Steward", "steward_inherited")],
"DataProduct": [("Product owner", "productOwner")],
}
# ---------------------------------------------------------------- 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 domain_of(subj, rec, subjects):
"""Owning domain, walking up belongsTo when not declared."""
d = rec.get("owningDomain")
if d:
return d.replace("ex:", "")
seen = set()
cur = rec
while cur is not None:
parent = cur.get("belongsTo") or cur.get("represents")
if not parent or parent in seen:
break
seen.add(parent)
cur = subjects.get(parent)
if cur and cur.get("owningDomain"):
return cur["owningDomain"].replace("ex:", "")
m = re.match(r"ex:[A-Z]+_(\d\d)_", subj)
return "DD_" + m.group(1) if m else "DD_06"
def build(subjects):
nodes = {}
for subj, rec in subjects.items():
t = (rec.get("type") or "").replace("pr:", "")
if not t:
continue
nodes[subj] = {
"id": subj,
"kind": t,
"label": LABELS.get(t, t),
"ident": rec.get("hasIdentifier", ""),
"name": rec.get("hasName", subj.replace("ex:", "")),
"domain": domain_of(subj, rec, subjects),
"arb": rec.get("arbitrationStatus", "") == "TO_ARBITRATE",
"status": rec.get("hasStatus", ""),
"definition": rec.get("hasBusinessDefinition", ""),
"rule": rec.get("hasBusinessRule", ""),
"formula": rec.get("hasFormula", ""),
"units": as_list(rec.get("hasUnit")),
"physical": rec.get("physicalName", ""),
"format": rec.get("hasFormat", ""),
"source": ", ".join(as_list(rec.get("hasSource"))),
"version": rec.get("hasVersion", ""),
"operatedBy": rec.get("operatedBy", ""),
"governedBy": rec.get("governedBy", ""),
"productOwner": rec.get("hasProductOwner", ""),
"packages": as_list(rec.get("packages")),
"exposes": as_list(rec.get("exposes")),
"belongsTo": rec.get("belongsTo", ""),
"about": rec.get("aboutConcept", ""),
"uses": as_list(rec.get("usesConcept")),
"represents": rec.get("represents", ""),
"measures": as_list(rec.get("measures")),
"hasMetric": as_list(rec.get("hasMetric")),
"computedBy": as_list(rec.get("computedBy")),
"hasElement": as_list(rec.get("hasElement")),
"grain": as_list(rec.get("hasGrainElement")),
"references": as_list(rec.get("references")),
"steward": as_list(rec.get("monitoredBy")) or as_list(rec.get("ownedBy")),
"dgl": rec.get("hasDGL", ""),
"owner": rec.get("hasDomainOwner", ""),
"subOwner": rec.get("hasSubDomainOwner", ""),
"activation": rec.get("hasActivationStatus", "").replace("pr:", ""),
}
# reverse index: which metric computes an element
for n in nodes.values():
for de in n["computedBy"]:
if de in nodes:
nodes[de].setdefault("computedFrom", []).append(n["id"])
for n in nodes.values():
n.setdefault("computedFrom", [])
n["route"] = ("measure" if n["computedFrom"]
else "dimensional" if n["represents"] else "")
# ---- derived physical layer -------------------------------------------
# Not asserted anywhere: these nodes are PROJECTED from physicalName on the
# data objects and elements. The real physical layer is meant to be
# harvested from Snowflake, not typed by hand, so showing it as derived is
# the honest option -- it makes the gap visible instead of pretending it is
# filled.
phys = {}
for n in list(nodes.values()):
if n["kind"] == "DataObject" and n["physical"]:
pid = "phys:" + n["physical"]
phys.setdefault(pid, {"id": pid, "kind": "BaseTable",
"label": LABELS["BaseTable"],
"name": n["physical"], "domain": n["domain"],
"derived": True, "backs": []})
phys[pid]["backs"].append(n["id"])
n["materializedAs"] = pid
for n in list(nodes.values()):
if n["kind"] == "DataElement" and n["physical"]:
owner = [d for d in nodes.values()
if d["kind"] == "DataObject" and n["id"] in d["hasElement"]]
rel = owner[0]["physical"] if owner and owner[0]["physical"] else ""
pid = "phys:" + (rel + "." if rel else "") + n["physical"]
phys.setdefault(pid, {"id": pid, "kind": "Field",
"label": LABELS["Field"], "name": n["physical"],
"domain": n["domain"], "derived": True,
"relation": "phys:" + rel if rel else "",
"backs": []})
phys[pid]["backs"].append(n["id"])
n["storedIn"] = pid
for p in phys.values():
p.setdefault("ident", ""); p.setdefault("arb", False)
p.setdefault("label", LABELS.get(p["kind"], p["kind"]))
for k in ("definition","rule","formula","physical","format","source","belongsTo",
"about","represents","status","dgl","owner","subOwner","activation",
"version","operatedBy","governedBy","productOwner","storedIn",
"materializedAs","relation","stewardFrom"):
p.setdefault(k, "")
for k in ("uses","measures","hasMetric","computedBy","hasElement","grain",
"references","steward","units","computedFrom","packages","exposes",
"steward_inherited"):
p.setdefault(k, [])
p.setdefault("route", "")
nodes[p["id"]] = p
# ---- inherited roles, resolved once -----------------------------------
# Stewardship is declared on the Business Object and inherited: a Data
# Object through represents, a Data Element through its Data Object. The
# card shows the inherited holder and says so, rather than leaving a blank
# that reads as "nobody".
for n in nodes.values():
if n["kind"] == "DataObject":
bo = nodes.get(n["represents"])
n["steward_inherited"] = bo["steward"] if bo else []
n["stewardFrom"] = n["represents"]
if n["kind"] == "DataElement":
do = [d for d in nodes.values()
if d["kind"] == "DataObject" and n["id"] in d["hasElement"]]
if do:
n["steward_inherited"] = do[0].get("steward_inherited", [])
n["stewardFrom"] = do[0]["id"]
domains = []
for code, (colour, label, role) in DOMAIN_STYLE.items():
members = [n for n in nodes.values() if n["domain"] == code]
if not members:
continue
d = nodes.get("ex:" + code, {})
domains.append({
"code": code,
"colour": colour,
"label": d.get("name") or label,
"role": role,
"dgl": nodes.get(d.get("dgl", ""), {}).get("name", "—"),
"owner": nodes.get(d.get("owner", ""), {}).get("name", "—"),
"activation": d.get("activation", ""),
"counts": {k: len([m for m in members if m["kind"] == k])
for k in ("SubDomain", "BusinessObject", "BusinessConcept",
"Metric", "DataObject", "DataElement",
"BaseTable", "Field")},
"arb": len([m for m in members if m["arb"]]),
})
return {"nodes": list(nodes.values()), "domains": domains,
"subject": SUBJECT_DOMAIN, "labels": LABELS,
"roles": ROLES,
"styles": {k: v[0] for k, v in DOMAIN_STYLE.items()}}
# ---------------------------------------------------------------- output
def main():
src = sys.argv[1] if len(sys.argv) > 1 else os.path.join(
os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "instances", "sodh.ttl")
out = sys.argv[2] if len(sys.argv) > 2 else os.path.join(
os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "generated", "sodh_viewer.html")
model = build(parse(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)
print("SODH VIEWER")
print(" source : %s" % src)
print(" output : %s" % out)
tot = {}
for n in model["nodes"]:
tot[n["kind"]] = tot.get(n["kind"], 0) + 1
print(" " + " | ".join("%s %d" % (k, v) for k, v in sorted(tot.items())))
print(" domains %d | to arbitrate %d"
% (len(model["domains"]), len([n for n in model["nodes"] if n["arb"]])))
for d in model["domains"]:
c = d["counts"]
print(" %-6s %-24s BO %2d BC %2d M %2d DE %3d arb %2d %s"
% (d["code"].replace("DD_", "DD-"), d["label"], c["BusinessObject"],
c["BusinessConcept"], c["Metric"], c["DataElement"], d["arb"],
"OWNER" if d["role"] == "owner" else ""))
TEMPLATE = r"""
Sell Out Data Hub · scope and dependencies