Add descriptive str and repr implementations

When debugging and logging objects it is very useful to get a
descriptive and clear string representation.

The new representation uses the class name and the user defined name
(if any) for `str` and a `repr` that tries to show how the class was
created but also important internal state.

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

# ------------------------ >8 ------------------------
# Do not modify or remove the line above.
# Everything below it will be ignored.
#
# Conflicts:
#	src/frequenz/channels/_base_classes.py
#	src/frequenz/channels/_bidirectional.py

Author: llucax

src/frequenz/channels/_anycast.py

@@ -171,6 +171,17 @@ def new_receiver(self) -> Receiver[T]:
         """
         return Receiver(self)
 
+    def __str__(self) -> str:
+        """Return a string representation of this channel."""
+        return f"{type(self).__name__}:{self._name}"
+
+    def __repr__(self) -> str:
+        """Return a string representation of this channel."""
+        return (
+            f"{type(self).__name__}(name={self._name!r}, limit={self.limit!r}):<"
+            f"current={len(self._deque)!r}, closed={self._closed!r}>"
+        )
+
 
 class Sender(BaseSender[T]):
     """A sender to send messages to an Anycast channel.
@@ -217,6 +228,14 @@ async def send(self, msg: T) -> None:
             self._chan._recv_cv.notify(1)
         # pylint: enable=protected-access
 
+    def __str__(self) -> str:
+        """Return a string representation of this sender."""
+        return f"{self._chan}:{type(self).__name__}"
+
+    def __repr__(self) -> str:
+        """Return a string representation of this sender."""
+        return f"{type(self).__name__}({self._chan!r})"
+
 
 class _Empty:
     """A sentinel value to indicate that a value has not been set."""
@@ -291,3 +310,11 @@ def consume(self) -> T:
         self._next = _Empty
 
         return next_val
+
+    def __str__(self) -> str:
+        """Return a string representation of this receiver."""
+        return f"{self._chan}:{type(self).__name__}"
+
+    def __repr__(self) -> str:
+        """Return a string representation of this receiver."""
+        return f"{type(self).__name__}({self._chan!r})"

src/frequenz/channels/_base_classes.py

@@ -206,3 +206,11 @@ def consume(self) -> U:  # noqa: DOC502
         return self._transform(
             self._receiver.consume()
         )  # pylint: disable=protected-access
+
+    def __str__(self) -> str:
+        """Return a string representation of the timer."""
+        return f"{type(self).__name__}:{self._receiver}:{self._transform}"
+
+    def __repr__(self) -> str:
+        """Return a string representation of the timer."""
+        return f"{type(self).__name__}({self._receiver!r}, {self._transform!r})"

src/frequenz/channels/_bidirectional.py

@@ -112,6 +112,17 @@ def consume(self) -> W:
                     err.__cause__ = this_chan_error
                 raise err
 
+        def __str__(self) -> str:
+            """Return a string representation of this handle."""
+            return f"{type(self).__name__}:{self._chan}"
+
+        def __repr__(self) -> str:
+            """Return a string representation of this handle."""
+            return (
+                f"{type(self).__name__}(channel={self._chan!r}, "
+                f"sender={self._sender!r}, receiver={self._receiver!r})"
+            )
+
     def __init__(self, *, name: str | None = None) -> None:
         """Create a `Bidirectional` instance.
 
@@ -180,3 +191,15 @@ def service_handle(self) -> Bidirectional.Handle[U, T]:
             Object to send/receive messages with.
         """
         return self._service_handle
+
+    def __str__(self) -> str:
+        """Return a string representation of this channel."""
+        return f"{type(self).__name__}:{self._name}"
+
+    def __repr__(self) -> str:
+        """Return a string representation of this channel."""
+        return (
+            f"{type(self).__name__}(name={self._name!r}):<"
+            f"request_channel={self._request_channel!r}, "
+            f"response_channel={self._response_channel!r}>"
+        )

src/frequenz/channels/_broadcast.py

@@ -200,6 +200,20 @@ def new_peekable(self) -> Peekable[T]:
         """
         return Peekable(self)
 
+    def __str__(self) -> str:
+        """Return a string representation of this receiver."""
+        return f"{type(self).__name__}:{self._name}"
+
+    def __repr__(self) -> str:
+        """Return a string representation of this channel."""
+        return (
+            f"{type(self).__name__}(name={self._name!r}, "
+            f"resend_latest={self.resend_latest!r}):<"
+            f"latest={self._latest!r}, "
+            f"receivers={len(self._receivers)!r}, "
+            f"closed={self._closed!r}>"
+        )
+
 
 class Sender(BaseSender[T]):
     """A sender to send messages to the broadcast channel.
@@ -248,6 +262,14 @@ async def send(self, msg: T) -> None:
             self._chan._recv_cv.notify_all()
         # pylint: enable=protected-access
 
+    def __str__(self) -> str:
+        """Return a string representation of this sender."""
+        return f"{self._chan}:{type(self).__name__}"
+
+    def __repr__(self) -> str:
+        """Return a string representation of this sender."""
+        return f"{type(self).__name__}({self._chan!r})"
+
 
 class Receiver(BaseReceiver[T]):
     """A receiver to receive messages from the broadcast channel.
@@ -396,6 +418,20 @@ def into_peekable(self) -> Peekable[T]:
         self._deactivate()
         return Peekable(self._chan)
 
+    def __str__(self) -> str:
+        """Return a string representation of this receiver."""
+        return f"{self._chan}:{type(self).__name__}"
+
+    def __repr__(self) -> str:
+        """Return a string representation of this receiver."""
+        limit = self._q.maxlen
+        assert limit is not None
+        return (
+            f"{type(self).__name__}(name={self._name!r}, limit={limit!r}, "
+            f"{self._chan!r}):<id={id(self)!r}, used={len(self._q)!r}, "
+            f"active={self._active!r}>"
+        )
+
 
 class Peekable(BasePeekable[T]):
     """A Peekable to peek into broadcast channels.
@@ -422,3 +458,11 @@ def peek(self) -> T | None:
                 has been sent to the channel yet, or if the channel is closed.
         """
         return self._chan._latest  # pylint: disable=protected-access
+
+    def __str__(self) -> str:
+        """Return a string representation of this receiver."""
+        return f"{self._chan}:{type(self).__name__}"
+
+    def __repr__(self) -> str:
+        """Return a string representation of this receiver."""
+        return f"{type(self).__name__}({self._chan!r}):<latest={self.peek()!r}>"

src/frequenz/channels/util/_file_watcher.py

@@ -137,3 +137,17 @@ def consume(self) -> Event:
         return FileWatcher.Event(
             type=FileWatcher.EventType(change), path=pathlib.Path(path_str)
         )
+
+    def __str__(self) -> str:
+        """Return a string representation of this receiver."""
+        if len(self._paths) > 3:
+            paths = [str(p) for p in self._paths[:3]]
+            paths.append("…")
+        else:
+            paths = [str(p) for p in self._paths]
+        event_types = [event_type.name for event_type in self.event_types]
+        return f"{type(self).__name__}:{','.join(event_types)}:{','.join(paths)}"
+
+    def __repr__(self) -> str:
+        """Return a string representation of this receiver."""
+        return f"{type(self).__name__}({self._paths!r}, {self.event_types!r})"

src/frequenz/channels/util/_merge.py

@@ -4,6 +4,7 @@
 """Merge messages coming from channels into a single stream."""
 
 import asyncio
+import itertools
 from collections import deque
 from typing import Any
 
@@ -117,3 +118,19 @@ def consume(self) -> T:
         assert self._results, "`consume()` must be preceded by a call to `ready()`"
 
         return self._results.popleft()
+
+    def __str__(self) -> str:
+        """Return a string representation of this receiver."""
+        if len(self._receivers) > 3:
+            receivers = [str(p) for p in itertools.islice(self._receivers.values(), 3)]
+            receivers.append("…")
+        else:
+            receivers = [str(p) for p in self._receivers.values()]
+        return f"{type(self).__name__}:{','.join(receivers)}"
+
+    def __repr__(self) -> str:
+        """Return a string representation of this receiver."""
+        return (
+            f"{type(self).__name__}("
+            f"{', '.join(f'{k}={v!r}' for k, v in self._receivers.items())})"
+        )

src/frequenz/channels/util/_merge_named.py

@@ -2,8 +2,8 @@
 # Copyright © 2022 Frequenz Energy-as-a-Service GmbH
 
 """Merge messages coming from channels into a single stream containing name of message."""
-
 import asyncio
+import itertools
 from collections import deque
 from typing import Any
 
@@ -102,3 +102,19 @@ def consume(self) -> tuple[str, T]:
         assert self._results, "`consume()` must be preceded by a call to `ready()`"
 
         return self._results.popleft()
+
+    def __str__(self) -> str:
+        """Return a string representation of this receiver."""
+        if len(self._receivers) > 3:
+            receivers = [str(p) for p in itertools.islice(self._receivers, 3)]
+            receivers.append("…")
+        else:
+            receivers = [str(p) for p in self._receivers]
+        return f"{type(self).__name__}:{','.join(receivers)}"
+
+    def __repr__(self) -> str:
+        """Return a string representation of this receiver."""
+        return (
+            f"{type(self).__name__}("
+            f"{', '.join(f'{k}={v!r}' for k, v in self._receivers.items())})"
+        )