From f7afc573839b065152e7e4bead5111be693a6f95 Mon Sep 17 00:00:00 2001 From: Caglar Pir Date: Thu, 22 Oct 2026 11:39:45 +0200 Subject: [PATCH 1/2] Report an unknown geotag source as a bad parameter A misspelled source type, as in --geotag_source foo or gpx,foo, failed with a ValueError traceback ("'foo' is not a valid SourceType"), and a JSON source with a key the schema does not know, such as {"source": "gpx", "path": "track.gpx"} failed with a jsonschema ValidationError traceback. Both exited 1, as for a crash. They now raise MapillaryBadParameterError (exit code 2), naming the supported source types and their aliases, or what the schema rejected ("'path' was unexpected"). This covers --geotag_source and --video_geotag_source in process and video_process, which parse them the same way. --- mapillary_tools/geotag/factory.py | 25 +++++++++- tests/unit/test_parse_source_option.py | 68 ++++++++++++++++++++++++++ 2 files changed, 91 insertions(+), 2 deletions(-) create mode 100644 tests/unit/test_parse_source_option.py diff --git a/mapillary_tools/geotag/factory.py b/mapillary_tools/geotag/factory.py index dc9d21e25..3a3d975f7 100644 --- a/mapillary_tools/geotag/factory.py +++ b/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( diff --git a/tests/unit/test_parse_source_option.py b/tests/unit/test_parse_source_option.py new file mode 100644 index 000000000..c3412f50e --- /dev/null +++ b/tests/unit/test_parse_source_option.py @@ -0,0 +1,68 @@ +# Copyright (c) Meta Platforms, Inc. and affiliates. +# +# This source code is licensed under the BSD license found in the +# LICENSE file in the root directory of this source tree. + +from __future__ import annotations + +from pathlib import Path + +import pytest +from mapillary_tools import exceptions, sample_video +from mapillary_tools.geotag.factory import parse_source_option, parse_source_options +from mapillary_tools.geotag.options import SourceType + + +class TestParseSourceOption: + def test_source_types(self): + assert [o.source for o in parse_source_option("gpx")] == [SourceType.GPX] + assert [o.source for o in parse_source_option("gopro_videos")] == [ + SourceType.GOPRO + ] + assert [o.source for o in parse_source_option("exif,native")] == [ + SourceType.EXIF, + SourceType.NATIVE, + ] + assert [o.source for o in parse_source_option('{"source": "nmea"}')] == [ + SourceType.NMEA + ] + + @pytest.mark.parametrize("source", ["foo", "gpx,foo", "foo,gpx"]) + def test_unknown_source_type_names_the_supported_ones(self, source): + with pytest.raises(exceptions.MapillaryBadParameterError) as excinfo: + parse_source_option(source) + + message = str(excinfo.value) + assert "'foo'" in message + assert ", ".join(s.value for s in SourceType) in message + assert "(aliases: blackvue_videos, gopro_videos, exiftool)" in message + + def test_unknown_key_is_named(self): + with pytest.raises(exceptions.MapillaryBadParameterError) as excinfo: + parse_source_option('{"source": "gpx", "path": "track.gpx"}') + + assert "'path' was unexpected" in str(excinfo.value) + + @pytest.mark.parametrize( + "source", + [ + '{"source": "foo"}', + '{"path": "track.gpx"}', + '{"source": "gpx", "filetypes": ["mp4"]}', + '{"source": "gpx", "interpolation_offset_time": "1s"}', + "123", + ], + ) + def test_invalid_json_source(self, source): + with pytest.raises(exceptions.MapillaryBadParameterError): + parse_source_option(source) + + def test_video_geotag_source(self): + with pytest.raises(exceptions.MapillaryBadParameterError): + parse_source_options( + geotag_source=[], video_geotag_source=["foo"], geotag_source_path=None + ) + + def test_video_process(self, tmp_path: Path): + with pytest.raises(exceptions.MapillaryBadParameterError): + sample_video._parse_video_sources(["foo"], None, None, tmp_path) From 294c0f9970886c91b32c6a45207ae5d6712d2fa5 Mon Sep 17 00:00:00 2001 From: Caglar Pir Date: Thu, 22 Oct 2026 11:48:48 +0200 Subject: [PATCH 2/2] Report a GPX file that images cannot be geotagged from as an error per image A truncated GPX file, or a file that is no GPX at all, given as the GPX to geotag images from failed the whole run with a RuntimeError traceback ("Error parsing GPX ...: GPXXMLSyntaxException"). A video geotagged from the same file gets a MapillaryInvalidGPXError in its description. A GPX path that does not exist failed with a RuntimeError too, and a GPX file with no timestamped point with an AssertionError. An NMEA file failed with a traceback in each of these cases, and when a sentence had no time or date (TypeError), such as the void ones a receiver writes before its first fix, or when the file had no RMC sentence to give the date (UnboundLocalError). Every image now gets a MapillaryInvalidGPXError in the description, saying what is wrong with the file, and the run exits as for any other file that fails to process. The file was given to geotag the images from, so a mistyped path fails them, as it did, rather than leaving them to the next geotag source without a word. --- .../geotag/geotag_images_from_gpx.py | 10 ++ .../geotag/geotag_images_from_gpx_file.py | 57 +++++-- .../geotag/geotag_images_from_nmea_file.py | 38 ++++- tests/integration/test_process.py | 28 ++++ .../test_geotag_images_from_track_file.py | 149 ++++++++++++++++++ 5 files changed, 267 insertions(+), 15 deletions(-) create mode 100644 tests/unit/test_geotag_images_from_track_file.py diff --git a/mapillary_tools/geotag/geotag_images_from_gpx.py b/mapillary_tools/geotag/geotag_images_from_gpx.py index 97b852a3c..f06c1d1f7 100644 --- a/mapillary_tools/geotag/geotag_images_from_gpx.py +++ b/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) diff --git a/mapillary_tools/geotag/geotag_images_from_gpx_file.py b/mapillary_tools/geotag/geotag_images_from_gpx_file.py index 010885f18..77f7ae42b 100644 --- a/mapillary_tools/geotag/geotag_images_from_gpx_file.py +++ b/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 diff --git a/mapillary_tools/geotag/geotag_images_from_nmea_file.py b/mapillary_tools/geotag/geotag_images_from_nmea_file.py index 04956c75f..cb4ef27aa 100644 --- a/mapillary_tools/geotag/geotag_images_from_nmea_file.py +++ b/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]: diff --git a/tests/integration/test_process.py b/tests/integration/test_process.py index 067ba3ddd..6e1482574 100644 --- a/tests/integration/test_process.py +++ b/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("") + 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: diff --git a/tests/unit/test_geotag_images_from_track_file.py b/tests/unit/test_geotag_images_from_track_file.py new file mode 100644 index 000000000..c1b5d1f93 --- /dev/null +++ b/tests/unit/test_geotag_images_from_track_file.py @@ -0,0 +1,149 @@ +# Copyright (c) Meta Platforms, Inc. and affiliates. +# +# This source code is licensed under the BSD license found in the +# LICENSE file in the root directory of this source tree. + +from __future__ import annotations + +from pathlib import Path + +import pytest +from mapillary_tools import exceptions, types +from mapillary_tools.geotag import factory +from mapillary_tools.geotag.geotag_images_from_gpx_file import GeotagImagesFromGPXFile +from mapillary_tools.geotag.geotag_images_from_nmea_file import GeotagImagesFromNMEAFile +from mapillary_tools.geotag.options import SourceOption, SourcePathOption, SourceType + +IMAGES = [Path("a.jpg"), Path("b.jpg")] + +GPX = ( + '\n' + '' + "{}\n" +) +GPX_POINT = '' + +NMEA_RMC = "$GPRMC,081754.14,A,5130.63420,N,00008.06160,W,0.0,0.0,150921,,,A*45\n" +NMEA_GGA = "$GPGGA,081754.14,5130.63420,N,00008.06160,W,1,08,1.0,22.4,M,0.0,M,,*74\n" + + +def _errors(geotag) -> list[Exception]: + descs = geotag.to_description(IMAGES) + assert [d.filename for d in descs] == IMAGES + assert all(isinstance(d, types.ErrorMetadata) for d in descs) + return [d.error for d in descs] + + +def _assert_fails_every_image(geotag, message: str) -> None: + for error in _errors(geotag): + assert isinstance(error, exceptions.MapillaryInvalidGPXError) + assert message in str(error) + + +class TestGPXFile: + @pytest.mark.parametrize( + "content", + [ + # Cut short, as by an interrupted download + GPX.format(GPX_POINT * 3)[:-30], + "This is not a GPX file.\n", + ], + ) + def test_a_file_that_is_no_gpx_fails_every_image(self, tmp_path, content): + gpx = tmp_path / "track.gpx" + gpx.write_text(content) + + _assert_fails_every_image( + GeotagImagesFromGPXFile(gpx), + "track.gpx is not a valid GPX file. Replace it with a valid GPX file.", + ) + + def test_a_missing_file_fails_every_image(self, tmp_path): + _assert_fails_every_image( + GeotagImagesFromGPXFile(tmp_path / "missing.gpx"), "GPX file not found" + ) + + def test_a_track_without_timestamps_fails_every_image(self, tmp_path): + gpx = tmp_path / "track.gpx" + gpx.write_text(GPX.format('')) + + _assert_fails_every_image( + GeotagImagesFromGPXFile(gpx), "No timestamped points found" + ) + + def test_a_valid_file_is_read(self, tmp_path): + gpx = tmp_path / "track.gpx" + gpx.write_text(GPX.format(GPX_POINT)) + + geotag = GeotagImagesFromGPXFile(gpx) + + assert geotag.track_error is None + assert [(p.lat, p.lon) for p in geotag.points] == [(37.0, -122.0)] + + +class TestNMEAFile: + @pytest.mark.parametrize( + "content", + [ + # The checksum does not match + NMEA_RMC.replace("*45", "*00") + NMEA_GGA, + # Void, as a receiver writes before its first fix: no time or date + "$GPRMC,,V,,,,,,,,,,N*53\n" + NMEA_GGA, + # No RMC sentence to give the date + NMEA_GGA, + ], + ) + def test_a_file_that_fails_to_parse_fails_every_image(self, tmp_path, content): + nmea = tmp_path / "track.nmea" + nmea.write_text(content) + + # Not a valid NMEA file, or none of its points timestamped if the + # parser skips the sentences it cannot time + _assert_fails_every_image(GeotagImagesFromNMEAFile(nmea), "track.nmea") + + def test_a_file_without_sentences_fails_every_image(self, tmp_path): + nmea = tmp_path / "track.nmea" + nmea.write_text("This is not a GPX file.\n") + + _assert_fails_every_image( + GeotagImagesFromNMEAFile(nmea), "No timestamped points found" + ) + + def test_a_missing_file_fails_every_image(self, tmp_path): + _assert_fails_every_image( + GeotagImagesFromNMEAFile(tmp_path / "missing.nmea"), "NMEA file not found" + ) + + def test_a_valid_file_is_read(self, tmp_path): + nmea = tmp_path / "track.nmea" + nmea.write_text(NMEA_RMC + NMEA_GGA) + + geotag = GeotagImagesFromNMEAFile(nmea) + + assert geotag.track_error is None + assert len(geotag.points) == 1 + + +def test_the_next_source_does_not_geotag_the_images_of_a_missing_gpx(tmp_path): + """ + The GPX was given to geotag the images from, and a mistyped path must not + pass unnoticed because the next source geotags them instead + """ + images = sorted( + Path(__file__).resolve().parent.parent.joinpath("data", "images").glob("*.JPG") + ) + assert images + options = [ + SourceOption( + SourceType.GPX, + source_path=SourcePathOption(source_path=tmp_path / "missing.gpx"), + ), + SourceOption(SourceType.EXIF), + ] + + descs = factory.process(images, options) + + assert sorted(d.filename for d in descs) == images + for desc in descs: + assert isinstance(desc, types.ErrorMetadata) + assert isinstance(desc.error, exceptions.MapillaryInvalidGPXError)