<s> import os <EOL> import re <EOL> import subprocess <EOL> from collections import namedtuple <EOL> import logging <EOL> import bisect <EOL> from common import SushiError , get_extension <EOL> import chapters <EOL> MediaStreamInfo = namedtuple ( '<STR_LIT>' , [ '<STR_LIT:id>' , '<STR_LIT:info>' , '<STR_LIT:default>' , '<STR_LIT:title>' ] ) <EOL> SubtitlesStreamInfo = namedtuple ( '<STR_LIT>' , [ '<STR_LIT:id>' , '<STR_LIT:info>' , '<STR_LIT:type>' , '<STR_LIT:default>' , '<STR_LIT:title>' ] ) <EOL> MediaInfo = namedtuple ( '<STR_LIT>' , [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] ) <EOL> class FFmpeg ( object ) : <EOL> @ staticmethod <EOL> def get_info ( path ) : <EOL> try : <EOL> process = subprocess . Popen ( [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , path ] , stderr = subprocess . PIPE ) <EOL> out , err = process . communicate ( ) <EOL> process . wait ( ) <EOL> return err <EOL> except OSError as e : <EOL> if e . errno == <NUM_LIT:2> : <EOL> raise SushiError ( "<STR_LIT>" ) <EOL> raise <EOL> @ staticmethod <EOL> def demux_file ( input_path , ** kwargs ) : <EOL> args = [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , input_path , '<STR_LIT>' ] <EOL> audio_stream = kwargs . get ( '<STR_LIT>' , None ) <EOL> audio_path = kwargs . get ( '<STR_LIT>' , None ) <EOL> audio_rate = kwargs . get ( '<STR_LIT>' , None ) <EOL> if audio_stream is not None : <EOL> args . extend ( ( '<STR_LIT>' , '<STR_LIT>' . format ( audio_stream ) ) ) <EOL> if audio_rate : <EOL> args . extend ( ( '<STR_LIT>' , str ( audio_rate ) ) ) <EOL> args . extend ( ( '<STR_LIT>' , '<STR_LIT:1>' , '<STR_LIT>' , '<STR_LIT>' , audio_path ) ) <EOL> script_stream = kwargs . get ( '<STR_LIT>' , None ) <EOL> script_path = kwargs . get ( '<STR_LIT>' , None ) <EOL> if script_stream is not None : <EOL> args . extend ( ( '<STR_LIT>' , '<STR_LIT>' . format ( script_stream ) , script_path ) ) <EOL> video_stream = kwargs . get ( '<STR_LIT>' , None ) <EOL> timecodes_path = kwargs . get ( '<STR_LIT>' , None ) <EOL> if timecodes_path is not None : <EOL> args . extend ( ( '<STR_LIT>' , '<STR_LIT>' . format ( video_stream ) , '<STR_LIT>' , '<STR_LIT>' , timecodes_path ) ) <EOL> logging . debug ( '<STR_LIT>' . format ( '<STR_LIT:U+0020>' . join ( ( '<STR_LIT>' if '<STR_LIT:U+0020>' in a else '<STR_LIT>' ) . format ( a ) for a in args ) ) ) <EOL> try : <EOL> subprocess . call ( args ) <EOL> except OSError as e : <EOL> if e . errno == <NUM_LIT:2> : <EOL> raise SushiError ( "<STR_LIT>" ) <EOL> raise <EOL> @ staticmethod <EOL> def _get_audio_streams ( info ) : <EOL> streams = re . findall ( r'<STR_LIT>' <EOL> r'<STR_LIT>' <EOL> r'<STR_LIT>' , <EOL> info , flags = re . VERBOSE ) <EOL> return [ MediaStreamInfo ( int ( x [ <NUM_LIT:0> ] ) , x [ <NUM_LIT:1> ] , x [ <NUM_LIT:2> ] != '<STR_LIT>' , x [ <NUM_LIT:3> ] ) for x in streams ] <EOL> @ staticmethod <EOL> def _get_video_streams ( info ) : <EOL> streams = re . findall ( r'<STR_LIT>' <EOL> r'<STR_LIT>' <EOL> r'<STR_LIT>' , <EOL> info , flags = re . VERBOSE ) <EOL> return [ MediaStreamInfo ( int ( x [ <NUM_LIT:0> ] ) , x [ <NUM_LIT:1> ] , x [ <NUM_LIT:2> ] != '<STR_LIT>' , x [ <NUM_LIT:3> ] ) for x in streams ] <EOL> @ staticmethod <EOL> def _get_chapters_times ( info ) : <EOL> return map ( float , re . findall ( r'<STR_LIT>' , info ) ) <EOL> @ staticmethod <EOL> def _get_subtitles_streams ( info ) : <EOL> maps = { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' <EOL> } <EOL> streams = re . findall ( r'<STR_LIT>' <EOL> r'<STR_LIT>' <EOL> r'<STR_LIT>' , <EOL> info , flags = re . VERBOSE ) <EOL> return [ SubtitlesStreamInfo ( int ( x [ <NUM_LIT:0> ] ) , x [ <NUM_LIT:1> ] , maps . get ( x [ <NUM_LIT:2> ] , x [ <NUM_LIT:2> ] ) , x [ <NUM_LIT:3> ] != '<STR_LIT>' , x [ <NUM_LIT:4> ] . strip ( ) ) for x in streams ] <EOL> @ classmethod <EOL> def get_media_info ( cls , path ) : <EOL> info = cls . get_info ( path ) <EOL> video_streams = cls . _get_video_streams ( info ) <EOL> audio_streams = cls . _get_audio_streams ( info ) <EOL> subs_streams = cls . _get_subtitles_streams ( info ) <EOL> chapter_times = cls . _get_chapters_times ( info ) <EOL> return MediaInfo ( video_streams , audio_streams , subs_streams , chapter_times ) <EOL> class MkvToolnix ( object ) : <EOL> @ classmethod <EOL> def extract_timecodes ( cls , mkv_path , stream_idx , output_path ) : <EOL> args = [ '<STR_LIT>' , '<STR_LIT>' , mkv_path , '<STR_LIT>' . format ( stream_idx , output_path ) ] <EOL> subprocess . call ( args ) <EOL> class SCXviD ( object ) : <EOL> @ classmethod <EOL> def make_keyframes ( cls , video_path , log_path ) : <EOL> try : <EOL> ffmpeg_process = subprocess . Popen ( [ '<STR_LIT>' , '<STR_LIT>' , video_path , <EOL> '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT:->' ] , stdout = subprocess . PIPE ) <EOL> except OSError as e : <EOL> if e . errno == <NUM_LIT:2> : <EOL> raise SushiError ( "<STR_LIT>" ) <EOL> raise <EOL> try : <EOL> scxvid_process = subprocess . Popen ( [ '<STR_LIT>' , log_path ] , stdin = ffmpeg_process . stdout ) <EOL> except OSError as e : <EOL> ffmpeg_process . kill ( ) <EOL> if e . errno == <NUM_LIT:2> : <EOL> raise SushiError ( "<STR_LIT>" ) <EOL> raise <EOL> scxvid_process . wait ( ) <EOL> class Timecodes ( object ) : <EOL> def __init__ ( self , times , default_fps ) : <EOL> super ( Timecodes , self ) . __init__ ( ) <EOL> self . times = times <EOL> self . default_frame_duration = <NUM_LIT:1.0> / default_fps if default_fps else None <EOL> def get_frame_time ( self , number ) : <EOL> try : <EOL> return self . times [ number ] <EOL> except IndexError : <EOL> if not self . default_frame_duration : <EOL> return self . get_frame_time ( len ( self . times ) - <NUM_LIT:1> ) <EOL> if self . times : <EOL> return self . times [ - <NUM_LIT:1> ] + ( self . default_frame_duration ) * ( number - len ( self . times ) + <NUM_LIT:1> ) <EOL> else : <EOL> return number * self . default_frame_duration <EOL> def get_frame_number ( self , timestamp ) : <EOL> if ( not self . times or self . times [ - <NUM_LIT:1> ] < timestamp ) and self . default_frame_duration : <EOL> return int ( ( timestamp - sum ( self . times ) ) / self . default_frame_duration ) <EOL> return bisect . bisect_left ( self . times , timestamp ) <EOL> def get_frame_size ( self , timestamp ) : <EOL> try : <EOL> number = bisect . bisect_left ( self . times , timestamp ) <EOL> except : <EOL> return self . default_frame_duration <EOL> c = self . get_frame_time ( number ) <EOL> if number == len ( self . times ) : <EOL> p = self . get_frame_time ( number - <NUM_LIT:1> ) <EOL> return c - p <EOL> else : <EOL> n = self . get_frame_time ( number + <NUM_LIT:1> ) <EOL> return n - c <EOL> @ classmethod <EOL> def _convert_v1_to_v2 ( cls , default_fps , overrides ) : <EOL> overrides = [ ( int ( x [ <NUM_LIT:0> ] ) , int ( x [ <NUM_LIT:1> ] ) , float ( x [ <NUM_LIT:2> ] ) ) for x in overrides ] <EOL> if not overrides : <EOL> return [ ] <EOL> fps = [ default_fps ] * ( overrides [ - <NUM_LIT:1> ] [ <NUM_LIT:1> ] + <NUM_LIT:1> ) <EOL> for o in overrides : <EOL> fps [ o [ <NUM_LIT:0> ] : o [ <NUM_LIT:1> ] + <NUM_LIT:1> ] = [ o [ <NUM_LIT:2> ] ] * ( o [ <NUM_LIT:1> ] - o [ <NUM_LIT:0> ] + <NUM_LIT:1> ) <EOL> v2 = [ <NUM_LIT:0> ] <EOL> for d in ( <NUM_LIT:1.0> / f for f in fps ) : <EOL> v2 . append ( v2 [ - <NUM_LIT:1> ] + d ) <EOL> return v2 <EOL> @ classmethod <EOL> def parse ( cls , text ) : <EOL> lines = text . splitlines ( ) <EOL> if not lines : <EOL> return [ ] <EOL> first = lines [ <NUM_LIT:0> ] . lower ( ) . lstrip ( ) <EOL> if first . startswith ( '<STR_LIT>' ) : <EOL> tcs = [ float ( x ) / <NUM_LIT> for x in lines [ <NUM_LIT:1> : ] ] <EOL> return Timecodes ( tcs , None ) <EOL> elif first . startswith ( '<STR_LIT>' ) : <EOL> default = float ( lines [ <NUM_LIT:1> ] . lower ( ) . replace ( '<STR_LIT>' , "<STR_LIT>" ) ) <EOL> overrides = ( x . split ( '<STR_LIT:U+002C>' ) for x in lines [ <NUM_LIT:2> : ] ) <EOL> return Timecodes ( cls . _convert_v1_to_v2 ( default , overrides ) , default ) <EOL> else : <EOL> raise SushiError ( '<STR_LIT>' ) <EOL> @ classmethod <EOL> def from_file ( cls , path ) : <EOL> with open ( path ) as file : <EOL> return cls . parse ( file . read ( ) ) <EOL> @ classmethod <EOL> def cfr ( cls , fps ) : <EOL> class CfrTimecodes ( object ) : <EOL> def __init__ ( self , fps ) : <EOL> self . frame_duration = <NUM_LIT:1.0> / fps <EOL> def get_frame_time ( self , number ) : <EOL> return number * self . frame_duration <EOL> def get_frame_size ( self , timestamp ) : <EOL> return self . frame_duration <EOL> def get_frame_number ( self , timestamp ) : <EOL> return int ( timestamp / self . frame_duration ) <EOL> return CfrTimecodes ( fps ) <EOL> class Demuxer ( object ) : <EOL> def __init__ ( self , path ) : <EOL> super ( Demuxer , self ) . __init__ ( ) <EOL> self . _path = path <EOL> self . _is_wav = get_extension ( self . _path ) == '<STR_LIT>' <EOL> self . _mi = None if self . _is_wav else FFmpeg . get_media_info ( self . _path ) <EOL> self . _demux_audio = self . _demux_subs = self . _make_timecodes = self . _make_keyframes = self . _write_chapters = False <EOL> @ property <EOL> def is_wav ( self ) : <EOL> return self . _is_wav <EOL> @ property <EOL> def path ( self ) : <EOL> return self . _path <EOL> @ property <EOL> def chapters ( self ) : <EOL> if self . is_wav : <EOL> return [ ] <EOL> return self . _mi . chapters <EOL> @ property <EOL> def has_video ( self ) : <EOL> return not self . is_wav and self . _mi . video <EOL> def set_audio ( self , stream_idx , output_path , sample_rate ) : <EOL> self . _audio_stream = self . _select_stream ( self . _mi . audio , stream_idx , '<STR_LIT>' ) <EOL> self . _audio_output_path = output_path <EOL> self . _audio_sample_rate = sample_rate <EOL> self . _demux_audio = True <EOL> def set_script ( self , stream_idx , output_path ) : <EOL> self . _script_stream = self . _select_stream ( self . _mi . subtitles , stream_idx , '<STR_LIT>' ) <EOL> self . _script_output_path = output_path <EOL> self . _demux_subs = True <EOL> def set_timecodes ( self , output_path ) : <EOL> self . _timecodes_output_path = output_path <EOL> self . _make_timecodes = True <EOL> def set_chapters ( self , output_path ) : <EOL> self . _write_chapters = True <EOL> self . _chapters_output_path = output_path <EOL> def set_keyframes ( self , output_path ) : <EOL> self . _keyframes_output_path = output_path <EOL> self . _make_keyframes = True <EOL> def get_subs_type ( self , stream_idx ) : <EOL> return self . _select_stream ( self . _mi . subtitles , stream_idx , '<STR_LIT>' ) . type <EOL> def demux ( self ) : <EOL> if self . _write_chapters : <EOL> with open ( self . _chapters_output_path , "<STR_LIT:w>" ) as output_file : <EOL> output_file . write ( chapters . format_ogm_chapters ( self . chapters ) ) <EOL> if self . _make_keyframes : <EOL> SCXviD . make_keyframes ( self . _path , self . _keyframes_output_path ) <EOL> ffargs = { } <EOL> if self . _demux_audio : <EOL> ffargs [ '<STR_LIT>' ] = self . _audio_stream . id <EOL> ffargs [ '<STR_LIT>' ] = self . _audio_output_path <EOL> ffargs [ '<STR_LIT>' ] = self . _audio_sample_rate <EOL> if self . _demux_subs : <EOL> ffargs [ '<STR_LIT>' ] = self . _script_stream . id <EOL> ffargs [ '<STR_LIT>' ] = self . _script_output_path <EOL> if self . _make_timecodes : <EOL> def set_ffmpeg_timecodes ( ) : <EOL> ffargs [ '<STR_LIT>' ] = self . _mi . video [ <NUM_LIT:0> ] . id <EOL> ffargs [ '<STR_LIT>' ] = self . _timecodes_output_path <EOL> if get_extension ( self . _path ) . lower ( ) == '<STR_LIT>' : <EOL> try : <EOL> MkvToolnix . extract_timecodes ( self . _path , <EOL> stream_idx = self . _mi . video [ <NUM_LIT:0> ] . id , <EOL> output_path = self . _timecodes_output_path ) <EOL> except OSError as e : <EOL> if e . errno == <NUM_LIT:2> : <EOL> set_ffmpeg_timecodes ( ) <EOL> else : <EOL> raise <EOL> else : <EOL> set_ffmpeg_timecodes ( ) <EOL> if ffargs : <EOL> FFmpeg . demux_file ( self . _path , ** ffargs ) <EOL> def cleanup ( self ) : <EOL> if self . _demux_audio : <EOL> os . remove ( self . _audio_output_path ) <EOL> if self . _demux_subs : <EOL> os . remove ( self . _script_output_path ) <EOL> if self . _make_timecodes : <EOL> os . remove ( self . _timecodes_output_path ) <EOL> if self . _write_chapters : <EOL> os . remove ( self . _chapters_output_path ) <EOL> @ classmethod <EOL> def _format_stream ( cls , stream ) : <EOL> return '<STR_LIT>' . format ( stream . id , '<STR_LIT>' % stream . title if stream . title else '<STR_LIT>' , stream . info ) <EOL> @ classmethod <EOL> def _format_streams_list ( cls , streams ) : <EOL> return '<STR_LIT:\n>' . join ( map ( cls . _format_stream , streams ) ) <EOL> def _select_stream ( self , streams , chosen_idx , name ) : <EOL> if not streams : <EOL> raise SushiError ( '<STR_LIT>' . format ( name , self . _path ) ) <EOL> if chosen_idx is None : <EOL> if len ( streams ) > <NUM_LIT:1> : <EOL> default_track = next ( ( s for s in streams if s . default ) , None ) <EOL> if default_track : <EOL> logging . warning ( '<STR_LIT>' <EOL> . format ( self . _format_stream ( default_track ) , self . _path ) ) <EOL> return default_track <EOL> raise SushiError ( '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' . format ( name , self . _path , self . _format_streams_list ( streams ) ) ) <EOL> return streams [ <NUM_LIT:0> ] <EOL> try : <EOL> return next ( x for x in streams if x . id == chosen_idx ) <EOL> except StopIteration : <EOL> raise SushiError ( "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" . format ( chosen_idx , self . _path , self . _format_streams_list ( streams ) ) ) </s>
