Apply design patterns across packages: shared scanner loop, polymorphic criteria nodes, label dispatch, split SLR parser

criteria/src/lambdafeedback/criteria/graph.py

@@ -115,15 +115,7 @@ def _check_label_is_free(self, label: str) -> None:
 
     def add_node(self, node: Node) -> None:
         """Add a copy of ``node`` (e.g. from another graph) to this graph, without its edges."""
-        if isinstance(node, EvaluationNode):
-            copy = self.add_evaluation_node(node.label, node.summary, node.details, node.sufficiencies, node.evaluate)
-            copy.replacement = node.replacement
-        elif isinstance(node, CriterionNode):
-            self.add_criterion_node(node.label, node.summary, node.details, node.feedback_string_generator)
-        elif isinstance(node, OutputNode):
-            self.add_output_node(node.label, node.summary, node.details)
-        else:
-            raise CriteriaGraphError("Can only add evaluation, criterion or output nodes to criteria graph.")
+        node.copy_into(self)
 
     def attach(self, source_label: str, target_label: str) -> None:
         """Add an edge between existing nodes: evaluation to criterion, or criterion to evaluation or output.
@@ -133,26 +125,12 @@ def attach(self, source_label: str, target_label: str) -> None:
         node is unknown, the target is of the wrong kind, or the edge exists.
         """
         source = self._source(source_label)
-        if isinstance(source, EvaluationNode):
-            target: Node | None = self.criteria.get(target_label)
-            if target is None and target_label in self.outputs:
-                raise CriteriaGraphError(
-                    f"{target_label} is an output node. Only criterion nodes can be attached to evaluation nodes."
-                )
-            if target is None and target_label in self.evaluations:
-                raise CriteriaGraphError(
-                    f"Both {source_label} and {target_label} are evaluation nodes. "
-                    "Only criterion nodes can be attached to evaluation nodes."
-                )
-        else:
-            target = self.evaluations.get(target_label) or self.outputs.get(target_label)
-            if target is None and target_label in self.criteria:
-                raise CriteriaGraphError(
-                    f"Both {source_label} and {target_label} are criterion nodes. "
-                    "Only evaluation nodes can be attached to criterion nodes."
-                )
+        target: Node | None = (
+            self.evaluations.get(target_label) or self.criteria.get(target_label) or self.outputs.get(target_label)
+        )
         if target is None:
             raise CriteriaGraphError(f"Unknown node {target_label}.")
+        source.check_target(target)
         edge = Edge(source, target)
         if edge in source.outgoing:
             raise CriteriaGraphError(f"{target_label} is already attached to {source_label}.")

criteria/src/lambdafeedback/criteria/nodes.py

@@ -9,7 +9,12 @@
 
 from collections.abc import Callable, Mapping
 from dataclasses import dataclass, field
-from typing import Any, Protocol, TypeAlias
+from typing import TYPE_CHECKING, Any, ClassVar, Literal, Protocol, TypeAlias
+
+from .errors import CriteriaGraphError
+
+if TYPE_CHECKING:
+    from .graph import CriteriaGraph
 
 ReachedCriteria: TypeAlias = Mapping[str, Mapping[str, Any] | None]
 """``{criterion_label: feedback_inputs_or_None}`` for the criteria that hold (or were reached)."""
@@ -20,6 +25,13 @@
 FeedbackStringGenerator: TypeAlias = Callable[[Mapping[str, Any]], str | None]
 """Renders a criterion's feedback text from the inputs its evaluation reported."""
 
+NodeKind: TypeAlias = Literal["evaluation", "criterion", "output"]
+
+# Mermaid node shapes (opening, closing).
+EVALUATION_STYLE = ("([", "])")
+CRITERION_STYLE = ("[", "]")
+OUTPUT_STYLE = ("{{", "}}")
+
 
 class ResultLike(Protocol):
     """The part of a result object ``CriteriaGraph.export_feedback`` needs (``lf_toolkit.evaluation.Result`` fits)."""
@@ -41,14 +53,31 @@ def no_feedback(inputs: Mapping[str, Any]) -> None:
 
 @dataclass(eq=False)
 class Node:
-    """A graph node. Nodes compare (and hash) by identity; within a graph, labels are unique."""
+    """A graph node. Nodes compare (and hash) by identity; within a graph, labels are unique.
+
+    Each kind of node (the subclasses) sets ``KIND`` and ``STYLE`` and says
+    which edges it allows (:meth:`check_target`); a plain ``Node`` has no kind
+    and can't be added to a graph.
+    """
+
+    KIND: ClassVar[NodeKind | None] = None
+    STYLE: ClassVar[tuple[str, str] | None] = None
+    """Mermaid shape (opening, closing)."""
 
     label: str
     summary: str
     details: str
     incoming: list[Edge] = field(default_factory=list, init=False, repr=False)
     outgoing: list[Edge] = field(default_factory=list, init=False, repr=False)
 
