服务端改用 aiortc 实现 WebRTC 视频流接收

- handle_client.py: 替换原始 JPEG WebSocket 为 aiortc RTCPeerConnection

- 实现 SDP offer/answer 协商和 ICE candidate 交换

- 通过 track.recv() 接收 RTP 视频帧并 cv2.imshow 显示

- 服务端口改为 8765 匹配 Android 端

- 新增 requirements.txt: aiortc, websockets, opencv-python 等

- .gitignore 添加 __pycache__

.gitignore

@@ -1,2 +1,3 @@
 .venv/
 .idea/
+__pycache__/

handle_client.py

@@ -1,59 +1,119 @@
 import asyncio
+import json
+import re
 import websockets
 import cv2
-import numpy as np
 from loguru import logger
+from aiortc import RTCPeerConnection, RTCSessionDescription, RTCIceCandidate
 
 
 async def handle_client(websocket):
     client = websocket.remote_address
     logger.info(f"Client connected: {client}")
 
+    pc = RTCPeerConnection()
+    video_task = None
+
+    def parse_ice(data):
+        match = re.match(
+            r'candidate:(\S+) (\d) (\S+) (\d+) (\S+) (\d+) typ (\S+)(?: raddr (\S+) rport (\d+))?',
+            data["candidate"]
+        )
+        if not match:
+            return None
+        g = match.groups()
+        cand = RTCIceCandidate(
+            foundation=g[0],
+            component=int(g[1]),
+            protocol=g[2].lower(),
+            priority=int(g[3]),
+            ip=g[4],
+            port=int(g[5]),
+            type=g[6],
+            relatedAddress=g[7],
+            relatedPort=int(g[8]) if g[8] else None,
+        )
+        cand.sdpMid = data.get("sdpMid")
+        cand.sdpMLineIndex = data.get("sdpMLineIndex", 0)
+        return cand
+
+    async def receive_video(track):
+        logger.info("Start receiving video frames")
         frame_count = 0
-
         try:
-        async for message in websocket:
+            while True:
+                frame = await track.recv()
                 frame_count += 1
-
+                img = frame.to_ndarray(format="bgr24")
-            data = np.frombuffer(message, dtype=np.uint8)
+                cv2.imshow("Android Camera (WebRTC)", img)
-            frame = cv2.imdecode(data, cv2.IMREAD_COLOR)
-
-            if frame is None:
-                logger.warning(f"Decode frame failed from client={client}")
-                continue
 
                 if frame_count % 100 == 0:
-                logger.info(f"Received frames={frame_count}, client={client}, size={len(message)} bytes")
+                    logger.info(f"Received {frame_count} frames, shape={img.shape}")
-
-            cv2.imshow("Android Camera", frame)
 
                 if cv2.waitKey(1) & 0xFF == 27:
                     logger.info("ESC pressed, closing display")
                     break
+        except asyncio.CancelledError:
+            logger.info("Video receive task cancelled")
+        except Exception as e:
+            logger.error(f"Video receive error: {e}")
+
+    @pc.on("track")
+    async def on_track(track):
+        logger.info(f"Track received: kind={track.kind}")
+        if track.kind == "video":
+            nonlocal video_task
+            video_task = asyncio.ensure_future(receive_video(track))
+
+    @pc.on("iceconnectionstatechange")
+    async def on_iceconnectionstatechange():
+        logger.info(f"ICE state: {pc.iceConnectionState}")
+        if pc.iceConnectionState in ("failed", "closed", "disconnected"):
+            await pc.close()
+
+    try:
+        async for message in websocket:
+            data = json.loads(message)
+            msg_type = data.get("type")
+
+            if msg_type == "offer":
+                offer = RTCSessionDescription(sdp=data["sdp"], type="offer")
+                await pc.setRemoteDescription(offer)
+
+                answer = await pc.createAnswer()
+                await pc.setLocalDescription(answer)
+
+                await websocket.send(json.dumps({
+                    "type": "answer",
+                    "sdp": pc.localDescription.sdp,
+                }))
+
+            elif msg_type == "candidate":
+                cand = parse_ice(data)
+                if cand:
+                    await pc.addIceCandidate(cand)
 
     except websockets.ConnectionClosed:
         logger.info(f"Client disconnected: {client}")
-
     except Exception as e:
-        logger.exception(f"WebSocket error, client={client}, error={e}")
+        logger.exception(f"Error: {e}")
-
     finally:
-        logger.info(f"Connection closed: client={client}, total_frames={frame_count}")
+        if video_task:
+            video_task.cancel()
+            try:
+                await video_task
+            except asyncio.CancelledError:
+                pass
+        await pc.close()
         cv2.destroyAllWindows()
+        logger.info(f"Connection closed: {client}")
 
 
 async def main():
     host = "0.0.0.0"
     port = 8765
-
+    logger.info(f"WebRTC signaling server: ws://{host}:{port}")
-    logger.info(f"WebSocket server started: ws://{host}:{port}")
+    async with websockets.serve(handle_client, host, port, max_size=10 * 1024 * 1024):
-
-    async with websockets.serve(
-        handle_client,
-        host,
-        port,
-        max_size=10 * 1024 * 1024
-    ):
         await asyncio.Future()
 
 

requirements.txt

@@ -0,0 +1,5 @@
+aiortc>=1.9.0
+websockets>=13.0
+opencv-python>=4.10.0
+numpy>=2.0.0
+loguru>=0.7.0