Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
54 changes: 40 additions & 14 deletions graphtage/graphtage.py
Original file line number Diff line number Diff line change
Expand Up @@ -8,8 +8,16 @@
from typing import Any, Generic, TypeVar

from .bounds import Range
from .edits import AbstractCompoundEdit, AbstractEdit, EditCollection, Insert, Match, Remove, Replace
from .levenshtein import EditDistance, levenshtein_distance
from .edits import (
AbstractCompoundEdit,
ConstantCostEdit,
EditCollection,
Insert,
Match,
Remove,
Replace,
)
from .levenshtein import EditDistance, exact_string_distance
from .multiset import MultiSetEdit
from .printer import NULL_PRINTER, Back, Fore, NullANSIContext, Printer
from .sequences import FixedLengthSequenceEdit, SequenceEdit, SequenceNode
Expand Down Expand Up @@ -73,7 +81,7 @@ def calculate_total_size(self) -> int:

def edits(self, node: TreeNode) -> Edit:
if isinstance(node, LeafNode):
return Match(self, node, levenshtein_distance(str(self.object), str(node.object)))
return Match(self, node, exact_string_distance(str(self.object), str(node.object)))
elif isinstance(node, ContainerNode):
return Replace(self, node)

Expand Down Expand Up @@ -672,30 +680,48 @@ def __iter__(self) -> Iterator[KeyValuePairNode]:
return iter(self._children.values())


class StringEdit(AbstractEdit):
"""An edit returned from a :class:`StringNode`"""
class StringEdit(ConstantCostEdit):
"""An edit returned from a :class:`StringNode`.

__slots__ = ('edit_distance',)
The cost of the edit is the Levenshtein distance between the two strings, which
:func:`graphtage.levenshtein.exact_string_distance` computes arithmetically at construction time. The
character-level lattice that renders the edit is built only when :attr:`StringEdit.edit_distance` is read,
which the formatters do for the edits they actually print. A diff costs many more pairs of strings than it
renders.

"""

__slots__ = ('_edit_distance',)

def __init__(
self,
from_node: 'StringNode',
to_node: 'StringNode'
):
self.edit_distance = string_edit_distance(from_node.object, to_node.object)
self._edit_distance: EditDistance | None = None
super().__init__(
from_node=from_node,
to_node=to_node
to_node=to_node,
cost=exact_string_distance(from_node.object, to_node.object)
)

def __repr__(self) -> str:
return f"{self.__class__.__name__}(from_node={self.from_node!r}, to_node={self.to_node!r})"
@property
def edit_distance(self) -> EditDistance:
"""The character-level edit lattice for this edit, constructed on first access.

def bounds(self) -> Range:
return self.edit_distance.bounds()
The lattice is the only source of the edit script, so the sequence of character matches, insertions, and
removals that a formatter renders is unaffected by the cost having been computed separately.

def tighten_bounds(self) -> bool:
return self.edit_distance.tighten_bounds()
Returns:
EditDistance: The lattice over the characters of the two strings.

"""
if self._edit_distance is None:
self._edit_distance = string_edit_distance(self.from_node.object, self.to_node.object)
return self._edit_distance

def __repr__(self) -> str:
return f"{self.__class__.__name__}(from_node={self.from_node!r}, to_node={self.to_node!r})"

def print(self, formatter: GraphtageFormatter, printer: Printer):
"""`StringEdit` does not implement :meth:`graphtage.tree.Edit.print`.
Expand Down
41 changes: 40 additions & 1 deletion graphtage/levenshtein.py
Original file line number Diff line number Diff line change
Expand Up @@ -30,7 +30,7 @@
log = logging.getLogger(__name__)


def levenshtein_distance(s: str, t: str) -> int:
def levenshtein_distance(s: str | bytes, t: str | bytes) -> int:
"""Canonical implementation of the Levenshtein distance metric.

Args:
Expand Down Expand Up @@ -64,6 +64,45 @@ def levenshtein_distance(s: str, t: str) -> int:
return dist[rows - 1][cols - 1]


def exact_string_distance(s: str | bytes, t: str | bytes) -> int:
"""Computes the Levenshtein distance between two strings without building an edit script.

:class:`EditDistance` computes the same number, but it does so by materializing one
:class:`graphtage.TreeNode` per character and one live :class:`graphtage.Edit` per cell of the Levenshtein
matrix. Callers that only need the cost, such as :class:`graphtage.StringEdit` and
:meth:`graphtage.LeafNode.edits`, go through this function instead and leave the lattice unbuilt.

Both arguments may be :class:`str` or :class:`bytes`, in either combination. Indexing :class:`bytes` yields
:class:`int` byte values, which never compare equal to a :class:`str` character, so a mixed pair costs one
per aligned position exactly as the lattice charges for it.

Args:
s: the string from which to match.
t: the string to which to match.

Returns:
int: The Levenshtein edit distance metric between the two strings.

"""
if s == t:
return 0
elif not s:
return len(t)
elif not t:
return len(s)
# Stripping a shared prefix and suffix quadratically reduces the size of the matrix that
# levenshtein_distance builds. It cannot change the answer: characters that align with themselves are
# always free, and no optimal alignment crosses them.
overlap = min(len(s), len(t))
prefix = 0
while prefix < overlap and s[prefix] == t[prefix]:
prefix += 1
suffix = 0
while suffix < overlap - prefix and s[len(s) - suffix - 1] == t[len(t) - suffix - 1]:
suffix += 1
return levenshtein_distance(s[prefix:len(s) - suffix], t[prefix:len(t) - suffix])


class EditDistance(SequenceEdit):
"""An edit that computes the minimum sequence of sub-edits necessary to transform one node to another.

Expand Down
130 changes: 130 additions & 0 deletions test/test_graphtage.py
Original file line number Diff line number Diff line change
Expand Up @@ -260,3 +260,133 @@ def test_rendering_closes_the_quote(self):

def test_rendering_escapes_unprintable_bytes(self):
self.assertEqual('b"he~~\\x00~~++\\xff++o"', self.render(b"he\x00o", b"he\xffo"))


class StringEditDistanceCounter:
"""Counts how many character-level lattices :class:`graphtage.StringEdit` builds inside a block."""

def __init__(self):
self.count = 0
self._original = graphtage.graphtage.string_edit_distance

def __enter__(self) -> 'StringEditDistanceCounter':
def counted(s1, s2):
self.count += 1
return self._original(s1, s2)

graphtage.graphtage.string_edit_distance = counted
return self

def __exit__(self, *args):
graphtage.graphtage.string_edit_distance = self._original


class TestStringEdit(TestCase):
"""Covers the two halves of :class:`graphtage.StringEdit`: its cost and its edit script.

The cost is computed arithmetically at construction; the lattice that produces the script is built only when
something reads :attr:`graphtage.StringEdit.edit_distance`. These tests pin that split, because a
:class:`graphtage.StringEdit` that eagerly builds its lattice is correct but quadratically slower, and a
lattice whose cost disagrees with the reported cost renders a diff that does not add up to its own price.

"""

@staticmethod
def edit(from_str: str | bytes, to_str: str | bytes) -> graphtage.StringEdit:
return graphtage.StringEdit(graphtage.StringNode(from_str), graphtage.StringNode(to_str))

def test_bounds_are_definitive_on_construction(self):
"""A caller that never tightens must still see the exact cost.

:class:`graphtage.matching.WeightedBipartiteMatcher` prices its edges with ``bounds().upper_bound``
after only partial tightening, so an edit whose initial upper bound is an over-estimate feeds the
assignment problem a weight that is too large.

