diff --git a/src/tscut/exceptions.py b/src/tscut/exceptions.py index 0aa19a0..84c96b4 100644 --- a/src/tscut/exceptions.py +++ b/src/tscut/exceptions.py @@ -3,10 +3,6 @@ # Copyright (C) 2026 Frédéric Tronel -from typing import IO - -from typeguard import typechecked - class TSCutError(Exception): """Base exception for tscut.""" @@ -21,3 +17,6 @@ class ExternalToolError(TSCutError): class InvalidMediaError(TSCutError): """Raised when input media cannot be processed.""" + +class UnimplementedFeature(TSCutError): + """Raised when a rare feature is encountered and not yet implemented""" diff --git a/src/tscut/h264/bitstream.py b/src/tscut/h264/bitstream.py index b8cae1f..1f3f352 100644 --- a/src/tscut/h264/bitstream.py +++ b/src/tscut/h264/bitstream.py @@ -284,7 +284,7 @@ def write_scaling_list(buf:bytearray, bit_position: int, size, matrix:list[int], bit_position = write_signed_exp_golomb(buf, bit_position, delta) else: logger.error('Not yet implemented') - exit(-1) + raise UnimplementedFeature("Optimized scaling list of H264 header is not implemented") # reverse = deltas.reverse() # compressed = False # while len(reverse)>0: diff --git a/src/tscut/tools/ffmpeg.py b/src/tscut/tools/ffmpeg.py index 4f42e74..7e6d68e 100644 --- a/src/tscut/tools/ffmpeg.py +++ b/src/tscut/tools/ffmpeg.py @@ -3,31 +3,24 @@ # Copyright (C) 2026 Frédéric Tronel import logging -from math import floor, ceil, log from datetime import timedelta -from os import ( - read, - SEEK_SET, - lseek, - memfd_create, - set_inheritable, - write, - close -) -from typing import IO -from subprocess import PIPE, Popen from io import TextIOWrapper +from math import ceil, floor, log +from os import SEEK_SET, close, lseek, memfd_create, read, set_inheritable, write +from subprocess import PIPE, Popen +from typing import IO -from typeguard import typechecked from tqdm import tqdm +from typeguard import typechecked +from tscut.exceptions import ExternalToolError, InvalidMediaError from tscut.tools.ffprobe import ( + get_frames_in_stream, get_video_dimensions, with_subtitles, - get_frames_in_stream, ) from tscut.tools.ppm import dump_ppm -from tscut.tools.timeframe import parse_timestamp, get_packet_duration +from tscut.tools.timeframe import get_packet_duration, parse_timestamp logger = logging.getLogger(__name__) @@ -46,10 +39,7 @@ def ffmpeg_convert(ffmpeg_path:str, ffprobe_path:str, input_file: IO[bytes], inp set_inheritable(infd, True) set_inheritable(outfd, True) - if logger.getEffectiveLevel() == logging.DEBUG: - log_level = [] - else: - log_level = [ '-loglevel', 'quiet' ] + log_level = [] if logger.getEffectiveLevel() == logging.DEBUG else ['-loglevel', 'quiet'] params = [ffmpeg_path, '-y',]+log_level+['-progress', '/dev/stdout', '-canvas_size', f'{width:d}x{height:d}', '-f', input_format, @@ -82,7 +72,7 @@ def ffmpeg_convert(ffmpeg_path:str, ffprobe_path:str, input_file: IO[bytes], inp @typechecked def extract_pictures(ffmpeg_path:str, input_file:IO[bytes], begin:timedelta, nb_frames:int, - width:int=640, height:int=480) -> tuple[bytes,int]|tuple[None,None]: + width:int=640, height:int=480) -> tuple[bytes,int]: """ Extract pictures from a video file using FFmpeg. @@ -126,13 +116,13 @@ def extract_pictures(ffmpeg_path:str, input_file:IO[bytes], begin:timedelta, nb_ status = ffmpeg.wait() if status != 0: logger.error('Conversion failed with status code: %d', status) - return None, None + raise ExternalToolError("") lseek(outfd, 0, SEEK_SET) images = read(outfd,length) if len(images) != length: logger.error("Received %d bytes but %d were expected.", len(images), length) - return None, None + raise InvalidMediaError("") lseek(outfd, 0, SEEK_SET) return images, outfd @@ -141,7 +131,7 @@ def extract_pictures(ffmpeg_path:str, input_file:IO[bytes], begin:timedelta, nb_ def extract_sound(ffmpeg_path:str, input_file: IO[bytes], begin:timedelta, output_filename:str, packet_duration:int, sub_channel:int=0, nb_packets:int=0, sample_rate:int=48000, - nb_channels:int=2) -> tuple[bytes,int]|tuple[None,None]: + nb_channels:int=2) -> tuple[bytes,int]: outfd = memfd_create(output_filename, flags=0) infd = input_file.fileno() lseek(infd, 0, SEEK_SET) @@ -160,7 +150,7 @@ def extract_sound(ffmpeg_path:str, input_file: IO[bytes], begin:timedelta, outpu status = ffmpeg.wait() if status != 0: logger.error('Sound extraction returns error code: %d', status) - return None, None + raise ExternalToolError("") lseek(outfd, 0, SEEK_SET) sound = read(outfd, length) @@ -169,7 +159,7 @@ def extract_sound(ffmpeg_path:str, input_file: IO[bytes], begin:timedelta, outpu logger.info("Received %d bytes but %d were expected (channels=%d, freq=%d, packets=%d,\ duration=%d ms).", len(sound), length, nb_channels, sample_rate, nb_packets, packet_duration) - return None, None + raise InvalidMediaError("") return sound, outfd @@ -236,9 +226,6 @@ def extract_all_streams(ffmpeg_path:str, ffprobe_path:str, input_file:IO[bytes], images_bytes, memfd = extract_pictures(ffmpeg_path, input_file=input_file, begin=begin, nb_frames=nb_frames, width=width, height=height) - if images_bytes is None or memfd is None: - logger.error('Impossible to extract picture from video stream.') - exit(-1) memfds.append(memfd) if dump_mem_fd: @@ -262,14 +249,10 @@ def extract_all_streams(ffmpeg_path:str, ffprobe_path:str, input_file:IO[bytes], logger.debug('Audio stream: %s', stream) sample_rate = int(stream['sample_rate']) nb_channels = int(stream['channels']) - if 'bit_rate' in stream: - bit_rate = int(stream['bit_rate']) - else: - bit_rate = 128000 + bit_rate = int(stream['bit_rate']) if 'bit_rate' in stream else 128000 codec = stream['codec_name'] - if 'tags' in stream: - if 'language' in stream['tags']: - generic_codec_params.extend([f'-metadata:s:a:{audio_id:d}', + if 'tags' in stream and 'language' in stream['tags']: + generic_codec_params.extend([f'-metadata:s:a:{audio_id:d}', f"language={stream['tags']['language']}"]) packets = get_frames_in_stream(ffprobe_path, input_file=input_file, begin=begin, end=end, stream_kind='a', sub_stream_id=audio_id) @@ -294,10 +277,6 @@ def extract_all_streams(ffmpeg_path:str, ffprobe_path:str, input_file:IO[bytes], output_filename=tmpname, sample_rate=sample_rate, nb_channels=nb_channels) - if sound_bytes is None or memfd is None: - logger.error('Impossible to extract sound track') - exit(-1) - memfds.append(memfd) if dump_mem_fd: @@ -325,9 +304,8 @@ def extract_all_streams(ffmpeg_path:str, ffprobe_path:str, input_file:IO[bytes], logger.info("Extracting a subtitle stream: s:%d", subtitle_id) codec = stream['codec_name'] generic_input_params.extend(['-i', './empty.idx']) - if 'tags' in stream: - if 'language' in stream['tags']: - generic_codec_params.extend([f'-metadata:s:s:{subtitle_id:d}', + if 'tags' in stream and 'language' in stream['tags']: + generic_codec_params.extend([f'-metadata:s:s:{subtitle_id:d}', f"language={stream['tags']['language']}"]) generic_codec_params.extend([f'-c:s:{subtitle_id:d}', 'copy']) subtitle_id=subtitle_id+1 diff --git a/src/tscut/tools/ffprobe.py b/src/tscut/tools/ffprobe.py index d451392..abbeb5b 100644 --- a/src/tscut/tools/ffprobe.py +++ b/src/tscut/tools/ffprobe.py @@ -4,17 +4,17 @@ import json import logging -from math import floor -from datetime import timedelta import os +from datetime import timedelta +from io import BytesIO +from math import floor from os import ( SEEK_SET, lseek, set_inheritable, ) -from typing import IO from subprocess import PIPE, Popen -from io import BytesIO +from typing import IO from typeguard import typechecked @@ -63,9 +63,8 @@ def get_frame_rate(ffprobe_path:str, input_file: IO[bytes]) -> float|None: out = json.load(BytesIO(out)) if 'frames' in out: for frame in out['frames']: - if 'interlaced_frame' in frame: - if frame['interlaced_frame'] == 1: - interlaced = True + if 'interlaced_frame' in frame and frame['interlaced_frame'] == 1: + interlaced = True if 'pts_time' in frame: ts = float(frame['pts_time']) if min_ts is None: @@ -165,8 +164,7 @@ def get_movie_duration(ffprobe_path:str, input_file: IO[bytes]) -> timedelta|Non out = json.load(BytesIO(out)) if 'format' in out and 'duration' in out['format']: duration = floor(float(out['format']['duration'])) - ts = timedelta(seconds=duration) - return ts + return timedelta(seconds=duration) logger.error('Impossible to retrieve duration of movie') return None @@ -327,22 +325,22 @@ def get_nearest_idr_frame(ffprobe_path: str, input_file: IO[bytes], timestamp:ti status = ffprobe.wait() if status != 0: logger.error('ffprobe failed with status code: %d', status) - return None + return if 'frames' in frames: frames = frames['frames'] for frame in frames: ts = get_ts_frame(frame) if ts is None: - return None + return if tbegin <= ts <= tend: idrs.append(frame) else: logger.error('Impossible to retrieve IDR frames inside file around [%s,%s]', tbegin, tend) - return None + return - return None + return @typechecked def get_nearest_iframe(ffprobe_path:str, input_file: IO[bytes], @@ -356,14 +354,8 @@ def get_nearest_iframe(ffprobe_path:str, input_file: IO[bytes], iframe = None while delta < delta_max: zero = timedelta() - if before: - tbegin = timestamp-delta - else: - tbegin = timestamp - if not before: - tend = timestamp+delta - else: - tend = timestamp + tbegin = timestamp - delta if before else timestamp + tend = timestamp + delta if not before else timestamp tbegin = max(tbegin, zero) logger.debug('Looking for an iframe in [%s, %s]', tbegin, tend) diff --git a/src/tscut/tools/ppm.py b/src/tscut/tools/ppm.py index 62d6800..adfbfa6 100644 --- a/src/tscut/tools/ppm.py +++ b/src/tscut/tools/ppm.py @@ -5,14 +5,14 @@ import logging import re +from io import BytesIO from math import ceil, log from os import write from typing import IO -from io import BytesIO - from typeguard import typechecked + @typechecked def dump_ppm(pictures: bytes, prefix: str, temporaries: list[IO[bytes]]) -> None: """ diff --git a/src/tscut/tools/timeframe.py b/src/tscut/tools/timeframe.py index 5899de4..03edab5 100644 --- a/src/tscut/tools/timeframe.py +++ b/src/tscut/tools/timeframe.py @@ -65,9 +65,8 @@ def parse_timestamp(ts:str) -> timedelta|None: if us < 0 or us > 1000000: logger.error("milliseconds must be in [0,1000000[") return None - res = timedelta(hours=hour, minutes=minute, seconds=second, microseconds=us) - return res + return timedelta(hours=hour, minutes=minute, seconds=second, microseconds=us) @typechecked def parse_time_interval(interval: str) -> tuple[timedelta, timedelta] | tuple[None, None]: @@ -204,8 +203,7 @@ def get_ts_frame(frame: dict) -> timedelta|None: logger.error('Impossible to find timestamp of frame %s', frame) return None - ts = timedelta(seconds=pts_time) - return ts + return timedelta(seconds=pts_time) @typechecked def get_packet_duration(packet: dict) -> int | None: diff --git a/src/tscut/tscut.py b/src/tscut/tscut.py index 8805908..c7bec65 100755 --- a/src/tscut/tscut.py +++ b/src/tscut/tscut.py @@ -143,8 +143,7 @@ def concatenate_h264_parts(h264parts: list[IO[bytes]], output: IO[bytes]) -> Non def concatenate_h264_ts_parts(h264_ts_parts: list[IO[bytes]], output: IO[bytes]) -> None: logger = logging.getLogger(__name__) - header = '# timestamp format v2\n'.encode('ascii') - + header = '# timestamp format v2\n' output.write(header) last = 0. @@ -166,7 +165,7 @@ def concatenate_h264_ts_parts(h264_ts_parts: list[IO[bytes]], output: IO[bytes]) break ts = offset + float(line) last = max(last,ts) - output.write(f'{ts:f}\n'.encode('ascii')) + output.write(f'{ts:f}\n') if first: first = False