Add more type hints

.gitignore

@@ -1,11 +1,11 @@
 *.egg-info
 *.pyc
+*.so
 .coverage
 .eggs
 .idea
+.mypy_cache
 .vscode
-/aiortc/codecs/_opus.*
-/aiortc/codecs/_vpx.*
 /build
 /dist
 /docs/_build

aiortc/codecs/__init__.py

@@ -1,5 +1,5 @@
 from collections import OrderedDict
-from typing import Dict, List, Optional
+from typing import Dict, List, Optional, Union
 
 from ..rtcrtpparameters import (
     RTCRtcpFeedback,
@@ -108,34 +108,36 @@ def depayload(codec: RTCRtpCodecParameters, payload: bytes) -> bytes:
         return payload
 
 
-def get_capabilities(kind):
-    if kind in CODECS:
-        codecs = []
-        rtx_added = False
-        for params in CODECS[kind]:
-            if not is_rtx(params):
-                codecs.append(
-                    RTCRtpCodecCapability(
-                        mimeType=params.mimeType,
-                        clockRate=params.clockRate,
-                        channels=params.channels,
-                        parameters=params.parameters,
-                    )
+def get_capabilities(kind: str) -> RTCRtpCapabilities:
+    if kind not in CODECS:
+        raise ValueError("cannot get capabilities for unknown media %s" % kind)
+
+    codecs = []
+    rtx_added = False
+    for params in CODECS[kind]:
+        if not is_rtx(params):
+            codecs.append(
+                RTCRtpCodecCapability(
+                    mimeType=params.mimeType,
+                    clockRate=params.clockRate,
+                    channels=params.channels,
+                    parameters=params.parameters,
                 )
-            elif not rtx_added:
-                # There will only be a single entry in codecs[] for retransmission
-                # via RTX, with sdpFmtpLine not present.
-                codecs.append(
-                    RTCRtpCodecCapability(
-                        mimeType=params.mimeType, clockRate=params.clockRate
-                    )
+            )
+        elif not rtx_added:
+            # There will only be a single entry in codecs[] for retransmission
+            # via RTX, with sdpFmtpLine not present.
+            codecs.append(
+                RTCRtpCodecCapability(
+                    mimeType=params.mimeType, clockRate=params.clockRate
                 )
-                rtx_added = True
+            )
+            rtx_added = True
 
-        headerExtensions = []
-        for params in HEADER_EXTENSIONS[kind]:
-            headerExtensions.append(RTCRtpHeaderExtensionCapability(uri=params.uri))
-        return RTCRtpCapabilities(codecs=codecs, headerExtensions=headerExtensions)
+    headerExtensions = []
+    for extension in HEADER_EXTENSIONS[kind]:
+        headerExtensions.append(RTCRtpHeaderExtensionCapability(uri=extension.uri))
+    return RTCRtpCapabilities(codecs=codecs, headerExtensions=headerExtensions)
 
 
 def get_decoder(codec: RTCRtpCodecParameters):
@@ -168,7 +170,7 @@ def get_encoder(codec: RTCRtpCodecParameters):
         return Vp8Encoder()
 
 
-def is_rtx(codec: RTCRtpCodecParameters) -> bool:
+def is_rtx(codec: Union[RTCRtpCodecCapability, RTCRtpCodecParameters]) -> bool:
     return codec.name.lower() == "rtx"
 
 

aiortc/codecs/_opus.pyi

@@ -0,0 +1,4 @@
+from typing import Any
+
+ffi: Any
+lib: Any

aiortc/codecs/_vpx.pyi

@@ -0,0 +1,4 @@
+from typing import Any
+
+ffi: Any
+lib: Any

aiortc/rtcdatachannel.py

@@ -1,4 +1,5 @@
 import logging
+from typing import Optional, Union
 
 import attr
 from pyee import AsyncIOEventEmitter
@@ -8,6 +9,41 @@
 logger = logging.getLogger("datachannel")
 
 
+@attr.s
+class RTCDataChannelParameters:
+    """
+    The :class:`RTCDataChannelParameters` dictionary describes the
+    configuration of an :class:`RTCDataChannel`.
+    """
+
+    label = attr.ib(default="")  # type: str
+    "A name describing the data channel."
+
+    maxPacketLifeTime = attr.ib(default=None)  # type: Optional[int]
+    "The maximum time in milliseconds during which transmissions are attempted."
+
+    maxRetransmits = attr.ib(default=None)  # type: Optional[int]
+    "The maximum number of retransmissions that are attempted."
+
+    ordered = attr.ib(default=True)  # type: bool
+    "Whether the data channel guarantees in-order delivery of messages."
+
+    protocol = attr.ib(default="")  # type: str
+    "The name of the subprotocol in use."
+
+    negotiated = attr.ib(default=False)  # type: bool
+    """
+    Whether data channel will be negotiated out of-band, where both sides
+    create data channel with an agreed-upon ID."""
+
+    id = attr.ib(default=None)  # type: Optional[int]
+    """
+    An numeric ID for the channel; permitted values are 0-65534.
+    If you don't include this option, the user agent will select an ID for you.
+    Must be set when negotiating out-of-band.
+    """
+
+
 class RTCDataChannel(AsyncIOEventEmitter):
     """
     The :class:`RTCDataChannel` interface represents a network channel which
@@ -17,7 +53,9 @@ class RTCDataChannel(AsyncIOEventEmitter):
     :param: parameters: An :class:`RTCDataChannelParameters`.
     """
 
-    def __init__(self, transport, parameters, send_open=True):
+    def __init__(
+        self, transport, parameters: RTCDataChannelParameters, send_open: bool = True
+    ) -> None:
         super().__init__()
         self.__bufferedAmount = 0
         self.__bufferedAmountLowThreshold = 0
@@ -43,43 +81,43 @@ def __init__(self, transport, parameters, send_open=True):
             self.__transport._data_channel_add_negotiated(self)
 
     @property
-    def bufferedAmount(self):
+    def bufferedAmount(self) -> int:
         """
         The number of bytes of data currently queued to be sent over the data channel.
         """
         return self.__bufferedAmount
 
     @property
-    def bufferedAmountLowThreshold(self):
+    def bufferedAmountLowThreshold(self) -> int:
         """
         The number of bytes of buffered outgoing data that is considered "low".
         """
         return self.__bufferedAmountLowThreshold
 
     @bufferedAmountLowThreshold.setter
-    def bufferedAmountLowThreshold(self, value):
+    def bufferedAmountLowThreshold(self, value: int) -> None:
         if value < 0 or value > 4294967295:
             raise ValueError(
                 "bufferedAmountLowThreshold must be in range 0 - 4294967295"
             )
         self.__bufferedAmountLowThreshold = value
 
     @property
-    def negotiated(self):
+    def negotiated(self) -> bool:
         """
         Whether data channel was negotiated out-of-band.
         """
         return self.__parameters.negotiated
 
     @property
-    def id(self):
+    def id(self) -> Optional[int]:
         """
         An ID number which uniquely identifies the data channel.
         """
         return self.__id
 
     @property
-    def label(self):
+    def label(self) -> str:
         """
         A name describing the data channel.
 
@@ -88,35 +126,35 @@ def label(self):
         return self.__parameters.label
 
     @property
-    def ordered(self):
+    def ordered(self) -> bool:
         """
         Indicates whether or not the data channel guarantees in-order delivery of messages.
         """
         return self.__parameters.ordered
 
     @property
-    def maxPacketLifeTime(self):
+    def maxPacketLifeTime(self) -> Optional[int]:
         """
         The maximum time in milliseconds during which transmissions are attempted.
         """
         return self.__parameters.maxPacketLifeTime
 
     @property
-    def maxRetransmits(self):
+    def maxRetransmits(self) -> Optional[int]:
         """
         "The maximum number of retransmissions that are attempted.
         """
         return self.__parameters.maxRetransmits
 
     @property
-    def protocol(self):
+    def protocol(self) -> str:
         """
         The name of the subprotocol in use.
         """
         return self.__parameters.protocol
 
     @property
-    def readyState(self):
+    def readyState(self) -> str:
         """
         A string indicating the current state of the underlying data transport.
         """
@@ -129,13 +167,13 @@ def transport(self):
         """
         return self.__transport
 
-    def close(self):
+    def close(self) -> None:
         """
         Close the data channel.
         """
         self.transport._data_channel_close(self)
 
-    def send(self, data):
+    def send(self, data: Union[bytes, str]) -> None:
         """
         Send `data` across the data channel to the remote peer.
         """
@@ -147,7 +185,7 @@ def send(self, data):
 
         self.transport._data_channel_send(self, data)
 
-    def _addBufferedAmount(self, amount):
+    def _addBufferedAmount(self, amount: int) -> None:
         crosses_threshold = (
             self.__bufferedAmount > self.bufferedAmountLowThreshold
             and self.__bufferedAmount + amount <= self.bufferedAmountLowThreshold
@@ -156,10 +194,10 @@ def _addBufferedAmount(self, amount):
         if crosses_threshold:
             self.emit("bufferedamountlow")
 
-    def _setId(self, id):
+    def _setId(self, id: int) -> None:
         self.__id = id
 
-    def _setReadyState(self, state):
+    def _setReadyState(self, state: str) -> None:
         if state != self.__readyState:
             self.__log_debug("- %s -> %s", self.__readyState, state)
             self.__readyState = state
@@ -173,40 +211,5 @@ def _setReadyState(self, state):
                 # to facilitate garbage collection.
                 self.remove_all_listeners()
 
-    def __log_debug(self, msg, *args):
+    def __log_debug(self, msg: str, *args) -> None:
         logger.debug(str(self.id) + " " + msg, *args)
-
-
-@attr.s
-class RTCDataChannelParameters:
-    """
-    The :class:`RTCDataChannelParameters` dictionary describes the
-    configuration of an :class:`RTCDataChannel`.
-    """
-
-    label = attr.ib(default="")
-    "A name describing the data channel."
-
-    maxPacketLifeTime = attr.ib(default=None)
-    "The maximum time in milliseconds during which transmissions are attempted."
-
-    maxRetransmits = attr.ib(default=None)
-    "The maximum number of retransmissions that are attempted."
-
-    ordered = attr.ib(default=True)
-    "Whether the data channel guarantees in-order delivery of messages."
-
-    protocol = attr.ib(default="")
-    "The name of the subprotocol in use."
-
-    negotiated = attr.ib(default=False)
-    """
-    Whether data channel will be negotiated out of-band, where both sides
-    create data channel with an agreed-upon ID."""
-
-    id = attr.ib(default=None)
-    """
-    An numeric ID for the channel; permitted values are 0-65534.
-    If you don't include this option, the user agent will select an ID for you.
-    Must be set when negotiating out-of-band.
-    """