Resolve an exiftool_xml pattern against the videos of video_process frames

README.md

@@ -443,7 +443,8 @@ possible to specify its location by setting the environment variable `MAPILLARY_
 * Pattern are case-sensitive or not depending on the filesystem - in Windows, `%g.gpx` will match both `basename.gpx`
 and `basename.GPX`, in MacOs, Linux or other Unix systems no.
 * If both `--video_geotag_source` and `--geotag_source` are specified, `--video_geotag_source` will apply to video files
-and `--geotag_source` to image files.
+and `--geotag_source` to image files. If only `--video_geotag_source` is specified, image files (including the frames
+that `video_process` samples) are geotagged by the default sources.
 
 ### Examples
 

mapillary_tools/geotag/factory.py

@@ -10,6 +10,8 @@
 import typing as T
 from pathlib import Path
 
+import jsonschema
+
 from .. import exceptions, types, utils
 from . import (
     base,
@@ -50,6 +52,9 @@ def parse_source_option(source: str) -> list[SourceOption]:
         "gpx,exif" -> [SourceOption(SourceType.GPX), SourceOption(SourceType.EXIF)]
         "exif,gpx" -> [SourceOption(SourceType.EXIF), SourceOption(SourceType.GPX)]
         '{"source": "gpx"}' -> [SourceOption(SourceType.GPX)]
+
+    Raise MapillaryBadParameterError for an unknown source type, and for a
+    JSON object that does not match SourceOptionSchema.
     """
 
     try:
@@ -64,11 +69,27 @@ def parse_source_option(source: str) -> list[SourceOption]:
     except json.JSONDecodeError:
         pass
     else:
-        return [SourceOption.from_dict(payload)]
+        try:
+            return [SourceOption.from_dict(payload)]
+        except jsonschema.ValidationError as ex:
+            raise exceptions.MapillaryBadParameterError(
+                f"Invalid geotag source {source}: {ex.message}"
+            ) from ex
 
     sources = source.split(",")
 
-    return [SourceOption(SourceType(SOURCE_TYPE_ALIAS.get(s, s))) for s in sources]
+    return [SourceOption(_parse_source_type(s)) for s in sources]
+
+
+def _parse_source_type(source: str) -> SourceType:
+    try:
+        return SourceType(SOURCE_TYPE_ALIAS.get(source, source))
+    except ValueError:
+        raise exceptions.MapillaryBadParameterError(
+            f"Invalid geotag source {source!r}. Supported source types: "
+            f"{', '.join(s.value for s in SourceType)} "
+            f"(aliases: {', '.join(SOURCE_TYPE_ALIAS)})"
+        ) from None
 
 
 def parse_source_options(

mapillary_tools/geotag/geotag_images_from_exiftool.py

@@ -125,6 +125,10 @@ def __init__(
     def geotag_samples(
         self, image_paths: T.Sequence[Path]
     ) -> list[types.ImageMetadataOrError]:
+        sample_metadata_or_errors = self._geotag_samples_by_pattern(image_paths)
+        named_paths = set(metadata.filename for metadata in sample_metadata_or_errors)
+        image_paths = [path for path in image_paths if path not in named_paths]
+
         # Find all video paths in self.xml_path
         rdf_by_path = GeotagVideosFromExifToolXML.find_rdf_by_path(
             self.source_path, image_paths
@@ -133,11 +137,45 @@ def geotag_samples(
             [Path(canonical_path) for canonical_path in rdf_by_path.keys()],
             skip_subfolders=True,
         )
+
+        return sample_metadata_or_errors + self._geotag_samples_of(
+            image_paths, video_paths, self.source_path
+        )
+
+    def _geotag_samples_by_pattern(
+        self, image_paths: T.Sequence[Path]
+    ) -> list[types.ImageMetadataOrError]:
+        """
+        Geotag the samples of the videos whose XML the pattern names. A pattern
+        names the XML of a video after the video, not after its samples: in
+        video_process, given the video import path (a video or a directory of
+        videos) as the source path, resolve it against the videos that the
+        samples were taken from, found there. The source path would override
+        the pattern otherwise, and read the video import path as the XML.
+        """
+        if self.source_path.pattern is None or self.source_path.source_path is None:
+            return []
+
+        pattern = options.SourcePathOption(pattern=self.source_path.pattern)
+        video_paths = [
+            video_path
+            for video_path in utils.find_videos([self.source_path.source_path])
+            if pattern.resolve(video_path).is_file()
+        ]
+
+        return self._geotag_samples_of(image_paths, video_paths, pattern)
+
+    def _geotag_samples_of(
+        self,
+        image_paths: T.Sequence[Path],
+        video_paths: T.Sequence[Path],
+        source_path: options.SourcePathOption,
+    ) -> list[types.ImageMetadataOrError]:
         # Find all video paths that have sample images
         samples_by_video = utils.find_all_image_samples(image_paths, video_paths)
 
         video_metadata_or_errors = GeotagVideosFromExifToolXML(
-            self.source_path, num_processes=self.num_processes
+            source_path, num_processes=self.num_processes
         ).to_description(list(samples_by_video.keys()))
         sample_paths = sum(samples_by_video.values(), [])
         sample_metadata_or_errors = GeotagImagesFromVideo(

mapillary_tools/geotag/geotag_images_from_gpx.py

@@ -39,6 +39,8 @@ def __init__(
         self.use_gpx_start_time = use_gpx_start_time
         self.use_image_start_time = use_image_start_time
         self.offset_time = offset_time
+        # Why the track could not be read, reported for every image instead
+        self.track_error: Exception | None = None
 
     def _interpolate_image_metadata_along(
         self,
@@ -96,6 +98,14 @@ def _generate_image_extractors(
     def to_description(
         self, image_paths: T.Sequence[Path]
     ) -> list[types.ImageMetadataOrError]:
+        if self.track_error is not None:
+            return [
+                types.describe_error_metadata(
+                    self.track_error, image_path, filetype=types.FileType.IMAGE
+                )
+                for image_path in image_paths
+            ]
+
         final_metadatas: list[types.ImageMetadataOrError] = []
 
         image_metadata_or_errors = super().to_description(image_paths)

mapillary_tools/geotag/geotag_images_from_gpx_file.py

@@ -8,6 +8,9 @@
 import logging
 from pathlib import Path
 
+import gpxpy
+
+from .. import exceptions, geo
 from . import utils
 from .geotag_images_from_gpx import GeotagImagesFromGPX
 
@@ -23,23 +26,51 @@ def __init__(
         offset_time: float = 0.0,
         num_processes: int | None = None,
     ):
+        track_error: Exception | None = None
         try:
-            tracks = utils.parse_gpx(source_path)
-        except Exception as ex:
-            raise RuntimeError(
-                f"Error parsing GPX {source_path}: {ex.__class__.__name__}: {ex}"
-            )
-
-        if 1 < len(tracks):
-            LOG.warning(
-                "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 = sum(tracks, [])
+            points = _read_gpx_points(source_path)
+        except exceptions.MapillaryInvalidGPXError as ex:
+            points, track_error = [], ex
+
         super().__init__(
             points,
             use_gpx_start_time=use_gpx_start_time,
             offset_time=offset_time,
             num_processes=num_processes,
         )
+        self.track_error = track_error
+
+
+def _read_gpx_points(source_path: Path) -> list[geo.Point]:
+    """
+    Read the points of a GPX file, merged into one track. Raise
+    MapillaryInvalidGPXError if the file cannot be parsed, as for a video
+    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.
+    """
+    if not source_path.is_file():
+        raise exceptions.MapillaryInvalidGPXError(f"GPX file not found: {source_path}")
+
+    try:
+        tracks = utils.parse_gpx(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(
+            "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, [])
+
+    if not points:
+        raise exceptions.MapillaryInvalidGPXError(
+            f"No timestamped points found in the GPX file {source_path}"
+        )
+
+    return points

mapillary_tools/geotag/geotag_images_from_nmea_file.py

@@ -10,7 +10,7 @@
 
 import pynmea2
 
-from .. import geo
+from .. import exceptions, geo
 from .geotag_images_from_gpx import GeotagImagesFromGPX
 
 
@@ -22,13 +22,47 @@ def __init__(
         offset_time: float = 0.0,
         num_processes: int | None = None,
     ):
-        points = get_lat_lon_time_from_nmea(source_path)
+        track_error: Exception | None = None
+        try:
+            points = _read_nmea_points(source_path)
+        except exceptions.MapillaryInvalidGPXError as ex:
+            points, track_error = [], ex
+
         super().__init__(
             points,
             use_gpx_start_time=use_gpx_start_time,
             offset_time=offset_time,
             num_processes=num_processes,
         )
+        self.track_error = track_error
+
+
+def _read_nmea_points(source_path: Path) -> list[geo.Point]:
+    """
+    Read the points of an NMEA file, raising as _read_gpx_points() does for a
+    GPX file
+    """
+    if not source_path.is_file():
+        raise exceptions.MapillaryInvalidGPXError(f"NMEA file not found: {source_path}")
+
+    try:
+        points = get_lat_lon_time_from_nmea(source_path)
+    # OSError: unreadable; ValueError: not text (UnicodeError) or a malformed
+    # sentence (pynmea2.ParseError); TypeError: a sentence without its time or
+    # date; UnboundLocalError: GGA sentences but no RMC sentence to give their
+    # date
+    except (OSError, ValueError, TypeError, UnboundLocalError) as ex:
+        raise exceptions.MapillaryInvalidGPXError(
+            f"{source_path.name} is not a valid NMEA file. "
+            "Replace it with a valid NMEA file."
+        ) from ex
+
+    if not points:
+        raise exceptions.MapillaryInvalidGPXError(
+            f"No timestamped points found in the NMEA file {source_path}"
+        )
+
+    return points
 
 
 def get_lat_lon_time_from_nmea(nmea_file: Path) -> list[geo.Point]:

mapillary_tools/geotag/geotag_images_from_video.py

@@ -146,7 +146,7 @@ def to_description(
                 num_processes=self.num_processes,
             )
 
-            image_metadatas = geotag.to_description(image_paths)
+            image_metadatas = geotag.to_description(sample_paths)
 
             for metadata in image_metadatas:
                 if isinstance(metadata, types.ImageMetadata):

mapillary_tools/process_geotag_properties.py

@@ -15,7 +15,7 @@
 
 from . import constants, exceptions, exif_write, types, utils
 from .geotag.factory import parse_source_options, process
-from .geotag.options import InterpolationOption, SourceType
+from .geotag.options import InterpolationOption, SourceOption, SourceType
 from .serializer.description import (
     DescriptionJSONSerializer,
     validate_and_fail_metadata,
@@ -75,6 +75,15 @@ def process_geotag_properties(
         geotag_source_path=geotag_source_path,
     )
 
+    # --video_geotag_source applies to videos only. Without --geotag_source,
+    # images (such as the frames that video_process samples) are geotagged by
+    # the default sources, as they are when neither option is given
+    if not geotag_source:
+        options.extend(
+            SourceOption(SourceType(source), filetypes={types.FileType.IMAGE})
+            for source in DEFAULT_GEOTAG_SOURCE_OPTIONS
+        )
+
     for option in options:
         option.filetypes = types.combine_filetype_filters(option.filetypes, filetypes)
         option.num_processes = num_processes

tests/integration/test_process.py

@@ -429,6 +429,34 @@ def test_geotagging_images_from_gpx(setup_data: py.path.local):
     )
 
 
+def test_geotagging_images_from_malformed_gpx(setup_data: py.path.local):
+    """Every image gets the error, as a video geotagged from the file does."""
+    gpx_file = setup_data.join("test.gpx")
+    gpx_file.write("<gpx><broken>")
+    images = setup_data.join("images")
+
+    descs = run_process_for_descs(
+        [
+            "--file_types=image",
+            "--geotag_source=gpx",
+            "--geotag_source_path",
+            str(gpx_file),
+            str(images),
+        ]
+    )
+
+    assert sorted(Path(desc["filename"]).name for desc in descs) == [
+        "DSC00001.JPG",
+        "DSC00497.JPG",
+        "V0370574.JPG",
+    ]
+    for desc in descs:
+        assert desc["error"] == {
+            "type": "MapillaryInvalidGPXError",
+            "message": "test.gpx is not a valid GPX file. Replace it with a valid GPX file.",
+        }
+
+
 def test_geotagging_images_from_gpx_with_offset(setup_data: py.path.local):
     gpx_file = setup_data.join("test.gpx")
     with gpx_file.open("w") as fp: