diff --git a/mapillary_tools/geotag/factory.py b/mapillary_tools/geotag/factory.py index 1071dfbb..4c9f7cfe 100644 --- a/mapillary_tools/geotag/factory.py +++ b/mapillary_tools/geotag/factory.py @@ -257,6 +257,8 @@ def process( options: T.Collection[SourceOption], # --filetypes, which the filetypes of every option are limited to already filetypes: set[types.FileType] | None = None, + # If given, filled with the option that geotagged each file, of those geotagged + sources: dict[Path, SourceOption] | None = None, ) -> list[types.MetadataOrError]: if not options: raise ValueError("No geotag options provided") @@ -378,10 +380,11 @@ def process( reprocessable_paths.remove(metadata.filename) pending_errors.pop(metadata.filename, None) rejection = gpx_rejections.pop(metadata.filename, None) - if rejection is not None and not isinstance( - metadata, types.ErrorMetadata - ): - rescued.append((metadata, *rejection)) + if not isinstance(metadata, types.ErrorMetadata): + if sources is not None: + sources[metadata.filename] = option + if rejection is not None: + rescued.append((metadata, *rejection)) _warn_geotagged_despite_gpx(rescued, option) diff --git a/mapillary_tools/geotag/geotag_images_from_gpx_file.py b/mapillary_tools/geotag/geotag_images_from_gpx_file.py index 77f7ae42..5fb022f3 100644 --- a/mapillary_tools/geotag/geotag_images_from_gpx_file.py +++ b/mapillary_tools/geotag/geotag_images_from_gpx_file.py @@ -48,12 +48,15 @@ def _read_gpx_points(source_path: Path) -> list[geo.Point]: geotagged from it, and if it is missing or has no timestamped point: the file was given to geotag the images from, so it fails them rather than leaving them to the next geotag source. + + The file is parsed once in a process (see utils.read_gpx_tracks()), and + its points are shared with its other readers, which must not change them. """ if not source_path.is_file(): raise exceptions.MapillaryInvalidGPXError(f"GPX file not found: {source_path}") try: - tracks = utils.parse_gpx(source_path) + tracks = utils.read_gpx_tracks(source_path) except (OSError, UnicodeError, gpxpy.gpx.GPXException) as ex: raise exceptions.MapillaryInvalidGPXError( f"{source_path.name} is not a valid GPX file. " @@ -61,12 +64,14 @@ def _read_gpx_points(source_path: Path) -> list[geo.Point]: ) from ex if 1 < len(tracks): - LOG.warning( + utils.warn_once( + source_path, + ("tracks", len(tracks)), "Found %s tracks in the GPX file %s. Will merge points in all the tracks as a single track for interpolation", len(tracks), source_path, ) - points: list[geo.Point] = sum(tracks, []) + points: list[geo.Point] = [point for track in tracks for point in track] if not points: raise exceptions.MapillaryInvalidGPXError( diff --git a/mapillary_tools/geotag/geotag_images_from_nmea_file.py b/mapillary_tools/geotag/geotag_images_from_nmea_file.py index 7d735ad0..0d88b5b5 100644 --- a/mapillary_tools/geotag/geotag_images_from_nmea_file.py +++ b/mapillary_tools/geotag/geotag_images_from_nmea_file.py @@ -37,9 +37,15 @@ # What a byte that is not text decodes to _NOT_TEXT = "\ufffd" -# The checksum that ends a sentence: *HH +# The checksum that ends a sentence: *HH. It is optional _CHECKSUM_RE = re.compile(r"\*[0-9A-Fa-f]{2}\s*$") +# The fields of a complete sentence: an RMC sentence has 11 (NMEA 2.0 to +# 2.2), 12 with the mode of NMEA 2.3, and 13 with the navigational status of +# NMEA 4.1; a GGA sentence has 14. One without its checksum that has fewer was +# cut short, such as the last one of a log when the recording stopped +_MIN_FIELDS = {"RMC": 11, "GGA": 14} + # The time of a sentence (hhmmss, and a fraction of a second), and the date of # an RMC sentence (ddmmyy). pynmea2 reads a field cut short as another time or # date: an RMC sentence cut inside its date "2212" as 2012-02-02 @@ -85,8 +91,7 @@ def _read_nmea_points(source_path: Path) -> list[geo.Point]: # UnboundLocalError: GGA sentences but no RMC sentence to give their date except (OSError, ValueError, TypeError, UnboundLocalError) as ex: raise exceptions.MapillaryInvalidGPXError( - f"{source_path.name} is not a valid NMEA file. " - "Replace it with a valid NMEA file." + describe_invalid_nmea(source_path, ex) ) from ex if not points: @@ -97,6 +102,29 @@ def _read_nmea_points(source_path: Path) -> list[geo.Point]: return points +def describe_invalid_nmea(nmea_file: Path, ex: Exception) -> str: + """ + The message of the error for the NMEA file **nmea_file** that reading it + failed with **ex**, which says why + + >>> describe_invalid_nmea(Path("a/track.nmea"), ValueError("it is not text")) + 'track.nmea is not a valid NMEA file: it is not text. Replace it with a valid NMEA file.' + >>> describe_invalid_nmea(Path("track.nmea"), PermissionError(13, "Permission denied")) + 'track.nmea is not a valid NMEA file: Permission denied. Replace it with a valid NMEA file.' + """ + if isinstance(ex, OSError): + reason = ex.strerror or str(ex) + elif isinstance(ex, ValueError): + reason = str(ex) + else: + reason = "" + because = f": {reason}" if reason else "" + return ( + f"{nmea_file.name} is not a valid NMEA file{because}. " + "Replace it with a valid NMEA file." + ) + + def get_lat_lon_time_from_nmea(nmea_file: Path) -> list[geo.Point]: """ Read the fixes of an NMEA file: the time and position of each GGA @@ -117,9 +145,10 @@ def get_lat_lon_time_from_nmea(nmea_file: Path) -> list[geo.Point]: time. A malformed sentence, such as one whose checksum does not match, one cut - short, or one with a byte that is not text, is skipped with a warning. A - file of which no fix can be read raises ValueError if it has malformed - sentences or bytes that are not text. + short, or one with a byte that is not text, is skipped with a warning. + Checksums are optional: a sentence without one is cut short only if its + own fields show it. A file of which no fix can be read raises ValueError, + which says why, if it has malformed sentences or bytes that are not text. A file is read once in a process, however many times it is asked for. """ @@ -158,29 +187,28 @@ def _read_nmea_data(nmea_file: Path, data: bytes) -> list[geo.Point]: for number, line in enumerate(lines, start=1) if _SENTENCE_RE.search(line) ] - # Receivers end each sentence with its checksum, or none of them. One - # without, where the others have theirs, was cut short, such as the last - # one of a log when the recording stopped - checksummed = any(_CHECKSUM_RE.search(line) for _, line in sentences) messages = [] # (line number, line, error) of each malformed sentence malformed: list[tuple[int, str, ValueError]] = [] for number, line in sentences: try: - messages.append(_parse_sentence(line, checksummed=checksummed)) + messages.append(_parse_sentence(line)) except ValueError as ex: malformed.append((number, line, ex)) points = _read_fixes(messages, nmea_file) if not points and not_text and not malformed: - raise ValueError(f"{nmea_file} is not text") + raise ValueError("it is not text") if malformed: number, line, error = malformed[0] if not points: - raise error + raise ValueError( + f"{len(malformed)} of its {len(sentences)} RMC and GGA sentences are malformed, " + f"the first on line {number} ({_quote(str(error))}): {_quote(line)}" + ) from error LOG.warning( "Skipping %d of the %d RMC and GGA sentences of the NMEA file %s, which are malformed, the first on line %d (%s): %s", len(malformed), @@ -259,14 +287,14 @@ def _read_fixes( return points -def _parse_sentence(line: str, checksummed: bool = False) -> pynmea2.NMEASentence: +def _parse_sentence(line: str) -> pynmea2.NMEASentence: """ Parse an NMEA sentence, raising ValueError if it is malformed: if it does - not parse or its checksum does not match, if it has no checksum but should - (**checksummed**), if its time or the date of an RMC sentence is not - hhmmss[.ss] or ddmmyy, or if a field that a fix is read from does not - parse, which pynmea2 leaves a string, or if it has a byte that is not - text, decoded as U+FFFD + not parse or its checksum does not match, if it has no checksum and fewer + fields than a complete sentence (see _MIN_FIELDS), if its time or the date + of an RMC sentence is not hhmmss[.ss] or ddmmyy, or if a field that a fix + is read from does not parse, which pynmea2 leaves a string, or if it has a + byte that is not text, decoded as U+FFFD >>> _parse_sentence("$GPGGA,0246,3735.30670,N,12220.25810,W,1,08,1.0,0.0,M,0.0,M,,*64") Traceback (most recent call last): @@ -276,21 +304,24 @@ def _parse_sentence(line: str, checksummed: bool = False) -> pynmea2.NMEASentenc Traceback (most recent call last): ... ValueError: checksum does not match: 00 != 4F - >>> _parse_sentence("$GPRMC,024636.00,A,3735.30670,N,12220.25810,W,0.0,0.0,2212", checksummed=True) - Traceback (most recent call last): - ... - ValueError: no checksum >>> _parse_sentence("$GPRMC,024636.00,A,3735.30670,N,12220.25810,W,0.0,0.0,2212") Traceback (most recent call last): ... ValueError: malformed datestamp: '2212' + >>> _parse_sentence("$GPRMC,024636.00,A,3735.30670,N,12220.25810,W,0.0,0.0,221222") + Traceback (most recent call last): + ... + ValueError: cut short: 9 of the 11 fields of an RMC sentence + >>> _parse_sentence("$GPGGA,024636.00,3735.30670,N,12220.25810,W,1,08,1.0,12") + Traceback (most recent call last): + ... + ValueError: cut short: 9 of the 14 fields of a GGA sentence + >>> type(_parse_sentence("$GPGGA,024636.00,3735.30670,N,12220.25810,W,1,08,1.0,12.5,M,0.0,M,,")).__name__ + 'GGA' """ if _NOT_TEXT in line: raise ValueError("a byte that is not text") - if checksummed and not _CHECKSUM_RE.search(line): - raise ValueError("no checksum") - try: message = pynmea2.parse(line) except pynmea2.ParseError as ex: @@ -306,6 +337,15 @@ def _parse_sentence(line: str, checksummed: bool = False) -> pynmea2.NMEASentenc if value and not pattern.fullmatch(value): raise ValueError(f"malformed {name}: {value!r}") + # A sentence with its checksum ended where it ends: pynmea2 checks it + if not _CHECKSUM_RE.search(line): + kind = message.sentence_type + if kind in _MIN_FIELDS and len(message.data) < _MIN_FIELDS[kind]: + article = "an" if kind == "RMC" else "a" + raise ValueError( + f"cut short: {len(message.data)} of the {_MIN_FIELDS[kind]} fields of {article} {kind} sentence" + ) + fields: dict[str, type] = {} if isinstance(message, pynmea2.RMC): fields = {"timestamp": datetime.time, "datestamp": datetime.date} diff --git a/mapillary_tools/geotag/geotag_videos_from_gpx.py b/mapillary_tools/geotag/geotag_videos_from_gpx.py index 2f34418d..859a95c9 100644 --- a/mapillary_tools/geotag/geotag_videos_from_gpx.py +++ b/mapillary_tools/geotag/geotag_videos_from_gpx.py @@ -5,6 +5,7 @@ from __future__ import annotations +import collections import logging import sys import typing as T @@ -16,7 +17,7 @@ from typing_extensions import override from .. import exceptions, types -from . import options +from . import options, utils from .base import GeotagVideosFromGeneric from .video_extractors.gpx import GPXVideoExtractor @@ -72,4 +73,28 @@ def _generate_video_extractors( filetype=types.FileType.VIDEO, ) ) + + # The videos are geotagged in a pool of processes, each of which would + # read the track file of its videos. Read here, once, and warn about + # once, a file that geotags several of them, or that this process read + # already (see _read_here()) + extractors = [ + result for result in results if isinstance(result, GPXVideoExtractor) + ] + num_videos = collections.Counter( + extractor.gpx_path.resolve() for extractor in extractors + ) + for extractor in extractors: + if self._read_here(extractor, num_videos[extractor.gpx_path.resolve()]): + extractor.read_tracks() + return results + + def _read_here(self, extractor: GPXVideoExtractor, num_videos: int) -> bool: + """ + Whether to read the track file of **extractor**, which geotags + **num_videos** videos, in this process rather than in the pool: if it + geotags several, or was parsed in this process already. A file of + each video is left to the pool, which parses them side by side + """ + return 1 < num_videos or utils.was_gpx_read(extractor.gpx_path) diff --git a/mapillary_tools/geotag/geotag_videos_from_nmea.py b/mapillary_tools/geotag/geotag_videos_from_nmea.py index 74d1efe4..6ff7fd8e 100644 --- a/mapillary_tools/geotag/geotag_videos_from_nmea.py +++ b/mapillary_tools/geotag/geotag_videos_from_nmea.py @@ -6,15 +6,12 @@ from __future__ import annotations import sys -import typing as T -from pathlib import Path if sys.version_info >= (3, 12): from typing import override else: from typing_extensions import override -from .. import types from .geotag_videos_from_gpx import GeotagVideosFromGPX from .video_extractors.gpx import GPXVideoExtractor from .video_extractors.nmea import NMEAVideoExtractor @@ -30,14 +27,7 @@ class GeotagVideosFromNMEA(GeotagVideosFromGPX): default_pattern = "%g.nmea" @override - def _generate_video_extractors( - self, video_paths: T.Sequence[Path] - ) -> T.Sequence[GPXVideoExtractor | types.ErrorMetadata]: - extractors = super()._generate_video_extractors(video_paths) - # Read each NMEA file here, once however many videos it geotags, and - # warn about it once: the videos are geotagged in a pool of processes, - # each of which would read it again - for extractor in extractors: - if isinstance(extractor, NMEAVideoExtractor): - extractor.read_tracks() - return extractors + def _read_here(self, extractor: GPXVideoExtractor, num_videos: int) -> bool: + # Every NMEA file, which is read once in a process however many videos + # it geotags, and warned about once + return True diff --git a/mapillary_tools/geotag/utils.py b/mapillary_tools/geotag/utils.py index e1b02e51..fa0a0b4a 100644 --- a/mapillary_tools/geotag/utils.py +++ b/mapillary_tools/geotag/utils.py @@ -5,7 +5,13 @@ from __future__ import annotations +import collections +import dataclasses +import hashlib +import io import logging +import multiprocessing +import os import typing as T import xml.etree.ElementTree as ET from pathlib import Path @@ -17,10 +23,95 @@ Track = T.List[geo.Point] LOG = logging.getLogger(__name__) +# The tracks parsed from the last GPX files read in this process, or the error +# parsing them raised, by path and content: a GPX file that geotags several +# videos, or the samples of a video and then their images, is parsed once. +# Only a few are kept, so that one GPX file per video does not pile up +_PARSED_GPX: collections.OrderedDict[ + tuple[str, str], list[Track] | gpxpy.gpx.GPXException | UnicodeError +] = collections.OrderedDict() +_MAX_PARSED_GPX = 4 + +# The warnings given in this process about the track files read, so that each +# is given once however many times a file is read +_WARNED: set[tuple[T.Hashable, ...]] = set() + def parse_gpx(gpx_file: Path) -> list[Track]: - with gpx_file.open("r") as f: - gpx = gpxpy.parse(f) + """ + Parse the tracks of a GPX file, each a list of points with Unix times + """ + # Copies, which the caller may change + return [ + [dataclasses.replace(point) for point in track] + for track in read_gpx_tracks(gpx_file) + ] + + +def read_gpx_tracks(gpx_file: Path) -> list[Track]: + """ + The tracks of a GPX file as parse_gpx() parses them, parsed once in a + process for as long as the file does not change. They are shared with + every other reader of the file, and must not be changed. + + A worker of a pool keeps none: it parses the GPX file of one video only, + as a file that geotags several videos is read before the pool (see + GeotagVideosFromGPX), and would free them only when it exits. + """ + data = gpx_file.read_bytes() + key = (str(gpx_file.resolve()), hashlib.sha256(data).hexdigest()) + parsed = _PARSED_GPX.get(key) + if parsed is None: + try: + # Decoded as a file opened in text mode would be, and parsed + # without the bytes, as a file is + text = io.TextIOWrapper(io.BytesIO(data)).read() + del data + parsed = _parse_gpx_text(text) + except (gpxpy.gpx.GPXException, UnicodeError) as ex: + parsed = ex + if multiprocessing.parent_process() is None: + _PARSED_GPX[key] = parsed + while _MAX_PARSED_GPX < len(_PARSED_GPX): + _PARSED_GPX.popitem(last=False) + else: + _PARSED_GPX.move_to_end(key) + + if isinstance(parsed, Exception): + raise parsed.with_traceback(None) + return parsed + + +def was_gpx_read(gpx_file: Path) -> bool: + """ + Whether the tracks of a GPX file were parsed in this process and are kept + """ + path = str(gpx_file.resolve()) + return any(key[0] == path for key in _PARSED_GPX) + + +def warn_once( + path: Path, what: tuple[T.Hashable, ...], message: str, *args: object +) -> None: + """ + Log a warning about the track file **path**, unless this process warned + already of **what** about the file as it is now, whoever read it + """ + try: + stat = os.stat(path) + except OSError: + file_key: tuple[T.Hashable, ...] = (str(path),) + else: + file_key = (str(path.resolve()), stat.st_size, stat.st_mtime_ns) + key = (*file_key, *what) + if key in _WARNED: + return + _WARNED.add(key) + LOG.warning(message, *args) + + +def _parse_gpx_text(text: str) -> list[Track]: + gpx = gpxpy.parse(text) tracks: list[Track] = [] diff --git a/mapillary_tools/geotag/video_extractors/gpx.py b/mapillary_tools/geotag/video_extractors/gpx.py index 1f145ba9..0af22d83 100644 --- a/mapillary_tools/geotag/video_extractors/gpx.py +++ b/mapillary_tools/geotag/video_extractors/gpx.py @@ -29,8 +29,7 @@ from ...mp4 import construct_mp4_parser as cparser, simple_mp4_parser as sparser from ...mp4.mp4_sample_parser import MovieBoxParser from ...serializer.description import build_capture_time -from .. import options -from ..utils import parse_gpx +from .. import options, utils as geotag_utils from .base import BaseVideoExtractor from .native import NativeVideoExtractor @@ -175,6 +174,10 @@ class GPXVideoExtractor(BaseVideoExtractor): # The format of the track file, as messages name it file_format = "GPX" + # The tracks of the track file, or the error reading it, once read_tracks() + # has read them + _tracks: list[list[geo.Point]] | exceptions.MapillaryInvalidGPXError | None = None + def __init__( self, video_path: Path, @@ -285,39 +288,93 @@ def extract(self) -> types.VideoMetadata: return dataclasses.replace(native_video_metadata, points=gpx_points) + def read_tracks(self) -> None: + """ + Read the tracks of the track file now, for extract() to place, rather + than when it runs, which can be in another process: a file that + geotags several videos is then read, and warned about, once + """ + try: + self._tracks = self._read_tracks() + except exceptions.MapillaryInvalidGPXError as ex: + self._tracks = ex + except Exception: + # Left for extract() to raise again, and to be reported for the + # video as any unexpected error is + pass + else: + self._warn_of_tracks(self._tracks) + def _read_gpx_points(self) -> list[geo.Point]: """ Read the points of the GPX file, merged into one track in time order """ + warned = self._tracks is not None gpx_tracks = self._parse_tracks() + if not warned: + self._warn_of_tracks(gpx_tracks) + + # Copies, which are moved to the video: the tracks may be shared with + # other readers of the file + gpx_points = [ + dataclasses.replace(p) + for track in gpx_tracks + for p in track + if _GPS_EPOCH <= p.time + ] + # The tracks need not be in time order, and interpolation needs the + # merged points to be + gpx_points.sort(key=lambda p: p.time) + return gpx_points + + def _warn_of_tracks(self, gpx_tracks: list[list[geo.Point]]) -> None: + """ + Warn of what reading the points of the tracks leaves out or merges, + once for each file + """ if 1 < len(gpx_tracks): - LOG.warning( - 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" + geotag_utils.warn_once( + self.gpx_path, + ("tracks", len(gpx_tracks)), + "Found %d tracks in the %s file %s. Will merge points in all the tracks as a single track for interpolation", + len(gpx_tracks), + self.file_format, + self.gpx_path, ) - 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( + num_points = sum(len(track) for track in gpx_tracks) + num_kept = sum(1 for track in gpx_tracks for p in track if _GPS_EPOCH <= p.time) + if num_kept < num_points: + geotag_utils.warn_once( + self.gpx_path, + ("before 1980", num_points - num_kept, num_points), "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), + num_points - num_kept, + num_points, self.file_format, self.gpx_path, ) - # The tracks need not be in time order, and interpolation needs the - # merged points to be - gpx_points.sort(key=lambda p: p.time) - return gpx_points - def _parse_tracks(self) -> list[list[geo.Point]]: """ - Parse the tracks of the GPX file, each a list of points with Unix times + The tracks of the track file, each a list of points with Unix times, + as read_tracks() read them, or read now. They may be shared with other + readers of the file, and must not be changed + """ + if self._tracks is None: + return self._read_tracks() + if isinstance(self._tracks, exceptions.MapillaryInvalidGPXError): + raise self._tracks + return self._tracks + + def _read_tracks(self) -> list[list[geo.Point]]: + """ + Read the tracks of the GPX file, parsed once in a process (see + read_gpx_tracks()) """ try: - return parse_gpx(self.gpx_path) + return geotag_utils.read_gpx_tracks(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. " diff --git a/mapillary_tools/geotag/video_extractors/nmea.py b/mapillary_tools/geotag/video_extractors/nmea.py index f8ac1082..32e596a0 100644 --- a/mapillary_tools/geotag/video_extractors/nmea.py +++ b/mapillary_tools/geotag/video_extractors/nmea.py @@ -13,7 +13,10 @@ from typing_extensions import override from ... import exceptions, geo -from ..geotag_images_from_nmea_file import get_lat_lon_time_from_nmea +from ..geotag_images_from_nmea_file import ( + describe_invalid_nmea, + get_lat_lon_time_from_nmea, +) from ..utils import track_point from .gpx import GPXVideoExtractor @@ -26,32 +29,7 @@ class NMEAVideoExtractor(GPXVideoExtractor): file_format = "NMEA" - # The tracks of the NMEA file, or the error reading it, once read_tracks() - # has read them - _tracks: list[list[geo.Point]] | exceptions.MapillaryInvalidGPXError | None = None - - def read_tracks(self) -> None: - """ - Read the tracks of the NMEA file now, for extract() to place, rather - than when it runs, which can be in another process - """ - try: - self._tracks = self._read_tracks() - except exceptions.MapillaryInvalidGPXError as ex: - self._tracks = ex - except Exception: - # Left for extract() to raise again, and to be reported for the - # video as any unexpected error is - pass - @override - def _parse_tracks(self) -> list[list[geo.Point]]: - if self._tracks is None: - return self._read_tracks() - if isinstance(self._tracks, exceptions.MapillaryInvalidGPXError): - raise self._tracks - return self._tracks - def _read_tracks(self) -> list[list[geo.Point]]: try: points = get_lat_lon_time_from_nmea(self.gpx_path) @@ -59,8 +37,7 @@ def _read_tracks(self) -> list[list[geo.Point]]: # malformed sentences and no fix except (OSError, ValueError) as ex: raise exceptions.MapillaryInvalidGPXError( - f"{self.gpx_path.name} is not a valid NMEA file. " - "Replace it with a valid NMEA file." + describe_invalid_nmea(self.gpx_path, ex) ) from ex return [[track_point(p.time, p.lat, p.lon, p.alt) for p in points]] diff --git a/mapillary_tools/sample_video.py b/mapillary_tools/sample_video.py index b2608ddf..22afd91c 100644 --- a/mapillary_tools/sample_video.py +++ b/mapillary_tools/sample_video.py @@ -880,12 +880,13 @@ def _sample_single_video_by_interval( # Located along the video's own GPS track, or the GPX (or NMEA) track of # --video_geotag_source as GPX sync placed it on the video, as distance # samples are - track = _read_track_for_interval_samples(video_path, geotag_options) + track_read = _read_track_for_interval_samples(video_path, geotag_options) + track = track_read[0] if track_read is not None else None # The gaps of a track placed from a file, outside it (see track_gaps()) track_gaps_: list[tuple[geo.Point, geo.Point]] = [] - track_from_file = track is not None and _placed_from_a_file(geotag_options) - if track is not None and track_from_file: - track = _start_at_the_first_sample(video_path, track, geotag_options) + track_from_file = track_read is not None and _placed_from_a_file(track_read[1]) + if track_read is not None and track_from_file: + track = _start_at_the_first_sample(video_path, *track_read) track_gaps_ = track_gaps(track.points, constants.MAX_GPX_GAP_SECONDS / rate) interpolator = geo.Interpolator([track.points]) if track is not None else None # Or located by the geotagging stage along a GPX (or NMEA) track, by their @@ -1011,17 +1012,20 @@ def _sample_single_video_by_interval( exif_edit.write() -def _placed_from_a_file(geotag_options: T.Sequence[SourceOption]) -> bool: +def _placed_from_a_file(track_source: SourceOption | None) -> bool: """ - Whether the track that the interval samples are located along was placed - on the video from a GPX (or NMEA) file: its only sources are such files + Whether the track of a video was placed on it from a GPX (or NMEA) file: + the source that read it, of those tried in turn, is such a file, whatever + the other sources are """ - return bool(geotag_options) and all( - option.source in factory.EXTERNAL_GPS_SOURCES for option in geotag_options + return ( + track_source is not None and track_source.source in factory.EXTERNAL_GPS_SOURCES ) -def _placed_from_a_video_file(geotag_options: T.Sequence[SourceOption]) -> bool: +def _placed_from_a_video_file( + track_source: SourceOption | None, geotag_options: T.Sequence[SourceOption] +) -> bool: """ Whether the track that distance samples are sampled along is placed on the video from a GPX (or NMEA) file of --video_geotag_source, as interval @@ -1029,8 +1033,9 @@ def _placed_from_a_video_file(geotag_options: T.Sequence[SourceOption]) -> bool: file of --geotag_source geotags the samples too, which the geotagging stage places along its track by their capture times """ - return _placed_from_a_file(geotag_options) and not any( - option.filetypes is None or types.FileType.IMAGE in option.filetypes + return _placed_from_a_file(track_source) and not any( + option.source in factory.EXTERNAL_GPS_SOURCES + and (option.filetypes is None or types.FileType.IMAGE in option.filetypes) for option in geotag_options ) @@ -1067,16 +1072,15 @@ def _describe_file_track_gap(before: geo.Point, after: geo.Point) -> str: def _start_at_the_first_sample( video_path: Path, track: types.VideoMetadata, - geotag_options: T.Sequence[SourceOption], + option: SourceOption, ) -> types.VideoMetadata: """ A track that --interpolation_use_gpx_start_time starts with the video, moved to start at the first interval sample: GPX sync starts it at the composition time of the first frame, at which distance sampling reads the first frame, and interval sampling stamps the samples from time 0, which - ffmpeg cuts the first one at + ffmpeg cuts the first one at. **option** is the source that placed it """ - option = geotag_options[0] if option.interpolation is None or not option.interpolation.use_gpx_start_time: return track # As GPX sync starts it @@ -1273,14 +1277,15 @@ def _read_file_track_for_interval_samples( def _read_track_for_interval_samples( video_path: Path, geotag_options: T.Sequence[SourceOption] -) -> types.VideoMetadata | None: +) -> tuple[types.VideoMetadata, SourceOption] | None: """ The track that the interval samples of a video are located along, in - video time, as distance samples are: the video's own GPS track, or that of - a GPX (or NMEA) file of --video_geotag_source as GPX sync placed it on the - video. None to leave their positions to the geotagging stage: that places - them along the track of a GPX (or NMEA) file of --geotag_source, which - geotags images, by their capture times + video time, as distance samples are, and the source that read it: the + video's own GPS track, or that of a GPX (or NMEA) file of + --video_geotag_source as GPX sync placed it on the video. None to leave + their positions to the geotagging stage: that places them along the track + of a GPX (or NMEA) file of --geotag_source, which geotags images, by their + capture times """ if any( option.source in factory.EXTERNAL_GPS_SOURCES @@ -1290,17 +1295,10 @@ def _read_track_for_interval_samples( return None # Through the factory with the same sources as distance sampling, which - # reads the video's own GPS by default (see _parse_geotag_options()). One - # video is read in this process rather than in a pool of its own - video_metadatas = factory.process( - [video_path], - [ - dataclasses.replace(option, num_processes=0) - for option in geotag_options or [SourceOption(SourceType.NATIVE)] - ], + # reads the video's own GPS by default (see _parse_geotag_options()) + video_metadata, track_source = _read_video_track( + video_path, geotag_options or [SourceOption(SourceType.NATIVE)] ) - assert len(video_metadatas) == 1, "expect 1 video metadata" - video_metadata = video_metadatas[0] if isinstance(video_metadata, types.ErrorMetadata): LOG.warning( "The samples of %s are not geotagged, as no GPS track was read from it: %s", @@ -1319,7 +1317,30 @@ def _read_track_for_interval_samples( ) return None - return video_metadata + assert track_source is not None, "expect the source of a video's track" + return video_metadata, track_source + + +def _read_video_track( + video_path: Path, geotag_options: T.Sequence[SourceOption] +) -> tuple[types.MetadataOrError, SourceOption | None]: + """ + The track of a video, read through the factory by the first of + **geotag_options** that geotags it, as in the geotagging stage, and that + source (None if none does). The video is read in this process rather than + in a pool of its own, so that a GPX (or NMEA) file that geotags several + videos is read once + """ + sources: dict[Path, SourceOption] = {} + video_metadatas = factory.process( + [video_path], + [dataclasses.replace(option, num_processes=0) for option in geotag_options], + sources=sources, + ) + # Every option can read videos (see _parse_geotag_options()), so process() + # returns metadata for the video even if all of them fail + assert len(video_metadatas) == 1, "expect 1 video metadata" + return video_metadatas[0], sources.get(video_path) def _add_position( @@ -1531,12 +1552,8 @@ def _sample_single_video_by_distance( # Go through the factory rather than reading the video's own telemetry # directly, so that --geotag_source is honoured here as well: a GPX is the # documented answer for a camera whose embedded GPS is unusable, and - # distance sampling needs positions just as much as geotagging does. - video_metadatas = factory.process([video_path], geotag_options) - # Every option can read videos (see _parse_geotag_options()), so process() - # returns metadata for the video even if all of them fail - assert len(video_metadatas) == 1, "expect 1 video metadata" - video_metadata = video_metadatas[0] + # distance sampling needs positions just as much as geotagging does + video_metadata, track_source = _read_video_track(video_path, geotag_options) # Distance sampling needs positions to decide which frames to cut, so # failing to read them is a failed sample, not something to carry on past. @@ -1574,7 +1591,7 @@ def _sample_single_video_by_distance( # GPS of their video instead gaps = ( _file_track_gaps(video_metadata.points) - if _placed_from_a_video_file(geotag_options) + if _placed_from_a_video_file(track_source, geotag_options) and not any( option.source in _FIRST_SAMPLE_SOURCES for option in sample_sources or [] ) diff --git a/tests/integration/test_process.py b/tests/integration/test_process.py index 5ec3a319..517d402d 100644 --- a/tests/integration/test_process.py +++ b/tests/integration/test_process.py @@ -978,6 +978,50 @@ def track(source: str, source_path: py.path.local): assert nmea_point == pytest.approx(gpx_point, abs=1e-6) +def test_process_video_from_nmea_with_some_checksums(setup_data: py.path.local): + """ + Checksums are optional in NMEA. Where some sentences had one, those + without were all taken for cut short, and the file failed + """ + video_path = setup_data.join("gopro_data").join("max-360mode.mp4") + start = datetime.datetime(2019, 11, 18, 23, 44, 40, tzinfo=datetime.timezone.utc) + + def write_nmea(name: str, with_checksum) -> py.path.local: + sentences = [] + for i in range(21): + time = start + datetime.timedelta(seconds=i) + 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) + ending = f"*{checksum:02X}" if with_checksum(body) else "" + sentences.append(f"${body}{ending}\r\n") + nmea_file = setup_data.join(name) + nmea_file.write_binary("".join(sentences).encode("ascii")) + return nmea_file + + def track(nmea_file: py.path.local): + [desc] = run_process_for_descs( + [ + *["--geotag_source", "nmea"], + *["--geotag_source_path", str(nmea_file)], + str(video_path), + ] + ) + assert "error" not in desc, desc + return desc["MAPGPSTrack"] + + every = track(write_nmea("every.nmea", lambda body: True)) + assert every + assert track(write_nmea("gga.nmea", lambda body: body.startswith("GPGGA"))) == every + assert track(write_nmea("one.nmea", lambda body: "234450" in body)) == every + + 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_video_process.py b/tests/integration/test_video_process.py index ebeb1970..4469e0ff 100644 --- a/tests/integration/test_video_process.py +++ b/tests/integration/test_video_process.py @@ -265,6 +265,53 @@ def test_video_process_by_interval_leaves_out_the_samples_past_the_gpx( assert "and were left out" in process.stderr +def test_video_process_reads_a_gpx_file_of_several_videos_once( + setup_data: py.path.local, +): + """ + Sampling by interval read the GPX file of --geotag_source again for each + video, to tell which samples fall outside it, and warned about it each + time. It is read, and warned about, once + """ + pytest_skip_if_not_ffmpeg_installed() + + gpx = setup_data.join("gpx").join("sf_30km_h.gpx").read_text("utf-8") + # Two tracks, which are merged with a warning + points = [m.start() for m in re.finditer("" + gpx[middle:] + + with tempfile.TemporaryDirectory() as dir: + videos = Path(dir, "videos") + videos.mkdir() + for name in ("a.mp4", "b.mp4"): + shutil.copy( + str(setup_data.join("videos").join("sample-5s.mp4")), videos / name + ) + gpx_file = Path(dir, "two_tracks.gpx") + gpx_file.write_bytes(two_tracks.encode("utf-8")) + process = run_video_process( + [ + "--video_sample_distance=-1", + "--video_sample_interval=1", + *["--video_start_time", "2025_03_14_07_01_31_000"], + "--geotag_source=gpx", + *["--geotag_source_path", str(gpx_file)], + *["--desc_path", str(Path(dir, "desc.json"))], + str(videos), + str(Path(dir, "samples")), + ], + capture_output=True, + text=True, + ) + with open(Path(dir, "desc.json")) as fp: + descs = json.load(fp) + + assert len(descs) == 6, descs + assert all("error" not in desc for desc in descs), descs + assert process.stderr.count("Found 2 tracks in the GPX file") == 1, process.stderr + + def test_video_process_by_interval_with_a_video_gpx(setup_data: py.path.local): """ The samples are located along the GPX of --video_geotag_source as GPX sync @@ -303,17 +350,11 @@ def test_video_process_by_interval_with_a_video_gpx(setup_data: py.path.local): assert desc["MAPLongitude"] == pytest.approx(gpx_desc["MAPLongitude"], abs=1e-7) -def test_video_process_by_distance_leaves_out_the_frames_in_a_gap_of_a_video_gpx( - setup_data: py.path.local, -): +def _gpx_with_a_gap(setup_data: py.path.local) -> str: """ - Without its point at 07:00:33.088, the GPX has a gap from 07:00:31.533 to - 07:00:34.379, over the limit, inside the video. The frames in it were - placed across it + The GPX without its point at 07:00:33.088: with a gap from 07:00:31.533 to + 07:00:34.379, over a limit of 2.5 seconds """ - pytest_skip_if_not_ffmpeg_installed() - - video_path = setup_data.join("videos").join("sample-5s.mp4") gpx = setup_data.join("gpx").join("sf_30km_h.gpx").read_text("utf-8") gpx_with_gap = re.sub( r").)*?.*?", @@ -322,13 +363,29 @@ def test_video_process_by_distance_leaves_out_the_frames_in_a_gap_of_a_video_gpx flags=re.S, ) assert gpx_with_gap != gpx + return gpx_with_gap + + +# Whatever source follows the GPX, which the video, without GPS of its own, +# leaves to geotag it +@pytest.mark.parametrize("then", [[], ["--video_geotag_source=native"]]) +def test_video_process_by_distance_leaves_out_the_frames_in_a_gap_of_a_video_gpx( + setup_data: py.path.local, then: list[str] +): + """ + The GPX has a gap inside the video. The frames in it were placed across it + """ + pytest_skip_if_not_ffmpeg_installed() + + video_path = setup_data.join("videos").join("sample-5s.mp4") with tempfile.TemporaryDirectory() as dir: gpx_file = Path(dir, "gap.gpx") - gpx_file.write_text(gpx_with_gap, encoding="utf-8") + gpx_file.write_bytes(_gpx_with_a_gap(setup_data).encode("utf-8")) descs = run_video_process_for_descs( [ "--video_geotag_source=gpx", + *then, *["--geotag_source_path", str(gpx_file)], *["--video_start_time", "2025_03_14_07_00_30_000"], str(video_path), @@ -348,6 +405,43 @@ def test_video_process_by_distance_leaves_out_the_frames_in_a_gap_of_a_video_gpx ], capture_times +@pytest.mark.parametrize("then", [[], ["--video_geotag_source=native"]]) +def test_video_process_by_interval_leaves_out_the_samples_in_a_gap_of_a_video_gpx( + setup_data: py.path.local, then: list[str] +): + """ + The GPX has a gap inside the video, from 1.533 to 4.379 seconds into it. + The samples in it were placed across it when another source followed the + GPX + """ + pytest_skip_if_not_ffmpeg_installed() + + video_path = setup_data.join("videos").join("sample-5s.mp4") + + with tempfile.TemporaryDirectory() as dir: + gpx_file = Path(dir, "gap.gpx") + gpx_file.write_bytes(_gpx_with_a_gap(setup_data).encode("utf-8")) + descs = run_video_process_for_descs( + [ + "--video_sample_distance=-1", + "--video_sample_interval=0.5", + "--video_geotag_source=gpx", + *then, + *["--geotag_source_path", str(gpx_file)], + *["--video_start_time", "2025_03_14_07_00_30_000"], + str(video_path), + str(Path(dir, "samples")), + ], + env={**os.environ, "MAPILLARY_TOOLS_MAX_GPX_GAP_SECONDS": "2.5"}, + ) + + assert all("error" not in desc for desc in descs), descs + # Of the 11 samples, from 0 to 5 seconds, not those at 2, 2.5, 3, 3.5 and 4 + assert sorted(Path(desc["filename"]).name for desc in descs) == [ + f"sample-5s_v_{idx + 1:06d}.jpg" for idx in [0, 1, 2, 3, 9, 10] + ] + + def _with_creation_time(video: py.path.local, creation_time: str, dir: str) -> Path: """A copy of the video whose container records the creation time.""" copy = Path(dir, video.basename) diff --git a/tests/unit/test_gpx_video_coverage.py b/tests/unit/test_gpx_video_coverage.py index 12500149..e72f3ac5 100644 --- a/tests/unit/test_gpx_video_coverage.py +++ b/tests/unit/test_gpx_video_coverage.py @@ -24,7 +24,15 @@ from pathlib import Path import pytest -from mapillary_tools import constants, exceptions, geo, telemetry, types, uploader +from mapillary_tools import ( + constants, + exceptions, + geo, + sample_video, + telemetry, + types, + uploader, +) from mapillary_tools.camm import camm_builder from mapillary_tools.geotag import factory from mapillary_tools.geotag.options import ( @@ -508,6 +516,53 @@ def test_video_is_geotagged_by_the_next_source_with_a_warning( assert "misses the video" in warning and "by 2.0 hours" in warning assert warning.endswith("Geotagged it from the native source instead") + @pytest.mark.parametrize( + "gpx_start, source", + [(A_UNIX_TIME, SourceType.GPX), (A_UNIX_TIME + 2 * 3600, SourceType.NATIVE)], + ) + def test_the_source_that_geotagged_the_video_is_given( + self, tmp_path, gpx_start, source + ): + """ + Interval and distance samples leave out the gaps of a track placed from + a GPX, and not those of the video's own track + """ + video_path = _write_camm_mp4(tmp_path / "video.mp4", _video_track()) + gpx_path = _write_gpx(tmp_path / "track.gpx", gpx_start, 10) + options = [ + # As --video_geotag_source gpx --video_geotag_source native + SourceOption( + SourceType.GPX, + num_processes=0, + source_path=SourcePathOption(source_path=gpx_path), + filetypes={types.FileType.VIDEO}, + ), + SourceOption( + SourceType.NATIVE, num_processes=0, filetypes={types.FileType.VIDEO} + ), + ] + + sources: dict[Path, SourceOption] = {} + [metadata] = factory.process([video_path], options, sources=sources) + assert isinstance(metadata, types.VideoMetadata) + assert sources[video_path].source is source + + read = sample_video._read_track_for_interval_samples(video_path, options) + assert read is not None + assert read[1].source is source + assert sample_video._placed_from_a_file(read[1]) is (source is SourceType.GPX) + + def test_no_source_is_given_for_a_video_none_geotagged(self, tmp_path): + sources: dict[Path, SourceOption] = {} + [metadata] = factory.process( + [NO_GPS_VIDEO], + [SourceOption(SourceType.NATIVE, num_processes=0)], + sources=sources, + ) + + assert isinstance(metadata, types.ErrorMetadata) + assert sources == {} + def _process_images( self, tmp_path: Path, sources: list[SourceType], count: int = 1 ) -> list[types.MetadataOrError]: diff --git a/tests/unit/test_nmea_parser.py b/tests/unit/test_nmea_parser.py index 9b08715a..74398fcd 100644 --- a/tests/unit/test_nmea_parser.py +++ b/tests/unit/test_nmea_parser.py @@ -260,7 +260,7 @@ def test_a_sentence_cut_inside_its_date_is_skipped(tmp_path, caplog, gga_first): ] + ([_fix("00:00:01")] if gga_first else []) assert "Skipping 1 of the" in caplog.text - assert "(no checksum)" in caplog.text + assert "(malformed datestamp: '2212')" in caplog.text def test_sentences_cut_short_are_skipped_with_one_warning(tmp_path, caplog): @@ -320,6 +320,123 @@ def bare(sentence: str) -> str: assert f"(malformed {error}: " in caplog.text +def _bare(sentence: str) -> str: + """The sentence without its checksum, which is optional""" + return sentence.rsplit("*", 1)[0] + + +def _log(seconds: int = 4) -> list[str]: + lines = [] + for second in range(seconds): + lines += [_rmc(f"02463{second}.00"), _gga(f"02463{second}.00")] + return lines + + +def _fixes(seconds: int = 4) -> list: + return [_fix(f"02:46:3{second}") for second in range(seconds)] + + +def test_a_log_with_one_checksum_is_read(tmp_path, caplog): + """The sentences without one were all taken for cut short: no fix was left""" + lines = [_bare(line) for line in _log()] + lines[3] = _log()[3] + + with caplog.at_level(logging.WARNING): + assert _read(tmp_path, *lines) == _fixes() + + assert not caplog.text + + +def test_a_log_with_checksums_on_gga_sentences_only_is_read(tmp_path, caplog): + lines = [line if "GGA," in line else _bare(line) for line in _log()] + + with caplog.at_level(logging.WARNING): + assert _read(tmp_path, *lines) == _fixes() + + assert not caplog.text + + +def test_logs_with_and_without_checksums_concatenated_are_read(tmp_path, caplog): + """The half without checksums was skipped""" + lines = _log() + lines = lines[:4] + [_bare(line) for line in lines[4:]] + + with caplog.at_level(logging.WARNING): + assert _read(tmp_path, *lines) == _fixes() + + assert not caplog.text + + +@pytest.mark.parametrize( + "rmc_tail", + [ + # NMEA 2.0 to 2.2: no mode + ",,", + # NMEA 2.3: the mode + ",,,A", + # NMEA 4.1: the mode and the navigational status + ",,,A,V", + ], +) +def test_complete_sentences_without_checksums_are_read(tmp_path, caplog, rmc_tail): + rmc = "$GPRMC,024636.00,A,3735.30670,N,12220.25810,W,0.0,0.0,221222" + rmc_tail + gga = _bare(_gga("024636.00")) + + with caplog.at_level(logging.WARNING): + assert _read(tmp_path, rmc, gga) == [_fix("02:46:36")] + + assert not caplog.text + + +@pytest.mark.parametrize( + "cut, reason", + [ + # Whole fields, the last of them maybe cut short too + ( + "$GPRMC,024637.00,A,3735.30670,N,12220.25810,W,0.0,0.0,221222", + "cut short: 9 of the 11 fields of an RMC sentence", + ), + ( + "$GPGGA,024637.00,3735.30670,N,12220.25810,W,1,08,1.0,12", + "cut short: 9 of the 14 fields of a GGA sentence", + ), + # A time that does not parse + ("$GPGGA,02463", "malformed timestamp: '02463'"), + ], +) +def test_sentences_cut_short_by_their_own_fields_are_skipped( + tmp_path, caplog, cut, reason +): + with caplog.at_level(logging.WARNING): + assert _read(tmp_path, *_log(1), cut) == _fixes(1) + + assert f"on line 3 ({reason}): {cut}" in caplog.text + + +def test_a_sentence_with_its_checksum_is_not_judged_by_its_fields(tmp_path, caplog): + """A checksum shows where the sentence ends, however few fields it has""" + short_gga = _sentence("GPGGA,024636.00,3735.30670,N,12220.25810,W,1,08,1.0,12.5,M") + + with caplog.at_level(logging.WARNING): + assert _read(tmp_path, _rmc("024636.00"), short_gga) == [_fix("02:46:36")] + + assert not caplog.text + + +def test_a_file_that_fails_says_why(tmp_path): + """It failed as "not a valid NMEA file", without a reason""" + good = _rmc("024636.00") + lines = [good[:-2] + "00", _bare(_gga("024636.00"))[:30]] + + with pytest.raises(ValueError) as excinfo: + _read(tmp_path, *lines) + + assert str(excinfo.value) == ( + "2 of its 2 RMC and GGA sentences are malformed, the first on line 1 " + f"(checksum does not match: 00 != {good[-2:]}): {lines[0]}" + ) + + def _read_bytes(tmp_path: Path, content: bytes): path = tmp_path / "track.nmea" path.write_bytes(content) diff --git a/tests/unit/test_nmea_video.py b/tests/unit/test_nmea_video.py index 4a5629ca..e3ccb01d 100644 --- a/tests/unit/test_nmea_video.py +++ b/tests/unit/test_nmea_video.py @@ -193,25 +193,33 @@ def test_messages_name_the_nmea_file(self, tmp_path): class TestInvalidNMEAFile: @pytest.mark.parametrize( - "content", + "content, reason", [ # not text - b"\xca\xfe\xba\xbe" * 100, + (b"\xca\xfe\xba\xbe" * 100, "it is not text"), # a wrong checksum - b"$GPRMC,024636.00,A,3735.30670,N,12220.25810,W,0.0,0.0,221222,,,A*00\r\n", + ( + b"$GPRMC,024636.00,A,3735.30670,N,12220.25810,W,0.0,0.0,221222,,,A*00\r\n", + "the first on line 1 (checksum does not match: 00 != 47)", + ), # 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", + ( + 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", + "the first on line 2 (malformed timestamp: '0246')", + ), ], ) - def test_it_is_reported_as_invalid(self, tmp_path, content): + def test_it_is_reported_as_invalid(self, tmp_path, content, reason): 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) + assert "track.nmea is not a valid NMEA file: " in str(excinfo.value) + # Why, which it did not say + assert reason in str(excinfo.value) class TestNMEAFileWithoutDatedFixes: diff --git a/tests/unit/test_read_track_files_once.py b/tests/unit/test_read_track_files_once.py new file mode 100644 index 00000000..960a74df --- /dev/null +++ b/tests/unit/test_read_track_files_once.py @@ -0,0 +1,243 @@ +# 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. + +""" +A GPX file that geotags several videos, or the samples of a video and then +their images, is parsed once in a run, and warned about once +""" + +from __future__ import annotations + +import logging +import multiprocessing +from pathlib import Path + +import gpxpy +import pytest + +from mapillary_tools import sample_video +from mapillary_tools.geotag import ( + geotag_images_from_gpx_file, + geotag_videos_from_gpx, + utils, +) +from mapillary_tools.geotag.options import SourceOption, SourcePathOption, SourceType +from mapillary_tools.geotag.video_extractors.gpx import GPXVideoExtractor + + +def _gpx(*segments: list[tuple[str, float]]) -> bytes: + """A GPX file of a track with **segments** of (time, latitude) points""" + trksegs = "".join( + "" + + "".join( + f'' + for time, lat in segment + ) + + "" + for segment in segments + ) + return ( + '' + '' + f"{trksegs}" + ).encode("utf-8") + + +ONE_TRACK = _gpx([("2025-01-01T00:00:00Z", 52.0), ("2025-01-01T00:00:01Z", 52.1)]) +TWO_TRACKS = _gpx( + [("2025-01-01T00:00:00Z", 52.0), ("2025-01-01T00:00:01Z", 52.1)], + [("2025-01-01T00:00:02Z", 52.2), ("2025-01-01T00:00:03Z", 52.3)], +) + + +@pytest.fixture +def parses(monkeypatch) -> list[int]: + """Counts the GPX files parsed, with nothing parsed or warned of before""" + if hasattr(utils, "_PARSED_GPX"): + monkeypatch.setattr(utils, "_PARSED_GPX", type(utils._PARSED_GPX)()) + if hasattr(utils, "_WARNED"): + monkeypatch.setattr(utils, "_WARNED", set()) + parsed: list[int] = [] + parse = gpxpy.parse + + def counting_parse(*args, **kwargs): + parsed.append(1) + return parse(*args, **kwargs) + + monkeypatch.setattr(gpxpy, "parse", counting_parse) + return parsed + + +def test_a_gpx_file_is_parsed_once(tmp_path: Path, parses: list[int]): + gpx = tmp_path / "track.gpx" + gpx.write_bytes(ONE_TRACK) + + first = utils.parse_gpx(gpx) + second = utils.parse_gpx(gpx) + + assert len(parses) == 1 + assert first == second + # Copies, which the caller may change + first[0][0].time = 0 + assert utils.parse_gpx(gpx)[0][0].time != 0 + + +def test_a_changed_gpx_file_is_parsed_again(tmp_path: Path, parses: list[int]): + gpx = tmp_path / "track.gpx" + gpx.write_bytes(ONE_TRACK) + assert len(utils.parse_gpx(gpx)) == 1 + + gpx.write_bytes(TWO_TRACKS) + assert len(utils.parse_gpx(gpx)) == 2 + assert len(parses) == 2 + + +def test_only_the_last_gpx_files_are_kept(tmp_path: Path, parses: list[int]): + paths = [] + for idx in range(utils._MAX_PARSED_GPX + 1): + gpx = tmp_path / f"track{idx}.gpx" + gpx.write_bytes(ONE_TRACK) + paths.append(gpx) + utils.read_gpx_tracks(gpx) + + assert len(utils._PARSED_GPX) == utils._MAX_PARSED_GPX + assert not utils.was_gpx_read(paths[0]) + assert utils.was_gpx_read(paths[-1]) + + +def test_a_worker_of_a_pool_keeps_no_gpx_file( + tmp_path: Path, parses: list[int], monkeypatch +): + """It parses the GPX file of one video only, and would keep it until it exits""" + gpx = tmp_path / "track.gpx" + gpx.write_bytes(ONE_TRACK) + monkeypatch.setattr(multiprocessing, "parent_process", lambda: object()) + + assert len(utils.read_gpx_tracks(gpx)) == 1 + assert not utils.was_gpx_read(gpx) + + +def test_an_invalid_gpx_file_is_parsed_once(tmp_path: Path, parses: list[int]): + gpx = tmp_path / "track.gpx" + gpx.write_bytes(b"not a GPX file") + + for _ in range(2): + with pytest.raises(Exception): + utils.read_gpx_tracks(gpx) + assert len(parses) == 1 + + +def test_images_warn_of_the_tracks_once(tmp_path: Path, parses: list[int], caplog): + gpx = tmp_path / "track.gpx" + gpx.write_bytes(TWO_TRACKS) + + with caplog.at_level(logging.WARNING): + for _ in range(3): + points = geotag_images_from_gpx_file._read_gpx_points(gpx) + + assert len(points) == 4 + assert len(parses) == 1 + assert caplog.text.count("Found 2 tracks in the GPX file") == 1 + + +def test_videos_warn_of_the_tracks_once(tmp_path: Path, parses: list[int], caplog): + gpx = tmp_path / "track.gpx" + gpx.write_bytes(TWO_TRACKS) + + with caplog.at_level(logging.WARNING): + geotag_images_from_gpx_file._read_gpx_points(gpx) + for name in ("a.mp4", "b.mp4"): + points = GPXVideoExtractor(tmp_path / name, gpx)._read_gpx_points() + + assert len(points) == 4 + assert len(parses) == 1 + assert caplog.text.count("Found 2 tracks") == 1 + + +def test_video_points_are_copies(tmp_path: Path, parses: list[int]): + """GPX sync moves the points of a track to the video""" + gpx = tmp_path / "track.gpx" + gpx.write_bytes(ONE_TRACK) + + points = GPXVideoExtractor(tmp_path / "a.mp4", gpx)._read_gpx_points() + for point in points: + point.time = 0 + + assert [p.time for p in utils.parse_gpx(gpx)[0]] == [ + 1735689600.0, + 1735689601.0, + ] + + +def test_a_gpx_file_of_several_videos_is_read_before_the_pool( + tmp_path: Path, parses: list[int] +): + shared = tmp_path / "shared.gpx" + shared.write_bytes(ONE_TRACK) + own = tmp_path / "c.gpx" + own.write_bytes(TWO_TRACKS) + videos = [] + for name in ("a.mp4", "b.mp4", "c.mp4"): + videos.append(tmp_path / name) + videos[-1].write_bytes(b"") + + shared_geotag = geotag_videos_from_gpx.GeotagVideosFromGPX( + source_path=SourcePathOption(source_path=shared) + ) + extractors = shared_geotag._generate_video_extractors(videos) + assert all( + isinstance(e, GPXVideoExtractor) and e._tracks is not None for e in extractors + ) + assert len(parses) == 1 + + # A GPX file of each video is left to the pool + own_geotag = geotag_videos_from_gpx.GeotagVideosFromGPX( + source_path=SourcePathOption(pattern="%g.gpx") + ) + (extractor,) = own_geotag._generate_video_extractors(videos[2:]) + assert isinstance(extractor, GPXVideoExtractor) + assert extractor._tracks is None + + # Unless this process parsed it already + utils.read_gpx_tracks(own) + (extractor,) = own_geotag._generate_video_extractors(videos[2:]) + assert isinstance(extractor, GPXVideoExtractor) + assert extractor._tracks is not None + assert len(parses) == 2 + + +def test_interval_samples_read_the_gpx_file_once(tmp_path: Path, parses: list[int]): + gpx = tmp_path / "track.gpx" + gpx.write_bytes(ONE_TRACK) + sources = [ + SourceOption(SourceType.GPX, source_path=SourcePathOption(source_path=gpx)) + ] + + for _ in range(3): + track = sample_video._read_file_track_for_interval_samples(sources) + assert track is not None + + assert len(parses) == 1 + + +def test_distance_sampling_reads_the_video_in_this_process(tmp_path: Path, monkeypatch): + """So that a GPX file that geotags several videos is parsed once""" + calls = [] + + def process(paths, options, filetypes=None, sources=None): + calls.append([option.num_processes for option in options]) + raise RuntimeError("stop here") + + monkeypatch.setattr(sample_video.factory, "process", process) + monkeypatch.setattr(sample_video, "_probe_video", lambda ffmpeg, path: None) + + with pytest.raises(RuntimeError, match="stop here"): + sample_video._sample_single_video_by_distance( + tmp_path / "a.mp4", + tmp_path / "samples", + sample_distance=3, + geotag_options=[SourceOption(SourceType.NATIVE)], + ) + assert calls == [[0]] diff --git a/tests/unit/test_sample_video.py b/tests/unit/test_sample_video.py index e5f3b99a..78485f36 100644 --- a/tests/unit/test_sample_video.py +++ b/tests/unit/test_sample_video.py @@ -456,7 +456,13 @@ def _video_metadata(video_path: Path, points: T.Sequence[geo.Point]): model="HERO7 Black", ) - def _sample(self, tmpdir, monkeypatch, track_points, **kwargs) -> list[Path]: + def _sample( + self, tmpdir, monkeypatch, track_points, read_by: int = 0, **kwargs + ) -> list[Path]: + """ + Sample the video every 2 seconds, its track read by the source at + index **read_by** of those sampling tries + """ clock_points = self._points(60) monkeypatch.setattr( sample_video, @@ -467,7 +473,7 @@ def _sample(self, tmpdir, monkeypatch, track_points, **kwargs) -> list[Path]: ) self.track_calls: list[list[SourceOption]] = [] - def fake_process(paths, options): + def fake_process(paths, options, sources=None): self.track_calls.append(list(options)) [video_path] = paths if isinstance(track_points, Exception): @@ -476,6 +482,8 @@ def fake_process(paths, options): track_points, video_path, filetype=FileType.VIDEO ) ] + if sources is not None: + sources[video_path] = options[read_by] return [self._video_metadata(video_path, track_points)] monkeypatch.setattr(sample_video.factory, "process", fake_process) @@ -653,8 +661,13 @@ def test_located_along_a_video_track_file( in caplog.text ) + @pytest.mark.parametrize( + "video_geotag_source", + # Whatever source follows the one that read the track + [["gpx"], ["gpx", "native"], ["nmea", "exiftool_runtime"]], + ) def test_samples_in_a_gap_of_a_video_track_file_are_left_out( - self, tmpdir, setup_mock, monkeypatch + self, tmpdir, setup_mock, monkeypatch, video_geotag_source ): monkeypatch.setattr(sample_video.constants, "MAX_GPX_GAP_SECONDS", 5.0) track = self._file_points([*range(11), *range(20, 61)]) @@ -663,7 +676,7 @@ def test_samples_in_a_gap_of_a_video_track_file_are_left_out( tmpdir, monkeypatch, track, - video_geotag_source=["gpx"], + video_geotag_source=video_geotag_source, geotag_source_path=Path(tmpdir) / "track.gpx", ) @@ -672,8 +685,9 @@ def test_samples_in_a_gap_of_a_video_track_file_are_left_out( f"hello_v_{idx + 1:06d}.jpg" for idx in [*range(6), *range(10, 30)] ] + @pytest.mark.parametrize("video_geotag_source", [["gpx"], ["gpx", "native"]]) def test_a_video_track_file_started_with_the_video_starts_at_the_first_sample( - self, tmpdir, setup_mock, monkeypatch + self, tmpdir, setup_mock, monkeypatch, video_geotag_source ): """ GPX sync starts it at the first frame, 0.5 seconds into the video, and @@ -686,7 +700,7 @@ def test_a_video_track_file_started_with_the_video_starts_at_the_first_sample( tmpdir, monkeypatch, track, - video_geotag_source=["gpx"], + video_geotag_source=video_geotag_source, geotag_source_path=Path(tmpdir) / "track.gpx", interpolation_use_gpx_start_time=True, ) @@ -696,6 +710,40 @@ def test_a_video_track_file_started_with_the_video_starts_at_the_first_sample( lon_lat = exif_read.ExifRead(samples[0]).extract_lon_lat() assert lon_lat == pytest.approx((-74.0, 40.0005)) + @pytest.mark.parametrize( + "video_geotag_source", + # After a GPX that misses the video, or before a GPX + [["gpx", "native"], ["native", "gpx"]], + ) + def test_a_track_read_by_the_video_is_located_along_as_its_own( + self, tmpdir, setup_mock, monkeypatch, video_geotag_source + ): + """ + The video's own track, with a gap over the limit from 10 to 20 seconds, + locates the samples as without a GPX + """ + monkeypatch.setattr(sample_video.constants, "MAX_GPX_GAP_SECONDS", 5.0) + track = [p for p in self._points(60) if not 10 < p.time < 20] + + samples = self._sample( + tmpdir.mkdir("both"), + monkeypatch, + track, + read_by=video_geotag_source.index("native"), + video_geotag_source=video_geotag_source, + geotag_source_path=Path(tmpdir) / "track.gpx", + ) + own = self._sample( + tmpdir.mkdir("own"), monkeypatch, track, video_geotag_source=["native"] + ) + + assert len(samples) == 30 + assert [s.name for s in samples] == [s.name for s in own] + for sample, own_sample in zip(samples, own): + exif, own_exif = exif_read.ExifRead(sample), exif_read.ExifRead(own_sample) + assert exif.extract_capture_time() == own_exif.extract_capture_time() + assert exif.extract_lon_lat() == own_exif.extract_lon_lat() + def test_none_left_out_of_a_video_track_file_where_the_video_source_may_geotag_them( self, tmpdir, setup_mock, monkeypatch ): @@ -1584,7 +1632,7 @@ def unreadable_gps(self, monkeypatch): monkeypatch.setattr( sample_video.factory, "process", - lambda paths, options: [ + lambda paths, options, sources=None: [ describe_error_metadata( error, filename=paths[0], filetype=FileType.GOPRO ) @@ -1651,7 +1699,7 @@ def test_an_empty_track_is_also_an_error(self, tmpdir, setup_mock, monkeypatch): monkeypatch.setattr( sample_video.factory, "process", - lambda paths, options: [ + lambda paths, options, sources=None: [ VideoMetadata( filename=paths[0], filesize=0, filetype=FileType.GOPRO, points=[] ) @@ -1667,7 +1715,7 @@ def test_a_missing_video_stream_is_also_an_error( monkeypatch.setattr( sample_video.factory, "process", - lambda paths, options: [ + lambda paths, options, sources=None: [ VideoMetadata( filename=paths[0], filesize=0, @@ -1780,6 +1828,8 @@ def _setup_mocks( num_gps_points: int = 10, gps_points: T.Sequence[geo.Point] | None = None, sample_times: T.Sequence[float] | None = None, + # The index of the source that reads the track, of those tried + read_by: int = 0, ) -> dict[str, T.Any]: """Set up all the mocks needed for _sample_single_video_by_distance.""" probe_output = _load_probe_output() @@ -1843,9 +1893,14 @@ def fake_extract_frames( mock_ffmpeg_class, ) + def fake_process(paths, options, sources=None): + if sources is not None: + sources[paths[0]] = options[read_by] + return [video_metadata] + patches["geotag_cls"] = mock.patch( "mapillary_tools.sample_video.factory.process", - return_value=[video_metadata], + side_effect=fake_process, ) patches["moov_parse"] = mock.patch.object( @@ -2325,9 +2380,16 @@ def _file_track(self, video_times, rate: float = 1.0) -> list[geo.Point]: ] def _sample_along_a_file_track( - self, tmp_path: Path, gps_points: T.Sequence[geo.Point], **kwargs + self, + tmp_path: Path, + gps_points: T.Sequence[geo.Point], + read_by: int = 0, + **kwargs, ) -> list[datetime.datetime | None]: - """Sample a frame a second by distance, and read their capture times""" + """ + Sample a frame a second by distance, the track read by the source at + index **read_by** of those tried, and read their capture times + """ video_file = tmp_path / "test.mp4" video_file.touch() output_dir = tmp_path / "output" @@ -2336,6 +2398,7 @@ def _sample_along_a_file_track( video_file, gps_points=gps_points, sample_times=[float(t) for t in range(int(gps_points[-1].time) + 1)], + read_by=read_by, ) with ( @@ -2354,9 +2417,11 @@ def _sample_along_a_file_track( frames = sorted((output_dir / "test.mp4").glob("*.jpg")) return [exif_read.ExifRead(frame).extract_capture_time() for frame in frames] + # Whatever source follows the one that read the track + @pytest.mark.parametrize("then", [[], ["native"], ["exiftool_runtime"]]) @pytest.mark.parametrize("source", ["gpx", "nmea"]) def test_frames_in_a_gap_of_a_video_track_file_are_left_out( - self, tmp_path: Path, monkeypatch, caplog, source + self, tmp_path: Path, monkeypatch, caplog, source, then ) -> None: """ The track of --video_geotag_source has a gap of 10 seconds, over the @@ -2368,7 +2433,9 @@ def test_frames_in_a_gap_of_a_video_track_file_are_left_out( with caplog.at_level(logging.WARNING, logger=sample_video.LOG.name): capture_times = self._sample_along_a_file_track( - tmp_path, self._file_track(video_times), video_geotag_source=[source] + tmp_path, + self._file_track(video_times), + video_geotag_source=[source, *then], ) assert capture_times == [ @@ -2424,6 +2491,28 @@ def test_frames_in_a_gap_are_left_where_the_geotagging_stage_places_them( self.GPS_START_TIME + datetime.timedelta(seconds=t) for t in range(31) ] + @pytest.mark.parametrize( + "video_geotag_source", + # After a GPX that misses the video, or before a GPX + [["gpx", "native"], ["native", "gpx"]], + ) + def test_frames_in_a_gap_of_a_track_read_by_the_video_are_kept( + self, tmp_path: Path, monkeypatch, video_geotag_source + ) -> None: + """The video's own track is sampled across its gaps, as without a GPX""" + monkeypatch.setattr(sample_video.constants, "MAX_GPX_GAP_SECONDS", 5.0) + + capture_times = self._sample_along_a_file_track( + tmp_path, + self._file_track([*range(11), *range(20, 31)]), + read_by=video_geotag_source.index("native"), + video_geotag_source=video_geotag_source, + ) + + assert capture_times == [ + self.GPS_START_TIME + datetime.timedelta(seconds=t) for t in range(31) + ] + def test_timelapse_frames_past_the_end_are_stamped_within_the_track( self, tmp_path: Path ) -> None: @@ -2700,7 +2789,7 @@ def test_unreadable_start_time_by_distance( monkeypatch.setattr( sample_video.factory, "process", - lambda paths, options: [ + lambda paths, options, sources=None: [ VideoMetadata( filename=paths[0], filesize=1, @@ -2877,7 +2966,7 @@ def test_the_sampler_asks_the_factory_for_them( seen: list = [] points = _make_gps_points(4, time_step=1.0) - def fake_process(paths, options): + def fake_process(paths, options, sources=None): seen.append(list(options)) return [ VideoMetadata( @@ -2967,7 +3056,7 @@ def test_every_option_carries_it(self): def test_sample_video_passes_it_on(self, tmpdir, setup_mock, monkeypatch): seen: list = [] - def fake_process(paths, options): + def fake_process(paths, options, sources=None): seen.extend(options) return [ VideoMetadata( @@ -3006,7 +3095,7 @@ def test_sample_video_passes_the_interpolation_options_on( """video_process registers them for its geotagging stage.""" seen: list = [] - def fake_process(paths, options): + def fake_process(paths, options, sources=None): seen.extend(options) return [ VideoMetadata( @@ -3077,7 +3166,7 @@ def test_it_fails_instead_of_writing_an_empty_sample_directory( monkeypatch.setattr( sample_video.factory, "process", - lambda paths, options: [ + lambda paths, options, sources=None: [ VideoMetadata( filename=paths[0], filesize=1,