Add type hints throughout

- consts.py: add FileInfo TypedDict (the shared media file record);
  annotate all Consts class attributes
- proc.py: add ProcResult TypedDict; annotate run() and __init__()
- tag.py: annotate all methods; fix remaining except BaseException and
  bare except: missed in the prior cleanup commit; fix get_int/get_str
  to return None explicitly on decode failure instead of falling through
  with bytes; replace hasattr(value, 'to_bytes') duck-test in set()
  with isinstance(value, int) now that the signature is typed
- probe.py: annotate probe() -> list[dict[str, Any]]
- scan.py: annotate scan(); remove unused imports of Mux and Lang
- mux.py: annotate mux() and internal locals
- lang.py: annotate lang()
- cli.py: annotate main() -> None
- __init__.py: export FileInfo and ProcResult

zz_remux/zz_remux/__init__.py

@@ -1,6 +1,6 @@
-from .consts import Consts
+from .consts import Consts, FileInfo
 from .tag import Tag
 from .scan import Scan
 from .mux import Mux
 from .lang import Lang
-from .proc import Proc
+from .proc import Proc, ProcResult

zz_remux/zz_remux/cli.py

@@ -2,7 +2,7 @@
 import re, sys, os, argparse, pathlib, logging
 import zz_remux as remux
 
-def main():
+def main() -> None:
   # Static config
   log_level = logging.WARN
   prog_name = 'zz_remux'

zz_remux/zz_remux/consts.py

@@ -1,6 +1,19 @@
+import pathlib
+from typing import TypedDict
+
+class FileInfo(TypedDict):
+  path:      pathlib.Path | None
+  name:      str
+  ext:       str
+  group:     str
+  lang:      str | None
+  sdh:       bool
+  mux_vers:  int | None
+  lang_vers: int | None
+
 class Consts:
   # Default file metadata
