improve: enable logging for multiprocessing (#774)

ptpt · web-flow · commit aff7bc17ffbc · 2025-08-20T22:36:22.000-07:00
* use LOG.isEnabledFor(DEBUG)

* configure logger in multiprocessing

* fix comments

* fix types

* use ProcessPoolExecutor

* imports

* fix mp_map
diff --git a/mapillary_tools/commands/__main__.py b/mapillary_tools/commands/__main__.py
@@ -8,6 +8,7 @@
 
 from .. import api_v4, constants, exceptions, VERSION
 from ..upload import log_exception
+from ..utils import configure_logger, get_app_name
 from . import (
     authenticate,
     process,
@@ -31,8 +32,8 @@
 ]
 
 
-# do not use __name__ here is because if you run tools as a module, __name__ will be "__main__"
-LOG = logging.getLogger("mapillary_tools")
+# Root logger of mapillary_tools (not including third-party libraries)
+LOG = logging.getLogger(get_app_name())
 
 
 # Handle shared arguments/options here
@@ -79,13 +80,6 @@ def add_general_arguments(parser, command):
         )
 
 
-def configure_logger(logger: logging.Logger, stream=None) -> None:
-    formatter = logging.Formatter("%(asctime)s - %(levelname)-7s - %(message)s")
-    handler = logging.StreamHandler(stream)
-    handler.setFormatter(formatter)
-    logger.addHandler(handler)
-
-
 def _log_params(argvars: dict) -> None:
     MAX_ENTRIES = 5
 
@@ -152,9 +146,7 @@ def main():
 
     args = parser.parse_args()
 
-    log_level = logging.DEBUG if args.verbose else logging.INFO
-    configure_logger(LOG, sys.stderr)
-    LOG.setLevel(log_level)
+    configure_logger(LOG, level=logging.DEBUG if args.verbose else logging.INFO)
 
     LOG.debug("%s", version_text)
     argvars = vars(args)
diff --git a/mapillary_tools/exif_read.py b/mapillary_tools/exif_read.py
@@ -871,7 +871,8 @@ def __init__(self, path_or_stream: Path | T.BinaryIO) -> None:
 
     def _xmp_with_reason(self, reason: str) -> ExifReadFromXMP | None:
         if not self._xml_extracted:
-            LOG.debug('Extracting XMP for "%s"', reason)
+            # TODO Disabled because too verbose but still useful to know
+            # LOG.debug('Extracting XMP for "%s"', reason)
             self._cached_xml = self._extract_xmp()
             self._xml_extracted = True
 
diff --git a/mapillary_tools/geotag/base.py b/mapillary_tools/geotag/base.py
@@ -46,7 +46,7 @@ def to_description(
                 map_results,
                 desc="Extracting images",
                 unit="images",
-                disable=LOG.getEffectiveLevel() <= logging.DEBUG,
+                disable=LOG.isEnabledFor(logging.DEBUG),
                 total=len(extractors),
             )
         )
@@ -62,6 +62,8 @@ def run_extraction(cls, extractor: TImageExtractor) -> types.ImageMetadataOrErro
         try:
             return extractor.extract()
         except exceptions.MapillaryDescriptionError as ex:
+            if LOG.isEnabledFor(logging.DEBUG):
+                LOG.error(f"{cls.__name__}({image_path.name}): {ex}")
             return types.describe_error_metadata(
                 ex, image_path, filetype=types.FileType.IMAGE
             )
@@ -112,7 +114,7 @@ def to_description(
                 map_results,
                 desc="Extracting videos",
                 unit="videos",
-                disable=LOG.getEffectiveLevel() <= logging.DEBUG,
+                disable=LOG.isEnabledFor(logging.DEBUG),
                 total=len(extractors),
             )
         )
@@ -128,6 +130,8 @@ def run_extraction(cls, extractor: TVideoExtractor) -> types.VideoMetadataOrErro
         try:
             return extractor.extract()
         except exceptions.MapillaryDescriptionError as ex:
+            if LOG.isEnabledFor(logging.DEBUG):
+                LOG.error(f"{cls.__name__}({video_path.name}): {ex}")
             return types.describe_error_metadata(
                 ex, video_path, filetype=types.FileType.VIDEO
             )
diff --git a/mapillary_tools/geotag/factory.py b/mapillary_tools/geotag/factory.py
@@ -67,7 +67,7 @@ def process(
     reprocessable_paths = set(paths)
 
     for idx, option in enumerate(options):
-        if LOG.getEffectiveLevel() <= logging.DEBUG:
+        if LOG.isEnabledFor(logging.DEBUG):
             LOG.info(
                 f"==> Processing {len(reprocessable_paths)} files with source {option}..."
             )
diff --git a/mapillary_tools/geotag/geotag_images_from_exiftool.py b/mapillary_tools/geotag/geotag_images_from_exiftool.py
@@ -93,7 +93,7 @@ def _generate_image_extractors(
             LOG.warning(
                 "Failed to parse ExifTool XML: %s",
                 str(ex),
-                exc_info=LOG.getEffectiveLevel() <= logging.DEBUG,
+                exc_info=LOG.isEnabledFor(logging.DEBUG),
             )
             rdf_by_path = {}
         else:
diff --git a/mapillary_tools/geotag/geotag_videos_from_exiftool.py b/mapillary_tools/geotag/geotag_videos_from_exiftool.py
@@ -110,7 +110,7 @@ def _generate_video_extractors(
             LOG.warning(
                 "Failed to parse ExifTool XML: %s",
                 str(ex),
-                exc_info=LOG.getEffectiveLevel() <= logging.DEBUG,
+                exc_info=LOG.isEnabledFor(logging.DEBUG),
             )
             rdf_by_path = {}
         else:
diff --git a/mapillary_tools/geotag/utils.py b/mapillary_tools/geotag/utils.py
@@ -46,7 +46,7 @@ def index_rdf_description_by_path(
         try:
             etree = ET.parse(xml_path)
         except ET.ParseError as ex:
-            verbose = LOG.getEffectiveLevel() <= logging.DEBUG
+            verbose = LOG.isEnabledFor(logging.DEBUG)
             if verbose:
                 LOG.warning("Failed to parse %s", xml_path, exc_info=True)
             else:
diff --git a/mapillary_tools/http.py b/mapillary_tools/http.py
@@ -88,7 +88,7 @@ def _log_debug_request(self, method: str | bytes, url: str | bytes, **kwargs):
         if self.disable_logging_request:
             return
 
-        if logging.getLogger().getEffectiveLevel() <= logging.DEBUG:
+        if not LOG.isEnabledFor(logging.DEBUG):
             return
 
         if isinstance(method, str) and isinstance(url, str):
@@ -124,7 +124,7 @@ def _log_debug_response(self, resp: requests.Response):
         if self.disable_logging_response:
             return
 
-        if logging.getLogger().getEffectiveLevel() <= logging.DEBUG:
+        if not LOG.isEnabledFor(logging.DEBUG):
             return
 
         elapsed = resp.elapsed.total_seconds() * 1000  # Convert to milliseconds
diff --git a/mapillary_tools/process_geotag_properties.py b/mapillary_tools/process_geotag_properties.py
@@ -167,7 +167,7 @@ def _overwrite_exif_tags(
         metadatas,
         desc="Overwriting EXIF",
         unit="images",
-        disable=LOG.getEffectiveLevel() <= logging.DEBUG,
+        disable=LOG.isEnabledFor(logging.DEBUG),
     ):
         dt = datetime.datetime.fromtimestamp(metadata.time, datetime.timezone.utc)
         dt = dt.replace(tzinfo=datetime.timezone.utc)
diff --git a/mapillary_tools/sample_video.py b/mapillary_tools/sample_video.py
@@ -125,12 +125,11 @@ def sample_video(
 
         except Exception as ex:
             if skip_sample_errors:
-                exc_info = LOG.getEffectiveLevel() <= logging.DEBUG
                 LOG.warning(
                     "Skipping the error sampling %s: %s",
                     video_path,
                     str(ex),
-                    exc_info=exc_info,
+                    exc_info=LOG.isEnabledFor(logging.DEBUG),
                 )
             else:
                 raise
diff --git a/mapillary_tools/upload.py b/mapillary_tools/upload.py
@@ -152,7 +152,7 @@ def zip_images(import_path: Path, zip_dir: Path, desc_path: str | None = None):
 
 
 def log_exception(ex: Exception) -> None:
-    if LOG.getEffectiveLevel() <= logging.DEBUG:
+    if LOG.isEnabledFor(logging.DEBUG):
         exc_info = ex
     else:
         exc_info = None
@@ -280,7 +280,7 @@ def upload_start(payload: uploader.Progress) -> None:
             unit_scale=True,
             unit_divisor=1024,
             initial=payload.get("offset", 0),
-            disable=LOG.getEffectiveLevel() <= logging.DEBUG,
+            disable=LOG.isEnabledFor(logging.DEBUG),
         )
 
     @emitter.on("upload_fetch_offset")
@@ -338,7 +338,7 @@ def upload_fetch_offset(payload: uploader.Progress) -> None:
     def upload_progress(payload: uploader.Progress):
         type: uploader.EventName = "upload_progress"
 
-        if LOG.getEffectiveLevel() <= logging.DEBUG:
+        if LOG.isEnabledFor(logging.DEBUG):
             # In debug mode, we want to see the progress every 30 seconds
             # instead of every chunk (which is too verbose)
             INTERVAL_SECONDS = 30
diff --git a/mapillary_tools/uploader.py b/mapillary_tools/uploader.py
@@ -1,7 +1,6 @@
 from __future__ import annotations
 
 import concurrent.futures
-
 import dataclasses
 import io
 import json
diff --git a/mapillary_tools/utils.py b/mapillary_tools/utils.py
@@ -1,9 +1,11 @@
 from __future__ import annotations
 
+import concurrent.futures
+
 import hashlib
+import logging
 import os
 import typing as T
-from multiprocessing import Pool
 from pathlib import Path
 
 
@@ -214,14 +216,57 @@ def mp_map_maybe(
     num_processes: int | None = None,
 ) -> T.Generator[TMapOut, None, None]:
     if num_processes is None:
-        pool_num_processes = None
+        max_workers = None
         disable_multiprocessing = False
     else:
-        pool_num_processes = max(num_processes, 1)
+        max_workers = max(num_processes, 1)
         disable_multiprocessing = num_processes <= 0
 
     if disable_multiprocessing:
         yield from map(func, iterable)
     else:
-        with Pool(processes=pool_num_processes) as pool:
-            yield from pool.imap(func, iterable)
+        app_logger = logging.getLogger(get_app_name())
+        with concurrent.futures.ProcessPoolExecutor(
+            max_workers=max_workers,
+            initializer=configure_logger,
+            initargs=(None, app_logger.getEffectiveLevel()),
+        ) as executor:
+            yield from executor.map(func, iterable)
+
+
+def configure_logger(
+    logger: logging.Logger | None = None, level: int = logging.INFO
+) -> logging.Logger:
+    """Configure logging in each worker process"""
+    if logger is None:
+        # Root logger if app name is ""
+        logger = logging.getLogger(get_app_name())
+
+    logger.setLevel(level)
+
+    try:
+        # Disable globally for now. TODO Disable it in non-interactive mode only
+        raise ImportError
+        from rich.console import Console  # type: ignore
+        from rich.logging import RichHandler  # type: ignore
+    except ImportError:
+        formatter = logging.Formatter(
+            "%(asctime)s.%(msecs)03d - %(levelname)-7s - %(message)s",
+            datefmt="%H:%M:%S",
+        )
+        handler: logging.Handler = logging.StreamHandler()
+        handler.setFormatter(formatter)
+    else:
+        handler = RichHandler(console=Console(stderr=True), rich_tracebacks=True)  # type: ignore
+
+    logger.addHandler(handler)
+
+    return logger
+
+
+def get_app_name() -> str:
+    if __name__:
+        return __name__.split(".")[0]
+    else:
+        # Rare case
+        return ""

Original file line number	Diff line number	Diff line change
`@@ -46,7 +46,7 @@ def to_description(`
`46`	`46`	`map_results,`
`47`	`47`	`desc="Extracting images",`
`48`	`48`	`unit="images",`
`49`		`- disable=LOG.getEffectiveLevel() <= logging.DEBUG,`
	`49`	`+ disable=LOG.isEnabledFor(logging.DEBUG),`
`50`	`50`	`total=len(extractors),`
`51`	`51`	`)`
`52`	`52`	`)`
`@@ -62,6 +62,8 @@ def run_extraction(cls, extractor: TImageExtractor) -> types.ImageMetadataOrErro`
`62`	`62`	`try:`
`63`	`63`	`return extractor.extract()`
`64`	`64`	`except exceptions.MapillaryDescriptionError as ex:`
	`65`	`+ if LOG.isEnabledFor(logging.DEBUG):`
	`66`	`+ LOG.error(f"{cls.__name__}({image_path.name}): {ex}")`
`65`	`67`	`return types.describe_error_metadata(`
`66`	`68`	`ex, image_path, filetype=types.FileType.IMAGE`
`67`	`69`	`)`
`@@ -112,7 +114,7 @@ def to_description(`
`112`	`114`	`map_results,`
`113`	`115`	`desc="Extracting videos",`
`114`	`116`	`unit="videos",`
`115`		`- disable=LOG.getEffectiveLevel() <= logging.DEBUG,`
	`117`	`+ disable=LOG.isEnabledFor(logging.DEBUG),`
`116`	`118`	`total=len(extractors),`
`117`	`119`	`)`
`118`	`120`	`)`
`@@ -128,6 +130,8 @@ def run_extraction(cls, extractor: TVideoExtractor) -> types.VideoMetadataOrErro`
`128`	`130`	`try:`
`129`	`131`	`return extractor.extract()`
`130`	`132`	`except exceptions.MapillaryDescriptionError as ex:`
	`133`	`+ if LOG.isEnabledFor(logging.DEBUG):`
	`134`	`+ LOG.error(f"{cls.__name__}({video_path.name}): {ex}")`
`131`	`135`	`return types.describe_error_metadata(`
`132`	`136`	`ex, video_path, filetype=types.FileType.VIDEO`
`133`	`137`	`)`
Original file line number	Diff line number	Diff line change
`@@ -67,7 +67,7 @@ def process(`
`67`	`67`	`reprocessable_paths = set(paths)`
`68`	`68`
`69`	`69`	`for idx, option in enumerate(options):`
`70`		`- if LOG.getEffectiveLevel() <= logging.DEBUG:`
	`70`	`+ if LOG.isEnabledFor(logging.DEBUG):`
`71`	`71`	`LOG.info(`
`72`	`72`	`f"==> Processing {len(reprocessable_paths)} files with source {option}..."`
`73`	`73`	`)`
Original file line number	Diff line number	Diff line change
`@@ -93,7 +93,7 @@ def _generate_image_extractors(`
`93`	`93`	`LOG.warning(`
`94`	`94`	`"Failed to parse ExifTool XML: %s",`
`95`	`95`	`str(ex),`
`96`		`- exc_info=LOG.getEffectiveLevel() <= logging.DEBUG,`
	`96`	`+ exc_info=LOG.isEnabledFor(logging.DEBUG),`
`97`	`97`	`)`
`98`	`98`	`rdf_by_path = {}`
`99`	`99`	`else:`
Original file line number	Diff line number	Diff line change
`@@ -110,7 +110,7 @@ def _generate_video_extractors(`
`110`	`110`	`LOG.warning(`
`111`	`111`	`"Failed to parse ExifTool XML: %s",`
`112`	`112`	`str(ex),`
`113`		`- exc_info=LOG.getEffectiveLevel() <= logging.DEBUG,`
	`113`	`+ exc_info=LOG.isEnabledFor(logging.DEBUG),`
`114`	`114`	`)`
`115`	`115`	`rdf_by_path = {}`
`116`	`116`	`else:`