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 14:30:38 +02:00
parent 51bab76c00
commit e1d7474e00
7 changed files with 29 additions and 47 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/tscut/cutting.py
+6 -5
View File
@@ -144,12 +144,13 @@ def cut_recording(media: PreparedMedia, options: ProcessingOptions,
nb_head_frames)
logger.info("Found %d frames between last I-frame and end of current part",
nb_tail_frames)
head_iframe_ts = get_ts_frame(head_iframe)
if head_iframe_ts is None:
try:
head_iframe_ts = get_ts_frame(head_iframe)
except InvalidMediaError:
raise InvalidMediaError("Impossible to retrieve timestamp of first i-frame.")
tail_iframe_ts = get_ts_frame(tail_iframe)
if tail_iframe_ts is None:
try:
tail_iframe_ts = get_ts_frame(tail_iframe)
except InvalidMediaError:
raise InvalidMediaError("Impossible to retrieve timestamp of final i-frame.")
checks.append(pos+head_iframe_ts-ts1)
src/tscut/pipeline.py
-3
View File
@@ -148,9 +148,6 @@ def process_recording(options: ProcessingOptions, temporaries: TemporaryFiles) -
logger.debug('Main video stream has following private data: %s',
hexdump.dump(main_codec_private_data, sep=':'))
if main_codec_private_data is None:
raise InvalidMediaError(f"Impossible to retrieve private data from MKV file {mkv}")
# We parse them
main_avc_config = parse_codec_private(main_codec_private_data)
logger.debug('AVC configuration: %s', main_avc_config)
src/tscut/tools/ffmpeg.py
+1 -3
View File
@@ -14,6 +14,7 @@ from tqdm import tqdm
from typeguard import typechecked
from tscut.exceptions import ExternalToolError, InvalidMediaError, TemporaryFileError
from tscut.temporaries import TemporaryFiles
from tscut.tools.ffprobe import (
get_frames_in_stream,
get_video_dimensions,
@@ -21,7 +22,6 @@ 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__)
@@ -263,8 +263,6 @@ def extract_all_streams(ffmpeg_path:str, ffprobe_path:str, input_file:IO[bytes],
logger.debug("Found %d packets to be extracted from audio track.", nb_packets)
if nb_packets > 0:
packet_duration = get_packet_duration(packets[0])
if packet_duration is None:
return None
else:
packet_duration = 0
src/tscut/tools/ffprobe.py
+11 -17
View File
@@ -18,7 +18,7 @@ from typing import IO
from typeguard import typechecked
from tscut.exceptions import InvalidMediaError
from tscut.exceptions import InvalidMediaError, ExternalToolError
from tscut.tools.timeframe import get_ts_frame
logger = logging.getLogger(__name__)
@@ -170,8 +170,7 @@ def get_movie_duration(ffprobe_path:str, input_file: IO[bytes]) -> timedelta:
# ffprobe -loglevel quiet -select_streams v:0 -show_entries stream=width,height -of json sample.ts
@typechecked
def get_video_dimensions(ffprobe_path:str,
input_file: IO[bytes]) -> tuple[int,int]| tuple[None,None]:
def get_video_dimensions(ffprobe_path:str, input_file: IO[bytes]) -> tuple[int,int]:
infd = input_file.fileno()
lseek(infd, 0, SEEK_SET)
set_inheritable(infd, True)
@@ -185,8 +184,7 @@ def get_video_dimensions(ffprobe_path:str,
if ('width' in video) and ('height' in video):
return int(video['width']), int(video['height'])
logger.error('Impossible to retrieve dimensions of video')
return None, None
raise InvalidMediaError('Impossible to retrieve dimensions of video')
@typechecked
def get_streams(ffprobe_path:str, input_file: IO[bytes]) -> list|None:
@@ -276,8 +274,7 @@ def get_frames_in_stream(ffprobe_path: str, input_file: IO[bytes], begin:timedel
frames = json.load(BytesIO(out))
status = ffprobe.wait()
if status != 0:
logger.error('ffprobe failed with status code: %d', status)
return None
raise ExternalToolError(f"ffprobe failed with status code: {status:d}")
# Sort frames by timestamp
tmp = {}
@@ -285,8 +282,6 @@ def get_frames_in_stream(ffprobe_path: str, input_file: IO[bytes], begin:timedel
frames = frames['frames']
for frame in frames:
ts = get_ts_frame(frame)
if ts is None:
return None
if begin <= ts <= end:
tmp[ts]=frame
@@ -322,15 +317,12 @@ def get_nearest_idr_frame(ffprobe_path: str, input_file: IO[bytes], timestamp:ti
frames = json.load(BytesIO(out))
status = ffprobe.wait()
if status != 0:
logger.error('ffprobe failed with status code: %d', status)
return
raise ExternalToolError(f"ffprobe failed with status code: {status:d}")
if 'frames' in frames:
frames = frames['frames']
for frame in frames:
ts = get_ts_frame(frame)
if ts is None:
return
if tbegin <= ts <= tend:
idrs.append(frame)
else:
@@ -371,8 +363,9 @@ def get_nearest_iframe(ffprobe_path:str, input_file: IO[bytes],
found = False
for frame in iframes:
ts = get_ts_frame(frame)
if ts is None:
try:
ts = get_ts_frame(frame)
except InvalidMediaError:
logger.warning('I-frame with no timestamp: %s', frame)
continue
@@ -397,8 +390,9 @@ def get_nearest_iframe(ffprobe_path:str, input_file: IO[bytes],
return None
nb_frames = 0
for frame in frames:
ts = get_ts_frame(frame)
if ts is None:
try:
ts = get_ts_frame(frame)
except InvalidMediaError:
logger.warning('Frame without timestamp: %s', frame)
continue
src/tscut/tools/mkvtoolnix.py
+6 -7
View File
@@ -17,14 +17,14 @@ import hexdump
from tqdm import tqdm
from typeguard import typechecked
from tscut.exceptions import InvalidMediaError
from tscut.exceptions import InvalidMediaError, ExternalToolError
from tscut.matroska.ebml import change_ebml_element_size
# Found codec private data using mkvinfo
@typechecked
def get_codec_private_data_from_mkv(mkvinfo_path:str,
input_file: IO[bytes]) -> tuple[int, bytes]|tuple[None,None]:
input_file: IO[bytes]) -> tuple[int, bytes]:
logger = logging.getLogger(__name__)
infd = input_file.fileno()
@@ -59,8 +59,7 @@ def get_codec_private_data_from_mkv(mkvinfo_path:str,
data = read(infd, size)
return position, data
logger.error("Impossible to retrieve codec private data from MKV !")
return None, None
raise InvalidMediaError("Impossible to retrieve codec private data from MKV !")
@typechecked
def parse_mkv_tree(mkvinfo_path:str, input_file: IO[bytes]) -> dict[str,tuple[int,int]]:
@@ -233,7 +232,7 @@ def extract_mkv_part(mkvmerge_path:str, input_file:IO[bytes], output_file:IO[byt
for w in warnings:
logger.warning(w)
elif status == 2:
logger.error('Extraction returns errors')
raise ExternalToolError("Extraction returns errors")
# Merge a list of mkv files passed as input, and produce a new MKV as output
@@ -367,7 +366,7 @@ def extract_track_from_mkv(mkvextract_path: str, input_file: IO[bytes], index,
extract.wait()
if extract.returncode != 0:
logger.error('Mkvextract returns an error code: %d', extract.returncode)
raise ExternalToolError(f"Mkvextract returns an error code: {extract.returncode:d}")
else:
logger.info('Track %d was succesfully extracted.', index)
@@ -408,7 +407,7 @@ def remove_video_tracks_from_mkv(mkvmerge_path:str, input_file: IO[bytes],
remove.wait()
if remove.returncode != 0:
logger.error('Mkvmerge returns an error code: %d', remove.returncode)
raise ExternalToolError(f"Mkvmerge returns an error code: {remove.returncode:d}")
else:
logger.info('Video tracks were succesfully extracted.')
src/tscut/tools/ppm.py
+2 -1
View File
@@ -12,8 +12,9 @@ from typing import IO
from typeguard import typechecked
from tscut.temporaries import TemporaryFiles
from tscut.exceptions import TemporaryFileError
from tscut.temporaries import TemporaryFiles
@typechecked
def dump_ppm(pictures: bytes, prefix: str, temporaries: TemporaryFiles) -> None:
src/tscut/tools/timeframe.py
+3 -11
View File
@@ -11,6 +11,7 @@ from typeguard import typechecked
from tscut.exceptions import InvalidMediaError
logger = logging.getLogger(__name__)
@typechecked
def parse_timestamp(ts:str) -> timedelta|None:
@@ -31,8 +32,6 @@ def parse_timestamp(ts:str) -> timedelta|None:
- the timestamp string is not in the correct format
- the timestamp values are out of range (e.g. hour > 23, minute > 59, etc.)
"""
logger = logging.getLogger(__name__)
ts_reg_exp = (r'^(?P<hour>[0-9]{1,2}):(?P<minute>[0-9]{1,2})'
r':(?P<second>[0-9]{1,2})(\.(?P<us>[0-9]{1,6}))?$')
p = re.compile(ts_reg_exp)
@@ -91,8 +90,6 @@ def parse_time_interval(interval: str) -> tuple[timedelta, timedelta] | tuple[No
- the time values are out of range (e.g. hour > 23, minute > 59, etc.)
- the end time is before the start time (non-monotonic interval)
"""
logger = logging.getLogger(__name__)
interval_reg_exp = (r'^(?P<hour1>[0-9]{1,2}):(?P<minute1>[0-9]{1,2}):(?P<second1>[0-9]{1,2})'
r'(\.(?P<ms1>[0-9]{1,3}))?-(?P<hour2>[0-9]{1,2}):(?P<minute2>[0-9]{1,2})'
r':(?P<second2>[0-9]{1,2})(\.(?P<ms2>[0-9]{1,3}))?$')
@@ -194,23 +191,18 @@ def compare_time_interval(interval1: tuple[timedelta, timedelta],
@typechecked
def get_ts_frame(frame: dict) -> timedelta|None:
logger = logging.getLogger(__name__)
def get_ts_frame(frame: dict) -> timedelta:
if 'pts_time' in frame:
pts_time = float(frame['pts_time'])
elif 'pkt_pts_time' in frame:
pts_time = float(frame['pkt_pts_time'])
else:
logger.error('Impossible to find timestamp of frame %s', frame)
return None
raise InvalidMediaError(f"Impossible to find timestamp of frame {frame}")
return timedelta(seconds=pts_time)
@typechecked
def get_packet_duration(packet: dict) -> int:
logger = logging.getLogger(__name__)
if 'duration' in packet:
duration = int(packet['duration'])
elif 'pkt_duration' in packet: