Hyperedge.distance()

Author: telmomenezes

CHANGELOG.md

@@ -5,6 +5,7 @@
 ### Added
 
 - `Hyperedge.transform()`: pattern-based rewrites.
+- `Hyperedge.distance()`: distance metric based on tree edit distance.
 - per-atom `tok_pos`/`text_span` and per-root `Hyperedge.tokens` source-position metadata.
 - continuity-aware sub-edge text derivation with verbatim character-offset slicing.
 - `transforms.tok_pos_tree(edge)` rebuilds the parallel `tok_pos` tree from in-memory atoms.

src/hyperbase/__init__.py

@@ -1,8 +1,10 @@
 from hyperbase.builders import hedge
+from hyperbase.distance import edge_distance
 from hyperbase.loaders import load_edges
 from hyperbase.parsers import get_parser
 
 __all__ = [
+    "edge_distance",
     "get_parser",
     "hedge",
     "load_edges",

src/hyperbase/distance.py

@@ -0,0 +1,174 @@
+"""Distance measure between two hyperedges.
+
+The distance is an ordered *tree edit distance* (Zhang-Shasha): the minimum
+cost of a sequence of node relabel / insert / delete operations that turns one
+edge-tree into the other. It accounts for both the nesting structure and the
+atom content of the edges, and is polynomial in the size of the edges, which
+makes it tractable for the (small) edges produced by sentence parsing.
+
+The raw cost is a true metric. By default it is normalised to ``[0, 1]`` by
+dividing by the combined node count of both edges, so ``0`` means identical and
+larger values mean more different.
+"""
+
+from __future__ import annotations
+
+from typing import cast
+
+from hyperbase.hyperedge import Atom, Hyperedge
+
+
+def _annotate(root: Hyperedge) -> tuple[list[Hyperedge], list[int], list[int]]:
+    """Post-order annotation of a tree for the Zhang-Shasha algorithm.
+
+    Returns ``(nodes, lmld, keyroots)`` where ``nodes`` lists every node of the
+    tree in post-order, ``lmld[i]`` is the post-order index of the leftmost-leaf
+    descendant of node ``i``, and ``keyroots`` lists the post-order indices of
+    the key roots (the root plus every node that has a left sibling).
+    """
+    nodes: list[Hyperedge] = []
+    lmld: list[int] = []
+
+    def visit(edge: Hyperedge) -> int:
+        if edge.atom or len(edge) == 0:
+            idx = len(nodes)
+            nodes.append(edge)
+            lmld.append(idx)
+            return idx
+        first_leaf = -1
+        for pos, child in enumerate(edge):
+            child_idx = visit(child)
+            if pos == 0:
+                first_leaf = lmld[child_idx]
+        idx = len(nodes)
+        nodes.append(edge)
+        lmld.append(first_leaf)
+        return idx
+
+    visit(root)
+
+    # The key root for a given leftmost leaf is the highest-indexed node that
+    # shares it; iterating in order and overwriting yields exactly that.
+    keyroot_for: dict[int, int] = {}
+    for i, leaf in enumerate(lmld):
+        keyroot_for[leaf] = i
+    keyroots = sorted(keyroot_for.values())
+    return nodes, lmld, keyroots
+
+
+def _safe_normalise(edge: Hyperedge) -> Hyperedge:
+    """Normalise argument roles, falling back to the edge on malformed input."""
+    from hyperbase.transforms import normalise
+
+    try:
+        return normalise(edge)
+    except (KeyError, RuntimeError):
+        return edge
+
+
+def _internal_label(edge: Hyperedge) -> str:
+    """Label used to compare two non-atom nodes (their inferred main type)."""
+    try:
+        return edge.mtype()
+    except RuntimeError:
+        return "?"
+
+
+def edge_distance(
+    edge1: Hyperedge,
+    edge2: Hyperedge,
+    *,
+    normalize: bool = True,
+    canonical: bool = True,
+    root_weight: float = 0.5,
+    type_weight: float = 0.5,
+) -> float:
+    """Tree edit distance between two hyperedges.
+
+    Keyword arguments:
+    normalize -- divide the raw edit cost by the combined node count, yielding a
+        value in ``[0, 1]`` (default: True). When False, the raw cost is
+        returned.
+    canonical -- normalise the argument-role ordering of both edges first, so
+        that edges differing only by a benign argument reordering are treated as
+        equal (default: True).
+    root_weight -- relative weight of the atom root (the word) when relabelling
+        two atoms (default: 0.5).
+    type_weight -- relative weight of the atom type/role when relabelling two
+        atoms (default: 0.5).
+
+    The atom weights are normalised so that any non-negative pair keeps a single
+    atom relabel cost within ``[0, 1]``; ``type_weight=0`` gives a purely
+    lexical comparison and ``root_weight=0`` a purely structural one.
+    """
+    if canonical:
+        edge1 = _safe_normalise(edge1)
+        edge2 = _safe_normalise(edge2)
+
+    total_weight = root_weight + type_weight
+    if total_weight <= 0:
+        w_root = w_type = 0.5
+    else:
+        w_root = root_weight / total_weight
+        w_type = type_weight / total_weight
+
+    def atom_cost(a: Atom, b: Atom) -> float:
+        root_part = 0.0 if a.root() == b.root() else 1.0
+        if a.type() == b.type():
+            type_part = 0.0
+        elif a.mtype() == b.mtype():
+            type_part = 0.5
+        else:
+            type_part = 1.0
+        return w_root * root_part + w_type * type_part
+
+    def relabel_cost(a: Hyperedge, b: Hyperedge) -> float:
+        if a.atom and b.atom:
+            return atom_cost(cast(Atom, a), cast(Atom, b))
+        if a.not_atom and b.not_atom:
+            return 0.0 if _internal_label(a) == _internal_label(b) else 0.5
+        # one atom and one non-atom: maximal substitution cost
+        return 1.0
+
+    nodes1, lmld1, keyroots1 = _annotate(edge1)
+    nodes2, lmld2, keyroots2 = _annotate(edge2)
+    n1 = len(nodes1)
+    n2 = len(nodes2)
+
+    treedist = [[0.0] * n2 for _ in range(n1)]
+
+    for i in keyroots1:
+        li = lmld1[i]
+        for j in keyroots2:
+            lj = lmld2[j]
+            rows = i - li + 2
+            cols = j - lj + 2
+            fd = [[0.0] * cols for _ in range(rows)]
+            for di in range(1, rows):
+                fd[di][0] = fd[di - 1][0] + 1.0
+            for dj in range(1, cols):
+                fd[0][dj] = fd[0][dj - 1] + 1.0
+            for di in range(1, rows):
+                ni = li + di - 1
+                for dj in range(1, cols):
+                    nj = lj + dj - 1
+                    delete = fd[di - 1][dj] + 1.0
+                    insert = fd[di][dj - 1] + 1.0
+                    if lmld1[ni] == li and lmld2[nj] == lj:
+                        relabel = fd[di - 1][dj - 1] + relabel_cost(
+                            nodes1[ni], nodes2[nj]
+                        )
+                        best = min(delete, insert, relabel)
+                        fd[di][dj] = best
+                        treedist[ni][nj] = best
+                    else:
+                        relabel = fd[lmld1[ni] - li][lmld2[nj] - lj] + treedist[ni][nj]
+                        fd[di][dj] = min(delete, insert, relabel)
+
+    ted = treedist[n1 - 1][n2 - 1]
+    if not normalize:
+        return ted
+    total = n1 + n2
+    if total == 0:
+        return 0.0
+    return ted / total

src/hyperbase/hyperedge.py

@@ -440,6 +440,30 @@ def normalise(self) -> Hyperedge:
 
         return _propagate_root_text(self, normalise(self))
 
+    def distance(
+        self,
+        other: Hyperedge,
+        *,
+        normalize: bool = True,
+        canonical: bool = True,
+        root_weight: float = 0.5,
+        type_weight: float = 0.5,
+    ) -> float:
+        """Tree edit distance between this edge and ``other``.
+
+        See ``hyperbase.distance.edge_distance`` for details.
+        """
+        from hyperbase.distance import edge_distance
+
+        return edge_distance(
+            self,
+            other,
+            normalize=normalize,
+            canonical=canonical,
+            root_weight=root_weight,
+            type_weight=type_weight,
+        )
+
     ############
     # patterns #
     ############

tests/test_distance.py

@@ -0,0 +1,85 @@
+import unittest
+
+from hyperbase import edge_distance
+from hyperbase.builders import hedge
+
+
+class TestDistance(unittest.TestCase):
+    def test_identity_atom(self):
+        assert edge_distance(hedge("berlin/C"), hedge("berlin/C")) == 0.0
+
+    def test_identity_nested(self):
+        edge = hedge("(is/P.so (the/M sky/C) blue/C)")
+        assert edge_distance(edge, edge) == 0.0
+
+    def test_symmetry(self):
+        a = hedge("(is/P.so berlin/C nice/C)")
+        b = hedge("(loves/P.so mary/C art/C)")
+        assert edge_distance(a, b) == edge_distance(b, a)
+
+    def test_bounds(self):
+        pairs = [
+            ("berlin/C", "paris/C"),
+            ("berlin/C", "loves/P"),
+            ("(is/P.so berlin/C nice/C)", "(loves/P.so mary/C art/C)"),
+            ("berlin/C", "(is/P.so berlin/C nice/C)"),
+        ]
+        for s1, s2 in pairs:
+            d = edge_distance(hedge(s1), hedge(s2))
+            assert 0.0 <= d <= 1.0
+
+    def test_graded_atom_cost(self):
+        base = hedge("(is/P.so berlin/C nice/C)")
+        subtype = hedge("(is/P.so berlin/C nice/Cp)")  # same root, subtype only
+        root = hedge("(is/P.so berlin/C ugly/C)")  # different root, same type
+        both = hedge("(is/P.so berlin/C loves/P)")  # different root and type
+        d_subtype = edge_distance(base, subtype)
+        d_root = edge_distance(base, root)
+        d_both = edge_distance(base, both)
+        assert 0.0 < d_subtype < d_root < d_both
+
+    def test_structural_change_costs_more(self):
+        base = hedge("(is/P.so berlin/C nice/C)")
+        relabel = hedge("(is/P.so berlin/C ugly/C)")  # one leaf relabel
+        # extra argument (valid connector with three argroles)
+        extra_arg = hedge("(is/P.soc berlin/C nice/C today/C)")
+        assert edge_distance(base, relabel) < edge_distance(base, extra_arg)
+
+    def test_canonical_invariance(self):
+        # Same relation, arguments and roles permuted (s,o vs o,s).
+        a = hedge("(is/P.so berlin/C nice/C)")
+        b = hedge("(is/P.os nice/C berlin/C)")
+        assert edge_distance(a, b, canonical=True) == 0.0
+        assert edge_distance(a, b, canonical=False) > 0.0
+
+    def test_raw_cost(self):
+        # A single leaf relabel (same type, different root) -> raw cost 0.5.
+        base = hedge("(is/P.so berlin/C nice/C)")
+        relabel = hedge("(is/P.so berlin/C ugly/C)")
+        assert edge_distance(base, relabel, normalize=False) == 0.5
+
+    def test_weight_extremes(self):
+        a = hedge("berlin/C")
+        b = hedge("berlin/P")  # same root, different type
+        # type_weight=0 -> purely lexical: same word -> identical
+        assert edge_distance(a, b, type_weight=0.0, root_weight=1.0) == 0.0
+        # root_weight=0 -> purely structural: different type -> non-zero
+        assert edge_distance(a, b, root_weight=0.0, type_weight=1.0) > 0.0
+
+    def test_method(self):
+        a = hedge("(is/P.so berlin/C nice/C)")
+        b = hedge("(is/P.so berlin/C ugly/C)")
+        assert a.distance(b) == edge_distance(a, b)
+
+    def test_triangle_inequality_raw(self):
+        a = hedge("(is/P.so berlin/C nice/C)")
+        b = hedge("(is/P.so berlin/C ugly/C)")
+        c = hedge("(loves/P.so mary/C art/C)")
+        d_ac = edge_distance(a, c, normalize=False)
+        d_ab = edge_distance(a, b, normalize=False)
+        d_bc = edge_distance(b, c, normalize=False)
+        assert d_ac <= d_ab + d_bc
+
+
+if __name__ == "__main__":
+    unittest.main()