Add a predicate to filter messages based on the previous one

Signed-off-by: Leandro Lucarella <[email protected]>

Author: llucax

RELEASE_NOTES.md

@@ -10,7 +10,9 @@
 
 ## New Features
 
-<!-- Here goes the main new features and examples or instructions on how to use them -->
+### Experimental
+
+- A new predicate, `OnlyIfPrevious`, to `filter()` messages based on the previous message.
 
 ## Bug Fixes
 

src/frequenz/channels/experimental/__init__.py

@@ -10,9 +10,11 @@
 """
 
 from ._pipe import Pipe
+from ._predicates import OnlyIfPrevious
 from ._relay_sender import RelaySender
 
 __all__ = [
+    "OnlyIfPrevious",
     "Pipe",
     "RelaySender",
 ]

src/frequenz/channels/experimental/_predicates.py

@@ -0,0 +1,129 @@
+# License: MIT
+# Copyright © 2024 Frequenz Energy-as-a-Service GmbH
+
+"""Predicates to be used in conjuntion with `Receiver.filter()`."""
+
+
+from typing import Callable, Final, Generic, TypeGuard
+
+from frequenz.channels._generic import ChannelMessageT
+
+
+class _Sentinel:
+    """A sentinel to denote that no value has been received yet."""
+
+    def __str__(self) -> str:
+        """Return a string representation of this sentinel."""
+        return "<no value received yet>"
+
+
+_SENTINEL: Final[_Sentinel] = _Sentinel()
+
+
+class OnlyIfPrevious(Generic[ChannelMessageT]):
+    """A predicate to check if a message has a particular relationship with the previous one.
+
+    This predicate can be used to filter out messages based on a custom condition on the
+    previous and current messages. This can be useful in cases where you want to
+    process messages only if they satisfy a particular condition with respect to the
+    previous message.
+
+    Tip:
+        If you want to use `==` as predicate, you can use the
+        [`ChangedOnly`][frequenz.channels.experimental.ChangedOnly] predicate.
+
+    Example: Receiving only messages that are not the same instance as the previous one.
+        ```python
+        from frequenz.channels import Broadcast
+        from frequenz.channels.experimental import OnlyIfPrevious
+
+        channel = Broadcast[int | bool](name="example")
+        receiver = channel.new_receiver().filter(OnlyIfPrevious(lambda old, new: old is not new))
+        sender = channel.new_sender()
+
+        # This message will be received as it is the first message.
+        await sender.send(1)
+        assert await receiver.receive() == 1
+
+        # This message will be skipped as it is the same instance as the previous one.
+        await sender.send(1)
+
+        # This message will be received as it is a different instance from the previous
+        # one.
+        await sender.send(True)
+        assert await receiver.receive() is True
+        ```
+
+    Example: Receiving only messages if they are bigger than the previous one.
+        ```python
+        from frequenz.channels import Broadcast
+        from frequenz.channels.experimental import OnlyIfPrevious
+
+        channel = Broadcast[int](name="example")
+        receiver = channel.new_receiver().filter(
+            OnlyIfPrevious(lambda old, new: new > old, first_is_true=False)
+        )
+        sender = channel.new_sender()
+
+        # This message will skipped as first_is_true is False.
+        await sender.send(1)
+
+        # This message will be received as it is bigger than the previous one (1).
+        await sender.send(2)
+        assert await receiver.receive() == 2
+
+        # This message will be skipped as it is smaller than the previous one (1).
+        await sender.send(0)
+
+        # This message will be skipped as it is not bigger than the previous one (0).
+        await sender.send(0)
+
+        # This message will be received as it is bigger than the previous one (0).
+        await sender.send(1)
+        assert await receiver.receive() == 1
+
+        # This message will be received as it is bigger than the previous one (1).
+        await sender.send(2)
+        assert await receiver.receive() == 2
+    """
+
+    def __init__(
+        self,
+        predicate: Callable[[ChannelMessageT, ChannelMessageT], bool],
+        *,
+        first_is_true: bool = True,
+    ) -> None:
+        """Initialize this instance.
+
+        Args:
+            predicate: A callable that takes two arguments, the previous message and the
+                current message, and returns a boolean indicating whether the current
+                message should be received.
+            first_is_true: Whether the first message should be considered as satisfying
+                the predicate. Defaults to `True`.
+        """
+        self._predicate = predicate
+        self._last_message: ChannelMessageT | _Sentinel = _SENTINEL
+        self._first_is_true = first_is_true
+
+    def __call__(self, message: ChannelMessageT) -> bool:
+        """Return whether `message` is the first one received or different from the previous one."""
+
+        def is_message(
+            value: ChannelMessageT | _Sentinel,
+        ) -> TypeGuard[ChannelMessageT]:
+            return value is not _SENTINEL
+
+        old_message = self._last_message
+        self._last_message = message
+        if is_message(old_message):
+            return self._predicate(old_message, message)
+        return self._first_is_true
+
+    def __str__(self) -> str:
+        """Return a string representation of this instance."""
+        return f"{type(self).__name__}:{self._predicate.__name__}"
+
+    def __repr__(self) -> str:
+        """Return a string representation of this instance."""
+        return f"<{type(self).__name__}: {self._predicate!r} first_is_true={self._first_is_true!r}>"

tests/experimental/test_predicates_only_if_previous.py

@@ -0,0 +1,158 @@
+# License: MIT
+# Copyright © 2024 Frequenz Energy-as-a-Service GmbH
+
+"""Tests for the OnlyIfPrevious implementation."""
+
+from dataclasses import dataclass
+from typing import Callable, Generic, TypeVar
+
+import pytest
+
+from frequenz.channels.experimental import OnlyIfPrevious
+
+_T = TypeVar("_T")
+
+
+@dataclass(frozen=True, kw_only=True)
+class PredicateTestCase(Generic[_T]):
+    """Test case for testing OnlyIfPrevious behavior with different predicates."""
+
+    title: str
+    messages: list[_T]
+    expected_results: list[bool]
+    predicate: Callable[[_T, _T], bool]
+    first_is_true: bool
+
+
+def always_true(old: object, new: object) -> bool:  # pylint: disable=unused-argument
+    """Return always True."""
+    return True
+
+
+def always_false(old: object, new: object) -> bool:  # pylint: disable=unused-argument
+    """Return always False."""
+    return False
+
+
+def is_greater(old: int, new: int) -> bool:
+    """Return weather the new value is greater than the old one."""
+    return new > old
+
+
+def is_not_same_instance(old: object, new: object) -> bool:
+    """Return weather the new value is not the same instance as the old one."""
+    return old is not new
+
+
+PREDICATE_TEST_CASES = [
+    # Basic cases with different predicates
+    PredicateTestCase(
+        title="Always true predicate",
+        messages=[1, 2, 3],
+        expected_results=[True, True, True],
+        predicate=always_true,
+        first_is_true=True,
+    ),
+    PredicateTestCase(
+        title="Always false predicate with first_is_true=False",
+        messages=[1, 2, 3],
+        expected_results=[False, False, False],
+        predicate=always_false,
+        first_is_true=False,
+    ),
+    PredicateTestCase(
+        title="Greater than predicate",
+        messages=[1, 2, 0, 0, 1, 2],
+        expected_results=[False, True, False, False, True, True],
+        predicate=is_greater,
+        first_is_true=False,
+    ),
+    # Edge cases
+    PredicateTestCase(
+        title="Empty sequence",
+        messages=[],
+        expected_results=[],
+        predicate=always_true,
+        first_is_true=True,
+    ),
+    PredicateTestCase(
+        title="Single value with first_is_true=True",
+        messages=[1],
+        expected_results=[True],
+        predicate=always_false,
+        first_is_true=True,
+    ),
+    PredicateTestCase(
+        title="Single value with first_is_true=False",
+        messages=[1],
+        expected_results=[False],
+        predicate=always_true,
+        first_is_true=False,
+    ),
+    # Instance comparison
+    PredicateTestCase(
+        title="Same instances",
+        messages=[1, 1],
+        expected_results=[True, False],
+        predicate=is_not_same_instance,
+        first_is_true=True,
+    ),
+    PredicateTestCase(
+        title="Different instances of same values",
+        messages=[[1], [1]],
+        expected_results=[True, True],
+        predicate=is_not_same_instance,
+        first_is_true=True,
+    ),
+]
+
+
+@pytest.mark.parametrize(
+    "test_case",
+    PREDICATE_TEST_CASES,
+    ids=lambda test_case: test_case.title,
+)
+def test_only_if_previous(test_case: PredicateTestCase[_T]) -> None:
+    """Test the OnlyIfPrevious with different predicates and sequences.
+
+    Args:
+        test_case: The test case containing the input values and expected results.
+    """
+    only_if_previous = OnlyIfPrevious(
+        test_case.predicate,
+        first_is_true=test_case.first_is_true,
+    )
+    results = [only_if_previous(msg) for msg in test_case.messages]
+    assert results == test_case.expected_results
+
+
+def test_only_if_previous_state_independence() -> None:
+    """Test that multiple OnlyIfPrevious instances maintain independent state."""
+    only_if_previous1 = OnlyIfPrevious(is_greater)
+    only_if_previous2 = OnlyIfPrevious(is_greater)
+
+    # First message should be accepted (first_is_true default is True)
+    assert only_if_previous1(1) is True
+    assert only_if_previous2(10) is True
+
+    # Second messages should be evaluated independently
+    assert only_if_previous1(0) is False  # 0 is not greater than 1
+    assert only_if_previous2(20) is True  # 20 is greater than 10
+
+
+def test_only_if_previous_str_representation() -> None:
+    """Test the string representation of OnlyIfPrevious."""
+    only_if_previous = OnlyIfPrevious(is_greater)
+    assert str(only_if_previous) == "OnlyIfPrevious:is_greater"
+    assert (
+        repr(only_if_previous) == f"<OnlyIfPrevious: {is_greater!r} first_is_true=True>"
+    )
+
+
+def test_only_if_previous_sentinel_str() -> None:
+    """Test the string representation of the sentinel value."""
+    only_if_previous = OnlyIfPrevious(always_true)
+
+    # Access the private attribute for testing purposes
+    # pylint: disable=protected-access
+    assert str(only_if_previous._last_message) == "<no value received yet>"