Remove custom formula channels and receivers

We will use regular broadcast channels, and provide composability on
FormulaEngine instances instead.

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

Author: shsms

src/frequenz/sdk/timeseries/_formula_engine/_formula_engine.py

@@ -416,192 +416,6 @@ def build(self) -> FormulaEngine:
 )
 
 
-class _BaseFormulaChannel(
-    Generic[_GenericSample, _GenericEngine],
-    Broadcast[_GenericSample],
-    ABC,
-):
-    """A broadcast channel implementation for use with formulas."""
-
-    ReceiverType: Type[FormulaReceiver | FormulaReceiver3Phase]
-
-    def __init__(
-        self, name: str, engine: _GenericEngine, resend_latest: bool = False
-    ) -> None:
-        """Create a `FormulaChannel` instance.
-
-        Args:
-            name: A name for the channel.
-            engine: A FormulaEngine instance that produces values for this channel.
-            resend_latest: Whether to resend latest channel values to newly created
-                receivers, like in `Broadcast` channels.
-        """
-        self._engine: _GenericEngine = engine
-        super().__init__(name, resend_latest)
-
-    @property
-    def engine(self) -> _GenericEngine:
-        """Return the formula engine attached to the channel.
-
-        Returns:
-            A FormulaEngine instance.
-        """
-        return self._engine
-
-    def new_receiver(
-        self, name: Optional[str] = None, maxsize: int = 50
-    ) -> _GenericFormulaReceiver:
-        """Create a new FormulaReceiver for the channel.
-
-        This implementation is similar to `Broadcast.new_receiver()`, except that it
-        creates and returns a `FormulaReceiver`.  The way the default name for the
-        receiver is constructed, is also slightly tweaked.
-
-        Args:
-            name: An optional name for the receiver.
-            maxsize: size of the receiver's buffer.
-
-        Returns:
-            A `FormulaReceiver` instance attached to the `FormulaChannel`.
-        """
-        uuid = uuid4()
-        if name is None:
-            name = self.name
-        recv = self.ReceiverType(uuid, name, maxsize, self)
-        self.receivers[uuid] = weakref.ReferenceType(recv)
-        if self._resend_latest and self._latest is not None:
-            recv.enqueue(self._latest)
-        return recv
-
-
-class _BaseFormulaReceiver(
-    Generic[_GenericSample, _GenericEngine],
-    BroadcastReceiver[_GenericSample],
-    ABC,
-):
-    """A receiver to receive calculated `Sample`s from a Formula channel.
-
-    They function as regular channel receivers, but can be composed to form higher order
-    formulas.
-    """
-
-    BuilderType: Type[HigherOrderFormulaBuilder | HigherOrderFormulaBuilder3Phase]
-
-    def __init__(
-        self,
-        uuid: UUID,
-        name: str,
-        maxsize: int,
-        chan: _GenericFormulaChannel,
-    ) -> None:
-        """Create a `FormulaReceiver` instance.
-
-        Args:
-            uuid: uuid to uniquely identify the receiver.  Forwarded to
-                BroadcastReceiver's `__init__` function.
-            name: Name for the receiver.
-            maxsize: Buffer size for the receiver.
-            chan: The `FormulaChannel` instance that this receiver is attached to.
-        """
-        self._engine = chan.engine
-        super().__init__(uuid, name, maxsize, chan)
-
-    @property
-    def name(self) -> str:
-        """Name of the receiver.
-
-        Returns:
-            Name of the receiver.
-        """
-        return self._name
-
-    @property
-    def engine(self) -> _GenericEngine:
-        """Return the formula engine attached to the receiver.
-
-        Returns:
-            Formula Engine attached to the receiver.
-        """
-        return self._engine
-
-    # The use of `Self` is necessary for mypy to deduce the type of `clone` method in
-    # the derived classes.  With `from __future__ import annotations`, both CPython and
-    # mypy accept this in python <= 3.10.
-    #
-    # Unfortunately pylint doesn't accept `Self` before python 3.11, even with `from
-    # __future__ import annotations`. So the pylint `undefined-variable` check is
-    # disabled to get `Self` to pass the checks.
-    def clone(self) -> Self:  # pylint: disable=undefined-variable
-        """Create a new receiver from the formula engine.
-
-        Returns:
-            New `FormulaReceiver` streaming a copy of the formula engine output.
-        """
-        return self._engine.new_receiver()
-
-    def __add__(
-        self,
-        other: _BaseFormulaReceiver | _GenericHOFormulaBuilder,
-    ) -> _GenericHOFormulaBuilder:
-        """Return a formula builder that adds (data in) `other` to `self`.
-
-        Args:
-            other: A formula receiver, or a formula builder instance corresponding to a
-                sub-expression.
-
-        Returns:
-            A formula builder that can take further expressions, or can be built
-                into a formula engine.
-        """
-        return self.BuilderType(self) + other
-
-    def __sub__(
-        self,
-        other: _BaseFormulaReceiver | _GenericHOFormulaBuilder,
-    ) -> _GenericHOFormulaBuilder:
-        """Return a formula builder that subtracts (data in) `other` from `self`.
-
-        Args:
-            other: A formula receiver, or a formula builder instance corresponding to a
-                sub-expression.
-
-        Returns:
-            A formula builder that can take further expressions, or can be built
-                into a formula engine.
-        """
-        return self.BuilderType(self) - other
-
-    def __mul__(
-        self, other: _BaseFormulaReceiver | _GenericHOFormulaBuilder
-    ) -> _GenericHOFormulaBuilder:
-        """Return a formula builder that multiplies (data in) `self` with `other`.
-
-        Args:
-            other: A formula receiver, or a formula builder instance corresponding to a
-                sub-expression.
-
-        Returns:
-            A formula builder that can take further expressions, or can be built
-                into a formula engine.
-        """
-        return self.BuilderType(self) * other
-
-    def __truediv__(
-        self, other: _BaseFormulaReceiver | _GenericHOFormulaBuilder
-    ) -> _GenericHOFormulaBuilder:
-        """Return a formula builder that divides (data in) `self` by `other`.
-
-        Args:
-            other: A formula receiver, or a formula builder instance corresponding to a
-                sub-expression.
-
-        Returns:
-            A formula builder that can take further expressions, or can be built
-                into a formula engine.
-        """
-        return self.BuilderType(self) / other
-
-
 class _BaseHOFormulaBuilder(
     ABC, Generic[_GenericFormulaReceiver, _GenericSample, _GenericEngine]
 ):
@@ -795,27 +609,3 @@ def build(self, name: str, nones_are_zeros: bool = False) -> FormulaEngine3Phase
             ),
         )
         return self._engine
-
-
-class FormulaReceiver(_BaseFormulaReceiver[Sample, FormulaEngine]):
-    """A specialization of the _BaseFormulaChannel for `Sample` objects."""
-
-    BuilderType = HigherOrderFormulaBuilder
-
-
-class FormulaReceiver3Phase(_BaseFormulaReceiver[Sample3Phase, FormulaEngine3Phase]):
-    """A specialization of the _BaseFormulaChannel for `Sample3Phase` objects."""
-
-    BuilderType = HigherOrderFormulaBuilder3Phase
-
-
-class FormulaChannel(_BaseFormulaChannel[Sample, FormulaEngine]):
-    """A specialization of the _BaseFormulaChannel for `Sample` objects."""
-
-    ReceiverType = FormulaReceiver
-
-
-class FormulaChannel3Phase(_BaseFormulaChannel[Sample3Phase, FormulaEngine3Phase]):
-    """A specialization of the _BaseFormulaChannel for `Sample3Phase` objects."""
-
-    ReceiverType = FormulaReceiver3Phase