Improve formula validation (#1359)

This PR improves formula validation and adds test coverage for all
parser and lexer validation errors.

### Changes

1. Switch to conventional token span start/end semantics: [start, end).
Spans are only used internally now, though (see (2)).
2. Introduced `FormulaError` and `FormulaSyntaxError` classes.
`FormulaSyntaxError.__str__()` produces a human-readable error message,
highlighting the offending span in the context of the original formula.
This should be fit for non-technical end users.
3. Removed `ASTNode.span` property as it was unused.
4. Added tests and adapted existing ones to the changes described above.

Author: simonvoelcker

RELEASE_NOTES.md

@@ -14,4 +14,4 @@
 
 ## Bug Fixes
 
-<!-- Here goes notable bug fixes that are worth a special mention or explanation -->
+- Improved formula validation: Consistent error messages for invalid formulas and conventional span semantics.

src/frequenz/sdk/timeseries/formulas/_base_ast_node.py

@@ -24,9 +24,6 @@
 class AstNode(abc.ABC, Generic[QuantityT]):
     """An abstract syntax tree node representing a formula expression."""
 
-    span: tuple[int, int] | None = None
-    """The span (start, end) of the expression in the input string."""
-
     @abc.abstractmethod
     async def evaluate(self) -> Sample[QuantityT] | QuantityT | None:
         """Evaluate the expression and return its numerical value."""

src/frequenz/sdk/timeseries/formulas/_exceptions.py

@@ -0,0 +1,53 @@
+# License: MIT
+# Copyright © 2026 Frequenz Energy-as-a-Service GmbH
+
+"""Exception classes for formulas parser and evaluator."""
+
+
+class FormulaError(Exception):
+    """Base class for exceptions in this module."""
+
+
+class FormulaSyntaxError(FormulaError):
+    """Exception raised for syntax errors in the formula."""
+
+    def __init__(
+        self, *, formula: str, span: tuple[int, int] | None, message: str
+    ) -> None:
+        """Initialize this instance."""
+        self._formula: str = formula
+        self._span: tuple[int, int] | None = span
+        self._message: str = message
+
+    def __str__(self) -> str:
+        """Return a string representation of the error.
+
+        The span (if present) will be used to highlight the error location.
+        For long formulas, the beginning and/or end will be truncated as needed.
+        """
+        if self._span is None:
+            return self._message
+
+        formula_line = self._formula
+        span_padding = " " * self._span[0]
+        span_highlight = "^" * (self._span[1] - self._span[0])
+        error_line = f"{span_padding}{span_highlight} {self._message}"
+
+        char_limit: int = 80 - 11  # Account for "  Formula: " prefix in final output
+
+        if len(error_line) > char_limit:
+            # Remove characters from the left to fit error line within char limit
+            num_chars_to_truncate = len(error_line) - char_limit
+            error_line = error_line[num_chars_to_truncate:]
+            # Shift formula line by the same amount and insert leading ellipsis
+            formula_line = f"... {formula_line[num_chars_to_truncate+4:]}"
+
+        if len(formula_line) > char_limit:
+            # Truncate the formula line, insert ellipsis
+            formula_line = f"{formula_line[:char_limit-4]} ..."
+
+        return (
+            "Formula syntax error:\n"
+            f"  Formula: {formula_line}\n"
+            f"           {error_line}"
+        )

src/frequenz/sdk/timeseries/formulas/_functions.py

@@ -85,19 +85,14 @@ async def unsubscribe(self) -> None:
         )
 
     @classmethod
-    def from_string(
-        cls, name: str, params: list[AstNode[QuantityT]]
-    ) -> Function[QuantityT]:
-        """Create a function instance from its name."""
-        match name.upper():
-            case "COALESCE":
-                return Coalesce(params)
-            case "MAX":
-                return Max(params)
-            case "MIN":
-                return Min(params)
-            case _:
-                raise ValueError(f"Unknown function name: {name}")
+    def function_class_by_name(cls, name: str) -> type[Function[QuantityT]] | None:
+        """Return the function class corresponding to the given name."""
+        known_functions: dict[str, type[Function[QuantityT]]] = {
+            "COALESCE": Coalesce,
+            "MAX": Max,
+            "MIN": Min,
+        }
+        return known_functions.get(name.upper())
 
 
 @dataclass

src/frequenz/sdk/timeseries/formulas/_lexer.py

@@ -11,6 +11,7 @@
 from typing_extensions import override
 
 from . import _token
+from ._exceptions import FormulaSyntaxError
 from ._peekable import Peekable
 
 
@@ -74,53 +75,58 @@ def __next__(self) -> _token.Token:  # pylint: disable=too-many-branches
             _ = next(self._iter)  # consume '#'
             comp_id = self._read_integer()
             if not comp_id:
-                raise ValueError(f"Expected integer after '#' at position {pos}")
-            end_pos = pos + len(comp_id)
+                raise FormulaSyntaxError(
+                    formula=self._formula,
+                    span=(pos + 1, pos + 2),
+                    message="Expected integer after '#'",
+                )
+            end_pos = pos + len(comp_id) + 1  # account for '#'
             return _token.Component(
-                span=(
-                    pos + 1,
-                    end_pos + 1,  # account for '#'
-                ),
+                span=(pos, end_pos),
                 id=comp_id,
                 value=self._formula[pos:end_pos],
             )
 
         if char == "+":
             _, char = next(self._iter)  # consume operator
-            return _token.Plus(span=(pos + 1, pos + 1), value=char)
+            return _token.Plus(span=(pos, pos + 1), value=char)
 
         if char == "-":
             _, char = next(self._iter)
-            return _token.Minus(span=(pos + 1, pos + 1), value=char)
+            return _token.Minus(span=(pos, pos + 1), value=char)
 
         if char == "*":
             _, char = next(self._iter)
-            return _token.Mul(span=(pos + 1, pos + 1), value=char)
+            return _token.Mul(span=(pos, pos + 1), value=char)
 
         if char == "/":
             _, char = next(self._iter)
-            return _token.Div(span=(pos + 1, pos + 1), value=char)
+            return _token.Div(span=(pos, pos + 1), value=char)
 
         if char == "(":
             _, char = next(self._iter)
-            return _token.OpenParen(span=(pos + 1, pos + 1), value=char)
+            return _token.OpenParen(span=(pos, pos + 1), value=char)
 
         if char == ")":
             _, char = next(self._iter)
-            return _token.CloseParen(span=(pos + 1, pos + 1), value=char)
+            return _token.CloseParen(span=(pos, pos + 1), value=char)
 
         if char == ",":
             _, char = next(self._iter)
-            return _token.Comma(span=(pos + 1, pos + 1), value=char)
+            return _token.Comma(span=(pos, pos + 1), value=char)
 
         if char.isdigit():
             num = self._read_number()
             end_pos = pos + len(num)
-            return _token.Number(span=(pos + 1, end_pos), value=num)
+            return _token.Number(span=(pos, end_pos), value=num)
 
         if char.isalpha():
             symbol = self._read_symbol()
             end_pos = pos + len(symbol)
-            return _token.Symbol(span=(pos + 1, end_pos), value=symbol)
+            return _token.Symbol(span=(pos, end_pos), value=symbol)
 
-        raise ValueError(f"Unexpected character '{char}' at position {pos}")
+        raise FormulaSyntaxError(
+            formula=self._formula,
+            span=(pos, pos + 1),
+            message="Unexpected character",
+        )