+    def copy_into(self, graph: CriteriaGraph) -> Node:
+        """Add a copy of this node, without its edges, to ``graph`` and return the copy."""
+        raise CriteriaGraphError("Can only add evaluation, criterion or output nodes to criteria graph.")
+
+    def check_target(self, target: Node) -> None:
+        """Raise :class:`CriteriaGraphError` unless an edge may lead from this node to ``target``."""
+        raise CriteriaGraphError(f"{self.label} can't have outgoing edges.")
+
 
 @dataclass(eq=False)
 class EvaluationNode(Node):
@@ -61,22 +90,60 @@ class EvaluationNode(Node):
     ``CriteriaGraph.starting_evaluations``).
     """
 
+    KIND: ClassVar[NodeKind | None] = "evaluation"
+    STYLE: ClassVar[tuple[str, str] | None] = EVALUATION_STYLE
+
     evaluate: Evaluate | None = None
     replacement: str | None = None
     sufficiencies: tuple[str, ...] = ()
 
+    def copy_into(self, graph: CriteriaGraph) -> EvaluationNode:
+        copy = graph.add_evaluation_node(self.label, self.summary, self.details, self.sufficiencies, self.evaluate)
+        copy.replacement = self.replacement
+        return copy
+
+    def check_target(self, target: Node) -> None:
+        if target.KIND == "output":
+            raise CriteriaGraphError(
+                f"{target.label} is an output node. Only criterion nodes can be attached to evaluation nodes."
+            )
+        if target.KIND == "evaluation":
+            raise CriteriaGraphError(
+                f"Both {self.label} and {target.label} are evaluation nodes. "
+                "Only criterion nodes can be attached to evaluation nodes."
+            )
+
 
 @dataclass(eq=False)
 class CriterionNode(Node):
     """An outcome of an evaluation, with the generator for its feedback text."""
 
+    KIND: ClassVar[NodeKind | None] = "criterion"
+    STYLE: ClassVar[tuple[str, str] | None] = CRITERION_STYLE
+
     feedback_string_generator: FeedbackStringGenerator = no_feedback
 
+    def copy_into(self, graph: CriteriaGraph) -> CriterionNode:
+        return graph.add_criterion_node(self.label, self.summary, self.details, self.feedback_string_generator)
+
+    def check_target(self, target: Node) -> None:
+        if target.KIND == "criterion":
+            raise CriteriaGraphError(
+                f"Both {self.label} and {target.label} are criterion nodes. "
+                "Only evaluation nodes can be attached to criterion nodes."
+            )
+
 
 @dataclass(eq=False)
 class OutputNode(Node):
     """A terminal node; output nodes have no outgoing edges."""
 
+    KIND: ClassVar[NodeKind | None] = "output"
+    STYLE: ClassVar[tuple[str, str] | None] = OUTPUT_STYLE
+
+    def copy_into(self, graph: CriteriaGraph) -> OutputNode:
+        return graph.add_output_node(self.label, self.summary, self.details)
+
 
 @dataclass(frozen=True, eq=False)
 class Edge:

criteria/src/lambdafeedback/criteria/render.py

@@ -6,17 +6,14 @@
 from collections.abc import Collection, Mapping
 from typing import TYPE_CHECKING, Any
 
-from .nodes import EvaluationNode, Node
+from .nodes import CRITERION_STYLE, EVALUATION_STYLE, OUTPUT_STYLE, EvaluationNode, Node
 
 if TYPE_CHECKING:
     from .graph import CriteriaGraph
     from .tree import CriteriaTree
 
-# Mermaid node shapes (opening, closing).
-EVALUATION_STYLE = ("([", "])")
+# Mermaid node shapes (opening, closing); each kind of node's own is its STYLE (see nodes).
 STARTING_EVALUATION_STYLE = (">", "]")
-CRITERION_STYLE = ("[", "]")
-OUTPUT_STYLE = ("{{", "}}")
 SPECIAL_STYLE = ("[[", "]]")
 
 

criteria/src/lambdafeedback/criteria/tree.py

@@ -8,8 +8,8 @@
 from typing import TYPE_CHECKING, Any, Literal
 
 from .errors import CriteriaGraphError
-from .nodes import CriterionNode, EvaluationNode, Node, OutputNode
-from .render import CRITERION_STYLE, EVALUATION_STYLE, OUTPUT_STYLE, tree_to_mermaid
+from .nodes import Node, NodeKind, OutputNode
+from .render import tree_to_mermaid
 
 if TYPE_CHECKING:
     from .graph import CriteriaGraph
@@ -56,18 +56,20 @@ def to_json(self) -> str:
 def _tree_node(
     node: Node, identifier: int, main_criteria: Collection[str], parent: CriteriaTree | None = None
 ) -> CriteriaTree:
-    type_label: NodeType
-    if isinstance(node, EvaluationNode):
-        style, type_label = EVALUATION_STYLE, "evaluation"
-    elif isinstance(node, CriterionNode):
-        style, type_label = CRITERION_STYLE, "main_criterion" if node.label in main_criteria else "criterion"
-    elif isinstance(node, OutputNode):
-        style, type_label = OUTPUT_STYLE, "output"
-    else:
+    kind, style = node.KIND, node.STYLE
+    if kind is None or style is None:
         raise CriteriaGraphError(f"Cannot display node {node.label} of type {type(node).__name__}.")
+    type_label = _type_label(kind, node.label, main_criteria)
     return CriteriaTree(node.label, node.summary, node.details, f"_{identifier}", style, type_label, parent)
 
 
+def _type_label(kind: NodeKind, label: str, main_criteria: Collection[str]) -> NodeType:
+    """The node's kind, with criteria in ``main_criteria`` marked as main criteria."""
+    if kind == "criterion" and label in main_criteria:
+        return "main_criterion"
+    return kind
+
+
 def build_tree(graph: CriteriaGraph, starting_evaluation: str, main_criteria: Collection[str] = ()) -> CriteriaTree:
     """Unroll ``graph`` depth-first from the evaluation labelled ``starting_evaluation``."""
     node = graph.evaluations.get(starting_evaluation)

