From 2efacb81c569098d40b71904d879b6e120d62e7c Mon Sep 17 00:00:00 2001 From: Caglar Pir Date: Thu, 22 Oct 2026 11:59:21 +0200 Subject: [PATCH 1/3] Geotag videos from NMEA files The README lists NMEA among the sources that geotag videos, but no video was read from an NMEA file: process and video_process gave the video no description at all, and exited 0, so it was silently left out. Sampling fell back to the GPS in the video, which a video recorded with an external logger does not have. Read the NMEA file into a track and place it on the video exactly as a GPX track is: synced by the GPS clock of the video, or by --video_start_time, or assumed to start with the video, with the interpolation options, the coverage checks and the clipping of process. GPXVideoExtractor now reads its file through _parse_tracks(), which NMEAVideoExtractor overrides, and names the format of the file in its messages. A file that is not valid NMEA is reported as an invalid GPX file is. Interval sampling checks an NMEA track started with the video by --interpolation_use_gpx_start_time as it checks a GPX track. On an action camera video with a GPX and the same track as NMEA, process gives the same track, and video_process samples the same frames at the same positions. --- README.md | 2 + mapillary_tools/geotag/factory.py | 19 +- .../geotag/geotag_images_from_nmea_file.py | 5 +- .../geotag/geotag_videos_from_gpx.py | 10 +- .../geotag/geotag_videos_from_nmea.py | 19 ++ mapillary_tools/geotag/utils.py | 38 +++- .../geotag/video_extractors/gpx.py | 52 +++-- .../geotag/video_extractors/nmea.py | 43 ++++ mapillary_tools/sample_video.py | 27 ++- tests/integration/test_process.py | 52 +++++ tests/unit/test_nmea_video.py | 207 ++++++++++++++++++ tests/unit/test_sample_video.py | 43 +++- 12 files changed, 462 insertions(+), 55 deletions(-) create mode 100644 mapillary_tools/geotag/geotag_videos_from_nmea.py create mode 100644 mapillary_tools/geotag/video_extractors/nmea.py create mode 100644 tests/unit/test_nmea_video.py 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") From 58249121f4fdeb180a70e7ccc10a2e658a325446 Mon Sep 17 00:00:00 2001 From: Caglar Pir Date: Thu, 22 Oct 2026 13:21:16 +0200 Subject: [PATCH 2/3] Read NMEA files that log sentences without a fix, or from other talkers Images geotagged from an NMEA file failed with a traceback when the file had an RMC sentence without a date, such as the last one of a log cut short (TypeError), or GGA sentences and no RMC sentence at all (UnboundLocalError), and a video was failed as if the file were not NMEA. A void RMC sentence, which receivers log before they acquire a fix, gave the date of their unset clock, and a GGA sentence without a fix added a point at 0, 0. Only $GP sentences were read, so a log of a receiver of several systems, which writes $GN sentences, gave no points at all. Skip RMC sentences that are void or have no date or time, and GGA sentences without a time or a fix, and read the sentences of any talker, keeping the first fix of each time, so that a receiver that logs a fix from two talkers does not give two points at the same time. GGA sentences without the date of any RMC sentence give no points, with a warning. The points read from valid files are the same. --- .../geotag/geotag_images_from_nmea_file.py | 80 ++++++--- .../geotag/video_extractors/nmea.py | 9 +- tests/unit/test_nmea_parser.py | 156 ++++++++++++++++++ tests/unit/test_nmea_video.py | 35 +++- 4 files changed, 252 insertions(+), 28 deletions(-) create mode 100644 tests/unit/test_nmea_parser.py diff --git a/mapillary_tools/geotag/geotag_images_from_nmea_file.py b/mapillary_tools/geotag/geotag_images_from_nmea_file.py index d0e1b07ee..eb779c736 100644 --- a/mapillary_tools/geotag/geotag_images_from_nmea_file.py +++ b/mapillary_tools/geotag/geotag_images_from_nmea_file.py @@ -6,6 +6,8 @@ from __future__ import annotations import datetime +import logging +import re from pathlib import Path import pynmea2 @@ -14,6 +16,13 @@ from .geotag_images_from_gpx import GeotagImagesFromGPX +LOG = logging.getLogger(__name__) + +# An RMC or a GGA sentence, from any talker: GP (GPS), GN (several systems), +# GL (GLONASS), GA (Galileo), ... +_SENTENCE_RE = re.compile(r"\$[A-Z]{2}(RMC|GGA),") + + class GeotagImagesFromNMEAFile(GeotagImagesFromGPX): def __init__( self, @@ -66,6 +75,20 @@ def _read_nmea_points(source_path: Path) -> list[geo.Point]: def get_lat_lon_time_from_nmea(nmea_file: Path) -> list[geo.Point]: + """ + Read the fixes of an NMEA file: the time and position of each GGA + sentence, on the date of the last RMC sentence before it, or of the first + RMC sentence for the GGA sentences before any. + + Receivers log sentences without a fix, such as before they acquire one: + RMC sentences that are void or have no date or time, and GGA sentences + without a time or a fix, which give their position as 0, 0. They are + skipped. A file without the date of any fix gives no points. + + 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. + """ # 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 @@ -73,29 +96,48 @@ 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)] + # Get initial date - for line in lines: - if "GPRMC" in line: - data = pynmea2.parse(line) - date = data.datetime.date() - break + dates = [_fix_date(m) for m in messages if isinstance(m, pynmea2.RMC)] + date = next((d for d in dates if d is not None), None) + if date is None: + fixes = [m for m in messages if isinstance(m, pynmea2.GGA) and _has_fix(m)] + if fixes: + LOG.warning( + "Ignoring the %d fixes in the NMEA file %s, as no RMC sentence gives their date", + len(fixes), + nmea_file, + ) + return [] # Parse GPS trace points = [] - for line in lines: - if "GPRMC" in line: - data = pynmea2.parse(line) - date = data.datetime.date() - - if "$GPGGA" in line: - data = pynmea2.parse(line) - dt = datetime.datetime.combine(date, data.timestamp) - lat, lon, alt = data.latitude, data.longitude, data.altitude - points.append( - geo.Point( - time=geo.as_unix_time(dt), lat=lat, lon=lon, alt=alt, angle=None - ) - ) + times: set[float] = set() + for message in messages: + if isinstance(message, pynmea2.RMC): + date = _fix_date(message) or date + + elif isinstance(message, pynmea2.GGA) and _has_fix(message): + dt = datetime.datetime.combine(date, message.timestamp) + time = geo.as_unix_time(dt) + if time in times: + continue + times.add(time) + lat, lon, alt = message.latitude, message.longitude, message.altitude + points.append(geo.Point(time=time, lat=lat, lon=lon, alt=alt, angle=None)) points.sort() return points + + +def _fix_date(message: pynmea2.RMC) -> datetime.date | None: + """The date of an RMC sentence, or None if it is void or has no date or time""" + if message.status != "A" or message.timestamp is None: + return None + return message.datestamp + + +def _has_fix(message: pynmea2.GGA) -> bool: + """Whether a GGA sentence has a time and a fix (quality 0 is none)""" + return message.timestamp is not None and bool(message.gps_qual) diff --git a/mapillary_tools/geotag/video_extractors/nmea.py b/mapillary_tools/geotag/video_extractors/nmea.py index d58b35762..466d00f84 100644 --- a/mapillary_tools/geotag/video_extractors/nmea.py +++ b/mapillary_tools/geotag/video_extractors/nmea.py @@ -30,11 +30,10 @@ 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) 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: + # 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: 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_nmea_parser.py b/tests/unit/test_nmea_parser.py new file mode 100644 index 000000000..49d6fd856 --- /dev/null +++ b/tests/unit/test_nmea_parser.py @@ -0,0 +1,156 @@ +# 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. + +""" +Reading the fixes of NMEA files as receivers log them: before a fix, with +void sentences, and from talkers other than GPS. +""" + +from __future__ import annotations + +import datetime +import logging +from pathlib import Path + +import pynmea2 +import pytest +from mapillary_tools.geotag.geotag_images_from_nmea_file import ( + get_lat_lon_time_from_nmea, +) + + +def _sentence(body: str) -> str: + return f"${body}*{pynmea2.NMEASentence.checksum(body):02X}" + + +def _rmc(time: str, date: str = "221222", status: str = "A", talker="GP") -> str: + return _sentence( + f"{talker}RMC,{time},{status},3735.30670,N,12220.25810,W,0.0,0.0,{date},,,A" + ) + + +def _gga(time: str, quality: int = 1, talker="GP", lat="3735.30670") -> str: + return _sentence( + f"{talker}GGA,{time},{lat},N,12220.25810,W,{quality},08,1.0,12.5,M,0.0,M,," + ) + + +def _read(tmp_path: Path, *lines: str): + path = tmp_path / "track.nmea" + path.write_text("".join(f"{line}\r\n" for line in lines)) + return [ + ( + datetime.datetime.fromtimestamp(p.time, datetime.timezone.utc).isoformat(), + round(p.lat, 6), + round(p.lon, 6), + p.alt, + ) + for p in get_lat_lon_time_from_nmea(path) + ] + + +def _fix(time: str, date: str = "2022-12-22", lat: float = 37.588445): + return (f"{date}T{time}+00:00", lat, -122.337635, 12.5) + + +def test_gps_fixes(tmp_path): + assert _read( + tmp_path, + _rmc("024636.00"), + _gga("024636.00"), + _rmc("024637.00"), + _gga("024637.00"), + ) == [_fix("02:46:36"), _fix("02:46:37")] + + +@pytest.mark.parametrize("talker", ["GN", "GL", "GA", "BD"]) +def test_any_talker(tmp_path, talker): + """Receivers of several systems log $GN sentences, which read 0 points""" + assert _read( + tmp_path, _rmc("024636.00", talker=talker), _gga("024636.00", talker=talker) + ) == [_fix("02:46:36")] + + +def test_one_point_per_time(tmp_path): + """ + Of the fixes that several talkers log for the same time, the first is + read, so that no two points share a time + """ + assert _read( + tmp_path, + _rmc("024636.00", talker="GP"), + _rmc("024636.00", talker="GN"), + _gga("024636.00", talker="GP"), + _gga("024636.00", talker="GN", lat="3735.30610"), + _gga("024637.00", talker="GN", lat="3735.30610"), + _gga("024637.00", talker="GP"), + ) == [_fix("02:46:36"), _fix("02:46:37", lat=37.588435)] + + +def test_sentences_before_the_first_fix_are_skipped(tmp_path): + """ + 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")] + + +def test_a_void_rmc_gives_no_date(tmp_path): + assert _read( + tmp_path, + _rmc("024636.00"), + _gga("024636.00"), + _rmc("024637.00", date="060180", status="V"), + _gga("024637.00"), + ) == [_fix("02:46:36"), _fix("02:46:37")] + + +def test_a_cut_off_sentence_is_skipped(tmp_path): + """The last sentence of a log cut short has no date, and failed the file""" + assert _read( + tmp_path, _rmc("024636.00"), _gga("024636.00"), "$GPRMC,024637.00,A,3735.3" + ) == [_fix("02:46:36")] + + +def test_gga_before_the_first_rmc_take_its_date(tmp_path): + assert _read( + tmp_path, + _gga("024635.00", lat="3735.30610"), + _rmc("024636.00"), + _gga("024636.00"), + ) == [_fix("02:46:35", lat=37.588435), _fix("02:46:36")] + + +def test_gga_without_any_date_give_no_points(tmp_path, caplog): + """They raised UnboundLocalError""" + with caplog.at_level(logging.WARNING): + assert _read(tmp_path, _gga("024636.00"), _gga("024637.00")) == [] + + assert "Ignoring the 2 fixes" in caplog.text + assert "no RMC sentence gives their date" in caplog.text + + +def test_no_fixes_give_no_points(tmp_path, caplog): + with caplog.at_level(logging.WARNING): + assert ( + _read( + tmp_path, + _rmc("024636.00", status="V"), + _gga("024636.00", quality=0), + ) + == [] + ) + + assert not caplog.text diff --git a/tests/unit/test_nmea_video.py b/tests/unit/test_nmea_video.py index 5bd1e611a..0b8a09728 100644 --- a/tests/unit/test_nmea_video.py +++ b/tests/unit/test_nmea_video.py @@ -11,6 +11,7 @@ from __future__ import annotations import datetime +import logging from pathlib import Path import pynmea2 @@ -152,10 +153,9 @@ class TestInvalidNMEAFile: 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", + # a malformed time + b"$GPRMC,024636.00,A,3735.30670,N,12220.25810,W,0.0,0.0,221222,,,A*47\r\n" + b"$GPGGA,0246,3735.30670,N,12220.25810,W,1,08,1.0,0.0,M,0.0,M,,*64\r\n", ], ) def test_it_is_reported_as_invalid(self, tmp_path, content): @@ -168,6 +168,33 @@ def test_it_is_reported_as_invalid(self, tmp_path, content): assert "track.nmea is not a valid NMEA file" in str(excinfo.value) +class TestNMEAFileWithoutDatedFixes: + @pytest.mark.parametrize( + "content, warning", + [ + # 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", + "as no RMC sentence gives their date", + ), + # an RMC sentence cut short, with no date, which is skipped + (b"$GPRMC,024636.00,A,3735.30670,N,12220.25810,W,0.0*07\r\n", None), + ], + ) + def test_no_points_are_read(self, tmp_path, caplog, content, warning): + nmea = tmp_path / "track.nmea" + nmea.write_bytes(content) + + with caplog.at_level(logging.WARNING): + points = NMEAVideoExtractor(GPS_VIDEO, nmea)._read_gpx_points() + + assert points == [] + if warning is None: + assert not caplog.text + else: + assert warning in caplog.text + + class TestGeotagVideosFromNMEA: def test_the_nmea_file_beside_the_video_is_found(self, tmp_path): video = tmp_path / "video.mp4" From d3f9e299c62fe9600c2494b2b66218c625e8ad7f Mon Sep 17 00:00:00 2001 From: Caglar Pir Date: Thu, 22 Oct 2026 12:16:31 +0200 Subject: [PATCH 3/3] Warn of a file named for process that is not an image or a video process searches a directory for the images and videos it can process, and skips the other files. It skipped a file named explicitly the same way, without a word: `process VID.insv` processed 0 files and exited 0, with nothing on stderr, so a user could not tell why the video was not there. Warn of such a file. It is still skipped, and the run does not fail: callers that pass the files a user selected rely on that, and find the skipped files in the process report (--process_report_path). A zip file is not warned of, as process_and_upload uploads it. --- mapillary_tools/process_geotag_properties.py | 11 ++++++++++ tests/integration/test_process_report.py | 21 ++++++++++++++++++++ 2 files changed, 32 insertions(+) diff --git a/mapillary_tools/process_geotag_properties.py b/mapillary_tools/process_geotag_properties.py index 5a1b5493b..3b9db8187 100644 --- a/mapillary_tools/process_geotag_properties.py +++ b/mapillary_tools/process_geotag_properties.py @@ -62,6 +62,17 @@ def process_geotag_properties( raise exceptions.MapillaryFileNotFoundError( f"Import file or directory not found: {path}" ) + # A directory is searched for the files it can process, but a file + # named explicitly is skipped only with a word. It is not failed: + # callers that pass the files a user selected find the skipped ones in + # the process report. A zip file is uploaded as it is, by + # process_and_upload too + if path.is_file() and not ( + utils.is_image_file(path) + or utils.is_video_file(path) + or path.suffix.lower() == ".zip" + ): + LOG.warning("Skipping %s, which is not a supported image or video", path) if geotag_source_path is None: geotag_source_path = video_import_path diff --git a/tests/integration/test_process_report.py b/tests/integration/test_process_report.py index 437fefe9e..1a1151684 100644 --- a/tests/integration/test_process_report.py +++ b/tests/integration/test_process_report.py @@ -112,3 +112,24 @@ def test_process_without_report_preserves_unsupported_omission(tmpdir: py.path.l assert json.loads(desc_path.read_text(encoding="utf-8")) == [] assert not (root / "process-report.json").exists() + + +def test_process_warns_of_an_explicit_unsupported_file(tmpdir: py.path.local): + """It was skipped without a word""" + root = Path(str(tmpdir)) + unsupported = root / "unsupported.webp" + desc_path = root / "description.json" + unsupported.write_bytes(b"not supported") + + result = run_command( + ["--desc_path", str(desc_path), str(unsupported)], + command="process", + capture_output=True, + text=True, + ) + + assert json.loads(desc_path.read_text(encoding="utf-8")) == [] + assert ( + f"Skipping {unsupported}, which is not a supported image or video" + in result.stderr + )