enhancement #29: fixing mypy for opencv

setup.cfg

@@ -50,7 +50,7 @@ show_missing=True
 
 [mypy]
 ignore_missing_imports = True
-exclude = (tests|src/example/videostream-cli|src/example/datachannel-cli)
+exclude = (tests|src/example/videostream-cli/cli.py|src/example/datachannel-cli/cli.py)
 strict = True
 plugins = numpy.typing.mypy_plugin
 explicit_package_bases = True

src/example/videostream-cli/opencv.py

@@ -2,6 +2,8 @@
 import asyncio
 import logging
 
+import aiortc
+import av
 import cv2
 from aiortc import (
     MediaStreamTrack,
@@ -10,25 +12,30 @@
     RTCSessionDescription,
 )
 from aiortc.contrib.media import MediaBlackhole
+from aiortc.rtcrtpreceiver import RemoteStreamTrack
 
-from gst_signalling.aiortc_adapter import BYE, add_signaling_arguments, create_signaling
+from gst_signalling.aiortc_adapter import (
+    BYE,
+    GstSignalingForAiortc,
+    add_signaling_arguments,
+    create_signaling,
+)
 
 
-class OpenCVTrack(MediaStreamTrack):
+class OpenCVTrack(MediaStreamTrack):  # type: ignore[misc]
     """
     A video track that processes frames with opencv.
     """
 
     kind = "video"
 
-    def __init__(self, track):
+    def __init__(self, track: RemoteStreamTrack):
         super().__init__()  # don't forget this!
         self.track = track
         self._window_name = f"track {self._id}"
 
-    async def recv(self):
+    async def recv(self) -> av.video.frame.VideoFrame:
         frame = await self.track.recv()
-
         img = frame.to_ndarray(format="bgr24")
         # process the frame here
         cv2.imshow(self._window_name, img)
@@ -37,9 +44,11 @@ async def recv(self):
         return frame
 
 
-async def run(pc, recorder, signaling):
-    @pc.on("track")
-    def on_track(track):
+async def run(
+    pc: RTCPeerConnection, recorder: MediaBlackhole, signaling: GstSignalingForAiortc
+) -> None:
+    @pc.on("track")  # type: ignore[misc]
+    def on_track(track: RemoteStreamTrack) -> None:
         print("Receiving %s" % track.kind)
         if track.kind == "video":
             recorder.addTrack(OpenCVTrack(track))
@@ -84,6 +93,7 @@ def on_track(track):
 
     # create media sink
     recorder = MediaBlackhole()
+    print(type(signaling))
 
     # run event loop
     loop = asyncio.get_event_loop()