Track Publications

from livekit import (
    TrackPublication,
    LocalTrackPublication,
    RemoteTrackPublication,
    TrackPublishOptions,
    VideoEncoding,
    TrackKind,
    TrackSource,
    EncryptionType,
)

class TrackPublication:
    """Base class for track publications.

    Provides metadata about a published track including encoding
    parameters, dimensions, and subscription state.
    
    Cannot be instantiated directly - use Local or Remote variants.
    """

    def __init__(
        self,
        owned_info: proto_track.OwnedTrackPublication
    ) -> None:
        """Initialize a TrackPublication.

        Args:
            owned_info: Internal publication information from FFI
            
        Note:
            Publications are created by SDK, not application.
        """

@property
def track(self) -> Optional[Track]:
    """The associated track (if any).

    Returns:
        Track: Track instance (LocalTrack or RemoteTrack)
        None: If track not available
        
    Note:
        For LocalTrackPublication: Always returns LocalTrack
        For RemoteTrackPublication: Returns RemoteTrack if subscribed, None otherwise
        
        Track may be None if:
        - Track not yet subscribed (remote)
        - Track unpublished
    """

@property
def sid(self) -> str:
    """Session ID of the publication.

    Returns:
        str: Unique publication session identifier
             Format: "TR_" followed by random string
             
    Note:
        Same as track.sid when track is available.
        Publication SID persists even when track is None.
    """

@property
def name(self) -> str:
    """Name of the publication.

    Returns:
        str: Publication/track name
        
    Note:
        Same as track name.
        Set when track is created.
    """

@property
def kind(self) -> proto_track.TrackKind.ValueType:
    """Kind of track (audio or video).

    Returns:
        TrackKind.KIND_AUDIO or TrackKind.KIND_VIDEO
        
    Example:
        >>> if publication.kind == TrackKind.KIND_AUDIO:
        ...     print("Audio publication")
        ... elif publication.kind == TrackKind.KIND_VIDEO:
        ...     print("Video publication")
    """

@property
def source(self) -> proto_track.TrackSource.ValueType:
    """Source of the track.

    Returns:
        TrackSource enum value:
        - SOURCE_CAMERA: Camera video
        - SOURCE_MICROPHONE: Microphone audio
        - SOURCE_SCREENSHARE: Screen share video
        - SOURCE_SCREENSHARE_AUDIO: Screen audio
        - SOURCE_UNKNOWN: Unknown/unspecified
        
    Note:
        Used to identify track purpose.
        Useful for selective subscription.
    """

@property
def simulcasted(self) -> bool:
    """Whether simulcast is enabled.

    Returns:
        bool: True if simulcast enabled, False otherwise
        
    Note:
        Simulcast means multiple quality layers sent simultaneously.
        Only applicable to video tracks.
        Audio tracks always return False.
    """

@property
def width(self) -> int:
    """Width of video track (0 for audio tracks).

    Returns:
        int: Width in pixels (0 for audio)
        
    Note:
        Represents encoded video width.
        May differ from source width if scaled.
    """

@property
def height(self) -> int:
    """Height of video track (0 for audio tracks).

    Returns:
        int: Height in pixels (0 for audio)
        
    Note:
        Represents encoded video height.
        May differ from source height if scaled.
    """

@property
def mime_type(self) -> str:
    """MIME type of the track.

    Returns:
        str: MIME type string
        
    Examples:
        Audio:
        - "audio/opus"
        
        Video:
        - "video/vp8"
        - "video/h264"
        - "video/vp9"
        - "video/av1"
        
    Note:
        Indicates codec used for encoding.
        Format: "type/codec" or "type/codec;profile=..."
    """

@property
def muted(self) -> bool:
    """Whether the track is muted.

    Returns:
        bool: True if muted, False otherwise
        
    Note:
        Muted state is maintained across subscriptions.
        Muted tracks send silence (audio) or black frames (video).
    """

@property
def encryption_type(self) -> proto_e2ee.EncryptionType.ValueType:
    """Encryption type for the track.

    Returns:
        EncryptionType enum value:
        - NONE: No encryption
        - GCM: AES-GCM encryption
        - CUSTOM: Custom encryption
        
    Note:
        Indicates if track is end-to-end encrypted.
    """

@property
def audio_features(self) -> List[int]:
    """List of audio features enabled for the track.

    Returns:
        List[int]: List of integer values representing audio features
        
    Examples:
        - Stereo
        - DTX (discontinuous transmission)
        - RED (redundant encoding)
        - FEC (forward error correction)
        
    Note:
        Only applicable to audio tracks.
        Video tracks return empty list.
        Values are audio feature flags (opaque integers).
    """

class LocalTrackPublication(TrackPublication):
    """Represents a local track publication.

    Local publications are created when publishing tracks from
    the local participant.
    
    Provides wait_for_subscription() to confirm delivery.
    """

    def __init__(
        self,
        owned_info: proto_track.OwnedTrackPublication
    ) -> None:
        """Initialize a LocalTrackPublication.

        Args:
            owned_info: Internal publication information
            
        Note:
            Created automatically by SDK when publishing.
        """

@property
def track(self) -> Optional[LocalTrack]:
    """The associated local track (if any).

    Returns:
        LocalTrack: LocalAudioTrack or LocalVideoTrack
        None: If track unpublished
    """

async def wait_for_subscription(self) -> None:
    """Wait until track subscribed by at least one participant.

    This is useful for ensuring that published tracks are being received
    before proceeding with operations that depend on active subscriptions.
    
    Returns:
        None (awaitable)
    
    Raises:
        RuntimeError: If track unpublished while waiting
        asyncio.CancelledError: If wait cancelled

    Example:
        >>> # Publish track
        >>> publication = await local.publish_track(track)
        >>> 
        >>> # Wait for confirmation
        >>> await publication.wait_for_subscription()
        >>> print("At least one participant is receiving the track")
        >>> 
        >>> # Now safe to start sending important data
        >>> await send_critical_data()
        
    Note:
        Blocks until at least one remote participant subscribes.
        Does not wait for specific participant.
        
        Use cases:
        - Confirm track delivery before important operation
        - Wait for audience before starting presentation
        - Ensure recording started before critical content
        
        Timeout pattern:
        >>> try:
        ...     await asyncio.wait_for(
        ...         publication.wait_for_subscription(),
        ...         timeout=10.0
        ...     )
        ... except asyncio.TimeoutError:
        ...     print("No subscribers after 10s")
    """

class RemoteTrackPublication(TrackPublication):
    """Represents a remote track publication.

    Remote publications are received from remote participants
    and can be subscribed to.
    
    Provides subscription control via set_subscribed().
    """

    def __init__(
        self,
        owned_info: proto_track.OwnedTrackPublication
    ) -> None:
        """Initialize a RemoteTrackPublication.

        Args:
            owned_info: Internal publication information
            
        Note:
            Created automatically by SDK.
            Received via track_published event.
        """

@property
def track(self) -> Optional[RemoteTrack]:
    """The associated remote track (if any).

    Returns:
        RemoteTrack: RemoteAudioTrack or RemoteVideoTrack if subscribed
        None: If not subscribed or subscription pending
        
    Note:
        Track becomes available after subscription completes.
        Check subscribed property to determine subscription state.
    """

@property
def subscribed(self) -> bool:
    """Whether the track is currently subscribed.

    Returns:
        bool: True if subscribed, False otherwise
        
    Note:
        Subscription state:
        - True: Track subscribed, receiving media
        - False: Track not subscribed or unsubscribed
        
        When auto_subscribe=True, this becomes True automatically.
        When auto_subscribe=False, set via set_subscribed(True).
    """

