Hyperedge.transform(): pattern-based rewrites

Author: telmomenezes

CHANGELOG.md

@@ -4,14 +4,15 @@
 
 ### Added
 
+- Hyperedge.transform(): pattern-based rewrites.
 - 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.
 
 ### Changed
 
-- repl shows multiple results if available.
+- REPL shows multiple results if available.
 
 ### Removed
 

src/hyperbase/hyperedge.py

@@ -227,6 +227,17 @@ def simplify(self, subtypes: bool = False, namespaces: bool = False) -> Hyperedg
 
         return simplify(self, subtypes=subtypes, namespaces=namespaces)
 
+    def transform(
+        self,
+        origin_pattern: Hyperedge | str | list | tuple,
+        target_pattern: Hyperedge | str | list | tuple,
+        recursive: bool = True,
+    ) -> Hyperedge:
+        """Pattern-driven rewrite. See ``hyperbase.transforms.transform``."""
+        from hyperbase.transforms import transform
+
+        return transform(self, origin_pattern, target_pattern, recursive=recursive)
+
     def type(self) -> str:
         """Returns the type of this edge as a string.
         Type inference is performed.

src/hyperbase/transforms.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+from collections.abc import Iterator
 from typing import cast
 
 import hyperbase.constants as const
@@ -153,3 +154,248 @@ def add_argument(
     new_edge = insert_argrole(edge, argrole, pos)
     combined = (*tuple(new_edge[: pos + 1]), arg, *tuple(new_edge[pos + 1 :]))
     return Hyperedge(combined)
+
+
+#############
+# transform #
+#############
+
+
+def transform(
+    edge: Hyperedge,
+    origin_pattern: Hyperedge | str | list | tuple,
+    target_pattern: Hyperedge | str | list | tuple,
+    recursive: bool = True,
+) -> Hyperedge:
+    """Pattern-driven rewrite of *edge*.
+
+    If ``origin_pattern`` matches *edge*, the edge is rewritten according to
+    ``target_pattern`` using the variable bindings extracted from the match.
+    If it does not match, the edge is returned unchanged.
+
+    With ``{}`` argroles in the target, un-consumed arguments and argroles
+    from the original edge at the matched (top) level are preserved alongside
+    the target's. Without ``{}`` only what appears in the target is kept.
+
+    Patterns must contain only named variables; anonymous wildcards
+    (``*``, ``.``, ``(*)``, ``...``) and functional patterns (``var``, ``any``,
+    ``atoms``, ``lemma``) raise ``ValueError``. The target's variables must be
+    a subset of the origin's variables.
+
+    By default the rewrite is applied recursively (depth-first, once per
+    level). Pass ``recursive=False`` for a shallow rewrite.
+    """
+    origin = hedge(origin_pattern)
+    target = hedge(target_pattern)
+    _validate_transform_patterns(origin, target)
+    return _transform_impl(edge, origin, target, recursive)
+
+
+def _transform_impl(
+    edge: Hyperedge,
+    origin: Hyperedge,
+    target: Hyperedge,
+    recursive: bool,
+) -> Hyperedge:
+    from hyperbase.patterns import match_pattern
+
+    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)
+    return edge
+
+
+def _validate_transform_patterns(origin: Hyperedge, target: Hyperedge) -> None:
+    from hyperbase.patterns.checks import is_variable, is_wildcard
+
+    for label, pattern in (("origin_pattern", origin), ("target_pattern", target)):
+        if _contains_fun_pattern(pattern):
+            raise ValueError(
+                f"Functional patterns are not supported in {label}: {pattern}"
+            )
+        for atom in _walk_atoms(pattern):
+            if is_wildcard(atom) and not is_variable(atom):
+                raise ValueError(
+                    f"Anonymous wildcard '{atom}' not allowed in {label}; "
+                    f"use a named variable"
+                )
+
+    origin_vars = _collect_var_names(origin)
+    target_vars = _collect_var_names(target)
+    extra = target_vars - origin_vars
+    if extra:
+        raise ValueError(
+            f"target_pattern uses variables not in origin_pattern: {sorted(extra)}"
+        )
+
+
+def _walk_atoms(edge: Hyperedge) -> Iterator[Atom]:
+    if edge.atom:
+        yield cast(Atom, edge)
+    else:
+        for child in edge:
+            yield from _walk_atoms(child)
+
+
+def _contains_fun_pattern(edge: Hyperedge) -> bool:
+    from hyperbase.patterns.checks import is_fun_pattern
+
+    if edge.atom:
+        return False
+    if is_fun_pattern(edge):
+        return True
+    return any(_contains_fun_pattern(c) for c in edge)
+
+
+def _collect_var_names(pattern: Hyperedge) -> set[str]:
+    from hyperbase.patterns.checks import is_variable, variable_name
+
+    names: set[str] = set()
+    for atom in _walk_atoms(pattern):
+        if is_variable(atom):
+            names.add(variable_name(atom))
+    return names
+
+
+def _find_var_atom(pattern: Hyperedge, name: str) -> Atom | None:
+    from hyperbase.patterns.checks import is_variable, variable_name
+
+    for atom in _walk_atoms(pattern):
+        if is_variable(atom) and variable_name(atom) == name:
+            return atom
+    return None
+
+
+def _strip_braces(s: str) -> str:
+    if len(s) >= 2 and s[0] == "{" and s[-1] == "}":
+        return s[1:-1]
+    return s
+
+
+def _atom_decoration_compatible(
+    origin_parts: list[str], target_parts: list[str]
+) -> bool:
+    """True if origin and target atom decorations differ only in argroles."""
+    o_role = origin_parts[1].split(".") if len(origin_parts) > 1 else [""]
+    t_role = target_parts[1].split(".") if len(target_parts) > 1 else [""]
+    if o_role[0] != t_role[0]:
+        return False
+    return origin_parts[2:] == target_parts[2:]
+
+
+def _instantiate(
+    target: Hyperedge,
+    origin: Hyperedge,
+    original: Hyperedge,
+    bindings: dict[str, Hyperedge],
+    top: bool,
+) -> Hyperedge:
+    from hyperbase.patterns.checks import is_variable
+
+    if target.atom:
+        if is_variable(target):
+            return _substitute_var_atom(cast(Atom, target), origin, bindings)
+        return target
+
+    if top and original.not_atom and target.argroles().startswith("{"):
+        return _instantiate_with_preserve(target, origin, original, bindings)
+
+    return Hyperedge(
+        tuple(_instantiate(c, origin, original, bindings, top=False) for c in target)
+    )
+
+
+def _substitute_var_atom(
+    target_atom: Atom,
+    origin: Hyperedge,
+    bindings: dict[str, Hyperedge],
+) -> Hyperedge:
+    from hyperbase.patterns.checks import variable_name
+
+    name = variable_name(target_atom)
+    if name not in bindings:
+        raise ValueError(f"Variable '{name}' has no binding in matched origin pattern")
+    binding = bindings[name]
+    target_parts = target_atom.parts()
+    if len(target_parts) == 1:
+        return binding
+
+    if binding.atom:
+        binding_atom = cast(Atom, binding)
+        new_parts = [binding_atom.root(), *target_parts[1:]]
+        return Atom("/".join(p for p in new_parts if p))
+
+    origin_atom = _find_var_atom(origin, name)
+    origin_parts = origin_atom.parts() if origin_atom else [name]
+    if not _atom_decoration_compatible(origin_parts, target_parts):
+        raise ValueError(
+            f"Cannot change type or namespace on non-atomic binding for "
+            f"variable '{name}': origin {origin_parts} vs target {target_parts}"
+        )
+
+    t_role = target_parts[1].split(".") if len(target_parts) > 1 else [""]
+    target_argroles = t_role[1] if len(t_role) > 1 else ""
+    o_role = origin_parts[1].split(".") if len(origin_parts) > 1 else [""]
+    origin_argroles = o_role[1] if len(o_role) > 1 else ""
+    if target_argroles == origin_argroles:
+        return binding
+    bare = _strip_braces(target_argroles)
+    if bare == "":
+        return binding
+    return replace_argroles(binding, bare)
+
+
+def _instantiate_with_preserve(
+    target: Hyperedge,
+    origin: Hyperedge,
+    original: Hyperedge,
+    bindings: dict[str, Hyperedge],
+) -> Hyperedge:
+    consumed = _consumed_arg_indices(original, origin)
+    edge_argroles = original.argroles()
+    if edge_argroles.startswith("{"):
+        edge_argroles = edge_argroles[1:-1]
+
+    args = list(original[1:])
+    preserved_args = [args[i] for i in range(len(args)) if i not in consumed]
+    preserved_argroles = "".join(
+        edge_argroles[i]
+        for i in range(min(len(edge_argroles), len(args)))
+        if i not in consumed
+    )
+
+    new_children = [
+        _instantiate(c, origin, original, bindings, top=False) for c in target
+    ]
+    final = Hyperedge((new_children[0], *preserved_args, *new_children[1:]))
+
+    target_spec = _strip_braces(target.argroles())
+    new_argroles = preserved_argroles + target_spec
+    if new_argroles == "":
+        return final
+    return replace_argroles(final, new_argroles)
+
+
+def _consumed_arg_indices(
+    original_edge: Hyperedge,
+    origin_pattern: Hyperedge,
+) -> set[int]:
+    from hyperbase.patterns import match_pattern
+
+    consumed: set[int] = set()
+    if origin_pattern.atom or original_edge.atom:
+        return consumed
+
+    args = list(original_edge[1:])
+    for arg_pattern in origin_pattern[1:]:
+        for i, arg in enumerate(args):
+            if i in consumed:
+                continue
+            if match_pattern(arg, arg_pattern):
+                consumed.add(i)
+                break
+    return consumed