-  _default = {
+  _default: FileInfo = {
     'path':      None,
     'name':      '',
     'ext':       '',
@@ -12,27 +25,27 @@ class Consts:
   }
 
   # Parsing constants
-  video_ext     = [ r'mkv', r'mp4', r'm4v' ]
-  subs_ext      = [ r'srt' ]
-  subs_sdh      = [ r'sdh' ]
-  subs_lang     = [ r'\w\w' ]
-  media_ext     = video_ext + subs_ext
+  video_ext:  list[str] = [ r'mkv', r'mp4', r'm4v' ]
+  subs_ext:   list[str] = [ r'srt' ]
+  subs_sdh:   list[str] = [ r'sdh' ]
+  subs_lang:  list[str] = [ r'\w\w' ]
+  media_ext:  list[str] = video_ext + subs_ext
 
   # Tag data
-  mux_tag   = 'zib.mux'
-  mux_vers  = 1
-  lang_tag  = 'zib.lang'
-  lang_vers = 2
+  mux_tag:  str = 'zib.mux'
+  mux_vers: int = 1
+  lang_tag:  str = 'zib.lang'
+  lang_vers: int = 2
 
   # Validation size factor
-  min_size = 0.9
+  min_size: float = 0.9
 
   # Stream constants
-  lang_undef   = 'und'
-  lang_default = 'eng'
-  lang_codecs  = [ 'subtitle', 'audio' ]
-  lang_missing = [ None, lang_undef ]
+  lang_undef:   str            = 'und'
+  lang_default: str            = 'eng'
+  lang_codecs:  list[str]      = [ 'subtitle', 'audio' ]
+  lang_missing: list[str|None] = [ None, lang_undef ]
 
   @staticmethod
-  def default():
+  def default() -> FileInfo:
     return Consts._default.copy()

zz_remux/zz_remux/lang.py

@@ -1,5 +1,5 @@
-import logging
-from .consts import Consts
+import pathlib, logging
+from .consts import Consts, FileInfo
 from .tag import Tag
 from .probe import Probe
 from .proc import Proc
@@ -7,18 +7,18 @@ from .proc import Proc
 class Lang:
   sdh_title = 'SDH'
 
-  def __init__(self, propeditExec='mkvpropedit', logLevel=logging.WARN):
+  def __init__(self, propeditExec: str = 'mkvpropedit', logLevel: int = logging.WARN) -> None:
     # Logging
     self.logger = logging.getLogger(__name__)
     self.logger.setLevel(logLevel)
-    
+
     # Module init
     self.propedit_exec = propeditExec
     self.tags = Tag(logLevel=logLevel)
     self.probes = Probe(logLevel=logLevel)
     self.proc = Proc(logLevel=logLevel)
 
-  def lang(self, media):
+  def lang(self, media: list[FileInfo]) -> None:
     self.logger.info(f"Processing {len(media)} files")
 
     for file in media:
@@ -41,8 +41,8 @@ class Lang:
 
       # Find missing and undefined languages for subtitle and audio streams
       # Find hearing_impaired subtitles without SDH title
-      add_lang = []
-      add_sdh  = []
+      add_lang: list[int] = []
+      add_sdh:  list[int] = []
       for stream in streams:
         if stream.get('codec_type') not in Consts.lang_codecs:
           self.logger.debug(f"Skipping missing or ignored codec_type: {file['name']}:{stream['index']}")
@@ -62,7 +62,7 @@ class Lang:
       # mkvpropedit
       if add_lang or add_sdh:
         # Remember mkvpropedit starts counting at 1
-        args = [ self.propedit_exec, file['path'] ]
+        args: list[str | pathlib.Path] = [ self.propedit_exec, file['path'] ]
         for index in add_lang:
           self.logger.info(f"Setting lang ({Consts.lang_default}): {file['name']}:{index}")
           args.extend([ '--edit', f"track:{index + 1}", '--set', f"language={Consts.lang_default}" ])

zz_remux/zz_remux/mux.py

@@ -1,10 +1,10 @@
 import pathlib, logging
-from .consts import Consts
+from .consts import Consts, FileInfo
 from .tag import Tag
 from .proc import Proc
 
 class Mux:
-  def __init__(self, logLevel=logging.WARN):
+  def __init__(self, logLevel: int = logging.WARN) -> None:
     # Logging
     self.logger = logging.getLogger(__name__)
     self.logger.setLevel(logLevel)
@@ -13,11 +13,11 @@ class Mux:
     self.tags = Tag(logLevel=logLevel)
     self.proc = Proc(logLevel=logLevel)
 
-  def mux(self, media):
+  def mux(self, media: list[FileInfo]) -> None:
     self.logger.info(f"Processing {len(media)} files")
 
     # Find groups
-    groups  = {}
+    groups: dict[str, list[FileInfo]] = {}
     for file in media:
       if not file['group']:
         self.logger.debug(f"Skipping non-grouped file: {file['name']}")
@@ -39,9 +39,9 @@ class Mux:
           continue
 
       # Find all sources for the group
-      size    = 0
-      subs    = []
-      videos  = []
+      size   = 0
+      subs:   list[FileInfo] = []
+      videos: list[FileInfo] = []
       for file in group:
         if file['ext'] in Consts.video_ext:
           self.logger.info(f"Found video: {file['name']}")
@@ -60,9 +60,9 @@ class Mux:
         continue
 
       # Build args
-      final   = videos[0]['path'].parent.joinpath(f'{videos[0]['group']}.mkv')
-      temp    = final.with_suffix('.mkv-remux')
-      args    = [ 'mkvmerge', '--default-language', Consts.lang_default, '-o', temp ]
+      final = videos[0]['path'].parent.joinpath(f'{videos[0]['group']}.mkv')
+      temp  = final.with_suffix('.mkv-remux')
+      args: list[str | pathlib.Path] = [ 'mkvmerge', '--default-language', Consts.lang_default, '-o', temp ]
       args.append(videos[0]['path'])
 
       # Subtitles get language and SDH metadata

zz_remux/zz_remux/probe.py

@@ -1,10 +1,11 @@
-import json, logging
+import json, logging, pathlib
+from typing import Any
 from .proc import Proc
 
 class Probe:
   ffprobe_exec = 'ffprobe'
-  
-  def __init__(self, ffprobe_exec='ffprobe', logLevel=logging.WARN):
+
+  def __init__(self, ffprobe_exec: str = 'ffprobe', logLevel: int = logging.WARN) -> None:
     self.logger = logging.getLogger(__name__)
     self.logger.setLevel(logLevel)
 
@@ -12,7 +13,7 @@ class Probe:
     self.ffprobe_exec = ffprobe_exec
     self.proc = Proc(logLevel=logLevel)
 
-  def probe(self, path):
+  def probe(self, path: pathlib.Path) -> list[dict[str, Any]]:
     self.logger.debug(f"Probing: {path}")
 
     # Sanity

zz_remux/zz_remux/proc.py

@@ -1,18 +1,24 @@
-import subprocess, logging
+import subprocess, logging, pathlib
+from typing import TypedDict
+
+class ProcResult(TypedDict):
+  out:  str | None
+  err:  str | None
+  code: int
 
 class Proc:
   encoding = 'utf-8'
 
-  def __init__(self, stdout=subprocess.PIPE, stderr=subprocess.PIPE, logLevel=logging.WARN):
+  def __init__(self, stdout=subprocess.PIPE, stderr=subprocess.PIPE, logLevel: int = logging.WARN) -> None:
     self.logger = logging.getLogger(__name__)
     self.logger.setLevel(logLevel)
 
     self.stdout = stdout
     self.stderr = stderr
 
-  def run(self, cmd):
+  def run(self, cmd: list[str | pathlib.Path]) -> ProcResult:
     # Return dict
-    ret = {'out': None, 'err': None, 'code': int(-111) }
+    ret: ProcResult = {'out': None, 'err': None, 'code': int(-111) }
 
     # Sanity
     if not cmd or len(cmd) < 1:

zz_remux/zz_remux/scan.py

@@ -1,12 +1,10 @@
 import re, pathlib, logging
 import langcodes
-from .consts import Consts
+from .consts import Consts, FileInfo
 from .tag import Tag
-from .mux import Mux
-from .lang import Lang
 
-class Scan:  
-  def __init__(self, logLevel=logging.WARN):
+class Scan:
+  def __init__(self, logLevel: int = logging.WARN) -> None:
     # Logging
     self.logger = logging.getLogger(__name__)
     self.logger.setLevel(logLevel)
@@ -24,14 +22,14 @@ class Scan:
       re.IGNORECASE
     )
 
