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

Fix of several typing errors.

Browse Source
This commit is contained in:
Frédéric Tronel
2026-08-30 13:56:41 +02:00
parent 447d04d13d
commit a6a41c112e
8 changed files with 54 additions and 43 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/tscut/cli.py
+1
View File
@@ -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
src/tscut/h264/avc.py
+1 -1
View File
@@ -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:
src/tscut/pipeline.py
+11 -12
View File
@@ -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)
src/tscut/subtitles/ocr.py
+6 -5
View File
@@ -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 = []
src/tscut/tools/ffmpeg.py
+1
View File
@@ -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:
src/tscut/tools/ffprobe.py
+8 -10
View File
@@ -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)
src/tscut/tools/mkvtoolnix.py
+23 -12
View File
@@ -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)
src/tscut/tscut.py
+3 -3
View File
@@ -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)