lcs.py
file-level
1
files
1
commits
0
hotspots
0
🧊 dead
0
💥 blast risk
| 1 | """LCS / Myers shortest-edit-script algorithm for ordered sequences. |
| 2 | |
| 3 | Operates on ``list[str]`` where each string is a content ID (SHA-256 or |
| 4 | deterministic hash). Two elements are considered identical iff their content |
| 5 | IDs are equal — the algorithm never inspects actual content. |
| 6 | |
| 7 | Public API |
| 8 | ---------- |
| 9 | - :func:`myers_ses` — compute shortest edit script (keep / insert / delete). |
| 10 | - :func:`detect_moves` — post-process insert+delete pairs into ``MoveOp``\\s. |
| 11 | - :func:`diff` — end-to-end: list[str] × list[str] → ``StructuredDelta``. |
| 12 | |
| 13 | Algorithm |
| 14 | --------- |
| 15 | ``myers_ses`` uses the classic O(nm) LCS dynamic-programming traceback. This |
| 16 | is the same algorithm as ``midi_diff.lcs_edit_script`` but operates on content |
| 17 | IDs (strings) rather than ``NoteKey`` dicts, making it fully generic. |
| 18 | |
| 19 | The patience-diff and O(nd) Myers variants (see ``SequenceSchema.diff_algorithm``) |
| 20 | are not yet implemented; both fall back to the O(nm) LCS. |
| 21 | as an optimisation without changing the public API. |
| 22 | """ |
| 23 | |
| 24 | from __future__ import annotations |
| 25 | |
| 26 | import logging |
| 27 | from dataclasses import dataclass |
| 28 | from typing import Literal |
| 29 | |
| 30 | from muse.core.schema import SequenceSchema |
| 31 | from muse.domain import DeleteOp, DomainOp, InsertOp, MoveOp, StructuredDelta |
| 32 | |
| 33 | |
| 34 | type _DeleteMap = dict[str, "DeleteOp"] |
| 35 | logger = logging.getLogger(__name__) |
| 36 | |
| 37 | EditKind = Literal["keep", "insert", "delete"] |
| 38 | |
| 39 | |
| 40 | @dataclass(frozen=True) |
| 41 | class EditStep: |
| 42 | """One step in the shortest edit script produced by :func:`myers_ses`.""" |
| 43 | |
| 44 | kind: EditKind |
| 45 | base_index: int # index in the base content-ID list |
| 46 | target_index: int # index in the target content-ID list |
| 47 | item: str # content ID of the element |
| 48 | |
| 49 | |
| 50 | # --------------------------------------------------------------------------- |
| 51 | # Core algorithm |
| 52 | # --------------------------------------------------------------------------- |
| 53 | |
| 54 | |
| 55 | def myers_ses(base: list[str], target: list[str]) -> list[EditStep]: |
| 56 | """Compute the shortest edit script transforming *base* into *target*. |
| 57 | |
| 58 | Uses the O(nm) LCS dynamic-programming table followed by a linear-time |
| 59 | traceback. Two elements are equal iff their content IDs match. |
| 60 | |
| 61 | Args: |
| 62 | base: Ordered list of content IDs for the base sequence. |
| 63 | target: Ordered list of content IDs for the target sequence. |
| 64 | |
| 65 | Returns: |
| 66 | A list of :class:`EditStep` entries (keep / insert / delete) that |
| 67 | transforms *base* into *target*. The number of "keep" steps equals |
| 68 | the LCS length; insert + delete steps are minimal. |
| 69 | """ |
| 70 | n, m = len(base), len(target) |
| 71 | |
| 72 | # dp[i][j] = length of LCS of base[i:] and target[j:] |
| 73 | dp: list[list[int]] = [[0] * (m + 1) for _ in range(n + 1)] |
| 74 | for i in range(n - 1, -1, -1): |
| 75 | for j in range(m - 1, -1, -1): |
| 76 | if base[i] == target[j]: |
| 77 | dp[i][j] = dp[i + 1][j + 1] + 1 |
| 78 | else: |
| 79 | dp[i][j] = max(dp[i + 1][j], dp[i][j + 1]) |
| 80 | |
| 81 | steps: list[EditStep] = [] |
| 82 | i, j = 0, 0 |
| 83 | while i < n or j < m: |
| 84 | if i < n and j < m and base[i] == target[j]: |
| 85 | steps.append(EditStep("keep", i, j, base[i])) |
| 86 | i += 1 |
| 87 | j += 1 |
| 88 | elif j < m and (i >= n or dp[i][j + 1] >= dp[i + 1][j]): |
| 89 | steps.append(EditStep("insert", i, j, target[j])) |
| 90 | j += 1 |
| 91 | else: |
| 92 | steps.append(EditStep("delete", i, j, base[i])) |
| 93 | i += 1 |
| 94 | |
| 95 | return steps |
| 96 | |
| 97 | |
| 98 | # --------------------------------------------------------------------------- |
| 99 | # Move detection post-pass |
| 100 | # --------------------------------------------------------------------------- |
| 101 | |
| 102 | |
| 103 | def detect_moves( |
| 104 | inserts: list[InsertOp], |
| 105 | deletes: list[DeleteOp], |
| 106 | ) -> tuple[list[MoveOp], list[InsertOp], list[DeleteOp]]: |
| 107 | """Collapse (delete, insert) pairs that share a content ID into ``MoveOp``\\s. |
| 108 | |
| 109 | A move is defined as a delete and an insert of the same content (same |
| 110 | ``content_id``) at different positions. Where the positions are the same, |
| 111 | the pair is left as separate insert/delete ops (idempotent round-trip). |
| 112 | |
| 113 | **Duplicate-content invariant:** when the input sequence contains multiple |
| 114 | elements with identical content (same ``content_id``), each delete is an |
| 115 | independent object. Only the *specific* ``DeleteOp`` object that was |
| 116 | paired into a ``MoveOp`` is removed from ``remaining_deletes``; any other |
| 117 | deletes with the same ``content_id`` are kept as true deletes. This |
| 118 | preserves the round-trip invariant ``apply(diff(A, B), A) == B`` when ``A`` |
| 119 | contains duplicate sections (same heading level, title, and body). |
| 120 | |
| 121 | The matching strategy is first-come, first-served per content ID: the first |
| 122 | delete for a given ``content_id`` is paired with the first insert for that |
| 123 | ``content_id`` when positions differ. |
| 124 | |
| 125 | Args: |
| 126 | inserts: ``InsertOp`` entries from the LCS edit script. |
| 127 | deletes: ``DeleteOp`` entries from the LCS edit script. |
| 128 | |
| 129 | Returns: |
| 130 | A tuple ``(moves, remaining_inserts, remaining_deletes)`` where |
| 131 | ``moves`` contains the detected ``MoveOp``\\s and the remaining lists |
| 132 | hold ops that could not be paired. |
| 133 | """ |
| 134 | # Map content_id → first unpaired DeleteOp for that content. Only the |
| 135 | # *first* delete per content_id is eligible for move-pairing; later ones |
| 136 | # with the same content_id remain as true deletes (see docstring). |
| 137 | delete_by_content: _DeleteMap = {} |
| 138 | for d in deletes: |
| 139 | if d["content_id"] not in delete_by_content: |
| 140 | delete_by_content[d["content_id"]] = d |
| 141 | |
| 142 | moves: list[MoveOp] = [] |
| 143 | remaining_inserts: list[InsertOp] = [] |
| 144 | # Track the object identity (id()) of each DeleteOp that was paired into a |
| 145 | # MoveOp. Using content_id here would incorrectly exclude sibling deletes |
| 146 | # that share the same content_id (duplicate sections) but are true deletes. |
| 147 | paired_delete_ids: set[int] = set() |
| 148 | |
| 149 | for ins in inserts: |
| 150 | cid = ins["content_id"] |
| 151 | if cid in delete_by_content and id(delete_by_content[cid]) not in paired_delete_ids: |
| 152 | d = delete_by_content[cid] |
| 153 | from_pos = d["position"] if d["position"] is not None else 0 |
| 154 | to_pos = ins["position"] if ins["position"] is not None else 0 |
| 155 | if from_pos != to_pos: |
| 156 | moves.append( |
| 157 | MoveOp( |
| 158 | op="move", |
| 159 | address=ins["address"], |
| 160 | from_position=from_pos, |
| 161 | to_position=to_pos, |
| 162 | content_id=cid, |
| 163 | ) |
| 164 | ) |
| 165 | paired_delete_ids.add(id(d)) |
| 166 | continue |
| 167 | remaining_inserts.append(ins) |
| 168 | |
| 169 | # Retain all deletes except the specific objects that became MoveOps. |
| 170 | remaining_deletes = [d for d in deletes if id(d) not in paired_delete_ids] |
| 171 | return moves, remaining_inserts, remaining_deletes |
| 172 | |
| 173 | |
| 174 | # --------------------------------------------------------------------------- |
| 175 | # Top-level diff entry point |
| 176 | # --------------------------------------------------------------------------- |
| 177 | |
| 178 | |
| 179 | def diff( |
| 180 | schema: SequenceSchema, |
| 181 | base: list[str], |
| 182 | target: list[str], |
| 183 | *, |
| 184 | domain: str, |
| 185 | address: str = "", |
| 186 | ) -> StructuredDelta: |
| 187 | """Diff two ordered sequences of content IDs, returning a ``StructuredDelta``. |
| 188 | |
| 189 | Runs :func:`myers_ses`, then :func:`detect_moves` to collapse paired |
| 190 | insert/delete entries into ``MoveOp``\\s. The resulting ``ops`` list |
| 191 | contains ``DeleteOp``, ``InsertOp``, and ``MoveOp`` entries. |
| 192 | |
| 193 | Args: |
| 194 | schema: The ``SequenceSchema`` declaring element type and identity. |
| 195 | base: Base (ancestor) sequence as a list of content IDs. |
| 196 | target: Target (newer) sequence as a list of content IDs. |
| 197 | domain: Domain tag for the returned ``StructuredDelta``. |
| 198 | address: Address prefix for generated op entries (e.g. file path). |
| 199 | |
| 200 | Returns: |
| 201 | A ``StructuredDelta`` with a human-readable ``summary`` and typed ops. |
| 202 | """ |
| 203 | steps = myers_ses(base, target) |
| 204 | |
| 205 | raw_inserts: list[InsertOp] = [] |
| 206 | raw_deletes: list[DeleteOp] = [] |
| 207 | elem = schema["element_type"] |
| 208 | |
| 209 | for step in steps: |
| 210 | if step.kind == "insert": |
| 211 | raw_inserts.append( |
| 212 | InsertOp( |
| 213 | op="insert", |
| 214 | address=address, |
| 215 | position=step.target_index, |
| 216 | content_id=step.item, |
| 217 | content_summary=f"{elem}:{step.item[:8]}", |
| 218 | ) |
| 219 | ) |
| 220 | elif step.kind == "delete": |
| 221 | raw_deletes.append( |
| 222 | DeleteOp( |
| 223 | op="delete", |
| 224 | address=address, |
| 225 | position=step.base_index, |
| 226 | content_id=step.item, |
| 227 | content_summary=f"{elem}:{step.item[:8]}", |
| 228 | ) |
| 229 | ) |
| 230 | |
| 231 | moves, remaining_inserts, remaining_deletes = detect_moves(raw_inserts, raw_deletes) |
| 232 | ops: list[DomainOp] = [*remaining_deletes, *remaining_inserts, *moves] |
| 233 | |
| 234 | n_ins = len(remaining_inserts) |
| 235 | n_del = len(remaining_deletes) |
| 236 | n_mov = len(moves) |
| 237 | |
| 238 | parts: list[str] = [] |
| 239 | if n_ins: |
| 240 | parts.append(f"{n_ins} {elem}{'s' if n_ins != 1 else ''} added") |
| 241 | if n_del: |
| 242 | parts.append(f"{n_del} {elem}{'s' if n_del != 1 else ''} removed") |
| 243 | if n_mov: |
| 244 | parts.append(f"{n_mov} {'moved' if n_mov != 1 else 'moved'}") |
| 245 | summary = ", ".join(parts) if parts else f"no {elem} changes" |
| 246 | |
| 247 | logger.debug( |
| 248 | "lcs.diff %r: +%d -%d ~%d ops on %d→%d elements", |
| 249 | address, |
| 250 | n_ins, |
| 251 | n_del, |
| 252 | n_mov, |
| 253 | len(base), |
| 254 | len(target), |
| 255 | ) |
| 256 | |
| 257 | return StructuredDelta(domain=domain, ops=ops, summary=summary) |