Unable to record to file using the Janus Example

[msalmonw]

I'm using the janus example. I have modified the script to have two versions. 1 only publishes the default green frames, while the other only subscribes and records the stream. My Janus instance is deployed on the cloud, the recorder always crashes. The complete log is given below, if anyone can help it would be really nice. I'm stuck and I don't know what to do.

C:\Users\bring\Documents\GitHub\janus_stream>python janus.py http://194.163.158.53:8088/janus --record-to sample2.mp4 id: 2399281696872851, display: aiortc INFO:aioice.ice:Connection(0) Could not bind to 169.254.23.128 - [WinError 10049] The requested address is not valid in its context INFO:aioice.ice:Connection(0) Could not bind to 169.254.61.247 - [WinError 10049] The requested address is not valid in its context INFO:aioice.ice:Connection(0) Could not bind to 169.254.228.123 - [WinError 10049] The requested address is not valid in its context INFO:aioice.ice:Connection(0) Check CandidatePair(('10.214.87.116', 62581) -> ('172.18.0.2', 49096)) State.FROZEN -> State.WAITING INFO:aioice.ice:Connection(0) Check CandidatePair(('10.214.87.116', 62581) -> ('194.163.158.53', 49096)) State.FROZEN -> State.WAITING INFO:aioice.ice:Connection(0) Check CandidatePair(('10.214.87.116', 62581) -> ('172.18.0.2', 49096)) State.WAITING -> State.IN_PROGRESS INFO:aioice.ice:Connection(0) Check CandidatePair(('10.214.87.116', 62581) -> ('194.163.158.53', 49096)) State.WAITING -> State.IN_PROGRESS INFO:aioice.ice:Connection(0) Check CandidatePair(('10.214.87.116', 62581) -> ('194.163.158.53', 49096)) State.IN_PROGRESS -> State.SUCCEEDED INFO:aioice.ice:Connection(0) ICE completed Track audio received Track video received INFO:aioice.ice:Connection(1) Could not bind to 169.254.23.128 - [WinError 10049] The requested address is not valid in its context INFO:aioice.ice:Connection(1) Could not bind to 169.254.61.247 - [WinError 10049] The requested address is not valid in its context INFO:aioice.ice:Connection(1) Could not bind to 169.254.228.123 - [WinError 10049] The requested address is not valid in its context INFO:aioice.ice:Connection(1) Check CandidatePair(('10.214.87.116', 62582) -> ('172.18.0.2', 38714)) State.FROZEN -> State.WAITING INFO:aioice.ice:Connection(1) Check CandidatePair(('10.214.87.116', 62582) -> ('194.163.158.53', 38714)) State.FROZEN -> State.WAITING INFO:aioice.ice:Connection(1) Check CandidatePair(('10.214.87.116', 62582) -> ('172.18.0.2', 38714)) State.WAITING -> State.IN_PROGRESS INFO:aioice.ice:Connection(1) Check CandidatePair(('10.214.87.116', 62582) -> ('194.163.158.53', 38714)) State.WAITING -> State.IN_PROGRESS Exchanging media INFO:aioice.ice:Connection(1) Check CandidatePair(('10.214.87.116', 62582) -> ('194.163.158.53', 38714)) State.IN_PROGRESS -> State.SUCCEEDED INFO:aioice.ice:Connection(1) ICE completed Exception in thread video-decoder: Traceback (most recent call last): File "C:\Users\bring\AppData\Local\Programs\Python\Python311\Lib\threading.py", line 1045, in _bootstrap_inner self.run() File "C:\Users\bring\AppData\Local\Programs\Python\Python311\Lib\threading.py", line 982, in run self._target(*self._args, **self._kwargs) File "C:\Users\bring\AppData\Local\Programs\Python\Python311\Lib\site-packages\aiortc\rtcrtpreceiver.py", line 68, in decoder_worker for frame in decoder.decode(encoded_frame): ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ File "C:\Users\bring\AppData\Local\Programs\Python\Python311\Lib\site-packages\aiortc\codecs\vpx.py", line 162, in decode return cast(list[Frame], self.codec.decode(packet)) ^^^^^^^^^^^^^^^^^^^^^^^^^ File "av/codec/context.pyx", line 433, in av.codec.context.CodecContext.decode cpdef decode(self, Packet packet=None): File "av/codec/context.pyx", line 448, in av.codec.context.CodecContext.decode for frame in self._send_packet_and_recv(packet): File "av/codec/context.pyx", line 340, in av.codec.context.CodecContext._send_packet_and_recv err_check(res, "avcodec_send_packet()") File "av/error.pyx", line 424, in av.error.err_check raise cls(code, message, filename, log) av.error.InvalidDataError: [Errno 1094995529] Invalid data found when processing input: 'avcodec_send_packet()'

[msalmonw]

This was the example of green frames, and I was on aiortc version 1.10.1. Today I set up an RPI as a publisher with a ribbon camera attached. I was able to receive the stream. Both record to file and live stream. The problem that occurred for then was:

video/rtx' MIMEType retransmission incorrectly handled as a codec needing a decoder

