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..3492eae7 100644 --- a/mapillary_tools/sample_video.py +++ b/mapillary_tools/sample_video.py @@ -1531,8 +1531,13 @@ 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) + # distance sampling needs positions just as much as geotagging 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 + video_metadatas = factory.process( + [video_path], + [dataclasses.replace(option, num_processes=0) for option in 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" 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..be416ce6 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 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..787422c5 --- /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): + 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]]