Unable to Subscribe to and Play Remote Audio in Janus Video

[MiguelASTavares]

Description
I'm working on a streaming project using the JanusVideoRoomPlugin and have successfully joined a room, published two camera streams, and an audio stream from a microphone. However, I'm encountering difficulties subscribing to the audio stream from another participant in the room and playing it back on a Raspberry Pi.

I've been experimenting with the subscribe_and_start() function but haven't had any success. Despite multiple attempts, the remote audio does not play.

Setup Details
Janus Plugin: JanusVideoRoomPlugin

Transport: Secure WebSockets (wss://)

Device: Raspberry Pi

Streaming:

✅ Successfully publishing two camera streams

✅ Successfully publishing an audio stream

❌ Unable to subscribe to and play back remote audio

Steps Taken
Successfully connected to the Janus server and joined the room.

Published video and audio streams without issues.

Attempted to subscribe to the remote audio stream using subscribe_and_start().

Verified that the other participant’s audio stream is active.

Any guidance would be greatly appreciated!

Thank you for your time and help!

[josephlim94]

Hi. What did you get when you do subscribe_and_start()?

I recommend the following debugging steps:

1. Refer to test\test_plugin_video_room.py unit test, test_publish_and_subscribe() test case. Did you subscribe to the correct participant? Created a correct on_track_created() callback? Are there any python exceptions?
2. If configurations are correct, did you really receive the media? Was on_track_created() really called? What is the input to it?
3. If everything until here is correct, next is you're responsible to playback the audio correctly. For this you can refer to aiortc.contrib.media.MediaRecorder.

Please let me know your progress. I'll help if I can.

[MiguelASTavares]

Thank you for your response. I ran the test\test_plugin_video_room.py code, and it returned "ok."

Regarding the other questions, I’m using plugin_handle.list_participants(room_id) to get the participant ID in the room and then use that as the feed ID. Is this the correct approach?

For example, in this code:

import asyncio
from janus_client import JanusSession, JanusVideoRoomPlugin
from aiortc.contrib.media import MediaPlayer, MediaRecorder
import logging
from aiortc import VideoStreamTrack  # Missing import added
import av
import cv2
import numpy as np

# Enable logging to monitor what's happening under the hood
logging.basicConfig(level=logging.INFO)


class ResizedVideoTrack(VideoStreamTrack):
    def __init__(self, track, width=640, height=480):
        super().__init__()
        self.track = track
        self.width = width
        self.height = height

    async def recv(self):
        frame = await self.track.recv()

        return frame
        # Convert to numpy array in RGB format (instead of BGR)
        img = frame.to_ndarray(format="yuv420p")

        # Resize frame
        # img_resized = cv2.resize(img, (self.width, self.height), interpolation=cv2.INTER_LINEAR)

        # Convert back to av.VideoFrame in RGB format
        new_frame = av.VideoFrame.from_ndarray(img, format="yuv420p")
        new_frame.pts = frame.pts
        new_frame.time_base = frame.time_base
        return new_frame


async def main():
    # Create session
    session = JanusSession(
        base_url="wss://192.168.1.136:8989/",  # Ensure this URL is correct and accessible
    )

    # Create plugin
    plugin_handle = JanusVideoRoomPlugin()

    # Attach to Janus session
    await plugin_handle.attach(session=session)

    # RTSP stream URL (replace this with your RTSP stream URL)
    rtsp_url = "rtsp://admin:123456@192.168.2.123"

    recorder = MediaRecorder("./videocall_record_out.mp4")

    # Room details
    room_id = 1234  # Room ID to join
    username_out = "RTSPStream"  # Display name for the participant

    player = MediaPlayer(rtsp_url, format="rtsp")
    resized_track = ResizedVideoTrack(player.video)

    stream_track = [resized_track]  # Use resized track instead of raw stream
    try:
        # Create a unique publisher ID (you can omit this and let Janus handle it)
        publisher_id = 12345678  # Use a positive integer for the publisher ID

        # Join the room with the publisher_id
        join_response = await plugin_handle.join(
            room_id=room_id, display_name=username_out, publisher_id=publisher_id
        )
        if join_response:
            logging.info(f"Joined room {room_id} as {username_out}")

            # Publish media to the room
            config = {
                "display": username_out,  # Display name for the room
                # Add any other configurations like codecs, bitrate, etc.
            }
            publish_response = await plugin_handle.publish(
                stream_track=stream_track, configuration=config
            )
            if publish_response:
                logging.info(f"Successfully published stream to room {room_id}")
                await recorder.start()  # Start recording if needed
            else:
                logging.error("Failed to publish stream.")
        else:
            logging.error(f"Failed to join room {room_id}")

        # Run until interrupted
        stop_event = asyncio.Event()
        try:
            await stop_event.wait()  # Wait indefinitely until interrupted
        except asyncio.CancelledError:
            pass
    except Exception as e:
        logging.error(f"Error occurred: {str(e)}")
    finally:
        # Clean up: Leave room and destroy plugin/connection
        await recorder.stop()  # Stop recording
        await plugin_handle.leave()  # No room_id argument needed here
        await plugin_handle.destroy()
        await session.destroy()
        logging.info("Cleaned up and destroyed session.")


if __name__ == "__main__":
    try:
        asyncio.run(main())
    except KeyboardInterrupt:
        logging.info("Call interrupted. Cleaning up...")

How could I subscribe to the audio from the video room?

Can I use the recorder = MediaRecorder("./videocall_record_out.mp4") that is being created to capture the audio as well?

[MiguelASTavares]

I also tried to create a code that only subscribes to the audio, but when I run this code, I only see the message print("This is an audio track.") once.

Here is the code I used:

import asyncio
import logging
from janus_client import JanusSession, JanusVideoRoomPlugin
from aiortc.contrib.media import MediaPlayer, MediaStreamTrack
import pyaudio


class AudioPlayer(MediaStreamTrack):
    def __init__(self, track):
        super().__init__()  # Initialize the parent class
        self.track = track
        self.pyaudio = pyaudio.PyAudio()
        self.stream = self.pyaudio.open(
            format=pyaudio.paInt16,
            channels=1,
            rate=48000,
            output=True,
            output_device_index=None,  # Use the default device
            frames_per_buffer=1024,
        )

    async def recv(self):
        frame = await self.track.recv()  # Receive the audio frame

        # Extract the raw audio data from the frame
        audio_data = (
            frame.to_bytes()
        )  # This method might need adjustments depending on your use case

        # Play the audio data using PyAudio
        self.stream.write(audio_data)

        return frame  # Return the frame for further processing (if needed)


async def on_track(track):
    print(
        "---------------------------------------------------------------------------------------------"
    )
    print(track)

    if track.kind == "audio":
        print("This is an audio track.")
    elif track.kind == "video":
        print("This is a video track.")
    else:
        print("Unknown track type.")

    print(
        "---------------------------------------------------------------------------------------------"
    )


async def main():
    session = JanusSession(base_url="wss://192.168.1.136:8989")

    plugin = JanusVideoRoomPlugin()
    await plugin.attach(session=session)

    room_id = 1234
    participantes = await plugin.list_participants(room_id)

    participant_id = []

    for participante in participantes:
        participant_id.append(participante["id"])

    # Use the participant_id to specify the participant to stream
    stream = {"feed": participant_id[0]}

    try:
        # Subscribe to the feed and start processing the tracks
        if await plugin.subscribe_and_start(room_id, on_track, stream):

            print("Subscribed")
        else:

            print("Failed")

        # Wait for an event (like termination or interruption)
        stop_event = asyncio.Event()
        await stop_event.wait()

    except Exception as e:
        logging.error(f"Error occurred: {e}")
    finally:
        # Cleanup
        await plugin.leave()
        await plugin.destroy()
        await session.destroy()
        logging.info("Cleaned up and destroyed session.")


if __name__ == "__main__":
    try:
        # Run the main asyncio loop
        asyncio.run(main())
    except KeyboardInterrupt:
        logging.info("Call interrupted. Cleaning up...")

[josephlim94]

Then it looks like you've successfully subscribed to the audio stream. Now I guess you want to play it using pyaudio.

Try following this guideline:

1. Refer to aiortc.contrib.media.MediaRecorder. Their methods are probably all necessary.
2. Create addTrack() to hold the track you received in async def on_track(track). Create start() to start playing the audio after the track is received and saved in async def on_track(track).
3. Create a "background task" that will continuously get the audio data from the track and then play it. In MediaRecorder it's created using asyncio.ensure_future and it runs the __run_track() method. The main point is that it's an infinite loop calling track.recv() to get the audio data and then play it. Using other parallel processing libraries might be possible, such as threads.
4. Create a stop() function that can end the background task you've created in step 3. Make sure you end it gracefully and cleanup as needed.
5. Really use it in your async def on_track(track) function. Create your audio player object, then add track and start it.
   

Let me know how it goes.

[MiguelASTavares]

Hi, first of all, thank you for your help.

I have tried combining the code I had before with the one present in the test file. Here, I am only trying to create a recording to confirm that I can receive both an audio and a video file. However, this code is not working as expected—it isn't even creating the .mp4 file. Am I doing something wrong?

In the terminal, the prints from the file indicate that the code is receiving a hangup message immediately after the ICE candidate exchange. Can you help me understand what I might be doing wrong?

import asyncio
import logging
from janus_client import JanusSession, JanusVideoRoomPlugin
from aiortc.contrib.media import MediaPlayer, MediaStreamTrack, MediaRecorder
import pyaudio
import os


format = "%(asctime)s: %(message)s"
logging.basicConfig(format=format, level=logging.INFO, datefmt="%H:%M:%S")
logger = logging.getLogger()


async def main():
    session = JanusSession(base_url="wss://192.168.1.136:8989")

    plugin = JanusVideoRoomPlugin()
    await plugin.attach(session=session)

    room_id = 1234
    participants = await plugin.list_participants(room_id)

    output_filename_out = "/home/mast/video_room_record_out.mp4"
    if os.path.exists(output_filename_out):
        os.remove(output_filename_out)
    recorder = MediaRecorder(output_filename_out)

    async def on_track_created(track):
        logger.info("Track %s received" % track.kind)
        if track.kind in ["video", "audio"]:
            recorder.addTrack(track)

        await recorder.start()

    try:
        # Subscribe to the feed and start processing the tracks
        if await plugin.subscribe_and_start(
            room_id=room_id,
            on_track_created=on_track_created,
            stream={"feed": participants[0]["id"]},
        ):
            print("Subscribed")
        else:
            print("Failed")

        # Wait for an event (like termination or interruption)
        stop_event = asyncio.Event()
        await stop_event.wait()

    except Exception as e:
        logging.error(f"An error occurred: {e}")
    finally:
        # Cleanup
        await plugin.leave()
        await plugin.destroy()
        await session.destroy()
        logging.info("Cleaned up and destroyed session.")


if __name__ == "__main__":
    try:
        # Run the main asyncio loop
        asyncio.run(main())
    except KeyboardInterrupt:
        logging.info("Call interrupted. Cleaning up...")

Terminal Output:

16:07:14: Received: {'janus': 'ack', 'session_id': 1603215246519783, 'transaction': '8baca414e2e8478eb089a959dc753c75'}
16:07:14: Received: {'janus': 'event', 'session_id': 1603215246519783, 'transaction': '8baca414e2e8478eb089a959dc753c75', 'sender': 1064975660855540, 'plugindata': {'plugin': 'janus.plugin.videoroom', 'data': {'videoroom': 'event', 'room': 1234, 'started': 'ok'}}}
16:07:14: Successfully subscribed to stream.
16:07:14: Connection(0) Discovered peer reflexive candidate Candidate(pjTVp5s3f3 1 udp 1847591423 192.168.1.100 57943 typ prflx)
16:07:14: Connection(0) Check CandidatePair(('192.168.1.176', 49131) -> ('192.168.1.100', 57943)) State.WAITING -> State.IN_PROGRESS
16:07:14: Connection(0) Check CandidatePair(('192.168.1.176', 49131) -> ('192.168.1.100', 57943)) State.IN_PROGRESS -> State.SUCCEEDED
16:07:14: Connection(0) ICE completed
16:07:14: Received: {'janus': 'hangup', 'session_id': 1603215246519783, 'sender': 1064975660855540, 'reason': 'DTLS alert'}
16:07:14: Unimplemented response handle: {'janus': 'hangup', 'session_id': 1603215246519783, 'sender': 1064975660855540, 'reason': 'DTLS alert'}
16:07:38: Send: {"janus": "keepalive", "transaction": "90439704571b43f481c5a0d20dfaaba4", "session_id": 1603215246519783}
16:07:38: Received: {'janus': 'ack', 'session_id': 1603215246519783, 'transaction': '90439704571b43f481c5a0d20dfaaba4'}
16:07:38: Async event for session: {'janus': 'ack', 'session_id': 1603215246519783, 'transaction': '90439704571b43f481c5a0d20dfaaba4'}
16:07:47: Connection(0) Consent to send expired

Thank you.

[josephlim94]

Looks like something is wrong with the webrtc connection. I can't tell. Is everything ok at the publisher side? Maybe check the logs of your Janus server too.