Reference for youtube_dl_scraper/core/stream_array.py

youtube_dl_scraper.core.stream_array.StreamArray

StreamArray(**kwargs: Union[List[Stream], Dict[str, Union[Callable, List[Stream]]]])

A utility class to manage and manipulate a collection of video and audio streams.

Parameters:

Name	Type	Description	Default
streams	List[Stream]	A list of initial streams to add.	required
video_api	Callable	A callable for video stream APIs (reserved for future use).	required
audio_api	Callable	A callable for audio stream APIs (reserved for future use).	required

Source code in youtube_dl_scraper/core/stream_array.py

def __init__(
    self, **kwargs: Union[List[Stream], Dict[str, Union[Callable, List[Stream]]]]
):
    """
    Initialize the StreamArray object.

    Args:
        streams (List[Stream], optional): A list of initial streams to add.
        video_api (Callable, optional): A callable for video stream APIs (reserved for future use).
        audio_api (Callable, optional): A callable for audio stream APIs (reserved for future use).
    """
    self._streams = []
    self._resolutions = []

    if kwargs.get("streams"):
        self._streams.extend(kwargs["streams"])

    # TODO: use the api_functions to get custom streams
    self._vid_stream_api = kwargs.get("video_api")
    self._aud_stream_api = kwargs.get("audio_api")

available_qualities property

available_qualities: dict[str, tuple[int, ...]]

Get available video and audio qualities.

Returns:

Type	Description
dict[str, tuple[int, ...]]	dict[str, tuple[int, ...]]: A dictionary containing video resolutions and audio bitrates.

bitrates property

bitrates: tuple[int, ...]

Get all unique audio bitrates in the stream collection.

Returns:

Type	Description
tuple[int, ...]	tuple[int, ...]: Tuple of audio bitrates.

frame_rates property

frame_rates: tuple[int, ...]

Get all unique frame rates from video streams.

Returns:

Type	Description
tuple[int, ...]	tuple[int, ...]: Tuple of frame rates.

resolutions property

resolutions: tuple[int, ...]

Get all unique video resolutions in the stream collection.

Returns:

Type	Description
tuple[int, ...]	tuple[int, ...]: Tuple of video resolutions.

streams property

streams: tuple[Stream, ...]

Return all streams, sorted by resolution for videos and bitrate for audio.

Returns:

Type	Description
tuple[Stream, ...]	tuple[Stream, ...]: Sorted tuple of streams.

__getitem__

__getitem__(index: Union[int, slice]) -> Stream

Access a stream by its index or sliced stream array by its slice.

Parameters:

Name	Type	Description	Default
index	Union[int, slice]	index (int): The index of the stream. slice (slice): The slice of the streams.	required

Returns:

Type	Description
Stream	outputs (Union[Stream, StreamArray]) Stream: The stream at the given index. StreamArray: The stream array from the slice.

Source code in youtube_dl_scraper/core/stream_array.py

def __getitem__(self, index: Union[int, slice]) -> Stream:
    """
    Access a stream by its index or sliced stream array by its slice.

    Args:
        index (Union[int, slice]):
            index (int): The index of the stream.
            slice (slice): The slice of the streams.

    Returns:
        outputs (Union[Stream, StreamArray])
            Stream: The stream at the given index.
            StreamArray: The stream array from the slice.
    """
    try:
        if isinstance(index, slice):
            return StreamArray(streams=list(self.streams[index]))
        elif isinstance(index, int):
            return self.streams[index]
        else:
            raise TypeError("Invalid argument type.")
    except IndexError:
        raise IndexError(f"No stream found at index: {index}")

__iter__

__iter__() -> Self

Initialize iteration over the streams.

Returns:

Name	Type	Description
Self	Self	The StreamArray object.

Source code in youtube_dl_scraper/core/stream_array.py

def __iter__(self) -> Self:
    """
    Initialize iteration over the streams.

    Returns:
        Self: The StreamArray object.
    """
    self.i = 0
    return self

__len__

__len__() -> int

Get the number of streams.

Returns:

Name	Type	Description
int	int	The number of streams.

Source code in youtube_dl_scraper/core/stream_array.py

def __len__(self) -> int:
    """
    Get the number of streams.

    Returns:
        int: The number of streams.
    """
    return len(self.streams)

__next__

__next__() -> Stream

Get the next stream during iteration.

Returns:

Name	Type	Description
Stream	Stream	The next stream.

Raises:

Type	Description
StopIteration	If no more streams are available.

Source code in youtube_dl_scraper/core/stream_array.py

def __next__(self) -> Stream:
    """
    Get the next stream during iteration.

    Returns:
        Stream: The next stream.

    Raises:
        StopIteration: If no more streams are available.
    """
    if self.i < len(self.streams):
        item = self.streams[self.i]
        self.i += 1
        return item
    else:
        raise StopIteration

__str__

__str__() -> str

Get a string representation of the streams.

Returns:

Name	Type	Description
str	str	String representation of the streams.

Source code in youtube_dl_scraper/core/stream_array.py

def __str__(self) -> str:
    """
    Get a string representation of the streams.

    Returns:
        str: String representation of the streams.
    """
    return str(self.streams)

add_stream

add_stream(*streams: Stream) -> None

Add one or more streams to the collection.

Parameters:

