5.2.2. Actions

Actions are high-level functions that work with a playback.

ytpb.actions.capture

Actions to capture frames as images.

ytpb.actions.capture.extract_frame_as_image(segment: Segment, target_date: datetime, last_as_fallback: bool = True) → Image

Extracts a frame as image from a segment.

Parameters:

  • segment – A Segment object.

  • target_date – A frame target date.

  • last_as_fallback – Whether to use a last frame as a fallback if the target date is out of a video.

Returns:

An extracted image as a PIL.Image.Image object.

ytpb.actions.capture.capture_frames(playback: Playback, target_dates: list[datetime], stream: AudioRepresentationInfo | VideoRepresentationInfo, reference_sequence: int | None = None) → Iterator[tuple[Image, Segment]]

Captures frames as images.

Examples

Here’s an example of capturing frames from one day every hour:

from datetime import datetime, timedelta, timezone

best_stream, = playback.streams.query(
    "type eq video and format eq mp4 and frame_rate eq 30 | best"
)
start_date = datetime(2024, 1, 2, 0, tzinfo=timezone.utc)
dates_to_capture = [start_date + timedelta(hours=h) for h in range(25)]

captured = capture_frames(
    playback, dates_to_capture, best_stream
)
for i, (image, _) in enumerate(captured):
    image.save(f"output-{i:02d}.jpg", quality=80)
Parameters:

  • playback – A Playback object.

  • target_dates – A list of dates to capture.

  • stream – A stream to which segments used for capturing belongs.

  • reference_sequence – A segment sequence number used as a start reference.

Returns:

An iterator with pairs of a captured frame and corresponding segment.

ytpb.actions.compose

Actions to compose and refresh MPEG-DASH MPDs.

ytpb.actions.compose.compose_static_mpd(playback: Playback, rewind_interval: RewindInterval, streams: Streams[AudioOrVideoStream]) → str

Composes a static MPEG-DASH MPD.

ytpb.actions.compose.compose_dynamic_mpd(playback: Playback, rewind_metadata: SegmentMetadata, streams: Streams[AudioOrVideoStream]) → str

Composes a dynamic MPEG-DASH MPD.

ytpb.actions.compose.refresh_mpd(manifest_content: str, streams: Streams[AudioOrVideoStream]) → str

Refreshes segment base URLs in a composed MPEG-DASH MPD.

ytpb.actions.download

Actions to download excerpts.

class ytpb.actions.download.ProgressReporter(*args, **kwargs)

Bases: Protocol

update(task: int)
class ytpb.actions.download.NullProgressReporter

Bases: object

update(task: int = 0) → None
class ytpb.actions.download.RichProgressReporter(progress: Progress | None = None)

Bases: object

update(task: int = 0) → None
class ytpb.actions.download.ExcerptDownloadResult(exception: Exception | None, merged_path: Path, audio_segment_paths: list[Path], video_segment_paths: list[Path])

Bases: NamedTuple

Represents an excerpt download result.

exception: Exception | None

An exception raised during downloading.

merged_path: Path

A path of the merged excerpt file.

audio_segment_paths: list[Path]

Paths of downloaded audio segments.

video_segment_paths: list[Path]

Paths of downloaded video segments.

ytpb.actions.download.chained_zip(*iterables) → Iterable[Any]

Makes an iterator that returns elements from iterables in round robin fashion.

Notes

The implementation assumes that the lengths of the iterables are the same. Otherwise, the iteration will stop after exhausting of the shortest iterable.

Examples

>>> list(chained_zip(["A", "B"], [0, 1]))
["A", 0, "B", 1]
ytpb.actions.download.download_segments(playback: Playback, sequence_numbers: Iterable[int], streams: Streams[AudioOrVideoStream] | list[AudioRepresentationInfo | VideoRepresentationInfo], output_directory: Path | None = None, output_filename: Callable[[int, str], str] = compose_default_segment_filename, progress_reporter: ProgressReporter | None = None) → list[Path]

Downloads segments.

Parameters:

  • playback – A playback.

  • sequence_numbers – Segment sequence numbers to rewind.

  • streams – Streams to download.

  • output_directory – A directory where to save downloaded segments.

  • output_filename – A callable to compose segment filenames.

  • progress_reporter – An instance of ProgressReporter-like class to show downloading progress. Defaults to dummy progress reporter.

Returns:

A list of downloaded segment paths.

ytpb.actions.download.download_excerpt(playback: Playback, rewind_interval: RewindInterval, output_stem: str | Path, segments_directory: Path, audio_stream: AudioRepresentationInfo | None = None, video_stream: VideoRepresentationInfo | None = None, need_cut: bool = True, merge_kwargs: dict[str, Any] | None = None, progress_reporter: ProgressReporter | None = None) → ExcerptDownloadResult

Downloads and merges audio and/or video segments.

Notes

Downloaded segments are not cleaned up after merging.

Parameters:

  • playback – A playback.

  • rewind_interval – A rewind interval.

  • output_stem – A full path stem of the merged excerpt file.

  • audio_stream – An audio stream.

  • video_stream – A video stream.

  • segments_directory – A directory where to store downloaded segments.

  • need_cut – Whether to cut boundary segments to exact times.

  • merge_kwargs – Arguments that ytpb.merge.merge_segments() takes.

  • progress_reporter – An instance of ProgressReporter-like class to show downloading progress. Defaults to dummy progress reporter.

Returns:

An ExcerptDownloadResult object.