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..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,33 +75,69 @@ 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:
+ """
+ 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
+ with nmea_file.open("r", encoding="utf-8") as f:
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/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..466d00f84
--- /dev/null
+++ b/mapillary_tools/geotag/video_extractors/nmea.py
@@ -0,0 +1,42 @@
+# 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), 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."
+ ) from ex
+
+ return [[track_point(p.time, p.lat, p.lon, p.alt) for p in points]]
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/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/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
+ )
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
new file mode 100644
index 000000000..0b8a09728
--- /dev/null
+++ b/tests/unit/test_nmea_video.py
@@ -0,0 +1,234 @@
+# 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
+import logging
+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 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):
+ 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 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"
+ 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")