improve: enable logging for multiprocessing

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)

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
 

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
             )

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}..."
             )

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:

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:

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:

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

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)

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

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

mapillary_tools/uploader.py

@@ -1,7 +1,6 @@
 from __future__ import annotations
 
 import concurrent.futures
-
 import dataclasses
 import io
 import json

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 ""