5.2.10. ytpb.playback

Playback for live streams.

class ytpb.playback.RewindTreeNode(key: float, value: int, left: Self | None = None, right: Self | None = None)

Bases: object

key: float
value: int
left: Self | None = None
right: Self | None = None
class ytpb.playback.RewindTreeMap(root: RewindTreeNode | None = None)

Bases: object

A binary search tree implementation to store key-value pairs.

Keys represent timestamps of segments, while values are sequence numbers.

root: RewindTreeNode | None = None
insert(key: float, value: int) → None

Inserts a pair of timestamp and sequence number into the tree.

closest(target: float) → RewindTreeNode | None

Finds the node closest to the target timestamp.

class ytpb.playback.RewindMoment(date: datetime, sequence: int, cut_at: float, is_end: bool = False, falls_in_gap: bool = False)

Bases: object

Represents a moment that has been rewound.

date: datetime

An actual rewound date.

sequence: int

A sequence number of a segment with a target date.

cut_at: float

A difference (in seconds) between segment start ingestion date and target date.

is_end: bool = False

Whether a moment represents the end of an interval.

falls_in_gap: bool = False

Whether a moment falls in gap.

class ytpb.playback.RewindInterval(start: RewindMoment, end: RewindMoment)

Bases: object

Represents an interval that has been rewound.

start: RewindMoment
end: RewindMoment
property duration: timedelta

An interval duration.

property sequences: Iterable[int]

Segment sequence numbers that represent the interval.

class ytpb.playback.PlaybackSession(playback: Playback = None, **kwargs)

Bases: Session

A session to use with Playback.

This session can be used to provide a retry mechanism for failed requests. Here is a list of handled HTTP status codes for segment URLs:

  • 403: Refresh segment base URL, and repeat a request

  • 404: Retry a request with no change

max_retries: int = 3
initial_delay: float = 3.0
backoff_multiplier: float = 1.5
set_playback(playback)
class ytpb.playback.Playback(video_url: str, session: Session | None = None, fetcher: InfoFetcher | None = None, write_to_cache: bool = False, **kwargs: Unpack[_PlaybackOptions])

Bases: object

The playback for live streams.

__init__(video_url: str, session: Session | None = None, fetcher: InfoFetcher | None = None, write_to_cache: bool = False, **kwargs: Unpack[_PlaybackOptions]) → None

Constructs a playback.

To work with a playback, fetch the essential information afterwards:

playback = Playback(video_url)
playback.fetch_and_set_essential()
Parameters:

  • video_url – A video URL.

  • session – An instance of requests.Session.

  • fetcher – A fetcher used to gather the video information and streams.

  • write_to_cache – Whether to write to cache.

Keyword Arguments:

user_agent – The HTTP User-Agent string used for requests. Note that this value has priority over the value from session headers.

classmethod from_url(video_url: str, **kwargs) → Playback

Creates a playback for the given video URL.

This also fetches the video information and streams.

Parameters:

  • video_url – A video URL.

  • **kwargs – Optional arguments that Playback takes.

Returns:

An instance of Playback.

classmethod from_manifest(manifest_path: Path, fetch_video_info: bool = True, **kwargs) → Playback
classmethod from_cache(video_url: str, **kwargs) → Playback

Creates a playback for the given URL from cache.

This also implies writing to cache. Note that an unexpired cache item for a stream should exist.

Example

To write a new cache item, create a playback with write_to_cache=True:

from ytpb.errors import CachedItemNotFoundError
try:
    playback = Playback.from_cache(video_url)
except CachedItemNotFoundError:
    playback = Playback.from_url(video_url, write_to_cache=True)
Parameters:

video_url – A video URL.

Raises:

CachedItemNotFoundError – If a cache item is not found or expired.

static get_cache_directory()

Gets the cache directory.

get_temp_directory() → Path

Gets the run temporary directory.

property video_url: str
property video_id: str
property info: YouTubeVideoInfo | LeftNotFetched
property streams: Streams[AudioRepresentationInfo | VideoRepresentationInfo]
property locations: dict[str, Path]
fetch_and_set_essential() → None

Fetches and sets essential information.

Such information includes information about a video and streams.

download_segment(sequence: int, stream: AudioRepresentationInfo | VideoRepresentationInfo, output_directory: Path | None = None, output_filename: str | Callable[[int, str], str] | None = compose_default_segment_filename, force_download: bool = False) → Path

Downloads a segment.

Examples

Download a segment to a file and read it:

from ytpb.segment import Segment
downloaded_path = playback.download_segment(
    0, playback.streams.get_by_itag("140")
)
segment = Segment.from_file(downloaded_path)
Parameters:

  • sequence – A segment sequence number.

  • stream – A stream to which segment belongs.

  • output_directory – Where to download a segment.

  • output_filename – A segment output filename.

  • force_download – Whether to force download a segment even if it exists.

Returns:

A path to the downloaded segment.

get_segment(sequence: int, stream: AudioRepresentationInfo | VideoRepresentationInfo, segment_directory: Path | None = None, segment_filename: str | Callable[[int, str], str] | None = compose_default_segment_filename, download: bool = True) → Segment

Gets a Segment representing a downloaded segment.

By default, if a segment file cannot be found, it will be downloaded.

Notes

See also ytpb.segment.Segment.from_file().

Parameters:

  • sequence – A segment sequence number.

  • stream – A stream to which segment belongs.

  • segment_directory – Where a segment is stored.

  • segment_filename – A segment filename.

  • download – Whether to download a segment if it doesn’t exist.

Returns:

A Segment object.

Raises:

FileNotFoundError – If couldn’t find a downloaded segment, when download is not requested.

locate_moment(point: datetime | int, stream: AudioRepresentationInfo | VideoRepresentationInfo | None = None, is_end: bool = False) → RewindMoment

Locates a moment by a point in stream.

Parameters:

  • point – An absolute point.

  • stream – A stream to which segments used in locating belong. If not set, the first available stream will be used.

  • is_end – Whether a moment represents the end of an interval.

Notes

See also ytpb.locate.SegmentLocator.

Returns:

A located moment of RewindMoment.

locate_interval(start_point: datetime | int | timedelta | RelativeSegmentSequence, end_point: datetime | int | timedelta | RelativeSegmentSequence, stream: AudioRepresentationInfo | VideoRepresentationInfo | None = None) → RewindInterval

Locates an interval by start and end points in stream.

Parameters:

  • start_point – A start absolute or relative point.

  • end_point – An end absolute or relative point.

  • stream – A stream to which segments used in locating belong. If not set, the first available stream will be used.

Notes

See also ytpb.locate.SegmentLocator.

Returns:

A located interval of RewindInterval.