"""
edit = self.edit("kitten", "sitting")
self.assertTrue(edit.bounds().definitive())
self.assertEqual(graphtage.Range(3, 3), edit.bounds())
self.assertEqual(graphtage.Range(3, 3), edit.initial_bounds)

def test_tighten_bounds_returns_false(self):
"""There is nothing left to tighten, so the loops that drive edits to completion terminate at once."""
edit = self.edit("kitten", "sitting")
self.assertFalse(edit.tighten_bounds())
self.assertEqual(graphtage.Range(3, 3), edit.bounds())

def test_the_lattice_is_not_built_until_it_is_read(self):
"""Constructing the edit must not build the lattice; reading the property must, exactly once."""
edit = self.edit("/usr/local/bin", "/usr/lib/bin")
self.assertIsNone(edit._edit_distance)
lattice = edit.edit_distance
self.assertIsNotNone(edit._edit_distance)
self.assertIs(lattice, edit.edit_distance)

def test_the_lattice_agrees_with_the_reported_cost(self):
"""The script a formatter renders has to add up to the cost the edit reported to the matcher."""
for from_str, to_str in (
("/usr/local/bin", "/usr/lib/bin"),
("kitten", "sitting"),
("aabc", "bcb"),
("hello", "hellp"),
("", "abc"),
("abc", ""),
("abcd", "abcd"),
(b"he\x00o", b"he\xffo"),
):
with self.subTest(from_str=from_str, to_str=to_str):
edit = self.edit(from_str, to_str)
lattice = edit.edit_distance
while lattice.tighten_bounds():
pass
self.assertEqual(edit.bounds().upper_bound, lattice.bounds().upper_bound)

def test_printing_a_string_node_builds_no_lattice(self):
"""``print_StringNode`` constructs two ``StringEdit``s per node only to ask whether it is quoted."""
out_stream = StringIO()
with StringEditDistanceCounter() as counter:
graphtage.StringNode("/usr/local/bin").print(Printer(ansi_color=False, out_stream=out_stream))
self.assertEqual(0, counter.count)
self.assertEqual('"/usr/local/bin"', out_stream.getvalue())

def test_a_diff_builds_a_lattice_only_for_what_it_renders(self):
"""A dense N-by-N match costs N*N pairs of key/value nodes but renders at most N of them.

No key survives, so the bipartite matcher prices every one of the 64 candidate pairs, each of which
holds a key string and a value string. Eagerly building a lattice per candidate is what made a diff of
a few dozen strings take seconds. The bound here is deliberately loose; it only has to stay far below
the 128 lattices that the candidates would otherwise account for.

"""
size = 8
from_obj = {f"from-key-{i}": f"/usr/local/share/value-{i}" for i in range(size)}
to_obj = {f"to-key-{i}": f"/opt/local/share/value-{i * 3}" for i in range(size)}
out_stream = StringIO()
with StringEditDistanceCounter() as counter:
diffed = graphtage.json.build_tree(from_obj).diff(graphtage.json.build_tree(to_obj))
graphtage.json.JSONFormatter.DEFAULT_INSTANCE.print(
Printer(ansi_color=False, quiet=True, out_stream=out_stream), diffed
)
self.assertLessEqual(counter.count, 4 * size)

def test_a_sixty_key_dict_diff_is_fast(self):
"""Guards the whole point of computing string edit costs without the lattice.

Every key and every value differs, so the bipartite matcher costs 3600 pairs of key/value nodes, and
therefore 7200 pairs of strings. Building a lattice for each of those takes the better part of a
minute; deriving the cost arithmetically takes under a second. The limit is generous so that a slow
runner cannot flake it.

"""
rng = random.Random(60)

def words(count: int, length: int) -> list[str]:
return [''.join(rng.choices('abcdefghijklmnopqrstuvwxyz', k=length)) for _ in range(count)]

from_obj = dict(zip(words(60, 12), words(60, 40), strict=True))
to_obj = dict(zip(words(60, 12), words(60, 40), strict=True))
with run_with_time_limit(seconds=30):
edit = graphtage.json.build_tree(from_obj).edits(graphtage.json.build_tree(to_obj))
while edit.tighten_bounds():
pass
self.assertTrue(edit.bounds().definitive())
73 changes: 72 additions & 1 deletion test/test_levenshtein.py
Original file line number Diff line number Diff line change
Expand Up @@ -5,10 +5,11 @@

from graphtage import EditDistance, string_edit_distance
from graphtage.edits import Edit, Insert, Match, Remove
from graphtage.levenshtein import levenshtein_distance
from graphtage.levenshtein import exact_string_distance, levenshtein_distance

LETTERS: str = 'abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ'
SMALL_ALPHABET: str = 'abcd'
NON_ASCII: str = 'αβγδεζηθ🙂é́'


def render_script(distance: EditDistance) -> list[str]:
Expand Down Expand Up @@ -281,3 +282,73 @@ def test_edit_script_realizes_the_reported_cost(self):
bounds = distance.bounds()
self.assertTrue(bounds.definitive(), f"{pair} has bounds {bounds!s}")
self.assertEqual(cost, bounds.upper_bound, pair)


class TestExactStringDistance(TestCase):
"""Covers :func:`graphtage.levenshtein.exact_string_distance`.

