video_transcoding.transcoding package

Submodules

video_transcoding.transcoding.analysis module

class video_transcoding.transcoding.analysis.FFProbeHLSAnalyzer(info: ProbeInfo)

Bases: Analyzer

Analyzer for multi-variant HLS results.

analyze() → List[Meta]

get_bitrate(track: Dict[str, Any]) → int

get_duration(track: Dict[str, Any]) → TS

class video_transcoding.transcoding.analysis.MKVPlaylistAnalyzer(info: ProbeInfo)

Bases: Analyzer

Analyzer for HLS playlist with .mkv fragments.

get_duration(track: Dict[str, Any]) → TS

Augment track duration with a value from container.

This is legit if media contains only a single stream.

class video_transcoding.transcoding.analysis.MKVSegmentAnalyzer(info: ProbeInfo)

Bases: MKVPlaylistAnalyzer

Analyzer for audio/video segments in .NUT container.

get_bitrate(track: Dict[str, Any]) → int

class video_transcoding.transcoding.analysis.SourceAnalyzer(info: MediaInfo)

Bases: Analyzer

Universal source media analyzer.

class video_transcoding.transcoding.analysis.VideoResultAnalyzer(info: ProbeInfo)

Bases: Analyzer

Analyzer for multi-stream video segments in MPEGTS container.

video_transcoding.transcoding.codecs module

class video_transcoding.transcoding.codecs.AudioCodec(codec: str = None, bitrate: int = 0, rate: float = None, channels: int = None)

Bases: AudioCodec

channels: int = None

rate: float = None

class video_transcoding.transcoding.codecs.Copy(codec: str = None, bitrate: Optional[int] = None, kind: fffw.graph.meta.StreamType = <factory>)

Bases: Copy

bitrate: int | None = None

class video_transcoding.transcoding.codecs.VideoCodec(codec: str = None, bitrate: int = 0, force_key_frames: str = None, constant_rate_factor: int = None, preset: str = None, max_rate: int = None, buf_size: int = None, profile: str = None, gop: int = None, rate: float = None, pix_fmt: str = None)

Bases: VideoCodec

buf_size: int = None

constant_rate_factor: int = None

force_key_frames: str = None

gop: int = None

max_rate: int = None

pix_fmt: str = None

preset: str = None

profile: str = None

rate: float = None

video_transcoding.transcoding.extract module

class video_transcoding.transcoding.extract.Extractor(*args: Any, **kwargs: Any)

Bases: LoggerMixin, ABC

ffprobe(uri: str, timeout: float = 60.0, **kwargs: Any) → ProbeInfo

abstract get_meta_data(uri: str) → Metadata

mediainfo(uri: str) → MediaInfo

class video_transcoding.transcoding.extract.HLSExtractor(*args: Any, **kwargs: Any)

Bases: Extractor

Extracts metadata from HLS results.

get_meta_data(uri: str) → Metadata

class video_transcoding.transcoding.extract.MKVExtractor(*args: Any, **kwargs: Any)

Bases: Extractor, ABC

Supports analyzing media from playlists with .mkv segments.

ffprobe(uri: str, timeout: float = 60.0, **kwargs: Any) → ProbeInfo

class video_transcoding.transcoding.extract.SourceExtractor(*args: Any, **kwargs: Any)

Bases: Extractor

get_meta_data(uri: str) → Metadata

class video_transcoding.transcoding.extract.SplitExtractor(video_playlist: str, audio_file: str)

Bases: MKVExtractor

Extracts source metadata from video and audio HLS playlists.

get_meta_data(uri: str) → Metadata

class video_transcoding.transcoding.extract.VideoResultExtractor(*args: Any, **kwargs: Any)

Bases: MKVExtractor

Extracts metadata from video segment transcoding results.

get_meta_data(uri: str) → Metadata

class video_transcoding.transcoding.extract.VideoSegmentExtractor(*args: Any, **kwargs: Any)

Bases: MKVExtractor

Extracts metadata from video segments

get_meta_data(uri: str) → Metadata

