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_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..84d59775 100644 --- a/mapillary_tools/geotag/video_extractors/nmea.py +++ b/mapillary_tools/geotag/video_extractors/nmea.py @@ -26,32 +26,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) 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_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_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]]