feat: Add progressbar support

`gallia scan uds identifiers` is the first scanner with a progressbar.

pyproject.toml

@@ -37,6 +37,7 @@ tomli = { version = "^2.0.1", python = "<3.11" }
 msgspec = ">=0.8,<0.10"
 pydantic = "^1.10"
 pygit2 = "^1.10"
+tqdm = "^4.64.1"
 
 [tool.poetry.group.dev.dependencies]
 black = "^22.10"
@@ -59,6 +60,7 @@ sphinx-rtd-theme = "^1.0"
 reuse = "^1.0.0"
 construct-typing = "^0.5.2"
 pytest-cov = "^4.0"
+types-tqdm = "^4.64.7.1"
 
 [tool.poetry.scripts]
 "gallia" = "gallia.cli:main"

src/gallia/cli.py

@@ -330,6 +330,7 @@ def cmd_template(args: argparse.Namespace) -> None:
 # dumpcap = <bool>
 # artifacts_dir = <string>
 # artifacts_base = <string>
+# progress = <bool>
 
 [gallia.protocols.uds]
 # dumpcap = <bool>

src/gallia/command.py

@@ -227,6 +227,15 @@ def configure_class_parser(self) -> None:
             default=self.config.get_value("gallia.lock_file", None),
             help="path to file used for a posix lock",
         )
+        group.add_argument(
+            "--progress",
+            action=argparse.BooleanOptionalAction,
+            default=self.config.get_value(
+                "gallia.scanner.progress",
+                default=sys.stderr.isatty(),
+            ),
+            help="Enable/Disable progress bar",
+        )
 
         if self.HAS_ARTIFACTS_DIR:
             mutex_group = group.add_mutually_exclusive_group()
@@ -344,9 +353,10 @@ def entry_point(self, args: Namespace) -> int:
                 self.get_log_level(args),
                 self.get_file_log_level(args),
                 self.artifacts_dir.joinpath(FileNames.LOGFILE.value),
+                progress=args.progress,
             )
         else:
-            setup_logging(self.get_log_level(args))
+            setup_logging(self.get_log_level(args), progress=args.progress)
 
         self.run_hook(HookVariant.PRE, args)
 

src/gallia/commands/scan/uds/identifiers.py

@@ -8,6 +8,8 @@
 from argparse import Namespace
 from itertools import product
 
+from tqdm import tqdm
+
 from gallia.command import UDSScanner
 from gallia.services.uds.core.client import UDSRequestConfig
 from gallia.services.uds.core.constants import RCSubFuncs, UDSErrorCodes, UDSIsoServices
@@ -152,8 +154,19 @@ async def main(self, args: Namespace) -> None:
                     )
                     args.end = 0xFF
 
-            for (DID, sub_function) in product(
-                range(args.start, args.end + 1), sub_functions
+            end_index = args.end + 1
+            n_iterations = (end_index - args.start) * len(sub_functions)
+            for (DID, sub_function) in tqdm(
+                product(
+                    range(args.start, end_index),
+                    sub_functions,
+                ),
+                disable=not args.progress,
+                total=n_iterations,
+                leave=False,
+                desc=f"session {session:#x}",
+                unit="",
+                miniters=100,
             ):
                 if session in args.skip and DID in args.skip[session]:
                     self.logger.info(f"{g_repr(DID)}: skipped")

src/gallia/log.py

@@ -26,6 +26,7 @@
 
 import msgspec
 import zstandard
+from tqdm.contrib import DummyTqdmFile
 
 if TYPE_CHECKING:
     from logging import _ExcInfoType
@@ -250,6 +251,7 @@ def setup_logging(
     file_level: Loglevel = Loglevel.DEBUG,
     path: Path | None = None,
     color_mode: ColorMode = ColorMode.AUTO,
+    progress: bool = False,
 ) -> None:
     """Enable and configure gallia's logging system.
     If this fuction is not called as early as possible,
@@ -265,6 +267,9 @@ def setup_logging(
     :param file_level: The loglevel to enable for the file handler.
     :param path: The path to the logfile containing json records.
     :param color_mode: The color mode to use for the console.
+    :param progress: If a progress bar via ``tqdm`` is used, set
+                     this to ``True`` to setup the console handler
+                     correctly.
     """
     set_color_mode(color_mode)
 
@@ -283,7 +288,8 @@ def setup_logging(
     logging.getLogger("asyncio").setLevel(logging.CRITICAL)
     logging.getLogger("aiosqlite").setLevel(logging.CRITICAL)
 
-    stderr_handler = logging.StreamHandler(sys.stderr)
+    stream = DummyTqdmFile(sys.stderr) if progress else sys.stderr
+    stderr_handler = logging.StreamHandler(stream)  # type: ignore
     stderr_handler.setLevel(level)
     stderr_handler.setFormatter(_ConsoleFormatter())