Read each GPX file once per run, and warn about it once

mapillary_tools/geotag/geotag_images_from_gpx_file.py

@@ -48,25 +48,30 @@ def _read_gpx_points(source_path: Path) -> list[geo.Point]:
     geotagged from it, and if it is missing or has no timestamped point: the
     file was given to geotag the images from, so it fails them rather than
     leaving them to the next geotag source.
+
+    The file is parsed once in a process (see utils.read_gpx_tracks()), and
+    its points are shared with its other readers, which must not change them.
     """
     if not source_path.is_file():
         raise exceptions.MapillaryInvalidGPXError(f"GPX file not found: {source_path}")
 
     try:
-        tracks = utils.parse_gpx(source_path)
+        tracks = utils.read_gpx_tracks(source_path)
     except (OSError, UnicodeError, gpxpy.gpx.GPXException) as ex:
         raise exceptions.MapillaryInvalidGPXError(
             f"{source_path.name} is not a valid GPX file. "
             "Replace it with a valid GPX file."
         ) from ex
 
     if 1 < len(tracks):
-        LOG.warning(
+        utils.warn_once(
+            source_path,
+            ("tracks", len(tracks)),
             "Found %s tracks in the GPX file %s. Will merge points in all the tracks as a single track for interpolation",
             len(tracks),
             source_path,
         )
-    points: list[geo.Point] = sum(tracks, [])
+    points: list[geo.Point] = [point for track in tracks for point in track]
 
     if not points:
         raise exceptions.MapillaryInvalidGPXError(

mapillary_tools/geotag/geotag_videos_from_gpx.py

@@ -5,6 +5,7 @@
 
 from __future__ import annotations
 
+import collections
 import logging
 import sys
 import typing as T
@@ -16,7 +17,7 @@
     from typing_extensions import override
 
 from .. import exceptions, types
-from . import options
+from . import options, utils
 from .base import GeotagVideosFromGeneric
 from .video_extractors.gpx import GPXVideoExtractor
 
@@ -72,4 +73,28 @@ def _generate_video_extractors(
                         filetype=types.FileType.VIDEO,
                     )
                 )
+
+        # The videos are geotagged in a pool of processes, each of which would
+        # read the track file of its videos. Read here, once, and warn about
+        # once, a file that geotags several of them, or that this process read
+        # already (see _read_here())
+        extractors = [
+            result for result in results if isinstance(result, GPXVideoExtractor)
+        ]
+        num_videos = collections.Counter(
+            extractor.gpx_path.resolve() for extractor in extractors
+        )
+        for extractor in extractors:
+            if self._read_here(extractor, num_videos[extractor.gpx_path.resolve()]):
+                extractor.read_tracks()
+
         return results
+
+    def _read_here(self, extractor: GPXVideoExtractor, num_videos: int) -> bool:
+        """
+        Whether to read the track file of **extractor**, which geotags
+        **num_videos** videos, in this process rather than in the pool: if it
+        geotags several, or was parsed in this process already. A file of
+        each video is left to the pool, which parses them side by side
+        """
+        return 1 < num_videos or utils.was_gpx_read(extractor.gpx_path)

mapillary_tools/geotag/geotag_videos_from_nmea.py

@@ -6,15 +6,12 @@
 from __future__ import annotations
 
 import sys
-import typing as T
-from pathlib import Path
 
 if sys.version_info >= (3, 12):
     from typing import override
 else:
     from typing_extensions import override
 
-from .. import types
 from .geotag_videos_from_gpx import GeotagVideosFromGPX
 from .video_extractors.gpx import GPXVideoExtractor
 from .video_extractors.nmea import NMEAVideoExtractor
@@ -30,14 +27,7 @@ class GeotagVideosFromNMEA(GeotagVideosFromGPX):
     default_pattern = "%g.nmea"
 
     @override
-    def _generate_video_extractors(
-        self, video_paths: T.Sequence[Path]
-    ) -> T.Sequence[GPXVideoExtractor | types.ErrorMetadata]:
-        extractors = super()._generate_video_extractors(video_paths)
-        # Read each NMEA file here, once however many videos it geotags, and
-        # warn about it once: the videos are geotagged in a pool of processes,
-        # each of which would read it again
-        for extractor in extractors:
-            if isinstance(extractor, NMEAVideoExtractor):
-                extractor.read_tracks()
-        return extractors
+    def _read_here(self, extractor: GPXVideoExtractor, num_videos: int) -> bool:
+        # Every NMEA file, which is read once in a process however many videos
+        # it geotags, and warned about once
+        return True

mapillary_tools/geotag/utils.py

@@ -5,7 +5,13 @@
 
 from __future__ import annotations
 
+import collections
+import dataclasses
+import hashlib
+import io
 import logging
+import multiprocessing
+import os
 import typing as T
 import xml.etree.ElementTree as ET
 from pathlib import Path
@@ -17,10 +23,95 @@
 Track = T.List[geo.Point]
 LOG = logging.getLogger(__name__)
 
+# The tracks parsed from the last GPX files read in this process, or the error
+# parsing them raised, by path and content: a GPX file that geotags several
+# videos, or the samples of a video and then their images, is parsed once.
+# Only a few are kept, so that one GPX file per video does not pile up
+_PARSED_GPX: collections.OrderedDict[
+    tuple[str, str], list[Track] | gpxpy.gpx.GPXException | UnicodeError
+] = collections.OrderedDict()
+_MAX_PARSED_GPX = 4
+
+# The warnings given in this process about the track files read, so that each
+# is given once however many times a file is read
+_WARNED: set[tuple[T.Hashable, ...]] = set()
+
 
 def parse_gpx(gpx_file: Path) -> list[Track]:
-    with gpx_file.open("r") as f:
-        gpx = gpxpy.parse(f)
+    """
+    Parse the tracks of a GPX file, each a list of points with Unix times
+    """
+    # Copies, which the caller may change
+    return [
+        [dataclasses.replace(point) for point in track]
+        for track in read_gpx_tracks(gpx_file)
+    ]
+
+
+def read_gpx_tracks(gpx_file: Path) -> list[Track]:
+    """
+    The tracks of a GPX file as parse_gpx() parses them, parsed once in a
+    process for as long as the file does not change. They are shared with
+    every other reader of the file, and must not be changed.
+
+    A worker of a pool keeps none: it parses the GPX file of one video only,
+    as a file that geotags several videos is read before the pool (see
+    GeotagVideosFromGPX), and would free them only when it exits.
+    """
+    data = gpx_file.read_bytes()
+    key = (str(gpx_file.resolve()), hashlib.sha256(data).hexdigest())
+    parsed = _PARSED_GPX.get(key)
+    if parsed is None:
+        try:
+            # Decoded as a file opened in text mode would be, and parsed
+            # without the bytes, as a file is
+            text = io.TextIOWrapper(io.BytesIO(data)).read()
+            del data
+            parsed = _parse_gpx_text(text)
+        except (gpxpy.gpx.GPXException, UnicodeError) as ex:
+            parsed = ex
+        if multiprocessing.parent_process() is None:
+            _PARSED_GPX[key] = parsed
+            while _MAX_PARSED_GPX < len(_PARSED_GPX):
+                _PARSED_GPX.popitem(last=False)
+    else:
+        _PARSED_GPX.move_to_end(key)
+
+    if isinstance(parsed, Exception):
+        raise parsed.with_traceback(None)
+    return parsed
+
+
+def was_gpx_read(gpx_file: Path) -> bool:
+    """
+    Whether the tracks of a GPX file were parsed in this process and are kept
+    """
+    path = str(gpx_file.resolve())
+    return any(key[0] == path for key in _PARSED_GPX)
+
+
+def warn_once(
+    path: Path, what: tuple[T.Hashable, ...], message: str, *args: object
+) -> None:
+    """
+    Log a warning about the track file **path**, unless this process warned
+    already of **what** about the file as it is now, whoever read it
+    """
+    try:
+        stat = os.stat(path)
+    except OSError:
+        file_key: tuple[T.Hashable, ...] = (str(path),)
+    else:
+        file_key = (str(path.resolve()), stat.st_size, stat.st_mtime_ns)
+    key = (*file_key, *what)
+    if key in _WARNED:
+        return
+    _WARNED.add(key)
+    LOG.warning(message, *args)
+
+
+def _parse_gpx_text(text: str) -> list[Track]:
+    gpx = gpxpy.parse(text)
 
     tracks: list[Track] = []
 

mapillary_tools/geotag/video_extractors/gpx.py

@@ -29,8 +29,7 @@
 from ...mp4 import construct_mp4_parser as cparser, simple_mp4_parser as sparser
 from ...mp4.mp4_sample_parser import MovieBoxParser
 from ...serializer.description import build_capture_time
-from .. import options
-from ..utils import parse_gpx
+from .. import options, utils as geotag_utils
 from .base import BaseVideoExtractor
 from .native import NativeVideoExtractor
 
@@ -175,6 +174,10 @@ class GPXVideoExtractor(BaseVideoExtractor):
     # The format of the track file, as messages name it
     file_format = "GPX"
 
+    # The tracks of the track file, or the error reading it, once read_tracks()
+    # has read them
+    _tracks: list[list[geo.Point]] | exceptions.MapillaryInvalidGPXError | None = None
+
     def __init__(
         self,
         video_path: Path,
@@ -285,39 +288,93 @@ def extract(self) -> types.VideoMetadata:
 
         return dataclasses.replace(native_video_metadata, points=gpx_points)
 
+    def read_tracks(self) -> None:
+        """
+        Read the tracks of the track file now, for extract() to place, rather
+        than when it runs, which can be in another process: a file that
+        geotags several videos is then read, and warned about, once
+        """
+        try:
+            self._tracks = self._read_tracks()
+        except exceptions.MapillaryInvalidGPXError as ex:
+            self._tracks = ex
+        except Exception:
+            # Left for extract() to raise again, and to be reported for the
+            # video as any unexpected error is
+            pass
+        else:
+            self._warn_of_tracks(self._tracks)
+
     def _read_gpx_points(self) -> list[geo.Point]:
         """
         Read the points of the GPX file, merged into one track in time order
         """
+        warned = self._tracks is not None
         gpx_tracks = self._parse_tracks()
+        if not warned:
+            self._warn_of_tracks(gpx_tracks)
+
+        # Copies, which are moved to the video: the tracks may be shared with
+        # other readers of the file
+        gpx_points = [
+            dataclasses.replace(p)
+            for track in gpx_tracks
+            for p in track
+            if _GPS_EPOCH <= p.time
+        ]
 
+        # The tracks need not be in time order, and interpolation needs the
+        # merged points to be
+        gpx_points.sort(key=lambda p: p.time)
+        return gpx_points
+
+    def _warn_of_tracks(self, gpx_tracks: list[list[geo.Point]]) -> None:
+        """
+        Warn of what reading the points of the tracks leaves out or merges,
+        once for each file
+        """
         if 1 < len(gpx_tracks):
-            LOG.warning(
-                f"Found {len(gpx_tracks)} tracks in the {self.file_format} file {self.gpx_path}. Will merge points in all the tracks as a single track for interpolation"
+            geotag_utils.warn_once(
+                self.gpx_path,
+                ("tracks", len(gpx_tracks)),
+                "Found %d tracks in the %s file %s. Will merge points in all the tracks as a single track for interpolation",
+                len(gpx_tracks),
+                self.file_format,
+                self.gpx_path,
             )
 
-        all_points: list[geo.Point] = sum(gpx_tracks, [])
-        gpx_points = [p for p in all_points if _GPS_EPOCH <= p.time]
-        if len(gpx_points) < len(all_points):
-            LOG.warning(
+        num_points = sum(len(track) for track in gpx_tracks)
+        num_kept = sum(1 for track in gpx_tracks for p in track if _GPS_EPOCH <= p.time)
+        if num_kept < num_points:
+            geotag_utils.warn_once(
+                self.gpx_path,
+                ("before 1980", num_points - num_kept, num_points),
                 "Ignoring %d of the %d points in the %s file %s, which are timestamped before 1980",
-                len(all_points) - len(gpx_points),
-                len(all_points),
+                num_points - num_kept,
+                num_points,
                 self.file_format,
                 self.gpx_path,
             )
 
-        # The tracks need not be in time order, and interpolation needs the
-        # merged points to be
-        gpx_points.sort(key=lambda p: p.time)
-        return gpx_points
-
     def _parse_tracks(self) -> list[list[geo.Point]]:
         """
