Use keyword-only arguments for AST node dataclasses

Also use `None` as a default value for span, to avoid having to type
it when constructing higher level formulas.

Signed-off-by: Sahas Subramanian <[email protected]>

Author: shsms

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

@@ -21,11 +21,11 @@
 _logger = logging.getLogger(__name__)
 
 
-@dataclass
+@dataclass(kw_only=True)
 class Node(abc.ABC):
     """An abstract syntax tree node representing a formula expression."""
 
-    span: tuple[int, int] | None
+    span: tuple[int, int] | None = None
 
     @abc.abstractmethod
     def evaluate(self) -> float | None:
@@ -41,7 +41,7 @@ def __str__(self) -> str:
         return self.format()
 
 
-@dataclass
+@dataclass(kw_only=True)
 class TelemetryStream(Node, Generic[QuantityT]):
     """A AST node that retrieves values from a component's telemetry stream."""
 
@@ -73,7 +73,7 @@ async def fetch_next(self) -> None:
         self._latest_sample = await anext(self.stream)
 
 
-@dataclass
+@dataclass(kw_only=True)
 class FunCall(Node):
     """A function call in the formula."""
 
@@ -92,7 +92,7 @@ def format(self, wrap: bool = False) -> str:
         return f"{self.function.name}({args_str})"
 
 
-@dataclass
+@dataclass(kw_only=True)
 class Constant(Node):
     """A constant numerical value in the formula."""
 
@@ -109,7 +109,7 @@ def format(self, wrap: bool = False) -> str:
         return str(self.value)
 
 
-@dataclass
+@dataclass(kw_only=True)
 class Add(Node):
     """Addition operation node."""
 
@@ -134,7 +134,7 @@ def format(self, wrap: bool = False) -> str:
         return expr
 
 
-@dataclass
+@dataclass(kw_only=True)
 class Sub(Node):
     """Subtraction operation node."""
 
@@ -159,7 +159,7 @@ def format(self, wrap: bool = False) -> str:
         return expr
 
 
-@dataclass
+@dataclass(kw_only=True)
 class Mul(Node):
     """Multiplication operation node."""
 
@@ -181,7 +181,7 @@ def format(self, wrap: bool = False) -> str:
         return f"{self.left.format(True)} * {self.right.format(True)}"
 
 
-@dataclass
+@dataclass(kw_only=True)
 class Div(Node):
     """Division operation node."""
 

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