-  def scan(self, files, parent=pathlib.Path(''), match=None):
+  def scan(self, files: list[str], parent: pathlib.Path = pathlib.Path(''), match: re.Pattern[str] | None = None) -> list[FileInfo]:
     self.logger.info(f"Processing {len(files)} files")
-    media = []
+    media: list[FileInfo] = []
 
     # Scan this directory for media
     for file in files:
       self.logger.info(f"Found file: {file}")
-      
+
       info = Consts.default()
       info['path'] = parent.joinpath(file)
       info['name'] = file

zz_remux/zz_remux/tag.py

@@ -5,70 +5,70 @@ class Tag:
   # Internals
   _encoding = 'utf-8'
 
-  def __init__(self, logLevel=logging.WARN):
+  def __init__(self, logLevel: int = logging.WARN) -> None:
     self.logger = logging.getLogger(__name__)
     self.logger.setLevel(logLevel)
 
-  def get_int(self, path, attr):
+  def get_int(self, path: pathlib.Path, attr: str) -> int | None:
     ret = self.get_bytes(path, attr)
     if ret:
       try:
-        ret = int.from_bytes(ret)
+        return int.from_bytes(ret)
       except Exception:
         self.logger.error(f"Unable to decode int ({path}): {attr} => {ret}")
-    return ret
+    return None
 
-  def get_str(self, path, attr):
+  def get_str(self, path: pathlib.Path, attr: str) -> str | None:
     ret = self.get_bytes(path, attr)
     if ret:
       try:
-        ret = ret.decode(self._encoding)
+        return ret.decode(self._encoding)
       except Exception:
         self.logger.error(f"Unable to decode str ({path}): {attr} => {ret}")
-    return ret
+    return None
 
-  def get(self, path, attr):
+  def get(self, path: pathlib.Path, attr: str) -> str | None:
     return self.get_str(path, attr)
 
-  def get_bytes(self, path, attr):
+  def get_bytes(self, path: pathlib.Path, attr: str) -> bytes | None:
     value = None
     try:
       if attr in xattr.listxattr(path):
         value = xattr.getxattr(path, attr)
         self.logger.debug(f"Got tag ({path}): {attr} => {value}")
-    except BaseException as e:
+    except Exception as e:
       self.logger.error(f"Get error ({path}:{attr}): {e}")
     return value
 
-  def set_bytes(self, path, attr, value):
+  def set_bytes(self, path: pathlib.Path, attr: str, value: bytes) -> bool:
     try:
       xattr.setxattr(path, attr, value)
       self.logger.debug(f"Wrote tag ({path}): {attr} => {value}")
       return True
-    except BaseException as e:
+    except Exception as e:
       self.logger.error(f"Set error ({path}:{attr}): {e}")
       return False
 
-  def set_int(self, path, attr, value):
+  def set_int(self, path: pathlib.Path, attr: str, value: int) -> bool:
     try:
       bval = value.to_bytes()
       return self.set_bytes(path, attr, bval)
-    except:
+    except Exception:
       self.logger.error(f"Unable to encode int: {value}")
       return False
 
-  def set_str(self, path, attr, value):
+  def set_str(self, path: pathlib.Path, attr: str, value: str) -> bool:
     try:
       bval = value.encode(self._encoding)
       return self.set_bytes(path, attr, bval)
-    except:
+    except Exception:
       self.logger.error(f"Unable to encode str: {value}")
       return False
 
-  def set(self, path, attr, value):
+  def set(self, path: pathlib.Path, attr: str, value: bytes | int | str) -> bool:
     if isinstance(value, bytes):
       return self.set_bytes(path, attr, value)
-    elif hasattr(value, 'to_bytes'):
+    elif isinstance(value, int):
       return self.set_int(path, attr, value)
     elif isinstance(value, str):
       return self.set_str(path, attr, value)