video_transcoding.transcoding.ffprobe module

class video_transcoding.transcoding.ffprobe.FFProbe(input: str = None, loglevel: str = None, show_streams: bool = False, show_format: bool = False, output_format: str = None, allowed_extensions: str | None = None)

Bases: FFProbe

Extends ffprobe wrapper with new arguments and output filtering.

handle_stderr(line: str) → str

handle_stdout(line: str) → str

allowed_extensions: str | None = None

video_transcoding.transcoding.inputs module

class video_transcoding.transcoding.inputs.Input(streams: Tuple[fffw.encoding.inputs.Stream, ...] = <factory>, hardware: str = None, device: str = None, output_format: str = None, fast_seek: Union[fffw.graph.meta.TS, float, int] = None, duration: Union[fffw.graph.meta.TS, float, int] = None, input_file: str = None, slow_seek: Union[fffw.graph.meta.TS, float, int] = None, allowed_extensions: str = None)

Bases: Input

allowed_extensions: str = None

video_transcoding.transcoding.inputs.input_file(filename: str, *streams: Stream, **kwargs: Any) → Input

video_transcoding.transcoding.metadata module

class video_transcoding.transcoding.metadata.Metadata(uri: str, videos: List[fffw.graph.meta.VideoMeta], audios: List[fffw.graph.meta.AudioMeta])

Bases: object

classmethod from_native(data: dict) → Metadata

property audio: AudioMeta

audios: List[AudioMeta]

property streams: List[Stream]

uri: str

property video: VideoMeta

videos: List[VideoMeta]

video_transcoding.transcoding.metadata.audio_meta_from_native(data: dict) → AudioMeta

video_transcoding.transcoding.metadata.get_meta_kwargs(data: dict) → dict

video_transcoding.transcoding.metadata.scene_from_native(data: dict) → Scene

video_transcoding.transcoding.metadata.video_meta_from_native(data: dict) → VideoMeta

video_transcoding.transcoding.outputs module

class video_transcoding.transcoding.outputs.FileOutput(codecs: List[fffw.encoding.outputs.Codec] = <factory>, no_video: Optional[bool] = None, no_audio: Optional[bool] = None, format: str = None, output_file: str = None, copyts: bool = False, avoid_negative_ts: str = None, method: Optional[str] = None, muxdelay: Optional[str] = None)

Bases: Output

method: str | None = None

muxdelay: str | None = None

class video_transcoding.transcoding.outputs.HLSOutput(codecs: List[fffw.encoding.outputs.Codec] = <factory>, no_video: Optional[bool] = None, no_audio: Optional[bool] = None, format: str = None, output_file: str = None, copyts: bool = False, avoid_negative_ts: str = None, hls_time: Optional[float] = 2, hls_playlist_type: Optional[str] = None, var_stream_map: Optional[str] = None, hls_segment_filename: Optional[str] = None, master_pl_name: Optional[str] = None, muxdelay: Optional[str] = None, reset_timestamps: Optional[int] = 0)

Bases: Output

hls_playlist_type: str | None = None

hls_segment_filename: str | None = None

hls_time: float | None = 2

master_pl_name: str | None = None

muxdelay: str | None = None

reset_timestamps: int | None = 0

var_stream_map: str | None = None

class video_transcoding.transcoding.outputs.Output(codecs: List[fffw.encoding.outputs.Codec] = <factory>, no_video: Optional[bool] = None, no_audio: Optional[bool] = None, format: str = None, output_file: str = None, copyts: bool = False, avoid_negative_ts: str = None)

Bases: Output

avoid_negative_ts: str = None

copyts: bool = False

class video_transcoding.transcoding.outputs.SegmentOutput(codecs: ~typing.List[~fffw.encoding.outputs.Codec] = <factory>, no_video: bool | None = None, no_audio: bool | None = None, format: str = None, output_file: str = None, copyts: bool = False, avoid_negative_ts: str = None, segment_format: str | None = None, segment_list: str | None = None, segment_list_type: str | None = None, segment_time: float | None = None)

Bases: Output

