diff --git a/README.md b/README.md
index b702a5a6..b8655af4 100644
--- a/README.md
+++ b/README.md
@@ -240,7 +240,9 @@ description-level processing error are included in `processed_file_count`.
Unsupported entries include their filename, category, stable reason code, and
normalized extension. Hidden/system files and common GPS or metadata sidecars
are ignored during directory discovery, but an explicitly supplied unsupported
-file is always reported. The report is written only to the requested local path
+file is always reported. Such a file also fails processing, with a
+`MapillaryUnsupportedFileError` entry in the image description file, whether
+or not the report is requested. The report is written only to the requested local path
and contains absolute source filenames, so treat it as local capture metadata.
## Upload
diff --git a/mapillary_tools/exceptions.py b/mapillary_tools/exceptions.py
index dac650ef..b477b6e6 100644
--- a/mapillary_tools/exceptions.py
+++ b/mapillary_tools/exceptions.py
@@ -58,6 +58,13 @@ class MapillaryGeoTaggingError(MapillaryDescriptionError):
pass
+class MapillaryUnsupportedFileError(MapillaryDescriptionError):
+ """
+ A file named for processing is neither an image nor a video that can be
+ processed
+ """
+
+
class MapillaryVideoGPSUnusableError(MapillaryDescriptionError):
"""
The video has no GPS track that can be used: none at all, an empty one, or
diff --git a/mapillary_tools/geotag/geotag_images_from_gpx.py b/mapillary_tools/geotag/geotag_images_from_gpx.py
index ef9fdca7..aafa7cf6 100644
--- a/mapillary_tools/geotag/geotag_images_from_gpx.py
+++ b/mapillary_tools/geotag/geotag_images_from_gpx.py
@@ -28,6 +28,75 @@
LOG = logging.getLogger(__name__)
+def check_within_track(
+ time: float,
+ sorted_points: T.Sequence[geo.Point],
+ gaps: T.Sequence[tuple[geo.Point, geo.Point]] = (),
+) -> None:
+ """
+ Raise MapillaryOutsideGPXTrackError if **time** falls outside the track of
+ **sorted_points**, with a tolerance of 1 ms: before its start, after its
+ end, or in one of its **gaps** (see track_gaps())
+
+ >>> track = [geo.Point(time=t, lat=0, lon=0, alt=None, angle=None) for t in (10, 20, 9000)]
+ >>> check_within_track(15.0, track)
+ >>> check_within_track(9000.0005, track)
+ >>> check_within_track(9.5, track)
+ Traceback (most recent call last):
+ ...
+ mapillary_tools.exceptions.MapillaryOutsideGPXTrackError: The image date time is 0.5 seconds behind the GPX start point
+ >>> check_within_track(4000.0, track, track_gaps(track, 3600))
+ Traceback (most recent call last):
+ ...
+ mapillary_tools.exceptions.MapillaryOutsideGPXTrackError: The GPX track has a gap of 2.5 hours, from 1970-01-01T00:00:20+00:00 at (0, 0) to 1970-01-01T02:30:00+00:00 at (0, 0), which the image time 1970-01-01T01:06:40+00:00 falls in
+ """
+ assert sorted_points, "must have at least one point"
+
+ if time < sorted_points[0].time:
+ delta = sorted_points[0].time - time
+ gpx_start_time = build_capture_time(sorted_points[0].time)
+ gpx_end_time = build_capture_time(sorted_points[-1].time)
+ # with the tolerance of 1ms
+ if 0.001 < delta:
+ raise exceptions.MapillaryOutsideGPXTrackError(
+ f"The image date time is {round(delta, 3)} seconds behind the GPX start point",
+ image_time=build_capture_time(time),
+ gpx_start_time=gpx_start_time,
+ gpx_end_time=gpx_end_time,
+ )
+
+ if sorted_points[-1].time < time:
+ delta = time - sorted_points[-1].time
+ gpx_start_time = build_capture_time(sorted_points[0].time)
+ gpx_end_time = build_capture_time(sorted_points[-1].time)
+ # with the tolerance of 1ms
+ if 0.001 < delta:
+ raise exceptions.MapillaryOutsideGPXTrackError(
+ f"The image time is {round(delta, 3)} seconds beyond the GPX end point",
+ image_time=build_capture_time(time),
+ gpx_start_time=gpx_start_time,
+ gpx_end_time=gpx_end_time,
+ )
+
+ # The last gap that starts at or before the image
+ idx = bisect.bisect_right([before.time for before, _ in gaps], time)
+ if idx:
+ before, after = gaps[idx - 1]
+ delta = min(time - before.time, after.time - time)
+ # with the tolerance of 1ms, as at the ends of the track
+ if 0.001 < delta:
+ captured_at = datetime.datetime.fromtimestamp(
+ time, tz=datetime.timezone.utc
+ ).isoformat()
+ track_gap = describe_gap(before, after, before.time, after.time)
+ raise exceptions.MapillaryOutsideGPXTrackError(
+ f"The GPX track has {track_gap}, which the image time {captured_at} falls in",
+ image_time=build_capture_time(time),
+ gpx_start_time=build_capture_time(sorted_points[0].time),
+ gpx_end_time=build_capture_time(sorted_points[-1].time),
+ )
+
+
class GeotagImagesFromGPX(GeotagImagesFromGeneric):
def __init__(
self,
@@ -55,55 +124,7 @@ def _interpolate_image_metadata_along(
sorted_points: T.Sequence[geo.Point],
gaps: T.Sequence[tuple[geo.Point, geo.Point]] = (),
) -> types.ImageMetadata:
- assert sorted_points, "must have at least one point"
-
- if image_metadata.time < sorted_points[0].time:
- delta = sorted_points[0].time - image_metadata.time
- gpx_start_time = build_capture_time(sorted_points[0].time)
- gpx_end_time = build_capture_time(sorted_points[-1].time)
- # with the tolerance of 1ms
- if 0.001 < delta:
- raise exceptions.MapillaryOutsideGPXTrackError(
- f"The image date time is {round(delta, 3)} seconds behind the GPX start point",
- image_time=build_capture_time(image_metadata.time),
- gpx_start_time=gpx_start_time,
- gpx_end_time=gpx_end_time,
- )
-
- if sorted_points[-1].time < image_metadata.time:
- delta = image_metadata.time - sorted_points[-1].time
- gpx_start_time = build_capture_time(sorted_points[0].time)
- gpx_end_time = build_capture_time(sorted_points[-1].time)
- # with the tolerance of 1ms
- if 0.001 < delta:
- raise exceptions.MapillaryOutsideGPXTrackError(
- f"The image time is {round(delta, 3)} seconds beyond the GPX end point",
- image_time=build_capture_time(image_metadata.time),
- gpx_start_time=gpx_start_time,
- gpx_end_time=gpx_end_time,
- )
-
- # The last gap that starts at or before the image
- idx = bisect.bisect_right(
- [before.time for before, _ in gaps], image_metadata.time
- )
- if idx:
- before, after = gaps[idx - 1]
- delta = min(
- image_metadata.time - before.time, after.time - image_metadata.time
- )
- # with the tolerance of 1ms, as at the ends of the track
- if 0.001 < delta:
- captured_at = datetime.datetime.fromtimestamp(
- image_metadata.time, tz=datetime.timezone.utc
- ).isoformat()
- track_gap = describe_gap(before, after, before.time, after.time)
- raise exceptions.MapillaryOutsideGPXTrackError(
- f"The GPX track has {track_gap}, which the image time {captured_at} falls in",
- image_time=build_capture_time(image_metadata.time),
- gpx_start_time=build_capture_time(sorted_points[0].time),
- gpx_end_time=build_capture_time(sorted_points[-1].time),
- )
+ check_within_track(image_metadata.time, sorted_points, gaps)
interpolated = geo.interpolate(sorted_points, image_metadata.time)
diff --git a/mapillary_tools/geotag/video_extractors/gpx.py b/mapillary_tools/geotag/video_extractors/gpx.py
index 3144a682..315bb9f9 100644
--- a/mapillary_tools/geotag/video_extractors/gpx.py
+++ b/mapillary_tools/geotag/video_extractors/gpx.py
@@ -204,7 +204,7 @@ def extract(self) -> types.VideoMetadata:
self._apply_offset_time(gpx_points, rate=rate)
self._check_coverage(gpx_points, gpx_unix_times, [], start_clock, rate)
if self.clip_to_video:
- gpx_points = self._clip_to_video(gpx_points, start_clock)
+ gpx_points = self._clip_to_video(gpx_points, start_clock, rate)
return types.VideoMetadata(
filename=self.video_path,
filesize=utils.get_file_size(self.video_path),
@@ -269,7 +269,7 @@ def extract(self) -> types.VideoMetadata:
rate,
)
if self.clip_to_video:
- gpx_points = self._clip_to_video(gpx_points, clock)
+ gpx_points = self._clip_to_video(gpx_points, clock, rate)
return dataclasses.replace(native_video_metadata, points=gpx_points)
@@ -313,7 +313,7 @@ def _parse_tracks(self) -> list[list[geo.Point]]:
) from ex
def _clip_to_video(
- self, gpx_points: list[geo.Point], clock: VideoClock | None
+ self, gpx_points: list[geo.Point], clock: VideoClock | None, rate: float
) -> list[geo.Point]:
"""
Clip a GPX track to the video, from video time 0 to where the video
@@ -330,10 +330,20 @@ def _clip_to_video(
past, so that the first frames keep a position. Its end is left as it
is: it may be that of a timelapse, which runs on past the end of the
video at the pace of the GPS clock rather than that of the video.
+
+ A gap of more than constants.MAX_GPX_GAP_SECONDS that video time 0, or
+ the end of the video at **rate** GPS seconds per second, falls in says
+ nothing of where the video was: the fixes on its far side are left
+ out, with no point interpolated across it, as if the track started
+ (or ended) there. One inside the video fails it (see _check_coverage).
"""
+ video_duration = _video_duration(self.video_path)
+ gpx_points = _leave_out_far_side_of_gaps(
+ gpx_points, constants.MAX_GPX_GAP_SECONDS / rate, video_duration
+ )
if clock is None:
return _clip_track(gpx_points, None)
- return _clip_track(gpx_points, _video_duration(self.video_path))
+ return _clip_track(gpx_points, video_duration)
def _place_without_video_gps(
self, gpx_points: T.Sequence[geo.Point]
@@ -446,7 +456,9 @@ def _check_coverage(
warns, and one that covers all of it is silent. The video can fall
between the ends of the track and still miss it, or part of it, in a
gap of more than constants.MAX_GPX_GAP_SECONDS between two of its
- points.
+ points. In process, which uploads the track with the video, a gap
+ inside the video raises too: sampling leaves out the frames in it,
+ while the track would be interpolated across it.
Without the duration of the video, the video is known only up to its
last GPS point, and a track that starts after it may still overlap the
@@ -506,6 +518,66 @@ def _check_coverage(
# In GPS seconds, which is where clock and time zone mistakes are made
gap = min(max(first - video_end, -last) for first, last in parts) * rate
+ if clock is None:
+ # Placed at the start of the video, only the offset moved it off
+ advice = "Check --interpolation_offset_time"
+ elif offset_time:
+ advice = (
+ f"Check --interpolation_offset_time, that the {self.file_format} file belongs to "
+ f"this video, then the camera clock and the time zone of the {self.file_format} "
+ "timestamps"
+ )
+ else:
+ advice = (
+ f"Check that the {self.file_format} file belongs to this video, then the camera "
+ f"clock and the time zone of the {self.file_format} timestamps"
+ )
+
+ if gap <= 0 and self.clip_to_video:
+ # process uploads the track with the video, and its frames would be
+ # placed along it across a gap that the video spans, which says
+ # nothing of where the video was. Sampling leaves out the frames
+ # in it (see _clip_to_video for the gaps at its ends)
+ for before, after in gap_points:
+ if before.time <= 0 and video_end <= after.time:
+ # The track touches the video at an instant at one end only
+ where = f"which {video} falls in{offset_clause}"
+ if clock is not None or offset_time:
+ where_advice = advice
+ elif self.sync_mode is SyncMode.REBASE:
+ # Said to start with the video
+ where_advice = f"Trim the {self.file_format} to the video"
+ else:
+ # Assumed to start with the video
+ where_advice = (
+ f"Trim the {self.file_format} to the video, or sample the video "
+ "with video_process and --video_start_time"
+ )
+ elif 0 <= before.time and after.time <= video_end:
+ where = (
+ f"inside {video}{offset_clause}, which says nothing of where "
+ "the video was in it"
+ )
+ where_advice = (
+ "Sample the video with video_process and --video_geotag_source "
+ f"{self.file_format.lower()}, which leaves out the frames in the gap"
+ )
+ else:
+ continue
+ # At the Unix times the points were recorded at
+ track_gap = describe_gap(
+ before,
+ after,
+ lookup_start + before.time * rate,
+ lookup_start + after.time * rate,
+ )
+ raise exceptions.MapillaryOutsideGPXTrackError(
+ f"{gpx_track} has {track_gap}, {where}. {where_advice}",
+ image_time=build_capture_time(lookup_start),
+ gpx_start_time=build_capture_time(gpx_unix_times[0]),
+ gpx_end_time=build_capture_time(gpx_unix_times[1]),
+ )
+
if gap <= 0:
# In video seconds
uncovered = (
@@ -523,20 +595,6 @@ def _check_coverage(
)
return
- if clock is None:
- # Placed at the start of the video, only the offset moved it off
- advice = "Check --interpolation_offset_time"
- elif offset_time:
- advice = (
- f"Check --interpolation_offset_time, that the {self.file_format} file belongs to "
- f"this video, then the camera clock and the time zone of the {self.file_format} "
- "timestamps"
- )
- else:
- advice = (
- f"Check that the {self.file_format} file belongs to this video, then the camera "
- f"clock and the time zone of the {self.file_format} timestamps"
- )
# Between the ends of the track, the video falls in one of its gaps
surrounding_gap = next(
(
@@ -1027,6 +1085,33 @@ def describe_gap(
)
+def _leave_out_far_side_of_gaps(
+ points: T.Sequence[geo.Point], max_gap: float, end: float | None
+) -> list[geo.Point]:
+ """
+ Leave out of a track, sorted by time, the points on the far side of a gap
+ of more than **max_gap** seconds that time 0, or **end** (None if
+ unknown), falls in
+
+ >>> track = [geo.Point(time=t, lat=t, lon=0, alt=None, angle=None) for t in (-9000, 99, 200, 9000)]
+ >>> [p.time for p in _leave_out_far_side_of_gaps(track, 3600, 300)]
+ [99, 200]
+ >>> [p.time for p in _leave_out_far_side_of_gaps(track, 3600, None)]
+ [99, 200, 9000]
+ >>> [p.time for p in _leave_out_far_side_of_gaps(track, 10000, 300)]
+ [-9000, 99, 200, 9000]
+ """
+ start, stop = 0, len(points)
+ for idx, (before, after) in enumerate(geo.pairwise(points)):
+ if after.time - before.time <= max_gap:
+ continue
+ if before.time < 0 < after.time:
+ start = idx + 1
+ if end is not None and before.time < end < after.time:
+ stop = idx + 1
+ return list(points[start:stop])
+
+
def _clip_track(
points: T.Sequence[geo.Point], duration: float | None
) -> list[geo.Point]:
diff --git a/mapillary_tools/process_geotag_properties.py b/mapillary_tools/process_geotag_properties.py
index 61ef91fb..68621c23 100644
--- a/mapillary_tools/process_geotag_properties.py
+++ b/mapillary_tools/process_geotag_properties.py
@@ -62,17 +62,21 @@ def process_geotag_properties(
raise exceptions.MapillaryFileNotFoundError(
f"Import file or directory not found: {path}"
)
- # A directory is searched for the files it can process, but a file
- # named explicitly is skipped only with a word. It is not failed:
- # callers that pass the files a user selected find the skipped ones in
- # the process report. A zip file is uploaded as it is, by
- # process_and_upload too
- if path.is_file() and not (
+
+ # A directory is searched for the files it can process, and the others in
+ # it are left alone, but a file named explicitly is failed if it cannot be
+ # processed: it was meant to be uploaded. A zip file is uploaded as it is,
+ # by process_and_upload
+ unsupported_metadatas = [
+ _describe_unsupported_file(path)
+ for path in import_paths
+ if path.is_file()
+ and not (
utils.is_image_file(path)
or utils.is_video_file(path)
or path.suffix.lower() == ".zip"
- ):
- LOG.warning("Skipping %s, which is not a supported image or video", path)
+ )
+ ]
if geotag_source_path is None:
geotag_source_path = video_import_path
@@ -110,7 +114,30 @@ def process_geotag_properties(
metadata_or_errors = process(image_paths + video_paths, options, filetypes)
- return metadata_or_errors
+ return metadata_or_errors + unsupported_metadatas
+
+
+def _describe_unsupported_file(path: Path) -> types.ErrorMetadata:
+ """
+ The error of a file named for processing that is neither an image nor a
+ video that can be processed. An Insta360 original (.insv) is stitched first
+
+ >>> metadata = _describe_unsupported_file(Path("notes.txt"))
+ >>> str(metadata.error)
+ 'Not a supported image or video'
+ >>> metadata.filetype is None
+ True
+ """
+ message = "Not a supported image or video"
+ if path.suffix.lower() == ".insv":
+ message += (
+ ". Stitch the Insta360 original with Insta360 Studio first, then"
+ " geotag the stitched video by the original with"
+ """ --video_geotag_source '{"source": "exiftool_runtime", "pattern": "../%g.insv"}'"""
+ )
+ return types.describe_error_metadata(
+ exceptions.MapillaryUnsupportedFileError(message), path, filetype=None
+ )
def _apply_offsets(
@@ -215,8 +242,10 @@ def _show_stats(
) -> None:
LOG.info("==> Process summary")
- metadatas_by_filetype: dict[types.FileType, list[types.MetadataOrError]] = {}
+ # None for the files that are neither images nor videos
+ metadatas_by_filetype: dict[types.FileType | None, list[types.MetadataOrError]] = {}
for metadata in metadatas:
+ filetype: types.FileType | None
if isinstance(metadata, types.ImageMetadata):
filetype = types.FileType.IMAGE
else:
@@ -240,13 +269,15 @@ def _show_stats(
def _show_stats_per_filetype(
metadatas: T.Collection[types.MetadataOrError],
- filetype: types.FileType,
+ filetype: types.FileType | None,
skipped_process_errors: set[T.Type[Exception]],
):
good_metadatas: list[types.Metadata]
good_metadatas, error_metadatas = types.separate_errors(metadatas)
- LOG.info(f"{len(metadatas)} {filetype.value} read in total")
+ LOG.info(
+ f"{len(metadatas)} {'other file' if filetype is None else filetype.value} read in total"
+ )
if good_metadatas:
total_filesize = sum(
[0 if m.filesize is None else m.filesize for m in good_metadatas]
diff --git a/mapillary_tools/sample_video.py b/mapillary_tools/sample_video.py
index b87525e7..d4fa05a2 100644
--- a/mapillary_tools/sample_video.py
+++ b/mapillary_tools/sample_video.py
@@ -29,7 +29,8 @@
utils,
)
from .exif_write import ExifEdit
-from .geotag import factory
+from .geotag import factory, geotag_images_from_gpx_file, geotag_images_from_nmea_file
+from .geotag.geotag_images_from_gpx import check_within_track
from .geotag.options import (
InterpolationOption,
SourceOption,
@@ -37,7 +38,7 @@
SourceType,
VideoSyncOption,
)
-from .geotag.video_extractors.gpx import GPXVideoExtractor, VideoClock
+from .geotag.video_extractors.gpx import GPXVideoExtractor, track_gaps, VideoClock
from .geotag.video_extractors.native import NativeVideoExtractor
from .geotag.video_extractors.nmea import NMEAVideoExtractor
from .mp4 import mp4_sample_parser, simple_mp4_parser as sparser
@@ -166,6 +167,47 @@ def _parse_video_sources(
return video_options
+def _parse_sample_sources(
+ geotag_source: list[str] | None,
+ geotag_source_path: Path | None,
+ video_import_path: Path,
+ interpolation: InterpolationOption,
+) -> list[SourceOption] | None:
+ """
+ The sources that the geotagging stage of video_process geotags the samples
+ by, as process_geotag_properties() picks them: those of --geotag_source,
+ or, without it, the default sources, which read the position that sampling
+ writes into a sample (an empty list here). None for the sample_video
+ command, which cannot tell how its samples will be geotagged
+ """
+ if geotag_source is None:
+ return None
+
+ if not geotag_source:
+ return []
+
+ options = factory.parse_source_options(
+ geotag_source=geotag_source,
+ video_geotag_source=[],
+ geotag_source_path=(
+ video_import_path if geotag_source_path is None else geotag_source_path
+ ),
+ )
+ for option in options:
+ if option.interpolation is None:
+ option.interpolation = interpolation
+ return options
+
+
+# Sources that geotag a sample by the time since the first sample of its video,
+# at which they place the start of the video: the legacy sources and
+# exiftool_xml, which geotag samples by the GPS of the videos they were taken
+# from (see GeotagImagesFromVideo)
+_FIRST_SAMPLE_SOURCES = frozenset(
+ {SourceType.GOPRO, SourceType.BLACKVUE, SourceType.CAMM, SourceType.EXIFTOOL_XML}
+)
+
+
def sample_video(
video_import_path: Path,
import_path: Path,
@@ -217,16 +259,22 @@ def sample_video(
duration_ratio=video_duration_ratio,
)
+ interpolation = InterpolationOption(
+ offset_time=interpolation_offset_time,
+ use_gpx_start_time=interpolation_use_gpx_start_time,
+ )
geotag_options = _parse_geotag_options(
geotag_source,
geotag_source_path,
video_geotag_source,
video_import_path,
video_sync,
- InterpolationOption(
- offset_time=interpolation_offset_time,
- use_gpx_start_time=interpolation_use_gpx_start_time,
- ),
+ interpolation,
+ )
+ # The sources that the geotagging stage of video_process geotags the
+ # samples by, which tell which interval samples may be left out
+ sample_sources = _parse_sample_sources(
+ geotag_source, geotag_source_path, video_import_path, interpolation
)
if rerun:
@@ -286,6 +334,8 @@ def sample_video(
start_time=video_start_time_dt,
geotag_options=geotag_options,
video_sync=video_sync,
+ sample_sources=sample_sources,
+ single_video=len(video_list) == 1,
)
except ffmpeglib.FFmpegNotFoundError as ex:
# fatal error
@@ -765,6 +815,10 @@ def _sample_single_video_by_interval(
start_time: datetime.datetime | None = None,
geotag_options: T.Sequence[SourceOption] = (),
video_sync: VideoSyncOption | None = None,
+ # How the samples are geotagged (see _parse_sample_sources())
+ sample_sources: T.Sequence[SourceOption] | None = None,
+ # Whether the video is the only one sampled
+ single_video: bool = True,
) -> None:
ffmpeg = ffmpeglib.FFMPEG(constants.FFMPEG_PATH, constants.FFPROBE_PATH)
@@ -812,6 +866,11 @@ def _sample_single_video_by_interval(
# Located along the video's own GPS track, as distance samples are
track = _read_track_for_interval_samples(video_path, geotag_options)
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
+ # capture times
+ file_track = (
+ _read_file_track_for_interval_samples(sample_sources) if track is None else None
+ )
stream_specifier = _interval_stream_specifier(probe)
# The stream sampled, whether it is the only one or not
@@ -830,6 +889,9 @@ def _sample_single_video_by_interval(
frame_samples = ffmpeglib.FFMPEG.sort_selected_samples(
wip_dir, video_path, selected_stream_specifiers=[stream_specifier]
)
+
+ # (path, video time, capture time) of each sample
+ samples: list[tuple[Path, float, datetime.datetime]] = []
for frame_idx_1based, sample_paths in frame_samples:
assert len(sample_paths) == 1
if sample_paths[0] is None:
@@ -837,18 +899,216 @@ def _sample_single_video_by_interval(
# extract_frames() produces 1-based frame indices so we need to subtract 1 here
video_time = (frame_idx_1based - 1) * sample_interval
timestamp = start_time + datetime.timedelta(seconds=video_time * rate)
- exif_edit = ExifEdit(sample_paths[0])
+ samples.append((sample_paths[0], video_time, timestamp))
+
+ if track is not None:
+ covered = [
+ _within_track_time_range_buffered(track.points, video_time)
+ for _, video_time, _ in samples
+ ]
+ left_out = _samples_to_leave_out(
+ covered,
+ keep_leading=_samples_placed_by_the_first(sample_sources),
+ )
+ track_range = (
+ start_time + datetime.timedelta(seconds=track.points[0].time * rate),
+ start_time + datetime.timedelta(seconds=track.points[-1].time * rate),
+ )
+ elif file_track is not None:
+ covered = file_track.covers(
+ [timestamp.timestamp() for _, _, timestamp in samples]
+ )
+ left_out = (
+ set()
+ if file_track.placed_by_the_first and not single_video
+ else _samples_to_leave_out(
+ covered, keep_leading=file_track.placed_by_the_first
+ )
+ )
+ track_range = (
+ datetime.datetime.fromtimestamp(
+ file_track.points[0].time, tz=datetime.timezone.utc
+ ),
+ datetime.datetime.fromtimestamp(
+ file_track.points[-1].time, tz=datetime.timezone.utc
+ ),
+ )
+ else:
+ covered = [False] * len(samples)
+ left_out = set()
+
+ if left_out:
+ LOG.warning(
+ "%d of the %d samples of %s %s outside its GPS track (%s to %s) and %s left out",
+ len(left_out),
+ len(samples),
+ video_path.name,
+ "falls" if len(left_out) == 1 else "fall",
+ track_range[0].isoformat(),
+ track_range[1].isoformat(),
+ "was" if len(left_out) == 1 else "were",
+ )
+
+ for idx, (sample_path, video_time, timestamp) in enumerate(samples):
+ if idx in left_out:
+ sample_path.unlink()
+ continue
+ exif_edit = ExifEdit(sample_path)
exif_edit.add_date_time_original(timestamp)
exif_edit.add_gps_datetime(timestamp)
- if (
- track is not None
- and interpolator is not None
- and _within_track_time_range_buffered(track.points, video_time)
- ):
+ if track is not None and interpolator is not None and covered[idx]:
_add_position(exif_edit, track, interpolator.interpolate(video_time))
exif_edit.write()
+def _samples_to_leave_out(covered: T.Sequence[bool], keep_leading: bool) -> set[int]:
+ """
+ The indices of the interval samples to leave out, given which of them the
+ track that geotags them covers: those it does not, so that they do not
+ fail geotagging. If it covers none of them, none, so that the samples
+ fail rather than leave the video out without a word. With
+ **keep_leading**, not those before the first sample it covers either:
+ the geotagging stage places the samples by the time since the first one
+ (see _samples_placed_by_the_first())
+
+ >>> sorted(_samples_to_leave_out([False, True, True, False], keep_leading=False))
+ [0, 3]
+ >>> sorted(_samples_to_leave_out([False, True, False, True, False], keep_leading=True))
+ [2, 4]
+ >>> _samples_to_leave_out([False, False], keep_leading=False)
+ set()
+ """
+ if not any(covered):
+ return set()
+ first = covered.index(True)
+ return {
+ idx
+ for idx, is_covered in enumerate(covered)
+ if not is_covered and (first < idx or not keep_leading)
+ }
+
+
+def _samples_placed_by_the_first(
+ sample_sources: T.Sequence[SourceOption] | None,
+) -> bool:
+ """
+ Whether the geotagging stage may place the samples of a video by the time
+ since its first sample, so that leaving out the samples before the track
+ starts would misplace the others: the sources of _FIRST_SAMPLE_SOURCES,
+ and a GPX (or NMEA) track that --interpolation_use_gpx_start_time starts
+ at the first sample. The sample_video command cannot tell which sources
+ will geotag its samples
+
+ >>> _samples_placed_by_the_first(None)
+ True
+ >>> _samples_placed_by_the_first([])
+ False
+ >>> _samples_placed_by_the_first([SourceOption(SourceType.GOPRO)])
+ True
+ >>> _samples_placed_by_the_first([SourceOption(SourceType.EXIF)])
+ False
+ """
+ if sample_sources is None:
+ return True
+ return any(
+ option.source in _FIRST_SAMPLE_SOURCES
+ or (
+ option.source in factory.EXTERNAL_GPS_SOURCES
+ and option.interpolation is not None
+ and option.interpolation.use_gpx_start_time
+ )
+ for option in sample_sources
+ )
+
+
+@dataclasses.dataclass(frozen=True)
+class _FileTrack:
+ """
+ A GPX (or NMEA) track that the geotagging stage locates samples along by
+ their capture times, as GeotagImagesFromGPX locates images
+ """
+
+ # Sorted by time
+ points: T.Sequence[geo.Point]
+ gaps: T.Sequence[tuple[geo.Point, geo.Point]]
+ # --interpolation_offset_time
+ offset_time: float
+ # --interpolation_use_gpx_start_time: the track starts at the first sample
+ placed_by_the_first: bool
+
+ def covers(self, capture_times: T.Sequence[float]) -> list[bool]:
+ """
+ Whether the track covers each sample of a video, given their capture
+ times, in order
+ """
+ offset = self.offset_time
+ if self.placed_by_the_first and capture_times:
+ offset += self.points[0].time - capture_times[0]
+
+ covered = []
+ for capture_time in capture_times:
+ try:
+ check_within_track(capture_time + offset, self.points, self.gaps)
+ except exceptions.MapillaryOutsideGPXTrackError:
+ covered.append(False)
+ else:
+ covered.append(True)
+ return covered
+
+
+def _read_file_track_for_interval_samples(
+ sample_sources: T.Sequence[SourceOption] | None,
+) -> _FileTrack | None:
+ """
+ The GPX (or NMEA) track that the geotagging stage locates the interval
+ samples along by their capture times: that of the first such source of
+ --geotag_source, which geotags images. None if there is none, if it cannot
+ be read (the geotagging stage reports why), or if another source may
+ geotag the samples it misses: one that geotags them by the GPS of their
+ video, or another track file
+ """
+ if not sample_sources:
+ return None
+
+ track_sources = [
+ option
+ for option in sample_sources
+ if option.source in factory.EXTERNAL_GPS_SOURCES
+ and (option.filetypes is None or types.FileType.IMAGE in option.filetypes)
+ ]
+ if len(track_sources) != 1 or any(
+ option.source in _FIRST_SAMPLE_SOURCES for option in sample_sources
+ ):
+ return None
+
+ option = track_sources[0]
+ if option.source_path is None or option.source_path.source_path is None:
+ return None
+
+ try:
+ if option.source is SourceType.GPX:
+ points = geotag_images_from_gpx_file._read_gpx_points(
+ option.source_path.source_path
+ )
+ else:
+ points = geotag_images_from_nmea_file._read_nmea_points(
+ option.source_path.source_path
+ )
+ except exceptions.MapillaryInvalidGPXError:
+ return None
+
+ sorted_points = sorted(points, key=lambda point: point.time)
+ interpolation = option.interpolation
+ return _FileTrack(
+ points=sorted_points,
+ gaps=track_gaps(sorted_points, constants.MAX_GPX_GAP_SECONDS),
+ offset_time=0.0 if interpolation is None else interpolation.offset_time,
+ placed_by_the_first=(
+ interpolation is not None and interpolation.use_gpx_start_time
+ ),
+ )
+
+
def _read_track_for_interval_samples(
video_path: Path, geotag_options: T.Sequence[SourceOption]
) -> types.VideoMetadata | None:
diff --git a/mapillary_tools/types.py b/mapillary_tools/types.py
index a57d5481..e3905957 100644
--- a/mapillary_tools/types.py
+++ b/mapillary_tools/types.py
@@ -87,7 +87,8 @@ def update_md5sum(self) -> None:
@dataclasses.dataclass
class ErrorMetadata:
filename: Path
- filetype: FileType
+ # None for a file that is neither an image nor a video
+ filetype: FileType | None
error: Exception
@@ -175,7 +176,7 @@ def separate_errors(
def describe_error_metadata(
- exc: Exception, filename: Path, filetype: FileType
+ exc: Exception, filename: Path, filetype: FileType | None
) -> ErrorMetadata:
return ErrorMetadata(filename=filename, filetype=filetype, error=exc)
diff --git a/tests/integration/test_process.py b/tests/integration/test_process.py
index b16b982d..7f6cf712 100644
--- a/tests/integration/test_process.py
+++ b/tests/integration/test_process.py
@@ -1311,6 +1311,77 @@ def test_process_geotag_with_exiftool_xml_pattern_fallback(
assert descs[0]["error"]["type"] == "MapillaryVideoGPSNotFoundError"
+def _write_track_parts(gpx_file: py.path.local, parts) -> None:
+ """A GPX track of parts, each (start, seconds, lat) with a point a second"""
+ points = "".join(
+ f''
+ f""
+ for start, seconds, lat in parts
+ for i in range(seconds + 1)
+ )
+ gpx_file.write(
+ '\n'
+ ''
+ f"{points}\n"
+ )
+
+
+def test_process_video_track_is_not_interpolated_across_a_gpx_gap(
+ setup_data: py.path.local,
+):
+ """
+ A video that starts in a gap of more than an hour of the GPX track takes
+ the track from the first point past the gap: a point interpolated across
+ the gap, at the start of the video, was uploaded with it
+ """
+ video_path = setup_data.join("gopro_data").join("max-360mode.mp4")
+ # The GPS clock of the video starts at 23:44:43.613, and the track
+ # resumes 3.387 seconds later, 2 hours after its previous point 5 km away
+ resumed = datetime.datetime(2019, 11, 18, 23, 44, 47, tzinfo=datetime.timezone.utc)
+ gpx_file = setup_data.join("track.gpx")
+ _write_track_parts(
+ gpx_file,
+ [(resumed - datetime.timedelta(hours=2), 0, 37.045), (resumed, 20, 37.0)],
+ )
+
+ [desc] = run_process_for_descs(
+ [
+ *["--geotag_source", "gpx"],
+ *["--geotag_source_path", str(gpx_file)],
+ str(video_path),
+ ]
+ )
+
+ track = desc["MAPGPSTrack"]
+ assert track[0][0] == pytest.approx(3387, abs=1)
+ assert track[0][2] == pytest.approx(37.0)
+
+
+def test_process_video_with_a_gpx_gap_inside_it_fails(setup_data: py.path.local):
+ """process would interpolate the track across it, which sampling leaves out"""
+ video_path = setup_data.join("gopro_data").join("max-360mode.mp4")
+ start = datetime.datetime(2019, 11, 18, 23, 44, 40, tzinfo=datetime.timezone.utc)
+ gpx_file = setup_data.join("track.gpx")
+ # No point from 4.4 to 9.4 seconds into the video
+ _write_track_parts(
+ gpx_file,
+ [(start, 8, 37.0), (start + datetime.timedelta(seconds=13), 10, 37.001)],
+ )
+
+ [desc] = run_process_for_descs(
+ [
+ *["--geotag_source", "gpx"],
+ *["--geotag_source_path", str(gpx_file)],
+ str(video_path),
+ ],
+ env={**os.environ, "MAPILLARY_TOOLS_MAX_GPX_GAP_SECONDS": "3"},
+ )
+
+ assert desc["error"]["type"] == "MapillaryOutsideGPXTrackError"
+ assert "has a gap of 5 seconds" in desc["error"]["message"]
+ assert "Sample the video with video_process" in desc["error"]["message"]
+
+
def test_process_video_with_a_gpx_far_longer_than_it_warns(
setup_data: py.path.local,
):
diff --git a/tests/integration/test_process_and_upload.py b/tests/integration/test_process_and_upload.py
index cb2347c1..ec7f61ad 100644
--- a/tests/integration/test_process_and_upload.py
+++ b/tests/integration/test_process_and_upload.py
@@ -14,6 +14,7 @@
assert_same_image_descs,
extract_all_uploaded_descs,
pytest_skip_if_not_ffmpeg_installed,
+ run_command,
run_process_and_upload_for_descs,
setup_config,
setup_data,
@@ -312,3 +313,28 @@ def test_video_process_and_upload_after_gpx(
uploaded_descs = sum(extract_all_uploaded_descs(Path(setup_upload)), [])
assert_same_image_descs(uploaded_descs, [])
+
+
+@pytest.mark.usefixtures("setup_config")
+def test_process_and_upload_named_zip_files(
+ tmpdir: py.path.local, setup_data: py.path.local, setup_upload: py.path.local
+):
+ """
+ A zip file named for upload is uploaded as it is, and a named file that is
+ neither an image nor a video gets an error of its own
+ """
+ run_command(["--file_types=image", str(setup_data)], command="process")
+ zip_dir = tmpdir.mkdir("zip_dir")
+ run_command([str(setup_data), str(zip_dir)], command="zip")
+ zip_paths = sorted(str(path) for path in zip_dir.listdir())
+ assert zip_paths
+ notes = tmpdir.join("notes.txt")
+ notes.write_binary(b"not an image")
+
+ descs = run_process_and_upload_for_descs([*zip_paths, str(notes)])
+
+ assert [(desc["filename"], desc["error"]["type"]) for desc in descs] == [
+ (str(Path(notes).resolve()), "MapillaryUnsupportedFileError")
+ ]
+ uploaded = extract_all_uploaded_descs(Path(setup_upload))
+ assert len(uploaded) == len(zip_paths)
diff --git a/tests/integration/test_process_report.py b/tests/integration/test_process_report.py
index 1a115168..1a77899a 100644
--- a/tests/integration/test_process_report.py
+++ b/tests/integration/test_process_report.py
@@ -4,9 +4,11 @@
# LICENSE file in the root directory of this source tree.
import json
+import subprocess
from pathlib import Path
import py.path
+import pytest
from .fixtures import run_command
@@ -42,9 +44,16 @@ def test_process_report_counts_recognized_errors_and_explicit_unsupported(
descs, report = _run_process_with_report([broken_image, unsupported], root)
- assert len(descs) == 1
- assert descs[0]["filename"] == str(broken_image.resolve())
- assert "error" in descs[0]
+ descs_by_filename = {desc["filename"]: desc for desc in descs}
+ assert set(descs_by_filename) == {
+ str(broken_image.resolve()),
+ str(unsupported.resolve()),
+ }
+ assert "error" in descs_by_filename[str(broken_image.resolve())]
+ assert descs_by_filename[str(unsupported.resolve())]["error"] == {
+ "type": "MapillaryUnsupportedFileError",
+ "message": "Not a supported image or video",
+ }
assert report == {
"schema_version": 1,
"discovered_file_count": 2,
@@ -86,7 +95,9 @@ def test_process_report_filters_folder_sidecars_but_reports_explicit_sidecars(
explicit_output = root / "explicit"
explicit_output.mkdir()
descs, explicit_report = _run_process_with_report([sidecar], explicit_output)
- assert descs == []
+ assert [desc["error"]["type"] for desc in descs] == [
+ "MapillaryUnsupportedFileError"
+ ]
assert explicit_report["discovered_file_count"] == 1
assert explicit_report["processed_file_count"] == 0
assert explicit_report["skipped_file_count"] == 1
@@ -94,7 +105,9 @@ def test_process_report_filters_folder_sidecars_but_reports_explicit_sidecars(
assert explicit_report["skipped_files"][0]["details"] == {"extension": ".gpx"}
-def test_process_without_report_preserves_unsupported_omission(tmpdir: py.path.local):
+def test_process_without_report_fails_an_explicit_unsupported_file(
+ tmpdir: py.path.local,
+):
root = Path(str(tmpdir))
unsupported = root / "unsupported.webp"
desc_path = root / "description.json"
@@ -110,26 +123,42 @@ def test_process_without_report_preserves_unsupported_omission(tmpdir: py.path.l
command="process",
)
- assert json.loads(desc_path.read_text(encoding="utf-8")) == []
+ assert json.loads(desc_path.read_text(encoding="utf-8")) == [
+ {
+ "error": {
+ "type": "MapillaryUnsupportedFileError",
+ "message": "Not a supported image or video",
+ },
+ "filename": str(unsupported.resolve()),
+ }
+ ]
assert not (root / "process-report.json").exists()
-def test_process_warns_of_an_explicit_unsupported_file(tmpdir: py.path.local):
- """It was skipped without a word"""
+def test_process_fails_an_explicit_unsupported_file(tmpdir: py.path.local):
+ """It was left out with only a warning, and the run succeeded"""
root = Path(str(tmpdir))
unsupported = root / "unsupported.webp"
+ unsupported_in_folder = root / "folder" / "unsupported.webp"
desc_path = root / "description.json"
unsupported.write_bytes(b"not supported")
-
- result = run_command(
- ["--desc_path", str(desc_path), str(unsupported)],
- command="process",
- capture_output=True,
- text=True,
- )
-
- assert json.loads(desc_path.read_text(encoding="utf-8")) == []
- assert (
- f"Skipping {unsupported}, which is not a supported image or video"
- in result.stderr
- )
+ unsupported_in_folder.parent.mkdir()
+ unsupported_in_folder.write_bytes(b"not supported")
+
+ with pytest.raises(subprocess.CalledProcessError) as exc_info:
+ run_command(
+ [
+ "--desc_path",
+ str(desc_path),
+ str(unsupported),
+ str(unsupported_in_folder.parent),
+ ],
+ command="process",
+ )
+
+ assert exc_info.value.returncode == 6
+ # Only the file named explicitly: those found in a folder are left alone
+ assert [
+ (desc["filename"], desc["error"]["type"])
+ for desc in json.loads(desc_path.read_text(encoding="utf-8"))
+ ] == [(str(unsupported.resolve()), "MapillaryUnsupportedFileError")]
diff --git a/tests/integration/test_video_process.py b/tests/integration/test_video_process.py
index 398ecb2c..b7cf5790 100644
--- a/tests/integration/test_video_process.py
+++ b/tests/integration/test_video_process.py
@@ -223,6 +223,48 @@ def test_video_process_by_interval_with_the_default_geotag_sources(
assert desc["MAPLongitude"] == pytest.approx(legacy["MAPLongitude"], abs=1e-7)
+def test_video_process_by_interval_leaves_out_the_samples_past_the_gpx(
+ setup_data: py.path.local,
+):
+ """
+ The GPX ends 2.6 seconds into the video. The samples after that would fail
+ geotagging, and fail the command: they are left out, with a warning
+ """
+ pytest_skip_if_not_ffmpeg_installed()
+
+ video_path = setup_data.join("videos").join("sample-5s.mp4")
+ gpx_file = setup_data.join("gpx").join("sf_30km_h.gpx")
+
+ with tempfile.TemporaryDirectory() as dir:
+ 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(video_path),
+ str(Path(dir, "samples")),
+ ],
+ capture_output=True,
+ text=True,
+ )
+ with open(Path(dir, "desc.json")) as fp:
+ descs = json.load(fp)
+ samples = sorted(p.name for p in Path(dir, "samples").rglob("*.jpg"))
+
+ assert all("error" not in desc for desc in descs), descs
+ assert sorted(desc["MAPCaptureTime"] for desc in descs) == [
+ "2025_03_14_07_01_31_000",
+ "2025_03_14_07_01_32_000",
+ "2025_03_14_07_01_33_000",
+ ]
+ assert len(samples) == 3, samples
+ assert "samples of sample-5s.mp4 fall outside its GPS track" in process.stderr
+ assert "and were left out" in process.stderr
+
+
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 b6e1f9b6..12500149 100644
--- a/tests/unit/test_gpx_video_coverage.py
+++ b/tests/unit/test_gpx_video_coverage.py
@@ -729,3 +729,155 @@ def test_an_image_on_a_point_by_a_gap_is_on_the_track(self, tmp_path):
tmp_path, [(self.IMAGE_TIME - 60, 60), (self.IMAGE_TIME + 7200, 60)]
)
assert isinstance(metadata, types.ImageMetadata)
+
+
+class TestGapsInTheTrackUploadedWithTheVideo:
+ """
+ process uploads the track with the video, and the frames are placed along
+ it. Nothing can be interpolated across a gap of more than an hour: a video
+ that starts or ends in one takes the track from the far side of it no
+ further, and one with a gap inside fails, as sampling leaves out the frames
+ in the gap.
+ """
+
+ def _extract(
+ self, tmp_path: Path, parts: T.Sequence[tuple[float, float]], **kwargs
+ ) -> types.VideoMetadata:
+ video_path = _write_camm_mp4(tmp_path / "video.mp4", _video_track())
+ gpx_path = _write_gpx_parts(tmp_path / "track.gpx", parts)
+ return GPXVideoExtractor(
+ video_path, gpx_path, clip_to_video=True, **kwargs
+ ).extract()
+
+ def test_a_start_in_a_gap_starts_the_track_past_it(self, tmp_path, caplog):
+ """The trip before ended 2 hours before the video, and the next one starts 5 seconds into it"""
+ with caplog.at_level(logging.WARNING):
+ metadata = self._extract(
+ tmp_path, [(A_UNIX_TIME - 7260, 60), (A_UNIX_TIME + 5, 60)]
+ )
+ times = [p.time for p in metadata.points]
+ # No point at video time 0, interpolated across the gap
+ assert times[0] == pytest.approx(5.0)
+ assert times[-1] == pytest.approx(VIDEO_DURATION)
+ [warning] = _warnings(caplog)
+ assert "5 seconds of the video fall outside the track" in warning
+
+ def test_an_end_in_a_gap_ends_the_track_before_it(self, tmp_path, caplog):
+ """The trip ends 6 seconds into the video, and the next one starts 2 hours later"""
+ with caplog.at_level(logging.WARNING):
+ metadata = self._extract(
+ tmp_path, [(A_UNIX_TIME - 60, 66), (A_UNIX_TIME + 7200, 60)]
+ )
+ times = [p.time for p in metadata.points]
+ assert times[0] == 0
+ # No point at the end of the video, interpolated across the gap
+ assert times[-1] == pytest.approx(6.0)
+ [warning] = _warnings(caplog)
+ assert "6 seconds of the video fall outside the track" in warning
+
+ def test_a_gap_inside_the_video_fails_it(self, tmp_path, monkeypatch):
+ monkeypatch.setattr(constants, "MAX_GPX_GAP_SECONDS", 3)
+ with pytest.raises(exceptions.MapillaryOutsideGPXTrackError) as info:
+ self._extract(tmp_path, [(A_UNIX_TIME - 10, 13), (A_UNIX_TIME + 7, 30)])
+ message = str(info.value)
+ assert (
+ "has a gap of 4 seconds, from 2026-08-12T08:26:30.979846+00:00 at (37.0, 14.0) "
+ "to 2026-08-12T08:26:34.979846+00:00 at (37.0, 14.0), inside the video"
+ ) in message
+ assert message.endswith(
+ "Sample the video with video_process and --video_geotag_source gpx, which leaves out the frames in the gap"
+ )
+ assert info.value.image_time == "2026_08_12_08_26_27_979"
+
+ def test_sampling_leaves_the_gap_inside_the_video_to_the_frames(
+ self, tmp_path, monkeypatch, caplog
+ ):
+ monkeypatch.setattr(constants, "MAX_GPX_GAP_SECONDS", 3)
+ video_path = _write_camm_mp4(tmp_path / "video.mp4", _video_track())
+ gpx_path = _write_gpx_parts(
+ tmp_path / "track.gpx", [(A_UNIX_TIME - 10, 13), (A_UNIX_TIME + 7, 30)]
+ )
+ with caplog.at_level(logging.WARNING):
+ metadata = GPXVideoExtractor(
+ video_path, gpx_path, video_sync=VideoSyncOption()
+ ).extract()
+ assert metadata.points
+ [warning] = _warnings(caplog)
+ assert "covers only part of the video" in warning
+
+ def test_a_gap_of_less_than_an_hour_is_interpolated(self, tmp_path):
+ metadata = self._extract(
+ tmp_path, [(A_UNIX_TIME - 1860, 60), (A_UNIX_TIME + 5, 60)]
+ )
+ first = metadata.points[0]
+ assert first.time == 0
+ assert first.get_unix_time() == pytest.approx(A_UNIX_TIME)
+
+ def test_the_limit_can_be_raised(self, tmp_path, monkeypatch):
+ """By MAPILLARY_TOOLS_MAX_GPX_GAP_SECONDS"""
+ monkeypatch.setattr(constants, "MAX_GPX_GAP_SECONDS", 3 * 3600)
+ metadata = self._extract(
+ tmp_path, [(A_UNIX_TIME - 7260, 60), (A_UNIX_TIME + 5, 60)]
+ )
+ assert metadata.points[0].time == 0
+
+ def test_started_with_the_video_ends_before_a_gap(self, tmp_path):
+ """--interpolation_use_gpx_start_time places the track, the clock gives its pace"""
+ metadata = self._extract(
+ tmp_path,
+ [(A_UNIX_TIME, 6), (A_UNIX_TIME + 7200, 60)],
+ sync_mode=SyncMode.REBASE,
+ )
+ times = [p.time for p in metadata.points]
+ assert times == [float(t) for t in range(7)]
+
+ def test_a_start_moved_into_a_gap_by_the_offset(self, tmp_path):
+ """
+ Without a GPS clock the track is assumed to start with the video, and
+ --interpolation_offset_time moves the start of the video into the gap
+ """
+ video_path = _write_camm_mp4(tmp_path / "video.mp4", UNTIMED_TRACK)
+ gpx_path = _write_gpx_parts(
+ tmp_path / "track.gpx", [(A_UNIX_TIME, 10), (A_UNIX_TIME + 7200, 60)]
+ )
+ metadata = GPXVideoExtractor(
+ video_path,
+ gpx_path,
+ interpolation=InterpolationOption(offset_time=7195),
+ clip_to_video=True,
+ ).extract()
+ times = [p.time for p in metadata.points]
+ assert times[0] == pytest.approx(5.0)
+ assert times[-1] == pytest.approx(65.0)
+
+ def test_a_video_started_at_a_lone_fix_falls_in_the_gap_after_it(self, tmp_path):
+ """
+ --interpolation_use_gpx_start_time starts the video at the first point
+ of the track, the only one before a gap of 2 hours: the track touches
+ the video at that instant alone
+ """
+ with pytest.raises(exceptions.MapillaryOutsideGPXTrackError) as info:
+ self._extract(
+ tmp_path,
+ [(A_UNIX_TIME, 0), (A_UNIX_TIME + 7200, 60)],
+ sync_mode=SyncMode.REBASE,
+ )
+ message = str(info.value)
+ assert (
+ "has a gap of 2.0 hours, from 2026-08-12T08:26:27.979846+00:00" in message
+ )
+ assert message.endswith("falls in. Trim the GPX to the video")
+
+ def test_a_video_assumed_to_start_at_a_lone_fix_falls_in_the_gap_after_it(
+ self, tmp_path
+ ):
+ video_path = _write_camm_mp4(tmp_path / "video.mp4", UNTIMED_TRACK)
+ gpx_path = _write_gpx_parts(
+ tmp_path / "track.gpx", [(A_UNIX_TIME, 0), (A_UNIX_TIME + 7200, 60)]
+ )
+ with pytest.raises(exceptions.MapillaryOutsideGPXTrackError) as info:
+ GPXVideoExtractor(video_path, gpx_path, clip_to_video=True).extract()
+ assert str(info.value).endswith(
+ "falls in. Trim the GPX to the video, or sample the video with "
+ "video_process and --video_start_time"
+ )
diff --git a/tests/unit/test_process_geotag_properties.py b/tests/unit/test_process_geotag_properties.py
index a741c811..06f50d69 100644
--- a/tests/unit/test_process_geotag_properties.py
+++ b/tests/unit/test_process_geotag_properties.py
@@ -9,7 +9,7 @@
from pathlib import Path
import pytest
-from mapillary_tools import process_geotag_properties as pgp, types
+from mapillary_tools import exceptions, process_geotag_properties as pgp, types
from mapillary_tools.geotag import factory
from mapillary_tools.geotag.options import SourceOption, SourceType
@@ -100,3 +100,75 @@ def test_video_process_geotags_the_frames_by_the_default(
assert self._sources(
monkeypatch, import_dir, [], ["native"], filetypes={types.FileType.IMAGE}
) == {"image": self.DEFAULT, "video": []}
+
+
+class TestNamedUnsupportedFiles:
+ """
+ A file named for processing that is neither an image nor a video was left
+ out of the results with only a warning, as if it had not been named
+ """
+
+ @staticmethod
+ def _process(monkeypatch, import_paths: list[Path]) -> list[types.MetadataOrError]:
+ processed: list[Path] = []
+
+ def process(paths, options, filetypes=None):
+ processed.extend(paths)
+ return []
+
+ monkeypatch.setattr(pgp, "process", process)
+ metadatas = pgp.process_geotag_properties(
+ import_paths,
+ None,
+ geotag_source=[],
+ geotag_source_path=None,
+ video_geotag_source=[],
+ )
+ assert not processed, "only images and videos are geotagged"
+ return metadatas
+
+ def test_named_file_is_failed(self, monkeypatch, tmp_path: Path):
+ notes = tmp_path / "notes.txt"
+ notes.write_bytes(b"not an image")
+
+ [metadata] = self._process(monkeypatch, [notes])
+
+ assert isinstance(metadata, types.ErrorMetadata)
+ assert metadata.filename == notes
+ assert metadata.filetype is None
+ assert isinstance(metadata.error, exceptions.MapillaryUnsupportedFileError)
+ assert str(metadata.error) == "Not a supported image or video"
+
+ def test_insta360_original_is_told_to_be_stitched(
+ self, monkeypatch, tmp_path: Path
+ ):
+ original = tmp_path / "VID_00_001.insv"
+ original.write_bytes(b"not stitched")
+
+ [metadata] = self._process(monkeypatch, [original])
+
+ assert isinstance(metadata, types.ErrorMetadata)
+ message = str(metadata.error)
+ assert message.startswith("Not a supported image or video. ")
+ assert "Insta360 Studio" in message
+ assert '{"source": "exiftool_runtime", "pattern": "../%g.insv"}' in message
+
+ def test_named_zip_is_left_to_upload(self, monkeypatch, tmp_path: Path):
+ archive = tmp_path / "sequence.zip"
+ archive.write_bytes(b"PK")
+
+ assert self._process(monkeypatch, [archive]) == []
+
+ def test_files_in_a_named_directory_are_left_alone(
+ self, monkeypatch, tmp_path: Path
+ ):
+ (tmp_path / "notes.txt").write_bytes(b"not an image")
+ (tmp_path / "VID_00_001.insv").write_bytes(b"not stitched")
+
+ assert self._process(monkeypatch, [tmp_path]) == []
+
+ def test_a_file_named_twice_is_failed_once(self, monkeypatch, tmp_path: Path):
+ notes = tmp_path / "notes.txt"
+ notes.write_bytes(b"not an image")
+
+ assert len(self._process(monkeypatch, [notes, notes])) == 1
diff --git a/tests/unit/test_sample_video.py b/tests/unit/test_sample_video.py
index 5aa857d5..632d9b0e 100644
--- a/tests/unit/test_sample_video.py
+++ b/tests/unit/test_sample_video.py
@@ -5,6 +5,7 @@
from __future__ import annotations
+import dataclasses
import datetime
import json
import logging
@@ -511,15 +512,69 @@ def test_located_along_the_video_track(self, tmpdir, setup_mock, monkeypatch):
(SourceType.NATIVE, 0)
]
- def test_samples_past_the_track_are_not_located(
- self, tmpdir, setup_mock, monkeypatch
+ def test_samples_past_the_track_are_left_out(
+ self, tmpdir, setup_mock, monkeypatch, caplog
):
# The track ends 20 seconds into the video
- samples = self._sample(tmpdir, monkeypatch, self._points(21))
+ with caplog.at_level(logging.WARNING, logger=sample_video.LOG.name):
+ samples = self._sample(tmpdir, monkeypatch, self._points(21))
+ assert [s.name for s in samples] == [
+ f"hello_v_{idx + 1:06d}.jpg" for idx in range(11)
+ ]
lon_lats = [exif_read.ExifRead(s).extract_lon_lat() for s in samples]
assert lon_lats[10] == pytest.approx((-74.0, 40.020))
- assert all(lon_lat == self.MOCK_SAMPLE_LON_LAT for lon_lat in lon_lats[11:])
+ assert (
+ "19 of the 30 samples of hello.mp4 fall outside its GPS track "
+ "(2021-08-10T06:38:06+00:00 to 2021-08-10T06:38:26+00:00) and were left out"
+ in caplog.text
+ )
+
+ @pytest.mark.parametrize(
+ "geotag_source, left_out",
+ [
+ # video_process with the default sources, which read the positions
+ # written into the samples
+ ([], True),
+ (["exif"], True),
+ # The sample_video command, whose samples may be geotagged later
+ # by the time since the first sample
+ (None, False),
+ (["gopro"], False),
+ (['{"source": "exiftool_xml", "pattern": "%g.xml"}'], False),
+ ],
+ )
+ def test_samples_before_the_track(
+ self, tmpdir, setup_mock, monkeypatch, geotag_source, left_out
+ ):
+ # The track starts 5 seconds into the video
+ track = self._points(60)[5:]
+ kwargs = {} if geotag_source is None else {"geotag_source": geotag_source}
+
+ samples = self._sample(tmpdir, monkeypatch, track, **kwargs)
+
+ names = [s.name for s in samples]
+ if left_out:
+ assert names[0] == "hello_v_000004.jpg"
+ assert len(samples) == 27
+ else:
+ # Left in, without a position, so that the others keep theirs
+ assert names[0] == "hello_v_000001.jpg"
+ assert len(samples) == 30
+ lon_lats = [exif_read.ExifRead(s).extract_lon_lat() for s in samples]
+ assert all(lon_lat == self.MOCK_SAMPLE_LON_LAT for lon_lat in lon_lats[:3])
+ assert lon_lats[3] == pytest.approx((-74.0, 40.006))
+
+ def test_none_left_out_if_the_track_covers_none(
+ self, tmpdir, setup_mock, monkeypatch, caplog
+ ):
+ """So that a video that misses its track does not go without a word"""
+ track = [dataclasses.replace(p, time=p.time + 100.0) for p in self._points(10)]
+ with caplog.at_level(logging.WARNING, logger=sample_video.LOG.name):
+ samples = self._sample(tmpdir, monkeypatch, track, geotag_source=[])
+
+ assert len(samples) == 30
+ assert "left out" not in caplog.text
def test_not_located_without_a_track(self, tmpdir, setup_mock, monkeypatch, caplog):
with caplog.at_level(logging.WARNING, logger=sample_video.LOG.name):
@@ -606,13 +661,15 @@ def reject_as_noise():
),
)
- def _write_gpx(self, path: Path, start: datetime.datetime) -> Path:
- # Over the 610 seconds that the 61 seconds of video span
+ def _write_gpx(
+ self, path: Path, start: datetime.datetime, num_points: int = 62
+ ) -> Path:
+ # By default over the 610 seconds that the 61 seconds of video span
trkpts = "".join(
f'"
- for i in range(62)
+ for i in range(num_points)
)
path.write_text(
'