RTP Transport

class RTCRtpSender:
    """RTP sender for outgoing media streams."""
    
    @property
    def kind(self) -> str:
        """Media kind: "audio" or "video" """
    
    @property
    def track(self) -> MediaStreamTrack:
        """Associated media track (None if no track)"""
    
    @property
    def transport(self) -> RTCDtlsTransport:
        """Associated DTLS transport"""
    
    @classmethod
    def getCapabilities(cls, kind: str) -> RTCRtpCapabilities:
        """
        Get sender capabilities for media kind.
        
        Parameters:
        - kind (str): Media kind ("audio" or "video")
        
        Returns:
        RTCRtpCapabilities: Supported codecs and extensions
        """
    
    async def getStats(self) -> RTCStatsReport:
        """
        Get sender statistics.
        
        Returns:
        RTCStatsReport: Outbound RTP stream statistics
        """
    
    async def replaceTrack(self, track: MediaStreamTrack) -> None:
        """
        Replace the current track without renegotiation.
        
        Parameters:
        - track (MediaStreamTrack): New track to send (None to stop sending)
        """
    
    async def send(self, parameters: RTCRtpParameters) -> None:
        """
        Start sending with RTP parameters.
        
        Parameters:
        - parameters (RTCRtpParameters): RTP configuration
        """
    
    def setStreams(self, streams) -> None:
        """
        Set associated media streams.
        
        Parameters:
        - streams: List of media streams (currently unused)
        """
    
    def stop(self) -> None:
        """Stop the sender"""

class RTCRtpReceiver:
    """RTP receiver for incoming media streams."""
    
    @property
    def track(self) -> MediaStreamTrack:
        """Associated received media track"""
    
    @property
    def transport(self) -> RTCDtlsTransport:
        """Associated DTLS transport"""
    
    @classmethod
    def getCapabilities(cls, kind: str) -> RTCRtpCapabilities:
        """
        Get receiver capabilities for media kind.
        
        Parameters:
        - kind (str): Media kind ("audio" or "video")
        
        Returns:
        RTCRtpCapabilities: Supported codecs and extensions
        """
    
    def getContributingSources(self) -> list:
        """
        Get contributing sources.
        
        Returns:
        list: List of RTCRtpContributingSource objects
        """
    
    async def getStats(self) -> RTCStatsReport:
        """
        Get receiver statistics.
        
        Returns:
        RTCStatsReport: Inbound RTP stream statistics
        """
    
    def getSynchronizationSources(self) -> list:
        """
        Get synchronization sources.
        
        Returns:
        list: List of RTCRtpSynchronizationSource objects
        """
    
    async def receive(self, parameters: RTCRtpParameters) -> None:
        """
        Start receiving with RTP parameters.
        
        Parameters:
        - parameters (RTCRtpParameters): RTP configuration
        """
    
    def stop(self) -> None:
        """Stop the receiver"""

class RTCRtpTransceiver:
    """RTP transceiver combining sender and receiver."""
    
    @property
    def currentDirection(self) -> str:
        """Negotiated direction: "sendrecv", "sendonly", "recvonly", "inactive", or None"""
    
    @property
    def direction(self) -> str:
        """Preferred direction: "sendrecv", "sendonly", "recvonly", "inactive" """
    
    @direction.setter
    def direction(self, value: str) -> None:
        """Set preferred direction"""
    
    @property
    def kind(self) -> str:
        """Media kind: "audio" or "video" """
    
    @property
    def mid(self) -> str:
        """Media identifier (None if not negotiated)"""
    
    @property
    def receiver(self) -> RTCRtpReceiver:
        """Associated RTP receiver"""
    
    @property
    def sender(self) -> RTCRtpSender:
        """Associated RTP sender"""
    
    @property
    def stopped(self) -> bool:
        """Whether transceiver is stopped"""
    
    def setCodecPreferences(self, codecs: list) -> None:
        """
        Override codec preferences for this transceiver.
        
        Parameters:
        - codecs (list): List of RTCRtpCodecCapability objects in preferred order
        """
    
    def stop(self) -> None:
        """Stop the transceiver"""

class RTCRtpSynchronizationSource:
    """Synchronization source information."""
    
    @property
    def audioLevel(self) -> float:
        """Audio level (0.0 to 1.0, None for video)"""
    
    @property
    def source(self) -> int:
        """SSRC identifier"""
    
    @property
    def timestamp(self) -> float:
        """Reception timestamp"""

class RTCRtpContributingSource:
    """Contributing source information."""
    
    @property
    def audioLevel(self) -> float:
        """Audio level (0.0 to 1.0, None for video)"""
    
    @property
    def source(self) -> int:
        """CSRC identifier"""
    
    @property
    def timestamp(self) -> float:
        """Reception timestamp"""

import aiortc
import asyncio

async def basic_sender():
    pc = aiortc.RTCPeerConnection()
    
    # Create and add track
    video_track = aiortc.VideoStreamTrack()
    sender = pc.addTrack(video_track)
    
    print(f"Sender kind: {sender.kind}")
    print(f"Sender track: {sender.track}")
    
    # Get sender capabilities
    capabilities = aiortc.RTCRtpSender.getCapabilities("video")
    print(f"Video codecs: {[codec.mimeType for codec in capabilities.codecs]}")
    
    # Replace track with different one
    new_track = aiortc.VideoStreamTrack()
    await sender.replaceTrack(new_track)
    print("Track replaced")
    
    # Get statistics
    stats = await sender.getStats()
    print(f"Sender stats: {stats}")

