import re, uuid, copy, io
from pathlib import Path
from rdflib import Graph, Namespace, RDF, RDFS, URIRef, Literal, BNode
from rdflib.namespace import XSD
from pyvis.network import Network
import IPython.display as display
import pandas as pd
import ipywidgets as widgets
AMT = Namespace("http://academic-meta-tool.xyz/vocab#")
# ── shared state (populated by upload callback) ───────────────────────────
amt = None
concepts = {}
roles = {}
nodes = {}
edges = []
axioms = []
PREFIX = ""
out = widgets.Output()
def _on_upload(change):
global amt, concepts, roles, nodes, edges, axioms, PREFIX
with out:
out.clear_output()
uploaded = uploader.value
if not uploaded:
return
file_info = uploaded[0] if isinstance(uploaded, tuple) else list(uploaded.values())[0]
filename = file_info["name"]
raw_bytes = file_info["content"]
ttl_str = (raw_bytes.tobytes() if hasattr(raw_bytes, "tobytes") else raw_bytes).decode("utf-8")
print(f"📂 {filename} ({len(ttl_str):,} chars)")
try:
amt = load_amt(ttl_str)
concepts = amt["concepts"]
roles = amt["roles"]
nodes = amt["nodes"]
edges = amt["edges"]
axioms = amt["axioms"]
if nodes:
PREFIX = next(iter(nodes.values()))["id"].rsplit("/", 1)[0] + "/"
print(f"✓ {len(concepts)} Concepts | {len(roles)} Roles | "
f"{len(nodes)} Nodes | {len(edges)} Edges | {len(axioms)} Axioms")
print("\nFile loaded. Run the cells below to analyse.")
except Exception as e:
print(f"✗ Error: {e}")
uploader = widgets.FileUpload(
accept=".ttl", multiple=False,
description="Upload TTL", button_style="primary",
layout=widgets.Layout(width="160px"),
)
uploader.observe(_on_upload, names="value")
display.display(widgets.VBox([
widgets.HTML("<b>Select an AMT-compatible Turtle (.ttl) file:</b>"),
uploader,
out,
]))Academic Meta Tool (AMT) – Python Implementation
About this notebook
This notebook is a full Python port of the Academic Meta Tool (AMT), originally written in JavaScript using N3.js and vis.js. It demonstrates how an archaeological domain expert can encode graded (fuzzy) attributions between finds, figure types, potters and find contexts as a small ontology, and then let a reasoner derive new weighted relations from the asserted ones.
Unlike the browser companion, which ships with a fixed example TTL file and runs in Pyodide, this notebook lets you upload any AMT-compatible Turtle (.ttl) file and analyse it with the full local scientific Python stack. Use it when you want to:
- point the tool at your own ontology,
- experiment with larger graphs than Pyodide comfortably handles,
- export the reasoned graph back to Turtle or Cypher (Neo4j),
- or integrate AMT into a larger local workflow.
What you’ll learn
- How AMT’s tiny RDF vocabulary (
amt:Concept,amt:Role,amt:weight,amt:RoleChainAxiom, …) models a fuzzy attribution graph. - How to parse such a graph with
rdfliband rebuild the concept / role / node / edge / axiom structures that AMT’s reasoner needs. - How
RoleChainAxiomandInverseAxiomderive new weighted edges under Łukasiewicz, Product or Gödel logic, and howDisjointAxiom/SelfDisjointAxiomflag inconsistencies. - How to export the reasoned graph to Turtle (round-trippable RDF) or Cypher (Neo4j).
Requirements
pip install rdflib pyvis pandas ipywidgets jupyterTested with Python ≥ 3.10 in JupyterLab and VS Code. ipywidgets needs the notebook extension enabled; JupyterLab ≥ 4 and recent VS Code builds handle this automatically.
How to use this notebook
- Run Step 1 below to show the file upload widget.
- Upload your
.ttlfile – wait for the ✓ confirmation. - Then use Run All Below, or execute the cells one by one.
Running Run All without uploading a file first stops at the first analysis cell with a clear error message, so you can’t accidentally analyse an empty graph.
Data-context notes
- Weights on edges live in the interval [0, 1] and are interpreted as degrees of attribution, not probabilities.
- The ontology uses the reified-statement pattern (
rdf:subject/rdf:predicate/rdf:object/amt:weight) so that each edge is itself an RDF resource and can carry its own weight literal. - Inferred edges shown in the visualisation are marked in red dashed style. They are reasoner output, not source data – you decide whether to round-trip them via the TTL export in Step 9.
Tooling notes
- The local variant uses
SPARQLWrapper-style freedom:rdflibfor parsing,pyvisfor interactive graph rendering,ipywidgetsfor file upload,pandasfor tabular inspection. - For a zero-install browser-only preview (fixed example file), see the companion interactive browser notebook.
Step 1 – Upload your TTL file
Run this cell and upload an AMT-compatible Turtle (.ttl) file. All cells below depend on this step.
Step 2 – AMT core engine
Direct Python port of amt.js. Three components:
load_amt()– parses a TTL file withrdfliband extracts Concepts / Roles / Nodes / Edges / Axioms.do_reasoning()– appliesRoleChainAxiomandInverseAxiomiteratively (fuzzy logic: Łukasiewicz / Product / Gödel).check_consistency()– validatesDisjointAxiomandSelfDisjointAxiomintegrity constraints.
Keeping the reasoner pure-Python (no OWL, no rdflib inference plug-ins) is deliberate: it mirrors the original AMT behaviour exactly and makes the fuzzy-logic choices explicit.
def _local(iri: str) -> str:
"""Return the local name of an IRI (after last # or /)."""
return str(iri).split("/")[-1].split("#")[-1]
def load_amt(ttl_source):
"""
Parse a Turtle file and extract all AMT components.
Parameters
----------
ttl_source : str or Path
Path to a .ttl file, or raw Turtle string.
Returns
-------
dict with keys: concepts, roles, nodes, edges, axioms, graph (rdflib.Graph)
"""
g = Graph()
src = str(ttl_source)
if Path(src).exists():
g.parse(src, format="turtle")
else:
g.parse(data=src, format="turtle")
# ── Concepts ──────────────────────────────────────────────────────────
concepts = {}
for c in g.subjects(RDF.type, AMT.Concept):
label = str(g.value(c, RDFS.label) or _local(c))
placeholder = str(g.value(c, AMT.placeholder) or label)
concepts[str(c)] = {"iri": str(c), "label": label, "placeholder": placeholder}
# ── Roles ─────────────────────────────────────────────────────────────
roles = {}
for r in sorted(g.subjects(RDF.type, AMT.Role),
key=lambda x: str(g.value(x, RDFS.label) or x)):
label = str(g.value(r, RDFS.label) or _local(r))
domain = str(g.value(r, RDFS.domain) or "")
range_ = str(g.value(r, RDFS.range) or "")
roles[str(r)] = {"iri": str(r), "label": label,
"domain": domain, "range": range_}
# ── Nodes (instances) ─────────────────────────────────────────────────
nodes = {}
for concept_iri in concepts:
for inst in g.subjects(AMT.instanceOf, URIRef(concept_iri)):
label = str(g.value(inst, RDFS.label) or _local(inst))
nodes[str(inst)] = {"id": str(inst), "label": label,
"concept": concept_iri}
# ── Edges (reified statements with weight) ────────────────────────────
edges = []
for stmt in g.subjects(AMT.weight, None):
frm = g.value(stmt, RDF.subject)
role = g.value(stmt, RDF.predicate)
to = g.value(stmt, RDF.object)
w = g.value(stmt, AMT.weight)
if frm and role and to and w is not None:
edges.append({"role": str(role),
"from": str(frm),
"to": str(to),
"weight": min(float(w), 1.0)})
# ── Axioms ────────────────────────────────────────────────────────────
axiom_types: set = set()
for cls in g.subjects(RDFS.subClassOf, AMT.Axiom):
axiom_types.add(cls)
for sub in g.subjects(RDFS.subClassOf, cls):
axiom_types.add(sub)
axioms = []
for atype in axiom_types:
for axiom in g.subjects(RDF.type, atype):
entry = {"type": _local(atype)}
for _, p, o in g.triples((axiom, None, None)):
if p != RDF.type:
entry[_local(p)] = str(o)
axioms.append(entry)
return {"concepts": concepts, "roles": roles, "nodes": nodes,
"edges": edges, "axioms": axioms, "graph": g}
# ── Fuzzy conjunction operators (mirrors amt.js) ──────────────────────────
def _conjunction(x: float, y: float, logic: str) -> float:
AMT_PFX = "http://academic-meta-tool.xyz/vocab#"
if logic == AMT_PFX + "LukasiewiczLogic": return max(x + y - 1, 0.0)
if logic == AMT_PFX + "ProductLogic": return x * y
if logic == AMT_PFX + "GoedelLogic": return min(x, y)
return 0.0
def do_reasoning(edges: list, axioms: list) -> list:
"""
Apply RoleChain and Inverse axioms iteratively until no new inferences.
Inferred edges are tagged with {'inferred': True}.
Returns a new list (original is not modified).
"""
result = copy.deepcopy(edges)
def _find(role, frm, to):
return next((e for e in result
if e["role"] == role
and e["from"] == frm
and e["to"] == to), None)
changed = True
while changed:
changed = False
for axiom in axioms:
if axiom["type"] == "RoleChainAxiom":
ant1 = axiom.get("antecedent1")
ant2 = axiom.get("antecedent2")
cons = axiom.get("consequent")
logic = axiom.get("logic", "")
for e1 in list(result):
for e2 in list(result):
if (e1["to"] == e2["from"]
and e1["role"] == ant1
and e2["role"] == ant2):
w = _conjunction(e1["weight"], e2["weight"], logic)
if w <= 0:
continue
w = min(round(w, 6), 1.0)
existing = _find(cons, e1["from"], e2["to"])
if existing is None:
result.append({"role": cons, "from": e1["from"],
"to": e2["to"], "weight": w,
"inferred": True})
changed = True
elif existing["weight"] < w:
existing["weight"] = w
changed = True
elif axiom["type"] == "InverseAxiom":
ant = axiom.get("antecedent")
inv = axiom.get("inverse")
for e in list(result):
if e["role"] == ant:
existing = _find(inv, e["to"], e["from"])
if existing is None:
result.append({"role": inv, "from": e["to"],
"to": e["from"], "weight": e["weight"],
"inferred": True})
changed = True
return result
def check_consistency(edges: list, axioms: list) -> tuple[bool, list]:
"""
Check DisjointAxiom and SelfDisjointAxiom integrity constraints.
Returns (is_consistent, list_of_violations).
"""
reasoned = do_reasoning(edges, axioms)
violations = []
def _find(role, frm, to):
return next((e for e in reasoned
if e["role"] == role
and e["from"] == frm
and e["to"] == to), None)
for axiom in axioms:
if axiom["type"] == "DisjointAxiom":
r1, r2 = axiom.get("role1"), axiom.get("role2")
for e in reasoned:
if e["role"] == r1 and _find(r2, e["from"], e["to"]):
violations.append(f"DisjointAxiom violated: {_local(e['from'])} "
f"has both {_local(r1)} and {_local(r2)} to {_local(e['to'])}")
if axiom["type"] == "SelfDisjointAxiom":
role = axiom.get("role")
for e in reasoned:
if e["role"] == role and e["from"] == e["to"]:
violations.append(f"SelfDisjointAxiom violated: "
f"{_local(e['from'])} has self-loop via {_local(role)}")
return (len(violations) == 0, violations)
print("✓ AMT engine loaded")Step 3 – Graph visualisation
Port of amt-render.js: same colour palette and node / edge styling logic, using pyvis (which wraps vis.js internally).
Each render is written to a small HTML file and embedded in the notebook cell as an iframe via its srcdoc attribute, so multiple graphs on the same page are sandboxed and don’t share state.
# Colour palette – same order as AMT_PALETTE in amt-render.js
_PALETTE = [
{"bg": "coral", "border": "#c05a00", "font": "#000"},
{"bg": "#5b9bd5", "border": "#2e75b6", "font": "#fff"},
{"bg": "#70ad47", "border": "#538135", "font": "#fff"},
{"bg": "#ffc000", "border": "#c07a00", "font": "#000"},
{"bg": "#7030a0", "border": "#4b1a6e", "font": "#fff"},
{"bg": "#ed7d31", "border": "#843c00", "font": "#fff"},
]
def visualise_amt(
nodes: dict,
edges: list,
concepts: dict,
reasoning: bool = False,
axioms: list = None,
height: str = "600px",
notebook: bool = True,
) -> Network:
"""
Build and display an interactive AMT graph.
Parameters
----------
nodes : dict from load_amt()
edges : list from load_amt() (or do_reasoning())
concepts : dict from load_amt()
reasoning : if True, run do_reasoning() and show inferred edges in red dashes
axioms : required when reasoning=True
height : CSS height string for the vis canvas
notebook : pass True when running in Jupyter/Quarto
"""
display_edges = (do_reasoning(edges, axioms or []) if reasoning else edges)
net = Network(
height=height, width="100%",
directed=True, notebook=notebook,
bgcolor="#ffffff", font_color="#333333",
cdn_resources="remote",
)
net.barnes_hut(
gravity=-8000, spring_length=350,
spring_strength=0.02, damping=0.5,
)
# concept → colour
concept_list = list(concepts.keys())
color_map = {c: _PALETTE[i % len(_PALETTE)]
for i, c in enumerate(concept_list)}
for node in nodes.values():
col = color_map.get(node["concept"], _PALETTE[0])
concept_label = concepts.get(node["concept"], {}).get("label", "?")
net.add_node(
node["id"],
label=node["label"],
color={"background": col["bg"], "border": col["border"],
"highlight": {"background": "white",
"border": col["border"]}},
shape="dot", size=20,
font={"size": 16, "face": "monospace", "color": "#000"},
title=f"Concept: {concept_label}",
)
for e in display_edges:
inferred = e.get("inferred", False)
role_short = _local(e["role"])
w = round(e["weight"], 3)
net.add_edge(
e["from"], e["to"],
label=f"{role_short}: {w}",
width=max(1, w * 5),
color={"color": "#cc0000" if inferred else "#333333",
"highlight": "#555555"},
arrows="to",
dashes=inferred,
font={"size": 11, "face": "monospace",
"color": "#cc0000" if inferred else "#333333",
"align": "middle"},
title="inferred" if inferred else "asserted",
)
return net
def show_amt(
nodes, edges, concepts,
reasoning=False, axioms=None,
height="600px",
filename="_amt_graph.html",
):
"""Render the AMT graph inline. Works in VS Code, JupyterLab, and classic Notebook."""
net = visualise_amt(nodes, edges, concepts,
reasoning=reasoning, axioms=axioms,
height=height, notebook=False)
net.save_graph(filename)
# Read back and embed as srcdoc to avoid iframe src path issues
html_content = Path(filename).read_text(encoding="utf-8")
# Escape for srcdoc attribute
html_escaped = html_content.replace('&', '&').replace('"', '"')
h = int(height.replace('px','')) + 20
display.display(display.HTML(
f'<iframe srcdoc="{html_escaped}" '
f'width="100%" height="{h}px" '
f'style="border:1px solid #ddd; border-radius:4px;"></iframe>'
))
print("✓ Visualisation helpers loaded")Step 4 – Export functions
Port of amt-export.js: Turtle serialisation (round-trippable RDF, with optional inferred edges marked amt:inferred "true"^^xsd:boolean) and Cypher export for Neo4j.
def export_ttl(
nodes: dict,
edges: list,
concepts: dict,
roles: dict,
axioms: list,
rdf_graph: Graph,
prefix: str,
with_reasoning: bool = False,
) -> str:
"""
Serialise the current AMT state to Turtle.
Mirrors exportTTL() from amt-export.js.
"""
AMT_NS = "http://academic-meta-tool.xyz/vocab#"
RDF_NS = "http://www.w3.org/1999/02/22-rdf-syntax-ns#"
RDFS_NS = "http://www.w3.org/2000/01/rdf-schema#"
XSD_NS = "http://www.w3.org/2001/XMLSchema#"
display_edges = do_reasoning(edges, axioms) if with_reasoning else edges
base_count = len(edges)
def pfx(iri):
if not iri: return iri
if iri.startswith(prefix): return "ex:" + iri[len(prefix):]
if iri.startswith(AMT_NS): return "amt:" + iri[len(AMT_NS):]
if iri.startswith(RDF_NS): return "rdf:" + iri[len(RDF_NS):]
if iri.startswith(RDFS_NS): return "rdfs:" + iri[len(RDFS_NS):]
if iri.startswith(XSD_NS): return "xsd:" + iri[len(XSD_NS):]
return f"<{iri}>"
lines = []
lines += [
f"@prefix amt: <{AMT_NS}> .",
f"@prefix rdf: <{RDF_NS}> .",
f"@prefix rdfs: <{RDFS_NS}> .",
f"@prefix xsd: <{XSD_NS}> .",
f"@prefix ex: <{prefix}> .", "",
]
# Concepts
lines.append("# Concepts")
for c in concepts.values():
lines += [pfx(c["iri"]),
' rdf:type amt:Concept ;',
f' rdfs:label "{c["label"]}" ;',
f' amt:placeholder "{c["placeholder"]}" .',
""]
# Roles
lines.append("# Roles")
for r in roles.values():
lines += [pfx(r["iri"]),
' rdf:type amt:Role ;',
f' rdfs:label "{r["label"]}" ;',
f' rdfs:domain {pfx(r["domain"])} ;',
f' rdfs:range {pfx(r["range"])} .',
""]
# Instances
lines.append("# Instances")
for n in nodes.values():
lines += [pfx(n["id"]),
f' amt:instanceOf {pfx(n["concept"])} ;',
f' rdfs:label "{n["label"]}" .',
""]
# Assertions
lines.append("# Original Assertions")
for j, e in enumerate(display_edges[:base_count]):
w = min(float(e["weight"]), 1.0)
bn = f"_:a{j+1}"
lines += [bn,
f' rdf:subject {pfx(e["from"])} ;',
f' rdf:predicate {pfx(e["role"])} ;',
f' rdf:object {pfx(e["to"])} ;',
f' amt:weight "{w:.6f}"^^xsd:double .',
""]
if with_reasoning and len(display_edges) > base_count:
lines.append("# Inferred Assertions")
for k, e in enumerate(display_edges[base_count:]):
w = min(float(e["weight"]), 1.0)
bn = f"_:i{k+1}"
lines += [bn,
f' rdf:subject {pfx(e["from"])} ;',
f' rdf:predicate {pfx(e["role"])} ;',
f' rdf:object {pfx(e["to"])} ;',
f' amt:weight "{w:.6f}"^^xsd:double ;',
' amt:inferred "true"^^xsd:boolean .',
""]
return "\n".join(lines)
def export_cypher(
nodes: dict,
edges: list,
axioms: list,
with_reasoning: bool = False,
) -> str:
"""Serialise to Neo4J Cypher. Mirrors exportCypher() from amt-export.js."""
def cypher_safe(s):
return re.sub(r"[^a-zA-Z0-9_]", "_", s)
display_edges = do_reasoning(edges, axioms) if with_reasoning else edges
base_count = len(edges)
var_map = {n["id"]: cypher_safe(_local(n["id"])) for n in nodes.values()}
lines = ["// AMT Cypher export",
f"// Nodes: {len(nodes)} Edges: {len(display_edges)}",
f"// (inferred: {len(display_edges)-base_count})", ""]
node_lines, var_list = [], []
for n in nodes.values():
var = var_map[n["id"]]
label = cypher_safe(_local(n["concept"]))
lbl = n["label"].replace('"', '\\"')
node_lines.append(
f'MERGE ({var}:{label} {{id: "{_local(n["id"])}"}})' "\n"
f' ON CREATE SET {var}.label = "{lbl}", '
f'{var}.concept = "{_local(n["concept"])}"'
)
var_list.append(var)
lines.append("// Step 1: nodes")
lines.append("\n".join(node_lines))
lines.append("WITH " + ", ".join(var_list))
lines.append("")
lines.append("// Step 2: relationships")
for j, e in enumerate(display_edges):
w = round(min(float(e["weight"]), 1.0), 6)
rel = cypher_safe(_local(e["role"])).upper()
fv = var_map.get(e["from"], cypher_safe(_local(e["from"])))
tv = var_map.get(e["to"], cypher_safe(_local(e["to"])))
inferred= "true" if e.get("inferred") else "false"
lines.append(
f'MERGE ({fv})-[:{rel} {{weight: {w}, '
f'role: "{_local(e["role"])}", inferred: {inferred}}}]->({tv})'
)
lines += ["", "RETURN *"]
return "\n".join(lines)
print("✓ Export functions loaded")Step 5 – Inspect the ontology structure
A quick pandas-based overview of what the uploaded TTL file contains: concepts, roles (with domain and range), and the axioms the reasoner will apply.
assert amt is not None, "⚠ Upload a TTL file in Step 1 first, then re-run this cell."
print("=== Concepts ===")
df_concepts = pd.DataFrame([
{"IRI (local)": _local(c["iri"]), "Label": c["label"], "Placeholder": c["placeholder"]}
for c in concepts.values()
])
display.display(df_concepts)
print("\n=== Roles ===")
df_roles = pd.DataFrame([
{"IRI (local)": _local(r["iri"]),
"Label": r["label"],
"Domain": _local(r["domain"]),
"Range": _local(r["range"])}
for r in roles.values()
])
display.display(df_roles)
print("\n=== Axioms ===")
for a in axioms:
details = {k: _local(v) for k, v in a.items() if k != "type"}
print(f" [{a['type']}] {details}")Step 6 – Visualise the assertion graph
Original assertions only – no reasoning applied. Node colour encodes the concept, edge width encodes the weight.
assert amt is not None, "⚠ Upload a TTL file in Step 1 first, then re-run this cell."
show_amt(nodes, edges, concepts, reasoning=False)Step 7 – Toggle reasoning
Inferred edges are rendered as red dashed arrows. Weights are computed via the fuzzy-logic operator declared on each RoleChainAxiom (amt:LukasiewiczLogic, amt:ProductLogic, or amt:GoedelLogic). Try editing the TTL file to switch the logic on a chain and re-running from Step 1 to see how derived weights change.
assert amt is not None, "⚠ Upload a TTL file in Step 1 first, then re-run this cell."
reasoned_edges = do_reasoning(edges, axioms)
inferred = [e for e in reasoned_edges if e.get("inferred")]
print(f"Original edges: {len(edges)} | Inferred: {len(inferred)}")
print("\nInferred edges:")
df_inf = pd.DataFrame([
{"From": _local(e["from"]), "Role": _local(e["role"]),
"To": _local(e["to"]), "Weight": round(e["weight"], 4)}
for e in inferred
])
display.display(df_inf)
show_amt(nodes, edges, concepts, reasoning=True, axioms=axioms)Step 8 – Consistency check
DisjointAxiom and SelfDisjointAxiom express what cannot be simultaneously true in the graph – for example: the same pair of nodes holding under two mutually exclusive roles, or a role being used reflexively when it must not be. Example ontologies such as the Potter Attribution dataset do not declare integrity axioms by default, so the check typically returns consistent. Add one to your TTL file to watch a violation show up here.
assert amt is not None, "⚠ Upload a TTL file in Step 1 first, then re-run this cell."
is_consistent, violations = check_consistency(edges, axioms)
if is_consistent:
print("✓ Graph is consistent – no integrity axioms violated.")
else:
print(f"✗ {len(violations)} violation(s) found:")
for v in violations:
print(f" • {v}")Step 9 – Export
Write the current state (optionally including inferred edges) to Turtle and to Cypher. The Turtle export is round-trippable – you can re-load amt-export.ttl in Step 1 and see the same graph.
assert amt is not None, "⚠ Upload a TTL file in Step 1 first, then re-run this cell."
ttl_output = export_ttl(
nodes, edges, concepts, roles, axioms,
rdf_graph=amt["graph"],
prefix=PREFIX,
with_reasoning=True,
)
Path("amt-export.ttl").write_text(ttl_output, encoding="utf-8")
print("Saved: amt-export.ttl")
print(ttl_output[:1200], "...")assert amt is not None, "⚠ Upload a TTL file in Step 1 first, then re-run this cell."
cypher_output = export_cypher(nodes, edges, axioms, with_reasoning=True)
Path("amt-export.cypher").write_text(cypher_output, encoding="utf-8")
print("Saved: amt-export.cypher")
print(cypher_output)This notebook is part of an Open Educational Resource built with the NFDI4Objects OER Quarto template. The Academic Meta Tool was originally developed by Martin Unold and Florian Thiery at i3mainz (2017) and is now maintained by Allard Mees and Florian Thiery at LEIZA within NFDI4Objects.