diff --git a/README.md b/README.md index d791e4690..dc1534441 100644 --- a/README.md +++ b/README.md @@ -327,7 +327,10 @@ mapillary_tools upload MY_SAMPLES --desc_path=/tmp/mapillary_description.json If you use external GPS devices for mapping, you will need to geotag your captures with the external GPS tracks. To geotag images with a GPX file, the capture time (extracted from EXIF tag "Date/Time Original" or "GPS Date/Time") is minimally required. -It is used to locate the images along the GPS tracks. +It is used to locate the images along the GPS tracks. Images captured before the track starts, after it ends, or in a +gap of more than an hour between two of its points (such as between two trips in one GPX file) fall outside the track, +and are not geotagged from it. The same applies to videos. To allow longer gaps, set the environment variable +`MAPILLARY_TOOLS_MAX_GPX_GAP_SECONDS` (3600 by default). ```sh mapillary_tools process MY_IMAGE_DIR --geotag_source "gpx" --geotag_source_path MY_EXTERNAL_GPS.gpx @@ -462,6 +465,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/constants.py b/mapillary_tools/constants.py index cab9e9b1a..d7dfdf89f 100644 --- a/mapillary_tools/constants.py +++ b/mapillary_tools/constants.py @@ -82,6 +82,11 @@ def _parse_scaled_integers( ################### USER_DATA_DIR = appdirs.user_data_dir(appname="mapillary_tools", appauthor="Mapillary") PROMPT_DISABLED: bool = _yes_or_no(os.getenv(_ENV_PREFIX + "PROMPT_DISABLED", "NO")) +# In seconds. Consecutive points of a GPX or NMEA track further apart than +# this leave a gap that no position is interpolated across, such as a GPS +# logger switched off between trips: the time inside it is outside the track, +# like the time before its start or after its end +MAX_GPX_GAP_SECONDS = float(os.getenv(_ENV_PREFIX + "MAX_GPX_GAP_SECONDS", 3600)) ############################ diff --git a/mapillary_tools/exif_read.py b/mapillary_tools/exif_read.py index 8a5dd3b8c..65189bd17 100644 --- a/mapillary_tools/exif_read.py +++ b/mapillary_tools/exif_read.py @@ -45,6 +45,18 @@ def eval_frac(value: Ratio) -> float: def gps_to_decimal(values: tuple[Ratio, Ratio, Ratio]) -> float | None: + """ + Convert degrees, minutes and seconds to decimal degrees, or None if they + are not a coordinate: a GPS IFD that a camera wrote at bad offsets reads + as minutes or seconds of more than 60 (60 itself may be 59.996 rounded) + + >>> gps_to_decimal((Ratio(10, 1), Ratio(22, 1), Ratio(30, 1))) + 10.375 + >>> round(gps_to_decimal((Ratio(10, 1), Ratio(22, 1), Ratio(60, 1))), 6) + 10.383333 + >>> gps_to_decimal((Ratio(0, 1), Ratio(1, 4), Ratio(381, 5))) is None + True + """ try: deg, min, sec, *_ = values except (TypeError, ValueError): @@ -61,6 +73,8 @@ def gps_to_decimal(values: tuple[Ratio, Ratio, Ratio]) -> float | None: seconds = eval_frac(sec) except ZeroDivisionError: return None + if not (0 <= minutes <= 60 and 0 <= seconds <= 60): + return None return degrees + minutes / 60 + seconds / 3600 diff --git a/mapillary_tools/exiftool_read_video.py b/mapillary_tools/exiftool_read_video.py index f666944ae..178849b7d 100644 --- a/mapillary_tools/exiftool_read_video.py +++ b/mapillary_tools/exiftool_read_video.py @@ -30,6 +30,7 @@ "XMP-aux": "http://ns.exiftool.org/XMP/XMP-aux/1.0/", "DJI": "http://ns.exiftool.org/MakerNotes/DJI/1.0/", "XMP-drone-dji": "http://ns.exiftool.org/XMP/XMP-drone-dji/1.0/", + "LIGO": "http://ns.exiftool.org/QuickTime/LIGO/1.0/", **{ f"Track{track_id}": f"http://ns.exiftool.org/QuickTime/Track{track_id}/1.0/" for track_id in range(1, MAX_TRACK_ID + 1) @@ -310,6 +311,121 @@ def _aggregate_gps_track( return track +def _group_gps_records( + elements: T.Iterable[ET.Element], + tags: T.Collection[str], + record_start_tag: str | None = None, + first_tag: str | None = None, +) -> T.Generator[dict[str, str], None, None]: + """ + Group the tags that ExifTool lists for one GPS record after another, one + dict per record from tag (expanded) to text. A record starts at + **record_start_tag**, at **first_tag**, which ExifTool lists first in a + record that has it, or at a tag that the record so far already has. + + A tag that a record does not have is left out of its dict, rather than + taken from a later record: a BlackVue writes no GPSTrack while stationary. + + >>> elements = [] + >>> for tag, text in [ + ... ("SampleTime", "0"), ("GPSLatitude", "1"), ("GPSTrack", "90"), + ... ("SampleTime", "1"), ("GPSLatitude", "2"), + ... ("GPSLatitude", "3"), ("GPSTrack", "180"), + ... ]: + ... elements.append(ET.Element(expand_tag(f"QuickTime:{tag}"))) + ... elements[-1].text = text + >>> records = _group_gps_records( + ... elements, + ... ["QuickTime:GPSLatitude", "QuickTime:GPSTrack"], + ... record_start_tag="QuickTime:SampleTime", + ... ) + >>> [sorted(record.values()) for record in records] + [['1', '90'], ['2'], ['180', '3']] + """ + expanded_tags = {expand_tag(tag) for tag in tags} + expanded_record_start_tag = ( + expand_tag(record_start_tag) if record_start_tag is not None else None + ) + expanded_first_tag = expand_tag(first_tag) if first_tag is not None else None + + record: dict[str, str] = {} + for element in elements: + if element.tag == expanded_record_start_tag: + if record: + yield record + record = {} + elif element.tag in expanded_tags and element.text is not None: + if record and (element.tag in record or element.tag == expanded_first_tag): + yield record + record = {} + record[element.tag] = element.text + if record: + yield record + + +def _aggregate_gps_track_by_record( + elements: T.Iterable[ET.Element], + time_tag: str, + lon_tag: str, + lat_tag: str, + alt_tag: str, + direction_tag: str, + record_start_tag: str | None = None, +) -> list[GPSPoint]: + """ + Aggregate the GPS records that ExifTool lists one after another, each with + its GPS time in **time_tag**, into a track whose times start at 0. Every + point takes its altitude and direction from its own record. + """ + expanded_time_tag = expand_tag(time_tag) + expanded_lon_tag = expand_tag(lon_tag) + expanded_lat_tag = expand_tag(lat_tag) + expanded_alt_tag = expand_tag(alt_tag) + expanded_direction_tag = expand_tag(direction_tag) + + track: list[GPSPoint] = [] + for record in _group_gps_records( + elements, + [time_tag, lon_tag, lat_tag, alt_tag, direction_tag], + record_start_tag=record_start_tag, + first_tag=time_tag, + ): + time_text = record.get(expanded_time_tag) + dt = exif_read.parse_gps_datetime(time_text) if time_text is not None else None + timestamp = geo.as_unix_time(dt) if dt is not None else None + lon = _maybe_float(record.get(expanded_lon_tag)) + lat = _maybe_float(record.get(expanded_lat_tag)) + if timestamp is None or lon is None or lat is None: + continue + + track.append( + GPSPoint( + time=timestamp, + lon=lon, + lat=lat, + alt=_maybe_float(record.get(expanded_alt_tag)), + angle=_maybe_float(record.get(expanded_direction_tag)), + epoch_time=timestamp, + fix=None, + precision=None, + ground_speed=None, + ) + ) + + track.sort(key=lambda point: point.time) + + # A BlackVue video holds a copy of its GPS log in the low resolution + # recording embedded in it, which ExifTool lists as well + track = _deduplicate_gps_points(track, same_gps_point=_same_gps_point) + + if track: + first_time = track[0].time + for point in track: + point.time = point.time - first_time + + return track + + def _aggregate_samples( elements: T.Iterable[ET.Element], sample_time_tag: str, @@ -436,6 +552,12 @@ def extract_gps_track(self) -> list[geo.Point]: if track_with_fix: return T.cast(T.List[geo.Point], track_with_fix) + # dashcams that write LIGOGPSINFO (e.g. GKU, Abask), which ExifTool + # decrypts. Its GPS times are in UTC, unlike the clocks of these cameras + track_with_fix = self._extract_gps_track_from_quicktime(namespace="LIGO") + if track_with_fix: + return T.cast(T.List[geo.Point], track_with_fix) + return [] def _extract_make_and_model(self) -> tuple[str | None, str | None]: @@ -585,12 +707,16 @@ def _extract_gps_track_from_quicktime( ): return [] - return _aggregate_gps_track( - self._texts_by_tag, + root = self.etree.getroot() + if root is None: + raise ValueError("ElementTree root is None") + + return _aggregate_gps_track_by_record( + root, time_tag=f"{namespace}:GPSDateTime", lon_tag=f"{namespace}:GPSLongitude", lat_tag=f"{namespace}:GPSLatitude", alt_tag=f"{namespace}:GPSAltitude", - gps_time_tag=f"{namespace}:GPSDateTime", direction_tag=f"{namespace}:GPSTrack", + record_start_tag=f"{namespace}:SampleTime", ) diff --git a/mapillary_tools/geotag/factory.py b/mapillary_tools/geotag/factory.py index cb6ede822..bb7d20b3b 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( @@ -163,10 +165,14 @@ def process( # Collection: ABC for sized iterable container classes paths: T.Iterable[Path], options: T.Collection[SourceOption], + # --filetypes, which the filetypes of every option are limited to already + filetypes: set[types.FileType] | None = None, ) -> list[types.MetadataOrError]: if not options: raise ValueError("No geotag options provided") + video_types = _video_types_asked_for(filetypes) + final_metadatas: list[types.MetadataOrError] = [] # Paths (image path or video path) that will be sent to the next geotag process @@ -234,7 +240,20 @@ def process( ] = [] for metadata in image_metadata_or_errors + video_metadata_or_errors: - if isinstance(metadata, types.ErrorMetadata) and _is_reprocessable( + if video_types is not None and _is_of_other_type( + metadata, option, video_types + ): + # Its type, known only once it is read, is not one asked for: + # leave it out, as the files of other kinds + LOG.debug( + "Skipping %s, which is not of the file types %s", + metadata.filename, + filetypes, + ) + reprocessable_paths.remove(metadata.filename) + pending_errors.pop(metadata.filename, None) + gpx_rejections.pop(metadata.filename, None) + elif isinstance(metadata, types.ErrorMetadata) and _is_reprocessable( metadata ): # Leave what it is for the next geotag process @@ -272,11 +291,14 @@ def process( break final_metadatas.extend(pending_errors.values()) - final_metadatas.extend( - _describe_unread_video(path, sources) - for path, sources in unread_videos.items() - if path in reprocessable_paths and path not in pending_errors - ) + # Limited to some types of video, a video that no source read may not be + # of any of them, and is left out as the videos found to be of other types + if video_types is None: + final_metadatas.extend( + _describe_unread_video(path, sources) + for path, sources in unread_videos.items() + if path in reprocessable_paths and path not in pending_errors + ) return final_metadatas @@ -342,6 +364,81 @@ def _warn_geotagged_despite_gpx( LOG.debug("Images outside %s: %s", gpx_track, [p.name for p in image_paths]) +def _video_types_asked_for( + filetypes: set[types.FileType] | None, +) -> set[types.FileType] | None: + """ + The types of video that --filetypes limits processing to, or None if it + lets videos of every type through + + >>> _video_types_asked_for({types.FileType.IMAGE, types.FileType.GOPRO}) + {} + >>> _video_types_asked_for({types.FileType.GOPRO, types.FileType.VIDEO}) is None + True + >>> _video_types_asked_for(None) is None + True + """ + if filetypes is None or types.FileType.VIDEO in filetypes: + return None + return filetypes & types.NATIVE_VIDEO_FILETYPES + + +def _is_of_other_type( + metadata: types.MetadataOrError, + option: SourceOption, + video_types: set[types.FileType], +) -> bool: + """ + Whether the source **option** found a video to be of none of the types + **video_types** asked for, which _filter_images_and_videos() cannot tell by + its name: it read the video as another type, or it is the native reader, + which tries only the readers of the types it is given, and finds no GPS + data only in a video that none of them recognizes. A generic video is of a + type that is not known, and is kept. + + >>> gopro = {types.FileType.GOPRO} + >>> video = types.VideoMetadata(Path("v.mp4"), types.FileType.VIDEO, points=[]) + >>> _is_of_other_type(video, SourceOption(SourceType.GPX), gopro) + False + >>> blackvue = types.VideoMetadata(Path("v.mp4"), types.FileType.BLACKVUE, points=[]) + >>> _is_of_other_type(blackvue, SourceOption(SourceType.EXIFTOOL_RUNTIME), gopro) + True + >>> not_found = types.describe_error_metadata( + ... exceptions.MapillaryVideoGPSNotFoundError("No GPS data found from the video"), + ... Path("v.mp4"), + ... filetype=types.FileType.VIDEO, + ... ) + >>> _is_of_other_type(not_found, SourceOption(SourceType.NATIVE, filetypes=gopro), gopro) + True + + Limited further by its own filetypes, a source found only that the video is + of none of those, and leaves it to the next source: + + >>> _is_of_other_type( + ... not_found, + ... SourceOption(SourceType.NATIVE, filetypes=gopro), + ... {types.FileType.GOPRO, types.FileType.BLACKVUE}, + ... ) + False + """ + if isinstance(metadata, types.VideoMetadata): + return ( + metadata.filetype in types.NATIVE_VIDEO_FILETYPES + and metadata.filetype not in video_types + ) + + if isinstance(metadata, types.ErrorMetadata): + return ( + option.source is SourceType.NATIVE + and isinstance(metadata.error, exceptions.MapillaryVideoGPSNotFoundError) + and option.filetypes is not None + and types.FileType.VIDEO not in option.filetypes + and video_types <= option.filetypes + ) + + return False + + def _is_reprocessable(metadata: types.MetadataOrError) -> bool: if isinstance(metadata, types.ErrorMetadata): if isinstance( @@ -478,8 +575,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 +621,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_gpx.py b/mapillary_tools/geotag/geotag_images_from_gpx.py index f06c1d1f7..ef9fdca73 100644 --- a/mapillary_tools/geotag/geotag_images_from_gpx.py +++ b/mapillary_tools/geotag/geotag_images_from_gpx.py @@ -5,7 +5,9 @@ from __future__ import annotations +import bisect import dataclasses +import datetime import logging import sys import typing as T @@ -16,10 +18,11 @@ else: from typing_extensions import override -from .. import exceptions, geo, types +from .. import constants, exceptions, geo, types from ..serializer.description import build_capture_time from .base import GeotagImagesFromGeneric from .geotag_images_from_exif import ImageEXIFExtractor +from .video_extractors.gpx import describe_gap, track_gaps LOG = logging.getLogger(__name__) @@ -33,12 +36,16 @@ def __init__( use_image_start_time: bool = False, offset_time: float = 0.0, num_processes: int | None = None, + # Whether an image in a gap of the track longer than + # constants.MAX_GPX_GAP_SECONDS is outside it + check_gaps: bool = True, ): super().__init__(num_processes=num_processes) self.points = points self.use_gpx_start_time = use_gpx_start_time self.use_image_start_time = use_image_start_time self.offset_time = offset_time + self.check_gaps = check_gaps # Why the track could not be read, reported for every image instead self.track_error: Exception | None = None @@ -46,6 +53,7 @@ def _interpolate_image_metadata_along( self, image_metadata: types.ImageMetadata, sorted_points: T.Sequence[geo.Point], + gaps: T.Sequence[tuple[geo.Point, geo.Point]] = (), ) -> types.ImageMetadata: assert sorted_points, "must have at least one point" @@ -75,6 +83,28 @@ def _interpolate_image_metadata_along( gpx_end_time=gpx_end_time, ) + # The last gap that starts at or before the image + idx = bisect.bisect_right( + [before.time for before, _ in gaps], image_metadata.time + ) + if idx: + before, after = gaps[idx - 1] + delta = min( + image_metadata.time - before.time, after.time - image_metadata.time + ) + # with the tolerance of 1ms, as at the ends of the track + if 0.001 < delta: + captured_at = datetime.datetime.fromtimestamp( + image_metadata.time, tz=datetime.timezone.utc + ).isoformat() + track_gap = describe_gap(before, after, before.time, after.time) + raise exceptions.MapillaryOutsideGPXTrackError( + f"The GPX track has {track_gap}, which the image time {captured_at} falls in", + image_time=build_capture_time(image_metadata.time), + gpx_start_time=build_capture_time(sorted_points[0].time), + gpx_end_time=build_capture_time(sorted_points[-1].time), + ) + interpolated = geo.interpolate(sorted_points, image_metadata.time) return dataclasses.replace( @@ -158,11 +188,17 @@ def to_description( # TODO: this time modification seems to affect final capture times image_metadata.time += image_time_offset + gaps = ( + track_gaps(sorted_points, constants.MAX_GPX_GAP_SECONDS) + if self.check_gaps + else [] + ) + for image_metadata in sorted_image_metadatas: try: final_metadatas.append( self._interpolate_image_metadata_along( - image_metadata, sorted_points + image_metadata, sorted_points, gaps ) ) except exceptions.MapillaryOutsideGPXTrackError as ex: 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_images_from_video.py b/mapillary_tools/geotag/geotag_images_from_video.py index a6035740f..886ff7de5 100644 --- a/mapillary_tools/geotag/geotag_images_from_video.py +++ b/mapillary_tools/geotag/geotag_images_from_video.py @@ -144,6 +144,9 @@ def to_description( use_image_start_time=True, offset_time=self.offset_time, num_processes=self.num_processes, + # Gaps between trips are those of a GPX file, not of the + # video's own track + check_gaps=False, ) image_metadatas = geotag.to_description(sample_paths) 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/image_extractors/exif.py b/mapillary_tools/geotag/image_extractors/exif.py index 01470964f..0997f7a16 100644 --- a/mapillary_tools/geotag/image_extractors/exif.py +++ b/mapillary_tools/geotag/image_extractors/exif.py @@ -6,6 +6,7 @@ from __future__ import annotations import contextlib +import datetime import sys import typing as T from pathlib import Path @@ -15,10 +16,25 @@ else: from typing_extensions import override -from ... import exceptions, exif_read, geo, types, utils +from ... import exceptions, exif_read, types, utils from .base import BaseImageExtractor +def _capture_unix_time(capture_time: datetime.datetime) -> float: + """ + The Unix time of the capture time of an image, assuming that a naive one + is in the local timezone. Some cameras write 0001:01:01 00:00:00 for no + date, which has no Unix time: raise as for an image with no capture time, + rather than let geo.as_unix_time() move it to 1970. + """ + try: + return capture_time.timestamp() + except (ValueError, OverflowError, OSError): + raise exceptions.MapillaryGeoTaggingError( + f"Unable to extract timestamp from the image: invalid capture time {capture_time}" + ) + + class ImageEXIFExtractor(BaseImageExtractor): def __init__(self, image_path: Path, skip_lonlat_error: bool = False): super().__init__(image_path) @@ -46,11 +62,12 @@ def extract(self) -> types.ImageMetadata: raise exceptions.MapillaryGeoTaggingError( "Unable to extract timestamp from the image" ) + unix_time = _capture_unix_time(capture_time) image_metadata = types.ImageMetadata( filename=self.image_path, filesize=utils.get_file_size(self.image_path), - time=geo.as_unix_time(capture_time), + time=unix_time, lat=lat, lon=lon, alt=exif.extract_altitude(), 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..d00f29d51 100644 --- a/mapillary_tools/geotag/video_extractors/gpx.py +++ b/mapillary_tools/geotag/video_extractors/gpx.py @@ -23,7 +23,7 @@ else: from typing_extensions import override -from ... import exceptions, geo, telemetry, types, utils +from ... import constants, exceptions, geo, telemetry, types, utils from ...camm import camm_parser from ...mp4 import construct_mp4_parser as cparser, simple_mp4_parser as sparser from ...mp4.mp4_sample_parser import MovieBoxParser @@ -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]]: + """ + Parse the tracks of the GPX file, each a list of points with Unix times """ - 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. + 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( @@ -357,7 +374,10 @@ def _check_coverage( A track that misses the video has no position for any moment of it, so it raises when its placement says so: the clock it was placed by, or --interpolation_offset_time. A track that covers only part of the video - warns, and one that covers all of it is silent. + warns, and one that covers all of it is silent. The video can fall + between the ends of the track and still miss it, or part of it, in a + gap of more than constants.MAX_GPX_GAP_SECONDS between two of its + points. Without the duration of the video, the video is known only up to its last GPS point, and a track that starts after it may still overlap the @@ -378,6 +398,18 @@ def _check_coverage( gpx_first = min(p.time for p in gpx_points) gpx_last = max(p.time for p in gpx_points) + # The gaps of the track, and its parts between them, in video seconds + gap_points = track_gaps( + sorted(gpx_points, key=lambda p: p.time), + constants.MAX_GPX_GAP_SECONDS / rate, + ) + gaps = [(before.time, after.time) for before, after in gap_points] + parts = list( + zip( + [gpx_first, *(end for _, end in gaps)], + [*(start for start, _ in gaps), gpx_last], + ) + ) offset_time = ( 0.0 if self.interpolation is None else self.interpolation.offset_time ) @@ -389,7 +421,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: @@ -404,10 +436,17 @@ def _check_coverage( ) # In GPS seconds, which is where clock and time zone mistakes are made - gap = max(gpx_first - video_end, -gpx_last) * rate + gap = min(max(first - video_end, -last) for first, last in parts) * rate if gap <= 0: # In video seconds - uncovered = max(gpx_first, 0.0) + max(video_end - gpx_last, 0.0) + uncovered = ( + max(gpx_first, 0.0) + + max(video_end - gpx_last, 0.0) + + sum( + max(min(end, video_end) - max(start, 0.0), 0.0) + for start, end in gaps + ) + ) if _LOGGER_TOLERANCE_SECONDS < uncovered * rate: LOG.warning( f"{gpx_track} covers only part of {video}{offset_clause}: " @@ -420,19 +459,42 @@ 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}. " - f"{advice}" + # Between the ends of the track, the video falls in one of its gaps + surrounding_gap = next( + ( + (before, after) + for before, after in gap_points + if before.time < 0 and video_end < after.time + ), + None, ) + if surrounding_gap is None: + message = ( + f"{gpx_track} misses {video} by {_format_duration(gap)}{offset_clause}. " + f"{advice}" + ) + else: + before, after = surrounding_gap + # At the Unix times the points were recorded at + track_gap = describe_gap( + before, + after, + lookup_start + before.time * rate, + lookup_start + after.time * rate, + ) + message = ( + f"{gpx_track} has {track_gap}, which {video} falls in{offset_clause}. " + f"{advice}" + ) if (clock is not None or offset_time) and ( video_duration is not None or _IMPLAUSIBLE_OFFSET_SECONDS < gap @@ -468,7 +530,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 +539,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 +568,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 +650,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"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 --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"{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]: @@ -730,6 +813,46 @@ def _median_step_rate(cls, pairs: T.Sequence[tuple[float, float]]) -> float | No return statistics.median(steps) if steps else None +def track_gaps( + points: T.Sequence[geo.Point], max_gap: float +) -> list[tuple[geo.Point, geo.Point]]: + """ + The gaps of a track, sorted by time, between consecutive points more than + **max_gap** seconds apart, as the points either side + + >>> track = [geo.Point(time=t, lat=0, lon=0, alt=None, angle=None) for t in (0, 1, 5000, 5001, 9000)] + >>> [(before.time, after.time) for before, after in track_gaps(track, 3600)] + [(1, 5000), (5001, 9000)] + >>> [(before.time, after.time) for before, after in track_gaps(track, 4000)] + [(1, 5000)] + >>> track_gaps(track[:2], 3600) + [] + """ + return [ + (left, right) + for left, right in geo.pairwise(points) + if max_gap < right.time - left.time + ] + + +def describe_gap( + before: geo.Point, after: geo.Point, before_time: float, after_time: float +) -> str: + """ + Describe a gap of a track between the points **before** and **after**, + which were recorded at the Unix times **before_time** and **after_time** + + >>> p = lambda lat, lon: geo.Point(time=0, lat=lat, lon=lon, alt=None, angle=None) + >>> describe_gap(p(55.174859, 61.256901), p(55.172986, 61.256636), 1739074271, 1739124451) + 'a gap of 13.9 hours, from 2025-02-09T04:11:11+00:00 at (55.174859, 61.256901) to 2025-02-09T18:07:31+00:00 at (55.172986, 61.256636)' + """ + return ( + f"a gap of {_format_duration(after_time - before_time)}, " + f"from {_isoformat(before_time)} at ({round(before.lat, 7)}, {round(before.lon, 7)}) " + f"to {_isoformat(after_time)} at ({round(after.lat, 7)}, {round(after.lon, 7)})" + ) + + def _clip_track( points: T.Sequence[geo.Point], duration: float | None ) -> list[geo.Point]: 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/gpmf/gpmf_gps_filter.py b/mapillary_tools/gpmf/gpmf_gps_filter.py index 61f2ff908..125bf4208 100644 --- a/mapillary_tools/gpmf/gpmf_gps_filter.py +++ b/mapillary_tools/gpmf/gpmf_gps_filter.py @@ -48,7 +48,17 @@ def remove_outliers( return sequence max_speed = gps_filter.upper_whisker(ground_speeds) - merged = gps_filter.dbscan(sequences, gps_filter.speed_le(max_speed)) + # Pieces whose ends are no farther apart than the points within a piece + # were split only by the points between them, such as a single sample + # far off the track. Over the fraction of a second such a sample lasts, + # GPS jitter alone can read faster than the camera moved, so merge them + # whatever the speed between them + merged = gps_filter.dbscan( + sequences, + gps_filter.either( + gps_filter.distance_le(max_distance), gps_filter.speed_le(max_speed) + ), + ) LOG.debug( "Found %d sequences after merging with max speed %f", len(merged), max_speed ) diff --git a/mapillary_tools/gpmf/gps_filter.py b/mapillary_tools/gpmf/gps_filter.py index 005191e40..f6cefae9e 100644 --- a/mapillary_tools/gpmf/gps_filter.py +++ b/mapillary_tools/gpmf/gps_filter.py @@ -80,6 +80,18 @@ def _split_or_not(p1, p2): return _split_or_not +def distance_le( + max_distance: float, +) -> Decider: + """Return a callable that checks if two points are no farther apart than the given distance.""" + + def _f(p1, p2): + distance = geo.gps_distance((p1.lat, p1.lon), (p2.lat, p2.lon)) + return distance <= max_distance + + return _f + + def speed_le(max_speed: float) -> Decider: """Return a callable that checks if the speed between two points are slower than the given speed.""" @@ -100,6 +112,16 @@ def _f(p1, p2): return _f +def either( + s1: Decider, + s2: Decider, +) -> Decider: + def _f(p1, p2): + return s1(p1, p2) or s2(p1, p2) + + return _f + + def dbscan( sequences: T.Sequence[PointSequence], merge_or_not: Decider, diff --git a/mapillary_tools/process_geotag_properties.py b/mapillary_tools/process_geotag_properties.py index 5a1b5493b..61ef91fb9 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 @@ -97,7 +108,7 @@ def process_geotag_properties( image_paths = utils.find_images(import_paths, skip_subfolders=skip_subfolders) video_paths = utils.find_videos(import_paths, skip_subfolders=skip_subfolders) - metadata_or_errors = process(image_paths + video_paths, options) + metadata_or_errors = process(image_paths + video_paths, options, filetypes) return metadata_or_errors 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/cli/gps_filter.py b/tests/cli/gps_filter.py index f63ae045b..0907581c1 100644 --- a/tests/cli/gps_filter.py +++ b/tests/cli/gps_filter.py @@ -122,7 +122,12 @@ def _filter_outliers( return points max_speed = gps_filter.upper_whisker(ground_speeds) - merged = gps_filter.dbscan(subseqs, gps_filter.speed_le(max_speed)) + merged = gps_filter.dbscan( + subseqs, + gps_filter.either( + gps_filter.distance_le(max_distance), gps_filter.speed_le(max_speed) + ), + ) return T.cast( T.List[telemetry.GPSPoint], diff --git a/tests/integration/test_process.py b/tests/integration/test_process.py index f21c093ba..46385de91 100644 --- a/tests/integration/test_process.py +++ b/tests/integration/test_process.py @@ -617,11 +617,11 @@ def test_process_videos_with_an_image_only_geotag_source(setup_data: py.path.loc def test_process_unsupported_filetypes(setup_data: py.path.local): video_dir = setup_data.join("gopro_data") for filetypes in ["blackvue"]: + # GoPro videos, which are not processed, as images are not descs = run_process_for_descs( ["--filetypes", filetypes, "--geotag_source=native", str(video_dir)] ) - assert 2 == len(descs) - assert 2 == len(find_desc_errors(descs)) + assert 0 == len(descs) for filetypes in ["image"]: descs = run_process_for_descs( @@ -630,6 +630,39 @@ def test_process_unsupported_filetypes(setup_data: py.path.local): assert 0 == len(descs) +def test_process_filetypes_of_videos(setup_data: py.path.local): + """ + The type of a video is known only once it is read. ExifTool, which the + default sources fall back to, processed the videos of other types anyway + """ + videos = [ + str(setup_data.join("gopro_data").join("max-360mode.mp4")), + str(setup_data.join("videos").join("sample-5s.mp4")), + ] + descs = run_process_for_descs(["--filetypes=gopro", *videos]) + assert [(Path(d["filename"]).name, d["filetype"]) for d in descs] == [ + ("max-360mode.mp4", "gopro") + ] + assert "error" not in descs[0] + + assert [] == run_process_for_descs(["--filetypes=blackvue", *videos]) + + # Limited by its own filetypes, a source leaves the video to the next + descs = run_process_for_descs( + [ + *[ + "--video_geotag_source", + '{"source": "native", "filetypes": ["blackvue"]}', + ], + *["--video_geotag_source", "native"], + videos[0], + ] + ) + assert [(Path(d["filename"]).name, d["filetype"]) for d in descs] == [ + ("max-360mode.mp4", "gopro") + ] + + def test_process_video_geotag_source_with_gpx_specified(setup_data: py.path.local): video_path = setup_data.join("videos").join("sample-5s.mp4") gpx_file = setup_data.join("gpx").join("sf_30km_h.gpx") @@ -799,6 +832,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_exifread.py b/tests/unit/test_exifread.py index 3bfb14b6f..be49b0012 100644 --- a/tests/unit/test_exifread.py +++ b/tests/unit/test_exifread.py @@ -366,6 +366,68 @@ def test_keeps_fractional_minutes(self): # Tests for extract_camera_uuid +def _read_lon_lat_image( + lat: T.Tuple[float, float, float], + lon: T.Tuple[float, float, float], + lat_ref: T.Optional[str], + lon_ref: T.Optional[str], +) -> ExifRead: + """ + Build an in-memory JPEG with the given GPSLatitude/GPSLongitude (degrees, + minutes, seconds) and refs (None for no tag), and read it back. + """ + img = Image.new("RGB", (4, 4), (0, 0, 0)) + exif = img.getexif() + gps = exif.get_ifd(IFD.GPSInfo) + gps[GPS.GPSLatitude] = tuple(IFDRational(v) for v in lat) + gps[GPS.GPSLongitude] = tuple(IFDRational(v) for v in lon) + if lat_ref is not None: + gps[GPS.GPSLatitudeRef] = lat_ref + if lon_ref is not None: + gps[GPS.GPSLongitudeRef] = lon_ref + exif[IFD.GPSInfo.value] = gps + buf = io.BytesIO() + img.save(buf, format="JPEG", exif=exif) + return ExifRead(io.BytesIO(buf.getvalue())) + + +class TestExtractLonLatFromEXIF: + """Test extract_lon_lat from EXIF tags""" + + def test_valid_coordinates(self): + reader = _read_lon_lat_image((56, 54, 20.5), (24, 25, 3.5), "N", "W") + lon, lat = reader.extract_lon_lat() + assert lat == pytest.approx(56 + 54 / 60 + 20.5 / 3600) + assert lon == pytest.approx(-(24 + 25 / 60 + 3.5 / 3600)) + + def test_without_refs(self): + # Some cameras write no refs, or empty ones, for the northern and + # eastern hemispheres + for ref in [None, ""]: + reader = _read_lon_lat_image((56, 54, 20.5), (24, 25, 3.5), ref, ref) + lon, lat = reader.extract_lon_lat() + assert lat == pytest.approx(56 + 54 / 60 + 20.5 / 3600) + assert lon == pytest.approx(24 + 25 / 60 + 3.5 / 3600) + + def test_accepts_rounded_up_seconds(self): + # 59.996 seconds, written rounded to 60.00 + reader = _read_lon_lat_image((56, 54, 60), (24, 25, 60), "N", "E") + lon, lat = reader.extract_lon_lat() + assert lat == pytest.approx(56 + 55 / 60) + assert lon == pytest.approx(24 + 26 / 60) + + def test_rejects_seconds_out_of_range(self): + # What a GPS IFD written at bad offsets reads as (FLIR P60 PAL) + reader = _read_lon_lat_image((0, 0.25, 76.2), (0, 0.25, 76.2), "", "") + assert reader.extract_lon_lat() is None + + def test_rejects_minutes_out_of_range(self): + reader = _read_lon_lat_image( + (16777215, 16777215, 167772.15), (24, 25, 3.5), "N", "E" + ) + assert reader.extract_lon_lat() is None + + class TestExtractCameraUuidFromEXIF: """Test extract_camera_uuid from EXIF tags""" diff --git a/tests/unit/test_exiftool_read_video.py b/tests/unit/test_exiftool_read_video.py index 9756c806c..491891bb4 100644 --- a/tests/unit/test_exiftool_read_video.py +++ b/tests/unit/test_exiftool_read_video.py @@ -1325,6 +1325,39 @@ def test_gps9_track_carries_dop_through_extract(self): pytest.approx(207.0), ] + def test_ligo_gps(self): + """ + Dashcams that write LIGOGPSINFO (GKU, Abask) keep their GPS in the LIGO + group, the last path in extract_gps_track. Its times are GPS times in + UTC, with or without a Z, not the camera clock in DateTimeOriginal. + """ + xml = """\ + + + + 2025:02:13 09:07:11Z + 44.11392 + 9.8478218 + 3.735484 + 2025:02:13 01:07:10 + 2025:02:13 09:07:12 + 44.11392 + 9.8478371 + 3.96328 + 2025:02:13 01:07:11 + + +""" + reader = ExifToolReadVideo(_etree_from_xml(xml)) + track = reader.extract_gps_track() + assert [(p.time, p.lat, p.lon) for p in track] == [ + (0.0, 44.11392, 9.8478218), + (1.0, 44.11392, 9.8478371), + ] + # 2025-02-13T09:07:11Z + assert track[0].epoch_time == pytest.approx(1739437631.0) + def test_empty_gps_track(self): """When no GPS data is present, returns empty list.""" xml = """\ @@ -1402,6 +1435,106 @@ def test_custom_namespace_not_present(self): track = reader._extract_gps_track_from_quicktime(namespace="Insta360") assert track == [] + def test_each_point_takes_the_fields_of_its_own_record(self): + """ + A BlackVue writes no GPSTrack while stationary, nor a GPSAltitude in + its last record. Neither shifts a later value onto an earlier point, + and the copy of the GPS log in the embedded low resolution recording + does not double the track. + """ + records = """\ + -10799.148 + 2019:09:02 10:23:28.00Z + 37.2655468 + 28.213496 + 133.46 + 402.9 + -10798.149 + 2019:09:02 10:23:29.00Z + 37.2655468 + 28.213496 + 402.8 + -10797.15 + 2019:09:02 10:23:30.00Z + 37.2654601 + 28.2136115 + 136.76 +""" + xml = f"""\ + + + +{records} test_NFS.mp4 +{records} + +""" + reader = ExifToolReadVideo(_etree_from_xml(xml)) + track = reader._extract_gps_track_from_quicktime() + assert [(p.time, p.angle, p.alt) for p in track] == [ + (0.0, 133.46, 402.9), + (1.0, None, 402.8), + (2.0, 136.76, None), + ] + + def test_a_record_without_a_time_keeps_the_next_time_off_it(self): + """ + Without SampleTime between them, a record that lacks GPSDateTime must + not take the time of the next record, which ExifTool lists first in it. + """ + xml = """\ + + + + 37.1 + 28.1 + 2019:09:02 10:23:29.00Z + 37.2 + 28.2 + 2019:09:02 10:23:30.00Z + 37.3 + 28.3 + 2019:09:02 10:23:31.00Z + 37.4 + 28.4 + + +""" + reader = ExifToolReadVideo(_etree_from_xml(xml)) + track = reader._extract_gps_track_from_quicktime() + assert [(p.time, p.lat, p.lon) for p in track] == [ + (0.0, 37.2, 28.2), + (1.0, 37.3, 28.3), + (2.0, 37.4, 28.4), + ] + + def test_records_without_sample_times(self): + """Insta360 lists its records with no SampleTime between them.""" + xml = """\ + + + + 2023:09:23 08:00:00Z + 47.371 + 8.542 + 410.0 + 2023:09:23 08:00:01Z + 47.372 + 8.543 + 45.0 + 411.0 + + +""" + reader = ExifToolReadVideo(_etree_from_xml(xml)) + track = reader._extract_gps_track_from_quicktime(namespace="Insta360") + assert [(p.time, p.angle, p.alt) for p in track] == [ + (0.0, None, 410.0), + (1.0, 45.0, 411.0), + ] + # --------------------------------------------------------------------------- # ExifToolReadVideo._extract_gps_track_from_track diff --git a/tests/unit/test_geotag_filetypes.py b/tests/unit/test_geotag_filetypes.py new file mode 100644 index 000000000..34629c0f9 --- /dev/null +++ b/tests/unit/test_geotag_filetypes.py @@ -0,0 +1,229 @@ +# 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. + +""" +--filetypes limits processing to the videos of the types given. The type of a +video is known only once it is read, so a video of another type is left out +once a source finds out, and no later source reads it. +""" + +from __future__ import annotations + +import typing as T +from pathlib import Path + +import pytest +from mapillary_tools import exceptions, types +from mapillary_tools.geotag import factory +from mapillary_tools.geotag.options import SourceOption, SourceType +from mapillary_tools.geotag.video_extractors.native import NativeVideoExtractor + +DATA = Path(__file__).parent.parent / "data" + +NOT_FOUND = exceptions.MapillaryVideoGPSNotFoundError( + "No GPS data found from the video" +) + +Outcome = T.Union[Exception, types.FileType] + + +def _run_chain( + monkeypatch: pytest.MonkeyPatch, + options: T.Sequence[SourceOption], + outcomes: dict[SourceType, Outcome], + filetypes: set[types.FileType] | None, +) -> tuple[list[types.MetadataOrError], list[SourceType]]: + """ + Run factory.process() on one video with --filetypes **filetypes**, which + every option is limited to, as process_geotag_properties() does. Each source + that geotags videos is replaced by a stub that reports outcomes[source]: an + error, or the file type it read the video as. Returns the results and the + sources that read the video. + """ + called: list[SourceType] = [] + + class StubGeotag: + def __init__(self, source: SourceType): + self.source = source + + def to_description(self, paths): + called.append(self.source) + outcome = outcomes[self.source] + if isinstance(outcome, types.FileType): + return [ + types.VideoMetadata(filename=path, filetype=outcome, points=[]) + for path in paths + ] + return [ + types.describe_error_metadata( + outcome, filename=path, filetype=types.FileType.VIDEO + ) + for path in paths + ] + + def build_video_geotag(option: SourceOption): + if option.source is SourceType.EXIF: + return None + return StubGeotag(option.source) + + monkeypatch.setattr(factory, "_build_video_geotag", build_video_geotag) + for option in options: + option.filetypes = types.combine_filetype_filters(option.filetypes, filetypes) + results = factory.process([Path("/tmp/video.mp4")], options, filetypes) + return results, called + + +def _options(*sources: SourceType) -> list[SourceOption]: + return [SourceOption(source) for source in sources] + + +DEFAULT_CHAIN = [SourceType.NATIVE, SourceType.EXIFTOOL_RUNTIME] +GOPRO = {types.FileType.GOPRO} + + +class TestProcess: + def test_video_of_another_type_is_left_out(self, monkeypatch): + """ExifTool processed it after the native reader, whatever its type""" + results, called = _run_chain( + monkeypatch, + _options(*DEFAULT_CHAIN), + { + SourceType.NATIVE: NOT_FOUND, + SourceType.EXIFTOOL_RUNTIME: types.FileType.VIDEO, + }, + GOPRO, + ) + assert results == [] + assert called == [SourceType.NATIVE] + + def test_video_read_as_another_type_is_left_out(self, monkeypatch): + results, _ = _run_chain( + monkeypatch, + _options(SourceType.GPX), + {SourceType.GPX: types.FileType.BLACKVUE}, + GOPRO, + ) + assert results == [] + + @pytest.mark.parametrize("filetype", [types.FileType.GOPRO, types.FileType.VIDEO]) + def test_video_of_a_type_asked_for_or_unknown_is_kept(self, monkeypatch, filetype): + """ + A GPX or ExifTool reads a video without telling its type, which may well + be one asked for + """ + results, _ = _run_chain( + monkeypatch, + _options(SourceType.EXIFTOOL_RUNTIME), + {SourceType.EXIFTOOL_RUNTIME: filetype}, + GOPRO, + ) + [result] = results + assert isinstance(result, types.VideoMetadata) + assert result.filetype is filetype + + def test_an_earlier_error_is_not_reported(self, monkeypatch): + results, _ = _run_chain( + monkeypatch, + _options(SourceType.GPX, SourceType.NATIVE), + {SourceType.GPX: NOT_FOUND, SourceType.NATIVE: NOT_FOUND}, + GOPRO, + ) + assert results == [] + + def test_gps_of_a_type_asked_for_that_is_unusable_is_reported(self, monkeypatch): + noise = exceptions.MapillaryGPSNoiseError("GPS is too noisy") + results, _ = _run_chain( + monkeypatch, + _options(*DEFAULT_CHAIN), + { + SourceType.NATIVE: noise, + SourceType.EXIFTOOL_RUNTIME: types.FileType.VIDEO, + }, + GOPRO, + ) + [result] = results + assert isinstance(result, types.ErrorMetadata) + assert result.error is noise + + @pytest.mark.parametrize("filetypes", [None, {types.FileType.VIDEO}]) + def test_any_video_type(self, monkeypatch, filetypes): + results, called = _run_chain( + monkeypatch, + _options(*DEFAULT_CHAIN), + { + SourceType.NATIVE: NOT_FOUND, + SourceType.EXIFTOOL_RUNTIME: types.FileType.VIDEO, + }, + filetypes, + ) + [result] = results + assert isinstance(result, types.VideoMetadata) + assert called == DEFAULT_CHAIN + + @pytest.mark.parametrize( + "filetypes", [None, {types.FileType.GOPRO, types.FileType.BLACKVUE}] + ) + def test_the_filetypes_of_a_source_leave_the_video_to_the_next( + self, monkeypatch, filetypes + ): + """ + A source limited to some types by its own filetypes found only that the + video is not of those + """ + results, called = _run_chain( + monkeypatch, + [ + SourceOption(SourceType.NATIVE, filetypes=GOPRO), + SourceOption(SourceType.EXIFTOOL_RUNTIME), + ], + { + SourceType.NATIVE: NOT_FOUND, + SourceType.EXIFTOOL_RUNTIME: types.FileType.VIDEO, + }, + filetypes, + ) + [result] = results + assert isinstance(result, types.VideoMetadata) + assert called == DEFAULT_CHAIN + + def test_video_that_no_source_can_read_is_left_out(self, monkeypatch): + """ + Its type is not known without reading it, so it may not be of any of the + types asked for + """ + results, _ = _run_chain(monkeypatch, _options(SourceType.EXIF), {}, GOPRO) + assert results == [] + + results, _ = _run_chain(monkeypatch, _options(SourceType.EXIF), {}, None) + [result] = results + assert isinstance(result, types.ErrorMetadata) + assert "cannot geotag videos" in str(result.error) + + +class TestNativeVideoExtractor: + """ + process() tells by the native reader's not found that a video is of none of + the types the reader was given + """ + + def test_video_of_none_of_the_types(self): + with pytest.raises(exceptions.MapillaryVideoGPSNotFoundError): + NativeVideoExtractor( + DATA / "videos" / "sample-5s.mp4", filetypes=GOPRO + ).extract() + + def test_video_of_another_type(self): + with pytest.raises(exceptions.MapillaryVideoGPSNotFoundError): + NativeVideoExtractor( + DATA / "gopro_data" / "max-360mode.mp4", + filetypes={types.FileType.BLACKVUE, types.FileType.CAMM}, + ).extract() + + def test_video_of_the_type_without_usable_gps(self): + """Recognized as a GoPro video, whose GPS is all noise""" + with pytest.raises(exceptions.MapillaryGPSNoiseError): + NativeVideoExtractor( + DATA / "gopro_data" / "hero8.mp4", filetypes=GOPRO + ).extract() diff --git a/tests/unit/test_gpmf_gps_filter.py b/tests/unit/test_gpmf_gps_filter.py index c15ef40e5..9182b06ea 100644 --- a/tests/unit/test_gpmf_gps_filter.py +++ b/tests/unit/test_gpmf_gps_filter.py @@ -99,6 +99,20 @@ def test_far_points_split(self): assert decider(p1, p2) is True +class TestDistanceLe: + def test_close_points(self): + decider = gps_filter.distance_le(100000) + p1 = _make_point(0.0, 48.0, 11.0) + p2 = _make_point(1.0, 48.001, 11.001) + assert decider(p1, p2) is True + + def test_far_points(self): + decider = gps_filter.distance_le(100) + p1 = _make_point(0.0, 0.0, 0.0) + p2 = _make_point(1.0, 1.0, 1.0) + assert decider(p1, p2) is False + + class TestSpeedLe: def test_slow_speed_true(self): decider = gps_filter.speed_le(1000) @@ -226,3 +240,28 @@ def test_outlier_removed(self): assert len(result) < len(cluster) + 1 # The cluster points should survive assert len(result) >= len(cluster) + + def test_sample_off_the_track_keeps_both_sides(self): + """ + A single sample far off a slow track splits it in two. The pieces on + either side of it are as close as consecutive points, but the jitter + between them, over the fraction of a second the sample lasts, reads + faster than the ground speeds: both sides must still be kept. + """ + # A walk at 0.5 m/s, sampled every 0.2s, which GPS jitter moves by 1m + # right after the off-track sample + walk = [ + _make_gps_point( + i * 0.2, + 48.0 + i * 0.0000009 + (0.00001 if 16 <= i else 0.0), + 11.0, + ground_speed=0.5, + ) + for i in range(41) + ] + off_track = _make_gps_point(15 * 0.2, 0.000001, 2.363944, ground_speed=0.5) + points = walk[:15] + [off_track] + walk[16:] + + result = remove_outliers(points) + + assert list(result) == walk[:15] + walk[16:] 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_coverage.py b/tests/unit/test_gpx_video_coverage.py index 2c3687472..b6e1f9b65 100644 --- a/tests/unit/test_gpx_video_coverage.py +++ b/tests/unit/test_gpx_video_coverage.py @@ -24,7 +24,7 @@ from pathlib import Path import pytest -from mapillary_tools import exceptions, geo, telemetry, types, uploader +from mapillary_tools import constants, exceptions, geo, telemetry, types, uploader from mapillary_tools.camm import camm_builder from mapillary_tools.geotag import factory from mapillary_tools.geotag.options import ( @@ -553,3 +553,179 @@ def test_gpx_as_the_only_source_fails_the_image(self, tmp_path): assert isinstance(metadata, types.ErrorMetadata) assert isinstance(metadata.error, exceptions.MapillaryOutsideGPXTrackError) + + +def _write_gpx_parts(path: Path, parts: T.Sequence[tuple[float, float]]) -> Path: + """A GPX track of several parts, each (start, seconds) with a point a second""" + trkpts = "".join( + f'' + for start, seconds in parts + for i in range(int(seconds) + 1) + ) + path.write_text( + '' + '' + f"{trkpts}" + ) + return path + + +class TestGapsInTheTrack: + """ + A GPX track can hold several trips, with the GPS logger switched off in + between. Nothing can be interpolated across a gap of more than an hour + between two of its points, which is outside the track like the time before + or after it. + """ + + def _extract( + self, + tmp_path: Path, + parts: T.Sequence[tuple[float, float]], + video_track: T.Sequence[geo.Point] | None = None, + duration: float = VIDEO_DURATION, + **kwargs, + ) -> types.VideoMetadata: + video_path = _write_camm_mp4( + tmp_path / "video.mp4", + _video_track() if video_track is None else video_track, + duration, + ) + gpx_path = _write_gpx_parts(tmp_path / "track.gpx", parts) + return GPXVideoExtractor(video_path, gpx_path, **kwargs).extract() + + def test_a_video_in_a_gap_misses_the_track(self, tmp_path): + """Trips ended 3 hours before the video and started 2 hours after it""" + with pytest.raises(exceptions.MapillaryOutsideGPXTrackError) as info: + self._extract( + tmp_path, + [ + (A_UNIX_TIME - 3 * 3600 - 60, 60), + (A_UNIX_TIME + VIDEO_DURATION + 2 * 3600, 60), + ], + ) + message = str(info.value) + # The gap, and the last point before it and the first after it + assert ( + "has a gap of 5.0 hours, from 2026-08-12T05:26:27.979846+00:00 at (37.0, 14.0) " + "to 2026-08-12T10:26:39.979846+00:00 at (37.0, 14.0), which the video" + ) in message + assert "falls in. Check that the GPX file belongs to this video" in message + assert info.value.image_time == "2026_08_12_08_26_27_979" + + def test_a_gap_of_less_than_an_hour_is_interpolated(self, tmp_path, caplog): + with caplog.at_level(logging.WARNING): + self._extract( + tmp_path, [(A_UNIX_TIME - 1800, 60), (A_UNIX_TIME + 1500, 60)] + ) + assert not _warnings(caplog) + + def test_a_gap_in_part_of_the_video_warns(self, tmp_path, caplog): + """The first trip ends 5 seconds into the video""" + with caplog.at_level(logging.WARNING): + self._extract(tmp_path, [(A_UNIX_TIME - 60, 65), (A_UNIX_TIME + 7200, 60)]) + [warning] = _warnings(caplog) + assert "7 seconds of the video fall outside the track" in warning + + def test_the_gap_of_a_timelapse_is_in_gps_time(self, tmp_path): + """ + 2 hours of GPS time are 8 minutes of a timelapse at 15x, a gap in GPS + time, where trips are recorded + """ + rate = 15.0 + with pytest.raises(exceptions.MapillaryOutsideGPXTrackError) as info: + self._extract( + tmp_path, + [(A_UNIX_TIME - 3600 - 60, 60), (A_UNIX_TIME + 11 * rate + 3600, 60)], + video_track=_video_track(rate), + duration=11.0, + ) + assert "has a gap of 2.0 hours" in str(info.value) + + def test_a_start_time_in_a_gap_misses_the_track(self, tmp_path): + gpx_path = _write_gpx_parts( + tmp_path / "track.gpx", + [(A_UNIX_TIME - 7200, 60), (A_UNIX_TIME + 7200, 60)], + ) + with pytest.raises(exceptions.MapillaryOutsideGPXTrackError) as info: + GPXVideoExtractor( + NO_GPS_VIDEO, + gpx_path, + video_sync=VideoSyncOption(start_time=A_UNIX_TIME), + ).extract() + assert "has a gap of 4.0 hours" in str(info.value) + + def test_the_limit_can_be_raised(self, tmp_path, caplog, monkeypatch): + """By MAPILLARY_TOOLS_MAX_GPX_GAP_SECONDS""" + monkeypatch.setattr(constants, "MAX_GPX_GAP_SECONDS", 6 * 3600) + with caplog.at_level(logging.WARNING): + self._extract( + tmp_path, + [ + (A_UNIX_TIME - 3 * 3600 - 60, 60), + (A_UNIX_TIME + VIDEO_DURATION + 2 * 3600, 60), + ], + ) + assert not _warnings(caplog) + + def _process_image( + self, tmp_path: Path, parts: T.Sequence[tuple[float, float]] + ) -> types.MetadataOrError: + image_path = tmp_path / "image.jpg" + shutil.copyfile(IMAGE, image_path) + gpx_path = _write_gpx_parts(tmp_path / "walk.gpx", parts) + [metadata] = factory.process( + [image_path], + [ + SourceOption( + SourceType.GPX, + num_processes=0, + source_path=SourcePathOption(source_path=gpx_path), + ) + ], + ) + return metadata + + # The capture time of IMAGE + IMAGE_TIME = datetime.datetime( + 2018, 6, 8, 20, 24, 11, tzinfo=datetime.timezone.utc + ).timestamp() + + def test_an_image_in_a_gap_is_outside_the_track(self, tmp_path): + metadata = self._process_image( + tmp_path, [(self.IMAGE_TIME - 3600, 60), (self.IMAGE_TIME + 600, 60)] + ) + assert isinstance(metadata, types.ErrorMetadata) + assert isinstance(metadata.error, exceptions.MapillaryOutsideGPXTrackError) + # The gap, and the last point before it and the first after it + assert str(metadata.error) == ( + "The GPX track has a gap of 1.1 hours, from 2018-06-08T19:25:11+00:00 at (37.0, 14.0) " + "to 2018-06-08T20:34:11+00:00 at (37.0, 14.0), which the image time " + "2018-06-08T20:24:11+00:00 falls in" + ) + assert vars(metadata.error) == { + "image_time": "2018_06_08_20_24_11_000", + "gpx_start_time": "2018_06_08_19_24_11_000", + "gpx_end_time": "2018_06_08_20_35_11_000", + } + + def test_the_image_limit_can_be_raised(self, tmp_path, monkeypatch): + """By MAPILLARY_TOOLS_MAX_GPX_GAP_SECONDS""" + monkeypatch.setattr(constants, "MAX_GPX_GAP_SECONDS", 2 * 3600) + metadata = self._process_image( + tmp_path, [(self.IMAGE_TIME - 3600, 60), (self.IMAGE_TIME + 600, 60)] + ) + assert isinstance(metadata, types.ImageMetadata) + + def test_an_image_in_a_gap_of_less_than_an_hour_is_interpolated(self, tmp_path): + metadata = self._process_image( + tmp_path, [(self.IMAGE_TIME - 1800, 60), (self.IMAGE_TIME + 600, 60)] + ) + assert isinstance(metadata, types.ImageMetadata) + + def test_an_image_on_a_point_by_a_gap_is_on_the_track(self, tmp_path): + """The last point before the gap is the image's own position""" + metadata = self._process_image( + tmp_path, [(self.IMAGE_TIME - 60, 60), (self.IMAGE_TIME + 7200, 60)] + ) + assert isinstance(metadata, types.ImageMetadata) 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_image_exif_extractor.py b/tests/unit/test_image_exif_extractor.py new file mode 100644 index 000000000..6c5c2f8ab --- /dev/null +++ b/tests/unit/test_image_exif_extractor.py @@ -0,0 +1,99 @@ +# 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. + +import datetime +import typing as T +import xml.etree.ElementTree as ET +from pathlib import Path + +import pytest +from mapillary_tools import exceptions +from mapillary_tools.exiftool_read import EXIFTOOL_NAMESPACES +from mapillary_tools.geotag.image_extractors.exif import ( + _capture_unix_time, + ImageEXIFExtractor, +) +from mapillary_tools.geotag.image_extractors.exiftool import ImageExifToolExtractor +from PIL import Image +from PIL.ExifTags import Base, GPS, IFD +from PIL.TiffImagePlugin import IFDRational + + +# What FLIR thermal cameras write for no date +NO_DATE = "0001:01:01 00:00:00" + + +def _write_image(path: Path, date_time_original: str) -> Path: + img = Image.new("RGB", (4, 4), (0, 0, 0)) + exif = img.getexif() + exif.get_ifd(IFD.Exif)[Base.DateTimeOriginal] = date_time_original + gps = exif.get_ifd(IFD.GPSInfo) + gps[GPS.GPSLatitudeRef] = "N" + gps[GPS.GPSLatitude] = (IFDRational(24), IFDRational(33), IFDRational(3636, 100)) + gps[GPS.GPSLongitudeRef] = "W" + gps[GPS.GPSLongitude] = (IFDRational(81), IFDRational(48), IFDRational(2772, 100)) + img.save(path, format="JPEG", exif=exif) + return path + + +def _exiftool_element(tags: T.Dict[str, str]) -> ET.Element: + root = ET.Element("rdf:Description") + for tag, value in tags.items(): + prefix, tag_name = tag.split(":", 1) + child = ET.SubElement(root, "{" + EXIFTOOL_NAMESPACES[prefix] + "}" + tag_name) + child.text = value + return root + + +class TestCaptureTimeWithoutUnixTime: + """ + A capture time with no Unix time is no capture time. It was read as 1970. + """ + + def test_exif(self, tmp_path: Path): + image = _write_image(tmp_path / "image.jpg", NO_DATE) + + with pytest.raises( + exceptions.MapillaryGeoTaggingError, match="Unable to extract timestamp" + ): + ImageEXIFExtractor(image).extract() + + def test_exiftool(self, tmp_path: Path): + image = tmp_path / "image.jpg" + element = _exiftool_element( + { + "GPS:GPSLatitude": "24.5601", + "GPS:GPSLatitudeRef": "N", + "GPS:GPSLongitude": "81.8077", + "GPS:GPSLongitudeRef": "W", + "ExifIFD:DateTimeOriginal": NO_DATE, + } + ) + + with pytest.raises( + exceptions.MapillaryGeoTaggingError, match="Unable to extract timestamp" + ): + ImageExifToolExtractor(image, element).extract() + + @pytest.mark.parametrize("error", [ValueError, OverflowError, OSError]) + def test_any_conversion_error(self, error: T.Type[Exception]): + # Naive times before 1970 raise OSError on Windows + class Unconvertible(datetime.datetime): + def timestamp(self) -> float: + raise error("no Unix time") + + with pytest.raises( + exceptions.MapillaryGeoTaggingError, match="Unable to extract timestamp" + ): + _capture_unix_time(Unconvertible(1969, 12, 31)) + + def test_a_valid_capture_time_is_read(self, tmp_path: Path): + image = _write_image(tmp_path / "image.jpg", "2020:01:02 03:04:05") + + metadata = ImageEXIFExtractor(image).extract() + + assert metadata.time == datetime.datetime(2020, 1, 2, 3, 4, 5).timestamp() + assert metadata.lat == pytest.approx(24.5601) + assert metadata.lon == pytest.approx(-81.8077) 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")