def set_subscribed(self, subscribed: bool) -> None:
    """Set whether to subscribe to this track.

    Args:
        subscribed: Desired subscription state
                   Type: bool
                   True: Subscribe to track
                   False: Unsubscribe from track

    Returns:
        None (synchronous operation)
        
    Raises:
        RuntimeError: If room not connected

    Example:
        >>> # Subscribe to track
        >>> publication.set_subscribed(True)
        >>> # Subscription happens asynchronously
        >>> # track_subscribed event emitted when complete
        >>>
        >>> # Unsubscribe from track
        >>> publication.set_subscribed(False)
        >>> # track_unsubscribed event emitted when complete
        
    Note:
        Subscription is asynchronous.
        Track becomes available in track_subscribed event.
        
        Calling set_subscribed(True) again when already subscribed is no-op.
        Calling set_subscribed(False) when not subscribed is no-op.
        
        Use cases:
        - Manual subscription when auto_subscribe=False
        - Selective subscription (bandwidth saving)
        - Unsubscribe when track no longer needed
    """

class TrackPublishOptions:
    """Options for publishing a track (protobuf message).

    This is a protobuf message class with the following fields.
    All fields are optional with defaults.
    """

# Fields:
video_encoding: VideoEncoding
"""Video encoding options.

Contains:
- max_bitrate: Maximum bitrate in bps
- max_framerate: Maximum frame rate in fps
"""

audio_encoding: AudioEncoding  
"""Audio encoding options.

Contains:
- max_bitrate: Maximum bitrate in bps
"""

video_codec: VideoCodec
"""Video codec to use.

Options:
- VP8: Best compatibility
- H264: Hardware acceleration
- AV1: Best compression
- VP9: Good middle ground
- H265: Better than H264
"""

dtx: bool
"""Enable discontinuous transmission (audio only).

When True:
- Stops transmission during silence
- Saves bandwidth
- Opus codec handles automatically
"""

red: bool
"""Enable redundant encoding (audio only).

When True:
- Sends redundant audio data
- Improves packet loss recovery
- Higher bandwidth usage
"""

simulcast: bool
"""Enable simulcast for adaptive quality.

When True:
- Sends multiple quality layers (low, medium, high)
- Receivers select appropriate quality
- Higher bandwidth and CPU usage
- Recommended for multi-party calls
"""

source: TrackSource
"""Track source type.

Options:
- SOURCE_CAMERA: Camera video
- SOURCE_MICROPHONE: Microphone audio
- SOURCE_SCREENSHARE: Screen video
- SOURCE_SCREENSHARE_AUDIO: Screen audio
- SOURCE_UNKNOWN: Unspecified
"""

stream: str
"""Stream name.

Optional identifier for track stream.
Rarely used.
"""

preconnect_buffer: bool
"""Enable preconnect buffering.

When True:
- Buffers frames before connection established
- Reduces initial delay
"""

from livekit import TrackPublishOptions, VideoEncoding, VideoCodec, TrackSource

options = TrackPublishOptions()
options.video_codec = VideoCodec.VP8
options.simulcast = True
options.source = TrackSource.SOURCE_CAMERA

# Set video encoding
encoding = VideoEncoding()
encoding.max_bitrate = 2_000_000  # 2 Mbps
encoding.max_framerate = 30.0
options.video_encoding = encoding

await local.publish_track(track, options)

from livekit import (
    AudioSource,
    LocalAudioTrack,
    TrackPublishOptions,
    TrackSource,
)

# Create audio track
source = AudioSource(48000, 1)
track = LocalAudioTrack.create_audio_track("microphone", source)

# Configure options
options = TrackPublishOptions()
options.source = TrackSource.SOURCE_MICROPHONE
options.dtx = True  # Enable DTX for bandwidth efficiency during silence
options.red = True  # Enable redundant encoding for packet loss recovery

# Publish
publication = await local.publish_track(track, options)

print(f"Published: {publication.sid}")
print(f"Name: {publication.name}")
print(f"MIME type: {publication.mime_type}")  # "audio/opus"
print(f"Kind: {publication.kind}")  # KIND_AUDIO
print(f"Source: {publication.source}")  # SOURCE_MICROPHONE

from livekit import (
    VideoSource,
    LocalVideoTrack,
    TrackPublishOptions,
    VideoEncoding,
    VideoCodec,
    TrackSource,
)

# Create video track
source = VideoSource(1920, 1080)
track = LocalVideoTrack.create_video_track("camera", source)

# Configure options
options = TrackPublishOptions()
options.video_codec = VideoCodec.VP8
options.simulcast = True
options.source = TrackSource.SOURCE_CAMERA

# Set encoding parameters
encoding = VideoEncoding()
encoding.max_bitrate = 3_000_000  # 3 Mbps
encoding.max_framerate = 30.0
options.video_encoding = encoding

# Publish
publication = await local.publish_track(track, options)

print(f"Published: {publication.sid}")
print(f"Name: {publication.name}")
print(f"Dimensions: {publication.width}x{publication.height}")
print(f"Simulcast: {publication.simulcasted}")
print(f"MIME type: {publication.mime_type}")  # "video/vp8"

# Publish track
publication = await local.publish_track(track, options)

# Wait until at least one participant subscribes
print("Waiting for subscription...")
await publication.wait_for_subscription()

print("Track is now being received by at least one participant")

# Now safe to proceed with operations requiring audience

from livekit import RemoteParticipant, RemoteTrackPublication

participant: RemoteParticipant = ...

# Subscribe to all tracks
for track_sid, publication in participant.track_publications.items():
    if isinstance(publication, RemoteTrackPublication):
        print(f"Subscribing to: {publication.name}")
        publication.set_subscribed(True)

# Check subscription status
for track_sid, publication in participant.track_publications.items():
    if isinstance(publication, RemoteTrackPublication):
        status = "subscribed" if publication.subscribed else "not subscribed"
        print(f"{publication.name}: {status}")

from livekit import RemoteParticipant, TrackKind, TrackSource

participant: RemoteParticipant = ...

# Subscribe only to camera video (not screen share)
for publication in participant.track_publications.values():
    if isinstance(publication, RemoteTrackPublication):
        should_subscribe = (
            publication.kind == TrackKind.KIND_VIDEO and
            publication.source == TrackSource.SOURCE_CAMERA
        )
        publication.set_subscribed(should_subscribe)

# Subscribe only to microphone audio
for publication in participant.track_publications.values():
    if isinstance(publication, RemoteTrackPublication):
        should_subscribe = (
            publication.kind == TrackKind.KIND_AUDIO and
            publication.source == TrackSource.SOURCE_MICROPHONE
        )
        publication.set_subscribed(should_subscribe)

# Unsubscribe from all tracks of a participant
for publication in participant.track_publications.values():
    if isinstance(publication, RemoteTrackPublication):
        publication.set_subscribed(False)

# Unsubscribe from specific track
specific_pub = participant.track_publications.get("TR_XXXXX")
if specific_pub and isinstance(specific_pub, RemoteTrackPublication):
    specific_pub.set_subscribed(False)

from livekit import TrackPublication, TrackKind

publication: TrackPublication = ...

# Basic info
print(f"SID: {publication.sid}")
print(f"Name: {publication.name}")
print(f"Kind: {publication.kind}")
print(f"Source: {publication.source}")
print(f"Muted: {publication.muted}")

# Video-specific (returns 0 for audio)
if publication.kind == TrackKind.KIND_VIDEO:
    print(f"Dimensions: {publication.width}x{publication.height}")
    print(f"Simulcast: {publication.simulcasted}")

# MIME type and codec
print(f"MIME type: {publication.mime_type}")
# Examples: "audio/opus", "video/vp8", "video/h264"

