chore: testing client working

Author: kieranmanning

intercomclient/camera_video_stream_track.py

@@ -1,9 +1,11 @@
-import cv2
+import asyncio
 import logging
-from intercomclient.config import Config
-from aiortc import VideoStreamTrack
+
 import av
-import asyncio
+import cv2
+from aiortc import VideoStreamTrack
+
+from intercomclient.config import Config
 
 LOG = logging.getLogger(__name__)
 
@@ -13,8 +15,9 @@ def __init__(self, config=Config):
         self.config = config
         self.target_fps = config.target_fps
         self.capture = cv2.VideoCapture(config.video_source)
+        LOG.info("Initialized video capture with source: %s", config.video_source)
         if not self.capture.isOpened():
-            raise ValueError("Unable to open video source")
+            raise ValueError("Unable to open video source: %s", config.video_source)
         self.segment_number = 0
         super().__init__()
 

intercomclient/config.py

@@ -1,8 +1,8 @@
-from dataclasses import dataclass
-from typing import Tuple
-from pathlib import Path
 import logging
 import os
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Tuple
 
 
 @dataclass()

intercomclient/device_authorization.py

@@ -1,9 +1,14 @@
-from .device import get_device_type, get_os_info
-from .config import Config
-from time import sleep
+import logging
 from datetime import datetime, timedelta
+from time import sleep
+
 import requests
 
+from .config import Config
+from .device import get_device_type, get_os_info
+
+LOG = logging.getLogger(__name__)
+
 
 def initiate_device_authorization(config: Config) -> dict:
     client_id = Config.oauth_client_id
@@ -65,4 +70,10 @@ def refresh_tokens(config: Config, refresh_token) -> dict:
         },
     )
 
-    return api_response.json()
+    try:
+        api_response.raise_for_status()
+        response_json = api_response.json()
+        return response_json
+    except requests.HTTPError:
+        LOG.error("HTTP error during token refresh: %s", api_response.json())
+        return response_json

intercomclient/token_store.py

@@ -1,10 +1,11 @@
+import datetime
 import json
 import logging
 import os
-from .config import Config
-import datetime
 from enum import Enum
 
+from .config import Config
+
 LOG = logging.getLogger(__name__)
 
 

intercomclient/video_writer.py

@@ -1,10 +1,12 @@
-from datetime import datetime
-import cv2
+import json
 import logging
 import os
-import json
 import time
+from datetime import datetime
 from pathlib import Path
+
+import cv2
+
 from intercomclient.segment import Segment
 
 LOG = logging.getLogger(__name__)

main.py

@@ -2,19 +2,20 @@
 import json
 import logging
 import signal
-from datetime import datetime, timedelta, UTC
+from datetime import UTC, datetime, timedelta
 
 import websockets
 from aiortc import RTCPeerConnection, RTCSessionDescription
+from requests import HTTPError
 
+from intercomclient.camera_video_stream_track import CameraVideoStreamTrack
 from intercomclient.config import Config
-from intercomclient.token_store import TokenStore, TokenStatus
 from intercomclient.device_authorization import (
     initiate_device_authorization,
     poll_for_token,
     refresh_tokens,
 )
-from intercomclient.camera_video_stream_track import CameraVideoStreamTrack
+from intercomclient.token_store import TokenStatus, TokenStore
 
 logging.basicConfig(level=logging.INFO)
 LOG = logging.getLogger("pi-client")
@@ -91,10 +92,15 @@ def device_authorization_flow(self):
     def refresh_flow(self):
         tokens = self.token_store.load_tokens()
 
