Container Operations

def open(file, mode='r', format=None, options=None, container_options=None, 
         stream_options=None, metadata_encoding='utf-8', metadata_errors='strict',
         buffer_size=32768, timeout=None, io_open=None, hwaccel=None):
    """
    Open a media container.
    
    Parameters:
    - file: str | file-like | bytes - File path, file object, or bytes
    - mode: str - 'r' for input, 'w' for output  
    - format: str - Container format name (auto-detected if None)
    - options: dict - General options
    - container_options: dict - Container-specific options
    - stream_options: list[dict] - Per-stream options
    - metadata_encoding: str - Metadata encoding
    - metadata_errors: str - Error handling for metadata
    - buffer_size: int - I/O buffer size in bytes
    - timeout: float - Operation timeout in seconds
    - io_open: callable - Custom I/O opener
    - hwaccel: str - Hardware acceleration device
    
    Returns:
    InputContainer | OutputContainer
    """

class InputContainer:
    """Container for reading media files."""
    
    # Properties
    streams: StreamContainer
    metadata: dict[str, str]
    format: ContainerFormat
    duration: int | None  # Duration in AV_TIME_BASE units
    start_time: int       # Start time in AV_TIME_BASE units  
    bit_rate: int         # Overall bitrate
    size: int            # File size in bytes
    
    def close(self) -> None:
        """Close the container and release resources."""
    
    def demux(self, *streams, video=None, audio=None, subtitles=None, data=None) -> Iterator[Packet]:
        """
        Demux packets from streams.
        
        Parameters:
        - *streams: Specific streams to demux
        - video: int | VideoStream - Video stream index or object
        - audio: int | AudioStream - Audio stream index or object  
        - subtitles: int | SubtitleStream - Subtitle stream index or object
        - data: int | DataStream - Data stream index or object
        
        Yields:
        Packet objects from the specified streams
        """
    
    def decode(self, *args, **kwargs) -> Iterator[AudioFrame | VideoFrame | SubtitleSet]:
        """
        Decode frames from streams.
        
        Parameters: Same as demux()
        
        Yields:
        Decoded frames based on stream type
        """
    
    def seek(self, offset, *, backward=True, any_frame=False, stream=None) -> None:
        """
        Seek to a specific position.
        
        Parameters:
        - offset: int - Target position in stream time_base units
        - backward: bool - Allow seeking backward
        - any_frame: bool - Seek to any frame (not just keyframes)
        - stream: Stream - Reference stream for offset units
        """
    
    def flush_buffers(self) -> None:
        """Flush internal decode buffers."""

class OutputContainer:
    """Container for writing media files."""
    
    # Properties  
    streams: StreamContainer
    metadata: dict[str, str]
    format: ContainerFormat
    default_video_codec: str
    default_audio_codec: str
    default_subtitle_codec: str
    supported_codecs: set[str]
    
    def add_stream(self, codec=None, rate=None, **kwargs) -> AudioStream | VideoStream:
        """
        Add a stream to the container.
        
        Parameters:
        - codec: str | Codec - Codec name or object
        - rate: int | Fraction - Frame rate for video, sample rate for audio
        - **kwargs: Additional codec parameters
        
        Returns:
        Stream object for the new stream
        """
    
    def add_stream_from_template(self, stream, **kwargs) -> Stream:
        """
        Add stream using another stream as template.
        
        Parameters:
        - stream: Stream - Template stream
        - **kwargs: Override parameters
        
        Returns:
        New stream object
        """
    
    def add_data_stream(self, codec=None) -> DataStream:
        """Add a data stream."""
    
    def start_encoding(self) -> None:
        """Begin encoding process - must be called before muxing packets."""
    
    def mux(self, packet) -> None:
        """
        Multiplex a packet into the container.
        
        Parameters:
        - packet: Packet - Encoded packet to write
        """
    
    def mux_one(self, packet) -> None:
        """Multiplex exactly one packet."""
    
    def close(self) -> None:
        """Close container and finalize file."""