async def handle_receiver():
    pc = aiortc.RTCPeerConnection()
    
    @pc.on("track")
    def on_track(track):
        print(f"Received {track.kind} track")
        
        # Get the receiver for this track
        for transceiver in pc.getTransceivers():
            if transceiver.receiver.track == track:
                receiver = transceiver.receiver
                break
        
        # Get receiver capabilities
        capabilities = aiortc.RTCRtpReceiver.getCapabilities(track.kind)
        print(f"Receiver capabilities: {capabilities}")
        
        # Monitor synchronization sources
        async def monitor_sources():
            while True:
                sync_sources = receiver.getSynchronizationSources()
                contrib_sources = receiver.getContributingSources()
                
                if sync_sources:
                    print(f"Sync sources: {[s.source for s in sync_sources]}")
                if contrib_sources:
                    print(f"Contributing sources: {[c.source for c in contrib_sources]}")
                
                await asyncio.sleep(1)
        
        # Start monitoring in background
        asyncio.create_task(monitor_sources())

async def manage_transceivers():
    pc = aiortc.RTCPeerConnection()
    
    # Add transceiver for bidirectional audio
    audio_transceiver = pc.addTransceiver("audio", direction="sendrecv")
    print(f"Audio transceiver: {audio_transceiver.kind}, direction: {audio_transceiver.direction}")
    
    # Add send-only video transceiver
    video_track = aiortc.VideoStreamTrack()
    video_transceiver = pc.addTransceiver(video_track, direction="sendonly")
    print(f"Video transceiver: {video_transceiver.kind}, direction: {video_transceiver.direction}")
    
    # Set codec preferences
    video_capabilities = aiortc.RTCRtpSender.getCapabilities("video")
    h264_codecs = [codec for codec in video_capabilities.codecs if "H264" in codec.mimeType]
    if h264_codecs:
        video_transceiver.setCodecPreferences(h264_codecs)
        print("Set H.264 codec preference")
    
    # List all transceivers
    transceivers = pc.getTransceivers()
    for i, t in enumerate(transceivers):
        print(f"Transceiver {i}: {t.kind}, {t.direction}, mid: {t.mid}, stopped: {t.stopped}")
    
    # Change direction after creation
    audio_transceiver.direction = "recvonly"
    print(f"Changed audio direction to: {audio_transceiver.direction}")
    
    # Stop a transceiver
    video_transceiver.stop()
    print(f"Video transceiver stopped: {video_transceiver.stopped}")

async def monitor_rtp_stats():
    pc = aiortc.RTCPeerConnection()
    
    # Add tracks
    audio_track = aiortc.AudioStreamTrack()
    video_track = aiortc.VideoStreamTrack()
    
    audio_sender = pc.addTrack(audio_track)
    video_sender = pc.addTrack(video_track)
    
    async def print_stats():
        while True:
            # Get overall connection stats
            all_stats = await pc.getStats()
            
            # Get specific sender stats
            audio_stats = await audio_sender.getStats()
            video_stats = await video_sender.getStats()
            
            print("=== RTP Statistics ===")
            
            # Process audio stats
            for stats in audio_stats.values():
                if hasattr(stats, 'bytesSent'):
                    print(f"Audio - Bytes sent: {stats.bytesSent}, Packets sent: {stats.packetsSent}")
            
            # Process video stats
            for stats in video_stats.values():
                if hasattr(stats, 'bytesSent'):
                    print(f"Video - Bytes sent: {stats.bytesSent}, Packets sent: {stats.packetsSent}")
            
            await asyncio.sleep(5)  # Update every 5 seconds
    
    # Start monitoring
    asyncio.create_task(print_stats())

async def advanced_codec_config():
    pc = aiortc.RTCPeerConnection()
    
    # Get available codecs
    audio_caps = aiortc.RTCRtpSender.getCapabilities("audio")
    video_caps = aiortc.RTCRtpSender.getCapabilities("video")
    
    print("Available audio codecs:")
    for codec in audio_caps.codecs:
        print(f"  {codec.mimeType}, clock rate: {codec.clockRate}, channels: {codec.channels}")
    
    print("Available video codecs:")
    for codec in video_caps.codecs:
        print(f"  {codec.mimeType}, clock rate: {codec.clockRate}")
    
    # Create transceiver with specific codec preferences
    video_transceiver = pc.addTransceiver("video")
    
    # Prefer VP8 over other codecs
    vp8_codecs = [codec for codec in video_caps.codecs if "VP8" in codec.mimeType]
    if vp8_codecs:
        video_transceiver.setCodecPreferences(vp8_codecs)
        print("Set VP8 codec preference")
    
    # Create audio transceiver preferring Opus
    audio_transceiver = pc.addTransceiver("audio")
    opus_codecs = [codec for codec in audio_caps.codecs if "opus" in codec.mimeType]
    if opus_codecs:
        audio_transceiver.setCodecPreferences(opus_codecs)
        print("Set Opus codec preference")

async def track_replacement():
    pc = aiortc.RTCPeerConnection()
    
    # Start with default video track
    initial_track = aiortc.VideoStreamTrack()
    sender = pc.addTrack(initial_track)
    
    print(f"Initial track ID: {initial_track.id}")
    
    # Create custom video track
    class ColoredVideoTrack(aiortc.VideoStreamTrack):
        def __init__(self, color="blue"):
            super().__init__()
            self.color = color
    
    # Replace with colored track
    colored_track = ColoredVideoTrack("red")
    await sender.replaceTrack(colored_track)
    
    print(f"Replaced with track ID: {colored_track.id}")
    print(f"Sender now has track: {sender.track.id}")
    
    # Replace with None to stop sending
    await sender.replaceTrack(None)
    print(f"Sender track after None replacement: {sender.track}")