From b8d2b9f364002936d720b4cfda0a5f04e2fdefb8 Mon Sep 17 00:00:00 2001
From: Fabien Danieau <fabien.danieau@pollen-robotics.com>
Date: Thu, 26 Oct 2023 17:36:21 +0200
Subject: [PATCH] enhancement #29: fixing mypy for opencv

---
 setup.cfg                             |  2 +-
 src/example/videostream-cli/opencv.py | 26 ++++++++++++++++++--------
 2 files changed, 19 insertions(+), 9 deletions(-)

diff --git a/setup.cfg b/setup.cfg
index 92de1e6..f366318 100644
--- a/setup.cfg
+++ b/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
\ No newline at end of file
diff --git a/src/example/videostream-cli/opencv.py b/src/example/videostream-cli/opencv.py
index ba94b55..47f1627 100644
--- a/src/example/videostream-cli/opencv.py
+++ b/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()