From f7afc573839b065152e7e4bead5111be693a6f95 Mon Sep 17 00:00:00 2001 From: Caglar Pir Date: Thu, 22 Oct 2026 11:39:45 +0200 Subject: [PATCH 1/3] 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/3] 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) From 7ba61fc5580912155d415d3f9049dabab44735d4 Mon Sep 17 00:00:00 2001 From: Caglar Pir Date: Thu, 22 Oct 2026 11:42:11 +0200 Subject: [PATCH 3/3] Geotag images by the default sources when only --video_geotag_source is given The README says that --video_geotag_source applies to videos and --geotag_source to images. But the default sources were used only when neither option was given, so with --video_geotag_source alone, images had no source at all, and went missing from the description file without a word, as if they had not been there. That includes the frames video_process samples: `video_process VIDEO --video_geotag_source native` sampled the frames, then wrote an empty description file and exited 0. Geotag images by the default sources when only --video_geotag_source is given, as when neither option is. The videos are still geotagged by --video_geotag_source alone. --- README.md | 3 +- mapillary_tools/process_geotag_properties.py | 11 +- tests/integration/test_video_process.py | 26 +++++ tests/unit/test_process_geotag_properties.py | 102 +++++++++++++++++++ 4 files changed, 140 insertions(+), 2 deletions(-) create mode 100644 tests/unit/test_process_geotag_properties.py diff --git a/README.md b/README.md index 53205badc..d791e4690 100644 --- a/README.md +++ b/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 diff --git a/mapillary_tools/process_geotag_properties.py b/mapillary_tools/process_geotag_properties.py index af49f68b0..5a1b5493b 100644 --- a/mapillary_tools/process_geotag_properties.py +++ b/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 diff --git a/tests/integration/test_video_process.py b/tests/integration/test_video_process.py index cad2b6b4a..616736d0e 100644 --- a/tests/integration/test_video_process.py +++ b/tests/integration/test_video_process.py @@ -125,6 +125,32 @@ def test_video_process_by_distance_with_an_image_only_geotag_source( assert all("error" not in desc for desc in descs), descs +def test_video_process_by_distance_with_a_video_geotag_source_only( + setup_data: py.path.local, +): + """ + --video_geotag_source applies to the video. The frames, which are images, + are geotagged by the default sources, as without it. They were geotagged by + none, and the description file came out empty. + """ + pytest_skip_if_not_ffmpeg_installed() + + video_path = setup_data.join("gopro_data").join("max-360mode.mp4") + + with tempfile.TemporaryDirectory() as dir: + descs = run_video_process_for_descs( + [ + "--video_sample_distance=6", + "--video_geotag_source=native", + str(video_path), + dir, + ] + ) + + assert descs + assert all("error" not in desc for desc in descs), descs + + def _with_creation_time(video: py.path.local, creation_time: str, dir: str) -> Path: """A copy of the video whose container records the creation time.""" copy = Path(dir, video.basename) diff --git a/tests/unit/test_process_geotag_properties.py b/tests/unit/test_process_geotag_properties.py new file mode 100644 index 000000000..a741c8110 --- /dev/null +++ b/tests/unit/test_process_geotag_properties.py @@ -0,0 +1,102 @@ +# 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 + +import typing as T +from pathlib import Path + +import pytest +from mapillary_tools import process_geotag_properties as pgp, types +from mapillary_tools.geotag import factory +from mapillary_tools.geotag.options import SourceOption, SourceType + + +class TestSourcesByFileType: + """ + The sources that each kind of file is geotagged by: --video_geotag_source + applies to videos, and --geotag_source to images, and to videos as well + when it is the only one given. A kind of file that neither option names + gets the default sources. + """ + + @pytest.fixture + def import_dir(self, tmp_path: Path) -> Path: + (tmp_path / "image.jpg").touch() + (tmp_path / "video.mp4").touch() + return tmp_path + + @staticmethod + def _sources( + monkeypatch, + import_dir: Path, + geotag_source: list[str], + video_geotag_source: list[str], + filetypes: set[types.FileType] | None = None, + ) -> dict[str, list[SourceType]]: + """The sources, in order, that process() tries on the image and the video""" + seen: list[T.Collection[SourceOption]] = [] + + def process(paths, options, filetypes=None): + seen.append(options) + return [] + + monkeypatch.setattr(pgp, "process", process) + pgp.process_geotag_properties( + import_dir, + filetypes, + geotag_source=geotag_source, + geotag_source_path=None, + video_geotag_source=video_geotag_source, + ) + + [options] = seen + sources: dict[str, list[SourceType]] = {"image": [], "video": []} + for option in options: + images, videos = factory._filter_images_and_videos( + [import_dir / "image.jpg", import_dir / "video.mp4"], option.filetypes + ) + if images: + sources["image"].append(option.source) + if videos: + sources["video"].append(option.source) + return sources + + DEFAULT = [SourceType.NATIVE, SourceType.EXIFTOOL_RUNTIME] + + def test_neither_option(self, monkeypatch, import_dir): + assert self._sources(monkeypatch, import_dir, [], []) == { + "image": self.DEFAULT, + "video": self.DEFAULT, + } + + def test_geotag_source_alone_applies_to_videos_too(self, monkeypatch, import_dir): + assert self._sources(monkeypatch, import_dir, ["exif"], []) == { + "image": [SourceType.EXIF], + "video": [SourceType.EXIF], + } + + def test_video_geotag_source_alone_leaves_images_the_default( + self, monkeypatch, import_dir + ): + """Images were left with no source at all, and went missing""" + assert self._sources(monkeypatch, import_dir, [], ["gpx"]) == { + "image": self.DEFAULT, + "video": [SourceType.GPX], + } + + def test_both_options(self, monkeypatch, import_dir): + assert self._sources(monkeypatch, import_dir, ["exif"], ["gpx"]) == { + "image": [SourceType.EXIF], + "video": [SourceType.GPX, SourceType.EXIF], + } + + def test_video_process_geotags_the_frames_by_the_default( + self, monkeypatch, import_dir + ): + """video_process geotags the frames it samples, which are images only""" + assert self._sources( + monkeypatch, import_dir, [], ["native"], filetypes={types.FileType.IMAGE} + ) == {"image": self.DEFAULT, "video": []}