Even more typing errors fixes.

2026-08-30 14:14:18 +02:00
parent a6a41c112e
commit 093ad733d3
3 changed files with 25 additions and 20 deletions
@@ -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
@@ -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:
@@ -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<lang>[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