diff --git a/mapillary_tools/exif_write.py b/mapillary_tools/exif_write.py index cd90aed9..4cba4a09 100644 --- a/mapillary_tools/exif_write.py +++ b/mapillary_tools/exif_write.py @@ -124,12 +124,18 @@ def add_lat_lon(self, lat: float, lon: float) -> None: ) def add_altitude(self, altitude: float) -> None: - """Add altitude.""" + """ + Add altitude. The terms of an EXIF rational are 32-bit, which limits the + denominator of altitudes above 4293 m + """ ref = 0 if altitude > 0 else 1 + max_denominator = min( + 1_000_000, max(1, (2**32 - 1) // (math.ceil(math.fabs(altitude)) + 1)) + ) self._ef["GPS"][piexif.GPSIFD.GPSAltitude] = ( Fraction.from_float(math.fabs(altitude)) - .limit_denominator() + .limit_denominator(max_denominator) .as_integer_ratio() ) self._ef["GPS"][piexif.GPSIFD.GPSAltitudeRef] = ref diff --git a/mapillary_tools/geotag/geotag_images_from_nmea_file.py b/mapillary_tools/geotag/geotag_images_from_nmea_file.py index 7b9bb684..1b868cbe 100644 --- a/mapillary_tools/geotag/geotag_images_from_nmea_file.py +++ b/mapillary_tools/geotag/geotag_images_from_nmea_file.py @@ -56,10 +56,9 @@ def _read_nmea_points(source_path: Path) -> list[geo.Point]: 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 + # OSError: unreadable; ValueError: not text (UnicodeError), or malformed + # sentences and no fix; 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. " @@ -92,6 +91,10 @@ def get_lat_lon_time_from_nmea(nmea_file: Path) -> list[geo.Point]: A receiver may log a fix from several talkers, such as $GPGGA and $GNGGA, for the same time: only the first is read, so that no two points share a time. + + A malformed sentence, such as one whose checksum does not match, is + skipped with a warning. A file of which no fix can be read and that has + malformed sentences raises ValueError, as one that is not text does. """ # NMEA sentences are ASCII. Decode them as UTF-8 on every platform, so # that a file that is not text fails to read, not only where UTF-8 is the @@ -100,8 +103,42 @@ def get_lat_lon_time_from_nmea(nmea_file: Path) -> list[geo.Point]: lines = f.readlines() lines = [line.rstrip("\n\r") for line in lines] - messages = [pynmea2.parse(line) for line in lines if _SENTENCE_RE.search(line)] + messages = [] + # (line number, line, error) of each malformed sentence + malformed: list[tuple[int, str, ValueError]] = [] + for number, line in enumerate(lines, start=1): + if _SENTENCE_RE.search(line): + try: + messages.append(_parse_sentence(line)) + except ValueError as ex: + malformed.append((number, line, ex)) + + points = _read_fixes(messages, nmea_file) + + if malformed: + number, line, error = malformed[0] + if not points: + raise error + LOG.warning( + "Skipping %d of the %d RMC and GGA sentences of the NMEA file %s, which are malformed, the first on line %d (%s): %s", + len(malformed), + len(malformed) + len(messages), + nmea_file, + number, + error, + line, + ) + + return points + +def _read_fixes( + messages: list[pynmea2.NMEASentence], nmea_file: Path +) -> list[geo.Point]: + """ + The points of the fixes of the NMEA sentences read from **nmea_file**, as + get_lat_lon_time_from_nmea() describes + """ # The RMC sentences with a date, and the GGA sentences with a fix fixes = [ m @@ -148,6 +185,45 @@ def get_lat_lon_time_from_nmea(nmea_file: Path) -> list[geo.Point]: return points +def _parse_sentence(line: str) -> pynmea2.NMEASentence: + """ + Parse an NMEA sentence, raising ValueError if it is malformed: if it does + not parse or its checksum does not match, or if a field that a fix is + read from does not parse, which pynmea2 leaves a string + + >>> _parse_sentence("$GPGGA,0246,3735.30670,N,12220.25810,W,1,08,1.0,0.0,M,0.0,M,,*64") + Traceback (most recent call last): + ... + ValueError: malformed timestamp: '0246' + >>> _parse_sentence("$GPGGA,024636.00,3735.30670,N,12220.25810,W,1,08,1.0,0.0,M,0.0,M,,*00") + Traceback (most recent call last): + ... + ValueError: checksum does not match: 00 != 4F + """ + try: + message = pynmea2.parse(line) + except pynmea2.ParseError as ex: + # Its argument is (reason, sentence) + raise ValueError(ex.args[0][0]) from ex + + fields: dict[str, type] = {} + if isinstance(message, pynmea2.RMC): + fields = {"timestamp": datetime.time, "datestamp": datetime.date} + elif isinstance(message, pynmea2.GGA): + fields = {"timestamp": datetime.time, "gps_qual": int, "altitude": float} + for name, expected_type in fields.items(): + value = getattr(message, name) + if value is not None and not isinstance(value, expected_type): + raise ValueError(f"malformed {name}: {value!r}") + + if isinstance(message, pynmea2.GGA) and _has_fix(message): + # pynmea2 computes the position from its fields when it is read, + # raising ValueError if they are malformed + _position = (message.latitude, message.longitude) + + return message + + def _rmc_datetime(message: pynmea2.RMC) -> datetime.datetime: return datetime.datetime.combine(message.datestamp, message.timestamp) diff --git a/mapillary_tools/geotag/video_extractors/nmea.py b/mapillary_tools/geotag/video_extractors/nmea.py index 466d00f8..30d2e70e 100644 --- a/mapillary_tools/geotag/video_extractors/nmea.py +++ b/mapillary_tools/geotag/video_extractors/nmea.py @@ -30,10 +30,9 @@ class NMEAVideoExtractor(GPXVideoExtractor): def _parse_tracks(self) -> list[list[geo.Point]]: try: points = get_lat_lon_time_from_nmea(self.gpx_path) - # OSError: unreadable; ValueError: not text (UnicodeError), a - # malformed sentence (pynmea2.ParseError) or position; TypeError: a - # malformed time, which pynmea2 leaves a string - except (OSError, ValueError, TypeError) as ex: + # OSError: unreadable; ValueError: not text (UnicodeError), or + # malformed sentences and no fix + except (OSError, ValueError) as ex: raise exceptions.MapillaryInvalidGPXError( f"{self.gpx_path.name} is not a valid NMEA file. " "Replace it with a valid NMEA file." diff --git a/mapillary_tools/sample_video.py b/mapillary_tools/sample_video.py index 17a47840..b87525e7 100644 --- a/mapillary_tools/sample_video.py +++ b/mapillary_tools/sample_video.py @@ -5,6 +5,7 @@ from __future__ import annotations +import dataclasses import datetime import logging import os @@ -808,6 +809,10 @@ def _sample_single_video_by_interval( f"Unable to extract video start time from {video_path}" ) + # Located along the video's own GPS track, as distance samples are + track = _read_track_for_interval_samples(video_path, geotag_options) + interpolator = geo.Interpolator([track.points]) if track is not None else None + stream_specifier = _interval_stream_specifier(probe) # The stream sampled, whether it is the only one or not video_stream = probe.probe_video_with_max_resolution() @@ -835,9 +840,80 @@ def _sample_single_video_by_interval( exif_edit = ExifEdit(sample_paths[0]) exif_edit.add_date_time_original(timestamp) exif_edit.add_gps_datetime(timestamp) + if ( + track is not None + and interpolator is not None + and _within_track_time_range_buffered(track.points, video_time) + ): + _add_position(exif_edit, track, interpolator.interpolate(video_time)) exif_edit.write() +def _read_track_for_interval_samples( + video_path: Path, geotag_options: T.Sequence[SourceOption] +) -> types.VideoMetadata | None: + """ + The video's own GPS track, which its interval samples are located along + as distance samples are, or None to leave their positions to the + geotagging stage. That places them by their capture times along a GPX or + NMEA track, which interval sampling, unlike distance sampling, does not + place along the video: it stamps a video without a GPS clock from its + creation time, rather than from the start of the track + """ + if any(option.source in factory.EXTERNAL_GPS_SOURCES for option in geotag_options): + return None + + # Through the factory with the same sources as distance sampling, which + # reads the video's own GPS by default (see _parse_geotag_options()). One + # video is read in this process rather than in a pool of its own + video_metadatas = factory.process( + [video_path], + [ + dataclasses.replace(option, num_processes=0) + for option in geotag_options or [SourceOption(SourceType.NATIVE)] + ], + ) + assert len(video_metadatas) == 1, "expect 1 video metadata" + video_metadata = video_metadatas[0] + if isinstance(video_metadata, types.ErrorMetadata): + LOG.warning( + "The samples of %s are not geotagged, as no GPS track was read from it: %s", + video_path.name, + video_metadata.error, + ) + return None + + assert isinstance(video_metadata, types.VideoMetadata), ( + f"expect VideoMetadata but got {type(video_metadata).__name__}" + ) + if not video_metadata.points: + LOG.warning( + "The samples of %s are not geotagged, as it has no GPS points", + video_path.name, + ) + return None + + return video_metadata + + +def _add_position( + exif_edit: ExifEdit, video_metadata: types.VideoMetadata, point: geo.Point +) -> None: + """ + Write the position of a sample, and the camera of the video it was cut + from, as distance sampling writes them + """ + exif_edit.add_lat_lon(point.lat, point.lon) + if point.alt is not None: + exif_edit.add_altitude(point.alt) + if point.angle is not None: + exif_edit.add_direction(point.angle) + if video_metadata.make: + exif_edit.add_make(video_metadata.make) + if video_metadata.model: + exif_edit.add_model(video_metadata.model) + + def _within_track_time_range_buffered(points, t: float) -> bool: # apply 1ms buffer, which is MAPCaptureTime's precision start_point_time = points[0].time - 0.001 @@ -1131,13 +1207,5 @@ def _sample_single_video_by_distance( exif_edit = ExifEdit(sample_paths[0]) exif_edit.add_date_time_original(timestamp) exif_edit.add_gps_datetime(timestamp) - exif_edit.add_lat_lon(interp.lat, interp.lon) - if interp.alt is not None: - exif_edit.add_altitude(interp.alt) - if interp.angle is not None: - exif_edit.add_direction(interp.angle) - if video_metadata.make: - exif_edit.add_make(video_metadata.make) - if video_metadata.model: - exif_edit.add_model(video_metadata.model) + _add_position(exif_edit, video_metadata, interp) exif_edit.write() diff --git a/tests/integration/test_video_process.py b/tests/integration/test_video_process.py index d805887e..398ecb2c 100644 --- a/tests/integration/test_video_process.py +++ b/tests/integration/test_video_process.py @@ -18,6 +18,7 @@ from .fixtures import ( assert_contains_image_descs, + assert_same_image_descs, EXIFTOOL_EXECUTABLE, pytest_skip_if_not_exiftool_installed, pytest_skip_if_not_ffmpeg_installed, @@ -184,6 +185,44 @@ def test_video_process_by_distance_with_a_video_geotag_source_only( assert all("error" not in desc for desc in descs), descs +def test_video_process_by_interval_with_the_default_geotag_sources( + setup_data: py.path.local, +): + """ + Interval samples are located along the video's own GPS track, as distance + samples are, where the legacy source that geotags images from videos would + place them. They used to carry no position, and the default sources failed + every one of them + """ + pytest_skip_if_not_ffmpeg_installed() + + video_path = setup_data.join("gopro_data").join("max-360mode.mp4") + interval = ["--video_sample_interval=2", "--video_sample_distance=-1"] + + with tempfile.TemporaryDirectory() as dir: + descs = run_video_process_for_descs([*interval, str(video_path), dir]) + sample_dir = Path(dir, "max-360mode.mp4") + assert descs + assert all("error" not in desc for desc in descs), descs + + # sample_video and process, as video_process runs them + process_descs = run_process_for_descs([str(sample_dir)]) + assert_same_image_descs(process_descs, descs) + + with tempfile.TemporaryDirectory() as dir: + legacy_descs = run_video_process_for_descs( + [*interval, "--geotag_source=gopro_videos", str(video_path), dir] + ) + + by_name = {Path(d["filename"]).name: d for d in descs} + assert len(by_name) == len(legacy_descs) + for legacy in legacy_descs: + desc = by_name[Path(legacy["filename"]).name] + assert desc["MAPCaptureTime"] == legacy["MAPCaptureTime"] + assert desc["MAPLatitude"] == pytest.approx(legacy["MAPLatitude"], abs=1e-7) + assert desc["MAPLongitude"] == pytest.approx(legacy["MAPLongitude"], abs=1e-7) + + 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_exifedit.py b/tests/unit/test_exifedit.py index 926ebbbe..69ef72e4 100644 --- a/tests/unit/test_exifedit.py +++ b/tests/unit/test_exifedit.py @@ -171,6 +171,20 @@ def test_add_lat_lon(self): def test_add_altitude(self): add_altitude_general(self, EMPTY_EXIF_FILE_TEST) + def test_add_altitude_above_4293_m(self): + # Written with the 1e-6 precision of lower altitudes, the numerator of + # the rational would not fit in its 32 bits + test_altitude = 7112.573717404068 + + empty_exifedit = ExifEdit(EMPTY_EXIF_FILE) + empty_exifedit.add_altitude(test_altitude) + empty_exifedit.write(EMPTY_EXIF_FILE_TEST) + + exif_data = ExifRead(EMPTY_EXIF_FILE_TEST) + altitude = exif_data.extract_altitude() + assert altitude is not None + self.assertAlmostEqual(test_altitude, altitude, places=5) + def test_add_direction(self): add_direction_general(self, EMPTY_EXIF_FILE_TEST) diff --git a/tests/unit/test_geotag_images_from_track_file.py b/tests/unit/test_geotag_images_from_track_file.py index c1b5d1f9..1b3c604b 100644 --- a/tests/unit/test_geotag_images_from_track_file.py +++ b/tests/unit/test_geotag_images_from_track_file.py @@ -114,6 +114,16 @@ def test_a_missing_file_fails_every_image(self, tmp_path): GeotagImagesFromNMEAFile(tmp_path / "missing.nmea"), "NMEA file not found" ) + def test_a_malformed_sentence_is_skipped(self, tmp_path): + """It failed every image""" + nmea = tmp_path / "track.nmea" + nmea.write_text(NMEA_RMC + NMEA_GGA.replace("$GPGGA", "$GLGGA") + NMEA_GGA) + + geotag = GeotagImagesFromNMEAFile(nmea) + + assert geotag.track_error is None + assert len(geotag.points) == 1 + def test_a_valid_file_is_read(self, tmp_path): nmea = tmp_path / "track.nmea" nmea.write_text(NMEA_RMC + NMEA_GGA) diff --git a/tests/unit/test_nmea_parser.py b/tests/unit/test_nmea_parser.py index 610a2b68..e1ff108b 100644 --- a/tests/unit/test_nmea_parser.py +++ b/tests/unit/test_nmea_parser.py @@ -39,7 +39,9 @@ def _gga(time: str, quality: int = 1, talker="GP", lat="3735.30670") -> str: def _read(tmp_path: Path, *lines: str): path = tmp_path / "track.nmea" - path.write_text("".join(f"{line}\r\n" for line in lines)) + # Bytes, so that the line ends are CRLF on every platform: write_text() on + # Windows would write each "\r\n" as "\r\r\n", an empty line more + path.write_bytes("".join(f"{line}\r\n" for line in lines).encode("ascii")) return [ ( datetime.datetime.fromtimestamp(p.time, datetime.timezone.utc).isoformat(), @@ -89,22 +91,26 @@ def test_one_point_per_time(tmp_path): ) == [_fix("02:46:36"), _fix("02:46:37", lat=37.588435)] -def test_sentences_before_the_first_fix_are_skipped(tmp_path): +def test_sentences_before_the_first_fix_are_skipped(tmp_path, caplog): """ A log that started before its first fix was rejected whole, and gave positions at 0, 0 """ - assert _read( - tmp_path, - # void, with the date of a clock that is not set yet, and empty - _rmc("000001.00", date="060180", status="V"), - _sentence("GPRMC,,V,,,,,,,,,,N"), - # no fix, with and without a time - _sentence("GPGGA,024635.00,,,,,0,00,99.99,,,,,,"), - _sentence("GPGGA,,,,,,0,00,,,M,,M,,"), - _rmc("024636.00"), - _gga("024636.00"), - ) == [_fix("02:46:36")] + with caplog.at_level(logging.WARNING): + assert _read( + tmp_path, + # void, with the date of a clock that is not set yet, and empty + _rmc("000001.00", date="060180", status="V"), + _sentence("GPRMC,,V,,,,,,,,,,N"), + # no fix, with and without a time + _sentence("GPGGA,024635.00,,,,,0,00,99.99,,,,,,"), + _sentence("GPGGA,,,,,,0,00,,,M,,M,,"), + _rmc("024636.00"), + _gga("024636.00"), + ) == [_fix("02:46:36")] + + # They are not malformed + assert not caplog.text def test_a_void_rmc_gives_no_date(tmp_path): @@ -184,6 +190,65 @@ def test_a_log_of_another_day_appended(tmp_path): ] +def test_a_malformed_sentence_is_skipped_with_a_warning(tmp_path, caplog): + """One failed the whole file, and every image geotagged from it""" + bad_checksum = _gga("024636.50", talker="GL")[:-2] + "00" + with caplog.at_level(logging.WARNING): + assert _read( + tmp_path, + _rmc("024636.00"), + _gga("024636.00"), + bad_checksum, + _rmc("024637.00"), + _gga("024637.00"), + ) == [_fix("02:46:36"), _fix("02:46:37")] + + assert "Skipping 1 of the 5 RMC and GGA sentences of the NMEA file" in caplog.text + assert "on line 3 (checksum does not match: 00 != " in caplog.text + assert bad_checksum in caplog.text + + +@pytest.mark.parametrize( + "malformed", + [ + # pynmea2 leaves the fields it cannot parse strings + _gga("0246"), + _sentence("GPGGA,024636.50,3735.30670,N,12220.25810,W,x,08,1.0,12.5,M,0.0,M,,"), + _gga("024636.50").replace(",12.5,", ",12.x,"), + _rmc("024636.50", date="2212x2"), + # and fails to compute the position from them + _gga("024636.50", lat="37x5.30670"), + # junk after the sentence type + "$GPGGA,\x01\x02", + ], +) +def test_malformed_fields_are_skipped(tmp_path, caplog, malformed): + with caplog.at_level(logging.WARNING): + assert _read( + tmp_path, + _rmc("024636.00"), + _gga("024636.00"), + malformed, + _rmc("024637.00"), + _gga("024637.00"), + ) == [_fix("02:46:36"), _fix("02:46:37")] + + assert "Skipping 1 of the 5 RMC and GGA sentences" in caplog.text + + +@pytest.mark.parametrize( + "lines", + [ + [_rmc("024636.00")[:-2] + "00", _gga("024636.00")[:-2] + "00"], + [_rmc("024636.00"), _gga("0246")], + ], +) +def test_malformed_sentences_and_no_fix_fail_the_file(tmp_path, lines): + """As a file that is not NMEA does, rather than as one without a fix""" + with pytest.raises(ValueError): + _read(tmp_path, *lines) + + def test_gga_without_any_date_give_no_points(tmp_path, caplog): """They raised UnboundLocalError""" with caplog.at_level(logging.WARNING): diff --git a/tests/unit/test_nmea_video.py b/tests/unit/test_nmea_video.py index c4131b90..f208c952 100644 --- a/tests/unit/test_nmea_video.py +++ b/tests/unit/test_nmea_video.py @@ -162,6 +162,21 @@ def test_a_track_across_midnight_is_placed_as_from_a_gpx(self, tmp_path): for nmea_point, gpx_point in zip(from_nmea, from_gpx): assert nmea_point == pytest.approx(gpx_point, abs=1e-6) + def test_a_malformed_sentence_is_skipped(self, tmp_path): + """It failed the whole file""" + gpx = _write_gpx(tmp_path / "track.gpx", 6) + nmea = _write_nmea(tmp_path / "track.nmea", 6) + lines = nmea.read_text().splitlines() + lines.insert(5, lines[4][:-2] + "00") + nmea.write_text("\r\n".join(lines)) + + from_gpx = _points(GPXVideoExtractor(NO_GPS_VIDEO, gpx)) + from_nmea = _points(NMEAVideoExtractor(NO_GPS_VIDEO, nmea)) + + assert len(from_nmea) == len(from_gpx) == 7 + for nmea_point, gpx_point in zip(from_nmea, from_gpx): + assert nmea_point == pytest.approx(gpx_point, abs=1e-6) + def test_messages_name_the_nmea_file(self, tmp_path): nmea = _write_nmea(tmp_path / "track.nmea", seconds=60) diff --git a/tests/unit/test_sample_video.py b/tests/unit/test_sample_video.py index ef29de51..5aa857d5 100644 --- a/tests/unit/test_sample_video.py +++ b/tests/unit/test_sample_video.py @@ -413,6 +413,150 @@ def test_the_duration_ratio_paces_a_video_without_a_gps_clock( ) +class TestLocateIntervalSamples: + """ + Interval samples are located along the video's own GPS track, as distance + samples are, so that the geotagging stage finds their positions in them + """ + + GPS_START_TIME = datetime.datetime( + 2021, 8, 10, 6, 38, 6, tzinfo=datetime.timezone.utc + ) + + # The position of the image that MOCK_FFMPEG copies for each sample + MOCK_SAMPLE_LON_LAT = (14.498805555555554, 35.91251666666667) + + @classmethod + def _points(cls, seconds: int) -> list[geo.Point]: + # One point per second of video, recorded in real time + return [ + telemetry.GPSPoint( + time=float(i), + lat=40.0 + i * 0.001, + lon=-74.0, + alt=100.0 + i, + angle=None, + epoch_time=cls.GPS_START_TIME.timestamp() + i, + fix=None, + precision=None, + ground_speed=None, + ) + for i in range(seconds) + ] + + @staticmethod + def _video_metadata(video_path: Path, points: T.Sequence[geo.Point]): + return VideoMetadata( + filename=video_path, + filetype=FileType.GOPRO, + points=list(points), + make="GoPro", + model="HERO7 Black", + ) + + def _sample(self, tmpdir, monkeypatch, track_points, **kwargs) -> list[Path]: + clock_points = self._points(60) + monkeypatch.setattr( + sample_video, + "NativeVideoExtractor", + lambda video_path: mock.Mock( + extract=lambda: self._video_metadata(video_path, clock_points) + ), + ) + self.track_calls: list[list[SourceOption]] = [] + + def fake_process(paths, options): + self.track_calls.append(list(options)) + [video_path] = paths + if isinstance(track_points, Exception): + return [ + describe_error_metadata( + track_points, video_path, filetype=FileType.VIDEO + ) + ] + return [self._video_metadata(video_path, track_points)] + + monkeypatch.setattr(sample_video.factory, "process", fake_process) + + root = _PWD.joinpath("data/mock_sample_video") + sample_dir = tmpdir.mkdir("sampled_video_frames") + sample_video.sample_video( + root.joinpath("videos"), + Path(sample_dir), + video_sample_distance=-1, + video_sample_interval=2, + rerun=True, + **kwargs, + ) + return sorted(Path(s) for s in sample_dir.join("hello.mp4").listdir()) + + def test_located_along_the_video_track(self, tmpdir, setup_mock, monkeypatch): + samples = self._sample(tmpdir, monkeypatch, self._points(60)) + + _validate_interval(samples, self.GPS_START_TIME) + for idx, sample in enumerate(samples): + exif = exif_read.ExifRead(sample) + # At the video time of the sample, 2 seconds apart + lon_lat = exif.extract_lon_lat() + assert lon_lat is not None + assert lon_lat == pytest.approx((-74.0, 40.0 + 2 * idx * 0.001)) + assert exif.extract_altitude() == pytest.approx(100.0 + 2 * idx) + # The bearing of the track, northwards + assert exif.extract_direction() == pytest.approx(0.0) + assert exif.extract_make() == "GoPro" + assert exif.extract_model() == "HERO7 Black" + # By the sources distance sampling reads, in this process + [options] = self.track_calls + assert [(o.source, o.num_processes) for o in options] == [ + (SourceType.NATIVE, 0) + ] + + def test_samples_past_the_track_are_not_located( + self, tmpdir, setup_mock, monkeypatch + ): + # The track ends 20 seconds into the video + samples = self._sample(tmpdir, monkeypatch, self._points(21)) + + lon_lats = [exif_read.ExifRead(s).extract_lon_lat() for s in samples] + assert lon_lats[10] == pytest.approx((-74.0, 40.020)) + assert all(lon_lat == self.MOCK_SAMPLE_LON_LAT for lon_lat in lon_lats[11:]) + + def test_not_located_without_a_track(self, tmpdir, setup_mock, monkeypatch, caplog): + with caplog.at_level(logging.WARNING, logger=sample_video.LOG.name): + samples = self._sample( + tmpdir, + monkeypatch, + exceptions.MapillaryVideoGPSNotFoundError("No GPS data found"), + ) + + _validate_interval(samples, self.GPS_START_TIME) + assert all( + exif_read.ExifRead(s).extract_lon_lat() == self.MOCK_SAMPLE_LON_LAT + for s in samples + ) + assert ( + "The samples of hello.mp4 are not geotagged, as no GPS track was read from it: No GPS data found" + in caplog.text + ) + + @pytest.mark.parametrize("source", [SourceType.GPX, SourceType.NMEA]) + def test_not_located_with_a_track_file(self, monkeypatch, source): + """ + The geotagging stage places the samples along a GPX or NMEA track by + their capture times, which interval sampling does not place the track by + """ + process = mock.Mock() + monkeypatch.setattr(sample_video.factory, "process", process) + + track = sample_video._read_track_for_interval_samples( + Path("video.mp4"), + [SourceOption(source), SourceOption(SourceType.NATIVE)], + ) + + assert track is None + process.assert_not_called() + + class TestSampleNoisyTimelapseByInterval: """ GPS rejected as noise gives interval sampling no telemetry to read a clock