# Encryption
print(f"Encryption: {publication.encryption_type}")
if publication.encryption_type != EncryptionType.NONE:
    print("Track is end-to-end encrypted")

# Audio features
if publication.kind == TrackKind.KIND_AUDIO:
    print(f"Audio features: {publication.audio_features}")

import asyncio
from livekit import (
    Room,
    RoomOptions,
    LocalParticipant,
    RemoteParticipant,
    LocalTrackPublication,
    RemoteTrackPublication,
    Track,
    TrackKind,
    TrackSource,
    AudioSource,
    LocalAudioTrack,
    TrackPublishOptions,
)

async def main():
    room = Room()

    # Handle remote track publications
    @room.on("track_published")
    def on_track_published(
        publication: RemoteTrackPublication,
        participant: RemoteParticipant
    ):
        print(f"Track published by {participant.identity}:")
        print(f"  Name: {publication.name}")
        print(f"  SID: {publication.sid}")
        print(f"  Kind: {publication.kind}")
        print(f"  Source: {publication.source}")
        print(f"  MIME: {publication.mime_type}")

        if publication.kind == TrackKind.KIND_VIDEO:
            print(f"  Resolution: {publication.width}x{publication.height}")
            print(f"  Simulcast: {publication.simulcasted}")

        # Selective subscription
        # Auto-subscribe to camera and microphone only
        if publication.source in [
            TrackSource.SOURCE_CAMERA,
            TrackSource.SOURCE_MICROPHONE
        ]:
            print(f"  -> Subscribing")
            publication.set_subscribed(True)
        else:
            print(f"  -> Not subscribing")

    @room.on("track_subscribed")
    def on_track_subscribed(
        track: Track,
        publication: RemoteTrackPublication,
        participant: RemoteParticipant
    ):
        print(f"Subscribed to {publication.name} from {participant.identity}")
        print(f"  MIME type: {publication.mime_type}")
        print(f"  Muted: {publication.muted}")
        print(f"  Encryption: {publication.encryption_type}")

    @room.on("track_unsubscribed")
    def on_track_unsubscribed(
        track: Track,
        publication: RemoteTrackPublication,
        participant: RemoteParticipant
    ):
        print(f"Unsubscribed from {publication.name}")

    # Connect with manual subscription
    await room.connect(
        url,
        token,
        RoomOptions(auto_subscribe=False)  # Manual subscription control
    )

    local: LocalParticipant = room.local_participant

    # Publish audio track
    source = AudioSource(48000, 1)
    track = LocalAudioTrack.create_audio_track("mic", source)

    options = TrackPublishOptions()
    options.source = TrackSource.SOURCE_MICROPHONE
    options.dtx = True
    options.red = True

    publication: LocalTrackPublication = await local.publish_track(
        track,
        options
    )

    print(f"Published local track: {publication.sid}")
    print(f"  Name: {publication.name}")
    print(f"  Kind: {publication.kind}")
    print(f"  MIME: {publication.mime_type}")

    # Wait for at least one subscriber
    print("Waiting for subscription...")
    await publication.wait_for_subscription()
    print("Track is now being received!")

    # Monitor remote participants' tracks
    for identity, participant in room.remote_participants.items():
        print(f"\nParticipant: {identity}")
        for sid, pub in participant.track_publications.items():
            if isinstance(pub, RemoteTrackPublication):
                print(f"  {pub.name} ({sid})")
                print(f"    Kind: {pub.kind}")
                print(f"    Source: {pub.source}")
                print(f"    Subscribed: {pub.subscribed}")
                print(f"    Muted: {pub.muted}")
                if pub.kind == TrackKind.KIND_VIDEO:
                    print(f"    Resolution: {pub.width}x{pub.height}")

    # Keep running
    await asyncio.sleep(30)

    # Cleanup
    await local.unpublish_track(publication.sid)
    await source.aclose()
    await room.disconnect()

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

# Auto-subscribe to all tracks (simple, higher bandwidth)
options = RoomOptions(auto_subscribe=True)
await room.connect(url, token, options)

# Manual subscription for fine control (bandwidth efficient)
options = RoomOptions(auto_subscribe=False)
await room.connect(url, token, options)

@room.on("track_published")
def on_track_published(publication, participant):
    # Subscribe selectively
    if publication.source == TrackSource.SOURCE_CAMERA:
        publication.set_subscribed(True)

# Publish track
publication = await local.publish_track(track)

# Wait for audience
await publication.wait_for_subscription()

# Now safe to assume someone is receiving
# Good for: presentations, critical announcements, recordings

@room.on("track_muted")
def on_track_muted(participant, publication):
    print(f"{publication.name} muted by {participant.identity}")
    # Update UI: Show muted icon
    # Adjust processing: Skip muted tracks

@room.on("track_unmuted")
def on_track_unmuted(participant, publication):
    print(f"{publication.name} unmuted by {participant.identity}")
    # Update UI: Remove muted icon
    # Resume processing

# Different handling for different sources
@room.on("track_subscribed")
def on_track_subscribed(track, publication, participant):
    if publication.source == TrackSource.SOURCE_CAMERA:
        # Display in main video area
        display_in_main_area(track)
    elif publication.source == TrackSource.SOURCE_SCREENSHARE:
        # Display in screen share area (larger)
        display_in_screenshare_area(track)
    elif publication.source == TrackSource.SOURCE_MICROPHONE:
        # Process audio
        process_audio(track)

from livekit import EncryptionType

@room.on("track_published")
def on_track_published(publication, participant):
    if publication.encryption_type != EncryptionType.NONE:
        print(f"Track '{publication.name}' is encrypted: {publication.encryption_type}")
    else:
        print(f"WARNING: Track '{publication.name}' is not encrypted!")

@room.on("track_published")
def on_track_published(publication, participant):
    """Subscribe based on quality requirements."""
    if publication.kind == TrackKind.KIND_VIDEO:
        # Only subscribe to high-quality video
        if publication.width >= 1280 and publication.height >= 720:
            publication.set_subscribed(True)
        else:
            print(f"Skipping low-res video: {publication.width}x{publication.height}")
            publication.set_subscribed(False)

class SubscriptionManager:
    """Manage track subscriptions with policies."""
    
    def __init__(self, room: Room, max_video_subs: int = 4):
        self.room = room
        self.max_video_subs = max_video_subs
        self.video_subscriptions = []
        
        @room.on("track_published")
        def on_track_published(pub, participant):
            self.handle_publication(pub, participant)
    
    def handle_publication(self, pub: RemoteTrackPublication, participant):
        """Handle new publication with subscription policy."""
        if pub.kind == TrackKind.KIND_AUDIO:
            # Always subscribe to audio
            pub.set_subscribed(True)
        elif pub.kind == TrackKind.KIND_VIDEO:
            # Limited video subscriptions
            if len(self.video_subscriptions) < self.max_video_subs:
                pub.set_subscribed(True)
                self.video_subscriptions.append((participant.identity, pub.sid))
            else:
                print(f"Video subscription limit reached ({self.max_video_subs})")
                pub.set_subscribed(False)
    
    def unsubscribe_participant(self, identity: str):
        """Unsubscribe from participant's videos."""
        participant = self.room.remote_participants.get(identity)
        if not participant:
            return
        
        for pub in participant.track_publications.values():
            if isinstance(pub, RemoteTrackPublication):
                if pub.kind == TrackKind.KIND_VIDEO and pub.subscribed:
                    pub.set_subscribed(False)
                    # Remove from tracking
                    self.video_subscriptions = [
                        (pid, sid) for pid, sid in self.video_subscriptions
                        if not (pid == identity and sid == pub.sid)
                    ]