Read a video's start time from its GPS clock, and tell whether creation_time marks the start or the end

README.md

@@ -284,7 +284,11 @@ It is used to locate the images along the GPS tracks.
 mapillary_tools process MY_IMAGE_DIR --geotag_source "gpx" --geotag_source_path MY_EXTERNAL_GPS.gpx
 ```
 
-To geotag videos with a GPX file, video start time (video creation time minus video duration) is required to locate the sample images along the GPS tracks.
+To geotag videos with a GPX file, video start time is required to locate the sample images along the GPS tracks.
+It is read from the video's own GPS track when it has one, and otherwise from the video creation time.
+Cameras disagree on whether the creation time marks the start or the end of the recording (most dashcams write the end),
+so mapillary_tools uses the camera model or a date and time in the file name to tell which, and assumes the start when neither says.
+Use `--video_start_time` to override it, in UTC, when the sample images end up one video duration off along the track.
 
 ```sh
 # Geotagging with GPX works with interval-based sampling only,

mapillary_tools/blackvue_parser.py

@@ -51,15 +51,19 @@ def extract_blackvue_info(fp: T.BinaryIO) -> BlackVueInfo | None:
     if gps_data is None:
         return None
 
-    points = _parse_gps_box(gps_data)
+    points, recording_start_time = _parse_gps_box_with_start_time(gps_data)
     points.sort(key=lambda p: p.time)
 
     if points:
-        # Convert the time field to relative time to the first point
+        # Convert the time field to the video time, i.e. relative to the start
+        # of the recording. That is the first NMEA line the camera logged, not
+        # the first valid fix: until the receiver gets a fix, which can take
+        # minutes after a cold start, the camera logs lines without positions.
         # epoch_time stays as the original time in seconds
-        first_point_time = points[0].time
+        assert recording_start_time is not None
         for p in points:
-            p.time = p.time - first_point_time
+            # Rounding needed to avoid floating point precision issues
+            p.time = round(p.time - recording_start_time, 3)
 
     # Camera model
     try:
@@ -76,6 +80,22 @@ def extract_blackvue_info(fp: T.BinaryIO) -> BlackVueInfo | None:
     return BlackVueInfo(model=model, gps=points)
 
 
+def is_blackvue(fp: T.BinaryIO) -> bool:
+    """
+    Tell whether a video was recorded by a BlackVue dashcam, which writes its
+    GPS log and its camera model into boxes nested in a top-level free box,
+    whether or not it ever got a GPS fix
+    """
+    for path in [[b"free", b"gps "], [b"free", b"cprt"]]:
+        fp.seek(0)
+        try:
+            if sparser.parse_mp4_data_first(fp, path) is not None:
+                return True
+        except sparser.ParsingError:
+            pass
+    return False
+
+
 def _extract_camera_model_from_cprt(cprt_bytes: bytes) -> str:
     """
     >>> _extract_camera_model_from_cprt(b' {"model":"DR900X Plus","ver":0.918,"lang":"English","direct":1,"psn":"","temp":34,"GPS":1}')
@@ -254,6 +274,29 @@ def _parse_gps_box(gps_data: bytes) -> list[telemetry.GPSPoint]:
     >>> list(_parse_gps_box(b"[1623057074211]$GPVTG,,T,,M,0.078,N,0.144,K,D*28[1623057075215]"))
     []
     """
+    points, _ = _parse_gps_box_with_start_time(gps_data)
+    return points
+
+
+def _parse_gps_box_with_start_time(
+    gps_data: bytes,
+) -> tuple[list[telemetry.GPSPoint], float | None]:
+    """
+    Parse the GPS points, and the time of the first NMEA line in the same
+    corrected clock, which is when the recording started
+
+    >>> _parse_gps_box_with_start_time(b"[1623057074211]$GPGGA,202530.00,5109.0262,N,11401.8407,W,5,40,0.5,1097.36,M,-17.00,M,18,TSTR*61")[1]
+    1623097530.0
+    >>> points, start_time = _parse_gps_box_with_start_time(b'''
+    ... [1623057072211]$GPGGA,,,,,,0,00,99.99,,,,,,*48
+    ... [1623057073211]$GPRMC,,V,,,,,,,,,,N*53
+    ... [1623057074211]$GPGGA,202530.00,5109.0262,N,11401.8407,W,5,40,0.5,1097.36,M,-17.00,M,18,TSTR*61
+    ... ''')
+    >>> len(points), points[0].time - start_time
+    (1, 2.0)
+    >>> _parse_gps_box_with_start_time(b"")
+    ([], None)
+    """
     parsed_lines: list[tuple[float, pynmea2.NMEASentence]] = []
 
     # First pass: collect parsed_lines
@@ -264,6 +307,13 @@ def _parse_gps_box(gps_data: bytes) -> list[telemetry.GPSPoint]:
 
     timezone_offset = _detect_timezone_offset(parsed_lines)
 
+    if parsed_lines:
+        start_time: float | None = round(
+            min(epoch_sec for epoch_sec, _ in parsed_lines) + timezone_offset, 3
+        )
+    else:
+        start_time = None
+
     points_by_sentence_type: dict[str, list[telemetry.GPSPoint]] = {}
 
     # Second pass: apply offset to all GPS points
@@ -298,12 +348,12 @@ def _parse_gps_box(gps_data: bytes) -> list[telemetry.GPSPoint]:
 
     # This is the extraction order in exiftool
     if "RMC" in points_by_sentence_type:
-        return points_by_sentence_type["RMC"]
+        return points_by_sentence_type["RMC"], start_time
 
     if "GGA" in points_by_sentence_type:
-        return points_by_sentence_type["GGA"]
+        return points_by_sentence_type["GGA"], start_time
 
     if "GLL" in points_by_sentence_type:
-        return points_by_sentence_type["GLL"]
+        return points_by_sentence_type["GLL"], start_time
 
-    return []
+    return [], start_time

mapillary_tools/ffmpeg.py

@@ -9,6 +9,7 @@
 import datetime
 import json
 import logging
+import math
 import os
 import re
 import subprocess
@@ -55,8 +56,13 @@ class Stream(T.TypedDict):
     nb_frames: str
 
 
-class ProbeOutput(T.TypedDict):
+class Format(T.TypedDict, total=False):
+    tags: dict[str, str]
+
+
+class ProbeOutput(T.TypedDict, total=False):
     streams: list[Stream]
+    format: Format
 
 
 class FFmpegNotFoundError(Exception):
@@ -605,40 +611,63 @@ def __init__(self, probe_output: ProbeOutput) -> None:
         """
         self.probe_output = probe_output
 
-    def probe_video_start_time(self) -> datetime.datetime | None:
+    def probe_video_creation_time(self) -> datetime.datetime | None:
         """
-        Determine the start time of the video by analyzing stream metadata.
+        Read the creation time the camera stamped into the stream metadata.
 
-        Searches for creation time and duration information in video streams first,
-        then falls back to other stream types. Calculates start time as:
-        creation_time - duration
+        Searches video streams first, then falls back to other stream types.
+        Whether the creation time marks the start or the end of the recording
+        depends on the camera (see sample_video._creation_time_to_start_time).
 
         Returns:
-            Video start time as datetime object, or None if cannot be determined
+            Creation time as datetime object, or None if cannot be determined
 
         Note:
             Prioritizes video streams with highest resolution when multiple exist.
         """
-        streams = self.probe_output.get("streams", [])
+        for stream in self._iterate_streams_by_priority():
+            creation_time = self.extract_stream_creation_time(stream)
+            if creation_time is not None:
+                return creation_time
+
+        return None
+
+    def probe_video_duration(self) -> float | None:
+        """
+        Read the duration of the video in seconds from the stream metadata.
+
+        Searches the streams in the same order as probe_video_creation_time.
+
+        Returns:
+            Duration in seconds, or None if cannot be determined
+        """
+        for stream in self._iterate_streams_by_priority():
+            duration = self.extract_stream_duration(stream)
+            if duration is not None:
+                return duration
 
-        # Search start time from video streams
+        return None
+
+    def probe_format_tag(self, key: str) -> str | None:
+        """
+        Read a tag of the container, such as "make" or "model".
+
+        Returns:
+            The tag value, or None if the container does not have the tag
+        """
+        return self.probe_output.get("format", {}).get("tags", {}).get(key)
+
+    def _iterate_streams_by_priority(self) -> T.Generator[Stream, None, None]:
+        # Video streams by resolution, from the highest, then the other streams
         video_streams = self.probe_video_streams()
         video_streams.sort(
             key=lambda s: s.get("width", 0) * s.get("height", 0), reverse=True
         )
-        for stream in video_streams:
-            start_time = self.extract_stream_start_time(stream)
-            if start_time is not None:
-                return start_time
+        yield from video_streams
 
-        # Search start time from the other streams
-        for stream in streams:
+        for stream in self.probe_output.get("streams", []):
             if stream.get("codec_type") != "video":
-                start_time = self.extract_stream_start_time(stream)
-                if start_time is not None:
-                    return start_time
-
-        return None
+                yield stream
 
     def probe_video_streams(self) -> list[Stream]:
         """
@@ -671,36 +700,54 @@ def probe_video_with_max_resolution(self) -> Stream | None:
         return video_streams[0]
 
     @classmethod
-    def extract_stream_start_time(cls, stream: Stream) -> datetime.datetime | None:
+    def extract_stream_creation_time(cls, stream: Stream) -> datetime.datetime | None:
         """
-        Calculate the start time of a specific stream.
-
-        Determines start time by subtracting stream duration from creation time:
-        start_time = creation_time - duration
+        Read the creation time of a specific stream.
 
         Args:
-            stream: Stream dictionary containing metadata including tags and duration
+            stream: Stream dictionary containing metadata including tags
 
         Returns:
-            Stream start time as datetime object, or None if required metadata is missing
+            Creation time as datetime object, or None if it is missing or malformed
 
         Note:
             Handles multiple datetime formats including ISO format and custom patterns.
         """
-        duration_str = stream.get("duration")
-        LOG.debug("Extracted video duration: %s", duration_str)
-        if duration_str is None:
-            return None
-        duration = float(duration_str)
-
         creation_time_str = stream.get("tags", {}).get("creation_time")
         LOG.debug("Extracted video creation time: %s", creation_time_str)
         if creation_time_str is None:
             return None
         try:
-            creation_time = datetime.datetime.fromisoformat(creation_time_str)
+            return datetime.datetime.fromisoformat(creation_time_str)
         except ValueError:
-            creation_time = datetime.datetime.strptime(
+            pass
+        try:
+            return datetime.datetime.strptime(
                 creation_time_str, "%Y-%m-%dT%H:%M:%S.%f%z"
             )
-        return creation_time - datetime.timedelta(seconds=duration)
+        except ValueError:
+            LOG.warning("Ignoring malformed video creation time: %s", creation_time_str)
+            return None
+
+    @classmethod
+    def extract_stream_duration(cls, stream: Stream) -> float | None:
+        """
+        Read the duration of a specific stream in seconds.
+
+        Args:
+            stream: Stream dictionary containing metadata
+
+        Returns:
+            Duration in seconds, or None if it is missing or malformed
+        """
+        duration_str = stream.get("duration")
+        LOG.debug("Extracted video duration: %s", duration_str)
+        if duration_str is None:
+            return None
+        try:
+            duration = float(duration_str)
+        except ValueError:
+            return None
+        if not math.isfinite(duration) or duration < 0:
+            return None
+        return duration