Name	Type	Description	Default
*streams	Stream	Streams to add.	()

Source code in youtube_dl_scraper/core/stream_array.py

def add_stream(self, *streams: Stream) -> None:
    """
    Add one or more streams to the collection.

    Args:
        *streams (Stream): Streams to add.
    """
    self._streams.extend(streams)

filter

filter(**kwargs: Union[str, int, bool]) -> Self

Filter streams based on attributes.

Parameters:

Name	Type	Description	Default
**kwargs	Union[str, int, bool]	Attribute-value pairs to filter by.	{}

Returns:

Name	Type	Description
Self	Self	A StreamArray with the filtered streams.

Source code in youtube_dl_scraper/core/stream_array.py

def filter(self, **kwargs: Union[str, int, bool]) -> Self:
    """
    Filter streams based on attributes.

    Args:
        **kwargs (Union[str, int, bool]): Attribute-value pairs to filter by.

    Returns:
        Self: A StreamArray with the filtered streams.
    """
    reverse = kwargs.pop("reverse", None)
    if reverse is not None and not isinstance(reverse, bool):
        raise ValueError("Reverse keyword argument must be of type bool")

    filtered = [
        stream
        for stream in self.streams
        if all(getattr(stream, key, None) == value for key, value in kwargs.items())
    ]
    return StreamArray(streams=filtered)

first

first() -> Optional[Stream]

Get the first stream in the collection.

Returns:

Type	Description
Optional[Stream]	Optional[Stream]: The first stream, or None if empty.

Source code in youtube_dl_scraper/core/stream_array.py

def first(self) -> Optional[Stream]:
    """
    Get the first stream in the collection.

    Returns:
        Optional[Stream]: The first stream, or None if empty.
    """
    if self.streams:
        return self.streams[0]

get_audio_streams

get_audio_streams() -> Self

Get a new StreamArray containing only audio streams.

Returns:

Name	Type	Description
Self	Self	A StreamArray with audio streams.

Source code in youtube_dl_scraper/core/stream_array.py

def get_audio_streams(self) -> Self:
    """
    Get a new StreamArray containing only audio streams.

    Returns:
        Self: A StreamArray with audio streams.
    """
    audio_streams = self._get_audio()
    return StreamArray(streams=audio_streams)

get_highest_bitrate

get_highest_bitrate() -> Optional[AudioStream]

Get the audio stream with the highest bitrate.

Returns:

Type	Description
Optional[AudioStream]	Optional[AudioStream]: The highest bitrate audio stream.

Source code in youtube_dl_scraper/core/stream_array.py

def get_highest_bitrate(self) -> Optional[AudioStream]:
    """
    Get the audio stream with the highest bitrate.

    Returns:
        Optional[AudioStream]: The highest bitrate audio stream.
    """
    ordered_by_abr = self._get_audio()
    if ordered_by_abr:
        return ordered_by_abr[0]

get_highest_resolution

get_highest_resolution() -> Optional[VideoStream]

Get the video stream with the highest resolution.

Returns:

Type	Description
Optional[VideoStream]	Optional[VideoStream]: The highest resolution video stream.

Source code in youtube_dl_scraper/core/stream_array.py

def get_highest_resolution(self) -> Optional[VideoStream]:
    """
    Get the video stream with the highest resolution.

    Returns:
        Optional[VideoStream]: The highest resolution video stream.
    """
    video_streams = self._get_video()
    if video_streams:
        return video_streams[0]

get_video_streams

get_video_streams() -> Self

Get a new StreamArray containing only video streams.

Returns:

Name	Type	Description
Self	Self	A StreamArray with video streams.

Source code in youtube_dl_scraper/core/stream_array.py

def get_video_streams(self) -> Self:
    """
    Get a new StreamArray containing only video streams.

    Returns:
        Self: A StreamArray with video streams.
    """
    video_streams = self._get_video()
    return StreamArray(streams=video_streams)

last

last() -> Optional[Stream]

Get the last stream in the collection.

Returns:

Type	Description
Optional[Stream]	Optional[Stream]: The last stream, or None if empty.

Source code in youtube_dl_scraper/core/stream_array.py

def last(self) -> Optional[Stream]:
    """
    Get the last stream in the collection.

    Returns:
        Optional[Stream]: The last stream, or None if empty.
    """
    if self.streams:
        return self.streams[-1]

order_by

order_by(key: str, reverse: bool = False) -> Self

Order streams by a specific attribute.

Parameters:

Name	Type	Description	Default
key	str	The attribute to order by.	required
reverse	bool	Whether to reverse the order.	False

Returns:

Name	Type	Description
Self	Self	A StreamArray with ordered streams.

Source code in youtube_dl_scraper/core/stream_array.py

def order_by(self, key: str, reverse: bool = False) -> Self:
    """
    Order streams by a specific attribute.

    Args:
        key (str): The attribute to order by.
        reverse (bool): Whether to reverse the order.

    Returns:
        Self: A StreamArray with ordered streams.
    """
    if not all(hasattr(stream, key) for stream in self.streams):
        raise ValueError(f"Key '{key}' not found in all streams.")
    return StreamArray(
        streams=sorted(
            self.streams,
            key=lambda stream: getattr(stream, key),
            reverse=(not reverse),
        )
    )