diff --git a/mapillary_tools/exif_read.py b/mapillary_tools/exif_read.py
index 8953d13c..08ecbd8c 100644
--- a/mapillary_tools/exif_read.py
+++ b/mapillary_tools/exif_read.py
@@ -320,13 +320,15 @@ def make_valid_timezone_offset(delta: datetime.timedelta) -> datetime.timedelta:
# No image was captured before 1970. Cameras without a clock write placeholder
# dates before it: 0001:01:01 00:00:00, a date in 1904 (the Mac epoch), or
# 1969:12:31 19:00:00 (Unix time 0 in a time zone west of UTC)
-_MIN_CAPTURE_YEAR = 1970
+_MIN_CAPTURE_TIME = datetime.datetime(1970, 1, 1)
def is_valid_capture_time(dt: datetime.datetime) -> bool:
"""
Whether **dt** can be the time an image was captured, rather than a
- placeholder date
+ placeholder date: from 1970 on, and before year 10000, which no date can
+ be written in. A time with a time zone offset is judged in UTC, one
+ without as it is
>>> is_valid_capture_time(datetime.datetime(2021, 7, 15, 15, 37, 30))
True
@@ -334,14 +336,63 @@ def is_valid_capture_time(dt: datetime.datetime) -> bool:
False
>>> is_valid_capture_time(datetime.datetime(1969, 12, 31, 19, 0))
False
+
+ 1969-12-31 22:30 in UTC, and 1970-01-01 00:30:
+
+ >>> plus_2h = datetime.timezone(datetime.timedelta(hours=2))
+ >>> is_valid_capture_time(datetime.datetime(1970, 1, 1, 0, 30, tzinfo=plus_2h))
+ False
+ >>> minus_1h = datetime.timezone(datetime.timedelta(hours=-1))
+ >>> is_valid_capture_time(datetime.datetime(1969, 12, 31, 23, 30, tzinfo=minus_1h))
+ True
+
+ Year 10000 in UTC:
+
+ >>> minus_5h = datetime.timezone(datetime.timedelta(hours=-5))
+ >>> is_valid_capture_time(datetime.datetime(9999, 12, 31, 23, 0, tzinfo=minus_5h))
+ False
"""
- return _MIN_CAPTURE_YEAR <= dt.year
+ utc = dt.replace(tzinfo=None)
+ offset = dt.utcoffset()
+ if offset is not None:
+ try:
+ utc -= offset
+ except OverflowError:
+ # Before year 1 or after year 9999 in UTC
+ return False
+ return _MIN_CAPTURE_TIME <= utc
_FIELD_TYPE = T.TypeVar("_FIELD_TYPE", int, float, str)
class ExifReadABC(abc.ABC):
+ # The first time that a date tag gave but that is no valid capture time
+ # (see is_valid_capture_time()), which tells why an image has no capture
+ # time when no other date tag gives one
+ rejected_capture_time: datetime.datetime | None = None
+
+ def _accept_capture_time(
+ self, dt: datetime.datetime | None
+ ) -> datetime.datetime | None:
+ """
+ **dt** if it is a valid capture time, otherwise None, so that the next
+ date tag is tried
+
+ >>> exif = ExifReadFromXMP(et.ElementTree(et.fromstring("")))
+ >>> exif._accept_capture_time(datetime.datetime(1, 1, 1)) is None
+ True
+ >>> exif._accept_capture_time(datetime.datetime(1904, 1, 1)) is None
+ True
+ >>> exif.rejected_capture_time
+ datetime.datetime(1, 1, 1, 0, 0)
+ """
+ if dt is None or is_valid_capture_time(dt):
+ return dt
+ if self.rejected_capture_time is None:
+ self.rejected_capture_time = dt
+ return None
+
@abc.abstractmethod
def extract_altitude(self) -> float | None:
raise NotImplementedError
@@ -412,10 +463,8 @@ def _extract_exif_datetime(
subsec = subsec.replace(" ", "0")
offset = self._extract_alternative_fields([offset_tag], str)
dt = parse_datetimestr_with_subsec_and_offset(dtstr, subsec, offset)
- if dt is None or not is_valid_capture_time(dt):
- # A placeholder date: let the next date tag be tried
- return None
- return dt
+ # A placeholder date lets the next date tag be tried
+ return self._accept_capture_time(dt)
def extract_exif_datetime(self) -> datetime.datetime | None:
dt = self._extract_exif_datetime(
@@ -458,12 +507,10 @@ def extract_gps_datetime(self) -> datetime.datetime | None:
def extract_capture_time(self) -> datetime.datetime | None:
dt = self.extract_gps_datetime()
- if (
- dt is not None
- and dt.date() != datetime.date(1970, 1, 1)
- and is_valid_capture_time(dt)
- ):
- return dt
+ if dt is not None and dt.date() != datetime.date(1970, 1, 1):
+ dt = self._accept_capture_time(dt)
+ if dt is not None:
+ return dt
dt = self.extract_exif_datetime()
if dt is not None:
@@ -764,10 +811,8 @@ def _extract_exif_datetime(
subsec = subsec.replace(" ", "0")
offset = self._extract_alternative_fields([offset_tag], field_type=str)
dt = parse_datetimestr_with_subsec_and_offset(dtstr, subsec, offset)
- if dt is None or not is_valid_capture_time(dt):
- # A placeholder date: let the next date tag be tried
- return None
- return dt
+ # A placeholder date lets the next date tag be tried
+ return self._accept_capture_time(dt)
def extract_exif_datetime(self) -> datetime.datetime | None:
# EXIF DateTimeOriginal: 0x9003 (date/time when original image was taken)
@@ -813,12 +858,10 @@ def extract_capture_time(self) -> datetime.datetime | None:
gps_dt = self.extract_gps_datetime()
except (ValueError, TypeError, ZeroDivisionError):
gps_dt = None
- if (
- gps_dt is not None
- and gps_dt.date() != datetime.date(1970, 1, 1)
- and is_valid_capture_time(gps_dt)
- ):
- return gps_dt
+ if gps_dt is not None and gps_dt.date() != datetime.date(1970, 1, 1):
+ gps_dt = self._accept_capture_time(gps_dt)
+ if gps_dt is not None:
+ return gps_dt
dt = self.extract_exif_datetime()
if dt is not None:
@@ -1049,6 +1092,8 @@ def extract_capture_time(self) -> datetime.datetime | None:
val = xmp.extract_capture_time()
if val is not None:
return val
+ if self.rejected_capture_time is None:
+ self.rejected_capture_time = xmp.rejected_capture_time
return None
def extract_lon_lat(self) -> tuple[float, float] | None:
diff --git a/mapillary_tools/exiftool_read.py b/mapillary_tools/exiftool_read.py
index f2bb6a1e..f8226e1a 100644
--- a/mapillary_tools/exiftool_read.py
+++ b/mapillary_tools/exiftool_read.py
@@ -188,10 +188,8 @@ def _extract_exif_datetime(
subsec = subsec.replace(" ", "0")
offset = self._extract_alternative_fields(offset_tags, str)
dt = exif_read.parse_datetimestr_with_subsec_and_offset(dtstr, subsec, offset)
- if dt is None or not exif_read.is_valid_capture_time(dt):
- # A placeholder date: let the next date tag be tried
- return None
- return dt
+ # A placeholder date lets the next date tag be tried
+ return self._accept_capture_time(dt)
def extract_exif_datetime_from_xmp(self) -> datetime.datetime | None:
# EXIF DateTimeOriginal: 0x9003 (date/time when original image was taken)
@@ -278,23 +276,19 @@ def extract_capture_time(self) -> datetime.datetime | None:
dt = self.extract_gps_datetime()
except (ValueError, TypeError, ZeroDivisionError):
dt = None
- if (
- dt is not None
- and dt.date() != datetime.date(1970, 1, 1)
- and exif_read.is_valid_capture_time(dt)
- ):
- return dt
+ if dt is not None and dt.date() != datetime.date(1970, 1, 1):
+ dt = self._accept_capture_time(dt)
+ if dt is not None:
+ return dt
try:
dt = self.extract_gps_datetime_from_xmp()
except (ValueError, TypeError, ZeroDivisionError):
dt = None
- if (
- dt is not None
- and dt.date() != datetime.date(1970, 1, 1)
- and exif_read.is_valid_capture_time(dt)
- ):
- return dt
+ if dt is not None and dt.date() != datetime.date(1970, 1, 1):
+ dt = self._accept_capture_time(dt)
+ if dt is not None:
+ return dt
dt = self.extract_exif_datetime()
if dt is not None:
diff --git a/mapillary_tools/geotag/factory.py b/mapillary_tools/geotag/factory.py
index 1071dfbb..87c0016d 100644
--- a/mapillary_tools/geotag/factory.py
+++ b/mapillary_tools/geotag/factory.py
@@ -152,10 +152,15 @@ def parse_source_options(
return parsed_options
-def _give_source_path(options: T.Sequence[SourceOption], source_path: Path) -> None:
+def _give_source_path(
+ options: T.Sequence[SourceOption],
+ source_path: Path,
+ given_by_the_user: bool = True,
+) -> None:
"""
Give **source_path** to the options without a source path of their own,
- keeping their patterns
+ keeping their patterns. Warn that the options with one ignore it, if it is
+ the --geotag_source_path that the user gave
"""
for option in options:
if option.source_path is None:
@@ -164,7 +169,7 @@ def _give_source_path(options: T.Sequence[SourceOption], source_path: Path) -> N
source_path_option = option.source_path
if source_path_option.source_path is None:
source_path_option.source_path = Path(source_path)
- else:
+ elif given_by_the_user:
LOG.warning(
"The option --geotag_source_path is ignored for source %s",
option,
@@ -183,7 +188,9 @@ def give_video_import_path(
for option in options:
if option.source is SourceType.EXIFTOOL_XML and option.source_path is None:
option.source_path = exiftool_xml_default_source_path(video_import_path)
- _give_source_path(options, video_import_path)
+ # Without --geotag_source_path: a source with a path of its own ignores the
+ # video import path without a word
+ _give_source_path(options, video_import_path, given_by_the_user=False)
def exiftool_xml_default_source_path(video_import_path: Path) -> SourcePathOption:
@@ -257,6 +264,8 @@ def process(
options: T.Collection[SourceOption],
# --filetypes, which the filetypes of every option are limited to already
filetypes: set[types.FileType] | None = None,
+ # If given, filled with the option that geotagged each file, of those geotagged
+ sources: dict[Path, SourceOption] | None = None,
) -> list[types.MetadataOrError]:
if not options:
raise ValueError("No geotag options provided")
@@ -378,10 +387,11 @@ def process(
reprocessable_paths.remove(metadata.filename)
pending_errors.pop(metadata.filename, None)
rejection = gpx_rejections.pop(metadata.filename, None)
- if rejection is not None and not isinstance(
- metadata, types.ErrorMetadata
- ):
- rescued.append((metadata, *rejection))
+ if not isinstance(metadata, types.ErrorMetadata):
+ if sources is not None:
+ sources[metadata.filename] = option
+ if rejection is not None:
+ rescued.append((metadata, *rejection))
_warn_geotagged_despite_gpx(rescued, option)
diff --git a/mapillary_tools/geotag/geotag_images_from_gpx.py b/mapillary_tools/geotag/geotag_images_from_gpx.py
index aafa7cf6..6fbe81b3 100644
--- a/mapillary_tools/geotag/geotag_images_from_gpx.py
+++ b/mapillary_tools/geotag/geotag_images_from_gpx.py
@@ -28,6 +28,23 @@
LOG = logging.getLogger(__name__)
+def _format_time(time: float) -> str:
+ """
+ The time as a description file writes it, or the Unix time itself if it
+ cannot be written as a date (a time offset can move it past year 9999),
+ so that the error that names it is still reported
+
+ >>> _format_time(0)
+ '1970_01_01_00_00_00_000'
+ >>> _format_time(1e12)
+ '1000000000000.0'
+ """
+ try:
+ return build_capture_time(time)
+ except exceptions.MapillaryMetadataValidationError:
+ return str(time)
+
+
def check_within_track(
time: float,
sorted_points: T.Sequence[geo.Point],
@@ -60,7 +77,7 @@ def check_within_track(
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),
+ image_time=_format_time(time),
gpx_start_time=gpx_start_time,
gpx_end_time=gpx_end_time,
)
@@ -73,7 +90,7 @@ def check_within_track(
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),
+ image_time=_format_time(time),
gpx_start_time=gpx_start_time,
gpx_end_time=gpx_end_time,
)
@@ -91,7 +108,7 @@ def check_within_track(
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),
+ image_time=_format_time(time),
gpx_start_time=build_capture_time(sorted_points[0].time),
gpx_end_time=build_capture_time(sorted_points[-1].time),
)
diff --git a/mapillary_tools/geotag/geotag_images_from_gpx_file.py b/mapillary_tools/geotag/geotag_images_from_gpx_file.py
index 77f7ae42..5fb022f3 100644
--- a/mapillary_tools/geotag/geotag_images_from_gpx_file.py
+++ b/mapillary_tools/geotag/geotag_images_from_gpx_file.py
@@ -48,12 +48,15 @@ def _read_gpx_points(source_path: Path) -> list[geo.Point]:
geotagged from it, and if it is missing or has no timestamped point: the
file was given to geotag the images from, so it fails them rather than
leaving them to the next geotag source.
+
+ The file is parsed once in a process (see utils.read_gpx_tracks()), and
+ its points are shared with its other readers, which must not change them.
"""
if not source_path.is_file():
raise exceptions.MapillaryInvalidGPXError(f"GPX file not found: {source_path}")
try:
- tracks = utils.parse_gpx(source_path)
+ tracks = utils.read_gpx_tracks(source_path)
except (OSError, UnicodeError, gpxpy.gpx.GPXException) as ex:
raise exceptions.MapillaryInvalidGPXError(
f"{source_path.name} is not a valid GPX file. "
@@ -61,12 +64,14 @@ def _read_gpx_points(source_path: Path) -> list[geo.Point]:
) from ex
if 1 < len(tracks):
- LOG.warning(
+ utils.warn_once(
+ source_path,
+ ("tracks", len(tracks)),
"Found %s tracks in the GPX file %s. Will merge points in all the tracks as a single track for interpolation",
len(tracks),
source_path,
)
- points: list[geo.Point] = sum(tracks, [])
+ points: list[geo.Point] = [point for track in tracks for point in track]
if not points:
raise exceptions.MapillaryInvalidGPXError(
diff --git a/mapillary_tools/geotag/geotag_images_from_nmea_file.py b/mapillary_tools/geotag/geotag_images_from_nmea_file.py
index 7d735ad0..e6b210dc 100644
--- a/mapillary_tools/geotag/geotag_images_from_nmea_file.py
+++ b/mapillary_tools/geotag/geotag_images_from_nmea_file.py
@@ -37,9 +37,22 @@
# What a byte that is not text decodes to
_NOT_TEXT = "\ufffd"
-# The checksum that ends a sentence: *HH
+# The byte order mark, which some editors write at the start of a UTF-8 file
+_BOM = "\ufeff"
+
+# How much of a file tells whether it is text at all, before it is read whole:
+# a video given as an NMEA file is gigabytes
+_SNIFF_BYTES = 64 * 1024
+
+# The checksum that ends a sentence: *HH. It is optional
_CHECKSUM_RE = re.compile(r"\*[0-9A-Fa-f]{2}\s*$")
+# The fields of a complete sentence: an RMC sentence has 11 (NMEA 2.0 to
+# 2.2), 12 with the mode of NMEA 2.3, and 13 with the navigational status of
+# NMEA 4.1; a GGA sentence has 14. One without its checksum that has fewer was
+# cut short, such as the last one of a log when the recording stopped
+_MIN_FIELDS = {"RMC": 11, "GGA": 14}
+
# The time of a sentence (hhmmss, and a fraction of a second), and the date of
# an RMC sentence (ddmmyy). pynmea2 reads a field cut short as another time or
# date: an RMC sentence cut inside its date "2212" as 2012-02-02
@@ -85,8 +98,7 @@ def _read_nmea_points(source_path: Path) -> list[geo.Point]:
# UnboundLocalError: GGA sentences but no RMC sentence to give their date
except (OSError, ValueError, TypeError, UnboundLocalError) as ex:
raise exceptions.MapillaryInvalidGPXError(
- f"{source_path.name} is not a valid NMEA file. "
- "Replace it with a valid NMEA file."
+ describe_invalid_nmea(source_path, ex)
) from ex
if not points:
@@ -97,6 +109,29 @@ def _read_nmea_points(source_path: Path) -> list[geo.Point]:
return points
+def describe_invalid_nmea(nmea_file: Path, ex: Exception) -> str:
+ """
+ The message of the error for the NMEA file **nmea_file** that reading it
+ failed with **ex**, which says why
+
+ >>> describe_invalid_nmea(Path("a/track.nmea"), ValueError("it is not text"))
+ 'track.nmea is not a valid NMEA file: it is not text. Replace it with a valid NMEA file.'
+ >>> describe_invalid_nmea(Path("track.nmea"), PermissionError(13, "Permission denied"))
+ 'track.nmea is not a valid NMEA file: Permission denied. Replace it with a valid NMEA file.'
+ """
+ if isinstance(ex, OSError):
+ reason = ex.strerror or str(ex)
+ elif isinstance(ex, ValueError):
+ reason = str(ex)
+ else:
+ reason = ""
+ because = f": {reason}" if reason else ""
+ return (
+ f"{nmea_file.name} is not a valid NMEA file{because}. "
+ "Replace it with a valid NMEA file."
+ )
+
+
def get_lat_lon_time_from_nmea(nmea_file: Path) -> list[geo.Point]:
"""
Read the fixes of an NMEA file: the time and position of each GGA
@@ -117,13 +152,21 @@ def get_lat_lon_time_from_nmea(nmea_file: Path) -> list[geo.Point]:
time.
A malformed sentence, such as one whose checksum does not match, one cut
- short, or one with a byte that is not text, is skipped with a warning. A
- file of which no fix can be read raises ValueError if it has malformed
- sentences or bytes that are not text.
+ short, or one with a byte that is not text, is skipped with a warning.
+ Checksums are optional: a sentence without one is cut short only if its
+ own fields show it. A file of which no fix can be read raises ValueError,
+ which says why, if it has malformed sentences or bytes that are not text.
+
+ A file that is not text at all, which its first bytes show, raises
+ ValueError without being read further.
A file is read once in a process, however many times it is asked for.
"""
- data = nmea_file.read_bytes()
+ with nmea_file.open("rb") as f:
+ head = f.read(_SNIFF_BYTES)
+ if not _looks_like_text(head):
+ raise ValueError("it is not text")
+ data = head + f.read()
key = (str(nmea_file.resolve()), hashlib.sha256(data).hexdigest())
read = _READ.get(key)
if read is None:
@@ -139,6 +182,27 @@ def get_lat_lon_time_from_nmea(nmea_file: Path) -> list[geo.Point]:
return [dataclasses.replace(p) for p in read]
+def _looks_like_text(head: bytes) -> bool:
+ r"""
+ Whether the first bytes **head** of a file can be those of a text file:
+ without NUL bytes, and mostly bytes that decode as UTF-8, where a log may
+ have a few bytes damaged
+
+ >>> _looks_like_text(b"$GPRMC,024636.00,A,3735.30670,N\r\n")
+ True
+ >>> _looks_like_text(b"$GPRMC,024636.00,\xe9,3735.30670,N\r\n")
+ True
+ >>> _looks_like_text(b"\x00\x00\x00\x18ftypmp42")
+ False
+ >>> _looks_like_text(b"\xca\xfe\xba\xbe" * 100)
+ False
+ """
+ if b"\x00" in head:
+ return False
+ text = head.decode("utf-8", errors="replace")
+ return text.count(_NOT_TEXT) * 2 <= len(text)
+
+
def _read_nmea_data(nmea_file: Path, data: bytes) -> list[geo.Point]:
"""
Read the fixes of the content **data** of an NMEA file, as
@@ -146,10 +210,14 @@ def _read_nmea_data(nmea_file: Path, data: bytes) -> list[geo.Point]:
"""
# 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 io.TextIOWrapper(io.BytesIO(data), encoding="utf-8", errors="replace") as f:
+ # such as one damaged in a sentence, as U+FFFD (see _parse_sentence()).
+ # Some editors on Windows start a file with a byte order mark, which
+ # "utf-8-sig" leaves out; so is one that starts a log appended to another
+ with io.TextIOWrapper(
+ io.BytesIO(data), encoding="utf-8-sig", errors="replace"
+ ) as f:
lines = f.readlines()
- lines = [line.rstrip("\n\r") for line in lines]
+ lines = [line.rstrip("\n\r").lstrip(_BOM) for line in lines]
not_text = any(_NOT_TEXT in line for line in lines)
# (line number, line) of each RMC and GGA sentence
@@ -158,29 +226,28 @@ def _read_nmea_data(nmea_file: Path, data: bytes) -> list[geo.Point]:
for number, line in enumerate(lines, start=1)
if _SENTENCE_RE.search(line)
]
- # Receivers end each sentence with its checksum, or none of them. One
- # without, where the others have theirs, was cut short, such as the last
- # one of a log when the recording stopped
- checksummed = any(_CHECKSUM_RE.search(line) for _, line in sentences)
messages = []
# (line number, line, error) of each malformed sentence
malformed: list[tuple[int, str, ValueError]] = []
for number, line in sentences:
try:
- messages.append(_parse_sentence(line, checksummed=checksummed))
+ messages.append(_parse_sentence(line))
except ValueError as ex:
malformed.append((number, line, ex))
points = _read_fixes(messages, nmea_file)
if not points and not_text and not malformed:
- raise ValueError(f"{nmea_file} is not text")
+ raise ValueError("it is not text")
if malformed:
number, line, error = malformed[0]
if not points:
- raise error
+ raise ValueError(
+ f"{len(malformed)} of its {len(sentences)} RMC and GGA sentences are malformed, "
+ f"the first on line {number} ({_quote(str(error))}): {_quote(line)}"
+ ) from error
LOG.warning(
"Skipping %d of the %d RMC and GGA sentences of the NMEA file %s, which are malformed, the first on line %d (%s): %s",
len(malformed),
@@ -259,14 +326,14 @@ def _read_fixes(
return points
-def _parse_sentence(line: str, checksummed: bool = False) -> pynmea2.NMEASentence:
+def _parse_sentence(line: str) -> pynmea2.NMEASentence:
"""
Parse an NMEA sentence, raising ValueError if it is malformed: if it does
- not parse or its checksum does not match, if it has no checksum but should
- (**checksummed**), if its time or the date of an RMC sentence is not
- hhmmss[.ss] or ddmmyy, or if a field that a fix is read from does not
- parse, which pynmea2 leaves a string, or if it has a byte that is not
- text, decoded as U+FFFD
+ not parse or its checksum does not match, if it has no checksum and fewer
+ fields than a complete sentence (see _MIN_FIELDS), if its time or the date
+ of an RMC sentence is not hhmmss[.ss] or ddmmyy, or if a field that a fix
+ is read from does not parse, which pynmea2 leaves a string, or if it has a
+ byte that is not text, decoded as U+FFFD
>>> _parse_sentence("$GPGGA,0246,3735.30670,N,12220.25810,W,1,08,1.0,0.0,M,0.0,M,,*64")
Traceback (most recent call last):
@@ -276,21 +343,24 @@ def _parse_sentence(line: str, checksummed: bool = False) -> pynmea2.NMEASentenc
Traceback (most recent call last):
...
ValueError: checksum does not match: 00 != 4F
- >>> _parse_sentence("$GPRMC,024636.00,A,3735.30670,N,12220.25810,W,0.0,0.0,2212", checksummed=True)
- Traceback (most recent call last):
- ...
- ValueError: no checksum
>>> _parse_sentence("$GPRMC,024636.00,A,3735.30670,N,12220.25810,W,0.0,0.0,2212")
Traceback (most recent call last):
...
ValueError: malformed datestamp: '2212'
+ >>> _parse_sentence("$GPRMC,024636.00,A,3735.30670,N,12220.25810,W,0.0,0.0,221222")
+ Traceback (most recent call last):
+ ...
+ ValueError: cut short: 9 of the 11 fields of an RMC sentence
+ >>> _parse_sentence("$GPGGA,024636.00,3735.30670,N,12220.25810,W,1,08,1.0,12")
+ Traceback (most recent call last):
+ ...
+ ValueError: cut short: 9 of the 14 fields of a GGA sentence
+ >>> type(_parse_sentence("$GPGGA,024636.00,3735.30670,N,12220.25810,W,1,08,1.0,12.5,M,0.0,M,,")).__name__
+ 'GGA'
"""
if _NOT_TEXT in line:
raise ValueError("a byte that is not text")
- if checksummed and not _CHECKSUM_RE.search(line):
- raise ValueError("no checksum")
-
try:
message = pynmea2.parse(line)
except pynmea2.ParseError as ex:
@@ -306,6 +376,15 @@ def _parse_sentence(line: str, checksummed: bool = False) -> pynmea2.NMEASentenc
if value and not pattern.fullmatch(value):
raise ValueError(f"malformed {name}: {value!r}")
+ # A sentence with its checksum ended where it ends: pynmea2 checks it
+ if not _CHECKSUM_RE.search(line):
+ kind = message.sentence_type
+ if kind in _MIN_FIELDS and len(message.data) < _MIN_FIELDS[kind]:
+ article = "an" if kind == "RMC" else "a"
+ raise ValueError(
+ f"cut short: {len(message.data)} of the {_MIN_FIELDS[kind]} fields of {article} {kind} sentence"
+ )
+
fields: dict[str, type] = {}
if isinstance(message, pynmea2.RMC):
fields = {"timestamp": datetime.time, "datestamp": datetime.date}
diff --git a/mapillary_tools/geotag/geotag_videos_from_gpx.py b/mapillary_tools/geotag/geotag_videos_from_gpx.py
index 2f34418d..859a95c9 100644
--- a/mapillary_tools/geotag/geotag_videos_from_gpx.py
+++ b/mapillary_tools/geotag/geotag_videos_from_gpx.py
@@ -5,6 +5,7 @@
from __future__ import annotations
+import collections
import logging
import sys
import typing as T
@@ -16,7 +17,7 @@
from typing_extensions import override
from .. import exceptions, types
-from . import options
+from . import options, utils
from .base import GeotagVideosFromGeneric
from .video_extractors.gpx import GPXVideoExtractor
@@ -72,4 +73,28 @@ def _generate_video_extractors(
filetype=types.FileType.VIDEO,
)
)
+
+ # The videos are geotagged in a pool of processes, each of which would
+ # read the track file of its videos. Read here, once, and warn about
+ # once, a file that geotags several of them, or that this process read
+ # already (see _read_here())
+ extractors = [
+ result for result in results if isinstance(result, GPXVideoExtractor)
+ ]
+ num_videos = collections.Counter(
+ extractor.gpx_path.resolve() for extractor in extractors
+ )
+ for extractor in extractors:
+ if self._read_here(extractor, num_videos[extractor.gpx_path.resolve()]):
+ extractor.read_tracks()
+
return results
+
+ def _read_here(self, extractor: GPXVideoExtractor, num_videos: int) -> bool:
+ """
+ Whether to read the track file of **extractor**, which geotags
+ **num_videos** videos, in this process rather than in the pool: if it
+ geotags several, or was parsed in this process already. A file of
+ each video is left to the pool, which parses them side by side
+ """
+ return 1 < num_videos or utils.was_gpx_read(extractor.gpx_path)
diff --git a/mapillary_tools/geotag/geotag_videos_from_nmea.py b/mapillary_tools/geotag/geotag_videos_from_nmea.py
index 74d1efe4..6ff7fd8e 100644
--- a/mapillary_tools/geotag/geotag_videos_from_nmea.py
+++ b/mapillary_tools/geotag/geotag_videos_from_nmea.py
@@ -6,15 +6,12 @@
from __future__ import annotations
import sys
-import typing as T
-from pathlib import Path
if sys.version_info >= (3, 12):
from typing import override
else:
from typing_extensions import override
-from .. import types
from .geotag_videos_from_gpx import GeotagVideosFromGPX
from .video_extractors.gpx import GPXVideoExtractor
from .video_extractors.nmea import NMEAVideoExtractor
@@ -30,14 +27,7 @@ class GeotagVideosFromNMEA(GeotagVideosFromGPX):
default_pattern = "%g.nmea"
@override
- def _generate_video_extractors(
- self, video_paths: T.Sequence[Path]
- ) -> T.Sequence[GPXVideoExtractor | types.ErrorMetadata]:
- extractors = super()._generate_video_extractors(video_paths)
- # Read each NMEA file here, once however many videos it geotags, and
- # warn about it once: the videos are geotagged in a pool of processes,
- # each of which would read it again
- for extractor in extractors:
- if isinstance(extractor, NMEAVideoExtractor):
- extractor.read_tracks()
- return extractors
+ def _read_here(self, extractor: GPXVideoExtractor, num_videos: int) -> bool:
+ # Every NMEA file, which is read once in a process however many videos
+ # it geotags, and warned about once
+ return True
diff --git a/mapillary_tools/geotag/image_extractors/exif.py b/mapillary_tools/geotag/image_extractors/exif.py
index 902dc755..fe31d358 100644
--- a/mapillary_tools/geotag/image_extractors/exif.py
+++ b/mapillary_tools/geotag/image_extractors/exif.py
@@ -20,22 +20,33 @@
from .base import BaseImageExtractor
+def _invalid_capture_time(capture_time: datetime.datetime) -> Exception:
+ return exceptions.MapillaryGeoTaggingError(
+ f"Unable to extract timestamp from the image: invalid capture time {capture_time}"
+ )
+
+
def _capture_unix_time(capture_time: datetime.datetime) -> float:
"""
The Unix time of the capture time of an image, assuming that a naive one
is in the local timezone. Some cameras write 0001:01:01 00:00:00 for no
date, which has no Unix time, and others other dates before 1970 (see
exif_read.is_valid_capture_time()): raise as for an image with no capture
- time, rather than let geo.as_unix_time() move it to 1970.
+ time, rather than let geo.as_unix_time() move it to 1970. So for a time
+ that the local timezone moves before 1970, or into year 10000, in which
+ no capture time can be written.
"""
+ if not exif_read.is_valid_capture_time(capture_time):
+ raise _invalid_capture_time(capture_time)
try:
- if not exif_read.is_valid_capture_time(capture_time):
- raise ValueError(f"placeholder date {capture_time}")
- return capture_time.timestamp()
+ unix_time = capture_time.timestamp()
+ # Raises for a time after year 9999
+ datetime.datetime.fromtimestamp(unix_time, datetime.timezone.utc)
except (ValueError, OverflowError, OSError):
- raise exceptions.MapillaryGeoTaggingError(
- f"Unable to extract timestamp from the image: invalid capture time {capture_time}"
- )
+ raise _invalid_capture_time(capture_time)
+ if unix_time < 0:
+ raise _invalid_capture_time(capture_time)
+ return unix_time
class ImageEXIFExtractor(BaseImageExtractor):
@@ -62,6 +73,9 @@ def extract(self) -> types.ImageMetadata:
capture_time = exif.extract_capture_time()
if capture_time is None:
+ if exif.rejected_capture_time is not None:
+ # Say which date was no capture time, as for the only one
+ raise _invalid_capture_time(exif.rejected_capture_time)
raise exceptions.MapillaryGeoTaggingError(
"Unable to extract timestamp from the image"
)
diff --git a/mapillary_tools/geotag/utils.py b/mapillary_tools/geotag/utils.py
index e1b02e51..fa0a0b4a 100644
--- a/mapillary_tools/geotag/utils.py
+++ b/mapillary_tools/geotag/utils.py
@@ -5,7 +5,13 @@
from __future__ import annotations
+import collections
+import dataclasses
+import hashlib
+import io
import logging
+import multiprocessing
+import os
import typing as T
import xml.etree.ElementTree as ET
from pathlib import Path
@@ -17,10 +23,95 @@
Track = T.List[geo.Point]
LOG = logging.getLogger(__name__)
+# The tracks parsed from the last GPX files read in this process, or the error
+# parsing them raised, by path and content: a GPX file that geotags several
+# videos, or the samples of a video and then their images, is parsed once.
+# Only a few are kept, so that one GPX file per video does not pile up
+_PARSED_GPX: collections.OrderedDict[
+ tuple[str, str], list[Track] | gpxpy.gpx.GPXException | UnicodeError
+] = collections.OrderedDict()
+_MAX_PARSED_GPX = 4
+
+# The warnings given in this process about the track files read, so that each
+# is given once however many times a file is read
+_WARNED: set[tuple[T.Hashable, ...]] = set()
+
def parse_gpx(gpx_file: Path) -> list[Track]:
- with gpx_file.open("r") as f:
- gpx = gpxpy.parse(f)
+ """
+ Parse the tracks of a GPX file, each a list of points with Unix times
+ """
+ # Copies, which the caller may change
+ return [
+ [dataclasses.replace(point) for point in track]
+ for track in read_gpx_tracks(gpx_file)
+ ]
+
+
+def read_gpx_tracks(gpx_file: Path) -> list[Track]:
+ """
+ The tracks of a GPX file as parse_gpx() parses them, parsed once in a
+ process for as long as the file does not change. They are shared with
+ every other reader of the file, and must not be changed.
+
+ A worker of a pool keeps none: it parses the GPX file of one video only,
+ as a file that geotags several videos is read before the pool (see
+ GeotagVideosFromGPX), and would free them only when it exits.
+ """
+ data = gpx_file.read_bytes()
+ key = (str(gpx_file.resolve()), hashlib.sha256(data).hexdigest())
+ parsed = _PARSED_GPX.get(key)
+ if parsed is None:
+ try:
+ # Decoded as a file opened in text mode would be, and parsed
+ # without the bytes, as a file is
+ text = io.TextIOWrapper(io.BytesIO(data)).read()
+ del data
+ parsed = _parse_gpx_text(text)
+ except (gpxpy.gpx.GPXException, UnicodeError) as ex:
+ parsed = ex
+ if multiprocessing.parent_process() is None:
+ _PARSED_GPX[key] = parsed
+ while _MAX_PARSED_GPX < len(_PARSED_GPX):
+ _PARSED_GPX.popitem(last=False)
+ else:
+ _PARSED_GPX.move_to_end(key)
+
+ if isinstance(parsed, Exception):
+ raise parsed.with_traceback(None)
+ return parsed
+
+
+def was_gpx_read(gpx_file: Path) -> bool:
+ """
+ Whether the tracks of a GPX file were parsed in this process and are kept
+ """
+ path = str(gpx_file.resolve())
+ return any(key[0] == path for key in _PARSED_GPX)
+
+
+def warn_once(
+ path: Path, what: tuple[T.Hashable, ...], message: str, *args: object
+) -> None:
+ """
+ Log a warning about the track file **path**, unless this process warned
+ already of **what** about the file as it is now, whoever read it
+ """
+ try:
+ stat = os.stat(path)
+ except OSError:
+ file_key: tuple[T.Hashable, ...] = (str(path),)
+ else:
+ file_key = (str(path.resolve()), stat.st_size, stat.st_mtime_ns)
+ key = (*file_key, *what)
+ if key in _WARNED:
+ return
+ _WARNED.add(key)
+ LOG.warning(message, *args)
+
+
+def _parse_gpx_text(text: str) -> list[Track]:
+ gpx = gpxpy.parse(text)
tracks: list[Track] = []
diff --git a/mapillary_tools/geotag/video_extractors/gpx.py b/mapillary_tools/geotag/video_extractors/gpx.py
index 1f145ba9..114e9382 100644
--- a/mapillary_tools/geotag/video_extractors/gpx.py
+++ b/mapillary_tools/geotag/video_extractors/gpx.py
@@ -29,8 +29,7 @@
from ...mp4 import construct_mp4_parser as cparser, simple_mp4_parser as sparser
from ...mp4.mp4_sample_parser import MovieBoxParser
from ...serializer.description import build_capture_time
-from .. import options
-from ..utils import parse_gpx
+from .. import options, utils as geotag_utils
from .base import BaseVideoExtractor
from .native import NativeVideoExtractor
@@ -175,6 +174,10 @@ class GPXVideoExtractor(BaseVideoExtractor):
# The format of the track file, as messages name it
file_format = "GPX"
+ # The tracks of the track file, or the error reading it, once read_tracks()
+ # has read them
+ _tracks: list[list[geo.Point]] | exceptions.MapillaryInvalidGPXError | None = None
+
def __init__(
self,
video_path: Path,
@@ -285,39 +288,93 @@ def extract(self) -> types.VideoMetadata:
return dataclasses.replace(native_video_metadata, points=gpx_points)
+ def read_tracks(self) -> None:
+ """
+ Read the tracks of the track file now, for extract() to place, rather
+ than when it runs, which can be in another process: a file that
+ geotags several videos is then read, and warned about, once
+ """
+ try:
+ self._tracks = self._read_tracks()
+ except exceptions.MapillaryInvalidGPXError as ex:
+ self._tracks = ex
+ except Exception:
+ # Left for extract() to raise again, and to be reported for the
+ # video as any unexpected error is
+ pass
+ else:
+ self._warn_of_tracks(self._tracks)
+
def _read_gpx_points(self) -> list[geo.Point]:
"""
Read the points of the GPX file, merged into one track in time order
"""
+ warned = self._tracks is not None
gpx_tracks = self._parse_tracks()
+ if not warned:
+ self._warn_of_tracks(gpx_tracks)
+
+ # Copies, which are moved to the video: the tracks may be shared with
+ # other readers of the file
+ gpx_points = [
+ dataclasses.replace(p)
+ for track in gpx_tracks
+ for p in track
+ if _GPS_EPOCH <= p.time
+ ]
+
+ # The tracks need not be in time order, and interpolation needs the
+ # merged points to be
+ gpx_points.sort(key=lambda p: p.time)
+ return gpx_points
+ def _warn_of_tracks(self, gpx_tracks: list[list[geo.Point]]) -> None:
+ """
+ Warn of what reading the points of the tracks leaves out or merges,
+ once for each file
+ """
if 1 < len(gpx_tracks):
- LOG.warning(
- f"Found {len(gpx_tracks)} tracks in the {self.file_format} file {self.gpx_path}. Will merge points in all the tracks as a single track for interpolation"
+ geotag_utils.warn_once(
+ self.gpx_path,
+ ("tracks", len(gpx_tracks)),
+ "Found %d tracks in the %s file %s. Will merge points in all the tracks as a single track for interpolation",
+ len(gpx_tracks),
+ self.file_format,
+ self.gpx_path,
)
- all_points: list[geo.Point] = sum(gpx_tracks, [])
- gpx_points = [p for p in all_points if _GPS_EPOCH <= p.time]
- if len(gpx_points) < len(all_points):
- LOG.warning(
+ num_points = sum(len(track) for track in gpx_tracks)
+ num_kept = sum(1 for track in gpx_tracks for p in track if _GPS_EPOCH <= p.time)
+ if num_kept < num_points:
+ geotag_utils.warn_once(
+ self.gpx_path,
+ ("before 1980", num_points - num_kept, num_points),
"Ignoring %d of the %d points in the %s file %s, which are timestamped before 1980",
- len(all_points) - len(gpx_points),
- len(all_points),
+ num_points - num_kept,
+ num_points,
self.file_format,
self.gpx_path,
)
- # The tracks need not be in time order, and interpolation needs the
- # merged points to be
- gpx_points.sort(key=lambda p: p.time)
- return gpx_points
-
def _parse_tracks(self) -> list[list[geo.Point]]:
"""
- Parse the tracks of the GPX file, each a list of points with Unix times
+ The tracks of the track file, each a list of points with Unix times,
+ as read_tracks() read them, or read now. They may be shared with other
+ readers of the file, and must not be changed
+ """
+ if self._tracks is None:
+ return self._read_tracks()
+ if isinstance(self._tracks, exceptions.MapillaryInvalidGPXError):
+ raise self._tracks
+ return self._tracks
+
+ def _read_tracks(self) -> list[list[geo.Point]]:
+ """
+ Read the tracks of the GPX file, parsed once in a process (see
+ read_gpx_tracks())
"""
try:
- return parse_gpx(self.gpx_path)
+ return geotag_utils.read_gpx_tracks(self.gpx_path)
except (OSError, UnicodeError, gpxpy.gpx.GPXException) as ex:
raise exceptions.MapillaryInvalidGPXError(
f"{self.gpx_path.name} is not a valid GPX file. "
@@ -351,7 +408,7 @@ def _clip_to_video(
"""
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
+ gpx_points, constants.MAX_GPX_GAP_SECONDS, video_duration, rate
)
if clock is None:
return _clip_track(gpx_points, None)
@@ -494,7 +551,8 @@ def _check_coverage(
# The gaps of the track, and its parts between them, in video seconds
gap_points = track_gaps(
sorted(gpx_points, key=lambda p: p.time),
- constants.MAX_GPX_GAP_SECONDS / rate,
+ constants.MAX_GPX_GAP_SECONDS,
+ rate,
)
gaps = [(before.time, after.time) for before, after in gap_points]
parts = list(
@@ -1087,11 +1145,12 @@ def _step_rates(cls, pairs: T.Sequence[tuple[float, float]]) -> list[float]:
def track_gaps(
- points: T.Sequence[geo.Point], max_gap: float
+ points: T.Sequence[geo.Point], max_gap: float, rate: float = 1.0
) -> list[tuple[geo.Point, geo.Point]]:
"""
The gaps of a track, sorted by time, between consecutive points more than
- **max_gap** seconds apart, as the points either side
+ **max_gap** GPS seconds apart (see gps_seconds_apart()), as the points
+ either side
>>> track = [geo.Point(time=t, lat=0, lon=0, alt=None, angle=None) for t in (0, 1, 5000, 5001, 9000)]
>>> [(before.time, after.time) for before, after in track_gaps(track, 3600)]
@@ -1100,14 +1159,45 @@ def track_gaps(
[(1, 5000)]
>>> track_gaps(track[:2], 3600)
[]
+
+ A gap of exactly **max_gap** is none, also along a video whose clock runs
+ at another rate than 1:
+
+ >>> placed = [geo.Point(time=t / 1.1426, lat=0, lon=0, alt=None, angle=None) for t in (0, 3600)]
+ >>> track_gaps(placed, 3600, rate=1.1426)
+ []
"""
return [
(left, right)
for left, right in geo.pairwise(points)
- if max_gap < right.time - left.time
+ if max_gap < gps_seconds_apart(left, right, rate)
]
+def gps_seconds_apart(left: geo.Point, right: geo.Point, rate: float = 1.0) -> float:
+ """
+ The GPS seconds between two points of a track: those between the GPS
+ times they keep, where both keep one, and otherwise between their times
+ at **rate** GPS seconds per second of their time, as a track placed along
+ a timelapse packs several GPS seconds into each second of video. Rounded to
+ the millisecond, which no GPS time is recorded more finely than, so that
+ neither a division by the rate nor a fractional timestamp makes a gap of
+ exactly the limit more than it
+
+ >>> p = lambda t: geo.Point(time=t, lat=0, lon=0, alt=None, angle=None)
+ >>> gps_seconds_apart(p(1.0), p(3.5), rate=2.0)
+ 5.0
+ >>> gps_seconds_apart(p(0), p(3600 / 1.1426), rate=1.1426)
+ 3600.0
+ """
+ left_time, right_time = left.get_unix_time(), right.get_unix_time()
+ if left_time is None or right_time is None:
+ seconds = (right.time - left.time) * rate
+ else:
+ seconds = right_time - left_time
+ return round(seconds, 3)
+
+
def describe_gap(
before: geo.Point, after: geo.Point, before_time: float, after_time: float
) -> str:
@@ -1151,12 +1241,15 @@ def describe_long_track(
def _leave_out_far_side_of_gaps(
- points: T.Sequence[geo.Point], max_gap: float, end: float | None
+ points: T.Sequence[geo.Point],
+ max_gap: float,
+ end: float | None,
+ rate: float = 1.0,
) -> 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
+ of more than **max_gap** GPS seconds (see gps_seconds_apart()) 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)]
@@ -1168,7 +1261,7 @@ def _leave_out_far_side_of_gaps(
"""
start, stop = 0, len(points)
for idx, (before, after) in enumerate(geo.pairwise(points)):
- if after.time - before.time <= max_gap:
+ if gps_seconds_apart(before, after, rate) <= max_gap:
continue
if before.time < 0 < after.time:
start = idx + 1
diff --git a/mapillary_tools/geotag/video_extractors/nmea.py b/mapillary_tools/geotag/video_extractors/nmea.py
index f8ac1082..32e596a0 100644
--- a/mapillary_tools/geotag/video_extractors/nmea.py
+++ b/mapillary_tools/geotag/video_extractors/nmea.py
@@ -13,7 +13,10 @@
from typing_extensions import override
from ... import exceptions, geo
-from ..geotag_images_from_nmea_file import get_lat_lon_time_from_nmea
+from ..geotag_images_from_nmea_file import (
+ describe_invalid_nmea,
+ get_lat_lon_time_from_nmea,
+)
from ..utils import track_point
from .gpx import GPXVideoExtractor
@@ -26,32 +29,7 @@ class NMEAVideoExtractor(GPXVideoExtractor):
file_format = "NMEA"
- # The tracks of the NMEA file, or the error reading it, once read_tracks()
- # has read them
- _tracks: list[list[geo.Point]] | exceptions.MapillaryInvalidGPXError | None = None
-
- def read_tracks(self) -> None:
- """
- Read the tracks of the NMEA file now, for extract() to place, rather
- than when it runs, which can be in another process
- """
- try:
- self._tracks = self._read_tracks()
- except exceptions.MapillaryInvalidGPXError as ex:
- self._tracks = ex
- except Exception:
- # Left for extract() to raise again, and to be reported for the
- # video as any unexpected error is
- pass
-
@override
- def _parse_tracks(self) -> list[list[geo.Point]]:
- if self._tracks is None:
- return self._read_tracks()
- if isinstance(self._tracks, exceptions.MapillaryInvalidGPXError):
- raise self._tracks
- return self._tracks
-
def _read_tracks(self) -> list[list[geo.Point]]:
try:
points = get_lat_lon_time_from_nmea(self.gpx_path)
@@ -59,8 +37,7 @@ def _read_tracks(self) -> list[list[geo.Point]]:
# malformed sentences and no fix
except (OSError, ValueError) as ex:
raise exceptions.MapillaryInvalidGPXError(
- f"{self.gpx_path.name} is not a valid NMEA file. "
- "Replace it with a valid NMEA file."
+ describe_invalid_nmea(self.gpx_path, ex)
) from ex
return [[track_point(p.time, p.lat, p.lon, p.alt) for p in points]]
diff --git a/mapillary_tools/sample_video.py b/mapillary_tools/sample_video.py
index b2608ddf..59b24b81 100644
--- a/mapillary_tools/sample_video.py
+++ b/mapillary_tools/sample_video.py
@@ -40,6 +40,7 @@
)
from .geotag.video_extractors.gpx import (
describe_gap,
+ gps_seconds_apart,
GPXVideoExtractor,
track_gaps,
VideoClock,
@@ -880,13 +881,14 @@ def _sample_single_video_by_interval(
# Located along the video's own GPS track, or the GPX (or NMEA) track of
# --video_geotag_source as GPX sync placed it on the video, as distance
# samples are
- track = _read_track_for_interval_samples(video_path, geotag_options)
+ track_read = _read_track_for_interval_samples(video_path, geotag_options)
+ track = track_read[0] if track_read is not None else None
# The gaps of a track placed from a file, outside it (see track_gaps())
track_gaps_: list[tuple[geo.Point, geo.Point]] = []
- track_from_file = track is not None and _placed_from_a_file(geotag_options)
- if track is not None and track_from_file:
- track = _start_at_the_first_sample(video_path, track, geotag_options)
- track_gaps_ = track_gaps(track.points, constants.MAX_GPX_GAP_SECONDS / rate)
+ track_from_file = track_read is not None and _placed_from_a_file(track_read[1])
+ if track_read is not None and track_from_file:
+ track = _start_at_the_first_sample(video_path, *track_read)
+ track_gaps_ = track_gaps(track.points, constants.MAX_GPX_GAP_SECONDS, rate)
interpolator = geo.Interpolator([track.points]) if track is not None else None
# Or located by the geotagging stage along a GPX (or NMEA) track, by their
# capture times
@@ -1011,17 +1013,20 @@ def _sample_single_video_by_interval(
exif_edit.write()
-def _placed_from_a_file(geotag_options: T.Sequence[SourceOption]) -> bool:
+def _placed_from_a_file(track_source: SourceOption | None) -> bool:
"""
- Whether the track that the interval samples are located along was placed
- on the video from a GPX (or NMEA) file: its only sources are such files
+ Whether the track of a video was placed on it from a GPX (or NMEA) file:
+ the source that read it, of those tried in turn, is such a file, whatever
+ the other sources are
"""
- return bool(geotag_options) and all(
- option.source in factory.EXTERNAL_GPS_SOURCES for option in geotag_options
+ return (
+ track_source is not None and track_source.source in factory.EXTERNAL_GPS_SOURCES
)
-def _placed_from_a_video_file(geotag_options: T.Sequence[SourceOption]) -> bool:
+def _placed_from_a_video_file(
+ track_source: SourceOption | None, geotag_options: T.Sequence[SourceOption]
+) -> bool:
"""
Whether the track that distance samples are sampled along is placed on
the video from a GPX (or NMEA) file of --video_geotag_source, as interval
@@ -1029,8 +1034,9 @@ def _placed_from_a_video_file(geotag_options: T.Sequence[SourceOption]) -> bool:
file of --geotag_source geotags the samples too, which the geotagging stage
places along its track by their capture times
"""
- return _placed_from_a_file(geotag_options) and not any(
- option.filetypes is None or types.FileType.IMAGE in option.filetypes
+ return _placed_from_a_file(track_source) and not any(
+ option.source in factory.EXTERNAL_GPS_SOURCES
+ and (option.filetypes is None or types.FileType.IMAGE in option.filetypes)
for option in geotag_options
)
@@ -1042,8 +1048,7 @@ def _file_track_gaps(
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
+ keep (see gps_seconds_apart()), which a timelapse packs into less video
"""
gaps: list[tuple[geo.Point, geo.Point]] = []
for before, after in geo.pairwise(points):
@@ -1051,7 +1056,7 @@ def _file_track_gaps(
if (
before_time is not None
and after_time is not None
- and constants.MAX_GPX_GAP_SECONDS < after_time - before_time
+ and constants.MAX_GPX_GAP_SECONDS < gps_seconds_apart(before, after)
):
gaps.append((before, after))
return gaps
@@ -1067,16 +1072,15 @@ def _describe_file_track_gap(before: geo.Point, after: geo.Point) -> str:
def _start_at_the_first_sample(
video_path: Path,
track: types.VideoMetadata,
- geotag_options: T.Sequence[SourceOption],
+ option: SourceOption,
) -> types.VideoMetadata:
"""
A track that --interpolation_use_gpx_start_time starts with the video,
moved to start at the first interval sample: GPX sync starts it at the
composition time of the first frame, at which distance sampling reads the
first frame, and interval sampling stamps the samples from time 0, which
- ffmpeg cuts the first one at
+ ffmpeg cuts the first one at. **option** is the source that placed it
"""
- option = geotag_options[0]
if option.interpolation is None or not option.interpolation.use_gpx_start_time:
return track
# As GPX sync starts it
@@ -1273,14 +1277,15 @@ def _read_file_track_for_interval_samples(
def _read_track_for_interval_samples(
video_path: Path, geotag_options: T.Sequence[SourceOption]
-) -> types.VideoMetadata | None:
+) -> tuple[types.VideoMetadata, SourceOption] | None:
"""
The track that the interval samples of a video are located along, in
- video time, as distance samples are: the video's own GPS track, or that of
- a GPX (or NMEA) file of --video_geotag_source as GPX sync placed it on the
- video. None to leave their positions to the geotagging stage: that places
- them along the track of a GPX (or NMEA) file of --geotag_source, which
- geotags images, by their capture times
+ video time, as distance samples are, and the source that read it: the
+ video's own GPS track, or that of a GPX (or NMEA) file of
+ --video_geotag_source as GPX sync placed it on the video. None to leave
+ their positions to the geotagging stage: that places them along the track
+ of a GPX (or NMEA) file of --geotag_source, which geotags images, by their
+ capture times
"""
if any(
option.source in factory.EXTERNAL_GPS_SOURCES
@@ -1290,17 +1295,10 @@ def _read_track_for_interval_samples(
return None
# Through the factory with the same sources as distance sampling, which
- # reads the video's own GPS by default (see _parse_geotag_options()). One
- # video is read in this process rather than in a pool of its own
- video_metadatas = factory.process(
- [video_path],
- [
- dataclasses.replace(option, num_processes=0)
- for option in geotag_options or [SourceOption(SourceType.NATIVE)]
- ],
+ # reads the video's own GPS by default (see _parse_geotag_options())
+ video_metadata, track_source = _read_video_track(
+ video_path, geotag_options or [SourceOption(SourceType.NATIVE)]
)
- assert len(video_metadatas) == 1, "expect 1 video metadata"
- video_metadata = video_metadatas[0]
if isinstance(video_metadata, types.ErrorMetadata):
LOG.warning(
"The samples of %s are not geotagged, as no GPS track was read from it: %s",
@@ -1319,7 +1317,30 @@ def _read_track_for_interval_samples(
)
return None
- return video_metadata
+ assert track_source is not None, "expect the source of a video's track"
+ return video_metadata, track_source
+
+
+def _read_video_track(
+ video_path: Path, geotag_options: T.Sequence[SourceOption]
+) -> tuple[types.MetadataOrError, SourceOption | None]:
+ """
+ The track of a video, read through the factory by the first of
+ **geotag_options** that geotags it, as in the geotagging stage, and that
+ source (None if none does). The video is read in this process rather than
+ in a pool of its own, so that a GPX (or NMEA) file that geotags several
+ videos is read once
+ """
+ sources: dict[Path, SourceOption] = {}
+ video_metadatas = factory.process(
+ [video_path],
+ [dataclasses.replace(option, num_processes=0) for option in geotag_options],
+ sources=sources,
+ )
+ # Every option can read videos (see _parse_geotag_options()), so process()
+ # returns metadata for the video even if all of them fail
+ assert len(video_metadatas) == 1, "expect 1 video metadata"
+ return video_metadatas[0], sources.get(video_path)
def _add_position(
@@ -1531,12 +1552,8 @@ def _sample_single_video_by_distance(
# Go through the factory rather than reading the video's own telemetry
# directly, so that --geotag_source is honoured here as well: a GPX is the
# documented answer for a camera whose embedded GPS is unusable, and
- # distance sampling needs positions just as much as geotagging does.
- video_metadatas = factory.process([video_path], geotag_options)
- # Every option can read videos (see _parse_geotag_options()), so process()
- # returns metadata for the video even if all of them fail
- assert len(video_metadatas) == 1, "expect 1 video metadata"
- video_metadata = video_metadatas[0]
+ # distance sampling needs positions just as much as geotagging does
+ video_metadata, track_source = _read_video_track(video_path, geotag_options)
# Distance sampling needs positions to decide which frames to cut, so
# failing to read them is a failed sample, not something to carry on past.
@@ -1574,7 +1591,7 @@ def _sample_single_video_by_distance(
# GPS of their video instead
gaps = (
_file_track_gaps(video_metadata.points)
- if _placed_from_a_video_file(geotag_options)
+ if _placed_from_a_video_file(track_source, geotag_options)
and not any(
option.source in _FIRST_SAMPLE_SOURCES for option in sample_sources or []
)
diff --git a/tests/integration/test_process.py b/tests/integration/test_process.py
index 5ec3a319..cc305b93 100644
--- a/tests/integration/test_process.py
+++ b/tests/integration/test_process.py
@@ -978,6 +978,113 @@ def track(source: str, source_path: py.path.local):
assert nmea_point == pytest.approx(gpx_point, abs=1e-6)
+def test_process_video_from_nmea_with_some_checksums(setup_data: py.path.local):
+ """
+ Checksums are optional in NMEA. Where some sentences had one, those
+ without were all taken for cut short, and the file failed
+ """
+ video_path = setup_data.join("gopro_data").join("max-360mode.mp4")
+ start = datetime.datetime(2019, 11, 18, 23, 44, 40, tzinfo=datetime.timezone.utc)
+
+ def write_nmea(name: str, with_checksum) -> py.path.local:
+ sentences = []
+ for i in range(21):
+ time = start + datetime.timedelta(seconds=i)
+ hhmmss = time.strftime("%H%M%S.00")
+ position = f"3700.{i * 100:04d},N,12200.0000,W"
+ for body in [
+ f"GPRMC,{hhmmss},A,{position},0.0,0.0,{time.strftime('%d%m%y')},,,A",
+ f"GPGGA,{hhmmss},{position},1,08,1.0,,M,,M,,",
+ ]:
+ checksum = 0
+ for char in body:
+ checksum ^= ord(char)
+ ending = f"*{checksum:02X}" if with_checksum(body) else ""
+ sentences.append(f"${body}{ending}\r\n")
+ nmea_file = setup_data.join(name)
+ nmea_file.write_binary("".join(sentences).encode("ascii"))
+ return nmea_file
+
+ def track(nmea_file: py.path.local):
+ [desc] = run_process_for_descs(
+ [
+ *["--geotag_source", "nmea"],
+ *["--geotag_source_path", str(nmea_file)],
+ str(video_path),
+ ]
+ )
+ assert "error" not in desc, desc
+ return desc["MAPGPSTrack"]
+
+ every = track(write_nmea("every.nmea", lambda body: True))
+ assert every
+ assert track(write_nmea("gga.nmea", lambda body: body.startswith("GPGGA"))) == every
+ assert track(write_nmea("one.nmea", lambda body: "234450" in body)) == every
+
+
+def test_process_video_from_nmea_with_a_byte_order_mark(setup_data: py.path.local):
+ """The byte order mark that some editors write cost the first sentence"""
+ video_path = setup_data.join("gopro_data").join("max-360mode.mp4")
+ start = datetime.datetime(2019, 11, 18, 23, 44, 40, tzinfo=datetime.timezone.utc)
+ sentences = []
+ for i in range(21):
+ time = start + datetime.timedelta(seconds=i)
+ hhmmss = time.strftime("%H%M%S.00")
+ position = f"3700.{i * 100:04d},N,12200.0000,W"
+ # GGA first, so that the first fix lost its position to the mark
+ for body in [
+ f"GPGGA,{hhmmss},{position},1,08,1.0,,M,,M,,",
+ f"GPRMC,{hhmmss},A,{position},0.0,0.0,{time.strftime('%d%m%y')},,,A",
+ ]:
+ checksum = 0
+ for char in body:
+ checksum ^= ord(char)
+ sentences.append(f"${body}*{checksum:02X}\r\n")
+ content = "".join(sentences).encode("ascii")
+
+ def track(name: str, data: bytes):
+ nmea_file = setup_data.join(name)
+ nmea_file.write_binary(data)
+ desc_path = setup_data.join(f"{name}.json")
+ result = run_command(
+ [
+ "--skip_process_errors",
+ *["--desc_path", str(desc_path)],
+ *["--geotag_source", "nmea"],
+ *["--geotag_source_path", str(nmea_file)],
+ str(video_path),
+ ],
+ "process",
+ capture_output=True,
+ text=True,
+ )
+ assert "malformed" not in result.stderr
+ [desc] = json.loads(desc_path.read())
+ assert "error" not in desc, desc
+ return desc["MAPGPSTrack"]
+
+ assert track("bom.nmea", b"\xef\xbb\xbf" + content) == track("plain.nmea", content)
+
+
+def test_process_video_from_an_nmea_file_that_is_a_video(setup_data: py.path.local):
+ """A video given as the NMEA file, which was decoded whole, fails at once"""
+ video_path = setup_data.join("gopro_data").join("max-360mode.mp4")
+
+ [desc] = run_process_for_descs(
+ [
+ *["--geotag_source", "nmea"],
+ *["--geotag_source_path", str(video_path)],
+ str(video_path),
+ ]
+ )
+
+ assert desc["error"]["type"] == "MapillaryInvalidGPXError"
+ assert (
+ "max-360mode.mp4 is not a valid NMEA file: it is not text"
+ in desc["error"]["message"]
+ )
+
+
def test_process_geotag_with_gpx_pattern_not_found(setup_data: py.path.local):
video_path = setup_data.join("gopro_data").join("max-360mode.mp4")
@@ -1520,3 +1627,51 @@ def test_process_image_with_a_placeholder_capture_time_and_offset(
by_name = {Path(desc["filename"]).name: desc for desc in descs}
assert by_name["placeholder.jpg"]["error"]["type"] == "MapillaryGeoTaggingError"
assert by_name["valid.jpg"]["MAPCaptureTime"] == "2020_01_02_02_04_05_000"
+
+
+def test_process_images_with_capture_times_outside_the_writable_range(
+ tmpdir: py.path.local,
+):
+ """
+ A capture time judged in UTC before 1970, or in year 10000, which no
+ description can be written with: geotagged by a GPX, the latter stopped
+ the run without a description
+ """
+ from PIL import Image
+ from PIL.ExifTags import Base, IFD
+
+ folder = tmpdir.mkdir("images")
+ for name, date, offset in [
+ ("valid.jpg", "2020:01:02 03:04:05", "+01:00"),
+ ("before_1970_in_utc.jpg", "1970:01:01 00:30:00", "+02:00"),
+ ("year_10000_in_utc.jpg", "9999:12:31 23:00:00", "-05:00"),
+ ]:
+ img = Image.new("RGB", (4, 4), (0, 0, 0))
+ exif = img.getexif()
+ exif.get_ifd(IFD.Exif)[Base.DateTimeOriginal] = date
+ exif.get_ifd(IFD.Exif)[Base.OffsetTimeOriginal] = offset
+ img.save(str(folder.join(name)), format="JPEG", exif=exif)
+ gpx = tmpdir.join("track.gpx")
+ gpx.write(
+ '\n'
+ ''
+ ""
+ ''
+ ''
+ ""
+ )
+
+ descs = run_process_for_descs(
+ ["--geotag_source=gpx", f"--geotag_source_path={gpx}", str(folder)]
+ )
+
+ by_name = {Path(desc["filename"]).name: desc for desc in descs}
+ assert by_name["valid.jpg"]["MAPCaptureTime"] == "2020_01_02_02_04_05_000"
+ for name, rejected in [
+ ("before_1970_in_utc.jpg", "1970-01-01 00:30:00+02:00"),
+ ("year_10000_in_utc.jpg", "9999-12-31 23:00:00-05:00"),
+ ]:
+ assert by_name[name]["error"] == {
+ "type": "MapillaryGeoTaggingError",
+ "message": f"Unable to extract timestamp from the image: invalid capture time {rejected}",
+ }
diff --git a/tests/integration/test_video_process.py b/tests/integration/test_video_process.py
index ebeb1970..4469e0ff 100644
--- a/tests/integration/test_video_process.py
+++ b/tests/integration/test_video_process.py
@@ -265,6 +265,53 @@ def test_video_process_by_interval_leaves_out_the_samples_past_the_gpx(
assert "and were left out" in process.stderr
+def test_video_process_reads_a_gpx_file_of_several_videos_once(
+ setup_data: py.path.local,
+):
+ """
+ Sampling by interval read the GPX file of --geotag_source again for each
+ video, to tell which samples fall outside it, and warned about it each
+ time. It is read, and warned about, once
+ """
+ pytest_skip_if_not_ffmpeg_installed()
+
+ gpx = setup_data.join("gpx").join("sf_30km_h.gpx").read_text("utf-8")
+ # Two tracks, which are merged with a warning
+ points = [m.start() for m in re.finditer("" + gpx[middle:]
+
+ with tempfile.TemporaryDirectory() as dir:
+ videos = Path(dir, "videos")
+ videos.mkdir()
+ for name in ("a.mp4", "b.mp4"):
+ shutil.copy(
+ str(setup_data.join("videos").join("sample-5s.mp4")), videos / name
+ )
+ gpx_file = Path(dir, "two_tracks.gpx")
+ gpx_file.write_bytes(two_tracks.encode("utf-8"))
+ process = run_video_process(
+ [
+ "--video_sample_distance=-1",
+ "--video_sample_interval=1",
+ *["--video_start_time", "2025_03_14_07_01_31_000"],
+ "--geotag_source=gpx",
+ *["--geotag_source_path", str(gpx_file)],
+ *["--desc_path", str(Path(dir, "desc.json"))],
+ str(videos),
+ str(Path(dir, "samples")),
+ ],
+ capture_output=True,
+ text=True,
+ )
+ with open(Path(dir, "desc.json")) as fp:
+ descs = json.load(fp)
+
+ assert len(descs) == 6, descs
+ assert all("error" not in desc for desc in descs), descs
+ assert process.stderr.count("Found 2 tracks in the GPX file") == 1, process.stderr
+
+
def test_video_process_by_interval_with_a_video_gpx(setup_data: py.path.local):
"""
The samples are located along the GPX of --video_geotag_source as GPX sync
@@ -303,17 +350,11 @@ def test_video_process_by_interval_with_a_video_gpx(setup_data: py.path.local):
assert desc["MAPLongitude"] == pytest.approx(gpx_desc["MAPLongitude"], abs=1e-7)
-def test_video_process_by_distance_leaves_out_the_frames_in_a_gap_of_a_video_gpx(
- setup_data: py.path.local,
-):
+def _gpx_with_a_gap(setup_data: py.path.local) -> str:
"""
- Without its point at 07:00:33.088, the GPX has a gap from 07:00:31.533 to
- 07:00:34.379, over the limit, inside the video. The frames in it were
- placed across it
+ The GPX without its point at 07:00:33.088: with a gap from 07:00:31.533 to
+ 07:00:34.379, over a limit of 2.5 seconds
"""
- pytest_skip_if_not_ffmpeg_installed()
-
- video_path = setup_data.join("videos").join("sample-5s.mp4")
gpx = setup_data.join("gpx").join("sf_30km_h.gpx").read_text("utf-8")
gpx_with_gap = re.sub(
r").)*?.*?",
@@ -322,13 +363,29 @@ def test_video_process_by_distance_leaves_out_the_frames_in_a_gap_of_a_video_gpx
flags=re.S,
)
assert gpx_with_gap != gpx
+ return gpx_with_gap
+
+
+# Whatever source follows the GPX, which the video, without GPS of its own,
+# leaves to geotag it
+@pytest.mark.parametrize("then", [[], ["--video_geotag_source=native"]])
+def test_video_process_by_distance_leaves_out_the_frames_in_a_gap_of_a_video_gpx(
+ setup_data: py.path.local, then: list[str]
+):
+ """
+ The GPX has a gap inside the video. The frames in it were placed across it
+ """
+ pytest_skip_if_not_ffmpeg_installed()
+
+ video_path = setup_data.join("videos").join("sample-5s.mp4")
with tempfile.TemporaryDirectory() as dir:
gpx_file = Path(dir, "gap.gpx")
- gpx_file.write_text(gpx_with_gap, encoding="utf-8")
+ gpx_file.write_bytes(_gpx_with_a_gap(setup_data).encode("utf-8"))
descs = run_video_process_for_descs(
[
"--video_geotag_source=gpx",
+ *then,
*["--geotag_source_path", str(gpx_file)],
*["--video_start_time", "2025_03_14_07_00_30_000"],
str(video_path),
@@ -348,6 +405,43 @@ def test_video_process_by_distance_leaves_out_the_frames_in_a_gap_of_a_video_gpx
], capture_times
+@pytest.mark.parametrize("then", [[], ["--video_geotag_source=native"]])
+def test_video_process_by_interval_leaves_out_the_samples_in_a_gap_of_a_video_gpx(
+ setup_data: py.path.local, then: list[str]
+):
+ """
+ The GPX has a gap inside the video, from 1.533 to 4.379 seconds into it.
+ The samples in it were placed across it when another source followed the
+ GPX
+ """
+ pytest_skip_if_not_ffmpeg_installed()
+
+ video_path = setup_data.join("videos").join("sample-5s.mp4")
+
+ with tempfile.TemporaryDirectory() as dir:
+ gpx_file = Path(dir, "gap.gpx")
+ gpx_file.write_bytes(_gpx_with_a_gap(setup_data).encode("utf-8"))
+ descs = run_video_process_for_descs(
+ [
+ "--video_sample_distance=-1",
+ "--video_sample_interval=0.5",
+ "--video_geotag_source=gpx",
+ *then,
+ *["--geotag_source_path", str(gpx_file)],
+ *["--video_start_time", "2025_03_14_07_00_30_000"],
+ str(video_path),
+ str(Path(dir, "samples")),
+ ],
+ env={**os.environ, "MAPILLARY_TOOLS_MAX_GPX_GAP_SECONDS": "2.5"},
+ )
+
+ assert all("error" not in desc for desc in descs), descs
+ # Of the 11 samples, from 0 to 5 seconds, not those at 2, 2.5, 3, 3.5 and 4
+ assert sorted(Path(desc["filename"]).name for desc in descs) == [
+ f"sample-5s_v_{idx + 1:06d}.jpg" for idx in [0, 1, 2, 3, 9, 10]
+ ]
+
+
def _with_creation_time(video: py.path.local, creation_time: str, dir: str) -> Path:
"""A copy of the video whose container records the creation time."""
copy = Path(dir, video.basename)
diff --git a/tests/unit/test_gpx_track_check.py b/tests/unit/test_gpx_track_check.py
new file mode 100644
index 00000000..4382725b
--- /dev/null
+++ b/tests/unit/test_gpx_track_check.py
@@ -0,0 +1,22 @@
+# 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.
+
+import pytest
+from mapillary_tools import exceptions, geo
+from mapillary_tools.geotag.geotag_images_from_gpx import check_within_track
+
+
+def test_a_time_past_year_9999_is_outside_the_track():
+ """
+ A time offset can move an image past year 9999. Writing its time into the
+ error stopped the run, rather than fail the image
+ """
+ track = [geo.Point(time=t, lat=0, lon=0, alt=None, angle=None) for t in (10, 20)]
+
+ with pytest.raises(exceptions.MapillaryOutsideGPXTrackError) as info:
+ check_within_track(3e11, track)
+
+ assert info.value.image_time == "300000000000.0"
+ assert info.value.gpx_end_time == "1970_01_01_00_00_20_000"
diff --git a/tests/unit/test_gpx_video_coverage.py b/tests/unit/test_gpx_video_coverage.py
index 12500149..8fa98c19 100644
--- a/tests/unit/test_gpx_video_coverage.py
+++ b/tests/unit/test_gpx_video_coverage.py
@@ -24,7 +24,15 @@
from pathlib import Path
import pytest
-from mapillary_tools import constants, exceptions, geo, telemetry, types, uploader
+from mapillary_tools import (
+ constants,
+ exceptions,
+ geo,
+ sample_video,
+ telemetry,
+ types,
+ uploader,
+)
from mapillary_tools.camm import camm_builder
from mapillary_tools.geotag import factory
from mapillary_tools.geotag.options import (
@@ -80,6 +88,17 @@ def _video_track(rate: float = 1.0) -> list[telemetry.CAMMGPSPoint]:
]
+def _long_video_track(rate: float) -> list[telemetry.CAMMGPSPoint]:
+ """
+ A GPS point every second of 2 minutes of video, its clock running rate
+ times faster: long enough for a rate near 1 to be measured
+ """
+ return [
+ _camm_point(time=float(t), epoch_time=A_UNIX_TIME + rate * t)
+ for t in range(121)
+ ]
+
+
# GPS without timestamps, so the GPX can only be assumed to start with the video
UNTIMED_TRACK = [
geo.Point(time=float(t), lat=37.0, lon=14.0, alt=None, angle=None)
@@ -508,6 +527,53 @@ def test_video_is_geotagged_by_the_next_source_with_a_warning(
assert "misses the video" in warning and "by 2.0 hours" in warning
assert warning.endswith("Geotagged it from the native source instead")
+ @pytest.mark.parametrize(
+ "gpx_start, source",
+ [(A_UNIX_TIME, SourceType.GPX), (A_UNIX_TIME + 2 * 3600, SourceType.NATIVE)],
+ )
+ def test_the_source_that_geotagged_the_video_is_given(
+ self, tmp_path, gpx_start, source
+ ):
+ """
+ Interval and distance samples leave out the gaps of a track placed from
+ a GPX, and not those of the video's own track
+ """
+ video_path = _write_camm_mp4(tmp_path / "video.mp4", _video_track())
+ gpx_path = _write_gpx(tmp_path / "track.gpx", gpx_start, 10)
+ options = [
+ # As --video_geotag_source gpx --video_geotag_source native
+ SourceOption(
+ SourceType.GPX,
+ num_processes=0,
+ source_path=SourcePathOption(source_path=gpx_path),
+ filetypes={types.FileType.VIDEO},
+ ),
+ SourceOption(
+ SourceType.NATIVE, num_processes=0, filetypes={types.FileType.VIDEO}
+ ),
+ ]
+
+ sources: dict[Path, SourceOption] = {}
+ [metadata] = factory.process([video_path], options, sources=sources)
+ assert isinstance(metadata, types.VideoMetadata)
+ assert sources[video_path].source is source
+
+ read = sample_video._read_track_for_interval_samples(video_path, options)
+ assert read is not None
+ assert read[1].source is source
+ assert sample_video._placed_from_a_file(read[1]) is (source is SourceType.GPX)
+
+ def test_no_source_is_given_for_a_video_none_geotagged(self, tmp_path):
+ sources: dict[Path, SourceOption] = {}
+ [metadata] = factory.process(
+ [NO_GPS_VIDEO],
+ [SourceOption(SourceType.NATIVE, num_processes=0)],
+ sources=sources,
+ )
+
+ assert isinstance(metadata, types.ErrorMetadata)
+ assert sources == {}
+
def _process_images(
self, tmp_path: Path, sources: list[SourceType], count: int = 1
) -> list[types.MetadataOrError]:
@@ -723,6 +789,14 @@ def test_an_image_in_a_gap_of_less_than_an_hour_is_interpolated(self, tmp_path):
)
assert isinstance(metadata, types.ImageMetadata)
+ def test_an_image_in_a_gap_of_exactly_the_limit_is_interpolated(self, tmp_path):
+ """A gap of exactly the limit is no gap, also between fractional times"""
+ before = self.IMAGE_TIME - 1800.123457
+ metadata = self._process_image(
+ tmp_path, [(before - 60, 60), (before + 3600, 60)]
+ )
+ assert isinstance(metadata, types.ImageMetadata)
+
def test_an_image_on_a_point_by_a_gap_is_on_the_track(self, tmp_path):
"""The last point before the gap is the image's own position"""
metadata = self._process_image(
@@ -821,6 +895,48 @@ def test_the_limit_can_be_raised(self, tmp_path, monkeypatch):
)
assert metadata.points[0].time == 0
+ # A LabPano clock rate, at which the gap in video seconds came out of a
+ # division a little more than the limit divided by the rate
+ @pytest.mark.parametrize("rate", [1.0, 1.142609374051617, 15.0])
+ @pytest.mark.parametrize("gap, interpolated", [(3600, True), (3600.5, False)])
+ def test_a_start_in_a_gap_of_the_limit(self, tmp_path, rate, gap, interpolated):
+ """
+ A gap of exactly the limit is no gap, also along a video whose clock
+ runs at another rate than 1: it is judged in GPS seconds
+ """
+ video_path = _write_camm_mp4(
+ tmp_path / "video.mp4", _long_video_track(rate), duration=125.0
+ )
+ # A fix before the video, and the track from a second into it
+ first = A_UNIX_TIME + rate
+ gpx_path = _write_gpx_parts(
+ tmp_path / "track.gpx", [(first - gap, 0), (first, 60)]
+ )
+ metadata = GPXVideoExtractor(video_path, gpx_path, clip_to_video=True).extract()
+ if interpolated:
+ assert metadata.points[0].time == 0
+ else:
+ assert metadata.points[0].time == pytest.approx(1.0)
+
+ @pytest.mark.parametrize("rate", [1.0, 1.142609374051617, 15.0])
+ def test_a_gap_of_the_limit_inside_the_video(self, tmp_path, monkeypatch, rate):
+ monkeypatch.setattr(constants, "MAX_GPX_GAP_SECONDS", 60)
+ video_path = _write_camm_mp4(
+ tmp_path / "video.mp4", _long_video_track(rate), duration=125.0
+ )
+
+ def extract(gap: float) -> types.VideoMetadata:
+ # The gap starts 22 GPS seconds into the video
+ gpx_path = _write_gpx_parts(
+ tmp_path / "track.gpx",
+ [(A_UNIX_TIME - 10, 32), (A_UNIX_TIME + 22 + gap, 120)],
+ )
+ return GPXVideoExtractor(video_path, gpx_path, clip_to_video=True).extract()
+
+ assert extract(60).points
+ with pytest.raises(exceptions.MapillaryOutsideGPXTrackError):
+ extract(60.5)
+
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(
diff --git a/tests/unit/test_image_exif_extractor.py b/tests/unit/test_image_exif_extractor.py
index 83ac7e25..30e1fdf6 100644
--- a/tests/unit/test_image_exif_extractor.py
+++ b/tests/unit/test_image_exif_extractor.py
@@ -218,3 +218,77 @@ def test_a_capture_time_from_1970_is_read(self, tmp_path: Path):
metadata = ImageEXIFExtractor(image).extract()
assert metadata.time == 24 * 3600
+
+
+class TestCaptureTimeEdgeCases:
+ @pytest.mark.parametrize(
+ "date, offset, rejected",
+ [
+ (NO_DATE, None, "0001-01-01 00:00:00"),
+ # 1969-12-31 22:30 in UTC
+ ("1970:01:01 00:30:00", "+02:00", "1970-01-01 00:30:00+02:00"),
+ # Year 10000 in UTC, in which no capture time can be written
+ ("9999:12:31 23:00:00", "-05:00", "9999-12-31 23:00:00-05:00"),
+ ],
+ )
+ def test_exif_names_the_rejected_time(
+ self, tmp_path: Path, date: str, offset: T.Optional[str], rejected: str
+ ):
+ tags = {Base.DateTimeOriginal: date}
+ if offset is not None:
+ tags[Base.OffsetTimeOriginal] = offset
+ image = _write_image_with_dates(tmp_path / "image.jpg", tags)
+
+ with pytest.raises(exceptions.MapillaryGeoTaggingError) as info:
+ ImageEXIFExtractor(image).extract()
+
+ assert str(info.value) == (
+ f"Unable to extract timestamp from the image: invalid capture time {rejected}"
+ )
+
+ def test_exiftool_names_the_rejected_time(self, tmp_path: Path):
+ element = _exiftool_element(
+ {
+ "GPS:GPSLatitude": "24.5601",
+ "GPS:GPSLatitudeRef": "N",
+ "GPS:GPSLongitude": "81.8077",
+ "GPS:GPSLongitudeRef": "W",
+ "ExifIFD:DateTimeOriginal": "1970:01:01 00:30:00",
+ "ExifIFD:OffsetTimeOriginal": "+02:00",
+ }
+ )
+
+ with pytest.raises(exceptions.MapillaryGeoTaggingError) as info:
+ ImageExifToolExtractor(tmp_path / "image.jpg", element).extract()
+
+ assert str(info.value).endswith(
+ "invalid capture time 1970-01-01 00:30:00+02:00"
+ )
+
+ def test_1970_is_judged_in_utc(self, tmp_path: Path):
+ """1969-12-31 23:30 one hour west of UTC is 1970-01-01 00:30 in UTC"""
+ image = _write_image_with_dates(
+ tmp_path / "image.jpg",
+ {
+ Base.DateTimeOriginal: "1969:12:31 23:30:00",
+ Base.OffsetTimeOriginal: "-01:00",
+ },
+ )
+
+ metadata = ImageEXIFExtractor(image).extract()
+
+ assert metadata.time == 1800
+
+ def test_the_next_date_tag_is_read_after_a_rejected_time(self, tmp_path: Path):
+ image = _write_image_with_dates(
+ tmp_path / "image.jpg",
+ {
+ Base.DateTimeOriginal: "9999:12:31 23:00:00",
+ Base.OffsetTimeOriginal: "-05:00",
+ Base.DateTimeDigitized: "2020:01:02 03:04:05",
+ },
+ )
+
+ metadata = ImageEXIFExtractor(image).extract()
+
+ assert metadata.time == datetime.datetime(2020, 1, 2, 3, 4, 5).timestamp()
diff --git a/tests/unit/test_nmea_parser.py b/tests/unit/test_nmea_parser.py
index 9b08715a..1df9f548 100644
--- a/tests/unit/test_nmea_parser.py
+++ b/tests/unit/test_nmea_parser.py
@@ -260,7 +260,7 @@ def test_a_sentence_cut_inside_its_date_is_skipped(tmp_path, caplog, gga_first):
] + ([_fix("00:00:01")] if gga_first else [])
assert "Skipping 1 of the" in caplog.text
- assert "(no checksum)" in caplog.text
+ assert "(malformed datestamp: '2212')" in caplog.text
def test_sentences_cut_short_are_skipped_with_one_warning(tmp_path, caplog):
@@ -320,6 +320,123 @@ def bare(sentence: str) -> str:
assert f"(malformed {error}: " in caplog.text
+def _bare(sentence: str) -> str:
+ """The sentence without its checksum, which is optional"""
+ return sentence.rsplit("*", 1)[0]
+
+
+def _log(seconds: int = 4) -> list[str]:
+ lines = []
+ for second in range(seconds):
+ lines += [_rmc(f"02463{second}.00"), _gga(f"02463{second}.00")]
+ return lines
+
+
+def _fixes(seconds: int = 4) -> list:
+ return [_fix(f"02:46:3{second}") for second in range(seconds)]
+
+
+def test_a_log_with_one_checksum_is_read(tmp_path, caplog):
+ """The sentences without one were all taken for cut short: no fix was left"""
+ lines = [_bare(line) for line in _log()]
+ lines[3] = _log()[3]
+
+ with caplog.at_level(logging.WARNING):
+ assert _read(tmp_path, *lines) == _fixes()
+
+ assert not caplog.text
+
+
+def test_a_log_with_checksums_on_gga_sentences_only_is_read(tmp_path, caplog):
+ lines = [line if "GGA," in line else _bare(line) for line in _log()]
+
+ with caplog.at_level(logging.WARNING):
+ assert _read(tmp_path, *lines) == _fixes()
+
+ assert not caplog.text
+
+
+def test_logs_with_and_without_checksums_concatenated_are_read(tmp_path, caplog):
+ """The half without checksums was skipped"""
+ lines = _log()
+ lines = lines[:4] + [_bare(line) for line in lines[4:]]
+
+ with caplog.at_level(logging.WARNING):
+ assert _read(tmp_path, *lines) == _fixes()
+
+ assert not caplog.text
+
+
+@pytest.mark.parametrize(
+ "rmc_tail",
+ [
+ # NMEA 2.0 to 2.2: no mode
+ ",,",
+ # NMEA 2.3: the mode
+ ",,,A",
+ # NMEA 4.1: the mode and the navigational status
+ ",,,A,V",
+ ],
+)
+def test_complete_sentences_without_checksums_are_read(tmp_path, caplog, rmc_tail):
+ rmc = "$GPRMC,024636.00,A,3735.30670,N,12220.25810,W,0.0,0.0,221222" + rmc_tail
+ gga = _bare(_gga("024636.00"))
+
+ with caplog.at_level(logging.WARNING):
+ assert _read(tmp_path, rmc, gga) == [_fix("02:46:36")]
+
+ assert not caplog.text
+
+
+@pytest.mark.parametrize(
+ "cut, reason",
+ [
+ # Whole fields, the last of them maybe cut short too
+ (
+ "$GPRMC,024637.00,A,3735.30670,N,12220.25810,W,0.0,0.0,221222",
+ "cut short: 9 of the 11 fields of an RMC sentence",
+ ),
+ (
+ "$GPGGA,024637.00,3735.30670,N,12220.25810,W,1,08,1.0,12",
+ "cut short: 9 of the 14 fields of a GGA sentence",
+ ),
+ # A time that does not parse
+ ("$GPGGA,02463", "malformed timestamp: '02463'"),
+ ],
+)
+def test_sentences_cut_short_by_their_own_fields_are_skipped(
+ tmp_path, caplog, cut, reason
+):
+ with caplog.at_level(logging.WARNING):
+ assert _read(tmp_path, *_log(1), cut) == _fixes(1)
+
+ assert f"on line 3 ({reason}): {cut}" in caplog.text
+
+
+def test_a_sentence_with_its_checksum_is_not_judged_by_its_fields(tmp_path, caplog):
+ """A checksum shows where the sentence ends, however few fields it has"""
+ short_gga = _sentence("GPGGA,024636.00,3735.30670,N,12220.25810,W,1,08,1.0,12.5,M")
+
+ with caplog.at_level(logging.WARNING):
+ assert _read(tmp_path, _rmc("024636.00"), short_gga) == [_fix("02:46:36")]
+
+ assert not caplog.text
+
+
+def test_a_file_that_fails_says_why(tmp_path):
+ """It failed as "not a valid NMEA file", without a reason"""
+ good = _rmc("024636.00")
+ lines = [good[:-2] + "00", _bare(_gga("024636.00"))[:30]]
+
+ with pytest.raises(ValueError) as excinfo:
+ _read(tmp_path, *lines)
+
+ assert str(excinfo.value) == (
+ "2 of its 2 RMC and GGA sentences are malformed, the first on line 1 "
+ f"(checksum does not match: 00 != {good[-2:]}): {lines[0]}"
+ )
+
+
def _read_bytes(tmp_path: Path, content: bytes):
path = tmp_path / "track.nmea"
path.write_bytes(content)
@@ -364,6 +481,78 @@ def test_a_byte_that_is_not_text_outside_the_sentences_is_ignored(tmp_path, capl
assert not caplog.text
+@pytest.mark.parametrize(
+ "head",
+ [
+ # the start of an MP4 video
+ b"\x00\x00\x00\x18ftypmp42\x00\x00\x00\x00mp42isom",
+ # mostly bytes that are not UTF-8
+ b"\xca\xfe\xba\xbe" * 1000,
+ ],
+ ids=["video", "not UTF-8"],
+)
+def test_a_file_that_is_not_text_fails_without_being_read_whole(tmp_path, head):
+ """
+ A video given as an NMEA file was decoded whole, gigabytes of it: what its
+ first 64 KiB show is enough
+ """
+ padding = b"\xff" * (64 * 1024)
+ sentences = (_rmc("024636.00") + "\r\n" + _gga("024636.00") + "\r\n").encode()
+
+ with pytest.raises(ValueError, match="^it is not text$"):
+ _read_bytes(tmp_path, head + padding + sentences)
+
+
+def test_a_log_longer_than_what_tells_text_is_read_whole(tmp_path):
+ lines = []
+ for minute in range(60):
+ for second in range(10):
+ time = f"02{minute:02d}0{second}.00"
+ lines += [_rmc(time), _gga(time)]
+ content = "".join(f"{line}\r\n" for line in lines).encode("ascii")
+ assert 64 * 1024 < len(content)
+
+ assert len(_read_bytes(tmp_path, content)) == 600
+
+
+def _crlf(lines: list[str]) -> bytes:
+ return "".join(f"{line}\r\n" for line in lines).encode("ascii")
+
+
+def test_a_log_with_a_byte_order_mark_is_read(tmp_path, caplog):
+ """The byte order mark that starts it cost its first sentence, and a warning"""
+ lines = [_rmc("024636.00"), _gga("024636.00")]
+
+ with caplog.at_level(logging.WARNING):
+ times = _read_bytes(tmp_path, b"\xef\xbb\xbf" + _crlf(lines))
+
+ assert len(times) == 1
+ assert not caplog.text
+
+
+def test_logs_with_byte_order_marks_appended_are_read(tmp_path, caplog):
+ first = [_rmc("024636.00"), _gga("024636.00")]
+ second = [_rmc("024637.00"), _gga("024637.00")]
+
+ with caplog.at_level(logging.WARNING):
+ times = _read_bytes(
+ tmp_path, b"\xef\xbb\xbf" + _crlf(first) + b"\xef\xbb\xbf" + _crlf(second)
+ )
+
+ assert len(times) == 2
+ assert not caplog.text
+
+
+def test_a_utf16_log_is_not_read(tmp_path):
+ """As one that is not text: NMEA is ASCII"""
+ content = "".join(
+ f"{line}\r\n" for line in [_rmc("024636.00"), _gga("024636.00")]
+ ).encode("utf-16")
+
+ with pytest.raises(ValueError, match="^it is not text$"):
+ _read_bytes(tmp_path, content)
+
+
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):
diff --git a/tests/unit/test_nmea_video.py b/tests/unit/test_nmea_video.py
index 4a5629ca..e3ccb01d 100644
--- a/tests/unit/test_nmea_video.py
+++ b/tests/unit/test_nmea_video.py
@@ -193,25 +193,33 @@ def test_messages_name_the_nmea_file(self, tmp_path):
class TestInvalidNMEAFile:
@pytest.mark.parametrize(
- "content",
+ "content, reason",
[
# not text
- b"\xca\xfe\xba\xbe" * 100,
+ (b"\xca\xfe\xba\xbe" * 100, "it is not text"),
# a wrong checksum
- b"$GPRMC,024636.00,A,3735.30670,N,12220.25810,W,0.0,0.0,221222,,,A*00\r\n",
+ (
+ b"$GPRMC,024636.00,A,3735.30670,N,12220.25810,W,0.0,0.0,221222,,,A*00\r\n",
+ "the first on line 1 (checksum does not match: 00 != 47)",
+ ),
# a malformed time
- b"$GPRMC,024636.00,A,3735.30670,N,12220.25810,W,0.0,0.0,221222,,,A*47\r\n"
- b"$GPGGA,0246,3735.30670,N,12220.25810,W,1,08,1.0,0.0,M,0.0,M,,*64\r\n",
+ (
+ b"$GPRMC,024636.00,A,3735.30670,N,12220.25810,W,0.0,0.0,221222,,,A*47\r\n"
+ b"$GPGGA,0246,3735.30670,N,12220.25810,W,1,08,1.0,0.0,M,0.0,M,,*64\r\n",
+ "the first on line 2 (malformed timestamp: '0246')",
+ ),
],
)
- def test_it_is_reported_as_invalid(self, tmp_path, content):
+ def test_it_is_reported_as_invalid(self, tmp_path, content, reason):
nmea = tmp_path / "track.nmea"
nmea.write_bytes(content)
with pytest.raises(exceptions.MapillaryInvalidGPXError) as excinfo:
NMEAVideoExtractor(GPS_VIDEO, nmea).extract()
- assert "track.nmea is not a valid NMEA file" in str(excinfo.value)
+ assert "track.nmea is not a valid NMEA file: " in str(excinfo.value)
+ # Why, which it did not say
+ assert reason in str(excinfo.value)
class TestNMEAFileWithoutDatedFixes:
diff --git a/tests/unit/test_process_geotag_properties.py b/tests/unit/test_process_geotag_properties.py
index 13651791..68266280 100644
--- a/tests/unit/test_process_geotag_properties.py
+++ b/tests/unit/test_process_geotag_properties.py
@@ -5,6 +5,7 @@
from __future__ import annotations
+import logging
import typing as T
from pathlib import Path
@@ -240,3 +241,39 @@ def test_geotag_source_path_is_read_as_it_is(self, monkeypatch, tmp_path):
assert option.source_path is not None
assert option.source_path.pattern is None
assert option.source_path.source_path == Path("/data/xmls")
+
+ @pytest.mark.parametrize(
+ "source",
+ [
+ "exiftool_xml",
+ '{"source": "exiftool_xml", "pattern": "%g.gpx.xml"}',
+ '{"source": "gpx", "source_path": "/data/track.gpx"}',
+ ],
+ )
+ def test_no_warning_of_a_geotag_source_path_not_given(
+ self, monkeypatch, tmp_path, caplog, source
+ ):
+ """
+ The sources with a path of their own read it, not the video import
+ path, without a warning about a --geotag_source_path that was not given
+ """
+ with caplog.at_level(logging.WARNING):
+ self._options(monkeypatch, tmp_path, [source])
+
+ assert "--geotag_source_path" not in caplog.text
+
+ def test_warning_of_a_geotag_source_path_ignored(
+ self, monkeypatch, tmp_path, caplog
+ ):
+ """A source with a path of its own ignores the one the user gave"""
+ with caplog.at_level(logging.WARNING):
+ [option, *_] = self._options(
+ monkeypatch,
+ tmp_path,
+ ['{"source": "gpx", "source_path": "/data/track.gpx"}'],
+ Path("/data/other.gpx"),
+ )
+
+ assert option.source_path is not None
+ assert option.source_path.source_path == Path("/data/track.gpx")
+ assert "The option --geotag_source_path is ignored for source" in caplog.text
diff --git a/tests/unit/test_read_track_files_once.py b/tests/unit/test_read_track_files_once.py
new file mode 100644
index 00000000..960a74df
--- /dev/null
+++ b/tests/unit/test_read_track_files_once.py
@@ -0,0 +1,243 @@
+# Copyright (c) Meta Platforms, Inc. and affiliates.
+#
+# This source code is licensed under the BSD license found in the
+# LICENSE file in the root directory of this source tree.
+
+"""
+A GPX file that geotags several videos, or the samples of a video and then
+their images, is parsed once in a run, and warned about once
+"""
+
+from __future__ import annotations
+
+import logging
+import multiprocessing
+from pathlib import Path
+
+import gpxpy
+import pytest
+
+from mapillary_tools import sample_video
+from mapillary_tools.geotag import (
+ geotag_images_from_gpx_file,
+ geotag_videos_from_gpx,
+ utils,
+)
+from mapillary_tools.geotag.options import SourceOption, SourcePathOption, SourceType
+from mapillary_tools.geotag.video_extractors.gpx import GPXVideoExtractor
+
+
+def _gpx(*segments: list[tuple[str, float]]) -> bytes:
+ """A GPX file of a track with **segments** of (time, latitude) points"""
+ trksegs = "".join(
+ ""
+ + "".join(
+ f''
+ for time, lat in segment
+ )
+ + ""
+ for segment in segments
+ )
+ return (
+ ''
+ ''
+ f"{trksegs}"
+ ).encode("utf-8")
+
+
+ONE_TRACK = _gpx([("2025-01-01T00:00:00Z", 52.0), ("2025-01-01T00:00:01Z", 52.1)])
+TWO_TRACKS = _gpx(
+ [("2025-01-01T00:00:00Z", 52.0), ("2025-01-01T00:00:01Z", 52.1)],
+ [("2025-01-01T00:00:02Z", 52.2), ("2025-01-01T00:00:03Z", 52.3)],
+)
+
+
+@pytest.fixture
+def parses(monkeypatch) -> list[int]:
+ """Counts the GPX files parsed, with nothing parsed or warned of before"""
+ if hasattr(utils, "_PARSED_GPX"):
+ monkeypatch.setattr(utils, "_PARSED_GPX", type(utils._PARSED_GPX)())
+ if hasattr(utils, "_WARNED"):
+ monkeypatch.setattr(utils, "_WARNED", set())
+ parsed: list[int] = []
+ parse = gpxpy.parse
+
+ def counting_parse(*args, **kwargs):
+ parsed.append(1)
+ return parse(*args, **kwargs)
+
+ monkeypatch.setattr(gpxpy, "parse", counting_parse)
+ return parsed
+
+
+def test_a_gpx_file_is_parsed_once(tmp_path: Path, parses: list[int]):
+ gpx = tmp_path / "track.gpx"
+ gpx.write_bytes(ONE_TRACK)
+
+ first = utils.parse_gpx(gpx)
+ second = utils.parse_gpx(gpx)
+
+ assert len(parses) == 1
+ assert first == second
+ # Copies, which the caller may change
+ first[0][0].time = 0
+ assert utils.parse_gpx(gpx)[0][0].time != 0
+
+
+def test_a_changed_gpx_file_is_parsed_again(tmp_path: Path, parses: list[int]):
+ gpx = tmp_path / "track.gpx"
+ gpx.write_bytes(ONE_TRACK)
+ assert len(utils.parse_gpx(gpx)) == 1
+
+ gpx.write_bytes(TWO_TRACKS)
+ assert len(utils.parse_gpx(gpx)) == 2
+ assert len(parses) == 2
+
+
+def test_only_the_last_gpx_files_are_kept(tmp_path: Path, parses: list[int]):
+ paths = []
+ for idx in range(utils._MAX_PARSED_GPX + 1):
+ gpx = tmp_path / f"track{idx}.gpx"
+ gpx.write_bytes(ONE_TRACK)
+ paths.append(gpx)
+ utils.read_gpx_tracks(gpx)
+
+ assert len(utils._PARSED_GPX) == utils._MAX_PARSED_GPX
+ assert not utils.was_gpx_read(paths[0])
+ assert utils.was_gpx_read(paths[-1])
+
+
+def test_a_worker_of_a_pool_keeps_no_gpx_file(
+ tmp_path: Path, parses: list[int], monkeypatch
+):
+ """It parses the GPX file of one video only, and would keep it until it exits"""
+ gpx = tmp_path / "track.gpx"
+ gpx.write_bytes(ONE_TRACK)
+ monkeypatch.setattr(multiprocessing, "parent_process", lambda: object())
+
+ assert len(utils.read_gpx_tracks(gpx)) == 1
+ assert not utils.was_gpx_read(gpx)
+
+
+def test_an_invalid_gpx_file_is_parsed_once(tmp_path: Path, parses: list[int]):
+ gpx = tmp_path / "track.gpx"
+ gpx.write_bytes(b"not a GPX file")
+
+ for _ in range(2):
+ with pytest.raises(Exception):
+ utils.read_gpx_tracks(gpx)
+ assert len(parses) == 1
+
+
+def test_images_warn_of_the_tracks_once(tmp_path: Path, parses: list[int], caplog):
+ gpx = tmp_path / "track.gpx"
+ gpx.write_bytes(TWO_TRACKS)
+
+ with caplog.at_level(logging.WARNING):
+ for _ in range(3):
+ points = geotag_images_from_gpx_file._read_gpx_points(gpx)
+
+ assert len(points) == 4
+ assert len(parses) == 1
+ assert caplog.text.count("Found 2 tracks in the GPX file") == 1
+
+
+def test_videos_warn_of_the_tracks_once(tmp_path: Path, parses: list[int], caplog):
+ gpx = tmp_path / "track.gpx"
+ gpx.write_bytes(TWO_TRACKS)
+
+ with caplog.at_level(logging.WARNING):
+ geotag_images_from_gpx_file._read_gpx_points(gpx)
+ for name in ("a.mp4", "b.mp4"):
+ points = GPXVideoExtractor(tmp_path / name, gpx)._read_gpx_points()
+
+ assert len(points) == 4
+ assert len(parses) == 1
+ assert caplog.text.count("Found 2 tracks") == 1
+
+
+def test_video_points_are_copies(tmp_path: Path, parses: list[int]):
+ """GPX sync moves the points of a track to the video"""
+ gpx = tmp_path / "track.gpx"
+ gpx.write_bytes(ONE_TRACK)
+
+ points = GPXVideoExtractor(tmp_path / "a.mp4", gpx)._read_gpx_points()
+ for point in points:
+ point.time = 0
+
+ assert [p.time for p in utils.parse_gpx(gpx)[0]] == [
+ 1735689600.0,
+ 1735689601.0,
+ ]
+
+
+def test_a_gpx_file_of_several_videos_is_read_before_the_pool(
+ tmp_path: Path, parses: list[int]
+):
+ shared = tmp_path / "shared.gpx"
+ shared.write_bytes(ONE_TRACK)
+ own = tmp_path / "c.gpx"
+ own.write_bytes(TWO_TRACKS)
+ videos = []
+ for name in ("a.mp4", "b.mp4", "c.mp4"):
+ videos.append(tmp_path / name)
+ videos[-1].write_bytes(b"")
+
+ shared_geotag = geotag_videos_from_gpx.GeotagVideosFromGPX(
+ source_path=SourcePathOption(source_path=shared)
+ )
+ extractors = shared_geotag._generate_video_extractors(videos)
+ assert all(
+ isinstance(e, GPXVideoExtractor) and e._tracks is not None for e in extractors
+ )
+ assert len(parses) == 1
+
+ # A GPX file of each video is left to the pool
+ own_geotag = geotag_videos_from_gpx.GeotagVideosFromGPX(
+ source_path=SourcePathOption(pattern="%g.gpx")
+ )
+ (extractor,) = own_geotag._generate_video_extractors(videos[2:])
+ assert isinstance(extractor, GPXVideoExtractor)
+ assert extractor._tracks is None
+
+ # Unless this process parsed it already
+ utils.read_gpx_tracks(own)
+ (extractor,) = own_geotag._generate_video_extractors(videos[2:])
+ assert isinstance(extractor, GPXVideoExtractor)
+ assert extractor._tracks is not None
+ assert len(parses) == 2
+
+
+def test_interval_samples_read_the_gpx_file_once(tmp_path: Path, parses: list[int]):
+ gpx = tmp_path / "track.gpx"
+ gpx.write_bytes(ONE_TRACK)
+ sources = [
+ SourceOption(SourceType.GPX, source_path=SourcePathOption(source_path=gpx))
+ ]
+
+ for _ in range(3):
+ track = sample_video._read_file_track_for_interval_samples(sources)
+ assert track is not None
+
+ assert len(parses) == 1
+
+
+def test_distance_sampling_reads_the_video_in_this_process(tmp_path: Path, monkeypatch):
+ """So that a GPX file that geotags several videos is parsed once"""
+ calls = []
+
+ def process(paths, options, filetypes=None, sources=None):
+ calls.append([option.num_processes for option in options])
+ raise RuntimeError("stop here")
+
+ monkeypatch.setattr(sample_video.factory, "process", process)
+ monkeypatch.setattr(sample_video, "_probe_video", lambda ffmpeg, path: None)
+
+ with pytest.raises(RuntimeError, match="stop here"):
+ sample_video._sample_single_video_by_distance(
+ tmp_path / "a.mp4",
+ tmp_path / "samples",
+ sample_distance=3,
+ geotag_options=[SourceOption(SourceType.NATIVE)],
+ )
+ assert calls == [[0]]
diff --git a/tests/unit/test_sample_video.py b/tests/unit/test_sample_video.py
index e5f3b99a..78485f36 100644
--- a/tests/unit/test_sample_video.py
+++ b/tests/unit/test_sample_video.py
@@ -456,7 +456,13 @@ def _video_metadata(video_path: Path, points: T.Sequence[geo.Point]):
model="HERO7 Black",
)
- def _sample(self, tmpdir, monkeypatch, track_points, **kwargs) -> list[Path]:
+ def _sample(
+ self, tmpdir, monkeypatch, track_points, read_by: int = 0, **kwargs
+ ) -> list[Path]:
+ """
+ Sample the video every 2 seconds, its track read by the source at
+ index **read_by** of those sampling tries
+ """
clock_points = self._points(60)
monkeypatch.setattr(
sample_video,
@@ -467,7 +473,7 @@ def _sample(self, tmpdir, monkeypatch, track_points, **kwargs) -> list[Path]:
)
self.track_calls: list[list[SourceOption]] = []
- def fake_process(paths, options):
+ def fake_process(paths, options, sources=None):
self.track_calls.append(list(options))
[video_path] = paths
if isinstance(track_points, Exception):
@@ -476,6 +482,8 @@ def fake_process(paths, options):
track_points, video_path, filetype=FileType.VIDEO
)
]
+ if sources is not None:
+ sources[video_path] = options[read_by]
return [self._video_metadata(video_path, track_points)]
monkeypatch.setattr(sample_video.factory, "process", fake_process)
@@ -653,8 +661,13 @@ def test_located_along_a_video_track_file(
in caplog.text
)
+ @pytest.mark.parametrize(
+ "video_geotag_source",
+ # Whatever source follows the one that read the track
+ [["gpx"], ["gpx", "native"], ["nmea", "exiftool_runtime"]],
+ )
def test_samples_in_a_gap_of_a_video_track_file_are_left_out(
- self, tmpdir, setup_mock, monkeypatch
+ self, tmpdir, setup_mock, monkeypatch, video_geotag_source
):
monkeypatch.setattr(sample_video.constants, "MAX_GPX_GAP_SECONDS", 5.0)
track = self._file_points([*range(11), *range(20, 61)])
@@ -663,7 +676,7 @@ def test_samples_in_a_gap_of_a_video_track_file_are_left_out(
tmpdir,
monkeypatch,
track,
- video_geotag_source=["gpx"],
+ video_geotag_source=video_geotag_source,
geotag_source_path=Path(tmpdir) / "track.gpx",
)
@@ -672,8 +685,9 @@ def test_samples_in_a_gap_of_a_video_track_file_are_left_out(
f"hello_v_{idx + 1:06d}.jpg" for idx in [*range(6), *range(10, 30)]
]
+ @pytest.mark.parametrize("video_geotag_source", [["gpx"], ["gpx", "native"]])
def test_a_video_track_file_started_with_the_video_starts_at_the_first_sample(
- self, tmpdir, setup_mock, monkeypatch
+ self, tmpdir, setup_mock, monkeypatch, video_geotag_source
):
"""
GPX sync starts it at the first frame, 0.5 seconds into the video, and
@@ -686,7 +700,7 @@ def test_a_video_track_file_started_with_the_video_starts_at_the_first_sample(
tmpdir,
monkeypatch,
track,
- video_geotag_source=["gpx"],
+ video_geotag_source=video_geotag_source,
geotag_source_path=Path(tmpdir) / "track.gpx",
interpolation_use_gpx_start_time=True,
)
@@ -696,6 +710,40 @@ def test_a_video_track_file_started_with_the_video_starts_at_the_first_sample(
lon_lat = exif_read.ExifRead(samples[0]).extract_lon_lat()
assert lon_lat == pytest.approx((-74.0, 40.0005))
+ @pytest.mark.parametrize(
+ "video_geotag_source",
+ # After a GPX that misses the video, or before a GPX
+ [["gpx", "native"], ["native", "gpx"]],
+ )
+ def test_a_track_read_by_the_video_is_located_along_as_its_own(
+ self, tmpdir, setup_mock, monkeypatch, video_geotag_source
+ ):
+ """
+ The video's own track, with a gap over the limit from 10 to 20 seconds,
+ locates the samples as without a GPX
+ """
+ monkeypatch.setattr(sample_video.constants, "MAX_GPX_GAP_SECONDS", 5.0)
+ track = [p for p in self._points(60) if not 10 < p.time < 20]
+
+ samples = self._sample(
+ tmpdir.mkdir("both"),
+ monkeypatch,
+ track,
+ read_by=video_geotag_source.index("native"),
+ video_geotag_source=video_geotag_source,
+ geotag_source_path=Path(tmpdir) / "track.gpx",
+ )
+ own = self._sample(
+ tmpdir.mkdir("own"), monkeypatch, track, video_geotag_source=["native"]
+ )
+
+ assert len(samples) == 30
+ assert [s.name for s in samples] == [s.name for s in own]
+ for sample, own_sample in zip(samples, own):
+ exif, own_exif = exif_read.ExifRead(sample), exif_read.ExifRead(own_sample)
+ assert exif.extract_capture_time() == own_exif.extract_capture_time()
+ assert exif.extract_lon_lat() == own_exif.extract_lon_lat()
+
def test_none_left_out_of_a_video_track_file_where_the_video_source_may_geotag_them(
self, tmpdir, setup_mock, monkeypatch
):
@@ -1584,7 +1632,7 @@ def unreadable_gps(self, monkeypatch):
monkeypatch.setattr(
sample_video.factory,
"process",
- lambda paths, options: [
+ lambda paths, options, sources=None: [
describe_error_metadata(
error, filename=paths[0], filetype=FileType.GOPRO
)
@@ -1651,7 +1699,7 @@ def test_an_empty_track_is_also_an_error(self, tmpdir, setup_mock, monkeypatch):
monkeypatch.setattr(
sample_video.factory,
"process",
- lambda paths, options: [
+ lambda paths, options, sources=None: [
VideoMetadata(
filename=paths[0], filesize=0, filetype=FileType.GOPRO, points=[]
)
@@ -1667,7 +1715,7 @@ def test_a_missing_video_stream_is_also_an_error(
monkeypatch.setattr(
sample_video.factory,
"process",
- lambda paths, options: [
+ lambda paths, options, sources=None: [
VideoMetadata(
filename=paths[0],
filesize=0,
@@ -1780,6 +1828,8 @@ def _setup_mocks(
num_gps_points: int = 10,
gps_points: T.Sequence[geo.Point] | None = None,
sample_times: T.Sequence[float] | None = None,
+ # The index of the source that reads the track, of those tried
+ read_by: int = 0,
) -> dict[str, T.Any]:
"""Set up all the mocks needed for _sample_single_video_by_distance."""
probe_output = _load_probe_output()
@@ -1843,9 +1893,14 @@ def fake_extract_frames(
mock_ffmpeg_class,
)
+ def fake_process(paths, options, sources=None):
+ if sources is not None:
+ sources[paths[0]] = options[read_by]
+ return [video_metadata]
+
patches["geotag_cls"] = mock.patch(
"mapillary_tools.sample_video.factory.process",
- return_value=[video_metadata],
+ side_effect=fake_process,
)
patches["moov_parse"] = mock.patch.object(
@@ -2325,9 +2380,16 @@ def _file_track(self, video_times, rate: float = 1.0) -> list[geo.Point]:
]
def _sample_along_a_file_track(
- self, tmp_path: Path, gps_points: T.Sequence[geo.Point], **kwargs
+ self,
+ tmp_path: Path,
+ gps_points: T.Sequence[geo.Point],
+ read_by: int = 0,
+ **kwargs,
) -> list[datetime.datetime | None]:
- """Sample a frame a second by distance, and read their capture times"""
+ """
+ Sample a frame a second by distance, the track read by the source at
+ index **read_by** of those tried, and read their capture times
+ """
video_file = tmp_path / "test.mp4"
video_file.touch()
output_dir = tmp_path / "output"
@@ -2336,6 +2398,7 @@ def _sample_along_a_file_track(
video_file,
gps_points=gps_points,
sample_times=[float(t) for t in range(int(gps_points[-1].time) + 1)],
+ read_by=read_by,
)
with (
@@ -2354,9 +2417,11 @@ def _sample_along_a_file_track(
frames = sorted((output_dir / "test.mp4").glob("*.jpg"))
return [exif_read.ExifRead(frame).extract_capture_time() for frame in frames]
+ # Whatever source follows the one that read the track
+ @pytest.mark.parametrize("then", [[], ["native"], ["exiftool_runtime"]])
@pytest.mark.parametrize("source", ["gpx", "nmea"])
def test_frames_in_a_gap_of_a_video_track_file_are_left_out(
- self, tmp_path: Path, monkeypatch, caplog, source
+ self, tmp_path: Path, monkeypatch, caplog, source, then
) -> None:
"""
The track of --video_geotag_source has a gap of 10 seconds, over the
@@ -2368,7 +2433,9 @@ def test_frames_in_a_gap_of_a_video_track_file_are_left_out(
with caplog.at_level(logging.WARNING, logger=sample_video.LOG.name):
capture_times = self._sample_along_a_file_track(
- tmp_path, self._file_track(video_times), video_geotag_source=[source]
+ tmp_path,
+ self._file_track(video_times),
+ video_geotag_source=[source, *then],
)
assert capture_times == [
@@ -2424,6 +2491,28 @@ def test_frames_in_a_gap_are_left_where_the_geotagging_stage_places_them(
self.GPS_START_TIME + datetime.timedelta(seconds=t) for t in range(31)
]
+ @pytest.mark.parametrize(
+ "video_geotag_source",
+ # After a GPX that misses the video, or before a GPX
+ [["gpx", "native"], ["native", "gpx"]],
+ )
+ def test_frames_in_a_gap_of_a_track_read_by_the_video_are_kept(
+ self, tmp_path: Path, monkeypatch, video_geotag_source
+ ) -> None:
+ """The video's own track is sampled across its gaps, as without a GPX"""
+ monkeypatch.setattr(sample_video.constants, "MAX_GPX_GAP_SECONDS", 5.0)
+
+ capture_times = self._sample_along_a_file_track(
+ tmp_path,
+ self._file_track([*range(11), *range(20, 31)]),
+ read_by=video_geotag_source.index("native"),
+ video_geotag_source=video_geotag_source,
+ )
+
+ assert capture_times == [
+ self.GPS_START_TIME + datetime.timedelta(seconds=t) for t in range(31)
+ ]
+
def test_timelapse_frames_past_the_end_are_stamped_within_the_track(
self, tmp_path: Path
) -> None:
@@ -2700,7 +2789,7 @@ def test_unreadable_start_time_by_distance(
monkeypatch.setattr(
sample_video.factory,
"process",
- lambda paths, options: [
+ lambda paths, options, sources=None: [
VideoMetadata(
filename=paths[0],
filesize=1,
@@ -2877,7 +2966,7 @@ def test_the_sampler_asks_the_factory_for_them(
seen: list = []
points = _make_gps_points(4, time_step=1.0)
- def fake_process(paths, options):
+ def fake_process(paths, options, sources=None):
seen.append(list(options))
return [
VideoMetadata(
@@ -2967,7 +3056,7 @@ def test_every_option_carries_it(self):
def test_sample_video_passes_it_on(self, tmpdir, setup_mock, monkeypatch):
seen: list = []
- def fake_process(paths, options):
+ def fake_process(paths, options, sources=None):
seen.extend(options)
return [
VideoMetadata(
@@ -3006,7 +3095,7 @@ def test_sample_video_passes_the_interpolation_options_on(
"""video_process registers them for its geotagging stage."""
seen: list = []
- def fake_process(paths, options):
+ def fake_process(paths, options, sources=None):
seen.extend(options)
return [
VideoMetadata(
@@ -3077,7 +3166,7 @@ def test_it_fails_instead_of_writing_an_empty_sample_directory(
monkeypatch.setattr(
sample_video.factory,
"process",
- lambda paths, options: [
+ lambda paths, options, sources=None: [
VideoMetadata(
filename=paths[0],
filesize=1,