-        Parse the tracks of the GPX file, each a list of points with Unix times
+        The tracks of the track file, each a list of points with Unix times,
+        as read_tracks() read them, or read now. They may be shared with other
+        readers of the file, and must not be changed
+        """
+        if self._tracks is None:
+            return self._read_tracks()
+        if isinstance(self._tracks, exceptions.MapillaryInvalidGPXError):
+            raise self._tracks
+        return self._tracks
+
+    def _read_tracks(self) -> list[list[geo.Point]]:
+        """
+        Read the tracks of the GPX file, parsed once in a process (see
+        read_gpx_tracks())
         """
         try:
-            return parse_gpx(self.gpx_path)
+            return geotag_utils.read_gpx_tracks(self.gpx_path)
         except (OSError, UnicodeError, gpxpy.gpx.GPXException) as ex:
             raise exceptions.MapillaryInvalidGPXError(
                 f"{self.gpx_path.name} is not a valid GPX file. "

mapillary_tools/geotag/video_extractors/nmea.py

@@ -26,32 +26,7 @@ class NMEAVideoExtractor(GPXVideoExtractor):
 
     file_format = "NMEA"
 
-    # The tracks of the NMEA file, or the error reading it, once read_tracks()
-    # has read them
-    _tracks: list[list[geo.Point]] | exceptions.MapillaryInvalidGPXError | None = None
-
-    def read_tracks(self) -> None:
-        """
-        Read the tracks of the NMEA file now, for extract() to place, rather
-        than when it runs, which can be in another process
-        """
-        try:
-            self._tracks = self._read_tracks()
-        except exceptions.MapillaryInvalidGPXError as ex:
-            self._tracks = ex
-        except Exception:
-            # Left for extract() to raise again, and to be reported for the
-            # video as any unexpected error is
-            pass
-
     @override
-    def _parse_tracks(self) -> list[list[geo.Point]]:
-        if self._tracks is None:
-            return self._read_tracks()
-        if isinstance(self._tracks, exceptions.MapillaryInvalidGPXError):
-            raise self._tracks
-        return self._tracks
-
     def _read_tracks(self) -> list[list[geo.Point]]:
         try:
             points = get_lat_lon_time_from_nmea(self.gpx_path)