@@ -177,9 +177,8 @@ def __init__(
 
         if isinstance(formula, Formula):
             self.root: _ast.Node = _ast.TelemetryStream(
-                None,
-                str(formula),
-                formula.new_receiver(),
+                source=str(formula),
+                stream=formula.new_receiver(),
             )
             self._streams.append(self.root)
             self._sub_formulas.append(formula)
@@ -195,13 +194,16 @@ def __add__(
             right_node = other.root
             self._streams.extend(other._streams)
         elif isinstance(other, Formula):
-            right_node = _ast.TelemetryStream(None, str(other), other.new_receiver())
+            right_node = _ast.TelemetryStream(
+                source=str(other),
+                stream=other.new_receiver(),
+            )
             self._streams.append(right_node)
             self._sub_formulas.append(other)
         else:
-            right_node = _ast.Constant(None, other.base_value)
+            right_node = _ast.Constant(value=other.base_value)
 
-        new_root = _ast.Add(None, self.root, right_node)
+        new_root = _ast.Add(left=self.root, right=right_node)
         return FormulaBuilder(
             new_root,
             self._create_method,
@@ -218,13 +220,15 @@ def __sub__(
             right_node = other.root
             self._streams.extend(other._streams)
         elif isinstance(other, Formula):
-            right_node = _ast.TelemetryStream(None, str(other), other.new_receiver())
+            right_node = _ast.TelemetryStream(
+                source=str(other), stream=other.new_receiver()
+            )
             self._streams.append(right_node)
             self._sub_formulas.append(other)
         else:
-            right_node = _ast.Constant(None, other.base_value)
+            right_node = _ast.Constant(value=other.base_value)
 
-        new_root = _ast.Sub(None, self.root, right_node)
+        new_root = _ast.Sub(left=self.root, right=right_node)
         return FormulaBuilder(
             new_root,
             self._create_method,
@@ -234,8 +238,8 @@ def __sub__(
 
     def __mul__(self, other: float) -> FormulaBuilder[QuantityT]:
         """Create a multiplication operation node."""
-        right_node = _ast.Constant(None, other)
-        new_root = _ast.Mul(None, self.root, right_node)
+        right_node = _ast.Constant(value=other)
+        new_root = _ast.Mul(left=self.root, right=right_node)
         return FormulaBuilder(
             new_root,
             self._create_method,
@@ -248,8 +252,8 @@ def __truediv__(
         other: float,
     ) -> FormulaBuilder[QuantityT]:
         """Create a division operation node."""
-        right_node = _ast.Constant(None, other)
-        new_root = _ast.Div(None, self.root, right_node)
+        right_node = _ast.Constant(value=other)
+        new_root = _ast.Div(left=self.root, right=right_node)
         return FormulaBuilder(
             new_root,
             self._create_method,
@@ -269,20 +273,18 @@ def coalesce(
                 self._streams.extend(item._streams)  # pylint: disable=protected-access
             elif isinstance(item, Formula):
                 right_node = _ast.TelemetryStream(
-                    None,
-                    str(item),
-                    item.new_receiver(),
+                    source=str(item),
+                    stream=item.new_receiver(),
                 )
                 right_nodes.append(right_node)
                 self._streams.append(right_node)
                 self._sub_formulas.append(item)
             else:
-                right_nodes.append(_ast.Constant(None, item.base_value))
+                right_nodes.append(_ast.Constant(value=item.base_value))
 
         new_root = _ast.FunCall(
-            None,
-            Coalesce(),
-            [self.root] + right_nodes,
+            function=Coalesce(),
+            args=[self.root] + right_nodes,
         )
 
         return FormulaBuilder(
@@ -304,20 +306,18 @@ def min(
                 self._streams.extend(item._streams)  # pylint: disable=protected-access
             elif isinstance(item, Formula):
                 right_node = _ast.TelemetryStream(
-                    None,
-                    str(item),
-                    item.new_receiver(),
+                    source=str(item),
+                    stream=item.new_receiver(),
                 )
                 right_nodes.append(right_node)
                 self._streams.append(right_node)
                 self._sub_formulas.append(item)
             else:
-                right_nodes.append(_ast.Constant(None, item.base_value))
+                right_nodes.append(_ast.Constant(value=item.base_value))
 
         new_root = _ast.FunCall(
-            None,
-            Min(),
-            [self.root] + right_nodes,
+            function=Min(),
+            args=[self.root] + right_nodes,
         )
 
         return FormulaBuilder(
@@ -339,20 +339,18 @@ def max(
                 self._streams.extend(item._streams)  # pylint: disable=protected-access
             elif isinstance(item, Formula):
                 right_node = _ast.TelemetryStream(
-                    None,
-                    str(item),
-                    item.new_receiver(),
+                    source=str(item),
+                    stream=item.new_receiver(),
                 )
                 right_nodes.append(right_node)
                 self._streams.append(right_node)
                 self._sub_formulas.append(item)
             else:
-                right_nodes.append(_ast.Constant(None, item.base_value))
+                right_nodes.append(_ast.Constant(value=item.base_value))
 
         new_root = _ast.FunCall(
-            None,
-            Max(),
-            [self.root] + right_nodes,
+            function=Max(),
+            args=[self.root] + right_nodes,
         )
 
         return FormulaBuilder(

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

@@ -44,19 +44,16 @@ def __init__(
         self._phase_3_formula: Formula[QuantityT] = phase_3
         self._components: list[_ast.TelemetryStream[QuantityT]] = [
             _ast.TelemetryStream(
-                None,
-                "phase_1",
-                phase_1.new_receiver(),
+                source="phase_1",
+                stream=phase_1.new_receiver(),
             ),
             _ast.TelemetryStream(
-                None,
-                "phase_2",
-                phase_2.new_receiver(),
+                source="phase_2",
+                stream=phase_2.new_receiver(),
             ),
             _ast.TelemetryStream(
-                None,
-                "phase_3",
-                phase_3.new_receiver(),
+                source="phase_3",
+                stream=phase_3.new_receiver(),
             ),
         ]
         self._output_channel: Broadcast[Sample3Phase[QuantityT]] = output_channel