diff --git a/README.md b/README.md index d791e4690..a932aca25 100644 --- a/README.md +++ b/README.md @@ -462,6 +462,8 @@ with a GPX file called `foo.gpx`. ```sh mapillary_tools process --video_geotag_source gpx VIDEO_DIR/ ``` +NMEA files work the same way with `--video_geotag_source nmea`, which associates `foo.mp4` with `foo.nmea`. What +follows about GPX tracks applies to NMEA tracks too. #### Insta360 stitched videos diff --git a/mapillary_tools/geotag/factory.py b/mapillary_tools/geotag/factory.py index cb6ede822..2dde06fef 100644 --- a/mapillary_tools/geotag/factory.py +++ b/mapillary_tools/geotag/factory.py @@ -22,6 +22,7 @@ geotag_images_from_video, geotag_videos_from_exiftool, geotag_videos_from_gpx, + geotag_videos_from_nmea, geotag_videos_from_video, ) from .options import ( @@ -38,8 +39,7 @@ _ALL_VIDEO_TYPES = {types.FileType.VIDEO, *types.NATIVE_VIDEO_FILETYPES} # Sources that read a GPS track from outside the video file, as opposed to -# re-reading the telemetry embedded in it. (NMEA cannot geotag videos yet: it -# skips them, and they keep the error they had) +# re-reading the telemetry embedded in it EXTERNAL_GPS_SOURCES = frozenset({SourceType.GPX, SourceType.NMEA}) @@ -146,6 +146,8 @@ def can_geotag_videos(option: SourceOption) -> bool: Examples: >>> can_geotag_videos(SourceOption(SourceType.GPX)) True + >>> can_geotag_videos(SourceOption(SourceType.NMEA)) + True >>> can_geotag_videos(SourceOption(SourceType.EXIF)) False >>> can_geotag_videos( @@ -478,8 +480,9 @@ def _build_video_geotag(option: SourceOption) -> base.GeotagVideosFromGeneric | 'GeotagVideosFromGPX' >>> opt = SourceOption(SourceType.NMEA, source_path=Path("/tmp/test.nmea")) - >>> _build_video_geotag(opt) is None - True + >>> geotagger = _build_video_geotag(opt) + >>> geotagger.__class__.__name__ + 'GeotagVideosFromNMEA' >>> opt = SourceOption(SourceType.EXIF) >>> _build_video_geotag(opt) is None @@ -523,8 +526,12 @@ def _build_video_geotag(option: SourceOption) -> base.GeotagVideosFromGeneric | ) elif option.source is SourceType.NMEA: - # TODO: geotag videos from NMEA - return None + return geotag_videos_from_nmea.GeotagVideosFromNMEA( + source_path=option.source_path, + num_processes=option.num_processes, + video_sync=option.video_sync, + interpolation=option.interpolation, + ) elif option.source is SourceType.EXIF: # Legacy image-specific geotag types diff --git a/mapillary_tools/geotag/geotag_images_from_nmea_file.py b/mapillary_tools/geotag/geotag_images_from_nmea_file.py index cb4ef27aa..d0e1b07ee 100644 --- a/mapillary_tools/geotag/geotag_images_from_nmea_file.py +++ b/mapillary_tools/geotag/geotag_images_from_nmea_file.py @@ -66,7 +66,10 @@ def _read_nmea_points(source_path: Path) -> list[geo.Point]: def get_lat_lon_time_from_nmea(nmea_file: Path) -> list[geo.Point]: - with nmea_file.open("r") as f: + # 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 + # default encoding + with nmea_file.open("r", encoding="utf-8") as f: lines = f.readlines() lines = [line.rstrip("\n\r") for line in lines] diff --git a/mapillary_tools/geotag/geotag_videos_from_gpx.py b/mapillary_tools/geotag/geotag_videos_from_gpx.py index 8e0acd4e6..2f34418d9 100644 --- a/mapillary_tools/geotag/geotag_videos_from_gpx.py +++ b/mapillary_tools/geotag/geotag_videos_from_gpx.py @@ -25,6 +25,10 @@ class GeotagVideosFromGPX(GeotagVideosFromGeneric): + # Reads the track file of each video, and where to find it by default + extractor_class: type[GPXVideoExtractor] = GPXVideoExtractor + default_pattern = "%g.gpx" + def __init__( self, source_path: options.SourcePathOption | None = None, @@ -34,7 +38,7 @@ def __init__( ): super().__init__(num_processes=num_processes) if source_path is None: - source_path = options.SourcePathOption(pattern="%g.gpx") + source_path = options.SourcePathOption(pattern=self.default_pattern) self.source_path = source_path self.video_sync = video_sync self.interpolation = interpolation @@ -48,7 +52,7 @@ def _generate_video_extractors( source_path = self.source_path.resolve(video_path) if source_path.is_file(): results.append( - GPXVideoExtractor( + self.extractor_class( video_path, source_path, video_sync=self.video_sync, @@ -62,7 +66,7 @@ def _generate_video_extractors( results.append( types.describe_error_metadata( exceptions.MapillaryVideoGPSNotFoundError( - "GPX file not found for video" + f"{self.extractor_class.file_format} file not found for video" ), filename=video_path, filetype=types.FileType.VIDEO, diff --git a/mapillary_tools/geotag/geotag_videos_from_nmea.py b/mapillary_tools/geotag/geotag_videos_from_nmea.py new file mode 100644 index 000000000..a4c84e45f --- /dev/null +++ b/mapillary_tools/geotag/geotag_videos_from_nmea.py @@ -0,0 +1,19 @@ +# 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 .geotag_videos_from_gpx import GeotagVideosFromGPX +from .video_extractors.nmea import NMEAVideoExtractor + + +class GeotagVideosFromNMEA(GeotagVideosFromGPX): + """ + Geotag videos from NMEA files, which are placed on the videos as GPX + files are + """ + + extractor_class = NMEAVideoExtractor + default_pattern = "%g.nmea" diff --git a/mapillary_tools/geotag/utils.py b/mapillary_tools/geotag/utils.py index 3c1e24103..e1b02e514 100644 --- a/mapillary_tools/geotag/utils.py +++ b/mapillary_tools/geotag/utils.py @@ -29,28 +29,42 @@ def parse_gpx(gpx_file: Path) -> list[Track]: tracks.append([]) for point in segment.points: if point.time is not None: - unix_time = geo.as_unix_time(point.time) tracks[-1].append( - telemetry.CAMMGPSPoint( - time=unix_time, + track_point( + geo.as_unix_time(point.time), lat=point.latitude, lon=point.longitude, alt=point.elevation, - angle=None, - epoch_time=unix_time, - gps_fix_type=3 if point.elevation is not None else 2, - horizontal_accuracy=0.0, - vertical_accuracy=0.0, - velocity_east=0.0, - velocity_north=0.0, - velocity_up=0.0, - speed_accuracy=0.0, ) ) return tracks +def track_point( + unix_time: float, lat: float, lon: float, alt: float | None +) -> telemetry.CAMMGPSPoint: + """ + A point of a track read from a file beside a video (GPX or NMEA), at its + Unix time, as it is written to the CAMM track uploaded with the video + """ + return telemetry.CAMMGPSPoint( + time=unix_time, + lat=lat, + lon=lon, + alt=alt, + angle=None, + epoch_time=unix_time, + gps_fix_type=3 if alt is not None else 2, + horizontal_accuracy=0.0, + vertical_accuracy=0.0, + velocity_east=0.0, + velocity_north=0.0, + velocity_up=0.0, + speed_accuracy=0.0, + ) + + def index_rdf_description_by_path(xml_paths: T.Sequence[Path]) -> dict[str, ET.Element]: rdf_by_path: dict[str, ET.Element] = {} diff --git a/mapillary_tools/geotag/video_extractors/gpx.py b/mapillary_tools/geotag/video_extractors/gpx.py index bd348bbea..05217d472 100644 --- a/mapillary_tools/geotag/video_extractors/gpx.py +++ b/mapillary_tools/geotag/video_extractors/gpx.py @@ -99,6 +99,9 @@ def to_unix_time(self, video_time: float) -> float: class GPXVideoExtractor(BaseVideoExtractor): + # The format of the track file, as messages name it + file_format = "GPX" + def __init__( self, video_path: Path, @@ -174,10 +177,11 @@ def extract(self) -> types.VideoMetadata: ): LOG.warning( "The GPS clock of %s, whose GPS was rejected as noise, is %.1f days away from %s. " - "Syncing the GPX from the start of the video instead", + "Syncing the %s from the start of the video instead", self.video_path, abs(offset) / 86400, self.gpx_path, + self.file_format, ) clock = None @@ -222,26 +226,21 @@ def _read_gpx_points(self) -> list[geo.Point]: """ Read the points of the GPX file, merged into one track in time order """ - try: - gpx_tracks = parse_gpx(self.gpx_path) - except (OSError, UnicodeError, gpxpy.gpx.GPXException) as ex: - raise exceptions.MapillaryInvalidGPXError( - f"{self.gpx_path.name} is not a valid GPX file. " - "Replace it with a valid GPX file." - ) from ex + gpx_tracks = self._parse_tracks() if 1 < len(gpx_tracks): LOG.warning( - f"Found {len(gpx_tracks)} tracks in the GPX file {self.gpx_path}. Will merge points in all the tracks as a single track for interpolation" + f"Found {len(gpx_tracks)} tracks in the {self.file_format} file {self.gpx_path}. Will merge points in all the tracks as a single track for interpolation" ) all_points: list[geo.Point] = sum(gpx_tracks, []) gpx_points = [p for p in all_points if _GPS_EPOCH <= p.time] if len(gpx_points) < len(all_points): LOG.warning( - "Ignoring %d of the %d points in the GPX file %s, which are timestamped before 1980", + "Ignoring %d of the %d points in the %s file %s, which are timestamped before 1980", len(all_points) - len(gpx_points), len(all_points), + self.file_format, self.gpx_path, ) @@ -250,6 +249,18 @@ def _read_gpx_points(self) -> list[geo.Point]: gpx_points.sort(key=lambda p: p.time) return gpx_points + def _parse_tracks(self) -> list[list[geo.Point]]: + """ + Parse the tracks of the GPX file, each a list of points with Unix times + """ + try: + return parse_gpx(self.gpx_path) + except (OSError, UnicodeError, gpxpy.gpx.GPXException) as ex: + raise exceptions.MapillaryInvalidGPXError( + f"{self.gpx_path.name} is not a valid GPX file. " + "Replace it with a valid GPX file." + ) from ex + def _clip_to_video(self, gpx_points: list[geo.Point]) -> list[geo.Point]: """ Clip a GPX track to the video when a clock gives its pace along the @@ -389,7 +400,7 @@ def _check_coverage( gpx_unix_times[0] if clock is None else clock.to_unix_time(0) ) + offset_time - gpx_track = f"The GPX track in {self.gpx_path} ({_isoformat(gpx_unix_times[0])} to {_isoformat(gpx_unix_times[1])})" + gpx_track = f"The {self.file_format} track in {self.gpx_path} ({_isoformat(gpx_unix_times[0])} to {_isoformat(gpx_unix_times[1])})" if clock is None: video = f"the video {self.video_path}" else: @@ -420,14 +431,14 @@ def _check_coverage( advice = "Check --interpolation_offset_time" elif offset_time: advice = ( - "Check --interpolation_offset_time, that the GPX file belongs to " - "this video, then the camera clock and the time zone of the GPX " + f"Check --interpolation_offset_time, that the {self.file_format} file belongs to " + f"this video, then the camera clock and the time zone of the {self.file_format} " "timestamps" ) else: advice = ( - "Check that the GPX file belongs to this video, then the camera " - "clock and the time zone of the GPX timestamps" + f"Check that the {self.file_format} file belongs to this video, then the camera " + f"clock and the time zone of the {self.file_format} timestamps" ) message = ( f"{gpx_track} misses {video} by {_format_duration(gap)}{offset_clause}. " @@ -468,7 +479,7 @@ def _check_assumed_start( if video_span < excess: raise exceptions.MapillaryGPXVideoStartUnknownError( - f"The GPX track in {self.gpx_path} lasts {_format_duration(gpx_duration)} " + f"The {self.file_format} track in {self.gpx_path} lasts {_format_duration(gpx_duration)} " f"and {self._describe_video_span(video_span, ratio)}, so the video " "could start anywhere along the track. Specify --video_start_time, or " "--interpolation_use_gpx_start_time if the track was started with the " @@ -477,11 +488,12 @@ def _check_assumed_start( if _LOGGER_TOLERANCE_SECONDS < excess: LOG.warning( - "Assuming that the GPX track in %s starts with the video %s, although " + "Assuming that the %s track in %s starts with the video %s, although " "the track lasts %s longer. If it was started first, the frames are " "placed up to that much too early along it. Specify --video_start_time " "to place them exactly, or --interpolation_use_gpx_start_time if the " "track was started with the video", + self.file_format, self.gpx_path, self.video_path, _format_duration(excess), @@ -505,7 +517,7 @@ def _check_given_start( return message = ( - f"The GPX track in {self.gpx_path} lasts {_format_duration(gpx_duration)} " + f"The {self.file_format} track in {self.gpx_path} lasts {_format_duration(gpx_duration)} " f"and {self._describe_video_span(video_span, ratio)}, so it runs on long " "after the video ends. " ) @@ -595,11 +607,11 @@ def _warn_of_a_long_track(self, gpx_points: T.Sequence[geo.Point]) -> None: gpx_duration = max(p.time for p in gpx_points) - min(p.time for p in gpx_points) if 2 * video_duration < gpx_duration: LOG.warning( - f"The GPX track in {self.gpx_path} lasts {_format_duration(gpx_duration)} " + f"The {self.file_format} track in {self.gpx_path} lasts {_format_duration(gpx_duration)} " f"and the video {self.video_path} only {_format_duration(video_duration)}, " "so the video could start anywhere along the track, and is assumed to " "start with it. Unless the video is a timelapse of the whole track, " - "trim the GPX to the video, or sample the video with video_process " + f"trim the {self.file_format} to the video, or sample the video with video_process " "and --video_start_time" ) diff --git a/mapillary_tools/geotag/video_extractors/nmea.py b/mapillary_tools/geotag/video_extractors/nmea.py new file mode 100644 index 000000000..d58b35762 --- /dev/null +++ b/mapillary_tools/geotag/video_extractors/nmea.py @@ -0,0 +1,43 @@ +# 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 sys + +if sys.version_info >= (3, 12): + from typing import override +else: + from typing_extensions import override + +from ... import exceptions, geo +from ..geotag_images_from_nmea_file import get_lat_lon_time_from_nmea +from ..utils import track_point +from .gpx import GPXVideoExtractor + + +class NMEAVideoExtractor(GPXVideoExtractor): + """ + Geotag a video from the track in an NMEA file, placed on the video as a + GPX track is + """ + + file_format = "NMEA" + + @override + 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) or a + # malformed sentence (pynmea2.ParseError); TypeError: a sentence + # without its time or date; UnboundLocalError: GGA sentences, and no + # RMC sentence in the file to give their date + except (OSError, ValueError, TypeError, UnboundLocalError) as ex: + raise exceptions.MapillaryInvalidGPXError( + f"{self.gpx_path.name} is not a valid NMEA file. " + "Replace it with a valid NMEA file." + ) from ex + + return [[track_point(p.time, p.lat, p.lon, p.alt) for p in points]] diff --git a/mapillary_tools/sample_video.py b/mapillary_tools/sample_video.py index 4714844f8..72337538d 100644 --- a/mapillary_tools/sample_video.py +++ b/mapillary_tools/sample_video.py @@ -39,6 +39,7 @@ ) from .geotag.video_extractors.gpx import GPXVideoExtractor from .geotag.video_extractors.native import NativeVideoExtractor +from .geotag.video_extractors.nmea import NMEAVideoExtractor from .mp4 import mp4_sample_parser from .serializer.description import build_capture_time, parse_capture_time @@ -132,8 +133,8 @@ def _parse_video_sources( if option.source_path is None: option.source_path = SourcePathOption(source_path=video_import_path) - # exif, nmea and the legacy gopro/blackvue/camm sources have nothing to read - # a video with; in video_process they are there to geotag the frames. Sample + # exif and the legacy gopro/blackvue/camm sources have nothing to read a + # video with; in video_process they are there to geotag the frames. Sample # by the sources that can read videos, or by the video's own GPS, as before, # if none of them can video_options = [option for option in options if factory.can_geotag_videos(option)] @@ -607,27 +608,35 @@ def _extract_video_start_time( return _creation_time_to_start_time(video_path, probe) +# The sources that place a track file on a video, by the extractor that reads it +_TRACK_FILE_EXTRACTORS: dict[SourceType, type[GPXVideoExtractor]] = { + SourceType.GPX: GPXVideoExtractor, + SourceType.NMEA: NMEAVideoExtractor, +} + + def _check_gpx_start_by_interval( video_path: Path, geotag_options: T.Sequence[SourceOption], video_sync: VideoSyncOption, ) -> None: """ - Check the GPX tracks that --interpolation_use_gpx_start_time says started - with a video without a GPS clock, as distance sampling does. The geotagging - stage puts the first frame at the start of the track and the others at - --video_duration_ratio, so a track that runs on long after the video was - not recorded for it alone, or needs the ratio. + Check the GPX (or NMEA) tracks that --interpolation_use_gpx_start_time says + started with a video without a GPS clock, as distance sampling does. The + geotagging stage puts the first frame at the start of the track and the + others at --video_duration_ratio, so a track that runs on long after the + video was not recorded for it alone, or needs the ratio. """ for option in geotag_options: - if option.source is not SourceType.GPX or option.source_path is None: + extractor_class = _TRACK_FILE_EXTRACTORS.get(option.source) + if extractor_class is None or option.source_path is None: continue if option.interpolation is None or not option.interpolation.use_gpx_start_time: continue gpx_path = option.source_path.resolve(video_path) if not gpx_path.is_file(): continue - extractor = GPXVideoExtractor( + extractor = extractor_class( video_path, gpx_path, video_sync=video_sync, diff --git a/tests/integration/test_process.py b/tests/integration/test_process.py index f21c093ba..8151d89f4 100644 --- a/tests/integration/test_process.py +++ b/tests/integration/test_process.py @@ -799,6 +799,58 @@ def video_time_of_recorded(*options: str) -> float: assert "by 40 seconds, as --interpolation_offset_time 60 " in error["message"] +def test_process_video_from_nmea(setup_data: py.path.local): + """ + A video is geotagged from an NMEA file as from a GPX file of the same + track. It got no description at all. + """ + video_path = setup_data.join("gopro_data").join("max-360mode.mp4") + start = datetime.datetime(2019, 11, 18, 23, 44, 40, tzinfo=datetime.timezone.utc) + trkpts = [] + sentences = [] + for i in range(21): + time = start + datetime.timedelta(seconds=i) + trkpts.append( + f'' + f"" + ) + hhmmss = time.strftime("%H%M%S.00") + position = f"3700.{i * 100:04d},N,12200.0000,W" + for body in [ + f"GPRMC,{hhmmss},A,{position},0.0,0.0,{time.strftime('%d%m%y')},,,A", + f"GPGGA,{hhmmss},{position},1,08,1.0,,M,,M,,", + ]: + checksum = 0 + for char in body: + checksum ^= ord(char) + sentences.append(f"${body}*{checksum:02X}\r\n") + gpx_file = setup_data.join("track.gpx") + gpx_file.write( + '\n' + '' + f"{''.join(trkpts)}\n" + ) + nmea_file = setup_data.join("track.nmea") + nmea_file.write("".join(sentences)) + + def track(source: str, source_path: py.path.local): + [desc] = run_process_for_descs( + [ + *["--geotag_source", source], + *["--geotag_source_path", str(source_path)], + str(video_path), + ] + ) + return desc["MAPGPSTrack"] + + from_nmea = track("nmea", nmea_file) + from_gpx = track("gpx", gpx_file) + + assert len(from_nmea) == len(from_gpx) > 0 + for nmea_point, gpx_point in zip(from_nmea, from_gpx): + assert nmea_point == pytest.approx(gpx_point, abs=1e-6) + + def test_process_geotag_with_gpx_pattern_not_found(setup_data: py.path.local): video_path = setup_data.join("gopro_data").join("max-360mode.mp4") diff --git a/tests/unit/test_nmea_video.py b/tests/unit/test_nmea_video.py new file mode 100644 index 000000000..5bd1e611a --- /dev/null +++ b/tests/unit/test_nmea_video.py @@ -0,0 +1,207 @@ +# 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. + +""" +Geotagging videos from NMEA files, which are placed on the videos as GPX files +are. NMEA files geotagged no videos: the videos got no description at all. +""" + +from __future__ import annotations + +import datetime +from pathlib import Path + +import pynmea2 +import pytest +from mapillary_tools import exceptions, types +from mapillary_tools.geotag import factory +from mapillary_tools.geotag.geotag_videos_from_nmea import GeotagVideosFromNMEA +from mapillary_tools.geotag.options import ( + InterpolationOption, + SourceOption, + SourcePathOption, + SourceType, + VideoSyncOption, +) +from mapillary_tools.geotag.video_extractors.gpx import GPXVideoExtractor +from mapillary_tools.geotag.video_extractors.nmea import NMEAVideoExtractor + +_DATA = Path(__file__).resolve().parent.parent.joinpath("data") + +# No GPS of its own, and 5.76 seconds long +NO_GPS_VIDEO = _DATA.joinpath("videos", "sample-5s.mp4") +# GPS of its own, recorded from 2019-11-18T23:44:42.59Z to 23:44:54.64Z +GPS_VIDEO = _DATA.joinpath("gopro_data", "max-360mode.mp4") + +# Shortly before the GPS track of GPS_VIDEO starts +START = datetime.datetime(2019, 11, 18, 23, 44, 40, tzinfo=datetime.timezone.utc) + + +def _track(seconds: int, start: datetime.datetime): + """A point every second, heading north at 10 m/s, 12 m high""" + for i in range(seconds + 1): + yield start + datetime.timedelta(seconds=i), 37.0 + i * 10 / 111_320, 12.0 + + +def _write_gpx(path: Path, seconds: int, start: datetime.datetime = START) -> Path: + points = "".join( + f'{alt}' + f"" + for time, lat, alt in _track(seconds, start) + ) + path.write_text( + '\n' + '' + f"{points}\n" + ) + return path + + +def _write_nmea(path: Path, seconds: int, start: datetime.datetime = START) -> Path: + """The track of _write_gpx(), as RMC and GGA sentences""" + lines = [] + for time, lat, alt in _track(seconds, start): + hhmmss = time.strftime("%H%M%S.00") + lat_field = f"{int(lat):02d}{(lat - int(lat)) * 60:010.7f}" + lon_field = "12200.0000000" + lines.append( + pynmea2.RMC( + "GP", + "RMC", + (hhmmss, "A", lat_field, "N", lon_field, "W", "19.4", "0.0") + + (time.strftime("%d%m%y"), "", "", "A"), + ) + ) + lines.append( + pynmea2.GGA( + "GP", + "GGA", + (hhmmss, lat_field, "N", lon_field, "W", "1", "08", "1.0") + + (f"{alt:.1f}", "M", "0.0", "M", "", ""), + ) + ) + path.write_text("".join(f"{line}\r\n" for line in lines)) + return path + + +def _points(extractor: GPXVideoExtractor): + return [ + (p.time, p.lat, p.lon, p.alt, p.get_unix_time()) + for p in extractor.extract().points + ] + + +class TestPlacedAsAGPX: + @pytest.mark.parametrize( + "video, seconds, kwargs", + [ + # synced by the GPS clock of the video, and clipped to it, in process + (GPS_VIDEO, 20, {"clip_to_video": True}), + # started with the video + ( + GPS_VIDEO, + 20, + {"interpolation": InterpolationOption(use_gpx_start_time=True)}, + ), + # moved by the offset + (GPS_VIDEO, 20, {"interpolation": InterpolationOption(offset_time=2)}), + # assumed to start with a video without GPS + (NO_GPS_VIDEO, 6, {}), + # placed by the start time of a video without GPS, in sampling + ( + NO_GPS_VIDEO, + 60, + { + "video_sync": VideoSyncOption( + start_time=START.timestamp() + 30, duration_ratio=2 + ) + }, + ), + ], + ) + def test_the_track_is_placed_as_from_a_gpx(self, tmp_path, video, seconds, kwargs): + gpx = _write_gpx(tmp_path / "track.gpx", seconds) + nmea = _write_nmea(tmp_path / "track.nmea", seconds) + + from_gpx = _points(GPXVideoExtractor(video, gpx, **kwargs)) + from_nmea = _points(NMEAVideoExtractor(video, nmea, **kwargs)) + + assert from_nmea + assert len(from_nmea) == len(from_gpx) + 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) + + with pytest.raises(exceptions.MapillaryGPXVideoStartUnknownError) as excinfo: + NMEAVideoExtractor( + NO_GPS_VIDEO, nmea, video_sync=VideoSyncOption() + ).extract() + + assert str(excinfo.value).startswith(f"The NMEA track in {nmea} lasts") + + +class TestInvalidNMEAFile: + @pytest.mark.parametrize( + "content", + [ + # not text + b"\xca\xfe\xba\xbe" * 100, + # a wrong checksum + b"$GPRMC,024636.00,A,3735.30670,N,12220.25810,W,0.0,0.0,221222,,,A*00\r\n", + # a GGA sentence, and no RMC sentence to give its date + b"$GPGGA,024636.00,3735.30670,N,12220.25810,W,1,08,1.0,0.0,M,0.0,M,,*4F\r\n", + # an RMC sentence cut short, with no date + b"$GPRMC,024636.00,A,3735.30670,N,12220.25810,W,0.0*07\r\n", + ], + ) + def test_it_is_reported_as_invalid(self, tmp_path, content): + nmea = tmp_path / "track.nmea" + nmea.write_bytes(content) + + with pytest.raises(exceptions.MapillaryInvalidGPXError) as excinfo: + NMEAVideoExtractor(GPS_VIDEO, nmea).extract() + + assert "track.nmea is not a valid NMEA file" in str(excinfo.value) + + +class TestGeotagVideosFromNMEA: + def test_the_nmea_file_beside_the_video_is_found(self, tmp_path): + video = tmp_path / "video.mp4" + video.write_bytes(GPS_VIDEO.read_bytes()) + _write_nmea(tmp_path / "video.nmea", seconds=20) + + [metadata] = GeotagVideosFromNMEA(num_processes=0).to_description([video]) + + assert isinstance(metadata, types.VideoMetadata) + assert metadata.points + + def test_a_missing_nmea_file_is_reported(self, tmp_path): + video = tmp_path / "video.mp4" + video.write_bytes(GPS_VIDEO.read_bytes()) + + [metadata] = GeotagVideosFromNMEA(num_processes=0).to_description([video]) + + assert isinstance(metadata, types.ErrorMetadata) + assert isinstance(metadata.error, exceptions.MapillaryVideoGPSNotFoundError) + assert str(metadata.error) == "NMEA file not found for video" + + def test_process_geotags_videos_from_nmea(self, tmp_path): + nmea = _write_nmea(tmp_path / "track.nmea", seconds=20) + + [metadata] = factory.process( + [GPS_VIDEO], + [ + SourceOption( + SourceType.NMEA, + source_path=SourcePathOption(source_path=nmea), + num_processes=0, + ) + ], + ) + + assert isinstance(metadata, types.VideoMetadata) + assert metadata.points diff --git a/tests/unit/test_sample_video.py b/tests/unit/test_sample_video.py index 1cd2b0c6d..552674748 100644 --- a/tests/unit/test_sample_video.py +++ b/tests/unit/test_sample_video.py @@ -16,6 +16,7 @@ from unittest import mock import py.path +import pynmea2 import pytest from mapillary_tools import ( exceptions, @@ -437,6 +438,33 @@ def test_a_track_that_ends_with_the_video_at_the_ratio_passes(self, tmp_path): sample_video._check_gpx_start_by_interval(self.VIDEO, options, video_sync) + def test_an_nmea_track_is_checked_as_a_gpx_track(self, tmp_path): + start = datetime.datetime(2025, 3, 14, 7, 0, 0, tzinfo=datetime.timezone.utc) + nmea = tmp_path / "track.nmea" + lines = [] + for i in range(61): + hhmmss = (start + datetime.timedelta(seconds=i)).strftime("%H%M%S.00") + position = (f"3700.{i:04d}", "N", "12200.0000", "W") + rmc = (hhmmss, "A", *position, "0.0", "0.0", "140325", "", "", "A") + gga = (hhmmss, *position, "1", "08", "1.0", "0.0", "M", "0.0", "M", "", "") + lines.append(str(pynmea2.RMC("GP", "RMC", rmc))) + lines.append(str(pynmea2.GGA("GP", "GGA", gga))) + nmea.write_text("".join(f"{line}\r\n" for line in lines)) + video_sync = VideoSyncOption() + options = sample_video._parse_geotag_options( + ["nmea"], + nmea, + None, + tmp_path, + video_sync, + InterpolationOption(use_gpx_start_time=True), + ) + + with pytest.raises(exceptions.MapillaryGPXVideoStartUnknownError) as excinfo: + sample_video._check_gpx_start_by_interval(self.VIDEO, options, video_sync) + + assert str(excinfo.value).startswith(f"The NMEA track in {nmea} lasts") + def test_gps_rejected_as_noise_still_has_a_clock(self, tmp_path, monkeypatch): """ Interval sampling reads no telemetry from GPS rejected as noise, but @@ -1822,12 +1850,12 @@ def fake_process(paths, options): class TestSourcesThatCannotReadVideos: """ - exif, nmea and the legacy gopro/blackvue/camm sources have no video - extractor, so process() returned no metadata for the video and the sampler - failed an assert, for a video it used to sample by its own GPS. + exif and the legacy gopro/blackvue/camm sources have no video extractor, + so process() returned no metadata for the video and the sampler failed an + assert, for a video it used to sample by its own GPS. """ - @pytest.mark.parametrize("source", ["exif", "nmea", "gopro", "blackvue", "camm"]) + @pytest.mark.parametrize("source", ["exif", "gopro", "blackvue", "camm"]) def test_sampling_falls_back_to_the_video_gps(self, source): options = sample_video._parse_geotag_options( [source], Path("/data/track"), None, Path("/data/v.mp4") @@ -1835,6 +1863,13 @@ def test_sampling_falls_back_to_the_video_gps(self, source): assert [option.source for option in options] == [SourceType.NATIVE] + def test_nmea_reads_videos(self): + options = sample_video._parse_geotag_options( + ["nmea"], Path("/data/track.nmea"), None, Path("/data/v.mp4") + ) + + assert [option.source for option in options] == [SourceType.NMEA] + def test_they_are_dropped_from_a_chain(self): options = sample_video._parse_geotag_options( ["exif", "gpx"], Path("/data/track.gpx"), None, Path("/data/v.mp4")