feat: add a new GPMF (GoPro) parser (#529)

* feat: add a new gpmf (GoPro) parser

* use this new parser for video_process

* refactor

* fix tests

* fix types

* usort

* types

Author: ptpt

mapillary_tools/constants.py

@@ -1,4 +1,5 @@
 import os
+import typing as T
 
 import appdirs
 
@@ -22,3 +23,9 @@
 )
 MAX_SEQUENCE_LENGTH = int(os.getenv(_ENV_PREFIX + "MAX_SEQUENCE_LENGTH", 500))
 USER_DATA_DIR = appdirs.user_data_dir(appname="mapillary_tools", appauthor="Mapillary")
+# This is DoP value, the lower the better
+# See https://github.com/gopro/gpmf-parser#hero5-black-with-gps-enabled-adds
+GOPRO_MAX_GPS_PRECISION = int(os.getenv(_ENV_PREFIX + "GOPRO_MAX_GPS_PRECISION", 1000))
+GOPRO_GPS_FIXES: T.Set[int] = set(
+    int(fix) for fix in os.getenv(_ENV_PREFIX + "GOPRO_GPS_FIXES", "2,3").split(",")
+)

mapillary_tools/geo.py

@@ -167,7 +167,7 @@ def as_unix_time(dt: T.Union[datetime.datetime, int, float]) -> float:
     return aware_dt.timestamp()
 
 
-def interpolate(points: T.List[Point], t: float) -> Point:
+def interpolate(points: T.Sequence[Point], t: float) -> Point:
     """
     Interpolate or extrapolate the point at time t along the sequence of points (sorted by time).
     """

mapillary_tools/geotag/geotag_from_exif.py

@@ -14,7 +14,7 @@
 
 
 class GeotagFromEXIF(GeotagFromGeneric):
-    def __init__(self, image_dir: str, images: T.List[str]):
+    def __init__(self, image_dir: str, images: T.Sequence[str]):
         self.image_dir = image_dir
         self.images = images
         super().__init__()

mapillary_tools/geotag/geotag_from_gopro.py

@@ -1,20 +1,15 @@
-import datetime
 import logging
 import os
-import sys
-import tempfile
 import typing as T
 from pathlib import Path
 
 from tqdm import tqdm
 
-from .. import constants, exceptions, ffmpeg as ffmpeglib, geo, types, utils
-from ..geo import get_max_distance_from_start, gps_distance, pairwise
-from . import utils as geotag_utils
+from .. import constants, exceptions, geo, types, utils
+from . import gpmf_parser, utils as geotag_utils
 from .geotag_from_generic import GeotagFromGeneric
 
 from .geotag_from_gpx import GeotagFromGPXWithProgress
-from .gpmf import interpolate_times, parse_bin
 
 
 LOG = logging.getLogger(__name__)
@@ -25,7 +20,6 @@ def __init__(
         self,
         image_dir: str,
         source_path: str,
-        use_gpx_start_time: bool = False,
         offset_time: float = 0.0,
     ):
         self.image_dir = image_dir
