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/commands/video_process.py b/mapillary_tools/commands/video_process.py
index d127b9b98..c6896cd43 100644
--- a/mapillary_tools/commands/video_process.py
+++ b/mapillary_tools/commands/video_process.py
@@ -6,8 +6,10 @@
from __future__ import annotations
import logging
+from pathlib import Path
-from .. import exceptions
+from .. import constants, exceptions
+from ..sample_video import find_sample_dirs
from ..types import FileType
from .process import Command as ProcessCommand
from .sample_video import Command as SampleCommand
@@ -43,7 +45,39 @@ def run(self, args: dict):
)
args[option] = {FileType.IMAGE}
+ if args["skip_subfolders"]:
+ _add_sample_dirs(args)
+
ProcessCommand().run(args)
if sample_error is not None:
raise sample_error
+
+
+def _add_sample_dirs(args: dict) -> None:
+ """
+ Add the folders that the videos were sampled into to the import paths.
+ --skip_subfolders chose the videos to sample, and hides the subfolders of
+ the import path from processing (and from uploading in
+ video_process_and_upload), which the samples of each video are in
+ """
+ import_path: Path = args["import_path"]
+ sample_dirs = [
+ sample_dir
+ for sample_dir in find_sample_dirs(
+ args["video_import_path"], import_path, skip_subfolders=True
+ )
+ # Not a video that failed to sample
+ if sample_dir.is_dir()
+ ]
+ if not sample_dirs:
+ return
+
+ if args.get("desc_path") is None:
+ # Where process writes it for the import path alone, rather than to
+ # STDOUT, as for several import paths
+ args["desc_path"] = str(
+ import_path.joinpath(constants.IMAGE_DESCRIPTION_FILENAME)
+ )
+
+ args["import_path"] = [import_path, *sample_dirs]
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..9b3c30965 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,
@@ -139,8 +142,8 @@ def extract(self) -> types.VideoMetadata:
rate = self._duration_ratio()
self._apply_offset_time(gpx_points, rate=rate)
self._check_coverage(gpx_points, gpx_unix_times, [], start_clock, rate)
- if self.clip_to_video and start_clock is not None:
- gpx_points = self._clip_to_video(gpx_points)
+ if self.clip_to_video:
+ gpx_points = self._clip_to_video(gpx_points, start_clock)
return types.VideoMetadata(
filename=self.video_path,
filesize=utils.get_file_size(self.video_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
@@ -213,8 +217,8 @@ def extract(self) -> types.VideoMetadata:
None if self.sync_mode is SyncMode.REBASE else clock,
rate,
)
- if self.clip_to_video and clock is not None:
- gpx_points = self._clip_to_video(gpx_points)
+ if self.clip_to_video:
+ gpx_points = self._clip_to_video(gpx_points, clock)
return dataclasses.replace(native_video_metadata, points=gpx_points)
@@ -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,21 +249,39 @@ def _read_gpx_points(self) -> list[geo.Point]:
gpx_points.sort(key=lambda p: p.time)
return gpx_points
- def _clip_to_video(self, gpx_points: list[geo.Point]) -> list[geo.Point]:
+ def _parse_tracks(self) -> list[list[geo.Point]]:
"""
- Clip a GPX track to the video when a clock gives its pace along the
- video, so that where the video ends along it is known: the GPS clock
- of the video, which places the track too unless
- --interpolation_use_gpx_start_time starts it with the video, or the
- --video_start_time of sampling. Its points outside the video are the
- positions of none of its frames, and a GPX of a longer trip can run on
- for long after it.
+ 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
- A track that is only assumed to start with a video without a GPS clock
- is left as it is: it may be that of a timelapse, which runs on past
- the end of the video at the pace of the GPS clock rather than that of
- the video.
+ def _clip_to_video(
+ self, gpx_points: list[geo.Point], clock: VideoClock | None
+ ) -> list[geo.Point]:
+ """
+ Clip a GPX track to the video, from video time 0 to where the video
+ ends along it, which is known when a **clock** gives its pace along
+ the video: the GPS clock of the video, which places the track too
+ unless --interpolation_use_gpx_start_time starts it with the video, or
+ the --video_start_time of sampling. Its points outside the video are
+ the positions of none of its frames, and a GPX of a longer trip can
+ run on for long after it.
+
+ A track placed at the start of a video without a GPS clock, assumed to
+ start with it or by --interpolation_use_gpx_start_time, is clipped
+ only at video time 0, which --interpolation_offset_time may move it
+ past, so that the first frames keep a position. Its end is left as it
+ is: it may be that of a timelapse, which runs on past the end of the
+ video at the pace of the GPS clock rather than that of the video.
"""
+ if clock is None:
+ return _clip_track(gpx_points, None)
return _clip_track(gpx_points, _video_duration(self.video_path))
def _place_without_video_gps(
@@ -389,7 +406,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 +437,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 +485,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 +494,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 +523,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. "
)
@@ -587,20 +605,40 @@ def _warn_of_a_long_track(self, gpx_points: T.Sequence[geo.Point]) -> None:
_check_assumed_start): the video could start anywhere along it. Only
warn, since the track of a timelapse without a GPS clock of its own
runs on past the end of the video too.
+
+ With --interpolation_offset_time, it is the offset that places the
+ video along the track, and one way to place it right.
"""
video_duration = _video_duration(self.video_path)
if video_duration is None:
return
- gpx_duration = max(p.time for p in gpx_points) - min(p.time for p in gpx_points)
- if 2 * video_duration < gpx_duration:
+ gpx_start = min(p.time for p in gpx_points)
+ gpx_duration = max(p.time for p in gpx_points) - gpx_start
+ if gpx_duration <= 2 * video_duration:
+ return
+
+ track = (
+ 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"
+ )
+ offset_time = (
+ 0.0 if self.interpolation is None else self.interpolation.offset_time
+ )
+ if offset_time:
+ LOG.warning(
+ f"{track}, and --interpolation_offset_time {offset_time:g} places its "
+ f"start at {_isoformat(gpx_start + offset_time)} along it. Unless the "
+ "video is a timelapse of the track from there, check that the offset "
+ f"is where the video starts along the track, trim the {self.file_format} to the "
+ "video, or sample the video with video_process and --video_start_time"
+ )
+ else:
LOG.warning(
- f"The GPX 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 "
- "and --video_start_time"
+ f"{track}, and is assumed to start with it. Unless the video is a "
+ f"timelapse of the whole track, trim the {self.file_format} to the video, or sample "
+ "the video with video_process and --video_start_time"
)
def _extract_native_video_metadata(self) -> tuple[types.VideoMetadata, bool]:
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..0e6323b6b 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
@@ -67,6 +68,24 @@ def _normalize_path(
return video_dir, video_list
+def find_sample_dirs(
+ video_import_path: Path, import_path: Path, skip_subfolders: bool = False
+) -> list[Path]:
+ """
+ The folder in import_path that sample_video() samples each of the videos
+ in video_import_path into
+ """
+ video_dir, video_list = _normalize_path(video_import_path, skip_subfolders)
+ return [
+ _sample_dir(import_path, video_dir, video_path) for video_path in video_list
+ ]
+
+
+def _sample_dir(import_path: Path, video_dir: Path, video_path: Path) -> Path:
+ # need to resolve video_path because video_dir might be absolute
+ return Path(import_path).joinpath(video_path.resolve().relative_to(video_dir))
+
+
def xor(a: bool, b: bool):
# xor https://stackoverflow.com/a/433161
return bool(a) ^ bool(b)
@@ -132,8 +151,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)]
@@ -218,9 +237,7 @@ def sample_video(
# - video_path: foo/bar/zzz.mp4
# Then:
# - sample_dir: mapillary_sampled_video_frames/bar/zzz.mp4/
- sample_dir = Path(import_path).joinpath(
- video_path.resolve().relative_to(video_dir)
- )
+ sample_dir = _sample_dir(import_path, video_dir, video_path)
LOG.info("Removing the sample directory %s", sample_dir)
if sample_dir.is_dir():
shutil.rmtree(sample_dir)
@@ -237,10 +254,7 @@ def sample_video(
failures: list[tuple[Path, Exception]] = []
for video_path in video_list:
- # need to resolve video_path because video_dir might be absolute
- sample_dir = Path(import_path).joinpath(
- video_path.resolve().relative_to(video_dir)
- )
+ sample_dir = _sample_dir(import_path, video_dir, video_path)
if sample_dir.exists():
LOG.warning(
"Skip sampling video %s as it has been sampled in %s. Specify --rerun to resample it",
@@ -607,27 +621,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/integration/test_video_process.py b/tests/integration/test_video_process.py
index c32bda19f..737d9563b 100644
--- a/tests/integration/test_video_process.py
+++ b/tests/integration/test_video_process.py
@@ -521,6 +521,107 @@ def test_video_process_skips_the_only_video(setup_data: py.path.local):
assert [] == descs
+def test_video_process_skip_subfolders(setup_data: py.path.local):
+ """
+ --skip_subfolders samples only the videos in the given folder, and their
+ samples are processed. Each video is sampled into a folder of its own,
+ which the option hid from processing, so nothing was processed.
+ """
+ pytest_skip_if_not_ffmpeg_installed()
+
+ video = setup_data.join("gopro_data").join("max-360mode.mp4")
+ video_dir = setup_data.join("skip_subfolders")
+ video_dir.ensure("nested", dir=True)
+ video.copy(video_dir.join("max-360mode.mp4"))
+ video.copy(video_dir.join("nested").join("nested-360mode.mp4"))
+ sample_dir = setup_data.join("samples")
+
+ descs = run_video_process_for_descs(
+ [
+ "--video_sample_distance=6",
+ "--skip_subfolders",
+ str(video_dir),
+ str(sample_dir),
+ ]
+ )
+
+ assert descs
+ for desc in descs:
+ assert "error" not in desc, desc
+ assert Path(desc["filename"]).parent == Path(sample_dir, "max-360mode.mp4")
+
+ descs = run_video_process_for_descs(
+ [
+ "--video_sample_distance=6",
+ "--skip_subfolders",
+ str(video),
+ str(setup_data.join("samples_of_one")),
+ ]
+ )
+
+ assert descs
+ assert all("error" not in desc for desc in descs), descs
+
+
+def test_video_process_skip_subfolders_skips_other_subfolders(
+ setup_data: py.path.local,
+):
+ """
+ Of the subfolders of the import path, only the samples of the videos are
+ processed. The description file goes to the import path, as it does
+ without the option.
+ """
+ pytest_skip_if_not_ffmpeg_installed()
+
+ video = setup_data.join("gopro_data").join("max-360mode.mp4")
+ sample_dir = setup_data.join("samples")
+ setup_data.join("images").join("DSC00001.JPG").copy(
+ sample_dir.ensure("photos", dir=True).join("DSC00001.JPG")
+ )
+
+ run_command(
+ ["--video_sample_distance=6", "--skip_subfolders", str(video), str(sample_dir)],
+ "video_process",
+ )
+
+ descs = json.loads(sample_dir.join("mapillary_image_description.json").read())
+ assert descs
+ for desc in descs:
+ assert "error" not in desc, desc
+ assert Path(desc["filename"]).parent == Path(sample_dir, "max-360mode.mp4")
+
+
+def test_video_process_and_upload_skip_subfolders(
+ setup_data: py.path.local, setup_config, setup_upload
+):
+ """
+ The samples are uploaded too, which the option hid from uploading, but
+ not the other subfolders of the import path
+ """
+ pytest_skip_if_not_ffmpeg_installed()
+
+ sample_dir = setup_data.join("samples")
+ setup_data.join("images").join("DSC00001.JPG").copy(
+ sample_dir.ensure("photos", dir=True).join("DSC00001.JPG")
+ )
+
+ run_command(
+ [
+ *["--dry_run", "--user_name", USERNAME],
+ "--video_sample_distance=6",
+ "--skip_subfolders",
+ str(setup_data.join("gopro_data").join("max-360mode.mp4")),
+ str(sample_dir),
+ ],
+ command="video_process_and_upload",
+ )
+ # The samples, in one sequence, and not the photo
+ assert len(setup_upload.listdir("*.jpg")) == len(
+ sample_dir.join("max-360mode.mp4").listdir("*.jpg")
+ )
+ assert len(setup_upload.listdir("*.json")) == 1
+
+
def test_video_process_and_upload_uploads_nothing_if_a_video_fails(
setup_data: py.path.local, setup_config, setup_upload
):
diff --git a/tests/unit/test_gpx_video_clip.py b/tests/unit/test_gpx_video_clip.py
index e6d713dbd..149edee45 100644
--- a/tests/unit/test_gpx_video_clip.py
+++ b/tests/unit/test_gpx_video_clip.py
@@ -6,7 +6,8 @@
"""
A GPX track whose pace along a video a clock gives is clipped to the video,
with a point interpolated at either end, when it geotags the video. One that
-is only assumed to start with a video without a GPS clock is not.
+is only placed at the start of a video without a GPS clock is clipped at the
+start of the video only.
"""
from __future__ import annotations
@@ -18,6 +19,7 @@
from mapillary_tools import types
from mapillary_tools.geotag import factory
from mapillary_tools.geotag.options import (
+ InterpolationOption,
SourceOption,
SourcePathOption,
SourceType,
@@ -145,13 +147,62 @@ def test_the_gpx_source_clips_it(self, tmp_path):
class TestTrackAssumedToStartWithTheVideo:
"""
Without a clock to say where the video ends along the track, a track that
- runs on past it may be that of a timelapse, at the pace of the GPS clock
+ runs on past it may be that of a timelapse, at the pace of the GPS clock.
+ It is clipped only at the start of the video, which
+ --interpolation_offset_time can move it past.
"""
+ START = datetime.datetime(2025, 3, 14, 7, 0, 0, tzinfo=datetime.timezone.utc)
+
def test_without_gps_is_not_clipped(self, tmp_path):
- start = datetime.datetime(2025, 3, 14, 7, 0, 0, tzinfo=datetime.timezone.utc)
- gpx = _write_gpx(tmp_path / "track.gpx", start.timestamp(), 60)
+ gpx = _write_gpx(tmp_path / "track.gpx", self.START.timestamp(), 60)
metadata = GPXVideoExtractor(NO_GPS_VIDEO, gpx, clip_to_video=True).extract()
assert _times(metadata) == [float(t) for t in range(61)]
+
+ def test_moved_by_the_offset_starts_at_the_video(self, tmp_path):
+ """
+ The first frames take their position from a point interpolated at the
+ start of the video, which the offset puts 2.5 seconds along the track
+ """
+ gpx = _write_gpx(tmp_path / "track.gpx", self.START.timestamp(), 60)
+
+ metadata = GPXVideoExtractor(
+ NO_GPS_VIDEO,
+ gpx,
+ interpolation=InterpolationOption(offset_time=2.5),
+ clip_to_video=True,
+ ).extract()
+
+ times = _times(metadata)
+ assert times == [0.0] + [t - 2.5 for t in range(3, 61)]
+ first = metadata.points[0]
+ assert first.lat == pytest.approx(37.0 + 2.5 * 10 / 111_320, abs=1e-7)
+ assert first.get_unix_time() == pytest.approx(self.START.timestamp() + 2.5)
+
+ def test_started_with_the_video_by_the_option_starts_at_the_video(self, tmp_path):
+ """--interpolation_use_gpx_start_time places it as well, without a pace"""
+ gpx = _write_gpx(tmp_path / "track.gpx", self.START.timestamp(), 60)
+
+ metadata = GPXVideoExtractor(
+ NO_GPS_VIDEO,
+ gpx,
+ interpolation=InterpolationOption(offset_time=2.5, use_gpx_start_time=True),
+ clip_to_video=True,
+ ).extract()
+
+ assert _times(metadata) == [0.0] + [t - 2.5 for t in range(3, 61)]
+
+ def test_moved_by_the_offset_is_left_as_it_is_for_sampling(self, tmp_path):
+ """Sampling reads its track only within the video"""
+ gpx = _write_gpx(tmp_path / "track.gpx", self.START.timestamp(), 5)
+
+ metadata = GPXVideoExtractor(
+ NO_GPS_VIDEO,
+ gpx,
+ video_sync=VideoSyncOption(),
+ interpolation=InterpolationOption(offset_time=2.5),
+ ).extract()
+
+ assert _times(metadata) == [t - 2.5 for t in range(6)]
diff --git a/tests/unit/test_gpx_video_long_track.py b/tests/unit/test_gpx_video_long_track.py
index 78a2093a3..8e1b85168 100644
--- a/tests/unit/test_gpx_video_long_track.py
+++ b/tests/unit/test_gpx_video_long_track.py
@@ -17,7 +17,12 @@
from mapillary_tools import types
from mapillary_tools.geotag import factory
-from mapillary_tools.geotag.options import SourceOption, SourcePathOption, SourceType
+from mapillary_tools.geotag.options import (
+ InterpolationOption,
+ SourceOption,
+ SourcePathOption,
+ SourceType,
+)
from mapillary_tools.geotag.video_extractors.gpx import GPXVideoExtractor, SyncMode
from mapillary_tools.geotag.video_extractors.native import NativeVideoExtractor
@@ -59,12 +64,36 @@ def test_a_track_more_than_twice_as_long_as_the_video_warns(tmp_path, caplog):
[warning] = _warnings(caplog)
assert "lasts 1.0 minutes and the video" in warning
- assert "could start anywhere along the track" in warning
+ assert "could start anywhere along the track, and is assumed to start" in warning
assert "video_process and --video_start_time" in warning
# and the track is still assumed to start with the video
assert metadata.points[0].time == 0
+def test_a_track_looked_up_at_an_offset_names_the_offset(tmp_path, caplog):
+ """
+ The offset, not the assumed start, places the video along the track, and
+ is one way to place it right
+ """
+ gpx = _write_gpx(tmp_path / "track.gpx", seconds=60)
+
+ with caplog.at_level(logging.WARNING):
+ GPXVideoExtractor(
+ NO_GPS_VIDEO, gpx, interpolation=InterpolationOption(offset_time=30)
+ ).extract()
+
+ [warning] = _warnings(caplog)
+ assert "could start anywhere along the track" in warning
+ assert "is assumed to start with it" not in warning
+ assert (
+ "--interpolation_offset_time 30 places its start at "
+ "2025-03-14T07:00:30+00:00 along it" in warning
+ )
+ assert "check that the offset is where the video starts" in warning
+ assert "trim the GPX to the video" in warning
+ assert "video_process and --video_start_time" in warning
+
+
def test_a_track_up_to_twice_as_long_as_the_video_is_silent(tmp_path, caplog):
# The video lasts 5.76 seconds
gpx = _write_gpx(tmp_path / "track.gpx", seconds=11)
diff --git a/tests/unit/test_gpx_video_sync.py b/tests/unit/test_gpx_video_sync.py
index 1a7cc252b..f2eb12594 100644
--- a/tests/unit/test_gpx_video_sync.py
+++ b/tests/unit/test_gpx_video_sync.py
@@ -453,7 +453,12 @@ def test_the_factory_hands_them_to_the_gpx_source(self, tmp_path):
)
assert isinstance(metadata, types.VideoMetadata)
- assert metadata.points[0].time == pytest.approx(-2)
+ # process clips the track at the start of the video, which the offset
+ # puts 2 seconds along it
+ assert metadata.points[0].time == 0
+ assert metadata.points[0].get_unix_time() == pytest.approx(
+ GPX_START.timestamp() + 2
+ )
def _write_gpx_tracks(path: Path, tracks: list[list[str]]) -> Path:
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")