expression_parsing/src/lambdafeedback/expression_parsing/_scanning.py

@@ -1,7 +1,10 @@
-"""Bracket helpers shared by the string scanners (no dependencies on the rest of the package)."""
+"""Helpers shared by the string scanners (no dependencies on the rest of the package)."""
 
 from __future__ import annotations
 
+from collections.abc import Callable
+from typing import TypeAlias
+
 OPENING_BRACKETS = "([{"
 CLOSING_BRACKETS = ")]}"
 GROUPING = OPENING_BRACKETS + CLOSING_BRACKETS
@@ -19,3 +22,24 @@ def find_matching_parenthesis(string: str, index: int, delimiters: tuple[str, st
             if depth == 0:
                 return k
     return -1
+
+
+Match: TypeAlias = tuple[int, int, str]
+"""``(start, end, replacement)``: replace ``text[start:end]`` by ``replacement``."""
+
+
+def replace_matches(text: str, find: Callable[[str, int], Match | None]) -> str:
+    """Replace each match ``find(text, index)`` reports, left to right.
+
+    ``find`` returns the first match starting at or after ``index`` (or
+    ``None``) and must end after ``index``; scanning resumes at the match's
+    end, so replacements are never scanned again.
+    """
+    pieces = []
+    done = 0
+    while (match := find(text, done)) is not None:
+        start, end, replacement = match
+        pieces += [text[done:start], replacement]
+        done = end
+    pieces.append(text[done:])
+    return "".join(pieces)

expression_parsing/src/lambdafeedback/expression_parsing/latex.py

@@ -3,7 +3,7 @@
 from __future__ import annotations
 
 import logging
-from collections.abc import Iterator, Mapping
+from collections.abc import Callable, Iterator, Mapping
 from string import ascii_letters
 from typing import Any, Literal, TypeAlias
 
@@ -17,7 +17,7 @@
 from sympy.printing.latex import LatexPrinter
 
 from ._safe_parse import TOO_DEEP_MESSAGE, check_input_length, safe_parse_expr
-from ._scanning import CLOSING_BRACKETS, OPENING_BRACKETS, find_matching_parenthesis
+from ._scanning import CLOSING_BRACKETS, OPENING_BRACKETS, Match, find_matching_parenthesis, replace_matches
 from .errors import LatexParseError
 from .params import SymbolSpec
 from .preprocessing import expand_plus_minus, split_equation
@@ -36,19 +36,17 @@
 
 def _replace_command(latex: str, command: str, replacement: str) -> str:
     """``command`` (e.g. ``\\pm``, but not the start of ``\\pmod``) and any whitespace after it as ``replacement``."""
-    pieces = []
-    done = 0
-    index = latex.find(command)
-    while index != -1:
-        after = index + len(command)
-        if latex[after : after + 1] in _LETTERS:
-            index = latex.find(command, index + 1)
-            continue
-        pieces += [latex[done:index], replacement]
-        done = len(latex) - len(latex[after:].lstrip())
-        index = latex.find(command, done)
-    pieces.append(latex[done:])
-    return "".join(pieces)
+
+    def find(text: str, index: int) -> Match | None:
+        start = text.find(command, index)
+        while start != -1 and text[start + len(command) : start + len(command) + 1] in _LETTERS:
+            start = text.find(command, start + 1)
+        if start == -1:
+            return None
+        after = start + len(command)
+        return start, len(text) - len(text[after:].lstrip()), replacement
+
+    return replace_matches(latex, find)
 
 
 class _LatexPrinter(LatexPrinter):
@@ -382,18 +380,20 @@ def sanitise_latex(response: str) -> str:
     """
     response = "".join(response.split()).replace("~", " ")
     for wrapper in (r"\mathrm", r"\text"):
-        processed = []
-        index = 0
-        while index < len(response):
-            wrapper_start = response.find(wrapper + "{", index)
-            if wrapper_start < 0:
-                processed.append(response[index:])
-                break
-            processed.append(response[index:wrapper_start])
-            wrapper_end = find_matching_parenthesis(response, wrapper_start + 1, delimiters=("{", "}"))
-            if wrapper_end < 0:
-                raise LatexParseError(f"Unclosed {wrapper}{{...}} in {response!r}.", expression=response)
-            processed.append(response[wrapper_start + len(wrapper) + 1 : wrapper_end])
-            index = wrapper_end + 1
-        response = "".join(processed)
+        response = replace_matches(response, _wrapped_content(wrapper))
     return response
+
+
+def _wrapped_content(wrapper: str) -> Callable[[str, int], Match | None]:
+    """Finder for :func:`replace_matches`: ``wrapper{...}`` as its content; raises if the brace is unclosed."""
+
+    def find(text: str, index: int) -> Match | None:
+        start = text.find(wrapper + "{", index)
+        if start < 0:
+            return None
+        end = find_matching_parenthesis(text, start + 1, delimiters=("{", "}"))
+        if end < 0:
+            raise LatexParseError(f"Unclosed {wrapper}{{...}} in {text!r}.", expression=text)
+        return start, end + 1, text[start + len(wrapper) + 1 : end]
+
+    return find

expression_parsing/src/lambdafeedback/expression_parsing/substitution.py

@@ -6,6 +6,7 @@
 from dataclasses import dataclass
 from typing import TYPE_CHECKING
 
+from ._scanning import Match, replace_matches
 from .symbol_tables import (
     ELEMENTARY_FUNCTIONS,
     EMBEDDABLE_FUNCTION_NAMES,
@@ -154,23 +155,21 @@ def normalise_unicode_operators(expr: str) -> str:
     :data:`UNICODE_OPERATORS` and :data:`SUPERSCRIPTS`); others, such as
     ``√`` or ``½``, are left for parsing to report.
     """
-    expr = expr.translate(_UNICODE_OPERATOR_TABLE)
-    pieces = []
-    done = index = 0
-    while index < len(expr):
-        if expr[index] not in SUPERSCRIPTS:
-            index += 1
-            continue
-        end = index
-        while end < len(expr) and expr[end] in SUPERSCRIPTS:
-            end += 1
-        # The power replaces any spaces before the run, and a redundant ^ or ** before those (x^²).
-        before = expr[done:index].rstrip(" ")
-        before = before.removesuffix("^") if before.endswith("^") else before.removesuffix("**")
-        pieces += [before, "**(" + "".join(SUPERSCRIPTS[char] for char in expr[index:end]) + ")"]
-        done = index = end
-    pieces.append(expr[done:])
-    return "".join(pieces)
+    return replace_matches(expr.translate(_UNICODE_OPERATOR_TABLE), _superscript_run)
+
+
+def _superscript_run(expr: str, index: int) -> Match | None:
+    """Finder for :func:`replace_matches`: the first run of superscripts at or after ``index``, as a power."""
+    start = next((k for k in range(index, len(expr)) if expr[k] in SUPERSCRIPTS), None)
+    if start is None:
+        return None
+    end = start
+    while end < len(expr) and expr[end] in SUPERSCRIPTS:
+        end += 1
+    # The power replaces any spaces before the run, and a redundant ^ or ** before those (x^²).
+    before = expr[index:start].rstrip(" ")
+    before = before.removesuffix("^") if before.endswith("^") else before.removesuffix("**")
+    return index + len(before), end, "**(" + "".join(SUPERSCRIPTS[char] for char in expr[start:end]) + ")"
 
 
 def substitute_input_symbols(expr: str, params: ExpressionParams) -> str: