repl command /transform

Author: telmomenezes

CHANGELOG.md

@@ -8,7 +8,7 @@
 - readers provide source information.
 - `hyperbase.parsers.badness` module for parser-agnostic combined structural + token-matching validation.
 - built-in REPL setting `check_badness` to render a badness panel after each parse, available regardless of the active parser plugin.
-- more REPL commands: /load, /search, /count and /types.
+- more REPL commands: /load, /search, /count, /types and /transform.
 
 ### Changed
 

src/hyperbase/cli/repl.py

@@ -20,7 +20,7 @@
 from rich.table import Table
 from rich.text import Text
 
-from hyperbase.builders import hedge
+from hyperbase.builders import hedge, split_edge_str
 from hyperbase.constants import EdgeType
 from hyperbase.hyperedge import Atom, Hyperedge
 from hyperbase.parsers import Parser, get_parser, list_parsers
@@ -367,6 +367,13 @@ def __init__(self, parser_name: str, settings: dict[str, Any]) -> None:
                 ),
                 "handler": self.cmd_types,
             },
+            "transform": {
+                "help": (
+                    "Rewrite loaded edges by pattern "
+                    "(e.g. /transform (X/P.{x} (Y/Ta Z)) ((Y/Mx X/P.{s}) Z))"
+                ),
+                "handler": self.cmd_transform,
+            },
         }
 
         self.session = PromptSession(
@@ -978,6 +985,107 @@ def cmd_types(self, args: list) -> bool:
         )
         return False
 
+    def cmd_transform(self, args: list) -> bool:
+        if not args:
+            self.console.print(
+                "[red]Error:[/red] /transform requires two patterns: "
+                "[cyan]/transform <origin> <target>[/cyan]"
+            )
+            return False
+        if not self.edges:
+            self.console.print("[yellow]No edges loaded.[/yellow]")
+            self.console.print(
+                "[dim]Use[/dim] [cyan]/load <path>[/cyan] [dim]first.[/dim]"
+            )
+            return False
+
+        text = " ".join(args)
+        try:
+            tokens = split_edge_str(text)
+        except ValueError as e:
+            self.console.print(f"[red]Error:[/red] {e}")
+            return False
+        if len(tokens) != 2:
+            self.console.print(
+                "[red]Error:[/red] expected exactly two patterns "
+                f"(got {len(tokens)}): [cyan]/transform <origin> <target>[/cyan]"
+            )
+            return False
+
+        try:
+            origin = hedge(tokens[0])
+            target = hedge(tokens[1])
+        except Exception as e:
+            self.console.print(f"[red]Error:[/red] failed to parse pattern: {e}")
+            return False
+        if origin is None or target is None:
+            self.console.print("[red]Error:[/red] could not parse patterns")
+            return False
+
+        try:
+            choice = (
+                self.session.prompt(
+                    "Dry run? [Y/n] (n applies the rewrite to loaded edges): "
+                )
+                .strip()
+                .lower()
+            )
+        except (KeyboardInterrupt, EOFError):
+            self.console.print("[dim](aborted)[/dim]")
+            return False
+        dry_run = choice not in ("n", "no")
+
+        changes: list[tuple[int, Hyperedge, Hyperedge]] = []
+        new_edges: list[Hyperedge] = []
+        try:
+            for idx, edge in enumerate(self.edges):
+                transformed = edge.transform(origin, target)
+                if transformed != edge:
+                    changes.append((idx, edge, transformed))
+                new_edges.append(transformed)
+        except ValueError as e:
+            self.console.print(f"[red]Error:[/red] {e}")
+            return False
+        except Exception as e:
+            self.console.print(f"[red]Error:[/red] transform failed: {e}")
+            return False
+
+        mode = "dry run" if dry_run else "applied"
+        self.console.print(
+            f"[green]{len(changes)}[/green] of [cyan]{len(self.edges)}[/cyan] "
+            f"edge(s) affected ([bold]{mode}[/bold])"
+        )
+        if not dry_run:
+            self.edges = new_edges
+
+        if not changes:
+            return False
+
+        page_size = int(self.settings.get("search_page_size", 10))
+        if page_size < 1:
+            page_size = 10
+
+        self._paginate(
+            changes,
+            lambda n, change: self._render_transform_change(n, *change),
+            page_size,
+        )
+        return False
+
+    def _render_transform_change(
+        self,
+        n: int,
+        idx: int,
+        before: Hyperedge,
+        after: Hyperedge,
+    ) -> None:
+        self.console.print(f"[bold]#{n}[/bold] [dim](edge {idx})[/dim]")
+        self.console.print("  [dim red]before:[/dim red]")
+        self.console.print(self.formatter.format(before))
+        self.console.print("  [dim green]after:[/dim green]")
+        self.console.print(self.formatter.format(after))
+        self.console.print()
+
     def _render_types_row(self, n: int, type_str: str, count: int) -> None:
         color = "white"
         if type_str:

src/hyperbase/transforms.py

@@ -202,10 +202,13 @@ def _transform_impl(
     if recursive and edge.not_atom:
         edge = Hyperedge(tuple(_transform_impl(c, origin, target, True) for c in edge))
 
-    matches = match_pattern(edge, origin)
-    if matches:
-        bindings = matches[0]
-        edge = _instantiate(target, origin, edge, bindings, top=True)
+    origin_vars = _collect_var_names(origin)
+    for bindings in match_pattern(edge, origin):
+        # The matcher may return partial bindings when constraint propagation
+        # fails on some pattern positions but enough vars satisfy its min_vars
+        # threshold. For a rewrite we require every origin variable bound.
+        if origin_vars.issubset(bindings.keys()):
+            return _instantiate(target, origin, edge, bindings, top=True)
     return edge
 
 

tests/test_hyperedge.py

@@ -1227,6 +1227,17 @@ def test_transform_bare_variable_substitution(self):
         result = edge.transform(hedge("(eats/Pd.so X Y)"), hedge("Y"))
         assert str(result) == "apples/C"
 
+    def test_transform_partial_match_unchanged(self):
+        # The matcher returns a partial binding {X: ...} (no Y, Z) when the
+        # x-position arg cannot match (Y/Ta Z). transform must treat this as
+        # no-match rather than blowing up on the unbound Y / Z.
+        edge = hedge("((had/Mm done/Pd.ox) (the/Md task) by/Ta)")
+        result = edge.transform(
+            hedge("(X/P.{x} (Y/Ta Z))"),
+            hedge("((Y/Mx X/P.{s}) Z)"),
+        )
+        assert result == edge
+
 
 if __name__ == "__main__":
     unittest.main()