Public/removeads
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Improve error handling.

Browse Source
This commit is contained in:
Frédéric Tronel
2026-08-31 13:54:03 +02:00
parent 153873637e
commit 51bab76c00
5 changed files with 45 additions and 49 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/tscut/temporaries.py
+2 -2
View File
@@ -6,7 +6,7 @@
import logging
import os
from typing import IO, Any
from typing import IO, Any, Self
logger = logging.getLogger(__name__)
@@ -34,5 +34,5 @@ class TemporaryFiles:
def __enter__(self) -> Self:
return self
def __exit__(self, ...) -> None:
def __exit__(self) -> None:
self.cleanup()
src/tscut/tools/ffmpeg.py
+31 -38
View File
@@ -8,12 +8,12 @@ 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 typing import IO, BinaryIO
from tqdm import tqdm
from typeguard import typechecked
from tscut.exceptions import ExternalToolError, InvalidMediaError
from tscut.exceptions import ExternalToolError, InvalidMediaError, TemporaryFileError
from tscut.tools.ffprobe import (
get_frames_in_stream,
get_video_dimensions,
@@ -21,13 +21,14 @@ from tscut.tools.ffprobe import (
)
from tscut.tools.ppm import dump_ppm
from tscut.tools.timeframe import get_packet_duration, parse_timestamp
from tscut.temporaries import TemporaryFiles
logger = logging.getLogger(__name__)
@typechecked
def ffmpeg_convert(ffmpeg_path:str, ffprobe_path:str, input_file: IO[bytes], input_format:str,
output_file: IO[bytes], output_format:str, duration: timedelta):
output_file: IO[bytes], output_format:str, duration: timedelta) -> None:
width, height = get_video_dimensions(ffprobe_path, input_file)
if width is None or height is None:
return
@@ -68,7 +69,7 @@ def ffmpeg_convert(ffmpeg_path:str, ffprobe_path:str, input_file: IO[bytes], inp
pb.update()
status = ffmpeg.wait()
if status != 0:
logger.error('Conversion failed with status code: %d', status)
raise ExternalToolError(f"Conversion failed with status code: {status:d}")
@typechecked
@@ -116,14 +117,12 @@ def extract_pictures(ffmpeg_path:str, input_file:IO[bytes], begin:timedelta, nb_
with Popen(command, stdout=PIPE, close_fds=False) as ffmpeg:
status = ffmpeg.wait()
if status != 0:
logger.error('Conversion failed with status code: %d', status)
raise ExternalToolError("")
raise ExternalToolError(f"Conversion failed with status code: {status:d}")
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)
raise InvalidMediaError("")
raise InvalidMediaError(f"Received {len(images)} bytes but {length} were expected.")
lseek(outfd, 0, SEEK_SET)
return images, outfd
@@ -150,24 +149,25 @@ def extract_sound(ffmpeg_path:str, input_file: IO[bytes], begin:timedelta, outpu
with Popen(command, stdout=PIPE, close_fds=False) as ffmpeg:
status = ffmpeg.wait()
if status != 0:
logger.error('Sound extraction returns error code: %d', status)
raise ExternalToolError("")
raise ExternalToolError(f"Sound extraction returns error code: {status}")
lseek(outfd, 0, SEEK_SET)
sound = read(outfd, length)
if len(sound) != length:
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)
raise InvalidMediaError("")
raise InvalidMediaError(f"Received {len(sound)} bytes but {length} were expected (\
channels={nb_channels}, freq={sample_rate} packets={nb_packets},\
duration={packet_duration} ms).")
return sound, outfd
@typechecked
def extract_all_streams(ffmpeg_path:str, ffprobe_path:str, input_file:IO[bytes], begin:timedelta,
end:timedelta, streams, files_prefix, nb_frames:int, framerate:float,
width:int, height:int, temporaries, dump_mem_fd:bool=False):
width:int, height:int, temporaries:TemporaryFiles,
dump_mem_fd:bool=False) -> tuple[BinaryIO|None,
TextIOWrapper|None,
BinaryIO|None]:
# The command line for encoding only video track
video_encoder_params = [ ffmpeg_path, '-y', '-loglevel', 'quiet']
video_input_params = []
@@ -258,8 +258,7 @@ def extract_all_streams(ffmpeg_path:str, ffprobe_path:str, input_file:IO[bytes],
packets = get_frames_in_stream(ffprobe_path, input_file=input_file, begin=begin,
end=end, stream_kind='a', sub_stream_id=audio_id)
if packets is None:
logger.error("Impossible to retrieve audio packets")
return None
raise InvalidMediaError("Impossible to retrieve audio packets")
nb_packets = len(packets)
logger.debug("Found %d packets to be extracted from audio track.", nb_packets)
if nb_packets > 0:
@@ -283,14 +282,13 @@ def extract_all_streams(ffmpeg_path:str, ffprobe_path:str, input_file:IO[bytes],
if dump_mem_fd:
try:
with open(tmpname,'wb') as output:
temporaries.append(output)
temporaries.add(output)
outfd = output.fileno()
pos = 0
while pos < len(sound_bytes):
pos+=write(outfd, sound_bytes[pos:])
except OSError:
logger.error('Impossible to create file: %s', tmpname)
return None
except OSError as e:
raise TemporaryFileError(f"Impossible to create file: {tmpname}") from e
# We rewind to zero the memory file descriptor
lseek(memfd, 0, SEEK_SET)
@@ -323,9 +321,8 @@ def extract_all_streams(ffmpeg_path:str, ffprobe_path:str, input_file:IO[bytes],
mkv_filename = f'{files_prefix}.mkv'
try:
mkv_output = open(mkv_filename,'wb+')
except OSError:
logger.error('Impossible to create file: %s', mkv_filename)
return None
except OSError as e:
raise TemporaryFileError(f"Impossible to create file: {mkv_filename}") from e
mkvoutfd = mkv_output.fileno()
set_inheritable(mkvoutfd, True)
@@ -336,17 +333,15 @@ def extract_all_streams(ffmpeg_path:str, ffprobe_path:str, input_file:IO[bytes],
with Popen(generic_encoder_params, stdout=PIPE, close_fds=False) as ffmpeg:
status = ffmpeg.wait()
if status != 0:
logger.error('Encoding failed with status code: %d', status)
return None
raise ExternalToolError(f"Encoding failed with status code: {status}")
temporaries.append(mkv_output)
temporaries.add(mkv_output)
h264_filename = f'{files_prefix}.h264'
try:
h264_output = open(h264_filename,'wb+')
except OSError:
logger.error('Impossible to create file: %s', h264_filename)
return None
except OSError as e:
raise TemporaryFileError(f"Impossible to create file {h264_filename}") from e
h264outfd = h264_output.fileno()
set_inheritable(h264outfd, True)
@@ -365,17 +360,15 @@ def extract_all_streams(ffmpeg_path:str, ffprobe_path:str, input_file:IO[bytes],
with Popen(video_encoder_params, stdout=PIPE, close_fds=False) as ffmpeg:
status = ffmpeg.wait()
if status != 0:
logger.error('Encoding failed with status code: %d', status)
return None
raise ExternalToolError(f"Encoding failed with status code: {status:d}")
temporaries.append(h264_output)
temporaries.add(h264_output)
h264_ts_filename = f'{files_prefix}-ts.txt'
try:
h264_ts_output = open(h264_ts_filename,'w+', encoding='utf8')
except OSError:
logger.error('Impossible to create file: %s', h264_ts_filename)
return None
except OSError as e:
raise TemporaryFileError(f"Impossible to create file: {h264_ts_filename}") from e
h264_ts_output.write('# timestamp format v2\n')
ts = 0
@@ -385,7 +378,7 @@ def extract_all_streams(ffmpeg_path:str, ffprobe_path:str, input_file:IO[bytes],
h264_ts_output.flush()
h264_ts_output.seek(0)
temporaries.append(h264_ts_output)
temporaries.add(h264_ts_output)
for memfd in memfds:
close(memfd)
@@ -393,5 +386,5 @@ def extract_all_streams(ffmpeg_path:str, ffprobe_path:str, input_file:IO[bytes],
return h264_output, h264_ts_output, mkv_output
# Nothing to be done. We are already at a i-frame boundary.
return None, None
return None, None, None
src/tscut/tools/mkvtoolnix.py
+2 -2
View File
@@ -11,7 +11,7 @@ from io import TextIOWrapper
from os import SEEK_SET, fstat, ftruncate, lseek, read, set_inheritable, write
from pathlib import Path
from subprocess import PIPE, Popen
from typing import IO
from typing import IO, Sequence
import hexdump
from tqdm import tqdm
@@ -238,7 +238,7 @@ def extract_mkv_part(mkvmerge_path:str, input_file:IO[bytes], output_file:IO[byt
# Merge a list of mkv files passed as input, and produce a new MKV as output
@typechecked
def merge_mkvs(mkvmerge_path:str, inputs: list[IO[bytes]], output_name:str,
def merge_mkvs(mkvmerge_path:str, inputs: Sequence[IO[bytes]], output_name:str,
concatenate: bool=True,
timestamps: dict[int, IO[str]] | None = None) -> IO[bytes]:
logger = logging.getLogger(__name__)
src/tscut/tools/ppm.py
+6 -4
View File
@@ -12,9 +12,11 @@ from typing import IO
from typeguard import typechecked
from tscut.temporaries import TemporaryFiles
from tscut.exceptions import TemporaryFileError
@typechecked
def dump_ppm(pictures: bytes, prefix: str, temporaries: list[IO[bytes]]) -> None:
def dump_ppm(pictures: bytes, prefix: str, temporaries: TemporaryFiles) -> None:
"""
Dump PPM pictures from a bytes buffer to files.
@@ -71,7 +73,7 @@ def dump_ppm(pictures: bytes, prefix: str, temporaries: list[IO[bytes]]) -> None
header_len=2+1+ceil(log(width, 10))+1+ceil(log(height, 10))+1+3+1
try:
with open(filename, 'wb') as out:
temporaries.append(out)
temporaries.add(out)
outfd = out.fileno()
length=header_len+3*width*height
nb_bytes = 0
@@ -79,5 +81,5 @@ def dump_ppm(pictures: bytes, prefix: str, temporaries: list[IO[bytes]]) -> None
nb_bytes+=write(outfd, pictures[pos+nb_bytes:pos+length])
pos+=length
picture+=1
except OSError:
logger.error('Impossible to create file: %s', filename)
except OSError as e:
raise TemporaryFileError(f"Impossible to create file {filename}") from e
src/tscut/tools/timeframe.py
+4 -3
View File
@@ -9,6 +9,8 @@ from datetime import timedelta
from typeguard import typechecked
from tscut.exceptions import InvalidMediaError
@typechecked
def parse_timestamp(ts:str) -> timedelta|None:
@@ -206,7 +208,7 @@ def get_ts_frame(frame: dict) -> timedelta|None:
return timedelta(seconds=pts_time)
@typechecked
def get_packet_duration(packet: dict) -> int | None:
def get_packet_duration(packet: dict) -> int:
logger = logging.getLogger(__name__)
if 'duration' in packet:
@@ -214,7 +216,6 @@ def get_packet_duration(packet: dict) -> int | None:
elif 'pkt_duration' in packet:
duration = int(packet['pkt_duration'])
else:
logger.error('Impossible to find duration of packet %s', packet)
return None
raise InvalidMediaError("Impossible to find duration of packet {packet}")
return duration