From 71217a646746008e1dc581f8bf337829a97cab07 Mon Sep 17 00:00:00 2001 From: Caglar Pir Date: Fri, 23 Oct 2026 11:47:30 +0200 Subject: [PATCH 1/2] Skip malformed NMEA sentences with a warning One malformed RMC or GGA sentence failed the whole file, as not a valid NMEA file, and since NMEA files are read from any talker, so did one from a talker that v0.14.7 did not read. A log of 222 sentences with one more, a $GLGGA sentence whose checksum does not match, failed all 113 images geotagged from it, where v0.14.7 geotagged 106 of them, the 7 others being duplicates of the previous image. A video geotagged from such a log failed too. v0.14.7 crashed on a malformed $GPGGA sentence. A sentence that does not parse, whose checksum does not match, or with a field a fix is read from that does not parse, is now skipped, and one warning per file says how many were skipped and quotes the first, with its line number and what is wrong with it. pynmea2 leaves a field it cannot parse a string: a malformed time failed the file, and a malformed altitude crashed process with a traceback, for images and videos alike. Void sentences and sentences without a fix are skipped as before, without a warning. A file of which no fix can be read and that has malformed sentences still fails as not a valid NMEA file. On that log the 106 images get the same positions, times, directions and altitudes as with v0.14.7 and as from the log without the bad sentence, and the 7 others are duplicates as before. --- .../geotag/geotag_images_from_nmea_file.py | 86 +++++++++++++++++- .../geotag/video_extractors/nmea.py | 7 +- .../test_geotag_images_from_track_file.py | 10 ++ tests/unit/test_nmea_parser.py | 91 ++++++++++++++++--- tests/unit/test_nmea_video.py | 15 +++ 5 files changed, 187 insertions(+), 22 deletions(-) diff --git a/mapillary_tools/geotag/geotag_images_from_nmea_file.py b/mapillary_tools/geotag/geotag_images_from_nmea_file.py index 7b9bb6844..1b868cbed 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 466d00f84..30d2e70e9 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/tests/unit/test_geotag_images_from_track_file.py b/tests/unit/test_geotag_images_from_track_file.py index c1b5d1f93..1b3c604be 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 610a2b68f..e1ff108bb 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 c4131b900..f208c9523 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) From cc717e37e22394b30ff5fe4b2b92946e9bc2996d Mon Sep 17 00:00:00 2001 From: Caglar Pir Date: Fri, 23 Oct 2026 13:06:44 +0200 Subject: [PATCH 2/2] Geotag interval samples by the video's own GPS, as distance samples are `video_process VIDEO OUT --video_sample_interval N --video_sample_distance -1` failed every sample with "Unable to extract GPS Longitude or GPS Latitude from the image", and so did sample_video followed by process. It did so on v0.13 and v0.14 too. Distance sampling writes the position of each sample into its EXIF, where the default geotag sources read it; interval sampling wrote only the capture time. Only the legacy gopro, camm and blackvue sources geotagged interval samples, by reading the video again in the geotagging stage. Locate each interval sample along the video's own GPS track at its video time, and write its position, altitude, direction and the make and model of the camera into its EXIF, as distance sampling does, reading the track by the same sources: the video's own GPS unless --geotag_source or --video_geotag_source says otherwise. The capture times are stamped by the video's GPS clock as before. A sample before the track starts or after it ends gets no position. A video whose GPS cannot be read, as it has none or it is rejected as noise, is named in a warning, and its samples fail as before. With a GPX or NMEA source, nothing changes: the geotagging stage places the samples along that track by their capture times, which for a video without a GPS clock come from its creation time, not from where GPX sync would place the track. An altitude above 4293 m, which a GoPro with poor GPS can report, failed the sample: written to the 1e-6 m, the numerator of its EXIF rational did not fit in 32 bits. Such altitudes are written to the precision that fits. That also fixes distance samples, which failed the same way. On a GoPro HERO7, a GoPro MAX (a normal video and a timelapse), a PANOX V2, a LabPano Pilot One, a BlackVue 900S, a Viofo A129 and a Ricoh Theta X, the samples now lie within 1 cm of the video's track at their video time, stamped at even steps of the clock (11.426 s for 10 s of the LabPano). They get the positions, capture times, altitudes and cameras that the legacy sources give them, within 1 cm on the LabPano, and sample_video followed by process describes them as video_process does. --geotag_source exif and native, which read the EXIF, now geotag them too. With a legacy source the descriptions do not change, but for the file sizes of the samples, which now hold their positions; nor do they with a GPX (also with --interpolation_use_gpx_start_time, and with native after it), with --video_geotag_source gpx, for distance sampling, for a video without GPS and for a GoPro MAX 2 whose GPS is noise. --- mapillary_tools/exif_write.py | 10 +- mapillary_tools/sample_video.py | 86 ++++++++++++-- tests/integration/test_video_process.py | 39 +++++++ tests/unit/test_exifedit.py | 14 +++ tests/unit/test_sample_video.py | 144 ++++++++++++++++++++++++ 5 files changed, 282 insertions(+), 11 deletions(-) diff --git a/mapillary_tools/exif_write.py b/mapillary_tools/exif_write.py index cd90aed97..4cba4a095 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/sample_video.py b/mapillary_tools/sample_video.py index 17a47840c..b87525e7f 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 d805887e5..398ecb2c9 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 926ebbbeb..69ef72e49 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_sample_video.py b/tests/unit/test_sample_video.py index ef29de517..5aa857d5c 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