Judge image capture times in UTC, and fail those no date can be written in

mapillary_tools/exif_read.py

@@ -320,28 +320,79 @@ 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
     >>> is_valid_capture_time(datetime.datetime(1, 1, 1))
     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("<x/>")))
+        >>> 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:

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:

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)
 

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),
             )

mapillary_tools/geotag/geotag_images_from_gpx_file.py

@@ -48,25 +48,30 @@ 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. "
             "Replace it with a valid GPX file."
         ) 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(