diff --git a/src/tscut/cli.py b/src/tscut/cli.py index 581fac5..240f2d8 100644 --- a/src/tscut/cli.py +++ b/src/tscut/cli.py @@ -13,6 +13,7 @@ from functools import cmp_to_key import coloredlogs +from tscut.exceptions import TSCutError from tscut.config import ProcessingOptions from tscut.tools.discovery import check_required_tools from tscut.tools.timeframe import parse_time_interval, compare_time_interval diff --git a/src/tscut/h264/avc.py b/src/tscut/h264/avc.py index 9cc0d92..b8f51de 100644 --- a/src/tscut/h264/avc.py +++ b/src/tscut/h264/avc.py @@ -243,7 +243,7 @@ class AVCDecoderConfiguration: # TODO: do the same with extended SPS ! @typechecked -def parse_codec_private(codec_private_data: bytes) -> AVCDecoderConfiguration: +def parse_codec_private(codec_private_data: bytes | bytearray) -> AVCDecoderConfiguration: if codec_private_data[0] != 0x63: raise ValueError(f'Matroska header is wrong: {codec_private_data[0]:x}') if codec_private_data[1] != 0xA2: diff --git a/src/tscut/pipeline.py b/src/tscut/pipeline.py index aed15c6..51c96a2 100644 --- a/src/tscut/pipeline.py +++ b/src/tscut/pipeline.py @@ -11,6 +11,7 @@ import logging import hexdump +from tscut.exceptions import InvalidMediaError from tscut.config import ProcessingOptions from tscut.tools.mkvtoolnix import ( get_codec_private_data_from_mkv, @@ -90,13 +91,13 @@ def process_recording(options: ProcessingOptions) -> None: if 'format_name' in format_of_file: if format_of_file['format_name'] == str(f): found = True - format_of_file = f + final_format_of_file = f break if not found: logger.error('Unsupported format of file') - if format_of_file == SupportedFormat.TS: + if final_format_of_file == SupportedFormat.TS: logger.info("Converting TS to MP4 (to fix timestamps).") try: with open(mp4filename, 'wb+') as mp4: @@ -116,7 +117,7 @@ def process_recording(options: ProcessingOptions) -> None: except OSError: logger.error('') - elif format_of_file == SupportedFormat.MP4: + elif final_format_of_file == SupportedFormat.MP4: logger.info("Converting MP4 to MKV") try: mkv = open(mkvfilename, 'wb+') @@ -221,15 +222,13 @@ def process_recording(options: ProcessingOptions) -> None: # Get the nearest I-frame whose timestamp is greater or equal to the beginning. head_frames = get_nearest_iframe(options.tools_paths['ffprobe'], mkv, ts1, before=False) if head_frames is None: - logger.error('Impossible to retrieve I-frame') - exit(-1) + raise InvalidMediaError("Impossible to retrieve first I-frame") # Get the nearest I-frame whose timestamp ... # TODO: wrong here ... tail_frames = get_nearest_iframe(options.tools_paths['ffprobe'], mkv, ts2, before=True) if tail_frames is None: - logger.error('Impossible to retrieve I-frame') - exit(-1) + raise InvalidMediaError("Impossible to retrieve last I-frame") nb_head_frames, head_iframe = head_frames nb_tail_frames, tail_iframe = tail_frames @@ -241,10 +240,10 @@ def process_recording(options: ProcessingOptions) -> None: head_iframe_ts = get_ts_frame(head_iframe) if head_iframe_ts is None: - exit(-1) + raise InvalidMediaError("Impossible to retrieve timestamp of first i-frame.") tail_iframe_ts = get_ts_frame(tail_iframe) if tail_iframe_ts is None: - exit(-1) + raise InvalidMediaError("Impossible to retrieve timestamp of final i-frame.") checks.append(pos+head_iframe_ts-ts1) @@ -513,10 +512,10 @@ def process_recording(options: ProcessingOptions) -> None: if not options.keep_temporaries: logger.info("Cleaning temporary files") - for f in temporaries: - path = os.path.realpath(f.name) + for tmp in temporaries: + path = os.path.realpath(tmp.name) logger.info("Removing: %s", path) - f.close() + tmp.close() unlink(path) d = datetime(1,1,1) diff --git a/src/tscut/subtitles/ocr.py b/src/tscut/subtitles/ocr.py index f62076f..c5abbaf 100644 --- a/src/tscut/subtitles/ocr.py +++ b/src/tscut/subtitles/ocr.py @@ -4,7 +4,7 @@ import logging -from typing import IO +from typing import IO, Sequence from io import TextIOWrapper from datetime import timedelta from os import ( @@ -24,6 +24,8 @@ from iso639.exceptions import InvalidLanguageValue from typeguard import typechecked from tqdm import tqdm +from tscut.exceptions import ExternalToolError + logger = logging.getLogger(__name__) @typechecked @@ -70,7 +72,7 @@ def get_tesseract_supported_lang(tesseract_path:str) -> dict[Lang, str]|None: @typechecked def extract_srt(mkvextract:str, filename:str, subtitles:dict[str, list[int]], - langs:dict[Lang,str]) -> list[tuple[str,str,str,str]]|None: + langs:dict[Lang,str]) -> list[tuple[str,str,str,str]]: params = [mkvextract, filename, 'tracks'] res = [] @@ -120,13 +122,12 @@ def extract_srt(mkvextract:str, filename:str, subtitles:dict[str, list[int]], case 1: logger.warning('Mkvextract returns warning') case 2: - logger.error('Mkvextract returns an error code: %d', extract.returncode) - res = None + raise ExternalToolError('Mkvextract returns an error code: %d', extract.returncode) return res @typechecked -def do_ocr(vobsubocr:str, idxs: list[tuple[str,str,str,str]], duration:timedelta, +def do_ocr(vobsubocr:str, idxs: Sequence[tuple[str,str,str,str]], duration:timedelta, temporaries:list[IO[bytes]], dump_mem_fd:bool=False): res = [] diff --git a/src/tscut/tools/ffmpeg.py b/src/tscut/tools/ffmpeg.py index 7e6d68e..9a68e24 100644 --- a/src/tscut/tools/ffmpeg.py +++ b/src/tscut/tools/ffmpeg.py @@ -56,6 +56,7 @@ def ffmpeg_convert(ffmpeg_path:str, ffprobe_path:str, input_file: IO[bytes], inp logger.debug('Executing %s', params) with Popen(params, stdout=PIPE, close_fds=False) as ffmpeg: + assert ffmpeg.stdout is not None pb = tqdm(TextIOWrapper(ffmpeg.stdout, encoding="utf-8"), total=int(duration/timedelta(seconds=1)), unit='s', desc='Conversion') for line in pb: diff --git a/src/tscut/tools/ffprobe.py b/src/tscut/tools/ffprobe.py index abbeb5b..217041e 100644 --- a/src/tscut/tools/ffprobe.py +++ b/src/tscut/tools/ffprobe.py @@ -19,6 +19,7 @@ from typing import IO from typeguard import typechecked from tscut.tools.timeframe import get_ts_frame +from tscut.exceptions import InvalidMediaError logger = logging.getLogger(__name__) @@ -154,7 +155,7 @@ def get_format(ffprobe_path:str, input_file: IO[bytes]) -> dict|None: return None @typechecked -def get_movie_duration(ffprobe_path:str, input_file: IO[bytes]) -> timedelta|None: +def get_movie_duration(ffprobe_path:str, input_file: IO[bytes]) -> timedelta: infd = input_file.fileno() lseek(infd, 0, SEEK_SET) set_inheritable(infd, True) @@ -165,9 +166,7 @@ def get_movie_duration(ffprobe_path:str, input_file: IO[bytes]) -> timedelta|Non if 'format' in out and 'duration' in out['format']: duration = floor(float(out['format']['duration'])) return timedelta(seconds=duration) - logger.error('Impossible to retrieve duration of movie') - - return None + raise InvalidMediaError("Impossible to retrieve duration of movie") # ffprobe -loglevel quiet -select_streams v:0 -show_entries stream=width,height -of json sample.ts @typechecked @@ -242,7 +241,7 @@ def with_subtitles(ffprobe_path:str, input_file: IO[bytes]) -> bool: return False -def find_subtitles_tracks(ffprobe_path:str, input_file: IO[bytes]) -> dict|None: +def find_subtitles_tracks(ffprobe_path:str, input_file: IO[bytes]) -> dict: infd = input_file.fileno() lseek(infd, 0, SEEK_SET) set_inheritable(infd, True) @@ -257,10 +256,9 @@ def find_subtitles_tracks(ffprobe_path:str, input_file: IO[bytes]) -> dict|None: out = json.load(BytesIO(out)) if 'streams' in out: return out['streams'] - logger.error('Impossible to retrieve format of file') ffprobe.wait() - return None + raise InvalidMediaError('Impossible to retrieve format of file') @typechecked def get_frames_in_stream(ffprobe_path: str, input_file: IO[bytes], begin:timedelta, end:timedelta, @@ -345,7 +343,7 @@ def get_nearest_idr_frame(ffprobe_path: str, input_file: IO[bytes], timestamp:ti @typechecked def get_nearest_iframe(ffprobe_path:str, input_file: IO[bytes], timestamp:timedelta, before:bool=True, - delta_max:timedelta=timedelta(seconds=15))-> tuple[int,dict | None]: + delta_max:timedelta=timedelta(seconds=15))-> tuple[int,dict] | None: infd = input_file.fileno() set_inheritable(infd, True) @@ -396,7 +394,7 @@ def get_nearest_iframe(ffprobe_path:str, input_file: IO[bytes], its = get_ts_frame(iframe) if its is None: logger.error("Impossible to retrieve timestamp of i-frame !") - return 0, None + return None nb_frames = 0 for frame in frames: ts = get_ts_frame(frame) @@ -414,6 +412,6 @@ def get_nearest_iframe(ffprobe_path:str, input_file: IO[bytes], nb_frames = nb_frames+1 else: logger.error("Impossible to find I-frame between: %s and %s", tbegin, tend) - return 0, None + return None return(nb_frames, iframe) diff --git a/src/tscut/tools/mkvtoolnix.py b/src/tscut/tools/mkvtoolnix.py index ab0be46..d6efad9 100644 --- a/src/tscut/tools/mkvtoolnix.py +++ b/src/tscut/tools/mkvtoolnix.py @@ -125,7 +125,8 @@ def parse_mkv_tree(mkvinfo_path:str, input_file: IO[bytes]) -> dict[str,tuple[in return elements @typechecked -def change_codec_private_data(mkvinfo_path:str, input_file: IO[bytes], codec_data:bytes) -> None: +def change_codec_private_data(mkvinfo_path:str, input_file: IO[bytes], + codec_data:bytes | bytearray) -> None: logger = logging.getLogger(__name__) infd = input_file.fileno() @@ -217,6 +218,7 @@ def extract_mkv_part(mkvmerge_path:str, input_file:IO[bytes], output_file:IO[byt logger.debug('Executing: %s', command) with Popen(command, stdout=PIPE, close_fds=False, env=env) as mkvmerge: + assert mkvmerge.stdout is not None pb = tqdm(TextIOWrapper(mkvmerge.stdout, encoding="utf-8"), total=100, unit='%', desc='Extraction') for line in pb: @@ -225,7 +227,8 @@ def extract_mkv_part(mkvmerge_path:str, input_file:IO[bytes], output_file:IO[byt m = p.match(line) if m is None: logger.error('Impossible to parse progress') - pb.update(int(m['progress'])-pb.n) + else: + pb.update(int(m['progress'])-pb.n) elif line.startswith('Warning'): warnings.append(line) pb.update(100-pb.n) @@ -245,7 +248,7 @@ def extract_mkv_part(mkvmerge_path:str, input_file:IO[bytes], output_file:IO[byt @typechecked def merge_mkvs(mkvmerge_path:str, inputs: list[IO[bytes]], output_name:str, concatenate: bool=True, - timestamps: dict[int, IO[str]] | None = None) -> IO[bytes]|None: + timestamps: dict[int, IO[str]] | None = None) -> IO[bytes]: logger = logging.getLogger(__name__) if timestamps is None: @@ -254,9 +257,9 @@ def merge_mkvs(mkvmerge_path:str, inputs: list[IO[bytes]], output_name:str, fds = [] try: out = open(output_name, 'wb+') - except OSError: + except OSError as e: logger.error('Impossible to create file: %s', output_name) - return None + raise e outfd = out.fileno() lseek(outfd, 0, SEEK_SET) @@ -299,6 +302,7 @@ def merge_mkvs(mkvmerge_path:str, inputs: list[IO[bytes]], output_name:str, logger.debug('Executing: LANG=C %s', merge_params) with Popen(merge_params, stdout=PIPE, close_fds=False, env=env) as mkvmerge: + assert mkvmerge.stdout is not None pb = tqdm(TextIOWrapper(mkvmerge.stdout, encoding="utf-8"), total=100, unit='%', desc='Merging') for line in pb: @@ -307,8 +311,9 @@ def merge_mkvs(mkvmerge_path:str, inputs: list[IO[bytes]], output_name:str, m = p.match(line) if m is None: logger.error('Impossible to parse progress') - pb.n = int(m['progress']) - pb.update() + else: + pb.n = int(m['progress']) + pb.update() elif line.startswith('Warning'): warnings.append(line) @@ -352,6 +357,7 @@ def extract_track_from_mkv(mkvextract_path: str, input_file: IO[bytes], index, logger.debug('Executing: LANG=C %s', params) with Popen(params, stdout=PIPE, close_fds=False, env=env) as extract: + assert extract.stdout is not None pb = tqdm(TextIOWrapper(extract.stdout, encoding="utf-8"), total=100, unit='%', desc='Extraction of track') for line in pb: @@ -360,7 +366,8 @@ def extract_track_from_mkv(mkvextract_path: str, input_file: IO[bytes], index, m = p.match(line) if m is None: logger.error('Impossible to parse progress') - pb.update(int(m['progress'])-pb.n) + else: + pb.update(int(m['progress'])-pb.n) pb.update(100-pb.n) pb.refresh() pb.close() @@ -391,6 +398,7 @@ def remove_video_tracks_from_mkv(mkvmerge_path:str, input_file: IO[bytes], env = {**os.environ, 'LANG': 'C'} with Popen(params, stdout=PIPE, close_fds=False, env=env) as remove: + assert remove.stdout is not None pb = tqdm(TextIOWrapper(remove.stdout, encoding="utf-8"), total=100, unit='%', desc='Removal of video track:') for line in pb: @@ -399,7 +407,8 @@ def remove_video_tracks_from_mkv(mkvmerge_path:str, input_file: IO[bytes], m = p.match(line) if m is None: logger.error('Impossible to parse progress') - pb.update(int(m['progress'])-pb.n) + else: + pb.update(int(m['progress'])-pb.n) pb.update(100-pb.n) pb.refresh() pb.close() @@ -420,7 +429,7 @@ def remux_srt_subtitles(mkvmerge_path:str, input_file: IO[bytes], output_file: P try: out = open(output_file, 'w', encoding='utf8') except OSError: - logger.error('Impossible to create file: %s', output_filename) + logger.error('Impossible to create file: %s', output_file) return None outfd = out.fileno() @@ -441,6 +450,7 @@ def remux_srt_subtitles(mkvmerge_path:str, input_file: IO[bytes], output_file: P env = {**os.environ, 'LANG': 'C'} logger.info('Remux subtitles: %s', mkv_merge_params) with Popen(mkv_merge_params, stdout=PIPE, close_fds=False, env=env) as mkvmerge: + assert mkvmerge.stdout is not None pb = tqdm(TextIOWrapper(mkvmerge.stdout, encoding="utf-8"), total=100, unit='%', desc='Remux subtitles:') for line in pb: @@ -449,8 +459,9 @@ def remux_srt_subtitles(mkvmerge_path:str, input_file: IO[bytes], output_file: P m = p.match(line) if m is None: logger.error('Impossible to parse progress') - pb.n = int(m['progress']) - pb.update() + else: + pb.n = int(m['progress']) + pb.update() elif line.startswith('Warning'): warnings.append(line) diff --git a/src/tscut/tscut.py b/src/tscut/tscut.py index 43989d4..5aaa4f8 100755 --- a/src/tscut/tscut.py +++ b/src/tscut/tscut.py @@ -12,7 +12,7 @@ from os import ( write, ) from sys import exit -from typing import IO +from typing import IO, BinaryIO, TextIO, Sequence # Third party libraries from tqdm import tqdm @@ -115,7 +115,7 @@ class SupportedFormat(IntEnum): @typechecked -def concatenate_h264_parts(h264parts: list[IO[bytes]], output: IO[bytes]) -> None: +def concatenate_h264_parts(h264parts: Sequence[BinaryIO], output: BinaryIO) -> None: logger = logging.getLogger(__name__) total_length = 0 @@ -142,7 +142,7 @@ def concatenate_h264_parts(h264parts: list[IO[bytes]], output: IO[bytes]) -> Non pb.update(nb_bytes) pos += nb_bytes -def concatenate_h264_ts_parts(h264_ts_parts: list[IO[bytes]], output: IO[bytes]) -> None: +def concatenate_h264_ts_parts(h264_ts_parts: Sequence[TextIO], output: TextIO) -> None: logger = logging.getLogger(__name__) header = '# timestamp format v2\n' output.write(header)