Сервер видеостриминга WebRTC (Python) застрял на получении кадров

Предыстория:

Я работаю над сервером видеостриминга WebRTC с использованием aiortc и aiohttp на Python. Цель состоит в том, чтобы сервер захватывал кадры с IP-камеры, обрабатывал их и затем стримил на клиент. Для тестирования это упрощённая версия кода, в которой кадры захватываются с IP-камеры и непосредственно стримятся на клиент без дополнительной обработки (в оригинальной версии кадры обрабатываются перед отправкой клиенту).

Проблема:

Несмотря на успешную переговора ICE и получение дорожки на стороне клиента, клиент зависает при попытке получить видеокадры, и кажется, что метод recv() сервера (который захватывает и отправляет кадры) никогда не срабатывает.

Я объясню настройку и поделюсь соответствующим кодом как для сервера (stream.py), так и для клиента (client.py).

Сервер: stream.py

import cv2
import asyncio
import logging
import time
import numpy as np
from aiortc import VideoStreamTrack, RTCPeerConnection, RTCSessionDescription
from aiortc.mediastreams import VideoFrame
from aiohttp import web

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("WebRTC")

# WebRTC дорожка для видеостриминга
class VideoProcessorTrack(VideoStreamTrack):
    def __init__(self, camera_url, desired_fps=15):
        super().__init__()
        self.camera_url = camera_url
        self.cap = cv2.VideoCapture(self.camera_url)
        self.desired_fps = desired_fps
        self.frame_interval = 1.0 / self.desired_fps  # Временной интервал между кадрами

        if not self.cap.isOpened():
            logger.error(f"Не удалось открыть поток камеры: {self.camera_url}")
        else:
            logger.info(f"Поток камеры успешно открыт: {self.camera_url}")

    async def recv(self):
        """Захват и обработка кадров с камеры."""
        logger.info("Вход в метод recv() для захвата кадра")
        await asyncio.sleep(self.frame_interval)

        # Захват кадра
        ret, frame = self.cap.read()
        if not ret:
            logger.warning("Не удалось считать кадр, отправляем пустой кадр...")
            frame = np.zeros((480, 640, 3), dtype=np.uint8)  # Заглушка черный кадр
        else:
            logger.info("Успешно захвачен кадр с камеры")

        # Конвертация кадра для WebRTC
        frame_rgb = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
        video_frame = VideoFrame.from_ndarray(frame_rgb, format="rgb24")
        video_frame.pts = time.time()
        video_frame.time_base = 1 / self.desired_fps
        logger.info("Кадр готов к отправке")
        return video_frame

async def offer(request):
    """Обработать входящее предложение WebRTC и отправить ответ."""
    try:
        logger.info("Получен запрос предложения от клиента")
        params = await request.json()
        logger.info(f"Разобрано предложение: {params}")

        offer = RTCSessionDescription(sdp=params["sdp"], type=params["type"])
        pc = RTCPeerConnection()

        async def on_iceconnectionstatechange():
            logger.info(f"Состояние ICE соединения: {pc.iceConnectionState}")
            if pc.iceConnectionState == "completed":
                logger.info("ICE соединение успешно завершено!")
            elif pc.iceConnectionState == "failed":
                logger.error("ICE соединение провалено.")

        pc.on("iceconnectionstatechange", on_iceconnectionstatechange)

        # Добавить видеодорожку в соединение
        player = VideoProcessorTrack(camera_url="http://129.125.136.20/axis-cgi/mjpg/video.cgi?camera=1", desired_fps=15)
        logger.info("Добавление видеодорожки в соединение")
        pc.addTrack(player)

        # Логировать каждый раз, когда мы получаем запрос на кадр
        @pc.on("track")
        async def on_track(track):
            logger.info(f"Дорожка {track.kind} получена на сервере")

        # Обработка обмена предложения/ответа WebRTC
        await pc.setRemoteDescription(offer)
        answer = await pc.createAnswer()
        await pc.setLocalDescription(answer)

        logger.info("Локальное описание установлено с ответом")
        return web.json_response({"sdp": pc.localDescription.sdp, "type": pc.localDescription.type})
    
    except Exception as e:
        logger.error(f"Ошибка в обработке предложения: {e}")
        return web.Response(text="Внутренняя ошибка сервера", status=500)

app = web.Application()
app.router.add_post("/offer", offer)

if __name__ == "__main__":
    logger.info("Запуск сервера видеостриминга WebRTC...")
    web.run_app(app, port=8081)

Клиент: client.py

import asyncio
import cv2
from aiortc import RTCPeerConnection, RTCSessionDescription, MediaStreamTrack
import aiohttp
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("WebRTC Client")

class DummyVideoTrack(MediaStreamTrack):
    kind = "video"

    async def recv(self):
        return None

async def run_client():
    pc = RTCPeerConnection()

    dummy_track = DummyVideoTrack()
    pc.addTrack(dummy_track)

    @pc.on("track")
    async def on_track(track):
        logger.info(f"Получение дорожки: {track.kind}")

        if track.kind == "video":
            cv2.namedWindow("WebRTC Stream", cv2.WINDOW_NORMAL)

            while True:
                try:
                    frame = await track.recv()  # Получение кадра из дорожки
                    img = frame.to_ndarray(format="bgr24")  # Конвертация кадра в массив numpy
                    logger.info(f"Кадр получен: {img.shape}")

                    # Отображение кадра с помощью OpenCV
                    cv2.imshow("WebRTC Stream", img)

                    if cv2.waitKey(1) & 0xFF == ord('q'):
                        break  # Выход при нажатии 'q'
                except Exception as e:
                    logger.error(f"Ошибка при обработке кадра: {e}")
                    break  # Прерывание цикла в случае ошибок

            cv2.destroyAllWindows()  # Закрытие окна по завершении
            track.stop()

    # Создание предложения WebRTC
    offer = await pc.createOffer()
    await pc.setLocalDescription(offer)

    # Отправка предложения на сервер
    offer_payload = {
        "sdp": pc.localDescription.sdp,
        "type": pc.localDescription.type,
    }
    logger.info("Отправка предложения на сервер")

    async with aiohttp.ClientSession() as session:
        async with session.post("http://localhost:8081/offer", json=offer_payload) as response:
            if response.status != 200:
                logger.error(f"Ошибка сервера: {response.status}")
                return
            answer = await response.json()
            logger.info("Получен ответ от сервера")

        await pc.setRemoteDescription(RTCSessionDescription(sdp=answer["sdp"], type=answer["type"]))
        logger.info("Удаленное описание установлено на клиенте")

    # Поддержание соединения открытым
    await asyncio.Future()

if __name__ == "__main__":
    asyncio.run(run_client())

Что работает:

• Клиент успешно отправляет предложение на сервер, а сервер отвечает ответом.
• Переговоры ICE завершаются успешно, и находится валидная пара кандидатов.
• Клиент логирует получение дорожки: видео, указывая, что сервер отправляет видеодорожку.

Проблема:

Несмотря на успешное соединение ICE и обмен медиа-дорожками, клиент зависает на:

frame = await track.recv()

Кажется, что метод recv() в VideoProcessorTrack сервера (который захватывает и отправляет кадры) никогда не срабатывает. Это означает, что клиент ждет кадры, которые никогда не отправляются.

С чем мне нужна помощь:

• Почему метод recv() на сервере не срабатывает?
• Есть ли что-то неправильное в переговорах дорожки WebRTC между клиентом и сервером, что мешает серверу отправлять кадры?
• Как я могу правильно запустить сервер, чтобы он начал отправлять кадры клиенту?

Любая помощь или советы будут очень полезны!