-        refresh_response = refresh_tokens(
-            self.config,
-            tokens["refresh"]["token_value"],
-        )
+        try:
+            refresh_response = refresh_tokens(
+                self.config,
+                tokens["refresh"]["token_value"],
+            )
+
+        except HTTPError as e:
+            LOG.error("Token refresh failed: %s", e)
+            print(refresh_response)
 
         expiry = (
             datetime.now(tz=UTC) + timedelta(seconds=refresh_response["expires_in"])
@@ -121,9 +127,6 @@ async def on_connectionstatechange():
             if self.pc.connectionState in ("failed", "closed"):
                 await self.shutdown()
 
-        camera_track = CameraVideoStreamTrack(self.config)
-        self.pc.addTrack(camera_track)
-
     async def signaling_loop(self):
         token_store = self.token_store.load_tokens()
         access_token = token_store["access"]["token_value"]
@@ -142,48 +145,57 @@ async def signaling_loop(self):
 
             await self.setup_peer_connection()
 
-            # Create offer
-            # offer = await self.pc.createOffer()
-            # await self.pc.setLocalDescription(offer)
-
-            # await ws.send(
-            #     json.dumps(
-            #         {
-            #             "type": "offer",
-            #             "sdp": self.pc.localDescription.sdp,
-            #         }
-            #     )
-            # )
+            @self.pc.on("icecandidate")
+            async def on_icecandidate(self, candidate):
+                print(f"ICE candidate generated: {candidate}")
+                if candidate is not None:
+                    # Send the candidate to the signaling server
+                    await self.ws.send_json(
+                        json.dumps(
+                            {
+                                "type": "ice",
+                                "candidate": {
+                                    "candidate": candidate.component,
+                                    "sdpMid": candidate.sdpMid,
+                                    "sdpMLineIndex": candidate.sdpMLineIndex,
+                                    "foundation": candidate.foundation,
+                                    "priority": candidate.priority,
+                                    "ip": candidate.address,
+                                    "port": candidate.port,
+                                    "protocol": candidate.protocol,
+                                    "type": candidate.type,
+                                },
+                            }
+                        )
+                    )
 
             async for message in ws:
                 data = json.loads(message)
                 print(f"Received message: {data}")
 
                 if data["type"] == "offer":
-                    try:
-                        await self.pc.setRemoteDescription(
-                            RTCSessionDescription(
-                                sdp=data["sdp"],
-                                type=data["type"],
-                            )
+                    await self.pc.setRemoteDescription(
+                        RTCSessionDescription(
+                            sdp=data["sdp"],
+                            type=data["type"],
                         )
-                        print("Remote description set successfully")
-                    except Exception as e:
-                        print("setRemoteDescription failed:", e)
-                        raise
+                    )
 
                     camera_track = CameraVideoStreamTrack(self.config)
-                    await self.pc.addTrack(camera_track)
-
-                    print("Sending answer...")
+                    self.pc.addTrack(camera_track)
                     answer = await self.pc.createAnswer()
+
                     await self.pc.setLocalDescription(answer)
-                    await self.ws.send_json(
-                        {
-                            "type": self.pc.localDescription.type,
-                            "sdp": self.pc.localDescription.sdp,
-                        }
+                    print(self.pc.localDescription.type)
+                    await self.ws.send(
+                        json.dumps(
+                            {
+                                "type": self.pc.localDescription.type,
+                                "sdp": self.pc.localDescription.sdp,
+                            }
+                        )
                     )
+                    print("Answer sent successfully")
 
                 elif data["type"] == "ice":
                     await self.pc.addIceCandidate(data["candidate"])

pyproject.toml

@@ -26,3 +26,6 @@ dev = [
 
 [tool.uv.sources]
 transparenc-sdk = { url = "https://github.com/kieranmanning/IntercomClientSDKs/releases/download/v0.9/transparenc_sdk-1.0.0-py3-none-any.whl" }
+
+[tool.ruff]
+lint.extend-select = ["I"] 

pytest.ini

@@ -1,2 +1,2 @@
 [pytest]
-addopts = -s -v
+addopts = -s - v

tests/integration/client.py

@@ -1,61 +1,97 @@
+import os
+
+os.environ["QT_QPA_PLATFORM"] = "xcb"
+
 import asyncio
 import json
+
+import cv2
+import websockets
 from aiortc import RTCPeerConnection, RTCSessionDescription
-from intercomclient.token_store import TokenStore
+
 from intercomclient.config import Config
-import websockets
+from intercomclient.token_store import TokenStore
+
 
+class TestClient:
+    def __init__(self):
+        self.pc = RTCPeerConnection()
+        self.pc.addTransceiver("video", direction="recvonly")
+        self.config = Config()
+        self.output_dir = "/tmp/intercom_client_testing/frames/"
+        if not os.path.exists(self.output_dir):
+            os.makedirs(self.output_dir)
+        self.device_code = TokenStore(self.config, verify=False).load_tokens()[
+            "device_code"
+        ]
+        self.websocket_api_url = (
+            f"{self.config.websocket_api_base_url}/{self.device_code}/"
+        )
+        self.remote_video = None
+
+    async def show_frame(self, frame):
+        await asyncio.to_thread(cv2.imshow, "Remote CCTV Feed", frame)
+
+    async def display_remote_video(self):
+        print("Displaying remote video track...")
+        track = self.remote_video
+        self.frame_count = 0
+        while True:
+            track = self.remote_video
+            if track is None:
+                await asyncio.sleep(1)
+                print("Waiting for remote video track to be available...")
+                continue
+            print("Waiting for video frame...")
+            frame = await track.recv()  # receive an AV frame
+            print("Video frame received, converting to OpenCV format...")
+            img = frame.to_ndarray(format="bgr24")  # convert to OpenCV format
+            print("Displaying video frame...")
+            frame_path = os.path.join(self.output_dir, "frame.jpg")
+            if self.frame_count % 3 == 0:  # every 3rd frame
+                cv2.imwrite(frame_path, img)
+            self.frame_count += 1
+
+    async def test_frame(self):
+        @self.pc.on("track")
+        async def on_track(track):
+            print("Track received:", track.kind)
 
-async def test_frame():
-    pc = RTCPeerConnection()
-    config = Config()
-    device_code = TokenStore(config, verify=False).load_tokens()["device_code"]
-    websocket_api_url = f"{config.websocket_api_base_url}/{device_code}/"
-
-    # # Optionally add a dummy track if server expects one
-    # class DummyTrack(VideoStreamTrack):
-    #     async def recv(self):
-    #         # just return None for testing
-    #         return None
-
-    # pc.addTrack(DummyTrack())
-
-    async with websockets.connect(websocket_api_url) as ws:
-        # Create offer
-        offer = await pc.createOffer()
-        await pc.setLocalDescription(offer)
-
-        await ws.send(json.dumps({"type": "offer", "sdp": pc.localDescription.sdp}))
-
-        # Wait for answer from server
-        async for message in ws:
-            print(f"Test client - Received message: {message}")
-            data = json.loads(message)
-            if data["type"] == "answer":
-                await pc.setRemoteDescription(
-                    RTCSessionDescription(sdp=data["sdp"], type="answer")
-                )
-            elif data["type"] == "ice":
-                await pc.addIceCandidate(data["candidate"])
-
-        print("WebRTC handshake done!")
-
-        # Wait for first frame from server
-        @pc.on("track")
-        def on_track(track):
-            print(f"Track received: {track.kind}")
             if track.kind == "video":
-                # grab a single frame
-                async def grab_frame():
-                    frame = await track.recv()
-                    print(f"Received frame: {frame}")
-                    await pc.close()
-                    return frame
+                self.remote_video = track
+
+        async with websockets.connect(self.websocket_api_url) as ws:
+            # Create offer
+            offer = await self.pc.createOffer()
+            await self.pc.setLocalDescription(offer)
+
+            await ws.send(
+                json.dumps({"type": "offer", "sdp": self.pc.localDescription.sdp})
+            )
+
+            # Wait for answer from server
+            async for message in ws:
+                print(f"Test client - Received message: {message}")
+                data = json.loads(message)
+                if data["type"] == "answer":
+                    await self.pc.setRemoteDescription(
+                        RTCSessionDescription(sdp=data["sdp"], type="answer")
+                    )
+                elif data["type"] == "ice":
+                    await self.pc.addIceCandidate(data["candidate"])
+            print("WebRTC handshake done!")
+
+            # Keep the event loop alive until frame is received
+            await asyncio.sleep(5)
+
+
+async def main():
+    test_client = TestClient()
 
-                asyncio.create_task(grab_frame())
+    signaling_task = asyncio.create_task(test_client.test_frame())
+    display_task = asyncio.create_task(test_client.display_remote_video())
 
-        # Keep the event loop alive until frame is received
-        await asyncio.sleep(5)
+    await asyncio.gather(signaling_task, display_task)
 
 
-asyncio.run(test_frame())
+asyncio.run(main())