#!/usr/bin/env python3 """Find individuals that ELK will place in owl:Nothing because their asserted types fall under two disjoint classes. Reads an N-Triples dump (raw/all.ttl from vfb-pipeline-dumps is N-Triples despite the extension). No reasoner, no Java, bounded memory. vfb_find_disjoint_clashes.py [mini.nt] The optional third argument writes a small ontology (TBox subclass/disjoint axioms plus only the clashing individuals' class assertions) that reproduces the inconsistency, so `robot explain -M inconsistency` runs in seconds instead of hours. """ from __future__ import print_function import os import subprocess import sys import tempfile from collections import defaultdict, deque RDF = "http://www.w3.org/1999/02/22-rdf-syntax-ns#" RDFS = "http://www.w3.org/2000/01/rdf-schema#" OWL = "http://www.w3.org/2002/07/owl#" TYPE = "<" + RDF + "type>" FIRST = "<" + RDF + "first>" REST = "<" + RDF + "rest>" NIL = "<" + RDF + "nil>" SUBCLASS = "<" + RDFS + "subClassOf>" LABEL = "<" + RDFS + "label>" DISJOINT = "<" + OWL + "disjointWith>" ALLDISJOINT = "<" + OWL + "AllDisjointClasses>" MEMBERS = "<" + OWL + "members>" DISJOINT_UNION = "<" + OWL + "disjointUnionOf>" OWL_CLASS = "<" + OWL + "Class>" NAMED_INDIVIDUAL = "<" + OWL + "NamedIndividual>" def parse(line): """Split one N-Triples line into (subject, predicate, object).""" line = line.rstrip("\n") if not line.endswith(" ."): return None body = line[:-2] if body.startswith("<"): i = body.find("> ") if i < 0: return None subj, body = body[: i + 1], body[i + 2 :] else: i = body.find(" ") if i < 0: return None subj, body = body[:i], body[i + 1 :] i = body.find("> ") if i < 0: return None return subj, body[: i + 1], body[i + 2 :] def short(iri): return iri.strip("<>").rsplit("/", 1)[-1] def main(): if len(sys.argv) < 3: sys.exit(__doc__) src, out_tsv = sys.argv[1], sys.argv[2] mini_nt = sys.argv[3] if len(sys.argv) > 3 else None workdir = os.path.dirname(os.path.abspath(out_tsv)) or "." # ---- pass 1: TBox ---- sys.stderr.write("pass 1: reading class axioms\n") parents = defaultdict(set) disjoint_direct = [] first, rest, members = {}, {}, {} alldisjoint = set() disjoint_union = [] declared_classes = set() total = 0 with open(src, "r", errors="replace") as fh: for line in fh: total += 1 if total % 20000000 == 0: sys.stderr.write(" %s lines\n" % format(total, ",")) sys.stderr.flush() if SUBCLASS in line: t = parse(line) if t and t[1] == SUBCLASS and t[0].startswith("<") and t[2].startswith("<"): parents[t[0]].add(t[2]) elif DISJOINT in line: t = parse(line) if t and t[1] == DISJOINT: disjoint_direct.append((t[0], t[2])) elif DISJOINT_UNION in line: t = parse(line) if t and t[1] == DISJOINT_UNION: disjoint_union.append(t[2]) elif MEMBERS in line: t = parse(line) if t and t[1] == MEMBERS: members[t[0]] = t[2] elif FIRST in line or REST in line: t = parse(line) if t and t[1] == FIRST: first[t[0]] = t[2] elif t and t[1] == REST: rest[t[0]] = t[2] elif TYPE in line: t = parse(line) if t and t[1] == TYPE: if t[2] == OWL_CLASS: declared_classes.add(t[0]) elif t[2] == ALLDISJOINT: alldisjoint.add(t[0]) def walk(head): out, seen = [], set() while head and head != NIL and head not in seen: seen.add(head) if head in first: out.append(first[head]) head = rest.get(head) return out pairs = set() for a, b in disjoint_direct: if a.startswith("<") and b.startswith("<"): pairs.add((a, b) if a < b else (b, a)) for head in [members.get(n) for n in alldisjoint] + disjoint_union: ms = [m for m in walk(head) if m.startswith("<")] for i in range(len(ms)): for j in range(i + 1, len(ms)): a, b = ms[i], ms[j] pairs.add((a, b) if a < b else (b, a)) anchors = sorted(set(c for pr in pairs for c in pr)) idx = dict((c, i) for i, c in enumerate(anchors)) sys.stderr.write(" %s triples, %s named disjoint pairs over %s classes\n" % (format(total, ","), format(len(pairs), ","), format(len(anchors), ","))) # ---- ancestor bitmasks ---- mask_cache = {} def mask(c, stack=None): if c in mask_cache: return mask_cache[c] if stack is None: stack = set() if c in stack: return 0 stack.add(c) m = 1 << idx[c] if c in idx else 0 for p in parents.get(c, ()): m |= mask(p, stack) stack.discard(c) mask_cache[c] = m return m sys.setrecursionlimit(100000) for c in list(parents): mask(c) for c in anchors: mask(c) anchored = dict((c, m) for c, m in mask_cache.items() if m) sys.stderr.write(" %s classes sit under a disjointness anchor\n" % format(len(anchored), ",")) pair_bits = dict(((1 << idx[a]) | (1 << idx[b]), (a, b)) for a, b in pairs) # ---- pass 2: class assertions + labels ---- sys.stderr.write("pass 2: streaming class assertions\n") labels = {} tmp = tempfile.NamedTemporaryFile(mode="w", suffix=".tsv", dir=workdir, delete=False, prefix="clash_") with open(src, "r", errors="replace") as fh: for line in fh: if TYPE in line: t = parse(line) if t and t[1] == TYPE and t[2] in anchored: tmp.write(t[0] + "\t" + t[2] + "\n") elif LABEL in line: t = parse(line) if t and t[1] == LABEL and (t[0] in anchored or t[0] in idx): if t[0] not in labels: try: labels[t[0]] = t[2].split('"')[1] except IndexError: pass tmp.close() sorted_path = tmp.name + ".sorted" sys.stderr.write("sorting\n") env = dict(os.environ, LC_ALL="C") subprocess.check_call(["sort", "-S", "2G", "-T", workdir, "-o", sorted_path, tmp.name], env=env) os.unlink(tmp.name) def lab(c): return "%s (%s)" % (short(c), labels.get(c, "?")) # ---- aggregate ---- clashes = {} n_clash = 0 signature = defaultdict(int) def flush(subj, types): acc = 0 for c in types: acc |= anchored[c] for bits, (a, b) in pair_bits.items(): if acc & bits == bits: left = sorted(c for c in types if anchored[c] & (1 << idx[a])) right = sorted(c for c in types if anchored[c] & (1 << idx[b])) clashes[subj] = (a, b, left, right) signature[(a, b, tuple(right))] += 1 return True return False cur, bag = None, [] with open(sorted_path) as fh: for line in fh: s, c = line.rstrip("\n").split("\t") if s != cur: if cur is not None and flush(cur, bag): n_clash += 1 cur, bag = s, [c] else: bag.append(c) if cur is not None and flush(cur, bag): n_clash += 1 os.unlink(sorted_path) with open(out_tsv, "w") as fh: fh.write("individual\tdisjoint_a\tdisjoint_b\ttypes_under_a\ttypes_under_b\n") for subj in sorted(clashes): a, b, left, right = clashes[subj] fh.write("\t".join([short(subj), short(a), short(b), ";".join(short(x) for x in left), ";".join(short(x) for x in right)]) + "\n") print("") print("CLASHING INDIVIDUALS: %s" % format(n_clash, ",")) print("") for (a, b, right), n in sorted(signature.items(), key=lambda kv: -kv[1])[:25]: print("%8s %s clashes with %s" % (format(n, ","), ", ".join(lab(x) for x in right), lab(a))) print("") print("wrote %s" % out_tsv) # ---- optional reproducer ---- if mini_nt: sys.stderr.write("pass 3: writing %s\n" % mini_nt) want = set(clashes) seen = set() with open(mini_nt, "w") as out, open(src, "r", errors="replace") as fh: for line in fh: keep = False if SUBCLASS in line or DISJOINT in line: t = parse(line) keep = bool(t and t[0].startswith("<") and t[2].startswith("<")) elif TYPE in line: t = parse(line) if t and t[1] == TYPE: keep = t[2] == OWL_CLASS or (t[0] in want and t[2].startswith("<")) if keep and line not in seen: seen.add(line) out.write(line) print("wrote %s (feed this to robot explain)" % mini_nt) if __name__ == "__main__": main()