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/factory.py b/mapillary_tools/geotag/factory.py
index e8f2f6bc..35b167a2 100644
--- a/mapillary_tools/geotag/factory.py
+++ b/mapillary_tools/geotag/factory.py
@@ -47,6 +47,12 @@
SourceType.CAMM: types.FileType.CAMM,
}
+_VIDEO_TYPE_NAMES = {
+ types.FileType.GOPRO: "GoPro",
+ types.FileType.BLACKVUE: "BlackVue",
+ types.FileType.CAMM: "CAMM",
+}
+
# The XML that the exiftool_xml source reads for a file, without a source path
# or a pattern: the output of `exiftool -X` for it, named after it
_EXIFTOOL_XML_DEFAULT_PATTERN = "%g.xml"
@@ -178,6 +184,7 @@ def _as_video_source(option: SourceOption) -> SourceOption:
option.filetypes = types.combine_filetype_filters(
option.filetypes, {video_type}
)
+ option.named_video_type = video_type
option.filetypes = types.combine_filetype_filters(
option.filetypes, {types.FileType.VIDEO}
)
@@ -283,6 +290,11 @@ def process(
video_metadata_or_errors = (
video_geotag.to_description(video_paths) if video_geotag else []
)
+ if option.named_video_type is not None:
+ video_metadata_or_errors = [
+ _say_of_another_type(metadata, option.named_video_type)
+ for metadata in video_metadata_or_errors
+ ]
else:
video_metadata_or_errors = []
@@ -356,13 +368,26 @@ def process(
sorted(reprocessable_paths - handed_paths), filetypes
)
- if video_types is not None and (unread_paths or sourceless_videos):
+ # Files that sources were let through, but that none of them described,
+ # nor said why, which they should not do: so that every file given gets
+ # one entry, but those that --filetypes leaves out
+ described_paths = {metadata.filename for metadata in final_metadatas}
+ unhandled_images, unhandled_videos = _filter_images_and_videos(
+ sorted(
+ (reprocessable_paths & handed_paths) - described_paths - set(unread_paths)
+ ),
+ filetypes,
+ )
+
+ if video_types is not None and (
+ unread_paths or sourceless_videos or unhandled_videos
+ ):
# Limited to some types of video, a video that no source read may not
# be of any of them. Those of other types are left out, as the videos
# found to be of other types
kept_paths = set(
_videos_not_of_other_types(
- unread_paths + sourceless_videos,
+ unread_paths + sourceless_videos + unhandled_videos,
video_types,
# The same for every option (see process_geotag_properties)
num_processes=next(iter(options)).num_processes,
@@ -370,6 +395,7 @@ def process(
)
unread_paths = [path for path in unread_paths if path in kept_paths]
sourceless_videos = [path for path in sourceless_videos if path in kept_paths]
+ unhandled_videos = [path for path in unhandled_videos if path in kept_paths]
final_metadatas.extend(
_describe_unread_video(path, unread_videos[path]) for path in unread_paths
@@ -378,6 +404,10 @@ def process(
_describe_sourceless_file(path, options, filetypes)
for path in sourceless_images + sourceless_videos
)
+ final_metadatas.extend(
+ _describe_unhandled_file(path, options)
+ for path in unhandled_images + unhandled_videos
+ )
return final_metadatas
@@ -461,6 +491,84 @@ def _describe_sourceless_file(
)
+def _describe_unhandled_file(
+ path: Path, options: T.Iterable[SourceOption]
+) -> types.ErrorMetadata:
+ """
+ Describe a file that the sources of **options** were let through, but that
+ none of them described, rather than leave it out of the results as if it
+ had not been there
+
+ >>> options = [SourceOption(SourceType.NATIVE), SourceOption(SourceType.EXIFTOOL_RUNTIME)]
+ >>> _describe_unhandled_file(Path("i.jpg"), options).error
+ MapillaryGeoTaggingError('The geotag sources native, exiftool_runtime gave no result for this image')
+ >>> _describe_unhandled_file(Path("v.mp4"), options[:1]).error
+ MapillaryVideoGPSNotFoundError('The geotag source native gave no result for this video')
+ """
+ names = list(dict.fromkeys(option.source.value for option in options))
+ sources = (
+ f"The geotag sources {', '.join(names)}"
+ if len(names) > 1
+ else f"The geotag source {names[0]}"
+ )
+ if utils.is_image_file(path):
+ return types.describe_error_metadata(
+ exceptions.MapillaryGeoTaggingError(
+ f"{sources} gave no result for this image"
+ ),
+ path,
+ filetype=types.FileType.IMAGE,
+ )
+ return types.describe_error_metadata(
+ exceptions.MapillaryVideoGPSNotFoundError(
+ f"{sources} gave no result for this video"
+ ),
+ path,
+ filetype=types.FileType.VIDEO,
+ )
+
+
+def _say_of_another_type(
+ metadata: types.VideoMetadataOrError, video_type: types.FileType
+) -> types.VideoMetadataOrError:
+ """
+ Say that a video in which the native reader, limited to the videos of
+ **video_type** by the name of the source, finds no GPS is of another
+ type, rather than that it has no GPS. The error stays the one the native
+ reader reports, which tells process() what to do with the video
+
+ >>> not_found = types.describe_error_metadata(
+ ... exceptions.MapillaryVideoGPSNotFoundError("No GPS data found from the video"),
+ ... Path("v.mp4"),
+ ... filetype=types.FileType.VIDEO,
+ ... )
+ >>> said = _say_of_another_type(not_found, types.FileType.GOPRO)
+ >>> type(said.error).__name__, str(said.error)
+ ('MapillaryVideoGPSNotFoundError', 'Not a GoPro video: the geotag source gopro reads GoPro videos only')
+ >>> noisy = types.describe_error_metadata(
+ ... exceptions.MapillaryGPSNoiseError("GPS is too noisy"),
+ ... Path("v.mp4"),
+ ... filetype=types.FileType.VIDEO,
+ ... )
+ >>> _say_of_another_type(noisy, types.FileType.GOPRO) is noisy
+ True
+ """
+ if not (
+ isinstance(metadata, types.ErrorMetadata)
+ and type(metadata.error) is exceptions.MapillaryVideoGPSNotFoundError
+ ):
+ return metadata
+
+ name = _VIDEO_TYPE_NAMES[video_type]
+ return types.describe_error_metadata(
+ exceptions.MapillaryVideoGPSNotFoundError(
+ f"Not a {name} video: the geotag source {video_type.value} reads {name} videos only"
+ ),
+ metadata.filename,
+ filetype=metadata.filetype,
+ )
+
+
def _describe_unread_video(
video_path: Path, sources: T.Sequence[SourceType]
) -> types.ErrorMetadata:
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/geotag_images_from_nmea_file.py b/mapillary_tools/geotag/geotag_images_from_nmea_file.py
index 1b868cbe..12982705 100644
--- a/mapillary_tools/geotag/geotag_images_from_nmea_file.py
+++ b/mapillary_tools/geotag/geotag_images_from_nmea_file.py
@@ -22,6 +22,18 @@
# GL (GLONASS), GA (Galileo), ...
_SENTENCE_RE = re.compile(r"\$[A-Z]{2}(RMC|GGA),")
+# What a byte that is not text decodes to
+_NOT_TEXT = "\ufffd"
+
+# The checksum that ends a sentence: *HH
+_CHECKSUM_RE = re.compile(r"\*[0-9A-Fa-f]{2}\s*$")
+
+# 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
+_TIME_RE = re.compile(r"\d{6}(\.\d+)?")
+_DATE_RE = re.compile(r"\d{6}")
+
class GeotagImagesFromNMEAFile(GeotagImagesFromGPX):
def __init__(
@@ -56,7 +68,7 @@ def _read_nmea_points(source_path: Path) -> list[geo.Point]:
try:
points = get_lat_lon_time_from_nmea(source_path)
- # OSError: unreadable; ValueError: not text (UnicodeError), or malformed
+ # OSError: unreadable; ValueError: not text, or malformed
# sentences and no fix; TypeError: a sentence without its time or date;
# UnboundLocalError: GGA sentences but no RMC sentence to give their date
except (OSError, ValueError, TypeError, UnboundLocalError) as ex:
@@ -92,29 +104,44 @@ def get_lat_lon_time_from_nmea(nmea_file: Path) -> list[geo.Point]:
for the same time: only the first is read, so that no two points share a
time.
- A malformed sentence, such as one whose checksum does not match, is
- skipped with a warning. A file of which no fix can be read and that has
- malformed sentences raises ValueError, as one that is not text does.
+ 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.
"""
- # NMEA sentences are ASCII. Decode them as UTF-8 on every platform, so
- # that a file that is not text fails to read, not only where UTF-8 is the
- # default encoding
- with nmea_file.open("r", encoding="utf-8") as f:
+ # NMEA sentences are ASCII. Decode them as UTF-8 on every platform, not as
+ # whatever the default encoding is, and decode a byte that is not text,
+ # such as one damaged in a sentence, as U+FFFD (see _parse_sentence())
+ with nmea_file.open("r", encoding="utf-8", errors="replace") as f:
lines = f.readlines()
lines = [line.rstrip("\n\r") for line in lines]
+ not_text = any(_NOT_TEXT in line for line in lines)
+
+ # (line number, line) of each RMC and GGA sentence
+ sentences = [
+ (number, line)
+ 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 enumerate(lines, start=1):
- if _SENTENCE_RE.search(line):
- try:
- messages.append(_parse_sentence(line))
- except ValueError as ex:
- malformed.append((number, line, ex))
+ for number, line in sentences:
+ try:
+ messages.append(_parse_sentence(line, checksummed=checksummed))
+ 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")
+
if malformed:
number, line, error = malformed[0]
if not points:
@@ -185,11 +212,14 @@ def _read_fixes(
return points
-def _parse_sentence(line: str) -> pynmea2.NMEASentence:
+def _parse_sentence(line: str, checksummed: bool = False) -> pynmea2.NMEASentence:
"""
Parse an NMEA sentence, raising ValueError if it is malformed: if it does
- not parse or its checksum does not match, or if a field that a fix is
- read from does not parse, which pynmea2 leaves a string
+ 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
>>> _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):
@@ -199,13 +229,36 @@ def _parse_sentence(line: str) -> pynmea2.NMEASentence:
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'
"""
+ 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:
# Its argument is (reason, sentence)
raise ValueError(ex.args[0][0]) from ex
+ formats = {"timestamp": _TIME_RE}
+ if isinstance(message, pynmea2.RMC):
+ formats["datestamp"] = _DATE_RE
+ for name, pattern in formats.items():
+ value = _raw_field(message, name)
+ # An empty field is no time or date, as before a fix
+ if value and not pattern.fullmatch(value):
+ raise ValueError(f"malformed {name}: {value!r}")
+
fields: dict[str, type] = {}
if isinstance(message, pynmea2.RMC):
fields = {"timestamp": datetime.time, "datestamp": datetime.date}
@@ -224,6 +277,12 @@ def _parse_sentence(line: str) -> pynmea2.NMEASentence:
return message
+def _raw_field(message: pynmea2.NMEASentence, name: str) -> str:
+ """The text of the field **name** of a sentence, empty if it is cut off"""
+ idx = message.name_to_idx[name]
+ return message.data[idx] if idx < len(message.data) else ""
+
+
def _rmc_datetime(message: pynmea2.RMC) -> datetime.datetime:
return datetime.datetime.combine(message.datestamp, message.timestamp)
diff --git a/mapillary_tools/geotag/geotag_images_from_video.py b/mapillary_tools/geotag/geotag_images_from_video.py
index 6422a5c9..e9e5bb6b 100644
--- a/mapillary_tools/geotag/geotag_images_from_video.py
+++ b/mapillary_tools/geotag/geotag_images_from_video.py
@@ -16,7 +16,7 @@
else:
from typing_extensions import override
-from .. import geo, types, utils
+from .. import exceptions, geo, types, utils
from .base import GeotagImagesFromGeneric
from .geotag_images_from_gpx import GeotagImagesFromGPX
from .geotag_videos_from_video import GeotagVideosFromVideo
@@ -151,4 +151,20 @@ def to_description(
offset_time=self.offset_time,
num_processes=self.num_processes,
)
- return geotag.to_description(image_paths)
+ sample_metadatas = geotag.to_description(image_paths)
+
+ # An image that is not a frame of any of the videos has no position
+ # here. Say so, rather than leave it out of the results, as a source
+ # that can geotag it (e.g. exif after this one) may still do
+ sample_paths = {metadata.filename for metadata in sample_metadatas}
+ return sample_metadatas + [
+ types.describe_error_metadata(
+ exceptions.MapillaryGeoTaggingError(
+ f"Not a frame of any video in {self.source_path}"
+ ),
+ image_path,
+ filetype=types.FileType.IMAGE,
+ )
+ for image_path in image_paths
+ if image_path not in sample_paths
+ ]
diff --git a/mapillary_tools/geotag/options.py b/mapillary_tools/geotag/options.py
index a4ed2514..da48bfd4 100644
--- a/mapillary_tools/geotag/options.py
+++ b/mapillary_tools/geotag/options.py
@@ -60,6 +60,11 @@ class SourceOption:
# Set by sample_video only, and not in the JSON schema
video_sync: VideoSyncOption | None = None
+ # The only type of video that a --video_geotag_source named after one
+ # (gopro, camm or blackvue) reads. Set when the options are parsed, and
+ # not in the JSON schema
+ named_video_type: types.FileType | None = None
+
@classmethod
def from_dict(cls, data: dict[str, T.Any]) -> SourceOption:
validate_option(data)
diff --git a/mapillary_tools/geotag/video_extractors/gpx.py b/mapillary_tools/geotag/video_extractors/gpx.py
index 3144a682..c45e3a9d 100644
--- a/mapillary_tools/geotag/video_extractors/gpx.py
+++ b/mapillary_tools/geotag/video_extractors/gpx.py
@@ -47,9 +47,7 @@
_MAX_RECORDING_SECONDS = 24 * 3600
# A GPS clock that runs this much faster than the video clock is a timelapse,
-# and the steps between its timestamps show it too. One timestamp far off the
-# rest, or one step of the clock, can pull the least-squares slope of a normal
-# video past it
+# and the steps between its timestamps show it too
_MIN_TIMELAPSE_RATE = 1.5
# No GPS fix can be timestamped before the GPS epoch, 1980-01-06. GPX files
@@ -74,7 +72,7 @@
# A GPS clock that runs within this fraction of the video's pace runs at it.
# Over 20 seconds or more, cameras that record in real time measure within
# 0.4% of 1x, or 1.6% for a GoPro MAX whose GPS glitches, while the clock of a
-# LabPano runs 9% to 14% faster than its video
+# LabPano runs 14% faster than its video
_MAX_CLOCK_RATE_ERROR = 0.05
# Nor can GPS timestamps measure a clock that departs from the video's pace by
@@ -83,6 +81,20 @@
# seconds long measure up to 15% off 1x, if by no more than 0.3 seconds
_MAX_CLOCK_DRIFT_SECONDS = 2.0
+# Nor, below the timelapse rate, do timestamps over less than this, of video
+# or of GPS time, measure a pace off 1x: over 8 to 20 seconds of GoPro and
+# PANOX V2 videos, a few timestamps seconds off the rest measure 0.24x to
+# 1.33x, by least squares and by the median of slopes alike. The LabPano
+# videos, whose clocks gain on them, span 3 minutes or more
+_MIN_CLOCK_SPAN_SECONDS = 60.0
+
+# The rate is the median of the slopes between pairs of timestamps, which a
+# few timestamps off the rest, or a step of the clock, do not move: a clock
+# that steps 18 seconds ahead two thirds into a 2-minute video pulls the
+# least-squares slope to 1.2x. Over more timestamps than this, an evenly spaced
+# selection of them measures it to within 0.01%
+_MAX_RATE_TIMESTAMPS = 300
+
class SyncMode(enum.Enum):
# Sync by video GPS timestamps if found, otherwise rebase
@@ -204,7 +216,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 +281,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 +325,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 +342,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 +468,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 +530,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 +607,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(
(
@@ -869,13 +939,25 @@ def _read_video_clock(
# See _MAX_CLOCK_RATE_ERROR and _MAX_CLOCK_DRIFT_SECONDS
rate = 1.0
elif _MIN_TIMELAPSE_RATE < rate:
- # A timestamp far off the rest, or a step of the clock, can pull the
- # least-squares slope far off, so a timelapse must show in the
- # median step between neighbouring pairs too. That is too coarse to
- # measure the rate, but not to tell a timelapse from a normal video
- step_rate = cls._median_step_rate(pairs)
- if step_rate is None or step_rate <= _MIN_TIMELAPSE_RATE:
+ # A timelapse must show in the steps between neighbouring pairs
+ # too: in their median, which is too coarse to measure the rate, but
+ # not to tell a timelapse from a normal video. Not in one step, nor
+ # in one of two: where a GPS stamps a fix seconds late, a few
+ # seconds of a video that records in real time can hold only 3
+ # timestamps, which measure 3.5x or 4x
+ steps = cls._step_rates(pairs)
+ if (
+ len(steps) < 2
+ or (min(steps) if len(steps) == 2 else statistics.median(steps))
+ <= _MIN_TIMELAPSE_RATE
+ ):
rate = 1.0
+ elif (
+ min(pairs[-1][0] - pairs[0][0], pairs[-1][1] - pairs[0][1])
+ < _MIN_CLOCK_SPAN_SECONDS
+ ):
+ # See _MIN_CLOCK_SPAN_SECONDS
+ rate = 1.0
# Anchored at its own video time, which is not 0 when the video's GPS
# starts late (e.g. GoPro drops noisy fixes at the start of a recording),
@@ -951,23 +1033,35 @@ def _estimate_clock_rate(
cls, pairs: T.Sequence[tuple[float, float]]
) -> float | None:
"""
- Least-squares slope of the Unix time over the video time, or None
- without two pairs at different video times
+ The slope of the Unix time over the video time: the median of the slopes
+ between pairs of the pairs (the Theil-Sen estimator), of an evenly
+ spaced selection of them if there are many (see _MAX_RATE_TIMESTAMPS).
+ None without two pairs at different video times
>>> GPXVideoExtractor._estimate_clock_rate([(0, 100), (1, 115), (2, 130)])
15.0
>>> GPXVideoExtractor._estimate_clock_rate([(0, 100)]) is None
True
- """
- if len(pairs) < 2 or pairs[-1][0] <= pairs[0][0]:
- return None
- n = len(pairs)
- mean_video_time = sum(v for v, _ in pairs) / n
- mean_unix_time = sum(u for _, u in pairs) / n
- covariance = sum((v - mean_video_time) * (u - mean_unix_time) for v, u in pairs)
- variance = sum((v - mean_video_time) ** 2 for v, _ in pairs)
- return covariance / variance
+ One timestamp far off the rest does not move it:
+
+ >>> GPXVideoExtractor._estimate_clock_rate([(i, 100 + i) for i in range(9)] + [(9, 130)])
+ 1.0
+ """
+ if len(pairs) > _MAX_RATE_TIMESTAMPS:
+ last = len(pairs) - 1
+ pairs = [
+ pairs[round(idx * last / (_MAX_RATE_TIMESTAMPS - 1))]
+ for idx in range(_MAX_RATE_TIMESTAMPS)
+ ]
+
+ slopes = [
+ (u2 - u1) / (v2 - v1)
+ for idx, (v1, u1) in enumerate(pairs)
+ for v2, u2 in pairs[idx + 1 :]
+ if v1 < v2
+ ]
+ return statistics.median(slopes) if slopes else None
@classmethod
def _median_step_rate(cls, pairs: T.Sequence[tuple[float, float]]) -> float | None:
@@ -979,12 +1073,22 @@ def _median_step_rate(cls, pairs: T.Sequence[tuple[float, float]]) -> float | No
>>> GPXVideoExtractor._median_step_rate(pairs)
1.0
"""
- steps = [
+ steps = cls._step_rates(pairs)
+ return statistics.median(steps) if steps else None
+
+ @classmethod
+ def _step_rates(cls, pairs: T.Sequence[tuple[float, float]]) -> list[float]:
+ """
+ The rates of the GPS clock between neighbouring pairs
+
+ >>> GPXVideoExtractor._step_rates([(0, 100), (1, 101), (1.5, 104)])
+ [1.0, 6.0]
+ """
+ return [
(u2 - u1) / (v2 - v1)
for (v1, u1), (v2, u2) in zip(pairs, pairs[1:])
if v1 < v2
]
- return statistics.median(steps) if steps else None
def track_gaps(
@@ -1027,6 +1131,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/geotag/video_extractors/nmea.py b/mapillary_tools/geotag/video_extractors/nmea.py
index 30d2e70e..7c19df4f 100644
--- a/mapillary_tools/geotag/video_extractors/nmea.py
+++ b/mapillary_tools/geotag/video_extractors/nmea.py
@@ -30,7 +30,7 @@ class NMEAVideoExtractor(GPXVideoExtractor):
def _parse_tracks(self) -> list[list[geo.Point]]:
try:
points = get_lat_lon_time_from_nmea(self.gpx_path)
- # OSError: unreadable; ValueError: not text (UnicodeError), or
+ # OSError: unreadable; ValueError: not text, or
# malformed sentences and no fix
except (OSError, ValueError) as ex:
raise exceptions.MapillaryInvalidGPXError(
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..a5b3acbf 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,12 @@
SourceType,
VideoSyncOption,
)
-from .geotag.video_extractors.gpx import GPXVideoExtractor, VideoClock
+from .geotag.video_extractors.gpx import (
+ describe_gap,
+ 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 +172,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 +264,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 samples may be left out
+ sample_sources = _parse_sample_sources(
+ geotag_source, geotag_source_path, video_import_path, interpolation
)
if rerun:
@@ -273,6 +326,7 @@ def sample_video(
start_time=video_start_time_dt,
geotag_options=geotag_options,
duration_ratio=video_duration_ratio,
+ sample_sources=sample_sources,
)
else:
assert 0 < video_sample_interval, (
@@ -286,6 +340,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 +821,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)
@@ -809,9 +869,22 @@ def _sample_single_video_by_interval(
f"Unable to extract video start time from {video_path}"
)
- # Located along the video's own GPS track, as distance samples are
+ # Located along the video's own GPS track, or the GPX (or NMEA) track of
+ # --video_geotag_source as GPX sync placed it on the video, as distance
+ # samples are
track = _read_track_for_interval_samples(video_path, geotag_options)
+ # The gaps of a track placed from a file, outside it (see track_gaps())
+ track_gaps_: list[tuple[geo.Point, geo.Point]] = []
+ track_from_file = track is not None and _placed_from_a_file(geotag_options)
+ if track is not None and track_from_file:
+ track = _start_at_the_first_sample(video_path, track, geotag_options)
+ track_gaps_ = track_gaps(track.points, constants.MAX_GPX_GAP_SECONDS / rate)
interpolator = geo.Interpolator([track.points]) if track is not None else None
+ # Or located by the geotagging stage along a GPX (or NMEA) track, by their
+ # 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 +903,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,30 +913,372 @@ 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 and track_from_file:
+ covered = [
+ _covers(track.points, track_gaps_, video_time)
+ for _, video_time, _ in samples
+ ]
+ # Unless the geotagging stage may geotag the samples by the GPS of
+ # their video instead
+ left_out = (
+ set()
+ if any(
+ option.source in _FIRST_SAMPLE_SOURCES
+ for option in sample_sources or []
+ )
+ else _samples_to_leave_out(covered, keep_leading=False)
+ )
+ track_range = (
+ _track_time(track.points[0], start_time, rate),
+ _track_time(track.points[-1], start_time, rate),
+ )
+ elif 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
+ interp = (
+ interpolator.interpolate(video_time)
+ if track is not None and interpolator is not None and covered[idx]
+ else None
+ )
+ if interp is not None and track_from_file:
+ # Stamped with the time of the track where the sample is, as
+ # distance samples of a track from a file are
+ timestamp = _track_time(interp, timestamp, 1.0)
+ 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)
- ):
- _add_position(exif_edit, track, interpolator.interpolate(video_time))
+ if track is not None and interp is not None:
+ _add_position(exif_edit, track, interp)
exif_edit.write()
+def _placed_from_a_file(geotag_options: T.Sequence[SourceOption]) -> bool:
+ """
+ Whether the track that the interval samples are located along was placed
+ on the video from a GPX (or NMEA) file: its only sources are such files
+ """
+ return bool(geotag_options) and all(
+ option.source in factory.EXTERNAL_GPS_SOURCES for option in geotag_options
+ )
+
+
+def _placed_from_a_video_file(geotag_options: T.Sequence[SourceOption]) -> bool:
+ """
+ Whether the track that distance samples are sampled along is placed on
+ the video from a GPX (or NMEA) file of --video_geotag_source, as interval
+ samples are located along it (see _read_track_for_interval_samples()). A
+ file of --geotag_source geotags the samples too, which the geotagging stage
+ places along its track by their capture times
+ """
+ return _placed_from_a_file(geotag_options) and not any(
+ option.filetypes is None or types.FileType.IMAGE in option.filetypes
+ for option in geotag_options
+ )
+
+
+def _file_track_gaps(
+ points: T.Sequence[geo.Point],
+) -> list[tuple[geo.Point, geo.Point]]:
+ """
+ The gaps of a track placed on the video from a GPX (or NMEA) file, outside
+ it as in the geotagging stage (see track_gaps()): between consecutive
+ points more than constants.MAX_GPX_GAP_SECONDS apart by the GPS times they
+ keep, which a timelapse packs into less video, as interval sampling scales
+ the limit by the rate of the video
+ """
+ gaps: list[tuple[geo.Point, geo.Point]] = []
+ for before, after in geo.pairwise(points):
+ before_time, after_time = before.get_unix_time(), after.get_unix_time()
+ if (
+ before_time is not None
+ and after_time is not None
+ and constants.MAX_GPX_GAP_SECONDS < after_time - before_time
+ ):
+ gaps.append((before, after))
+ return gaps
+
+
+def _describe_file_track_gap(before: geo.Point, after: geo.Point) -> str:
+ """Describe a gap that _file_track_gaps() found, by its GPS times"""
+ before_time, after_time = before.get_unix_time(), after.get_unix_time()
+ assert before_time is not None and after_time is not None
+ return describe_gap(before, after, before_time, after_time)
+
+
+def _start_at_the_first_sample(
+ video_path: Path,
+ track: types.VideoMetadata,
+ geotag_options: T.Sequence[SourceOption],
+) -> types.VideoMetadata:
+ """
+ A track that --interpolation_use_gpx_start_time starts with the video,
+ moved to start at the first interval sample: GPX sync starts it at the
+ composition time of the first frame, at which distance sampling reads the
+ first frame, and interval sampling stamps the samples from time 0, which
+ ffmpeg cuts the first one at
+ """
+ option = geotag_options[0]
+ if option.interpolation is None or not option.interpolation.use_gpx_start_time:
+ return track
+ # As GPX sync starts it
+ start = GPXVideoExtractor(
+ video_path,
+ video_path,
+ video_sync=option.video_sync,
+ interpolation=option.interpolation,
+ )._gpx_start_video_time()
+ if not start:
+ return track
+ return dataclasses.replace(
+ track,
+ points=[dataclasses.replace(p, time=p.time - start) for p in track.points],
+ )
+
+
+def _covers(
+ points: T.Sequence[geo.Point],
+ gaps: T.Sequence[tuple[geo.Point, geo.Point]],
+ video_time: float,
+) -> bool:
+ """
+ Whether a track placed on the video covers a sample at **video_time**, as
+ the geotagging stage tells whether a GPX track covers an image
+ """
+ try:
+ check_within_track(video_time, points, gaps)
+ except exceptions.MapillaryOutsideGPXTrackError:
+ return False
+ return True
+
+
+def _track_time(
+ point: geo.Point, start_time: datetime.datetime, rate: float
+) -> datetime.datetime:
+ """
+ The time of a point of a track placed on the video: its own GPS time, or,
+ without one, that of its video time from **start_time** at **rate**
+ """
+ unix_time = point.get_unix_time()
+ if unix_time is None:
+ return start_time + datetime.timedelta(seconds=point.time * rate)
+ return datetime.datetime.fromtimestamp(unix_time, tz=datetime.timezone.utc)
+
+
+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:
"""
- The video's own GPS track, which its interval samples are located along
- as distance samples are, or None to leave their positions to the
- geotagging stage. That places them by their capture times along a GPX or
- NMEA track, which interval sampling, unlike distance sampling, does not
- place along the video: it stamps a video without a GPS clock from its
- creation time, rather than from the start of the track
+ The track that the interval samples of a video are located along, in
+ video time, as distance samples are: the video's own GPS track, or that of
+ a GPX (or NMEA) file of --video_geotag_source as GPX sync placed it on the
+ video. None to leave their positions to the geotagging stage: that places
+ them along the track of a GPX (or NMEA) file of --geotag_source, which
+ geotags images, by their capture times
"""
- if any(option.source in factory.EXTERNAL_GPS_SOURCES for option in geotag_options):
+ if any(
+ option.source in factory.EXTERNAL_GPS_SOURCES
+ and (option.filetypes is None or types.FileType.IMAGE in option.filetypes)
+ for option in geotag_options
+ ):
return None
# Through the factory with the same sources as distance sampling, which
@@ -925,6 +1343,10 @@ def _sample_video_stream_by_distance(
points: T.Sequence[geo.Point],
video_track_parser: mp4_sample_parser.TrackBoxParser,
sample_distance: float,
+ # The gaps of the track, outside it, whose frames are left out (see
+ # _file_track_gaps()), and the name of the video, to warn of them
+ gaps: T.Sequence[tuple[geo.Point, geo.Point]] = (),
+ video_name: str = "the video",
) -> dict[int, tuple[mp4_sample_parser.Sample, geo.Point]]:
"""
Locate video frames along the track (points), then resample them by the minimal sample_distance, and return the sparse frames.
@@ -955,6 +1377,47 @@ def _sample_video_stream_by_distance(
points, video_sample.exact_composition_time
)
]
+ if gaps:
+ # Clamped to the ends of the track, past which frames are kept by 1 ms,
+ # so that only its gaps leave frames out
+ covered = [
+ _covers(
+ points,
+ gaps,
+ min(
+ max(video_sample.exact_composition_time, points[0].time),
+ points[-1].time,
+ ),
+ )
+ for _, video_sample, _ in interp_sample_points
+ ]
+ times_in_gaps = [
+ video_sample.exact_composition_time
+ for (_, video_sample, _), c in zip(interp_sample_points, covered)
+ if not c
+ ]
+ if times_in_gaps:
+ gaps_left_out = [
+ (before, after)
+ for before, after in gaps
+ if any(before.time < t < after.time for t in times_in_gaps)
+ ]
+ LOG.warning(
+ "%d %s of %s %s in %s of its GPS track and %s left out: %s",
+ len(times_in_gaps),
+ "frame" if len(times_in_gaps) == 1 else "frames",
+ video_name,
+ "falls" if len(times_in_gaps) == 1 else "fall",
+ "a gap" if len(gaps_left_out) == 1 else f"{len(gaps_left_out)} gaps",
+ "was" if len(times_in_gaps) == 1 else "were",
+ "; ".join(
+ _describe_file_track_gap(before, after)
+ for before, after in gaps_left_out
+ ),
+ )
+ interp_sample_points = [
+ x for x, c in zip(interp_sample_points, covered) if c
+ ]
LOG.info("Found total %d interpolated video samples", len(interp_sample_points))
# select sample points by sample distance
@@ -1045,6 +1508,8 @@ def _sample_single_video_by_distance(
start_time: datetime.datetime | None = None,
geotag_options: T.Sequence[SourceOption] | None = None,
duration_ratio: float = 1.0,
+ # How the samples are geotagged (see _parse_sample_sources())
+ sample_sources: T.Sequence[SourceOption] | None = None,
) -> None:
if geotag_options is None:
geotag_options = [SourceOption(SourceType.NATIVE)]
@@ -1094,8 +1559,25 @@ def _sample_single_video_by_distance(
video_stream_idx = video_stream["index"]
moov_parser = mp4_sample_parser.MovieBoxParser.parse_file(video_path)
video_track_parser = moov_parser.extract_track_at(video_stream_idx)
+ # A track placed on the video from a GPX (or NMEA) file of
+ # --video_geotag_source is not interpolated across its gaps, outside it as
+ # in the geotagging stage: the frames in them are left out, as interval
+ # samples are. Unless the geotagging stage may geotag the samples by the
+ # GPS of their video instead
+ gaps = (
+ _file_track_gaps(video_metadata.points)
+ if _placed_from_a_video_file(geotag_options)
+ and not any(
+ option.source in _FIRST_SAMPLE_SOURCES for option in sample_sources or []
+ )
+ else []
+ )
sample_points_by_frame_idx = _sample_video_stream_by_distance(
- video_metadata.points, video_track_parser, sample_distance
+ video_metadata.points,
+ video_track_parser,
+ sample_distance,
+ gaps,
+ video_path.name,
)
# A track that misses the video, such as a GPX from another trip or one
# placed by a wrong --video_start_time, selects no frames, and an empty
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..e9b1d5d8 100644
--- a/tests/integration/test_process.py
+++ b/tests/integration/test_process.py
@@ -725,6 +725,38 @@ def test_process_filetypes_no_source_is_given(setup_data: py.path.local):
assert {d["error"]["type"] for d in descs} == {"MapillaryGeoTaggingError"}
+def test_process_image_not_a_frame_of_any_video(setup_data: py.path.local):
+ """
+ The legacy sources that geotag images from videos left out an image that is
+ not a frame of any of the videos, as if it had not been given
+ """
+ video_path = setup_data.join("gopro_data").join("max-360mode.mp4")
+ image_dir = setup_data.mkdir("frames")
+ setup_data.join("images").join("DSC00001.JPG").copy(image_dir)
+ legacy_source = [
+ "--geotag_source=gopro_videos",
+ f"--geotag_source_path={video_path}",
+ ]
+
+ descs = run_process_for_descs([*legacy_source, str(image_dir)])
+ assert [(Path(d["filename"]).name, d["error"]) for d in descs] == [
+ (
+ "DSC00001.JPG",
+ {
+ "type": "MapillaryGeoTaggingError",
+ "message": f"Not a frame of any video in {video_path}",
+ },
+ )
+ ]
+
+ # A later source can still geotag it
+ descs = run_process_for_descs(
+ [*legacy_source, "--geotag_source=exif", str(image_dir)]
+ )
+ assert [Path(d["filename"]).name for d in descs] == ["DSC00001.JPG"]
+ assert "error" not in descs[0]
+
+
def test_process_video_geotag_source_with_gpx_specified(setup_data: py.path.local):
video_path = setup_data.join("videos").join("sample-5s.mp4")
gpx_file = setup_data.join("gpx").join("sf_30km_h.gpx")
@@ -1049,10 +1081,15 @@ def test_process_video_geotag_source_names_in_readme(setup_data: py.path.local):
descs = run_process_for_descs(["--video_geotag_source", source, video_path])
assert_descs_exact_equal(native_descs, descs)
- # A GoPro video has none of the GPS of the other formats
- for source in ["camm", "blackvue"]:
+ # A GoPro video is of none of the other formats, which says so
+ for source, name in [("camm", "CAMM"), ("blackvue", "BlackVue")]:
descs = run_process_for_descs(["--video_geotag_source", source, video_path])
- assert [d["error"]["type"] for d in descs] == ["MapillaryVideoGPSNotFoundError"]
+ assert [d["error"] for d in descs] == [
+ {
+ "type": "MapillaryVideoGPSNotFoundError",
+ "message": f"Not a {name} video: the geotag source {source} reads {name} videos only",
+ }
+ ]
def test_process_video_geotag_source_with_exiftool_runtime_pattern(
@@ -1311,6 +1348,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..ae11ef3f 100644
--- a/tests/integration/test_video_process.py
+++ b/tests/integration/test_video_process.py
@@ -223,6 +223,131 @@ 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 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
+ placed it on the video, where a GPX of --geotag_source puts them. They used
+ to carry no position, and failed geotagging
+ """
+ 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")
+ args = [
+ "--video_sample_distance=-1",
+ "--video_sample_interval=1",
+ *["--video_start_time", "2025_03_14_07_00_30_000"],
+ *["--geotag_source_path", str(gpx_file)],
+ str(video_path),
+ ]
+
+ with tempfile.TemporaryDirectory() as dir:
+ descs = run_video_process_for_descs(
+ ["--video_geotag_source=gpx", *args, str(Path(dir, "samples"))]
+ )
+ with tempfile.TemporaryDirectory() as dir:
+ gpx_descs = run_video_process_for_descs(
+ ["--geotag_source=gpx", *args, str(Path(dir, "samples"))]
+ )
+
+ assert descs
+ assert all("error" not in desc for desc in descs), descs
+ by_name = {Path(d["filename"]).name: d for d in gpx_descs}
+ assert len(by_name) == len(descs)
+ for desc in descs:
+ gpx_desc = by_name[Path(desc["filename"]).name]
+ assert desc["MAPCaptureTime"] == gpx_desc["MAPCaptureTime"]
+ assert desc["MAPLatitude"] == pytest.approx(gpx_desc["MAPLatitude"], abs=1e-7)
+ assert desc["MAPLongitude"] == pytest.approx(gpx_desc["MAPLongitude"], abs=1e-7)
+
+
+def test_video_process_by_distance_leaves_out_the_frames_in_a_gap_of_a_video_gpx(
+ setup_data: py.path.local,
+):
+ """
+ Without its point at 07:00:33.088, the GPX has a gap from 07:00:31.533 to
+ 07:00:34.379, over the limit, inside the video. The frames in it were
+ placed across it
+ """
+ pytest_skip_if_not_ffmpeg_installed()
+
+ video_path = setup_data.join("videos").join("sample-5s.mp4")
+ gpx = setup_data.join("gpx").join("sf_30km_h.gpx").read_text("utf-8")
+ gpx_with_gap = re.sub(
+ r").)*?.*?",
+ "",
+ gpx,
+ flags=re.S,
+ )
+ assert gpx_with_gap != gpx
+
+ with tempfile.TemporaryDirectory() as dir:
+ gpx_file = Path(dir, "gap.gpx")
+ gpx_file.write_text(gpx_with_gap, encoding="utf-8")
+ descs = run_video_process_for_descs(
+ [
+ "--video_geotag_source=gpx",
+ *["--geotag_source_path", str(gpx_file)],
+ *["--video_start_time", "2025_03_14_07_00_30_000"],
+ str(video_path),
+ str(Path(dir, "samples")),
+ ],
+ env={**os.environ, "MAPILLARY_TOOLS_MAX_GPX_GAP_SECONDS": "2.5"},
+ )
+
+ assert all("error" not in desc for desc in descs), descs
+ capture_times = sorted(desc["MAPCaptureTime"] for desc in descs)
+ assert capture_times[0] < "2025_03_14_07_00_31_533", capture_times
+ assert "2025_03_14_07_00_34_379" < capture_times[-1], capture_times
+ assert not [
+ capture_time
+ for capture_time in capture_times
+ if "2025_03_14_07_00_31_534" < capture_time < "2025_03_14_07_00_34_378"
+ ], capture_times
+
+
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_geotag_entries.py b/tests/unit/test_geotag_entries.py
new file mode 100644
index 00000000..80147c2f
--- /dev/null
+++ b/tests/unit/test_geotag_entries.py
@@ -0,0 +1,168 @@
+# 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.
+
+"""
+Every file given to geotag/factory.process gets exactly one entry, but those
+that --filetypes leaves out.
+"""
+
+from __future__ import annotations
+
+import collections
+import shutil
+import typing as T
+from pathlib import Path
+
+import pytest
+from mapillary_tools import exceptions, types
+from mapillary_tools.geotag import factory
+from mapillary_tools.geotag.options import SourceOption, SourcePathOption, SourceType
+
+DATA = Path(__file__).parent.parent / "data"
+UNIT_DATA = Path(__file__).parent / "data"
+
+GOPRO_VIDEO = DATA / "gopro_data" / "max-360mode.mp4"
+OTHER_VIDEO = DATA / "videos" / "sample-5s.mp4"
+GPX = DATA / "gpx" / "sf_30km_h.gpx"
+
+
+def _inputs(tmp_path: Path) -> tuple[list[Path], list[Path]]:
+ """
+ Images (one with GPS, one without, one in the folder that sample_video
+ writes the frames of the GoPro video to) and videos (a GoPro video, and a
+ video without GPS)
+ """
+ gps_image = tmp_path / "images" / "with_gps.jpg"
+ gps_image.parent.mkdir()
+ shutil.copyfile(DATA / "images" / "DSC00001.JPG", gps_image)
+ no_gps_image = tmp_path / "images" / "no_gps.jpg"
+ shutil.copyfile(UNIT_DATA / "empty_exif.jpg", no_gps_image)
+ frame = tmp_path / "frames" / GOPRO_VIDEO.name / f"{GOPRO_VIDEO.stem}_0_000001.jpg"
+ frame.parent.mkdir(parents=True)
+ shutil.copyfile(UNIT_DATA / "test_exif.jpg", frame)
+ return [gps_image, no_gps_image, frame], [GOPRO_VIDEO, OTHER_VIDEO]
+
+
+def _legacy(source: SourceType) -> SourceOption:
+ # Geotag images from the videos beside the GoPro video
+ return SourceOption(
+ source, source_path=SourcePathOption(source_path=GOPRO_VIDEO.parent)
+ )
+
+
+def _limited(source: SourceType, filetypes: set[types.FileType]) -> SourceOption:
+ return SourceOption(source, filetypes=filetypes)
+
+
+def _gpx() -> SourceOption:
+ return SourceOption(SourceType.GPX, source_path=SourcePathOption(source_path=GPX))
+
+
+SOURCES: dict[str, T.Callable[[], list[SourceOption]]] = {
+ "native": lambda: [SourceOption(SourceType.NATIVE)],
+ "exif": lambda: [SourceOption(SourceType.EXIF)],
+ "gpx": lambda: [_gpx()],
+ "gopro": lambda: [_legacy(SourceType.GOPRO)],
+ "gopro,exif": lambda: [_legacy(SourceType.GOPRO), SourceOption(SourceType.EXIF)],
+ "blackvue,native": lambda: [
+ _legacy(SourceType.BLACKVUE),
+ SourceOption(SourceType.NATIVE),
+ ],
+ "native for images": lambda: [_limited(SourceType.NATIVE, {types.FileType.IMAGE})],
+ "native for gopro, gpx": lambda: [
+ _limited(SourceType.NATIVE, {types.FileType.GOPRO}),
+ _gpx(),
+ ],
+}
+
+FILETYPES: dict[str, set[types.FileType] | None] = {
+ "all": None,
+ "image": {types.FileType.IMAGE},
+ "video": {types.FileType.VIDEO},
+ "gopro": {types.FileType.GOPRO},
+ "image,blackvue": {types.FileType.IMAGE, types.FileType.BLACKVUE},
+}
+
+
+@pytest.mark.parametrize("filetypes_name", FILETYPES)
+@pytest.mark.parametrize("sources_name", SOURCES)
+def test_every_input_gets_one_entry(
+ tmp_path: Path, sources_name: str, filetypes_name: str
+):
+ images, videos = _inputs(tmp_path)
+ filetypes = FILETYPES[filetypes_name]
+ options = SOURCES[sources_name]()
+ for option in options:
+ # As process_geotag_properties limits them
+ option.filetypes = types.combine_filetype_filters(option.filetypes, filetypes)
+ option.num_processes = 0
+
+ metadatas = factory.process(images + videos, options, filetypes)
+
+ counts = collections.Counter(metadata.filename for metadata in metadatas)
+ assert all(count == 1 for count in counts.values()), counts
+
+ # --filetypes leaves out the files of other types: images by their names,
+ # and videos once a source reads them as of other types. A source that does
+ # not tell the type of a video (gpx) may keep a video of another type
+ expected = set()
+ if filetypes is None or types.FileType.IMAGE in filetypes:
+ expected.update(images)
+ if filetypes is None or types.FileType.VIDEO in filetypes:
+ expected.update(videos)
+ elif types.FileType.GOPRO in filetypes:
+ expected.add(GOPRO_VIDEO)
+ assert expected <= set(counts) <= expected | set(videos)
+
+
+def test_an_image_not_a_frame_of_any_video_is_reported(tmp_path: Path):
+ """
+ The legacy sources that geotag images from videos left out an image that is
+ not a frame of any of the videos, as if it had not been given
+ """
+ images, _ = _inputs(tmp_path)
+
+ metadatas = factory.process(images[:1], [_legacy(SourceType.GOPRO)])
+
+ assert len(metadatas) == 1
+ assert isinstance(metadatas[0], types.ErrorMetadata)
+ assert isinstance(metadatas[0].error, exceptions.MapillaryGeoTaggingError)
+ assert str(metadatas[0].error) == (
+ f"Not a frame of any video in {GOPRO_VIDEO.parent}"
+ )
+
+
+def test_a_later_source_geotags_an_image_not_a_frame_of_any_video(tmp_path: Path):
+ images, _ = _inputs(tmp_path)
+
+ metadatas = factory.process(
+ images[:1], [_legacy(SourceType.GOPRO), SourceOption(SourceType.EXIF)]
+ )
+
+ assert len(metadatas) == 1
+ assert isinstance(metadatas[0], types.ImageMetadata)
+ assert metadatas[0].lat == pytest.approx(45.5169031)
+
+
+def test_a_file_no_source_describes_is_reported(monkeypatch, tmp_path: Path):
+ """
+ A file that the sources were let through, but none described, is reported,
+ and not left out
+ """
+ images, _ = _inputs(tmp_path)
+
+ class Silent:
+ def to_description(self, paths):
+ return []
+
+ monkeypatch.setattr(factory, "_build_image_geotag", lambda option: Silent())
+
+ metadatas = factory.process(images[:1], [SourceOption(SourceType.EXIF)])
+
+ assert len(metadatas) == 1
+ assert isinstance(metadatas[0], types.ErrorMetadata)
+ assert str(metadatas[0].error) == (
+ "The geotag source exif gave no result for this image"
+ )
diff --git a/tests/unit/test_geotag_filetypes.py b/tests/unit/test_geotag_filetypes.py
index 5310fc1c..326edb35 100644
--- a/tests/unit/test_geotag_filetypes.py
+++ b/tests/unit/test_geotag_filetypes.py
@@ -335,3 +335,71 @@ def test_video_of_the_type_without_usable_gps(self):
NativeVideoExtractor(
DATA / "gopro_data" / "hero8.mp4", filetypes=GOPRO
).extract()
+
+
+class TestSourceNamedAfterAVideoType:
+ """
+ --video_geotag_source gopro, camm or blackvue reads only the videos of its
+ type. A video of another type was reported as having no GPS
+ """
+
+ @staticmethod
+ def _process(
+ video: Path, source: str, filetypes: set[types.FileType] | None = None
+ ) -> list[types.MetadataOrError]:
+ options = factory.parse_source_options(
+ geotag_source=[], video_geotag_source=[source], geotag_source_path=None
+ )
+ return _process_without_stubs([video], options, filetypes)
+
+ @pytest.mark.parametrize(
+ "video, source, message",
+ [
+ (
+ DATA / "videos" / "sample-5s.mp4",
+ "gopro",
+ "Not a GoPro video: the geotag source gopro reads GoPro videos only",
+ ),
+ (
+ DATA / "gopro_data" / "max-360mode.mp4",
+ "camm",
+ "Not a CAMM video: the geotag source camm reads CAMM videos only",
+ ),
+ (
+ DATA / "gopro_data" / "max-360mode.mp4",
+ "blackvue_videos",
+ "Not a BlackVue video: the geotag source blackvue reads BlackVue videos only",
+ ),
+ (
+ DATA / "gopro_data" / "max-360mode.mp4",
+ '{"source": "camm"}',
+ "Not a CAMM video: the geotag source camm reads CAMM videos only",
+ ),
+ ],
+ )
+ def test_video_of_another_type(self, video, source, message):
+ [result] = self._process(video, source)
+
+ assert isinstance(result, types.ErrorMetadata)
+ # The error the native reader gives, which process() acts on
+ assert type(result.error) is exceptions.MapillaryVideoGPSNotFoundError
+ assert str(result.error) == message
+
+ def test_video_of_its_type(self):
+ [result] = self._process(DATA / "gopro_data" / "max-360mode.mp4", "gopro")
+
+ assert isinstance(result, types.VideoMetadata)
+ assert result.filetype is types.FileType.GOPRO
+
+ def test_native_source_limited_by_its_filetypes(self):
+ """Not named after a type of video: its error is as before"""
+ [result] = self._process(
+ DATA / "videos" / "sample-5s.mp4",
+ '{"source": "native", "filetypes": ["gopro"]}',
+ )
+
+ assert isinstance(result, types.ErrorMetadata)
+ assert str(result.error) == "No GPS data found from the video"
+
+ def test_video_of_another_type_than_filetypes_is_left_out(self):
+ assert self._process(DATA / "videos" / "sample-5s.mp4", "gopro", GOPRO) == []
diff --git a/tests/unit/test_geotag_images_from_video.py b/tests/unit/test_geotag_images_from_video.py
index ae350c03..3392a98d 100644
--- a/tests/unit/test_geotag_images_from_video.py
+++ b/tests/unit/test_geotag_images_from_video.py
@@ -166,13 +166,14 @@ def test_samples_are_placed_along_a_clock_that_gains(tmp_path: Path):
"""
A clock that gains on the video, as a LabPano's runs 8/7 times as fast,
is no timelapse, and its track timed by the video placed the samples
- ahead
+ ahead. Over more than a minute, as less is too short to measure a pace
+ off 1x
"""
video_path = tmp_path / "hello.mp4"
sample_dir = tmp_path / "samples" / "hello.mp4"
sample_dir.mkdir(parents=True)
- gps_seconds = [i * 8 / 7 for i in range(30)]
+ gps_seconds = [i * 8 / 7 for i in range(70)]
samples = [
_sample(
sample_dir, k + 1, START_TIME + datetime.timedelta(seconds=2 * k * 8 / 7)
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_gpx_video_timestamps.py b/tests/unit/test_gpx_video_timestamps.py
index 71e3ccda..ea3346c6 100644
--- a/tests/unit/test_gpx_video_timestamps.py
+++ b/tests/unit/test_gpx_video_timestamps.py
@@ -190,11 +190,54 @@ def test_a_stale_first_timestamp_off_the_line(self):
assert clock.to_unix_time(0) == pytest.approx(START.timestamp() + 0.1)
def test_a_timestamp_far_off_makes_no_timelapse(self):
- """It pulls the least-squares slope past the timelapse rate"""
+ """It would pull a least-squares slope past the timelapse rate"""
pairs = [(i, i) for i in range(9)] + [(9, 30)]
assert _clock(pairs).rate == 1.0
+ def test_a_few_timestamps_make_no_timelapse(self):
+ """
+ A few seconds of a video whose GPS stamps a fix seconds late hold 3
+ timestamps, which measure 4x
+ """
+ assert _clock([(0, 0), (1, 1), (2, 8)]).rate == 1.0
+
+ def test_a_timelapse_of_3_timestamps(self):
+ """Stamped once a second of video, each step of which is 15 seconds"""
+ assert _clock([(0, 0), (1, 15), (2, 30)]).rate == pytest.approx(15)
+
+ def test_one_step_makes_no_timelapse(self):
+ """Nor can it tell a timelapse from a fix stamped seconds late"""
+ assert _clock([(0, 0), (1, 15)]).rate == 1.0
+
+ def test_a_rate_measured_over_less_than_a_minute_is_1(self):
+ """
+ Over 9 seconds, a last fix 6 seconds late pulls the least-squares slope
+ to 1.33x, a drift of 3 seconds; nor can less than a minute of timestamps
+ measure the pace of a clock that gains, by a second over 50 seconds
+ """
+ assert _clock([(i, i + (6 if i == 9 else 0)) for i in range(10)]).rate == 1.0
+ assert _clock([(i, i * 8 / 7) for i in range(50)]).rate == 1.0
+
+ def test_a_step_of_the_clock_does_not_tilt_it(self):
+ """
+ A clock that steps 18 seconds ahead two thirds into a 2-minute video
+ would pull the least-squares slope to 1.2x
+ """
+ clock = _clock([(i, i + (18 if 80 <= i else 0)) for i in range(120)])
+
+ assert clock.rate == 1.0
+ # By the timestamps before the step, which are the most
+ assert clock.to_unix_time(0) == pytest.approx(START.timestamp())
+
+ def test_timestamps_off_the_line_do_not_tilt_it(self):
+ """Late fixes of a camera that records in real time"""
+ late = {30, 31, 32, 95, 96, 119}
+ clock = _clock([(i, i + (6 if i in late else 0)) for i in range(120)])
+
+ assert clock.rate == 1.0
+ assert clock.to_unix_time(0) == pytest.approx(START.timestamp())
+
def _gpx(path: Path, seconds: T.Iterable[float]) -> Path:
"""A GPX track with a point at each of the GPS seconds from START"""
@@ -258,8 +301,8 @@ def test_by_the_straight_clock(self, tmp_path, monkeypatch, video_sync):
assert placed[seconds] == pytest.approx(seconds - 0.1)
def test_at_the_pace_of_a_clock_that_gains(self, tmp_path, monkeypatch):
- # Over 30 seconds, as 12 are too few to tell 8/7 from 1x (by 1.7 s)
- self.mock_video_gps(monkeypatch, [(i, i * 8 / 7) for i in range(30)])
+ # Over more than a minute, as less is too short to measure a pace off 1x
+ self.mock_video_gps(monkeypatch, [(i, i * 8 / 7) for i in range(70)])
self.GPX_SECONDS = [float(s) for s in range(15)]
placed = self._extract(tmp_path, video_sync=VideoSyncOption())
@@ -286,7 +329,7 @@ def test_started_with_the_video_at_the_pace_of_the_clock(
):
# The timestamps of the video 100 seconds before the GPX, 8/7 times as
# fast as the video
- self.mock_video_gps(monkeypatch, [(i, i * 8 / 7 - 100) for i in range(30)])
+ self.mock_video_gps(monkeypatch, [(i, i * 8 / 7 - 100) for i in range(70)])
placed = self._extract(
tmp_path,
diff --git a/tests/unit/test_nmea_parser.py b/tests/unit/test_nmea_parser.py
index e1ff108b..2458bcb8 100644
--- a/tests/unit/test_nmea_parser.py
+++ b/tests/unit/test_nmea_parser.py
@@ -236,6 +236,141 @@ def test_malformed_fields_are_skipped(tmp_path, caplog, malformed):
assert "Skipping 1 of the 5 RMC and GGA sentences" in caplog.text
+@pytest.mark.parametrize("gga_first", [True, False])
+def test_a_sentence_cut_inside_its_date_is_skipped(tmp_path, caplog, gga_first):
+ """
+ pynmea2 read the date "2212" of an RMC sentence cut short, the last of a
+ log, as 2012-02-02, and the last fix took it: a track of 10 years
+ """
+ lines = []
+ for time, date in [("235959.00", "211222"), ("000000.00", "221222")]:
+ if gga_first:
+ lines += [_gga(time), _rmc(time, date=date)]
+ else:
+ lines += [_rmc(time, date=date), _gga(time)]
+ if gga_first:
+ lines.append(_gga("000001.00"))
+ cut = _rmc("000001.00").split(",221222,")[0] + ",2212"
+
+ with caplog.at_level(logging.WARNING):
+ assert _read(tmp_path, *lines, cut) == [
+ _fix("23:59:59", date="2022-12-21"),
+ _fix("00:00:00"),
+ ] + ([_fix("00:00:01")] if gga_first else [])
+
+ assert "Skipping 1 of the" in caplog.text
+ assert "(no checksum)" in caplog.text
+
+
+def test_sentences_cut_short_are_skipped_with_one_warning(tmp_path, caplog):
+ lines = []
+ for second in range(10):
+ lines += [_rmc(f"02463{second}.00"), _gga(f"02463{second}.00")]
+ # Every other fix cut short, at another field each time
+ for idx in range(1, 20, 4):
+ lines[idx] = lines[idx][: 10 + 3 * idx]
+
+ with caplog.at_level(logging.WARNING):
+ points = _read(tmp_path, *lines)
+
+ assert [p[0][11:19] for p in points] == [
+ f"02:46:3{second}" for second in range(10) if second % 2 == 1
+ ]
+ assert caplog.text.count("Skipping 5 of the 20 RMC and GGA sentences") == 1
+
+
+def test_sentences_without_checksums_are_read(tmp_path, caplog):
+ """In a log whose sentences carry no checksum, as some converters write"""
+
+ def bare(sentence: str) -> str:
+ return sentence.rsplit("*", 1)[0]
+
+ with caplog.at_level(logging.WARNING):
+ assert _read(
+ tmp_path,
+ bare(_rmc("024636.00")),
+ bare(_gga("024636.00")),
+ bare(_rmc("024637.00")),
+ bare(_gga("024637.00")),
+ ) == [_fix("02:46:36"), _fix("02:46:37")]
+
+ assert not caplog.text
+
+
+@pytest.mark.parametrize(
+ "cut, error",
+ [
+ # pynmea2 read these as 2012-02-02 and 00:00:02
+ ("$GPRMC,024637.00,A,3735.30670,N,12220.25810,W,0.0,0.0,2212", "datestamp"),
+ ("$GPGGA,00002", "timestamp"),
+ ],
+)
+def test_fields_cut_short_without_checksums_are_skipped(tmp_path, caplog, cut, error):
+ """Where no sentence has a checksum, a field cut short tells it"""
+
+ def bare(sentence: str) -> str:
+ return sentence.rsplit("*", 1)[0]
+
+ with caplog.at_level(logging.WARNING):
+ assert _read(
+ tmp_path, bare(_rmc("024636.00")), bare(_gga("024636.00")), cut
+ ) == [_fix("02:46:36")]
+
+ assert f"(malformed {error}: " in caplog.text
+
+
+def _read_bytes(tmp_path: Path, content: bytes):
+ path = tmp_path / "track.nmea"
+ path.write_bytes(content)
+ return [p.time for p in get_lat_lon_time_from_nmea(path)]
+
+
+def test_a_sentence_with_a_byte_that_is_not_text_is_skipped(tmp_path, caplog):
+ """A byte that is not UTF-8 failed the whole file"""
+ damaged = _gga("024636.50").replace(",N,", ",N,\0", 1).encode("ascii")
+ damaged = damaged.replace(b"\0", b"\xe9")
+ with caplog.at_level(logging.WARNING):
+ times = _read_bytes(
+ tmp_path,
+ b"\r\n".join(
+ [
+ _rmc("024636.00").encode("ascii"),
+ _gga("024636.00").encode("ascii"),
+ damaged,
+ _rmc("024637.00").encode("ascii"),
+ _gga("024637.00").encode("ascii"),
+ ]
+ ),
+ )
+
+ assert len(times) == 2
+ assert "Skipping 1 of the 5 RMC and GGA sentences" in caplog.text
+ assert "on line 3 (a byte that is not text)" in caplog.text
+
+
+def test_a_byte_that_is_not_text_outside_the_sentences_is_ignored(tmp_path, caplog):
+ """Such as in a header line that a camera writes before the sentences"""
+ with caplog.at_level(logging.WARNING):
+ times = _read_bytes(
+ tmp_path,
+ b"@log/Z\xfcrich/\r\n"
+ + _rmc("024636.00").encode("ascii")
+ + b"\r\n"
+ + _gga("024636.00").encode("ascii"),
+ )
+
+ assert len(times) == 1
+ assert not caplog.text
+
+
+def test_bytes_that_are_not_text_and_no_fix_fail_the_file(tmp_path):
+ """As a file that is not text did before"""
+ with pytest.raises(ValueError):
+ _read_bytes(
+ tmp_path, b"\xca\xfe\r\n" + _rmc("024636.00", status="V").encode("ascii")
+ )
+
+
@pytest.mark.parametrize(
"lines",
[
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..66395e18 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
@@ -270,7 +271,8 @@ class TestSampleTimelapseByInterval:
@classmethod
def _mock_telemetry(cls, monkeypatch, gps_seconds: T.Callable[[int], float]):
- # One point per second of video, at gps_seconds(i) from GPS_START_TIME
+ # One point per second of video, at gps_seconds(i) from GPS_START_TIME,
+ # over more than a minute, which a pace off 1x takes to measure
points = [
telemetry.GPSPoint(
time=float(i),
@@ -283,7 +285,7 @@ def _mock_telemetry(cls, monkeypatch, gps_seconds: T.Callable[[int], float]):
precision=None,
ground_speed=None,
)
- for i in range(60)
+ for i in range(70)
]
monkeypatch.setattr(
sample_video,
@@ -511,15 +513,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):
@@ -539,11 +595,130 @@ def test_not_located_without_a_track(self, tmpdir, setup_mock, monkeypatch, capl
in caplog.text
)
+ def _file_points(self, video_times, gps_offset: float = 100.0) -> list[geo.Point]:
+ """
+ The points of a track from a file as GPX sync places them on the
+ video, at video times, keeping their own GPS times
+ """
+ return [
+ telemetry.CAMMGPSPoint(
+ time=float(t),
+ lat=40.0 + t * 0.001,
+ lon=-74.0,
+ alt=None,
+ angle=None,
+ epoch_time=self.GPS_START_TIME.timestamp() + gps_offset + t,
+ gps_fix_type=2,
+ horizontal_accuracy=0.0,
+ vertical_accuracy=0.0,
+ velocity_east=0.0,
+ velocity_north=0.0,
+ velocity_up=0.0,
+ speed_accuracy=0.0,
+ )
+ for t in video_times
+ ]
+
+ @pytest.mark.parametrize("source", ["gpx", "nmea"])
+ def test_located_along_a_video_track_file(
+ self, tmpdir, setup_mock, monkeypatch, caplog, source
+ ):
+ """
+ Along the track of --video_geotag_source as GPX sync placed it on the
+ video, and stamped with its times, as distance samples are. It ends 40
+ seconds into the video
+ """
+ track = self._file_points(range(41))
+ with caplog.at_level(logging.WARNING, logger=sample_video.LOG.name):
+ samples = self._sample(
+ tmpdir,
+ monkeypatch,
+ track,
+ video_geotag_source=[source],
+ geotag_source_path=Path(tmpdir) / f"track.{source}",
+ )
+
+ assert len(samples) == 21
+ _validate_interval(
+ samples, self.GPS_START_TIME + datetime.timedelta(seconds=100)
+ )
+ for idx, sample in enumerate(samples):
+ lon_lat = exif_read.ExifRead(sample).extract_lon_lat()
+ assert lon_lat == pytest.approx((-74.0, 40.0 + 2 * idx * 0.001))
+ [options] = self.track_calls
+ assert [o.source.value for o in options] == [source]
+ assert (
+ "9 of the 30 samples of hello.mp4 fall outside its GPS track "
+ "(2021-08-10T06:39:46+00:00 to 2021-08-10T06:40:26+00:00) and were left out"
+ in caplog.text
+ )
+
+ def test_samples_in_a_gap_of_a_video_track_file_are_left_out(
+ self, tmpdir, setup_mock, monkeypatch
+ ):
+ monkeypatch.setattr(sample_video.constants, "MAX_GPX_GAP_SECONDS", 5.0)
+ track = self._file_points([*range(11), *range(20, 61)])
+
+ samples = self._sample(
+ tmpdir,
+ monkeypatch,
+ track,
+ video_geotag_source=["gpx"],
+ geotag_source_path=Path(tmpdir) / "track.gpx",
+ )
+
+ # Not those at 12, 14, 16 and 18 seconds
+ assert [s.name for s in samples] == [
+ f"hello_v_{idx + 1:06d}.jpg" for idx in [*range(6), *range(10, 30)]
+ ]
+
+ def test_a_video_track_file_started_with_the_video_starts_at_the_first_sample(
+ self, tmpdir, setup_mock, monkeypatch
+ ):
+ """
+ GPX sync starts it at the first frame, 0.5 seconds into the video, and
+ ffmpeg cuts the first interval sample at time 0
+ """
+ monkeypatch.setattr(gpx_extractor, "_first_frame_time", lambda path: 0.5)
+ track = self._file_points([0.5 + t for t in range(61)], gps_offset=-0.5)
+
+ samples = self._sample(
+ tmpdir,
+ monkeypatch,
+ track,
+ video_geotag_source=["gpx"],
+ geotag_source_path=Path(tmpdir) / "track.gpx",
+ interpolation_use_gpx_start_time=True,
+ )
+
+ assert len(samples) == 30
+ _validate_interval(samples, self.GPS_START_TIME)
+ lon_lat = exif_read.ExifRead(samples[0]).extract_lon_lat()
+ assert lon_lat == pytest.approx((-74.0, 40.0005))
+
+ def test_none_left_out_of_a_video_track_file_where_the_video_source_may_geotag_them(
+ self, tmpdir, setup_mock, monkeypatch
+ ):
+ """The legacy gopro source geotags the samples by the GPS of the video"""
+ track = self._file_points(range(41))
+
+ samples = self._sample(
+ tmpdir,
+ monkeypatch,
+ track,
+ video_geotag_source=["gpx"],
+ geotag_source=["gopro"],
+ geotag_source_path=Path(tmpdir) / "track.gpx",
+ )
+
+ assert len(samples) == 30
+
@pytest.mark.parametrize("source", [SourceType.GPX, SourceType.NMEA])
def test_not_located_with_a_track_file(self, monkeypatch, source):
"""
- The geotagging stage places the samples along a GPX or NMEA track by
- their capture times, which interval sampling does not place the track by
+ The geotagging stage places the samples along a GPX or NMEA track of
+ --geotag_source by their capture times, which interval sampling does not
+ place the track by
"""
process = mock.Mock()
monkeypatch.setattr(sample_video.factory, "process", process)
@@ -606,13 +781,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(
' None:
gps_start_time + datetime.timedelta(seconds=seconds) for seconds in range(6)
]
+ GPS_START_TIME = datetime.datetime(
+ 2021, 8, 10, 6, 38, 6, tzinfo=datetime.timezone.utc
+ )
+
+ def _file_track(self, video_times, rate: float = 1.0) -> list[geo.Point]:
+ """
+ The points of a track from a GPX (or NMEA) file as GPX sync places
+ them on the video, at video times, keeping their own GPS times, of
+ which a timelapse packs **rate** into each second of video
+ """
+ return [
+ telemetry.CAMMGPSPoint(
+ time=float(t),
+ lat=40.0 + t * 0.001,
+ lon=-74.0,
+ alt=None,
+ angle=None,
+ epoch_time=self.GPS_START_TIME.timestamp() + t * rate,
+ gps_fix_type=2,
+ horizontal_accuracy=0.0,
+ vertical_accuracy=0.0,
+ velocity_east=0.0,
+ velocity_north=0.0,
+ velocity_up=0.0,
+ speed_accuracy=0.0,
+ )
+ for t in video_times
+ ]
+
+ def _sample_along_a_file_track(
+ self, tmp_path: Path, gps_points: T.Sequence[geo.Point], **kwargs
+ ) -> list[datetime.datetime | None]:
+ """Sample a frame a second by distance, and read their capture times"""
+ video_file = tmp_path / "test.mp4"
+ video_file.touch()
+ output_dir = tmp_path / "output"
+ mocks = self._setup_mocks(
+ tmp_path,
+ video_file,
+ gps_points=gps_points,
+ sample_times=[float(t) for t in range(int(gps_points[-1].time) + 1)],
+ )
+
+ with (
+ mocks["patches"]["ffmpeg_cls"],
+ mocks["patches"]["geotag_cls"],
+ mocks["patches"]["moov_parse"],
+ ):
+ sample_video.sample_video(
+ video_import_path=video_file,
+ import_path=output_dir,
+ video_sample_distance=0.0,
+ geotag_source_path=tmp_path / "track.gpx",
+ **kwargs,
+ )
+
+ frames = sorted((output_dir / "test.mp4").glob("*.jpg"))
+ return [exif_read.ExifRead(frame).extract_capture_time() for frame in frames]
+
+ @pytest.mark.parametrize("source", ["gpx", "nmea"])
+ def test_frames_in_a_gap_of_a_video_track_file_are_left_out(
+ self, tmp_path: Path, monkeypatch, caplog, source
+ ) -> None:
+ """
+ The track of --video_geotag_source has a gap of 10 seconds, over the
+ limit, from 10 to 20 seconds into the video. The frames in it were
+ placed across it, as they would be outside the track
+ """
+ monkeypatch.setattr(sample_video.constants, "MAX_GPX_GAP_SECONDS", 5.0)
+ video_times = [*range(11), *range(20, 31)]
+
+ with caplog.at_level(logging.WARNING, logger=sample_video.LOG.name):
+ capture_times = self._sample_along_a_file_track(
+ tmp_path, self._file_track(video_times), video_geotag_source=[source]
+ )
+
+ assert capture_times == [
+ self.GPS_START_TIME + datetime.timedelta(seconds=t) for t in video_times
+ ]
+ assert (
+ "9 frames of test.mp4 fall in a gap of its GPS track and were left out: "
+ "a gap of 10 seconds, from 2021-08-10T06:38:16+00:00 at (40.01, -74.0) "
+ "to 2021-08-10T06:38:26+00:00 at (40.02, -74.0)"
+ ) in caplog.text
+
+ def test_the_gap_limit_is_in_gps_seconds_in_a_timelapse(
+ self, tmp_path: Path, monkeypatch
+ ) -> None:
+ """
+ The gap of 2 seconds of video, from 10 to 12 seconds into it, spans 20
+ seconds of GPS time at 10x, over the limit
+ """
+ monkeypatch.setattr(sample_video.constants, "MAX_GPX_GAP_SECONDS", 15.0)
+ video_times = [*range(11), *range(12, 21)]
+
+ capture_times = self._sample_along_a_file_track(
+ tmp_path,
+ self._file_track(video_times, rate=10.0),
+ video_geotag_source=["gpx"],
+ )
+
+ assert capture_times == [
+ self.GPS_START_TIME + datetime.timedelta(seconds=t * 10)
+ for t in video_times
+ ]
+
+ @pytest.mark.parametrize(
+ "kwargs",
+ [
+ # The geotagging stage places the frames along the track of a file
+ # of --geotag_source by their capture times
+ {"geotag_source": ["gpx"]},
+ # The legacy gopro source geotags them by the GPS of their video
+ {"video_geotag_source": ["gpx"], "geotag_source": ["gopro"]},
+ ],
+ )
+ def test_frames_in_a_gap_are_left_where_the_geotagging_stage_places_them(
+ self, tmp_path: Path, monkeypatch, kwargs
+ ) -> None:
+ monkeypatch.setattr(sample_video.constants, "MAX_GPX_GAP_SECONDS", 5.0)
+
+ capture_times = self._sample_along_a_file_track(
+ tmp_path, self._file_track([*range(11), *range(20, 31)]), **kwargs
+ )
+
+ assert capture_times == [
+ self.GPS_START_TIME + datetime.timedelta(seconds=t) for t in range(31)
+ ]
+
def test_timelapse_frames_past_the_end_are_stamped_within_the_track(
self, tmp_path: Path
) -> None: