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

@@ -15,7 +15,7 @@
 from gallia.services.uds.core.service import NegativeResponse, UDSResponse
 from gallia.services.uds.core.utils import service_repr
 from gallia.services.uds.helpers import suggests_service_not_supported
-from gallia.utils import ParseSkips, auto_int, g_repr
+from gallia.utils import ParseSkips, auto_int, g_repr, wrap_progress
 
 
 class ScanIdentifiers(UDSScanner):
@@ -152,8 +152,16 @@ 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 wrap_progress(
+                product(
+                    range(args.start, end_index),
+                    sub_functions,
+                ),
+                enabled=args.progress,
+                total=n_iterations,
+                description=f"session {session:#x}",
             ):
                 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())
 

src/gallia/utils.py

@@ -14,10 +14,11 @@
 from enum import Enum
 from pathlib import Path
 from types import ModuleType
-from typing import TYPE_CHECKING, Any, TypeVar
+from typing import TYPE_CHECKING, Any, Iterable, TypeVar
 from urllib.parse import urlparse
 
 import aiofiles
+from tqdm import tqdm
 
 from gallia.services.uds.core.service import NegativeResponse
 from gallia.services.uds.core.utils import bytes_repr, int_repr
@@ -239,3 +240,25 @@ def lazy_import(name: str) -> ModuleType:
     sys.modules[name] = module
     loader.exec_module(module)
     return module
+
+
+_T = TypeVar("_T")
+
+
+def wrap_progress(
+    generator: Iterable[_T],
+    description: str = "",
+    unit: str = "",
+    total: int | None = None,
+    enabled: bool = True,
+) -> Iterable[_T]:
+    return tqdm(
+        generator,
+        disable=not enabled,
+        total=total,
+        leave=False,
+        desc=description,
+        unit=unit,
+        miniters=100,
+        dynamic_ncols=True,
+    )