Segment muxer

segment_format: str | None = None

segment_list: str | None = None

segment_list_type: str | None = None

segment_time: float | None = None

video_transcoding.transcoding.profiles module

class video_transcoding.transcoding.profiles.AudioCondition(min_sample_rate: int = 0, min_bitrate: int = 0)

Bases: object

Condition for source audio stream for video profile selection

is_valid(meta: AudioMeta) → bool

Parameters:

meta – source video stream metadata

Returns:

True if source stream satisfies condition.

min_bitrate: int = 0

min_sample_rate: int = 0

class video_transcoding.transcoding.profiles.AudioProfile(condition: AudioCondition, audio: List[str])

Bases: object

Audio transcoding profile.

audio: List[str]

condition: AudioCondition

class video_transcoding.transcoding.profiles.AudioTrack(id: str, codec: str, bitrate: int, channels: int, sample_rate: int)

Bases: object

Settings for a single audio stream in resulting media file

classmethod from_native(data: Dict[str, Any]) → AudioTrack

bitrate: int

channels: int

codec: str

id: str

sample_rate: int

class video_transcoding.transcoding.profiles.Container(segment_duration: float | None = None)

Bases: object

Output file format

classmethod from_native(data: Dict[str, Any]) → Container

segment_duration: float | None = None

class video_transcoding.transcoding.profiles.Preset(video_profiles: List[VideoProfile], audio_profiles: List[AudioProfile], video: List[VideoTrack], audio: List[AudioTrack])

Bases: object

A set of video and audio profiles to select from.

select_profile(video: VideoMeta, audio: AudioMeta) → Profile

audio: List[AudioTrack]

audio_profiles: List[AudioProfile]

video: List[VideoTrack]

video_profiles: List[VideoProfile]

class video_transcoding.transcoding.profiles.Profile(video: List[VideoTrack], audio: List[AudioTrack], container: Container)

Bases: object

Selected transcoding profile containing a number of audio and video streams.

classmethod from_native(data: Dict[str, Any]) → Profile

audio: List[AudioTrack]

container: Container

video: List[VideoTrack]

class video_transcoding.transcoding.profiles.VideoCondition(min_width: int = 0, min_height: int = 0, min_bitrate: int = 0, min_frame_rate: float = 0.0, min_dar: float = 0.0, max_dar: float = 0.0)

Bases: object

Condition for source video stream for video profile selection

is_valid(meta: VideoMeta) → bool

Parameters:

meta – source video stream metadata

Returns:

True if source stream satisfies condition.

max_dar: float = 0.0

min_bitrate: int = 0

min_dar: float = 0.0

min_frame_rate: float = 0.0

min_height: int = 0

min_width: int = 0

class video_transcoding.transcoding.profiles.VideoProfile(condition: VideoCondition, segment_duration: float, video: List[str])

Bases: object

Video transcoding profile.

condition: VideoCondition

segment_duration: float

video: List[str]

class video_transcoding.transcoding.profiles.VideoTrack(id: str, codec: str, constant_rate_factor: int, preset: str, max_rate: int, buf_size: int, profile: str, pix_fmt: str, width: int, height: int, frame_rate: float, gop_size: int, force_key_frames: str)

Bases: object

Settings for a single video stream in resulting media file

classmethod from_native(data: Dict[str, Any]) → VideoTrack

buf_size: int

codec: str

constant_rate_factor: int

force_key_frames: str

frame_rate: float

gop_size: int

height: int

id: str

max_rate: int

pix_fmt: str

preset: str

profile: str

width: int

video_transcoding.transcoding.transcoder module

class video_transcoding.transcoding.transcoder.Processor(src: str, dst: str, *, profile: Profile, meta: Metadata)

Bases: LoggerMixin, ABC

A single processing step abstract class.

abstract get_result_metadata(uri: str) → Metadata

Get result metadata.

Parameters:

uri – analyzed media

Returns:

metadata object with video and audio stream

abstract prepare_ffmpeg(src: Metadata) → FFMPEG

process() → Metadata