To resolve this I switched to aiortc version 1.12. On 1.12 even on fetching the valid RPI stream -which was working on 1.10.1- again gave the following issue:

File "C:\Users\bring\AppData\Local\Programs\Python\Python311\Lib\site-packages\aiortc\codecs\vpx.py", line 162, in decode return cast(list[Frame], self.codec.decode(packet)) ^^^^^^^^^^^^^^^^^^^^^^^^^ File "av/codec/context.pyx", line 433, in av.codec.context.CodecContext.decode cpdef decode(self, Packet packet=None): File "av/codec/context.pyx", line 448, in av.codec.context.CodecContext.decode for frame in self._send_packet_and_recv(packet): File "av/codec/context.pyx", line 340, in av.codec.context.CodecContext._send_packet_and_recv err_check(res, "avcodec_send_packet()") File "av/error.pyx", line 424, in av.error.err_check raise cls(code, message, filename, log) av.error.InvalidDataError: [Errno 1094995529] Invalid data found when processing input: 'avcodec_send_packet()

So, then I downgraded to V1.11. And now the stream can be viewed indefinitely without either 'rtx' issue or the 'invalid data' issue.

I'm now really confused as to what's happening, what is actually causing the invalid data issue on the latest version, and why the invalid data issue occurred for the dummy stream as well in the first place.

I hope I explained my working and investigation in a good way. If there's still any confusion, I can do more testing for you guys with the different versions that I've been using.

Hope to get an explanation for this.

Cheers!

[msalmonw]

Now after running my subscriber script powered by V1.11. I get the following issue after a whille:

C:\Users\bring\Documents\GitHub\janus_stream>python subscriber_opencv.py http://194.163.158.53:8088/janus id: 7818223790335686, display: picam2 INFO:aioice.ice:Connection(0) Could not bind to 169.254.23.128 - [WinError 10049] The requested address is not valid in its context INFO:aioice.ice:Connection(0) Could not bind to 169.254.61.247 - [WinError 10049] The requested address is not valid in its context INFO:aioice.ice:Connection(0) Could not bind to 169.254.228.123 - [WinError 10049] The requested address is not valid in its context INFO:aioice.ice:Connection(0) Check CandidatePair(('192.168.10.8', 59984) -> ('172.18.0.2', 45564)) State.FROZEN -> State.WAITING INFO:aioice.ice:Connection(0) Check CandidatePair(('192.168.10.8', 59984) -> ('194.163.158.53', 45564)) State.FROZEN -> State.WAITING INFO:aioice.ice:Connection(0) Check CandidatePair(('192.168.10.8', 59984) -> ('172.18.0.2', 45564)) State.WAITING -> State.IN_PROGRESS INFO:aioice.ice:Connection(0) Check CandidatePair(('192.168.10.8', 59984) -> ('194.163.158.53', 45564)) State.WAITING -> State.IN_PROGRESS Displaying video... Press 'q' in the video window to quit. INFO:aioice.ice:Connection(0) Check CandidatePair(('192.168.10.8', 59984) -> ('194.163.158.53', 45564)) State.IN_PROGRESS -> State.SUCCEEDED INFO:aioice.ice:Connection(0) ICE completed ERROR:asyncio:Task exception was never retrieved future: <Task finished name='Task-15' coro=<subscribe.<locals>.on_track.<locals>.display_video() done, defined at C:\Users\bring\Documents\GitHub\janus_stream\subscriber_opencv.py:149> exception=MediaStreamError()> Traceback (most recent call last): File "C:\Users\bring\Documents\GitHub\janus_stream\subscriber_opencv.py", line 151, in display_video frame = await local_track.recv() ^^^^^^^^^^^^^^^^^^^^^^^^ File "C:\Users\bring\AppData\Local\Programs\Python\Python311\Lib\site-packages\aiortc\contrib\media.py", line 526, in recv raise MediaStreamError aiortc.mediastreams.MediaStreamError

And the code that I use:

``
import argparse
import asyncio
import logging
import random
import string
import time
import threading
import cv2
import numpy as np

import aiohttp
from aiortc import RTCPeerConnection, RTCSessionDescription, RTCConfiguration, RTCIceServer, VideoStreamTrack
from aiortc.contrib.media import MediaRelay

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("janus-subscriber")

RTC_CONFIG = RTCConfiguration(iceServers=[
RTCIceServer(urls="stun:stun2.l.google.com:19302")
])

pcs = set()

relay = MediaRelay()

def transaction_id():
return "".join(random.choice(string.ascii_letters) for x in range(12))

class JanusPlugin:
def init(self, session, url):
self._queue = asyncio.Queue()
self._session = session
self._url = url

async def send(self, payload):
    message = {"janus": "message", "transaction": transaction_id()}
    message.update(payload)
    async with self._session._http.post(self._url, json=message) as response:
        data = await response.json()
        assert data["janus"] == "ack"

    response = await self._queue.get()
    assert response["transaction"] == message["transaction"]
    return response

class JanusSession:
def init(self, url):
self._http = None
self._poll_task = None
self._plugins = {}
self._root_url = url
self._session_url = None

async def attach(self, plugin_name: str) -> JanusPlugin:
    message = {
        "janus": "attach",
        "plugin": plugin_name,
        "transaction": transaction_id(),
    }
    async with self._http.post(self._session_url, json=message) as response:
        data = await response.json()
        assert data["janus"] == "success"
        plugin_id = data["data"]["id"]
        plugin = JanusPlugin(self, self._session_url + "/" + str(plugin_id))
        self._plugins[plugin_id] = plugin
        return plugin

async def create(self):
    self._http = aiohttp.ClientSession()
    message = {"janus": "create", "transaction": transaction_id()}
    async with self._http.post(self._root_url, json=message) as response:
        data = await response.json()
        assert data["janus"] == "success"
        session_id = data["data"]["id"]
        self._session_url = self._root_url + "/" + str(session_id)

    self._poll_task = asyncio.ensure_future(self._poll())

async def destroy(self):
    if self._poll_task:
        self._poll_task.cancel()
        self._poll_task = None

    if self._session_url:
        message = {"janus": "destroy", "transaction": transaction_id()}
        async with self._http.post(self._session_url, json=message) as response:
            data = await response.json()
            assert data["janus"] == "success"
        self._session_url = None

    if self._http:
        await self._http.close()
        self._http = None

async def _poll(self):
    while True:
        params = {"maxev": 1, "rid": int(time.time() * 1000)}
        async with self._http.get(self._session_url, params=params) as response:
            data = await response.json()
            if data["janus"] == "event":
                plugin = self._plugins.get(data["sender"], None)
                if plugin:
                    await plugin._queue.put(data)
                else:
                    print(data)

class OpenCVVideoSink:
def init(self):
self._frame_queue = asyncio.Queue()
threading.Thread(target=self._display_loop, daemon=True).start()

async def consume(self, track: VideoStreamTrack):
    while True:
        frame = await track.recv()
        img = frame.to_ndarray(format="bgr24")
        await self._frame_queue.put(img)

def _display_loop(self):
    while True:
        frame = asyncio.run(self._frame_queue.get())
        if frame is None:
            break
        cv2.imshow("Janus Stream", frame)
        if cv2.waitKey(1) & 0xFF == ord("q"):
            break
    cv2.destroyAllWindows()

async def subscribe(session, room, feed):
pc = RTCPeerConnection()
pcs.add(pc)

sink = OpenCVVideoSink()

@pc.on("track")
async def on_track(track):
    #codec = track.codec
    #if codec and "rtx" in codec.mimeType.lower():
    #    logger.warning(f"Ignoring RTX track: {codec.mimeType}")
    #    return

    if track.kind == "video":
        local_track = relay.subscribe(track)

        async def display_video():
            while True:
                frame = await local_track.recv()
                img = frame.to_ndarray(format="bgr24")
                cv2.imshow("Video", img)
                if cv2.waitKey(1) & 0xFF == ord("q"):
                    break
            cv2.destroyAllWindows()

        asyncio.ensure_future(display_video())

    elif track.kind == "audio":
        print("Audio track received (not played in this example).")

# subscribe
plugin = await session.attach("janus.plugin.videoroom")
response = await plugin.send(
    {"body": {"request": "join", "ptype": "subscriber", "room": room, "feed": feed}}
)

# apply offer
await pc.setRemoteDescription(
    RTCSessionDescription(
        sdp=response["jsep"]["sdp"], type=response["jsep"]["type"]
    )
)

# send answer
await pc.setLocalDescription(await pc.createAnswer())
await plugin.send(
    {
        "body": {"request": "start"},
        "jsep": {
            "sdp": pc.localDescription.sdp,
            "trickle": False,
            "type": pc.localDescription.type,
        },
    }
)

async def run(room, session):
await session.create()

# join video room
plugin = await session.attach("janus.plugin.videoroom")
response = await plugin.send(
    {
        "body": {
            "display": "aiortc",
            "ptype": "publisher",
            "request": "join",
            "room": room,
        }
    }
)

publishers = response["plugindata"]["data"].get("publishers", [])
for publisher in publishers:
    print("id: %(id)s, display: %(display)s" % publisher)

if publishers:
    await subscribe(session=session, room=room, feed=publishers[0]["id"])

    print("Displaying video... Press 'q' in the video window to quit.")
    while True:
        await asyncio.sleep(1)

if name == "main":
parser = argparse.ArgumentParser(description="Janus")
parser.add_argument("url", help="Janus root URL, e.g. http://localhost:8088/janus")
parser.add_argument("--room", type=int, default=1234, help="Video room ID (default: 1234)")
parser.add_argument("--verbose", "-v", action="count")
args = parser.parse_args()

if args.verbose:
    logging.basicConfig(level=logging.DEBUG)

session = JanusSession(args.url)
loop = asyncio.get_event_loop()
try:
    loop.run_until_complete(run(room=args.room, session=session))
except KeyboardInterrupt:
    pass
finally:
    loop.run_until_complete(session.destroy())
    coros = [pc.close() for pc in pcs]
    loop.run_until_complete(asyncio.gather(*coros))``

How can I make this reliable?

[msalmonw]

Any help or just your insights would really be appreciated. im very new webrtc and janus