Initial commit

tools/oracle/align.py

@@ -0,0 +1,127 @@
+"""Whole-document sentence alignment, in the manner of a patience diff.
+
+Non-normative; see spec/05-hashes.md, "Alignment".
+Each reading is a list of items: {"id", "strict", "lossy", "complex", "container", "context"}.
+"""
+
+import unicodedata
+
+from .hashing import prose_length
+
+MIN_PROSE = 20
+
+
+def _numeric_skeleton(strict):
+    keep = []
+    i, n = 0, len(strict)
+    while i < n:
+        ch = strict[i]
+        if ch in "_^" and i + 1 < n and strict[i + 1] == "{":
+            keep.append(strict[i:i + 2])
+            i += 2
+            continue
+        if unicodedata.category(ch) == "Nd" or ch in "\u2212.,+}":
+            keep.append(ch)
+        i += 1
+    return "".join(keep)
+
+
+def classify(a, b):
+    """`None` when two paired sentences agree; else the suspect kind."""
+    if a["strict"] == b["strict"] and a["complex"] == b["complex"]:
+        return None
+    if _numeric_skeleton(a["strict"]) != _numeric_skeleton(b["strict"]):
+        return "number"
+    if a["complex"] != b["complex"]:
+        return "equation"
+    return "word"
+
+
+def _lis(pairs):
+    """Longest subsequence increasing in B among pairs sorted by A; lexicographically smallest A on ties."""
+    n = len(pairs)
+    if not n:
+        return []
+    best = [1] * n
+    nxt = [-1] * n
+    for i in range(n - 1, -1, -1):
+        for j in range(i + 1, n):
+            if pairs[j][1] > pairs[i][1] and best[j] + 1 > best[i]:
+                best[i] = best[j] + 1
+                nxt[i] = j
+    top = max(best)
+    i = next(k for k in range(n) if best[k] == top)
+    out = []
+    while i != -1:
+        out.append(pairs[i])
+        i = nxt[i]
+    return out
+
+
+def _lcs(A, B, a0, a1, b0, b1):
+    """LCS over align keys of A[a0:a1] and B[b0:b1], as index pairs, with the spec's tie rule."""
+    n, m = a1 - a0, b1 - b0
+    L = [[0] * (m + 1) for _ in range(n + 1)]
+    for i in range(n - 1, -1, -1):
+        for j in range(m - 1, -1, -1):
+            if A[a0 + i]["lossy"] == B[b0 + j]["lossy"]:
+                L[i][j] = L[i + 1][j + 1] + 1
+            else:
+                L[i][j] = max(L[i + 1][j], L[i][j + 1])
+    out = []
+    i = j = 0
+    while i < n and j < m:
+        if A[a0 + i]["lossy"] == B[b0 + j]["lossy"]:
+            out.append((a0 + i, b0 + j))
+            i += 1
+            j += 1
+        elif L[i + 1][j] >= L[i][j + 1]:
+            i += 1
+        else:
+            j += 1
+    return out
+
+
+def align(A, B):
+    """Align reading B to reading A. Returns {"pairs": [...], "suspects": [...]}."""
+    count_a, count_b = {}, {}
+    for x in A:
+        count_a[x["lossy"]] = count_a.get(x["lossy"], 0) + 1
+    for x in B:
+        count_b[x["lossy"]] = count_b.get(x["lossy"], 0) + 1
+    pos_b = {x["lossy"]: j for j, x in enumerate(B)}
+    cands = [
+        (i, pos_b[x["lossy"]]) for i, x in enumerate(A)
+        if count_a[x["lossy"]] == 1 and count_b.get(x["lossy"]) == 1 and prose_length(x["lossy"]) >= MIN_PROSE
+    ]
+    anchors = _lis(cands)
+    pairs = []
+    prev_a, prev_b = 0, 0
+    for ai, bi in anchors + [(len(A), len(B))]:
+        pairs += _lcs(A, B, prev_a, ai, prev_b, bi)
+        if ai < len(A):
+            pairs.append((ai, bi))
+        prev_a, prev_b = ai + 1, bi + 1
+    out_pairs, suspects = [], []
+    for ai, bi in pairs:
+        kind = classify(A[ai], B[bi])
+        out_pairs.append({"a": A[ai]["id"], "b": B[bi]["id"], "agree": kind is None})
+        if kind:
+            suspects.append({"kind": kind, "a": A[ai]["id"], "b": B[bi]["id"]})
+        if A[ai]["context"] != B[bi]["context"]:
+            suspects.append({"kind": "structure", "a": A[ai]["id"], "b": B[bi]["id"]})
+    for (a1, b1), (a2, b2) in zip(pairs, pairs[1:]):
+        if a2 == a1 + 1 and b2 == b1 + 1:
+            same_a = A[a1]["container"] == A[a2]["container"]
+            same_b = B[b1]["container"] == B[b2]["container"]
+            if same_a != same_b:
+                suspects.append({"kind": "join", "a": A[a2]["id"], "b": B[b2]["id"]})
+    paired_a = {a for a, _ in pairs}
+    paired_b = {b for _, b in pairs}
+    for i, x in enumerate(A):
+        if i not in paired_a:
+            suspects.append({"kind": "dropped", "a": x["id"], "b": None})
+    for j, x in enumerate(B):
+        if j not in paired_b:
+            suspects.append({"kind": "added", "a": None, "b": x["id"]})
+    return {"pairs": out_pairs, "suspects": suspects}