@@ -34,10 +28,43 @@ def __init__(
         else:
             # it is okay to not suffix with .mp4
             self.videos = [source_path]
-        self.use_gpx_start_time = use_gpx_start_time
         self.offset_time = offset_time
         super().__init__()
 
+    def _filter_noisy_points(
+        self, points: T.Sequence[gpmf_parser.PointWithFix], video: Path
+    ) -> T.Sequence[gpmf_parser.PointWithFix]:
+        num_points = len(points)
+        points = [
+            p
+            for p in points
+            if p.gps_fix is not None and p.gps_fix.value in constants.GOPRO_GPS_FIXES
+        ]
+        if len(points) < num_points:
+            LOG.warning(
+                "Removed %d points with the GPS fix not in %s from %s",
+                num_points - len(points),
+                constants.GOPRO_GPS_FIXES,
+                video,
+            )
+
+        num_points = len(points)
+        points = [
+            p
+            for p in points
+            if p.gps_precision is not None
+            and p.gps_precision <= constants.GOPRO_MAX_GPS_PRECISION
+        ]
+        if len(points) < num_points:
+            LOG.warning(
+                "Removed %d points with DoP value higher than %d from %s",
+                num_points - len(points),
+                constants.GOPRO_MAX_GPS_PRECISION,
+                video,
+            )
+
+        return points
+
     def to_description(self) -> T.List[types.ImageDescriptionFileOrError]:
         descs: T.List[types.ImageDescriptionFileOrError] = []
 
@@ -55,11 +82,13 @@ def to_description(self) -> T.List[types.ImageDescriptionFileOrError]:
             if not sample_images:
                 continue
 
-            points = get_points_from_gpmf(Path(video))
+            points = self._filter_noisy_points(
+                gpmf_parser.parse_gpx(Path(video)), Path(video)
+            )
 
             # bypass empty points to raise MapillaryGPXEmptyError
             if points and geotag_utils.is_video_stationary(
-                get_max_distance_from_start([(p.lat, p.lon) for p in points])
+                geo.get_max_distance_from_start([(p.lat, p.lon) for p in points])
             ):
                 LOG.warning(
                     "Fail %d sample images due to stationary video %s",
@@ -85,103 +114,11 @@ def to_description(self) -> T.List[types.ImageDescriptionFileOrError]:
                     self.image_dir,
                     sample_images,
                     points,
-                    use_gpx_start_time=self.use_gpx_start_time,
+                    use_gpx_start_time=False,
+                    use_image_start_time=True,
                     offset_time=self.offset_time,
                     progress_bar=pbar,
                 )
                 descs.extend(geotag.to_description())
 
         return descs
-
-
-def extract_and_parse_bin(path: Path) -> T.List:
-    ffmpeg = ffmpeglib.FFMPEG(constants.FFMPEG_PATH, constants.FFPROBE_PATH)
-    probe = ffmpeg.probe_format_and_streams(path)
-
-    format_name = probe["format"]["format_name"].lower()
-    if "mp4" not in format_name:
-        raise IOError("File must be an mp4")
-
-    stream_id = None
-    for stream in probe["streams"]:
-        if (
-            "codec_tag_string" in stream
-            and "gpmd" in stream["codec_tag_string"].lower()
-        ):
-            stream_id = stream["index"]
-
-    if stream_id is None:
-        raise IOError("No GoPro metadata track found - was GPS turned on?")
-
-    # https://github.com/mapillary/mapillary_tools/issues/503
-    if sys.platform == "win32":
-        delete = False
-    else:
-        delete = True
-
-    with tempfile.NamedTemporaryFile(delete=delete) as tmp:
-        try:
-            LOG.debug("Extracting GoPro stream %s to %s", stream_id, tmp.name)
-            ffmpeg.extract_stream(path, Path(tmp.name), stream_id)
-            LOG.debug("Parsing GoPro GPMF %s", tmp.name)
-            return parse_bin(tmp.name)
-        finally:
-            if not delete:
-                try:
-                    os.remove(tmp.name)
-                except FileNotFoundError:
-                    pass
-
-
-def get_points_from_gpmf(path: Path) -> T.List[geo.Point]:
-    gpmf_data = extract_and_parse_bin(path)
-
-    rows = len(gpmf_data)
-
-    points: T.List[geo.Point] = []
-    for i, frame in enumerate(gpmf_data):
-        t = frame["time"]
-
-        if i < rows - 1:
-            next_ts = gpmf_data[i + 1]["time"]
-        else:
-            next_ts = t + datetime.timedelta(seconds=1)
-
-        interpolate_times(frame, next_ts)
-
-        for point in frame["gps"]:
-            points.append(
-                geo.Point(
-                    time=geo.as_unix_time(point["time"]),
-                    lat=point["lat"],
-                    lon=point["lon"],
-                    alt=point["alt"],
-                    angle=None,
-                )
-            )
-
-    return points
-
-
-if __name__ == "__main__":
-    import sys
-
-    points = get_points_from_gpmf(Path(sys.argv[1]))
-    gpx = geotag_utils.convert_points_to_gpx(points)
-    print(gpx.to_xml())
-
-    LOG.setLevel(logging.INFO)
-    handler = logging.StreamHandler(sys.stderr)
-    handler.setLevel(logging.INFO)
-    LOG.addHandler(handler)
-    LOG.info(
-        "Stationary: %s",
-        geotag_utils.is_video_stationary(
-            get_max_distance_from_start([(p.lat, p.lon) for p in points])
-        ),
-    )
-    distance = sum(
-        gps_distance((cur.lat, cur.lon), (nex.lat, nex.lon))
-        for cur, nex in pairwise(points)
-    )
-    LOG.info("Total distance: %f", distance)

mapillary_tools/geotag/geotag_from_gpx.py

@@ -20,8 +20,8 @@ class GeotagFromGPX(GeotagFromGeneric):
     def __init__(
         self,
         image_dir: str,
-        images: T.List[str],
-        points: T.List[geo.Point],
+        images: T.Sequence[str],
+        points: T.Sequence[geo.Point],
         use_gpx_start_time: bool = False,
         use_image_start_time: bool = False,
         offset_time: float = 0.0,
@@ -172,8 +172,8 @@ class GeotagFromGPXWithProgress(GeotagFromGPX):
     def __init__(
         self,
         image_dir: str,
-        images: T.List[str],
-        points: T.List[geo.Point],
+        images: T.Sequence[str],
+        points: T.Sequence[geo.Point],
         use_gpx_start_time: bool = False,
         use_image_start_time: bool = False,
         offset_time: float = 0.0,

mapillary_tools/geotag/geotag_from_nmea_file.py

@@ -12,7 +12,7 @@ class GeotagFromNMEAFile(GeotagFromGPX):
     def __init__(
         self,
         image_dir: str,
-        images: T.List[str],
+        images: T.Sequence[str],
         source_path: str,
         use_gpx_start_time: bool = False,
         offset_time: float = 0.0,