From 093ad733d3505e3eaba2ec9c2aa517f84421ce61 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Fr=C3=A9d=C3=A9ric=20Tronel?= Date: Sun, 30 Aug 2026 14:14:18 +0200 Subject: [PATCH] Even more typing errors fixes. --- src/tscut/h264/bitstream.py | 22 +++++++++++----------- src/tscut/pipeline.py | 5 +++-- src/tscut/subtitles/ocr.py | 18 +++++++++++------- 3 files changed, 25 insertions(+), 20 deletions(-) diff --git a/src/tscut/h264/bitstream.py b/src/tscut/h264/bitstream.py index adca126..8824add 100644 --- a/src/tscut/h264/bitstream.py +++ b/src/tscut/h264/bitstream.py @@ -18,7 +18,7 @@ from tscut.exceptions import UnimplementedFeatureError logger = logging.getLogger(__name__) @typechecked -def read_bit(buf:bytes, bit_position: int) -> tuple[int, int]: +def read_bit(buf:bytes|bytearray, bit_position: int) -> tuple[int, int]: """ Read a single bit from a byte buffer. @@ -41,7 +41,7 @@ def read_bit(buf:bytes, bit_position: int) -> tuple[int, int]: return bit_position+1, bit @typechecked -def read_boolean(buf:bytes, bit_position: int) -> tuple[int, bool]: +def read_boolean(buf:bytes|bytearray, bit_position: int) -> tuple[int, bool]: """ Read a boolean value from a byte buffer. @@ -61,7 +61,7 @@ def read_boolean(buf:bytes, bit_position: int) -> tuple[int, bool]: return bit_position, b==1 @typechecked -def read_bits(buf:bytes, bit_position: int, nb_bits: int) -> tuple[int, int]: +def read_bits(buf:bytes|bytearray, bit_position: int, nb_bits: int) -> tuple[int, int]: v = 0 for _ in range(nb_bits): bit_position, bit = read_bit(buf, bit_position) @@ -69,22 +69,22 @@ def read_bits(buf:bytes, bit_position: int, nb_bits: int) -> tuple[int, int]: return bit_position, v @typechecked -def read_byte(buf:bytes, bit_position: int) -> tuple[int, int]: +def read_byte(buf:bytes|bytearray, bit_position: int) -> tuple[int, int]: bit_position, b = read_bits(buf, bit_position, 8) return bit_position, b @typechecked -def read_word(buf:bytes, bit_position: int) -> tuple[int, int]: +def read_word(buf:bytes|bytearray, bit_position: int) -> tuple[int, int]: bit_position, w = read_bits(buf, bit_position, 16) return bit_position, w @typechecked -def read_long(buf:bytes, bit_position: int) -> tuple[int, int]: +def read_long(buf:bytes|bytearray, bit_position: int) -> tuple[int, int]: bit_position, value = read_bits(buf, bit_position, 32) return bit_position, value @typechecked -def read_unsigned_exp_golomb(buf:bytes, bit_position: int) -> tuple[int, int]: +def read_unsigned_exp_golomb(buf:bytes|bytearray, bit_position: int) -> tuple[int, int]: nb_zeroes=0 while True: bit_position, b = read_bit(buf, bit_position) @@ -98,7 +98,7 @@ def read_unsigned_exp_golomb(buf:bytes, bit_position: int) -> tuple[int, int]: return bit_position, v-1 @typechecked -def read_signed_exp_golomb(buf:bytes, bit_position: int) -> tuple[int, int]: +def read_signed_exp_golomb(buf:bytes|bytearray, bit_position: int) -> tuple[int, int]: bit_position, v = read_unsigned_exp_golomb(buf, bit_position) match v%2: case 0: @@ -173,7 +173,7 @@ def write_signed_exp_golomb(buf:bytearray, bit_position: int, v) -> int: return bit_position @typechecked -def parse_rbsp_trailing_bits(buf:bytes, bit_position: int) -> int: +def parse_rbsp_trailing_bits(buf:bytes|bytearray, bit_position: int) -> int: bit_position, one = read_bit(buf, bit_position) if one==0: raise ValueError(f'Stop bit should be equal to one. Read: {one:d}') @@ -193,7 +193,7 @@ def write_rbsp_trailing_bits(buf:bytearray, bit_position: int) -> int: return bit_position @typechecked -def more_rbsp_data(buf:bytes, bit_position: int) -> bool: +def more_rbsp_data(buf:bytes|bytearray, bit_position: int) -> bool: logger.debug('Is there more data in buffer of length: %d at bit position: %d', len(buf), bit_position) @@ -250,7 +250,7 @@ def sodb_to_rbsp(buf:bytes) -> bytes: # Useful for SPS and PPS @typechecked -def parse_scaling_list(buf:bytes, bit_position: int, size) -> tuple[int,list[int]]: +def parse_scaling_list(buf:bytes|bytearray, bit_position: int, size) -> tuple[int,list[int]]: res = [] last_scale = 8 next_scale = 8 diff --git a/src/tscut/pipeline.py b/src/tscut/pipeline.py index 51c96a2..8faddf2 100644 --- a/src/tscut/pipeline.py +++ b/src/tscut/pipeline.py @@ -8,6 +8,7 @@ from os import unlink from shutil import copyfile, move from sys import exit import logging +from typing import IO, Any import hexdump @@ -55,7 +56,7 @@ logger = logging.getLogger(__name__) def process_recording(options: ProcessingOptions) -> None: nb_parts = len(options.parts) - temporaries = [] + temporaries : list[IO[Any]] = [] basename = os.path.splitext(os.path.basename(options.input_file))[0] mp4filename = basename+'.mp4' @@ -460,7 +461,7 @@ def process_recording(options: ProcessingOptions) -> None: logger.info('Find subtitles tracks and language.') subtitles = find_subtitles_tracks(options.tools_paths['ffprobe'], final_with_video) logger.info(subtitles) - sts = {} + sts: dict[str, list[int]] = {} for subtitle in subtitles: index = subtitle['index'] if 'tags' in subtitle: diff --git a/src/tscut/subtitles/ocr.py b/src/tscut/subtitles/ocr.py index c5abbaf..7f3daed 100644 --- a/src/tscut/subtitles/ocr.py +++ b/src/tscut/subtitles/ocr.py @@ -18,6 +18,7 @@ from os import ( import re import os from subprocess import PIPE, Popen +from typing import IO, Any from iso639 import Lang from iso639.exceptions import InvalidLanguageValue @@ -29,7 +30,7 @@ from tscut.exceptions import ExternalToolError logger = logging.getLogger(__name__) @typechecked -def get_tesseract_supported_lang(tesseract_path:str) -> dict[Lang, str]|None: +def get_tesseract_supported_lang(tesseract_path:str) -> dict[Lang, str]: """ Retrieves the set of natural languages supported by the Tesseract OCR tool. @@ -48,6 +49,7 @@ def get_tesseract_supported_lang(tesseract_path:str) -> dict[Lang, str]|None: res = {} with Popen([tesseract_path, '--list-langs'], stdout=PIPE) as tesseract: + assert tesseract.stdout is not None for line in tesseract.stdout: line = line.decode('utf8') p = re.compile('(?P[a-z]{3})\n') @@ -63,8 +65,7 @@ def get_tesseract_supported_lang(tesseract_path:str) -> dict[Lang, str]|None: tesseract.wait() if tesseract.returncode != 0: - logger.error("Tesseract returns an error code: %d",tesseract.returncode) - return None + raise ExternalToolError("Tesseract returns an error code: %d",tesseract.returncode) return res @@ -100,6 +101,7 @@ def extract_srt(mkvextract:str, filename:str, subtitles:dict[str, list[int]], env = {**os.environ, 'LANG': 'C'} 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:') for line in pb: @@ -108,7 +110,8 @@ def extract_srt(mkvextract:str, filename:str, subtitles:dict[str, list[int]], 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() @@ -128,7 +131,7 @@ def extract_srt(mkvextract:str, filename:str, subtitles:dict[str, list[int]], @typechecked def do_ocr(vobsubocr:str, idxs: Sequence[tuple[str,str,str,str]], duration:timedelta, - temporaries:list[IO[bytes]], dump_mem_fd:bool=False): + temporaries:list[IO[Any]], dump_mem_fd:bool=False): res = [] for idx_name, _, lang, iso in idxs: @@ -141,6 +144,7 @@ def do_ocr(vobsubocr:str, idxs: Sequence[tuple[str,str,str,str]], duration:timed srtfd = memfd_create(srtname, flags=0) with Popen([vobsubocr, '--lang', iso, idx_name], stdout=PIPE) as ocr: + assert ocr.stdout is not None pb = tqdm(TextIOWrapper(ocr.stdout, encoding="utf-8"), total= int(duration/timedelta(seconds=1)), unit='s', desc='OCR') for line in pb: @@ -153,7 +157,7 @@ def do_ocr(vobsubocr:str, idxs: Sequence[tuple[str,str,str,str]], duration:timed m = re.match(timestamps, line) if m is not None: hours = int(m.group('hours')) - minutes = int(m.group('hours')) + minutes = int(m.group('minutes')) seconds = int(m.group('seconds')) ts = timedelta(hours=hours, minutes=minutes, seconds=seconds) pb.n = int(ts/timedelta(seconds=1)) @@ -174,7 +178,7 @@ def do_ocr(vobsubocr:str, idxs: Sequence[tuple[str,str,str,str]], duration:timed pos = 0 while pos < srt_length: pos+=write(outfd, buf[pos:]) - temporaries.append(dump_srt) + temporaries.append(dump_srt) except OSError: logger.error('Impossible to create file: %s', srtname) return None