OSC typing, harden Capture (#462)

* Tighten OSC typing

* Harden Capture

Author: josephine-wolf-oberholtzer

supriya/contexts/core.py

@@ -10,6 +10,7 @@
 import itertools
 import re
 import threading
+from collections.abc import Sequence as SequenceABC
 from os import PathLike
 from typing import (
     Callable,
@@ -1179,7 +1180,7 @@ def read_buffer(
         return self._add_request_with_completion(request, on_completion)
 
     @abc.abstractmethod
-    def send(self, message: SupportsOsc):
+    def send(self, message: Union[SupportsOsc, SequenceABC, str]):
         """
         Send a message to the execution context.
 

supriya/contexts/nonrealtime.py

@@ -9,6 +9,7 @@
 import shlex
 import shutil
 import struct
+from collections.abc import Sequence as SequenceABC
 from contextlib import ExitStack
 from os import PathLike
 from pathlib import Path
@@ -238,7 +239,7 @@ def iterate_request_bundles(
         if until and until > timestamp:
             yield RequestBundle(timestamp=until, contents=[DoNothing()])
 
-    def send(self, message: SupportsOsc) -> None:
+    def send(self, message: Union[SupportsOsc, SequenceABC, str]):
         """
         Send a message to the execution context.
 

supriya/contexts/realtime.py

@@ -44,7 +44,6 @@
 from ..osc import (
     AsyncOscProtocol,
     HealthCheck,
-    OscBundle,
     OscCallback,
     OscMessage,
     OscProtocol,
@@ -374,9 +373,7 @@ def _validate_moment_timestamp(self, seconds: Optional[float]) -> None:
 
     ### PUBLIC METHODS ###
 
-    def send(
-        self, message: Union[OscMessage, OscBundle, SupportsOsc, SequenceABC, str]
-    ) -> None:
+    def send(self, message: Union[SupportsOsc, SequenceABC, str]) -> None:
         """
         Send a message to the execution context.
 
@@ -385,9 +382,7 @@ def send(
         if self._boot_status == BootStatus.OFFLINE:
             raise ServerOffline
         osc_protocol: OscProtocol = getattr(self, "_osc_protocol")
-        osc_protocol.send(
-            message.to_osc() if isinstance(message, SupportsOsc) else message
-        )
+        osc_protocol.send(message)
 
     def set_latency(self, latency: float) -> None:
         """

supriya/osc/asynchronous.py

@@ -3,8 +3,8 @@
 from typing import Awaitable, Callable, Dict, Optional, Sequence, Set, Tuple, Union
 
 from ..enums import BootStatus
-from ..typing import FutureLike
-from .messages import OscBundle, OscMessage
+from ..typing import FutureLike, SupportsOsc
+from .messages import OscMessage
 from .protocols import (
     HealthCheck,
     OscCallback,
@@ -196,7 +196,7 @@ def register(
         )
         return callback
 
-    def send(self, message: Union[OscBundle, OscMessage, SequenceABC, str]) -> None:
+    def send(self, message: Union[SupportsOsc, SequenceABC, str]) -> None:
         self.transport.sendto(self._send(message))
 
     def unregister(self, callback: OscCallback) -> None:

supriya/osc/messages.py

@@ -297,6 +297,9 @@ def to_list(self):
                 result.append(x)
         return result
 
+    def to_osc(self) -> "OscMessage":
+        return self
+
 
 class OscBundle:
     """
@@ -471,3 +474,6 @@ def to_list(self):
         result = [self.timestamp]
         result.append([x.to_list() for x in self.contents])
         return result
+
+    def to_osc(self) -> "OscBundle":
+        return self

supriya/osc/protocols.py

@@ -9,7 +9,9 @@
     Awaitable,
     Callable,
     Dict,
+    Iterator,
     List,
+    Literal,
     NamedTuple,
     Optional,
     Sequence,
@@ -19,7 +21,7 @@
 )
 
 from ..enums import BootStatus
-from ..typing import FutureLike
+from ..typing import FutureLike, SupportsOsc
 from .messages import OscBundle, OscMessage
 
 osc_protocol_logger = logging.getLogger(__name__)
@@ -65,31 +67,35 @@ class HealthCheck:
 
 class CaptureEntry(NamedTuple):
     timestamp: float
-    label: str
+    label: Literal["R", "S"]
     message: Union[OscMessage, OscBundle]
+    raw_message: Union[SupportsOsc, SequenceABC, str] | None = None
 
 
 class Capture:
     ### INITIALIZER ###
 
-    def __init__(self, osc_protocol):
+    def __init__(self, osc_protocol: "OscProtocol") -> None:
         self.osc_protocol = osc_protocol
-        self.messages = []
+        self.messages: list[CaptureEntry] = []
 
     ### SPECIAL METHODS ###
 
-    def __enter__(self):
+    def __enter__(self) -> "Capture":
         self.osc_protocol.captures.add(self)
         self.messages[:] = []
         return self
 
-    def __exit__(self, exc_type, exc_value, traceback):
+    def __exit__(self, exc_type, exc_value, traceback) -> None:
         self.osc_protocol.captures.remove(self)
 
-    def __iter__(self):
+    def __getitem__(self, i: int | slice) -> CaptureEntry | list[CaptureEntry]:
+        return self.messages[i]
+
+    def __iter__(self) -> Iterator[CaptureEntry]:
         return iter(self.messages)
 
-    def __len__(self):
+    def __len__(self) -> int:
         return len(self.messages)
 
     ### PUBLIC METHODS ###
@@ -98,7 +104,7 @@ def filtered(
         self, sent=True, received=True, status=True
     ) -> List[Union[OscBundle, OscMessage]]:
         messages = []
-        for _, label, message in self.messages:
+        for _, label, message, _ in self.messages:
             if label == "R" and not received:
                 continue
             if label == "S" and not sent:
@@ -112,24 +118,6 @@ def filtered(
             messages.append(message)
         return messages
 
-    ### PUBLIC PROPERTIES ###
-
-    @property
-    def received_messages(self):
-        return [
-            (timestamp, osc_message)
-            for timestamp, label, osc_message in self.messages
-            if label == "R"
-        ]
-
-    @property
-    def sent_messages(self):
-        return [
-            (timestamp, osc_message)
-            for timestamp, label, osc_message in self.messages
-            if label == "S"
-        ]
-
 
 class OscProtocol:
     ### INITIALIZER ###
@@ -293,27 +281,35 @@ def _register(
             kwargs=kwargs,
         )
 
-    def _send(self, message):
+    def _send(self, raw_message: Union[SupportsOsc, SequenceABC, str]) -> bytes:
         if self.status not in (BootStatus.BOOTING, BootStatus.ONLINE):
             raise OscProtocolOffline
-        if not isinstance(message, (str, SequenceABC, OscBundle, OscMessage)):
-            raise ValueError(message)
-        if isinstance(message, str):
-            message = OscMessage(message)
-        elif isinstance(message, SequenceABC):
-            message = OscMessage(*message)
+        if not isinstance(raw_message, (str, SequenceABC, SupportsOsc)):
+            raise ValueError(raw_message)
+        message: OscBundle | OscMessage
+        if isinstance(raw_message, str):
+            message = OscMessage(raw_message)
+        elif isinstance(raw_message, SequenceABC):
+            message = OscMessage(*raw_message)
+        else:
+            message = raw_message.to_osc()
         osc_out_logger.debug(
             f"[{self.ip_address}:{self.port}/{self.name or hex(id(self))}] "
             f"{message!r}"
         )
         for capture in self.captures:
             capture.messages.append(
-                CaptureEntry(timestamp=time.time(), label="S", message=message)
+                CaptureEntry(
+                    timestamp=time.time(),
+                    label="S",
+                    message=message,
+                    raw_message=raw_message,
+                )
             )
         datagram = message.to_datagram()
         udp_out_logger.debug(
             f"[{self.ip_address}:{self.port}/{self.name or hex(id(self))}] "
-            f"{datagram}"
+            f"{datagram!r}"
         )
         return datagram
 
@@ -373,7 +369,7 @@ def register(
     ) -> OscCallback:
         raise NotImplementedError
 
-    def send(self, message: Union[OscBundle, OscMessage, SequenceABC, str]) -> None:
+    def send(self, message: Union[SupportsOsc, SequenceABC, str]) -> None:
         raise NotImplementedError
 
     def unregister(self, callback: OscCallback) -> None:

supriya/osc/threaded.py

@@ -17,8 +17,8 @@
 )
 
 from ..enums import BootStatus
-from ..typing import FutureLike
-from .messages import OscBundle, OscMessage
+from ..typing import FutureLike, SupportsOsc
+from .messages import OscMessage
 from .protocols import (
     HealthCheck,
     OscCallback,
@@ -229,7 +229,7 @@ def register(
         )
         return callback
 
-    def send(self, message: Union[OscBundle, OscMessage, SequenceABC, str]) -> None:
+    def send(self, message: Union[SupportsOsc, SequenceABC, str]) -> None:
         try:
             self.osc_server.socket.sendto(
                 self._send(message),