static run(ff: FFMPEG) → None

Starts ffmpeg process and captures errors from it’s logs

requires_audio: bool = True

requires_video: bool = True

class video_transcoding.transcoding.transcoder.Segmentor(*, video_source: str, audio_source: str, dst: str, profile: Profile, meta: Metadata)

Bases: Processor

Result segmentation logic.

get_output_kwargs(codecs_list: List[Codec]) → Dict[str, Any]

get_result_metadata(uri: str) → Metadata

Get result metadata.

Parameters:

uri – analyzed media

Returns:

metadata object with video and audio stream

static get_var_stream_map(codecs_list: List[Codec]) → str

prepare_audio_codecs() → List[AudioCodec]

prepare_ffmpeg(src: Metadata) → FFMPEG

prepare_output(codecs_list: List[Codec]) → Output

class video_transcoding.transcoding.transcoder.Splitter(src: str, dst: str, *, profile: Profile, meta: Metadata, source_video_playlist: str, source_video_chunk: str, source_audio: str)

Bases: Processor

Source splitting logic.

get_audio_output_kwargs(codecs_list: List[Codec]) → Dict[str, Any]

get_result_metadata(uri: str) → Metadata

Get result metadata.

Parameters:

uri – analyzed media

Returns:

metadata object with video and audio stream

get_video_output_kwargs(codecs_list: List[Codec]) → Dict[str, Any]

prepare_audio_output(codecs_list: List[Codec]) → Output

prepare_ffmpeg(src: Metadata) → FFMPEG

prepare_video_output(codecs_list: List[Codec]) → Output

class video_transcoding.transcoding.transcoder.Transcoder(src: str, dst: str, *, profile: Profile, meta: Metadata)

Bases: Processor

Source transcoding logic.

get_result_metadata(uri: str) → Metadata

Get result metadata.

Parameters:

uri – analyzed media

Returns:

metadata object with video and audio stream

prepare_ffmpeg(src: Metadata) → FFMPEG

Prepares ffmpeg command for a given source :param src: input file metadata :return: ffmpeg wrapper

static prepare_input(src: Metadata) → Input

prepare_output(video_codecs: List[VideoCodec]) → Output

prepare_video_codecs() → List[VideoCodec]

scale_and_encode(source: Input, video_codecs: List[VideoCodec], dst: Output) → SIMD

requires_audio: bool = False

video_transcoding.transcoding.workspace module

class video_transcoding.transcoding.workspace.Collection(*parts: str)

Bases: Resource

Directory in a workspace

collection(*parts: str) → Collection

file(*parts: str) → File

property trailing_slash: str

class video_transcoding.transcoding.workspace.File(*parts: str)

Bases: Resource

File in a workspace.

property trailing_slash: str

class video_transcoding.transcoding.workspace.FileSystemWorkspace(base: str)

Bases: Workspace

create_collection(c: Collection) → None

delete_collection(c: Collection) → None

exists(r: Resource) → bool

read(r: File) → str

write(r: File, content: str) → None

class video_transcoding.transcoding.workspace.Resource(*parts: str)

Bases: ABC

Directory or file in a workspace.

property basename: str

property parent: Collection | None

property path: str

abstract property trailing_slash: str

class video_transcoding.transcoding.workspace.WebDAVWorkspace(base: str)

Bases: Workspace

create_collection(c: Collection) → None

delete_collection(c: Collection) → None

exists(r: Resource) → bool

read(r: File) → str

write(r: File, content: str) → None

class video_transcoding.transcoding.workspace.Workspace(uri: ParseResult)

Bases: LoggerMixin, ABC

abstract create_collection(c: Collection) → None

abstract delete_collection(c: Collection) → None

ensure_collection(path: str) → Collection

Ensures that a directory with relative path exists.

Returns:

complete uri for a directory.

abstract exists(r: Resource) → bool

get_absolute_uri(r: Resource) → ParseResult

abstract read(f: File) → str

abstract write(f: File, content: str) → None

video_transcoding.transcoding.workspace.init(base: str) → Workspace

Module contents