diff --git a/mapillary_tools/geotag/factory.py b/mapillary_tools/geotag/factory.py index 4d1eeea0..522a25fb 100644 --- a/mapillary_tools/geotag/factory.py +++ b/mapillary_tools/geotag/factory.py @@ -22,7 +22,13 @@ geotag_videos_from_gpx, geotag_videos_from_video, ) -from .options import InterpolationOption, SOURCE_TYPE_ALIAS, SourceOption, SourceType +from .options import ( + InterpolationOption, + SOURCE_TYPE_ALIAS, + SourceOption, + SourcePathOption, + SourceType, +) LOG = logging.getLogger(__name__) @@ -58,6 +64,51 @@ def parse_source_option(source: str) -> list[SourceOption]: return [SourceOption(SourceType(SOURCE_TYPE_ALIAS.get(s, s))) for s in sources] +def parse_source_options( + geotag_source: T.Sequence[str], + video_geotag_source: T.Sequence[str], + geotag_source_path: Path | None, +) -> list[SourceOption]: + """ + Turn the raw --geotag_source / --video_geotag_source / --geotag_source_path + arguments into the option list that process() consumes. + """ + parsed_options: list[SourceOption] = [] + + if video_geotag_source and geotag_source: + LOG.warning( + "Video source options will be processed BEFORE the generic source options" + ) + + for s in video_geotag_source: + for video_option in parse_source_option(s): + video_option.filetypes = types.combine_filetype_filters( + video_option.filetypes, {types.FileType.VIDEO} + ) + parsed_options.append(video_option) + + for s in geotag_source: + parsed_options.extend(parse_source_option(s)) + + if geotag_source_path is not None: + for parsed_option in parsed_options: + if parsed_option.source_path is None: + parsed_option.source_path = SourcePathOption( + source_path=Path(geotag_source_path) + ) + else: + source_path_option = parsed_option.source_path + if source_path_option.source_path is None: + source_path_option.source_path = Path(geotag_source_path) + else: + LOG.warning( + "The option --geotag_source_path is ignored for source %s", + parsed_option, + ) + + return parsed_options + + def process( # Collection: ABC for sized iterable container classes paths: T.Iterable[Path], diff --git a/mapillary_tools/process_geotag_properties.py b/mapillary_tools/process_geotag_properties.py index 41d8e8a2..73ee9d45 100644 --- a/mapillary_tools/process_geotag_properties.py +++ b/mapillary_tools/process_geotag_properties.py @@ -14,11 +14,9 @@ from tqdm import tqdm from . import constants, exceptions, exif_write, types, utils -from .geotag.factory import parse_source_option, process +from .geotag.factory import parse_source_options, process from .geotag.options import ( InterpolationOption, - SourceOption, - SourcePathOption, SourceType, ) from .serializer.description import ( @@ -44,47 +42,6 @@ def _normalize_import_paths(import_path: Path | T.Sequence[Path]) -> T.Sequence[ return import_paths -def _parse_source_options( - geotag_source: list[str], - video_geotag_source: list[str], - geotag_source_path: Path | None, -) -> list[SourceOption]: - parsed_options: list[SourceOption] = [] - - if video_geotag_source and geotag_source: - LOG.warning( - "Video source options will be processed BEFORE the generic source options" - ) - - for s in video_geotag_source: - for video_option in parse_source_option(s): - video_option.filetypes = types.combine_filetype_filters( - video_option.filetypes, {types.FileType.VIDEO} - ) - parsed_options.append(video_option) - - for s in geotag_source: - parsed_options.extend(parse_source_option(s)) - - if geotag_source_path is not None: - for parsed_option in parsed_options: - if parsed_option.source_path is None: - parsed_option.source_path = SourcePathOption( - source_path=Path(geotag_source_path) - ) - else: - source_path_option = parsed_option.source_path - if source_path_option.source_path is None: - source_path_option.source_path = Path(geotag_source_path) - else: - LOG.warning( - "The option --geotag_source_path is ignored for source %s", - parsed_option, - ) - - return parsed_options - - def process_geotag_properties( import_path: Path | T.Sequence[Path], filetypes: set[types.FileType] | None, @@ -115,7 +72,7 @@ def process_geotag_properties( if not geotag_source and not video_geotag_source: geotag_source = [*DEFAULT_GEOTAG_SOURCE_OPTIONS] - options = _parse_source_options( + options = parse_source_options( geotag_source=geotag_source or [], video_geotag_source=video_geotag_source or [], geotag_source_path=geotag_source_path, diff --git a/mapillary_tools/sample_video.py b/mapillary_tools/sample_video.py index 1e3e7764..ab864d6f 100644 --- a/mapillary_tools/sample_video.py +++ b/mapillary_tools/sample_video.py @@ -16,12 +16,30 @@ from . import constants, exceptions, ffmpeg as ffmpeglib, geo, types, utils from .exif_write import ExifEdit -from .geotag import geotag_videos_from_video +from .geotag import factory +from .geotag.options import SourceOption, SourceType from .mp4 import mp4_sample_parser from .serializer.description import parse_capture_time LOG = logging.getLogger(__name__) +# Sampling errors are suppressed by --skip_sample_errors, not by +# --skip_process_errors, which governs the later geotagging stage. Say so in +# the message: the two flags are easy to reach for the wrong one. +_SKIP_HINT = "To skip these errors, specify --skip_sample_errors" + + +def _sampling_error(message: str) -> exceptions.MapillaryVideoError: + """ + Build an error for a failed sample, naming the flag that skips it. + + Everything raised out of the per-video body of sample_video() is + suppressible by --skip_sample_errors, so every one of those messages should + say so. Going through here rather than appending the hint at each raise + keeps that true of raises added later. + """ + return exceptions.MapillaryVideoError(f"{message}. {_SKIP_HINT}") + def _normalize_path( video_import_path: Path, skip_subfolders: bool @@ -49,6 +67,36 @@ def xor(a: bool, b: bool): return bool(a) ^ bool(b) +def _parse_geotag_options( + geotag_source: list[str] | None, + geotag_source_path: Path | None, + video_geotag_source: list[str] | None, + video_import_path: Path, +) -> list[SourceOption]: + """ + Resolve which sources distance sampling may take its positions from. + + Defaults to the video's own telemetry rather than to + DEFAULT_GEOTAG_SOURCE_OPTIONS: the default chain continues on to + exiftool_runtime, which reads the same telemetry through a parser that + cannot see every field the noise filter rejects on, so routing sampling + through it would accept tracks the native parser refuses. + """ + if not geotag_source and not video_geotag_source: + return [SourceOption(SourceType.NATIVE)] + + # Mirrors process_geotag_properties(): a sidecar is looked for next to the + # video when no explicit path is given + if geotag_source_path is None: + geotag_source_path = video_import_path + + return factory.parse_source_options( + geotag_source=geotag_source or [], + video_geotag_source=video_geotag_source or [], + geotag_source_path=geotag_source_path, + ) + + def sample_video( video_import_path: Path, import_path: Path, @@ -60,9 +108,18 @@ def sample_video( video_start_time: str | None = None, skip_sample_errors: bool = False, rerun: bool = False, + # Absent when called from the sample_video command, which does not register + # the process command's arguments + geotag_source: list[str] | None = None, + geotag_source_path: Path | None = None, + video_geotag_source: list[str] | None = None, ) -> None: video_dir, video_list = _normalize_path(video_import_path, skip_subfolders) + geotag_options = _parse_geotag_options( + geotag_source, geotag_source_path, video_geotag_source, video_import_path + ) + if not xor(0 <= video_sample_distance, 0 < video_sample_interval): raise exceptions.MapillaryBadParameterError( f"Expect either non-negative video_sample_distance or positive video_sample_interval but got {video_sample_distance} and {video_sample_interval} respectively" @@ -112,6 +169,7 @@ def sample_video( sample_dir, sample_distance=video_sample_distance, start_time=video_start_time_dt, + geotag_options=geotag_options, ) else: assert 0 < video_sample_interval, ( @@ -193,7 +251,7 @@ def _sample_single_video_by_interval( ffmpeg.probe_format_and_streams(video_path) ).probe_video_start_time() if start_time is None: - raise exceptions.MapillaryVideoError( + raise _sampling_error( f"Unable to extract video start time from {video_path}" ) @@ -281,7 +339,11 @@ def _sample_single_video_by_distance( sample_dir: Path, sample_distance: float, start_time: datetime.datetime | None = None, + geotag_options: T.Sequence[SourceOption] | None = None, ) -> None: + if geotag_options is None: + geotag_options = [SourceOption(SourceType.NATIVE)] + ffmpeg = ffmpeglib.FFMPEG(constants.FFMPEG_PATH, constants.FFPROBE_PATH) probe = ffmpeglib.Probe(ffmpeg.probe_format_and_streams(video_path)) @@ -289,28 +351,47 @@ def _sample_single_video_by_distance( if start_time is None: start_time = probe.probe_video_start_time() if start_time is None: - raise exceptions.MapillaryVideoError( + raise _sampling_error( f"Unable to extract video start time from {video_path}" ) LOG.info("Extracting video metdata") - video_metadatas = geotag_videos_from_video.GeotagVideosFromVideo().to_description( - [video_path] - ) + # Go through the factory rather than reading the video's own telemetry + # directly, so that --geotag_source is honoured here as well: a GPX is the + # documented answer for a camera whose embedded GPS is unusable, and + # distance sampling needs positions just as much as geotagging does. + video_metadatas = factory.process([video_path], geotag_options) assert len(video_metadatas) == 1, "expect 1 video metadata" video_metadata = video_metadatas[0] + + # Distance sampling needs positions to decide which frames to cut, so + # failing to read them is a failed sample, not something to carry on past. + # Warning and returning left the caller with a success exit code, an empty + # (or missing) sample directory and nothing to upload. sample_video() + # already funnels these through --skip_sample_errors for callers who do + # want to tolerate them. if isinstance(video_metadata, types.ErrorMetadata): - LOG.warning(str(video_metadata.error)) - return - assert video_metadata.points, "expect non-empty points" + raise _sampling_error( + f"Unable to sample {video_path} by distance: {video_metadata.error}" + ) from video_metadata.error + + # Only a video path was passed in, so only video metadata can come back + assert isinstance(video_metadata, types.VideoMetadata), ( + f"expect VideoMetadata but got {type(video_metadata).__name__}" + ) + + if not video_metadata.points: + raise _sampling_error( + f"Unable to sample {video_path} by distance: no GPS points found" + ) + LOG.info("Found total %d GPS points", len(video_metadata.points)) # find the video stream with maximum resolution video_stream = probe.probe_video_with_max_resolution() if not video_stream: - LOG.warning("no video streams found from ffprobe") - return + raise _sampling_error(f"No video streams found in {video_path} by ffprobe") LOG.info("Extracting video samples") video_stream_idx = video_stream["index"] @@ -333,7 +414,7 @@ def _sample_single_video_by_distance( wip_dir, video_path, selected_stream_specifiers=[str(video_stream_idx)] ) if len(frame_samples) != len(sorted_sample_indices): - raise exceptions.MapillaryVideoError( + raise _sampling_error( f"Expect {len(sorted_sample_indices)} samples but extracted {len(frame_samples)} samples" ) for idx, (frame_idx_1based, sample_paths) in enumerate(frame_samples): @@ -341,7 +422,7 @@ def _sample_single_video_by_distance( "Expect 1 sample path at {frame_idx_1based} but got {sample_paths}" ) if idx + 1 != frame_idx_1based: - raise exceptions.MapillaryVideoError( + raise _sampling_error( f"Expect {sample_paths[0]} to be {idx + 1}th sample but got {frame_idx_1based}" ) diff --git a/tests/integration/test_video_process.py b/tests/integration/test_video_process.py index a2a2a05e..aff97e68 100644 --- a/tests/integration/test_video_process.py +++ b/tests/integration/test_video_process.py @@ -31,13 +31,24 @@ def test_sample_video_relpath(): pytest_skip_if_not_ffmpeg_installed() + # hero8.mp4's embedded GPS is filtered out entirely as noise, so it cannot + # be sampled by distance. This test is about relative paths, not about that, + # hence --skip_sample_errors. with tempfile.TemporaryDirectory() as dir: - run_sample_video(["--rerun", "tests/data/gopro_data/hero8.mp4", str(dir)]) + run_sample_video( + [ + "--rerun", + "--skip_sample_errors", + "tests/data/gopro_data/hero8.mp4", + str(dir), + ] + ) with tempfile.TemporaryDirectory() as dir: run_sample_video( [ "--rerun", + "--skip_sample_errors", "--video_start_time", "2021_10_10_10_10_10_123", "tests/data", @@ -46,6 +57,37 @@ def test_sample_video_relpath(): ) +def test_sample_video_by_distance_without_usable_gps(setup_data: py.path.local): + """ + A video whose GPS cannot be read is not silently sampled into nothing. + + hero8.mp4 has 32 embedded GPS points and remove_noisy_points() drops all of + them, so distance sampling has no positions to choose frames with. It used + to log a warning, write no frames and still exit 0, which left the caller a + success code and an empty import directory. + """ + pytest_skip_if_not_ffmpeg_installed() + + video_path = setup_data.join("gopro_data").join("hero8.mp4") + + with tempfile.TemporaryDirectory() as dir: + with pytest.raises(subprocess.CalledProcessError) as ex: + run_sample_video(["--video_sample_distance=6", "--rerun", video_path, dir]) + assert 7 == ex.value.returncode, ex.value.stderr + assert not list(Path(dir).iterdir()) + + with tempfile.TemporaryDirectory() as dir: + run_sample_video( + [ + "--video_sample_distance=6", + "--skip_sample_errors", + "--rerun", + video_path, + dir, + ] + ) + + def test_sample_video_without_video_time(setup_data: py.path.local): pytest_skip_if_not_ffmpeg_installed() @@ -143,6 +185,9 @@ def test_video_process_sample_with_multiple_distances(setup_data: py.path.local) [ "--video_sample_distance", str(distance), + # gopro_data also holds hero8.mp4, whose GPS is all noise and so + # cannot be distance-sampled; this test is about max-360mode.mp4 + "--skip_sample_errors", "--rerun", str(video_dir), str(video_dir.join("my_samples")), @@ -167,6 +212,8 @@ def test_video_process_sample_with_distance(setup_data: py.path.local): descs = run_video_process_for_descs( [ *options, + # see test_video_process_sample_with_multiple_distances + "--skip_sample_errors", str(video_dir), str(video_dir.join("my_samples")), ] diff --git a/tests/unit/test_sample_video.py b/tests/unit/test_sample_video.py index 0743eeb4..5441302a 100644 --- a/tests/unit/test_sample_video.py +++ b/tests/unit/test_sample_video.py @@ -22,9 +22,10 @@ geo, sample_video, ) +from mapillary_tools.geotag.options import SourceType from mapillary_tools.mp4 import mp4_sample_parser from mapillary_tools.serializer import description -from mapillary_tools.types import FileType, VideoMetadata +from mapillary_tools.types import describe_error_metadata, FileType, VideoMetadata _PWD = Path(os.path.dirname(os.path.abspath(__file__))) @@ -334,6 +335,120 @@ def test_empty_samples(self) -> None: assert len(result) == 0 +# --------------------------------------------------------------------------- +# A failed distance sample must not look like a successful one +# --------------------------------------------------------------------------- + + +class TestDistanceSamplingFailsLoudly: + """ + Distance sampling used to warn and return when it could not read the GPS, + which left the caller a success exit code, no sample directory and nothing + to upload -- `video_process` went on to geotag zero files and printed an + empty summary. Reported for a GoPro whose embedded GPS is all noise, where + an attached GPX made no difference. + """ + + VIDEO = _PWD.joinpath("data/mock_sample_video/videos/hello.mp4") + + @pytest.fixture + def unreadable_gps(self, monkeypatch): + """Every GPS read fails the way a fully filtered noisy track does.""" + error = exceptions.MapillaryGPSNoiseError("GPS is too noisy") + monkeypatch.setattr( + sample_video.factory, + "process", + lambda paths, options: [ + describe_error_metadata( + error, filename=paths[0], filetype=FileType.GOPRO + ) + ], + ) + return error + + def _sample(self, tmpdir: py.path.local, **kwargs): + return sample_video.sample_video( + self.VIDEO, Path(tmpdir), video_sample_distance=2, rerun=True, **kwargs + ) + + def test_it_raises_instead_of_returning(self, tmpdir, setup_mock, unreadable_gps): + with pytest.raises(exceptions.MapillaryVideoError) as excinfo: + self._sample(tmpdir) + + # the underlying reason has to survive into the message the user sees + assert "GPS is too noisy" in str(excinfo.value) + assert excinfo.value.__cause__ is unreadable_gps + + def test_it_names_the_flag_that_suppresses_it( + self, tmpdir, setup_mock, unreadable_gps + ): + """ + --skip_process_errors governs the later geotagging stage and does not + cover sampling, so the message has to name --skip_sample_errors or + users reach for the wrong flag. + """ + with pytest.raises(exceptions.MapillaryVideoError) as excinfo: + self._sample(tmpdir) + + assert "--skip_sample_errors" in str(excinfo.value) + + def test_the_exit_code_is_a_clean_one(self, tmpdir, setup_mock, unreadable_gps): + """Not a MapillaryUserError means a traceback instead of an exit code.""" + with pytest.raises(exceptions.MapillaryUserError) as excinfo: + self._sample(tmpdir) + + assert excinfo.value.exit_code == 7 + + def test_nothing_is_left_behind(self, tmpdir, setup_mock, unreadable_gps): + with pytest.raises(exceptions.MapillaryVideoError): + self._sample(tmpdir) + + assert not Path(tmpdir).joinpath(self.VIDEO.name).exists() + + def test_skip_sample_errors_still_tolerates_it( + self, tmpdir, setup_mock, unreadable_gps + ): + """The opt-out that several existing callers rely on.""" + self._sample(tmpdir, skip_sample_errors=True) + + def test_an_empty_track_is_also_an_error(self, tmpdir, setup_mock, monkeypatch): + """Previously an assert, so it vanished under `python -O`.""" + monkeypatch.setattr( + sample_video.factory, + "process", + lambda paths, options: [ + VideoMetadata( + filename=paths[0], filesize=0, filetype=FileType.GOPRO, points=[] + ) + ], + ) + + with pytest.raises(exceptions.MapillaryVideoError): + self._sample(tmpdir) + + def test_a_missing_video_stream_is_also_an_error( + self, tmpdir, setup_mock, monkeypatch + ): + monkeypatch.setattr( + sample_video.factory, + "process", + lambda paths, options: [ + VideoMetadata( + filename=paths[0], + filesize=0, + filetype=FileType.GOPRO, + points=_make_gps_points(3, time_step=1.0), + ) + ], + ) + monkeypatch.setattr( + ffmpeglib.Probe, "probe_video_with_max_resolution", lambda _self: None + ) + + with pytest.raises(exceptions.MapillaryVideoError): + self._sample(tmpdir) + + # --------------------------------------------------------------------------- # sample_video() parameter validation & rerun # --------------------------------------------------------------------------- @@ -475,11 +590,9 @@ def fake_extract_frames( mock_ffmpeg_class, ) - mock_geotag_instance = mock.MagicMock() - mock_geotag_instance.to_description.return_value = [video_metadata] patches["geotag_cls"] = mock.patch( - "mapillary_tools.sample_video.geotag_videos_from_video.GeotagVideosFromVideo", - return_value=mock_geotag_instance, + "mapillary_tools.sample_video.factory.process", + return_value=[video_metadata], ) patches["moov_parse"] = mock.patch.object( @@ -608,3 +721,172 @@ def test_exif_lat_lon_written(self, tmp_path: Path) -> None: # First GPS point is at (40.0, -74.0) assert abs(lat - 40.0) < 0.01 assert abs(lon - (-74.0)) < 0.01 + + +class TestEverySuppressibleErrorNamesTheFlag: + """ + Every error raised out of the per-video body of sample_video() is + suppressed by --skip_sample_errors, so every one of those messages has to + name it. Half of them naming it is worse than none: a user who hits one of + the silent ones reaches for --skip_process_errors, which governs the later + geotagging stage and leaves the run failing with the same message. + """ + + VIDEO = _PWD.joinpath("data/mock_sample_video/videos/hello.mp4") + + @pytest.fixture + def no_start_time(self, monkeypatch): + monkeypatch.setattr( + ffmpeglib.Probe, "probe_video_start_time", lambda _self: None + ) + + @pytest.mark.parametrize( + "kwargs", + [ + pytest.param({"video_sample_distance": 2}, id="by_distance"), + pytest.param( + {"video_sample_distance": -1, "video_sample_interval": 2}, + id="by_interval", + ), + ], + ) + def test_unreadable_start_time_names_the_flag( + self, tmpdir, setup_mock, no_start_time, kwargs + ): + with pytest.raises(exceptions.MapillaryVideoError) as excinfo: + sample_video.sample_video(self.VIDEO, Path(tmpdir), rerun=True, **kwargs) + + assert "Unable to extract video start time" in str(excinfo.value) + assert "--skip_sample_errors" in str(excinfo.value) + + def test_the_helper_appends_it_once(self): + error = sample_video._sampling_error("something went wrong") + + assert str(error) == ( + "something went wrong. To skip these errors, specify --skip_sample_errors" + ) + assert isinstance(error, exceptions.MapillaryUserError) + + def test_a_missing_ffmpeg_does_not_claim_to_be_skippable( + self, tmpdir, setup_mock, monkeypatch + ): + """ + FFmpegNotFoundError is re-raised by its own handler before the skip + check, so --skip_sample_errors does not suppress it and its message + must not offer the flag. + """ + + def boom(*args, **kwargs): + raise ffmpeglib.FFmpegNotFoundError("ffmpeg not found") + + monkeypatch.setattr(MOCK_FFMPEG, "extract_frames_by_interval", boom) + + with pytest.raises(exceptions.MapillaryFFmpegNotFoundError) as excinfo: + sample_video.sample_video( + self.VIDEO, + Path(tmpdir), + video_sample_distance=-1, + video_sample_interval=2, + rerun=True, + skip_sample_errors=True, + ) + + assert "--skip_sample_errors" not in str(excinfo.value) + + +# --------------------------------------------------------------------------- +# Distance sampling honours --geotag_source +# --------------------------------------------------------------------------- + + +class TestGeotagSourceReachesDistanceSampling: + """ + Distance sampling read the video's own telemetry directly, bypassing the + geotag factory, so --geotag_source was ignored: a GPX attached to rescue a + camera whose embedded GPS is unusable never got a chance to supply the + positions the sampler picks frames with. + """ + + def test_no_sources_requested_means_native_only(self): + """ + Not DEFAULT_GEOTAG_SOURCE_OPTIONS: that chain continues on to + exiftool_runtime, whose parser cannot see every field the noise filter + rejects on, so it would accept tracks the native parser refuses. + """ + options = sample_video._parse_geotag_options( + None, None, None, Path("/data/v.mp4") + ) + + assert [option.source for option in options] == [SourceType.NATIVE] + + def test_requested_source_is_passed_through(self): + options = sample_video._parse_geotag_options( + ["gpx"], Path("/data/track.gpx"), None, Path("/data/v.mp4") + ) + + assert [option.source for option in options] == [SourceType.GPX] + assert options[0].source_path is not None + assert options[0].source_path.source_path == Path("/data/track.gpx") + + def test_video_geotag_source_is_honoured_too(self): + options = sample_video._parse_geotag_options( + None, Path("/data/track.gpx"), ["gpx"], Path("/data/v.mp4") + ) + + assert [option.source for option in options] == [SourceType.GPX] + + def test_sidecar_is_looked_for_beside_the_video(self): + """Mirrors process_geotag_properties() when no explicit path is given.""" + options = sample_video._parse_geotag_options( + ["gpx"], None, None, Path("/data/v.mp4") + ) + + assert options[0].source_path is not None + assert options[0].source_path.source_path == Path("/data/v.mp4") + + def test_chained_sources_are_preserved(self): + options = sample_video._parse_geotag_options( + ["native", "gpx"], Path("/data/track.gpx"), None, Path("/data/v.mp4") + ) + + assert [option.source for option in options] == [ + SourceType.NATIVE, + SourceType.GPX, + ] + + def test_the_sampler_asks_the_factory_for_them( + self, tmpdir, setup_mock, monkeypatch + ): + """The options reach factory.process() rather than being dropped.""" + seen: list = [] + points = _make_gps_points(4, time_step=1.0) + + def fake_process(paths, options): + seen.append(list(options)) + return [ + VideoMetadata( + filename=paths[0], + filesize=1, + filetype=FileType.GOPRO, + points=points, + ) + ] + + monkeypatch.setattr(sample_video.factory, "process", fake_process) + # stop after the metadata is read; the rest needs real ffmpeg + monkeypatch.setattr( + ffmpeglib.Probe, "probe_video_with_max_resolution", lambda _self: None + ) + + with pytest.raises(exceptions.MapillaryVideoError): + sample_video.sample_video( + _PWD.joinpath("data/mock_sample_video/videos/hello.mp4"), + Path(tmpdir), + video_sample_distance=2, + rerun=True, + geotag_source=["gpx"], + geotag_source_path=Path("/data/track.gpx"), + ) + + assert len(seen) == 1 + assert [option.source for option in seen[0]] == [SourceType.GPX]