class Flags(Flag):
    """Container behavior flags."""
    gen_pts = 1      # Generate missing PTS
    ign_idx = 2      # Ignore index
    non_block = 4    # Non-blocking mode
    ign_dts = 8      # Ignore DTS
    no_fillin = 16   # Do not fill in missing values
    no_parse = 32    # Do not use packet parsing
    no_buffer = 64   # Do not buffer
    custom_io = 128  # Custom I/O
    discard_corrupt = 256   # Discard corrupt packets
    flush_packets = 512     # Flush output packets
    bitexact = 1024        # Bitexact mode
    sort_dts = 2048        # Sort DTS
    fast_seek = 4096       # Fast seeking
    shortest = 8192        # Stop at shortest stream
    auto_bsf = 16384       # Automatic bitstream filtering

class StreamContainer:
    """Container managing streams in a media file."""
    
    # Stream collections by type
    video: tuple[VideoStream, ...]
    audio: tuple[AudioStream, ...]
    subtitles: tuple[SubtitleStream, ...]
    attachments: tuple[AttachmentStream, ...]
    data: tuple[DataStream, ...]
    other: tuple[Stream, ...]
    
    def __len__(self) -> int:
        """Total number of streams."""
    
    def __iter__(self) -> Iterator[Stream]:
        """Iterate over all streams."""
    
    def __getitem__(self, index: int) -> Stream:
        """Get stream by index."""
    
    def get(self, *, video=None, audio=None, subtitles=None, data=None) -> list[Stream]:
        """
        Get streams by type and criteria.
        
        Parameters:
        - video: int | tuple - Video stream selection
        - audio: int | tuple - Audio stream selection  
        - subtitles: int | tuple - Subtitle stream selection
        - data: int | tuple - Data stream selection
        
        Returns:
        List of matching streams
        """
    
    def best(self, kind) -> Stream | None:
        """
        Get the best stream of a given type.
        
        Parameters:
        - kind: str - Stream type ('video', 'audio', 'subtitle')
        
        Returns:
        Best stream of the specified type or None
        """

class ContainerFormat:
    """Media container format information."""
    
    name: str           # Format name
    long_name: str      # Descriptive name
    extensions: set[str] # File extensions
    is_input: bool      # Can read this format
    is_output: bool     # Can write this format
    flags: int          # Format flags
    no_file: bool       # Format doesn't need files

import av

# Open container
container = av.open('video.mp4')

# Get basic information
print(f"Duration: {container.duration / av.time_base}")
print(f"Bitrate: {container.bit_rate}")
print(f"Video streams: {len(container.streams.video)}")
print(f"Audio streams: {len(container.streams.audio)}")

# Access metadata
for key, value in container.metadata.items():
    print(f"{key}: {value}")

# Get best streams
video_stream = container.streams.video[0]
audio_stream = container.streams.audio[0]

# Decode frames
for frame in container.decode(video_stream):
    print(f"Video frame: {frame.width}x{frame.height} at {frame.time}")
    
for frame in container.decode(audio_stream):
    print(f"Audio frame: {frame.samples} samples at {frame.time}")

container.close()

import av
import numpy as np

# Create output container
output = av.open('output.mp4', 'w')

# Add video stream
video_stream = output.add_stream('h264', rate=30)
video_stream.width = 1280
video_stream.height = 720
video_stream.pix_fmt = 'yuv420p'

# Add audio stream  
audio_stream = output.add_stream('aac', rate=44100)
audio_stream.channels = 2
audio_stream.layout = 'stereo'

# Start encoding
output.start_encoding()

# Generate and encode frames
for i in range(150):  # 5 seconds
    # Create video frame
    frame = av.VideoFrame.from_ndarray(
        np.random.randint(0, 255, (720, 1280, 3), dtype=np.uint8),
        format='rgb24'
    )
    frame.pts = i
    frame.time_base = video_stream.time_base
    
    # Encode and mux
    for packet in video_stream.encode(frame):
        output.mux(packet)

# Flush encoders
for packet in video_stream.encode():
    output.mux(packet)
for packet in audio_stream.encode():
    output.mux(packet)

output.close()

import av

# Open network stream
container = av.open('rtmp://example.com/stream', options={
    'timeout': '5000000',  # 5 second timeout
    'user_agent': 'PyAV Client'
})

# Read packets
for packet in container.demux():
    if packet.stream.type == 'video':
        for frame in packet.decode():
            print(f"Video frame: {frame.width}x{frame.height}")
    elif packet.stream.type == 'audio':
        for frame in packet.decode():
            print(f"Audio frame: {frame.samples} samples")

container.close()