The function short-circuits equal and empty operands and strips a shared prefix and suffix before calling
:func:`graphtage.levenshtein.levenshtein_distance`. Every one of those steps is an opportunity to return a
number that is not the Levenshtein distance, and the number it returns is the cost of a
:class:`graphtage.StringEdit`, so a wrong answer silently changes which nodes a diff matches.

"""

def assert_agrees(self, s, t):
expected = levenshtein_distance(s, t)
pair = f"{s!r} -> {t!r}"
self.assertEqual(expected, exact_string_distance(s, t), pair)
self.assertEqual(expected, exact_string_distance(t, s), f"{t!r} -> {s!r}")

def test_agrees_on_random_strings(self):
"""A small alphabet maximizes the number of pairs with a shared prefix or suffix to strip."""
for _ in trange(500):
s = ''.join(random.choices(SMALL_ALPHABET, k=random.randint(0, 12)))
t = ''.join(random.choices(SMALL_ALPHABET, k=random.randint(0, 12)))
self.assert_agrees(s, t)

def test_agrees_on_random_bytes(self):
""":class:`StringNode` wraps ``bytes`` as well as ``str``, and indexing ``bytes`` yields ``int``."""
for _ in trange(500):
s = bytes(random.choices(range(0, 8), k=random.randint(0, 12)))
t = bytes(random.choices(range(0, 8), k=random.randint(0, 12)))
self.assert_agrees(s, t)

def test_agrees_on_random_non_ascii(self):
"""Astral characters and combining marks are single positions to both implementations."""
for _ in trange(200):
s = ''.join(random.choices(NON_ASCII, k=random.randint(0, 10)))
t = ''.join(random.choices(NON_ASCII, k=random.randint(0, 10)))
self.assert_agrees(s, t)

def test_empty_operands(self):
self.assertEqual(0, exact_string_distance('', ''))
self.assertEqual(0, exact_string_distance(b'', b''))
self.assertEqual(3, exact_string_distance('', 'abc'))
self.assertEqual(3, exact_string_distance('abc', ''))
self.assertEqual(3, exact_string_distance(b'', b'abc'))
self.assertEqual(3, exact_string_distance(b'abc', b''))

def test_shared_prefix_and_suffix_do_not_overlap(self):
"""An operand that is wholly a prefix of the other must not have its characters counted twice."""
self.assertEqual(1, exact_string_distance('aa', 'aaa'))
self.assertEqual(2, exact_string_distance('aaa', 'aaaaa'))
self.assertEqual(4, exact_string_distance('a', 'aaaaa'))
self.assertEqual(1, exact_string_distance('ab', 'aab'))
self.assertEqual(0, exact_string_distance('aaaa', 'aaaa'))

def test_str_never_equals_bytes(self):
"""A mixed pair costs one per aligned position, which is what the character lattice charges."""
self.assertEqual(5, exact_string_distance('hello', b'hello'))
self.assertEqual(5, exact_string_distance(b'hello', 'hello'))
self.assertEqual(1, exact_string_distance('a', b'a'))

def test_agrees_with_the_lattice(self):
"""The lattice is what renders the edit, so the reported cost has to be the cost of its script."""
for _ in trange(100):
s = ''.join(random.choices(SMALL_ALPHABET, k=random.randint(0, 10)))
t = ''.join(random.choices(SMALL_ALPHABET, k=random.randint(0, 10)))
lattice = string_edit_distance(s, t)
while lattice.tighten_bounds():
pass
self.assertEqual(lattice.bounds().upper_bound, exact_string_distance(s, t), f"{s!r} -> {t!r}")
Loading