diff --git a/criteria/src/lambdafeedback/criteria/graph.py b/criteria/src/lambdafeedback/criteria/graph.py index 7528ec9..b6f1566 100644 --- a/criteria/src/lambdafeedback/criteria/graph.py +++ b/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}.") diff --git a/criteria/src/lambdafeedback/criteria/nodes.py b/criteria/src/lambdafeedback/criteria/nodes.py index cd3c2ca..e947556 100644 --- a/criteria/src/lambdafeedback/criteria/nodes.py +++ b/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,7 +53,16 @@ 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 @@ -49,6 +70,14 @@ class Node: 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: diff --git a/criteria/src/lambdafeedback/criteria/render.py b/criteria/src/lambdafeedback/criteria/render.py index edf3975..200eae7 100644 --- a/criteria/src/lambdafeedback/criteria/render.py +++ b/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 = ("[[", "]]") diff --git a/criteria/src/lambdafeedback/criteria/tree.py b/criteria/src/lambdafeedback/criteria/tree.py index 0aa19e0..427eeb3 100644 --- a/criteria/src/lambdafeedback/criteria/tree.py +++ b/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) diff --git a/expression_parsing/src/lambdafeedback/expression_parsing/_scanning.py b/expression_parsing/src/lambdafeedback/expression_parsing/_scanning.py index 62ef5ce..faaa543 100644 --- a/expression_parsing/src/lambdafeedback/expression_parsing/_scanning.py +++ b/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) diff --git a/expression_parsing/src/lambdafeedback/expression_parsing/latex.py b/expression_parsing/src/lambdafeedback/expression_parsing/latex.py index 70168fd..f069345 100644 --- a/expression_parsing/src/lambdafeedback/expression_parsing/latex.py +++ b/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 diff --git a/expression_parsing/src/lambdafeedback/expression_parsing/substitution.py b/expression_parsing/src/lambdafeedback/expression_parsing/substitution.py index c874c13..17054d1 100644 --- a/expression_parsing/src/lambdafeedback/expression_parsing/substitution.py +++ b/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: diff --git a/slr_parsing/CHANGELOG.md b/slr_parsing/CHANGELOG.md index acc8236..368aa5f 100644 --- a/slr_parsing/CHANGELOG.md +++ b/slr_parsing/CHANGELOG.md @@ -34,7 +34,9 @@ The first public release, after the extraction refactor. The import path is now - **Errors:** bare `Exception`s are replaced by `GrammarError` (invalid grammar), `ScanError` (text no token matches) and `ParseError` (input doesn't match the grammar). All derive from `SLRError` and `ValueError`. `ParseError`'s message is a one-line summary ("Unexpected end of input.", "Unexpected OP '+' at position 3."); the full parser-state dump that used to be the message is available from `ParseError.details()`, with the parts as attributes. An invalid `ExprNode` child raises `TypeError`. - **Logging instead of printing:** `parse()` no longer takes `verbose`; set the `lambdafeedback.slr_parsing.parser` logger to DEBUG to trace shifts and reductions. Unreachable grammar states and reductions are logged as a WARNING instead of printed, and the start state is no longer falsely reported. - `build_expression_parser` drops the unused `group_node` parameter; sequence defaults are tuples instead of shared mutable lists. -- `SLRParser` keeps its construction internals private. The public surface is `scan`, `parse`, `parsing_action`, `parsing_table`, `states` (now a list of kernel item tuples, was a dict of kernel → closure), `productions`, `token_list`, `start_token`/`end_token`/`null_token`, `error_handlers`, `tag_handler` and the debug helpers `state_string` and `parsing_table_to_string`. `state_string_list` is removed. +- `SLRParser` keeps its construction internals private. The public surface is `scan`, `parse`, `parsing_action`, `parsing_table`, `states` (now a list of kernel item tuples, was a dict of kernel → closure), `productions`, `token_list`, `start_token`/`end_token`/`null_token`, `error_handlers`, `tag_handler`, `scanner`, `table` and the debug helpers `state_string` and `parsing_table_to_string`. `state_string_list` is removed. +- **`SLRParser` is split in three:** a `Scanner` (`parser.scanner`) turns text into tokens, a `ParseTable` (`parser.table`) holds the LR(0) states, FIRST and FOLLOW sets and the entries, and `SLRParser` runs the parse with them. Its constructor and methods are unchanged. +- **Parse-table entries are typed:** `parsing_table` cells and `parsing_action(state, symbol)` are `Shift(state)`, `Reduce(production)`, `Accept()`, `Reject()` (a syntax error) or `Recover(handler)` (handled by `error_handlers[handler]`), all exported with the `Entry` union, instead of integers encoding shifts, accept, reductions (offset by the number of states) and handlers (as negatives). `parsing_table_to_string()` prints them as before. The entry classes are final. - Error handlers are normalised to `ErrorHandler(condition, action)` named tuples; plain 2-tuples are still accepted. - `replace_tag(node, t, t)` keeps the tag (it used to remove it). - **`ExprNode` equality is by identity.** Nodes used to inherit `Token`'s label-only equality, so any two trees with the same root label compared equal (`x+x == x+(x+x)`). `Token`s still compare by label, and a `Token` never equals an `ExprNode`. Compare trees by content (e.g. `content_string()`) instead. @@ -47,7 +49,7 @@ The first public release, after the extraction refactor. The import path is now - an error handler that isn't a `(condition, action)` pair of callables (was `TypeError`). - **Token patterns are literals, not regular expressions.** The pattern of a `(pattern, label)` token specification is a plain string, matched exactly, or the new `Lexeme(forms, space_before=False, space_after=False)`: the first of several literal forms, optionally with spaces around it, where a space inside a form matches one or more spaces. `(r" *\+ *", "PLUS")` becomes `(Lexeme(("+",), True, True), "PLUS")`, and `(r"\(", "LPAR")` becomes `("(", "LPAR")`. Anything else (a run of digits, say) is scanned with a matcher, the `(symbol, label, matcher)` shape, which was already supported. Ties between equally long matches go to the specification with the larger `Lexeme.weight` (the longest form, plus 2 per side allowing spaces) or the longer literal, mirroring the old order by pattern length. `build_expression_parser` builds `Lexeme`s for its operators and delimiters, so its callers are unaffected. - `repr(ExprNode)` matches `str(ExprNode)`. -- Modules: `tokens`, `actions`, `tags`, `grammar` (with the grammar format documented), `errors`, `parser`, `builder`. Fully type-annotated (`py.typed`). The type aliases used in public signatures (`ErrorCondition`, `ErrorAction`, `TraverseStep`, `TagRule`, `TagCondition`) are exported from the package root. +- Modules: `tokens`, `actions`, `tags`, `grammar` (with the grammar format documented), `errors`, `scanner`, `table`, `parser`, `builder`. Fully type-annotated (`py.typed`). The type aliases used in public signatures (`ErrorCondition`, `ErrorAction`, `TraverseStep`, `TagRule`, `TagCondition`) are exported from the package root. ### Fixed diff --git a/slr_parsing/README.md b/slr_parsing/README.md index fd03c19..5756a8c 100644 --- a/slr_parsing/README.md +++ b/slr_parsing/README.md @@ -159,7 +159,10 @@ content (for example with `content_string()`), not with `==`. Set the `lambdafeedback.slr_parsing.parser` logger to `DEBUG` to trace every shift and reduction. `parser.parsing_table_to_string()` and -`parser.state_string(state)` print the generated table and states. +`parser.state_string(state)` print the generated table and states; the +table itself is `parser.table` (a `ParseTable`, with its `states`, `first` +and `follow` sets and typed entries such as `Shift(state)` and +`Reduce(production)`). ## Changelog diff --git a/slr_parsing/src/lambdafeedback/slr_parsing/__init__.py b/slr_parsing/src/lambdafeedback/slr_parsing/__init__.py index a110d6e..c34e71e 100644 --- a/slr_parsing/src/lambdafeedback/slr_parsing/__init__.py +++ b/slr_parsing/src/lambdafeedback/slr_parsing/__init__.py @@ -49,6 +49,8 @@ TokenSpec, ) from .parser import SLRParser +from .scanner import Scanner +from .table import Accept, Entry, ParseTable, Recover, Reduce, Reject, Shift from .tags import ( TagCondition, TagRule, @@ -81,8 +83,10 @@ "END_DELIMITER_LABEL", "GROUP_LABEL", "START_DELIMITER_LABEL", + "Accept", "Action", "CatchAll", + "Entry", "ErrorAction", "ErrorCondition", "ErrorHandler", @@ -93,10 +97,16 @@ "Matcher", "ParseError", "ParseState", + "ParseTable", "Production", + "Recover", + "Reduce", + "Reject", "SLRError", "SLRParser", "ScanError", + "Scanner", + "Shift", "TagCondition", "TagHandler", "TagRule", diff --git a/slr_parsing/src/lambdafeedback/slr_parsing/parser.py b/slr_parsing/src/lambdafeedback/slr_parsing/parser.py index 47da132..ee0a06a 100644 --- a/slr_parsing/src/lambdafeedback/slr_parsing/parser.py +++ b/slr_parsing/src/lambdafeedback/slr_parsing/parser.py @@ -1,4 +1,4 @@ -"""SLR(1) parser: table construction, scanning and parsing.""" +"""SLR(1) parser: a scanner and a parse table, and the parse that drives them.""" from __future__ import annotations @@ -7,47 +7,15 @@ from collections.abc import Sequence from typing import Any, Literal -from .errors import GrammarError, ParseError, ScanError -from .grammar import ( - Action, - CatchAll, - ErrorAction, - ErrorCondition, - ErrorHandler, - GrammarSymbol, - Lexeme, - Matcher, - ParseState, - TagHandler, - TokenMatcher, - TokenPattern, - TokenSpec, - match_pattern, - pattern_weight, -) +from .errors import GrammarError, ParseError +from .grammar import Action, ErrorAction, ErrorCondition, ErrorHandler, ParseState, TagHandler, TokenSpec +from .scanner import Scanner +from .table import Accept, Entry, ParseTable, Recover, Reduce, Reject, Shift, State from .tags import inherit_tags from .tokens import ExprNode, Token logger = logging.getLogger(__name__) -# An LR(0) item: (production index, position of the dot in the body). -Item = tuple[int, int] -# A state is identified by its kernel items. -State = tuple[Item, ...] - -# Parse-table entries: 0 <= e < len(states) shifts to state e; e == len(states) -# accepts; len(states) + p reduces by production p; -1 is a syntax error and -# -k (k >= 2) runs error handler k - 2. -_ERROR = -1 - - -def _handler_entry(handler_index: int) -> int: - return -2 - handler_index - - -def _handler_index(entry: int) -> int: - return -2 - entry - class SLRParser: """An SLR(1) parser generated from token specifications and productions. @@ -55,6 +23,11 @@ class SLRParser: See :mod:`lambdafeedback.slr_parsing.grammar` for how grammars are written. ``tag_handler`` computes the tags of nodes built by reduction actions; ``error_handlers`` recover from (or explain) syntax errors. + + The work is split between a :class:`~lambdafeedback.slr_parsing.scanner.Scanner` + (``scanner``), which turns text into tokens, and a + :class:`~lambdafeedback.slr_parsing.table.ParseTable` (``table``), built + from the productions; the parser runs the parse with them. """ def __init__( @@ -67,10 +40,8 @@ def __init__( error_handlers: Sequence[ErrorHandler | tuple[ErrorCondition, ErrorAction]] = (), tag_handler: TagHandler = inherit_tags, ) -> None: - # Heaviest patterns first: scan() keeps the longest match, so this - # order only breaks ties between equal-length matches. - self._check_token_specs(token_list) - self.token_list: list[TokenSpec] = sorted(token_list, key=lambda spec: -pattern_weight(spec.pattern)) + self.scanner = Scanner(token_list) + self.token_list = self.scanner.token_list self.productions = list(productions) self.start_symbol = start_symbol self.end_symbol = end_symbol @@ -79,118 +50,30 @@ def __init__( self.error_handlers = [self._error_handler(handler) for handler in error_handlers] self.tag_handler = tag_handler - self._prepare_scanner() - self.start_token = self._grammar_symbol(start_symbol, "Start symbol") - self.end_token = self._grammar_symbol(end_symbol, "End symbol") - self.null_token = self._grammar_symbol(null_symbol, "Null symbol") + self.start_token = self.scanner.grammar_symbol(start_symbol, "Start symbol") + self.end_token = self.scanner.grammar_symbol(end_symbol, "End symbol") + self.null_token = self.scanner.grammar_symbol(null_symbol, "Null symbol") - self._closures: dict[State, tuple[Item, ...]] = {} self._check_duplicate_productions() - self._tokenise_productions() - self._classify_symbols() - self._compute_first() - self._compute_follow() - self._build_states() - self._build_table() + # Indexed like the productions: productions sharing a body keep their own actions. + self._reductions: list[Action | None] = [production[2] for production in self.productions] + self.table = ParseTable( + self._tokenise_productions(), self.start_token, self.end_token, self.null_token, self.error_handlers + ) + self.states = self.table.states + self.parsing_table = self.table.rows + # Looked up on every parse step, so bound here rather than reached through the table's methods. + self._symbol_ids = self.table.symbol_ids + self._gotos = self.table.gotos self._report_unreachable() # ------------------------------------------------------------------ - # Scanner + # Grammar # ------------------------------------------------------------------ - @staticmethod - def _check_token_specs(token_list: Sequence[TokenSpec]) -> None: - for spec in token_list: - if not isinstance(spec, TokenPattern | TokenMatcher | CatchAll | GrammarSymbol): - raise GrammarError( - f"Token specification {spec!r} is not a TokenPattern, TokenMatcher, CatchAll or GrammarSymbol." - ) - if sum(isinstance(spec, CatchAll) for spec in token_list) > 1: - raise GrammarError("Only one token type can be used to catch undefined lexemes.") - - def _prepare_scanner(self) -> None: - self._catch_all = next((spec for spec in self.token_list if isinstance(spec, CatchAll)), None) - self._expression_symbols = [ - (spec.pattern, spec.label) for spec in self.token_list if isinstance(spec, TokenPattern) - ] - self._expression_rules: list[tuple[str, Matcher]] = [ - (spec.label, spec.matcher) for spec in self.token_list if isinstance(spec, TokenMatcher) - ] - self._bnf_symbols: list[tuple[str | Lexeme, str]] = [(spec.pattern, spec.label) for spec in self.token_list] - def scan(self, expr: str, mode: Literal["expression", "bnf"] = "expression") -> list[Token]: - """Split ``expr`` into tokens, preferring the longest match at each position. - - Of equal-length matches, the specification with the longest pattern - string wins; a matcher only wins over a pattern with a strictly - longer match. - - ``mode="bnf"`` scans production strings into grammar symbols: every - token specification is used as a plain pattern and no matchers run. - Text that no specification matches is a :class:`GrammarError` in bnf - mode (the catch-all only applies to expressions). - """ - if mode == "expression": - symbols, rules = self._expression_symbols, self._expression_rules - elif mode == "bnf": - symbols, rules = self._bnf_symbols, [] - else: - raise ValueError(f"Unknown scan mode {mode!r}, expected 'expression' or 'bnf'.") - - tokens: list[Token] = [] - index = 0 - undefined = "" - while index - len(undefined) < len(expr): - label: str | None = None - content: Any = "" - length = 0 - for pattern, symbol_label in symbols: - matched_length = match_pattern(pattern, expr, index) - if matched_length is not None and matched_length > length: - length = matched_length - content = expr[index : index + length] - label = symbol_label - rest = expr[index:] if rules else "" # matchers read the text from where they are tried - for rule_label, matcher in rules: - matched, rule_content = matcher(rest) - if matched is not None and len(matched) > length: - content = rule_content - length = len(matched) - label = rule_label - - token = None - if label is None: - undefined += expr[index] - index += 1 - else: - token = Token(label, content, expr, index, index + length - 1) - if undefined and (token is not None or index >= len(expr)): - if mode == "bnf": - raise GrammarError(f"Undeclared grammar symbol {undefined!r} in {expr!r}.") - tokens.append(self._undefined_token(undefined, expr, index)) - undefined = "" - if token is not None: - tokens.append(token) - index += length - return tokens - - def _undefined_token(self, text: str, expr: str, index: int) -> Token: - if self._catch_all is None: - raise ScanError(f"Undefined input: {text}") - return Token(self._catch_all.label, text, expr, index - len(text), index - 1) - - def _grammar_symbol(self, symbol: str, role: str) -> Token: - try: - tokens = self.scan(symbol, mode="bnf") - except GrammarError: - tokens = [] - if len(tokens) != 1: - raise GrammarError(f"{role} {symbol!r} is not exactly one declared grammar symbol.") - return tokens[0] - - # ------------------------------------------------------------------ - # Table construction - # ------------------------------------------------------------------ + """Split ``expr`` into tokens (see :meth:`~lambdafeedback.slr_parsing.scanner.Scanner.scan`).""" + return self.scanner.scan(expr, mode) def _check_production_shapes(self) -> None: for production in self.productions: @@ -219,209 +102,36 @@ def _check_duplicate_productions(self) -> None: if duplicates: raise GrammarError("There are duplicate productions:\n" + "\n".join(duplicates)) - def _tokenise_productions(self) -> None: - self._productions_token: list[tuple[Token, list[Token]]] = [] + def _tokenise_productions(self) -> list[tuple[Token, list[Token]]]: + productions = [] for head, body, *_ in self.productions: - body_tokens = self.scan(body, mode="bnf") + body_tokens = self.scanner.scan(body, mode="bnf") if not body_tokens: raise GrammarError(f"Production {(head, body)} has an empty body; ε-productions are not supported.") - self._productions_token.append((self._grammar_symbol(head, "Production head"), body_tokens)) - # Indexed like the productions: productions sharing a body keep their own actions. - self._reductions: list[Action | None] = [production[2] for production in self.productions] - # Production indices by head, in production order. - self._productions_by_head: dict[Token, list[int]] = {} - for p, (head_token, _) in enumerate(self._productions_token): - self._productions_by_head.setdefault(head_token, []).append(p) - - def _classify_symbols(self) -> None: - self._nonterminals: list[Token] = [] - for head, _ in self._productions_token: - if head not in self._nonterminals: - self._nonterminals.append(head) - self._terminals: list[Token] = [self.end_token, self.null_token] - for _, body in self._productions_token: - for token in body: - if token not in self._terminals and token not in self._nonterminals: - self._terminals.append(token) - self._symbols = self._terminals + self._nonterminals - self._symbol_ids = {symbol: j for j, symbol in enumerate(self._symbols)} - - def _compute_first(self) -> None: - """FIRST sets of single symbols (ε-productions are not supported).""" - # Dicts as ordered sets: membership is constant-time and the order stays deterministic. - first: dict[Token, dict[Token, None]] = {t: {t: None} for t in self._terminals} - first.update({nt: {} for nt in self._nonterminals}) - changed = True - while changed: - changed = False - for nt in self._nonterminals: - for p in self._productions_by_head[nt]: - for token in self._productions_token[p][1]: - for x in first[token]: - if x not in first[nt]: - first[nt][x] = None - changed = True - if self.null_token not in first[token]: - break - self._first_dict = {symbol: list(symbols) for symbol, symbols in first.items()} - - def _first(self, tokens: Sequence[Token]) -> list[Token]: - """FIRST set of a string of symbols.""" - if len(tokens) == 1: - return self._first_dict[tokens[0]] - result: list[Token] = [] - for token in tokens: - for item in self._first_dict[token]: - if item not in result and item != self.null_token: - result.append(item) - if token != self.null_token: - break - return result - - def _compute_follow(self) -> None: - follow: dict[Token, dict[Token, None]] = {nt: {} for nt in self._nonterminals} - follow[self.start_token][self.end_token] = None - changed = True - while changed: - changed = False - for head, body in self._productions_token: - for k, token in enumerate(body): - if token not in follow: - continue - rest = body[k + 1 :] - rest_first = self._first(rest) if rest else [] - additions = rest_first - if not rest or self.null_token in rest_first: - additions = [*follow[head], *rest_first] - for item in additions: - if item != self.null_token and item not in follow[token]: - follow[token][item] = None - changed = True - self._follow = {symbol: list(symbols) for symbol, symbols in follow.items()} - - def _closure(self, item_set: Sequence[Item]) -> tuple[Item, ...]: - key = tuple(item_set) - cached = self._closures.get(key) - if cached is not None: - return cached - productions = self._productions_token - closure = list(item_set) - expanded: set[Token] = set() # nonterminals whose productions are already in the closure - for i, j in closure: # the loop also visits the items it appends - body = productions[i][1] - if j < len(body) and body[j] in self._productions_by_head and body[j] not in expanded: - expanded.add(body[j]) - closure += [(p, 0) for p in self._productions_by_head[body[j]]] - self._closures[key] = result = tuple(closure) - return result - - def _compute_transitions(self, item_set: Sequence[Item]) -> list[tuple[Token, list[Item]]]: - transitions: dict[Token, list[Item]] = {} # in order of first appearance - for i, j in item_set: - body = self._productions_token[i][1] - if j < len(body): - transitions.setdefault(body[j], []).append((i, j + 1)) - return list(transitions.items()) - - def _build_states(self) -> None: - start: State = tuple((p, 0) for p, (head, _) in enumerate(self._productions_token) if head == self.start_token) - self.states: list[State] = [start] - self._state_ids: dict[State, int] = {start: 0} - self._transitions: dict[State, list[tuple[Token, list[Item]]]] = {} - queue = deque([start]) - while queue: - state = queue.popleft() - transitions = self._compute_transitions(self._closure(state)) - self._transitions[state] = transitions - for _, kernel in transitions: - target = tuple(kernel) - if target not in self._state_ids: - self._state_ids[target] = len(self.states) - self.states.append(target) - queue.append(target) - - def _build_table(self) -> None: - candidates: list[list[list[tuple[int, int]]]] = [[[] for _ in self._symbols] for _ in self.states] - for s, state in enumerate(self.states): - targets = {symbol: self._state_ids[tuple(kernel)] for symbol, kernel in self._transitions[state]} - for production_index, dot in self._closure(state): - head, body = self._productions_token[production_index] - if dot < len(body): # shift / goto - symbol = body[dot] - candidates[s][self._symbol_ids[symbol]].append((production_index, targets[symbol])) - elif head == self.start_token: # accept - candidates[s][self._symbol_ids[self.end_token]].append((production_index, self._accept)) - else: # reduce - for symbol in self._follow[head]: - reduction = (production_index, self._reduce_entry(production_index)) - candidates[s][self._symbol_ids[symbol]].append(reduction) - - self.parsing_table: list[list[int]] = [ - [self._resolve(s, j, entries) for j, entries in enumerate(row)] for s, row in enumerate(candidates) - ] - - def _resolve(self, s: int, j: int, entries: list[tuple[int, int]]) -> int: - """Pick the table entry for state ``s`` and symbol ``j``. - - Conflicts go to the production listed last (ties to the first - candidate); empty entries become errors, handled by the first error - handler whose condition holds. - """ - if not entries: - items = [ - (self._productions_token[i][1][:dot], self._productions_token[i][1][dot:]) for i, dot in self.states[s] - ] - next_symbol = self._symbols[j] - for k, handler in enumerate(self.error_handlers): - if handler.condition(items, next_symbol): - return _handler_entry(k) - return _ERROR - best = entries[0] - for entry in entries[1:]: - if best[0] < entry[0]: - best = entry - return best[1] + productions.append((self.scanner.grammar_symbol(head, "Production head"), body_tokens)) + return productions def _report_unreachable(self) -> None: - targets = {entry for row in self.parsing_table for entry in row} - states = [self.state_string(self.states[s]) for s in range(1, len(self.states)) if s not in targets] - reductions = [ - f"{head} --> {body}" - for p, (head, body, *_) in enumerate(self.productions) - if self._productions_token[p][0] != self.start_token and self._reduce_entry(p) not in targets - ] + states, reductions = self.table.unreachable() if states or reductions: logger.warning("Unreachable states: %s; unreachable reductions: %s", states, reductions) - # Table entries (see _ERROR): shifts are state numbers, then accept, then one reduction per production. - - @property - def _accept(self) -> int: - return len(self.states) - - def _reduce_entry(self, production_index: int) -> int: - return len(self.states) + production_index - - def _reduction_index(self, entry: int) -> int | None: - """The production an entry reduces by (``None`` if it doesn't reduce).""" - p = entry - len(self.states) - return p if 0 <= p < len(self._productions_token) else None - # ------------------------------------------------------------------ # Parsing # ------------------------------------------------------------------ - def parsing_action(self, state: int, symbol: Token) -> int: + def parsing_action(self, state: int, symbol: Token) -> Entry: + """The parse-table entry for ``state`` with ``symbol`` next.""" return self.parsing_table[state][self._symbol_ids[symbol]] - def _next_action(self, state: ParseState) -> int: + def _next_action(self, state: ParseState) -> Shift | Reduce | Accept: """The table entry for the current state and lookahead, after any error-recovery handlers, or raise.""" action = self.parsing_action(state.stack[-1], state.lookahead) - while action < 0: - if action == _ERROR: - raise self.parse_error(state) - self.error_handlers[_handler_index(action)].action(self, state) + while type(action) is Recover: + self.error_handlers[action.handler].action(self, state) action = self.parsing_action(state.stack[-1], state.lookahead) + if type(action) is Reject: + raise self.parse_error(state) return action def _shift(self, target: int, state: ParseState, trace: bool) -> None: @@ -432,13 +142,13 @@ def _shift(self, target: int, state: ParseState, trace: bool) -> None: state.lookahead = state.remaining.popleft() def _reduce(self, p: int, state: ParseState, trace: bool) -> None: - production = self._productions_token[p] + production = self.table.productions[p] reduction = self._reductions[p] if reduction is None: raise GrammarError(f"Production {tuple(self.productions[p][:2])} has no reduction action.") state.output = reduction(production, state.output, self.tag_handler) del state.stack[-len(production[1]) :] - state.stack.append(self.parsing_action(state.stack[-1], production[0])) + state.stack.append(self._gotos[state.stack[-1]][production[0]]) if trace: body = "".join(x.content for x in production[1]) logger.debug("reduce by %s --> %s %s", production[0].content, body, state.output) @@ -468,19 +178,12 @@ def _parse_root(self, input_tokens: list[Token]) -> tuple[list[Any], list[Token] ) trace = logger.isEnabledFor(logging.DEBUG) action = self._next_action(state) - while action != self._accept: - if action < self._accept: - self._shift(action, state, trace) - elif (p := self._reduction_index(action)) is not None: - self._reduce(p, state, trace) - else: - raise ParseError( - f"Invalid parse table entry {action}.", - lookahead=state.lookahead, - remaining=list(state.remaining), - stack=state.stack, - output=state.output, - ) + # type() checks rather than match or isinstance: this is the parser's innermost loop. + while type(action) is not Accept: + if type(action) is Shift: + self._shift(action.state, state, trace) + elif type(action) is Reduce: + self._reduce(action.production, state, trace) action = self._next_action(state) logger.debug("accept") return state.output, list(state.remaining) @@ -506,30 +209,9 @@ def parse_error(self, state: ParseState) -> ParseError: # Debugging aids # ------------------------------------------------------------------ - def _item_strings(self, state: State) -> list[str]: - strings = [] - for production_index, dot in self._closure(state): - body = self._productions_token[production_index][1] - strings.append("".join(x.content for x in body[:dot]) + "." + "".join(x.content for x in body[dot:])) - return strings - def state_string(self, state: State) -> str: - return f"I{self._state_ids[state]}: ({', '.join(self._item_strings(state))})" + return self.table.state_string(state) def parsing_table_to_string(self) -> str: - lines = ["\t" + "\t".join(x.content for x in self._symbols)] - for s, row in enumerate(self.parsing_table): - lines.append(f"{s}\t" + "\t".join(self._entry_string(entry) for entry in row)) - return "\n".join(lines) + "\n" - - def _entry_string(self, entry: int) -> str: - """``s`` (shift), ``acc``, ``r``, ``h`` (error handler) or ``-`` (error).""" - if entry == _ERROR: - return "-" - if entry < 0: - return f"h{_handler_index(entry)}" - if entry < self._accept: - return f"s{entry}" - if entry == self._accept: - return "acc" - return f"r{self._reduction_index(entry)}" + """The parse table, tab-separated (see :meth:`~lambdafeedback.slr_parsing.table.ParseTable.to_string`).""" + return self.table.to_string() diff --git a/slr_parsing/src/lambdafeedback/slr_parsing/scanner.py b/slr_parsing/src/lambdafeedback/slr_parsing/scanner.py new file mode 100644 index 0000000..b7702b8 --- /dev/null +++ b/slr_parsing/src/lambdafeedback/slr_parsing/scanner.py @@ -0,0 +1,127 @@ +"""Scanning: splitting text into tokens with a grammar's token specifications.""" + +from __future__ import annotations + +from collections.abc import Sequence +from typing import Any, Literal + +from .errors import GrammarError, ScanError +from .grammar import ( + CatchAll, + GrammarSymbol, + Lexeme, + Matcher, + TokenMatcher, + TokenPattern, + TokenSpec, + match_pattern, + pattern_weight, +) +from .tokens import Token + + +class Scanner: + """Splits input text, and production strings, into tokens. + + Raises :class:`GrammarError` if a token specification is not a + :class:`TokenPattern`, :class:`TokenMatcher`, :class:`CatchAll` or + :class:`GrammarSymbol`, or if there is more than one :class:`CatchAll`. + """ + + def __init__(self, token_list: Sequence[TokenSpec]) -> None: + _check_token_specs(token_list) + # Heaviest patterns first: scan() keeps the longest match, so this + # order only breaks ties between equal-length matches. + self.token_list: list[TokenSpec] = sorted(token_list, key=lambda spec: -pattern_weight(spec.pattern)) + self._catch_all = next((spec for spec in self.token_list if isinstance(spec, CatchAll)), None) + self._expression_symbols = [ + (spec.pattern, spec.label) for spec in self.token_list if isinstance(spec, TokenPattern) + ] + self._expression_rules: list[tuple[str, Matcher]] = [ + (spec.label, spec.matcher) for spec in self.token_list if isinstance(spec, TokenMatcher) + ] + self._bnf_symbols: list[tuple[str | Lexeme, str]] = [(spec.pattern, spec.label) for spec in self.token_list] + + def scan(self, expr: str, mode: Literal["expression", "bnf"] = "expression") -> list[Token]: + """Split ``expr`` into tokens, preferring the longest match at each position. + + Of equal-length matches, the specification with the longest pattern + string wins; a matcher only wins over a pattern with a strictly + longer match. + + ``mode="bnf"`` scans production strings into grammar symbols: every + token specification is used as a plain pattern and no matchers run. + Text that no specification matches is a :class:`GrammarError` in bnf + mode (the catch-all only applies to expressions). + """ + if mode == "expression": + symbols, rules = self._expression_symbols, self._expression_rules + elif mode == "bnf": + symbols, rules = self._bnf_symbols, [] + else: + raise ValueError(f"Unknown scan mode {mode!r}, expected 'expression' or 'bnf'.") + + tokens: list[Token] = [] + index = 0 + undefined = "" + while index - len(undefined) < len(expr): + label: str | None = None + content: Any = "" + length = 0 + for pattern, symbol_label in symbols: + matched_length = match_pattern(pattern, expr, index) + if matched_length is not None and matched_length > length: + length = matched_length + content = expr[index : index + length] + label = symbol_label + rest = expr[index:] if rules else "" # matchers read the text from where they are tried + for rule_label, matcher in rules: + matched, rule_content = matcher(rest) + if matched is not None and len(matched) > length: + content = rule_content + length = len(matched) + label = rule_label + + token = None + if label is None: + undefined += expr[index] + index += 1 + else: + token = Token(label, content, expr, index, index + length - 1) + if undefined and (token is not None or index >= len(expr)): + if mode == "bnf": + raise GrammarError(f"Undeclared grammar symbol {undefined!r} in {expr!r}.") + tokens.append(self._undefined_token(undefined, expr, index)) + undefined = "" + if token is not None: + tokens.append(token) + index += length + return tokens + + def _undefined_token(self, text: str, expr: str, index: int) -> Token: + if self._catch_all is None: + raise ScanError(f"Undefined input: {text}") + return Token(self._catch_all.label, text, expr, index - len(text), index - 1) + + def grammar_symbol(self, symbol: str, role: str) -> Token: + """The grammar symbol ``symbol`` names; ``role`` (e.g. ``"Start symbol"``) is for the error message. + + Raises :class:`GrammarError` unless ``symbol`` scans as exactly one declared symbol. + """ + try: + tokens = self.scan(symbol, mode="bnf") + except GrammarError: + tokens = [] + if len(tokens) != 1: + raise GrammarError(f"{role} {symbol!r} is not exactly one declared grammar symbol.") + return tokens[0] + + +def _check_token_specs(token_list: Sequence[TokenSpec]) -> None: + for spec in token_list: + if not isinstance(spec, TokenPattern | TokenMatcher | CatchAll | GrammarSymbol): + raise GrammarError( + f"Token specification {spec!r} is not a TokenPattern, TokenMatcher, CatchAll or GrammarSymbol." + ) + if sum(isinstance(spec, CatchAll) for spec in token_list) > 1: + raise GrammarError("Only one token type can be used to catch undefined lexemes.") diff --git a/slr_parsing/src/lambdafeedback/slr_parsing/table.py b/slr_parsing/src/lambdafeedback/slr_parsing/table.py new file mode 100644 index 0000000..871c843 --- /dev/null +++ b/slr_parsing/src/lambdafeedback/slr_parsing/table.py @@ -0,0 +1,336 @@ +"""SLR(1) parse tables: LR(0) states, FIRST and FOLLOW sets, and the table's entries.""" + +from __future__ import annotations + +from collections import deque +from collections.abc import Sequence +from dataclasses import dataclass +from typing import TypeAlias, final + +from .errors import GrammarError +from .grammar import ErrorHandler, TokenProduction +from .tokens import Token + +Item: TypeAlias = tuple[int, int] +"""An LR(0) item: (production index, position of the dot in the body).""" +State: TypeAlias = tuple[Item, ...] +"""A state, identified by its kernel items.""" + + +# ---------------------------------------------------------------------- +# Table entries +# ---------------------------------------------------------------------- + + +@final +@dataclass(frozen=True, slots=True) +class Shift: + """Shift the lookahead and go to ``state`` (for a nonterminal: the goto after a reduction).""" + + state: int + + def __str__(self) -> str: + return f"s{self.state}" + + +@final +@dataclass(frozen=True, slots=True) +class Reduce: + """Reduce by production ``production`` (its index in the grammar).""" + + production: int + + def __str__(self) -> str: + return f"r{self.production}" + + +@final +@dataclass(frozen=True, slots=True) +class Accept: + """Accept the input: a production of the start symbol is complete.""" + + def __str__(self) -> str: + return "acc" + + +@final +@dataclass(frozen=True, slots=True) +class Reject: + """A syntax error that no error handler applies to.""" + + def __str__(self) -> str: + return "-" + + +@final +@dataclass(frozen=True, slots=True) +class Recover: + """A syntax error, handled by error handler ``handler`` (its index in ``error_handlers``).""" + + handler: int + + def __str__(self) -> str: + return f"h{self.handler}" + + +Entry: TypeAlias = Shift | Reduce | Accept | Reject | Recover +"""A parse-table entry; ``str()`` gives its form in :meth:`ParseTable.to_string`. + +The entry classes are final: the parser tells them apart with ``type(entry) is ...``, +which is faster than ``isinstance`` or ``match`` in its innermost loop. +""" + +ACCEPT = Accept() +REJECT = Reject() + + +# ---------------------------------------------------------------------- +# The table +# ---------------------------------------------------------------------- + + +class ParseTable: + """The SLR(1) parse table of a grammar, built from its scanned productions. + + ``rows[state][symbol_id]`` is the :data:`Entry` for a state and a symbol + (see :meth:`entry`). When productions conflict, the one listed last wins; + empty entries go to the first error handler whose condition holds, or + are :class:`Reject`. Productions headed by ``start`` accept rather than + reduce. ε-productions are not supported. + """ + + def __init__( + self, + productions: Sequence[TokenProduction], + start: Token, + end: Token, + null: Token, + error_handlers: Sequence[ErrorHandler] = (), + ) -> None: + self.productions = list(productions) + self.start_token = start + self.end_token = end + self.null_token = null + self._error_handlers = list(error_handlers) + self._closures: dict[State, tuple[Item, ...]] = {} + # Production indices by head, in production order. + self.productions_by_head: dict[Token, list[int]] = {} + for p, (head, _) in enumerate(self.productions): + self.productions_by_head.setdefault(head, []).append(p) + + self._classify_symbols() + self._compute_first() + self._compute_follow() + self._build_states() + self._build_rows() + + # ------------------------------------------------------------------ + # Lookup + # ------------------------------------------------------------------ + + def entry(self, state: int, symbol: Token) -> Entry: + """The entry for ``state`` with ``symbol`` next.""" + return self.rows[state][self.symbol_ids[symbol]] + + def goto(self, state: int, nonterminal: Token) -> int: + """The state to go to from ``state`` after reducing to ``nonterminal``.""" + target = self.gotos[state].get(nonterminal) + if target is None: + raise GrammarError(f"No goto from state {state} on {nonterminal.label}.") + return target + + # ------------------------------------------------------------------ + # Symbols, FIRST and FOLLOW + # ------------------------------------------------------------------ + + def _classify_symbols(self) -> None: + self.nonterminals: list[Token] = [] + for head, _ in self.productions: + if head not in self.nonterminals: + self.nonterminals.append(head) + self.terminals: list[Token] = [self.end_token, self.null_token] + for _, body in self.productions: + for token in body: + if token not in self.terminals and token not in self.nonterminals: + self.terminals.append(token) + self.symbols = self.terminals + self.nonterminals + self.symbol_ids = {symbol: j for j, symbol in enumerate(self.symbols)} + + def _compute_first(self) -> None: + """FIRST sets of single symbols (ε-productions are not supported).""" + # Dicts as ordered sets: membership is constant-time and the order stays deterministic. + first: dict[Token, dict[Token, None]] = {t: {t: None} for t in self.terminals} + first.update({nt: {} for nt in self.nonterminals}) + changed = True + while changed: + changed = False + for nt in self.nonterminals: + for p in self.productions_by_head[nt]: + for token in self.productions[p][1]: + for x in first[token]: + if x not in first[nt]: + first[nt][x] = None + changed = True + if self.null_token not in first[token]: + break + self.first: dict[Token, list[Token]] = {symbol: list(symbols) for symbol, symbols in first.items()} + + def first_of(self, tokens: Sequence[Token]) -> list[Token]: + """FIRST set of a string of symbols.""" + if len(tokens) == 1: + return self.first[tokens[0]] + result: list[Token] = [] + for token in tokens: + for item in self.first[token]: + if item not in result and item != self.null_token: + result.append(item) + if token != self.null_token: + break + return result + + def _compute_follow(self) -> None: + follow: dict[Token, dict[Token, None]] = {nt: {} for nt in self.nonterminals} + follow[self.start_token][self.end_token] = None + changed = True + while changed: + changed = False + for head, body in self.productions: + for k, token in enumerate(body): + if token not in follow: + continue + rest = body[k + 1 :] + rest_first = self.first_of(rest) if rest else [] + additions = rest_first + if not rest or self.null_token in rest_first: + additions = [*follow[head], *rest_first] + for item in additions: + if item != self.null_token and item not in follow[token]: + follow[token][item] = None + changed = True + self.follow: dict[Token, list[Token]] = {symbol: list(symbols) for symbol, symbols in follow.items()} + + # ------------------------------------------------------------------ + # LR(0) states + # ------------------------------------------------------------------ + + def closure(self, item_set: Sequence[Item]) -> tuple[Item, ...]: + """The LR(0) closure of ``item_set`` (cached).""" + key = tuple(item_set) + cached = self._closures.get(key) + if cached is not None: + return cached + closure = list(item_set) + expanded: set[Token] = set() # nonterminals whose productions are already in the closure + for i, j in closure: # the loop also visits the items it appends + body = self.productions[i][1] + if j < len(body) and body[j] in self.productions_by_head and body[j] not in expanded: + expanded.add(body[j]) + closure += [(p, 0) for p in self.productions_by_head[body[j]]] + self._closures[key] = result = tuple(closure) + return result + + def _compute_transitions(self, item_set: Sequence[Item]) -> list[tuple[Token, list[Item]]]: + transitions: dict[Token, list[Item]] = {} # in order of first appearance + for i, j in item_set: + body = self.productions[i][1] + if j < len(body): + transitions.setdefault(body[j], []).append((i, j + 1)) + return list(transitions.items()) + + def _build_states(self) -> None: + start: State = tuple((p, 0) for p, (head, _) in enumerate(self.productions) if head == self.start_token) + self.states: list[State] = [start] + self.state_ids: dict[State, int] = {start: 0} + self._transitions: dict[State, list[tuple[Token, list[Item]]]] = {} + queue = deque([start]) + while queue: + state = queue.popleft() + transitions = self._compute_transitions(self.closure(state)) + self._transitions[state] = transitions + for _, kernel in transitions: + target = tuple(kernel) + if target not in self.state_ids: + self.state_ids[target] = len(self.states) + self.states.append(target) + queue.append(target) + + # ------------------------------------------------------------------ + # Entries + # ------------------------------------------------------------------ + + def _build_rows(self) -> None: + # (production index, entry) candidates per state and symbol; _resolve picks one. + candidates: list[list[list[tuple[int, Entry]]]] = [[[] for _ in self.symbols] for _ in self.states] + for s, state in enumerate(self.states): + targets = {symbol: self.state_ids[tuple(kernel)] for symbol, kernel in self._transitions[state]} + for production_index, dot in self.closure(state): + head, body = self.productions[production_index] + if dot < len(body): # shift / goto + symbol = body[dot] + candidates[s][self.symbol_ids[symbol]].append((production_index, Shift(targets[symbol]))) + elif head == self.start_token: # accept + candidates[s][self.symbol_ids[self.end_token]].append((production_index, ACCEPT)) + else: # reduce + for symbol in self.follow[head]: + candidates[s][self.symbol_ids[symbol]].append((production_index, Reduce(production_index))) + + self.rows: list[list[Entry]] = [ + [self._resolve(s, j, entries) for j, entries in enumerate(row)] for s, row in enumerate(candidates) + ] + self.gotos: list[dict[Token, int]] = [ + {nt: entry.state for nt in self.nonterminals if isinstance(entry := row[self.symbol_ids[nt]], Shift)} + for row in self.rows + ] + """``gotos[state][nonterminal]``: the state after reducing to ``nonterminal`` (the :class:`Shift` entries).""" + + def _resolve(self, s: int, j: int, entries: list[tuple[int, Entry]]) -> Entry: + """Pick the entry for state ``s`` and symbol ``j``. + + Conflicts go to the production listed last (ties to the first + candidate); empty entries become errors, handled by the first error + handler whose condition holds. + """ + if not entries: + items = [(self.productions[i][1][:dot], self.productions[i][1][dot:]) for i, dot in self.states[s]] + next_symbol = self.symbols[j] + for k, handler in enumerate(self._error_handlers): + if handler.condition(items, next_symbol): + return Recover(k) + return REJECT + best = entries[0] + for entry in entries[1:]: + if best[0] < entry[0]: + best = entry + return best[1] + + def unreachable(self) -> tuple[list[str], list[str]]: + """The states no entry leads to and the productions never reduced by (as strings), for diagnostics.""" + targets = {entry for row in self.rows for entry in row} + states = [self.state_string(self.states[s]) for s in range(1, len(self.states)) if Shift(s) not in targets] + reductions = [ + f"{head.content} --> {''.join(token.content for token in body)}" + for p, (head, body) in enumerate(self.productions) + if head != self.start_token and Reduce(p) not in targets + ] + return states, reductions + + # ------------------------------------------------------------------ + # Debugging aids + # ------------------------------------------------------------------ + + def _item_strings(self, state: State) -> list[str]: + strings = [] + for production_index, dot in self.closure(state): + body = self.productions[production_index][1] + strings.append("".join(x.content for x in body[:dot]) + "." + "".join(x.content for x in body[dot:])) + return strings + + def state_string(self, state: State) -> str: + return f"I{self.state_ids[state]}: ({', '.join(self._item_strings(state))})" + + def to_string(self) -> str: + """The table, tab-separated: ``s`` (shift), ``acc``, ``r``, ``h`` or ``-``.""" + lines = ["\t" + "\t".join(x.content for x in self.symbols)] + for s, row in enumerate(self.rows): + lines.append(f"{s}\t" + "\t".join(str(entry) for entry in row)) + return "\n".join(lines) + "\n" diff --git a/slr_parsing/tests/test_engine.py b/slr_parsing/tests/test_engine.py index 729abdb..d8abf8a 100644 --- a/slr_parsing/tests/test_engine.py +++ b/slr_parsing/tests/test_engine.py @@ -9,6 +9,7 @@ import pytest from lambdafeedback.slr_parsing import ( + Accept, CatchAll, ErrorHandler, GrammarError, @@ -74,14 +75,14 @@ def parse(text, parser=None): class TestTableConstruction: def test_first_and_follow_sets(self): - # Checked through private attributes: these are the core of SLR(1). - parser = make_parser() + # These are the core of SLR(1). + table = make_parser().table def labels(tokens): return {t.label for t in tokens} - first = {nt.label: labels(parser._first_dict[nt]) for nt in parser._nonterminals} - follow = {nt.label: labels(parser._follow[nt]) for nt in parser._nonterminals} + first = {nt.label: labels(table.first[nt]) for nt in table.nonterminals} + follow = {nt.label: labels(table.follow[nt]) for nt in table.nonterminals} assert first == {"START": {"LPAR", "ID"}, "E": {"LPAR", "ID"}, "T": {"LPAR", "ID"}, "F": {"LPAR", "ID"}} assert follow == { "START": {"END"}, @@ -230,7 +231,7 @@ def test_parsing_table_string_tells_accept_from_reductions(): rows = [line.split("\t")[1:] for line in parser.parsing_table_to_string().splitlines()[1:]] for table_row, printed_row in zip(parser.parsing_table, rows, strict=True): for entry, cell in zip(table_row, printed_row, strict=True): - assert (cell == "acc") == (entry == len(parser.states)) + assert (cell == "acc") == isinstance(entry, Accept) def test_walk_lists_each_node_before_its_children(): diff --git a/units/src/lambdafeedback/units/parser.py b/units/src/lambdafeedback/units/parser.py index 573d5a5..f561d16 100644 --- a/units/src/lambdafeedback/units/parser.py +++ b/units/src/lambdafeedback/units/parser.py @@ -110,44 +110,66 @@ def unit_dictionaries(unit_sets: frozenset[str], strictness: Strictness) -> Unit ) +_LEAF_TAGS: dict[str, QuantityTag] = { + Label.UNIT: QuantityTag.UNIT, + Label.NUMBER: QuantityTag.NUMBER, + Label.NON_UNIT: QuantityTag.NON_UNIT, +} +"""The tag each kind of leaf (a scanned token) carries.""" + +_NOT_A_UNIT = frozenset({QuantityTag.NUMBER, QuantityTag.NON_UNIT, QuantityTag.REJECTED_UNIT}) +"""Tags that stop an operator's result from being a unit.""" + + +def _is_unit_power(node: ExprNode) -> bool: + """``m^2``: a unit to a numeric power, which is a unit.""" + base, exponent = node.children + return QuantityTag.UNIT in base.tags and exponent.tags == {QuantityTag.NUMBER} + + +def _is_reciprocal_unit(node: ExprNode) -> bool: + """``1/s``: one over a unit, which is a unit.""" + numerator, denominator = node.children + return numerator.content == "1" and denominator.tags == {QuantityTag.UNIT} + + +def _is_number_ratio(node: ExprNode) -> bool: + """``3/4``: a number over a number, which stays a number.""" + numerator, denominator = node.children + return numerator.tags == {QuantityTag.NUMBER} == denominator.tags + + def quantity_tag_handler(strictness: Strictness) -> Callable[[ExprNode], set[Any]]: """Tag handler deciding which subtrees are units, numbers or other (value) text.""" def tag_handler(node: ExprNode) -> set[Any]: tags: set[Any] = set().union(*(child.tags for child in node.children)) - children = node.children - if node.label == Label.UNIT: - tags.add(QuantityTag.UNIT) - elif node.label == Label.NUMBER: - tags.add(QuantityTag.NUMBER) - elif node.label == Label.NON_UNIT: - tags.add(QuantityTag.NON_UNIT) - elif ( - node.label == Label.POWER - and QuantityTag.UNIT in children[0].tags - and children[1].tags == {QuantityTag.NUMBER} - ): - tags.remove(QuantityTag.NUMBER) # a unit to a numeric power is a unit - elif node.label == Label.SOLIDUS and children[0].content == "1" and children[1].tags == {QuantityTag.UNIT}: - tags.remove(QuantityTag.NUMBER) # 1/unit is a unit - elif node.label == Label.SOLIDUS and children[0].tags == {QuantityTag.NUMBER} == children[1].tags: - pass # number/number stays a number - elif node.label in (Label.PRODUCT, Label.SOLIDUS, Label.POWER): - if tags & {QuantityTag.NUMBER, QuantityTag.NON_UNIT, QuantityTag.REJECTED_UNIT}: + # The cases are tried in order: a case whose guard fails falls through to the next. + match node.label: + case Label.UNIT | Label.NUMBER | Label.NON_UNIT: + tags.add(_LEAF_TAGS[node.label]) + case Label.POWER if _is_unit_power(node): + tags.remove(QuantityTag.NUMBER) + case Label.SOLIDUS if _is_reciprocal_unit(node): + tags.remove(QuantityTag.NUMBER) + case Label.SOLIDUS if _is_number_ratio(node): + pass + case Label.PRODUCT | Label.SOLIDUS | Label.POWER if tags & _NOT_A_UNIT: tags.discard(QuantityTag.UNIT) if QuantityTag.NUMBER in tags: tags.add(QuantityTag.NON_UNIT) - elif node.label == Label.SPACE and QuantityTag.NON_UNIT in children[1].tags: - tags.discard(QuantityTag.UNIT) - elif node.label == GROUP_LABEL and not node.content: - # Juxtaposition (a group without delimiters) - if strictness == "strict": - node.children = [remove_tag(child, QuantityTag.UNIT) for child in children] - if QuantityTag.UNIT in tags: - tags.remove(QuantityTag.UNIT) - tags.add(QuantityTag.REJECTED_UNIT) - elif any(QuantityTag.NON_UNIT in child.tags for child in children): + case Label.SPACE if QuantityTag.NON_UNIT in node.children[1].tags: tags.discard(QuantityTag.UNIT) + # GROUP_LABEL is a bare name, so "case GROUP_LABEL:" would capture (match anything), not compare. + case label if label == GROUP_LABEL and not node.content: + # Juxtaposition (a group without delimiters) + if strictness == "strict": + node.children = [remove_tag(child, QuantityTag.UNIT) for child in node.children] + if QuantityTag.UNIT in tags: + tags.remove(QuantityTag.UNIT) + tags.add(QuantityTag.REJECTED_UNIT) + elif any(QuantityTag.NON_UNIT in child.tags for child in node.children): + tags.discard(QuantityTag.UNIT) return tags return tag_handler diff --git a/units/src/lambdafeedback/units/quantity.py b/units/src/lambdafeedback/units/quantity.py index 86f1b9f..b7ab7a4 100644 --- a/units/src/lambdafeedback/units/quantity.py +++ b/units/src/lambdafeedback/units/quantity.py @@ -3,6 +3,7 @@ from __future__ import annotations import logging +from collections.abc import Callable from dataclasses import dataclass from functools import cache from typing import Any @@ -151,6 +152,70 @@ def _unsplittable_names(params: QuantityParams, all_forms: bool) -> list[str]: return sorted(names) +# ---------------------------------------------------------------------- +# LaTeX for units: one renderer per node label (see _UNIT_LATEX) +# ---------------------------------------------------------------------- + + +def _unit_latex(node: ExprNode) -> list[str]: + """LaTeX for a unit subtree, as pieces to join.""" + render = _UNIT_LATEX.get(node.label, _plain_latex) + return render(node) + + +def _braced_latex(node: ExprNode) -> list[str]: + """LaTeX for a node already grouped by braces (a numerator, denominator or exponent): no brackets of its own.""" + while node.label == GROUP_LABEL and len(node.children) == 1: # (x) or ((x)), not juxtaposition + node = node.children[0] + return _unit_latex(node) + + +def _plain_latex(node: ExprNode) -> list[str]: + """A number, or any other node without a renderer of its own: its content.""" + return [node.content] + + +def _unit_name_latex(node: ExprNode) -> list[str]: + return ["\\mathrm{", node.content, "}"] + + +def _product_latex(node: ExprNode) -> list[str]: + left, right = node.children + return [*_unit_latex(left), "\\cdot", *_unit_latex(right)] + + +def _space_latex(node: ExprNode) -> list[str]: + left, right = node.children + return [*_unit_latex(left), "~", *_unit_latex(right)] + + +def _power_latex(node: ExprNode) -> list[str]: + base, exponent = node.children + return [*_unit_latex(base), "^{", *_braced_latex(exponent), "}"] + + +def _solidus_latex(node: ExprNode) -> list[str]: + numerator, denominator = node.children + return ["\\frac{", *_braced_latex(numerator), "}{", *_braced_latex(denominator), "}"] + + +def _group_latex(node: ExprNode) -> list[str]: + if node.delimiters is None: + return _plain_latex(node) + opening, closing = node.delimiters + return [opening.content, *(part for child in node.children for part in _unit_latex(child)), closing.content] + + +_UNIT_LATEX: dict[str, Callable[[ExprNode], list[str]]] = { + Label.UNIT: _unit_name_latex, + Label.PRODUCT: _product_latex, + Label.SPACE: _space_latex, + Label.POWER: _power_latex, + Label.SOLIDUS: _solidus_latex, + GROUP_LABEL: _group_latex, +} + + class PhysicalQuantity: """A response read as a quantity: its value and unit, their LaTeX, and SI forms. @@ -213,7 +278,7 @@ def __init__(self, name: str, params: QuantityParams, ast_root: ExprNode, parser if self.value is not None: self._revert_value_text(self.value) self.value_latex = self._value_latex() - self.unit_latex = "".join(self._unit_latex(self.unit)) if self.unit is not None else None + self.unit_latex = "".join(_unit_latex(self.unit)) if self.unit is not None else None separator = "~" if self.value_latex is not None and self.unit_latex is not None else "" self.latex = (self.value_latex or "") + separator + (self.unit_latex or "") self._set_sympy_forms() @@ -350,29 +415,6 @@ def _value_latex(self) -> str | None: return "\\left\\{" + ",~".join(parts) + "\\right\\}" return sympy_to_latex(parsed[0], params.symbols, settings=settings) - def _unit_latex(self, node: ExprNode) -> list[str]: - children = node.children - if node.label == Label.PRODUCT: - return [*self._unit_latex(children[0]), "\\cdot", *self._unit_latex(children[1])] - if node.label == Label.SPACE: - return [*self._unit_latex(children[0]), "~", *self._unit_latex(children[1])] - if node.label == Label.UNIT: - return ["\\mathrm{", node.content, "}"] - if node.label == GROUP_LABEL and node.delimiters is not None: - opening, closing = node.delimiters - return [opening.content, *(part for child in children for part in self._unit_latex(child)), closing.content] - if node.label == Label.POWER: - return [*self._unit_latex(children[0]), "^{", *self._braced_latex(children[1]), "}"] - if node.label == Label.SOLIDUS: - return ["\\frac{", *self._braced_latex(children[0]), "}{", *self._braced_latex(children[1]), "}"] - return [node.content] # NUMBER and anything else - - def _braced_latex(self, node: ExprNode) -> list[str]: - """LaTeX for a node already grouped by braces (a numerator, denominator or exponent): no brackets of its own.""" - while node.label == GROUP_LABEL and len(node.children) == 1: # (x) or ((x)), not juxtaposition - node = node.children[0] - return self._unit_latex(node) - # ------------------------------------------------------------------ # Conversion to SI base units # ------------------------------------------------------------------ diff --git a/units/tests/test_regressions.py b/units/tests/test_regressions.py index 25ee3f2..52bf86f 100644 --- a/units/tests/test_regressions.py +++ b/units/tests/test_regressions.py @@ -3,7 +3,7 @@ import pytest from sympy import simplify -from lambdafeedback.units import DisallowedUnitError, QuantityParams, QuantityParseError, parse_quantity +from lambdafeedback.units import DisallowedUnitError, QuantityParams, QuantityParseError, RevertedUnit, parse_quantity NATURAL = QuantityParams(strictness="natural") @@ -157,6 +157,21 @@ def test_unit_between_value_parts_raises_instead_of_recursing_forever(self): with pytest.raises(QuantityParseError, match="Cannot separate the value from the unit"): parse("5 m/s x") + @pytest.mark.parametrize( + ("expr", "reverted"), + [ + ("kg 2x", [("", "kg", " 2x")]), + ("m 3y", [("", "m", " 3y")]), + ("kg m 2x", [("kg", "kilogram metre", "m 2x"), ("", "kg", " m 2x"), ("kg ", "m", " 2x")]), + ], + ) + def test_strict_unit_before_juxtaposed_value_is_reverted_once(self, expr, reverted): + # Splitting rotates the juxtaposition "2x" to the root; retagging it must strip the unit + # from the "kg 2" it now holds, or that span is reported as a unit too ("kilogram 2"). + quantity = parse(expr, QuantityParams(strictness="strict")) + assert quantity.value.content_string() == expr and quantity.unit is None + assert quantity.reverted_units == [RevertedUnit(*parts) for parts in reverted] + class TestPickling: def test_params_error_round_trips(self):