From 2a9087078e3a00933e33cc09a38e04233a7d7d93 Mon Sep 17 00:00:00 2001 From: Caglar Pir Date: Wed, 23 Sep 2026 12:43:44 +0200 Subject: [PATCH 1/2] Parse Novatek freeGPS natively so distance sampling can use it Distance sampling reads GPS only through the native extractors (GoPro, CAMM, BlackVue). Novatek based dashcams -- Viofo A129/A139, Rove R2-4K Pro, AZDome GS63H, Vantrue N4, Anker Roav C1 Pro -- store GPS in "freeGPS " blocks indexed by a moov/"gps " box, which none of them parse. So `video_process --video_sample_distance` finds no GPS in these videos, even though `process` geotags them fine through the ExifTool fallback. Add novatek_parser, tried last by NativeVideoExtractor. ExifTool tells the vendor layouts apart by probing each block in a fixed order (ProcessFreeGPS). The parser probes in the same order and decodes only the three layouts these cameras use, ExifTool GPSType 1, 3 and 15. The points are the ones the ExifTool fallback produces: same date handling, same dedupe/sort/rebase as _aggregate_gps_track, and numbers rounded to the 15 significant digits ExifTool prints. Anything the parser is not certain about returns None and still falls through to ExifTool as before: other layouts, blocks that also parse as Nextbase records, videos mixing layouts, type 3 timestamps that ExifTool converts from the time zone of the machine it runs on, and non-finite numbers. Failure raises MapillaryVideoGPSNotFoundError, which the geotag factory treats as reprocessable, so the ExifTool fallback in `process` is unchanged. Checked against ExifTool 13.40 on 669 dashcam and action camera videos: the 27 Novatek videos ExifTool reads now extract natively with points identical to ExifTool's, no other video changes, and `process` writes identical descriptions for all 35 videos that have a "gps " box. --- .../geotag/video_extractors/native.py | 37 +- mapillary_tools/novatek_parser.py | 419 ++++++++++++++++++ tests/unit/test_novatek_parser.py | 253 +++++++++++ 3 files changed, 708 insertions(+), 1 deletion(-) create mode 100644 mapillary_tools/novatek_parser.py create mode 100644 tests/unit/test_novatek_parser.py diff --git a/mapillary_tools/geotag/video_extractors/native.py b/mapillary_tools/geotag/video_extractors/native.py index a4a329e7..a1f1c532 100644 --- a/mapillary_tools/geotag/video_extractors/native.py +++ b/mapillary_tools/geotag/video_extractors/native.py @@ -14,7 +14,15 @@ else: from typing_extensions import override -from ... import blackvue_parser, exceptions, geo, telemetry, types, utils +from ... import ( + blackvue_parser, + exceptions, + geo, + novatek_parser, + telemetry, + types, + utils, +) from ...camm import camm_parser from ...gpmf import gpmf_gps_filter, gpmf_parser from ...mp4 import construct_mp4_parser, simple_mp4_parser @@ -105,6 +113,26 @@ def extract(self) -> types.VideoMetadata: return video_metadata +class NovatekVideoExtractor(BaseVideoExtractor): + @override + def extract(self) -> types.VideoMetadata: + with self.video_path.open("rb") as fp: + points = novatek_parser.extract_points(fp) + + # Unsupported or empty GPS data is left to ExifTool + if not points: + raise exceptions.MapillaryVideoGPSNotFoundError( + "No GPS data found from the video" + ) + + return types.VideoMetadata( + filename=self.video_path, + filesize=utils.get_file_size(self.video_path), + filetype=types.FileType.VIDEO, + points=T.cast(T.List[geo.Point], points), + ) + + class NativeVideoExtractor(BaseVideoExtractor): def __init__(self, video_path: Path, filetypes: set[types.FileType] | None = None): super().__init__(video_path) @@ -160,6 +188,13 @@ def extract(self) -> types.VideoMetadata: except exceptions.MapillaryVideoGPSNotFoundError: pass + if ft is None or types.FileType.VIDEO in ft: + extractor = NovatekVideoExtractor(self.video_path) + try: + return extractor.extract() + except exceptions.MapillaryVideoGPSNotFoundError: + pass + raise exceptions.MapillaryVideoGPSNotFoundError( "No GPS data found from the video" ) diff --git a/mapillary_tools/novatek_parser.py b/mapillary_tools/novatek_parser.py new file mode 100644 index 00000000..47f3c777 --- /dev/null +++ b/mapillary_tools/novatek_parser.py @@ -0,0 +1,419 @@ +# 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. + +""" +Extract GPS from the "freeGPS " blocks written by Novatek based dashcams. + +The moov/"gps " box indexes the "freeGPS " blocks stored in mdat. The block layout +varies by vendor. ExifTool (ProcessFreeGPS in QuickTimeStream.pl) tells the layouts +apart by probing them in a fixed order, and numbers them with its GPSType. + +This parser probes the blocks in the same order and decodes only these layouts: +- Type 1: XOR encrypted text (AZDome GS63H) +- Type 3: float32 values (Viofo A129, Viofo A139, Anker Roav C1 Pro) +- Type 15: float64 values (Vantrue N4, Rove R2-4K Pro) + +The decoded points match what the ExifTool fallback extracts from the same video. +If a video contains any block that can't be decoded that way, extract_points returns +None and the video is left to ExifTool. +""" + +from __future__ import annotations + +import datetime +import math +import re +import struct +import typing as T + +from . import telemetry +from .mp4 import simple_mp4_parser as sparser + + +# ExifTool ignores smaller blocks +_MIN_BLOCK_SIZE = 82 + +# Enough to probe and decode every layout +_MAX_BLOCK_READ_SIZE = 0x1000 + +_FREEGPS_SIGNATURE = b"freeGPS " + +_TYPE1_KEY = b"\xaa\xaa\xf2\xe1\xf0\xee\x54\x54" +_TYPE1_GPS = re.compile( + rb"^.{8}(\d{4})(\d{2})(\d{2})(\d{2})(\d{2})(\d{2}).(.{15})([NS])(\d{8})([EW])(\d{9})", + re.S, +) +_TYPE1_DATE = re.compile( + rb"^.{8}(\d{4})(\d{2})(\d{2})(\d{2})(\d{2})(\d{2}).(.{15})", re.S +) +_TYPE1_ACC = re.compile(rb"^.{65}([-+]\d{3}){3}", re.S) +_TYPE1_ACC_AZDOME = re.compile(rb"^.{173}([-+]\d{3}){3}", re.S) + +_TYPE2 = re.compile(rb"^.{52}\d{14}", re.S) +_TYPE3 = re.compile(rb"^.{37}\x00\x00\x00A([NS])([EW])\x00", re.S) +_TYPE3_BASE64 = re.compile(rb"^[A-Za-z0-9+/]{8,20}={0,2}\x00*$") +_TYPE3_DECIMAL = re.compile(rb"^\d{1,5}\.\d+\x00*$") +_TYPE5 = re.compile(rb"^(.{16}|.{48}|.{80})LIGOGPSINFO\x00", re.S) +_TYPE6 = re.compile(rb"^.{60}A\x00{3}.{4}[NS]\x00{3}.{4}[EW]\x00{3}", re.S) +_TYPE7 = re.compile(rb"^.{60}4W`b]S<", re.S) +_TYPE8 = re.compile( + rb"^.{64}[\x01-\x0c]\x00{3}[\x01-\x1f]\x00{3}A[NS][EW]\x00{5}", re.S +) +_TYPE9 = re.compile(rb"^.{12}\xac\x00\x00\x00.{116}", re.S) +_TYPE10 = re.compile(rb"^.{64}A[NS][EW]\x00", re.S) +_TYPE12 = re.compile(rb"^.{60}A\x00.{10}[NS]\x00.{14}[EW]\x00", re.S) +_TYPE13 = re.compile(rb"^.{16}A[NS][EW]\x00", re.S) +_TYPE14 = re.compile(rb"^.{20}[\x00-\x18][\x00-\x3b]{2}[\x00-\x09]A[NS][EW]", re.S) +_TYPE15 = re.compile(rb"^.{28}A.{11}([NS]).{15}([EW])", re.S) +_TYPE16 = re.compile(rb"^.{72}A[NS][EW]\x00", re.S) +_TYPE18 = re.compile(rb"^.{23}\d{4}/\d{2}/\d{2} \d{2}:\d{2}:\d{2} [N|S]", re.S) +# No re.S here, as in ExifTool +_TYPE19 = re.compile(rb"^.{30}A.{20}VV") + + +class _UnsupportedBlockError(Exception): + pass + + +def extract_points(fp: T.BinaryIO) -> list[telemetry.GPSPoint] | None: + """ + Return the GPS points sorted by time, or None if the video has no "gps " box, + no points, or any block this parser does not support. + Like other parsers, point time is relative to the first point, + and epoch_time is the Unix time. + """ + try: + gps_box = sparser.parse_mp4_data_first(fp, [b"moov", b"gps "]) + except sparser.ParsingError: + return None + + if gps_box is None: + return None + + gps_types: set[int] = set() + points: list[telemetry.GPSPoint] = [] + + for block in _read_blocks(fp, _parse_gps_box(gps_box)): + gps_type = _probe_block(block) + gps_types.add(gps_type) + + if gps_type == 20: + # ExifTool falls back to type 20 (Nextbase binary records) for anything unknown, + # including blocks without a fix. It extracts nothing unless the first record is valid. + if _has_valid_nextbase_record(block): + return None + continue + + decode = _DECODERS.get(gps_type) + if decode is None: + return None + + try: + point = decode(block) + except _UnsupportedBlockError: + return None + + if point is None: + continue + + if not points or _point_key(points[-1]) != _point_key(point): + points.append(point) + + # ExifTool aggregates GPSTrack by position, so it misaligns the track values + # if some blocks have them (types 3 and 15) and others don't (type 1) + if len(gps_types & _DECODERS.keys()) > 1: + return None + + if not points: + return None + + points.sort(key=lambda p: p.time) + + deduplicated: list[telemetry.GPSPoint] = [] + for point in points: + if not deduplicated or _point_key(deduplicated[-1]) != _point_key(point): + deduplicated.append(point) + + first_point_time = deduplicated[0].time + for point in deduplicated: + point.time = point.time - first_point_time + + return deduplicated + + +def _parse_gps_box(data: bytes) -> list[tuple[int, int]]: + """ + Parse the "gps " box into a list of (offset, size) of the "freeGPS " blocks + + >>> _parse_gps_box(bytes.fromhex("00000101 00000002 00001000 00008000 00009000 00008000")) + [(4096, 32768), (36864, 32768)] + + The count is limited by the box size, as in ExifTool + >>> _parse_gps_box(bytes.fromhex("00000101 00000009 00001000 00008000 00009000")) + [(4096, 32768)] + >>> _parse_gps_box(bytes.fromhex("00000101 00000009")) + [] + """ + if len(data) <= 8: + return [] + (count,) = struct.unpack_from(">I", data, 4) + count = min(count, (len(data) - 8) // 8) + return [struct.unpack_from(">II", data, 8 + i * 8) for i in range(count)] + + +def _read_blocks( + fp: T.BinaryIO, index: list[tuple[int, int]] +) -> T.Generator[bytes, None, None]: + for offset, size in index: + fp.seek(offset, 0) + # The layouts can be told apart by the first bytes and a few size thresholds, + # all smaller than _MAX_BLOCK_READ_SIZE, so reading more doesn't change the result + block = fp.read(min(size, _MAX_BLOCK_READ_SIZE)) + if block[4:12] != _FREEGPS_SIGNATURE: + continue + if len(block) < _MIN_BLOCK_SIZE: + continue + yield block + + +def _probe_block(block: bytes) -> int: + """ + Return ExifTool's GPSType of the block, i.e. the first layout that matches it + """ + if block[18:26] == _TYPE1_KEY: + return 1 + if _TYPE2.match(block): + return 2 + if _TYPE3.match(block): + return 3 if _is_type3_binary(block) else 4 + matched = _TYPE5.match(block) + if matched and len(block) >= len(matched.group(1)) + 0x84: + return 5 + if _TYPE6.match(block): + return 6 + if _TYPE7.match(block) and len(block) >= 140: + return 7 + if _TYPE8.match(block): + return 8 + if _TYPE9.match(block): + return 9 + if _TYPE10.match(block): + return 10 + if block[0x45:0x48] == b"ATC": + return 11 + if _TYPE12.match(block) and len(block) >= 0x88: + return 12 + if _TYPE13.match(block): + return 13 + if _TYPE14.match(block): + return 14 + if _TYPE15.match(block): + return 15 + if _TYPE16.match(block): + # Types 16 and 17 + return 16 + if _TYPE18.match(block): + return 18 + if _TYPE19.match(block): + return 19 + return 20 + + +def _is_type3_binary(block: bytes) -> bool: + """ + Type 3 stores lat/lon as floats, while type 4 (E-ACE B44) stores them + as decimal or base64 encoded (and encrypted) strings in the same place + """ + if len(block) < 0x78: + return True + lat, lon = block[0x2C : 0x2C + 20], block[0x40 : 0x40 + 20] + is_base64 = all(_TYPE3_BASE64.match(s) for s in (lat, lon)) + is_decimal = all(_TYPE3_DECIMAL.match(s) for s in (lat, lon)) + return not is_base64 and not is_decimal + + +def _has_valid_nextbase_record(block: bytes) -> bool: + """ + Check the first record in the same way as ExifTool + """ + year, month, day, hour, minute, second = struct.unpack_from(">HBBBBH", block, 0x36) + return ( + 2000 <= year <= 2200 + and 1 <= month <= 12 + and 1 <= day <= 31 + and hour <= 59 + and minute <= 59 + and second <= 600 + ) + + +def _decode_type1(block: bytes) -> telemetry.GPSPoint | None: + n = min(len(block) - 18, 0x101) + decrypted = bytes(b ^ 0xAA for b in block[18 : 18 + n]) + + matched = _TYPE1_GPS.match(decrypted) + if matched is None: + # AZDome may store the date without GPS, which ExifTool extracts + # when the accelerometer data is found at the AZDome location + if ( + not _TYPE1_ACC.match(decrypted) + and _TYPE1_ACC_AZDOME.match(decrypted) + and _TYPE1_DATE.match(decrypted) + ): + raise _UnsupportedBlockError("Date without GPS") + return None + + year, month, day, hour, minute, second = ( + int(g) for g in matched.group(1, 2, 3, 4, 5, 6) + ) + return _build_point( + (year, month, day, hour, minute, second), + lat=int(matched.group(9)) / 1e4, + lat_ref=matched.group(8), + lon=int(matched.group(11)) / 1e4, + lon_ref=matched.group(10), + angle=None, + ) + + +def _decode_type3(block: bytes) -> telemetry.GPSPoint | None: + matched = _TYPE3.match(block) + assert matched is not None + hour, minute, second, year, month, day = struct.unpack_from("<6I", block, 16) + if year >= 2000: + # ExifTool assumes it is local time (Kenwood) and converts it to UTC + # using the time zone of the machine it runs on + raise _UnsupportedBlockError("Local time") + lat, lon, _speed, track = struct.unpack_from("<4f", block, 0x2C) + return _build_point( + (year, month, day, hour, minute, second), + lat=lat, + lat_ref=matched.group(1), + lon=lon, + lon_ref=matched.group(2), + angle=track, + ) + + +def _decode_type15(block: bytes) -> telemetry.GPSPoint | None: + matched = _TYPE15.match(block) + assert matched is not None + try: + hour, minute, second = struct.unpack_from("<3I", block, 16) + year, month, day = struct.unpack_from("<3I", block, 80) + (lat,) = struct.unpack_from(" telemetry.GPSPoint | None: + year, month, day, hour, minute, second = date_time + + # ExifTool drops the block + if not 1 <= month <= 12: + return None + + if year < 2000: + year += 2000 + + epoch_time = _epoch_time(year, month, day, hour, minute, second) + if epoch_time is None: + return None + + if not all(math.isfinite(v) for v in (lat, lon, 0.0 if angle is None else angle)): + raise _UnsupportedBlockError("Invalid number") + + lat = _to_exiftool_number(_ddmm_to_degrees(lat) * (-1 if lat_ref == b"S" else 1)) + lon = _to_exiftool_number(_ddmm_to_degrees(lon) * (-1 if lon_ref == b"W" else 1)) + if angle is not None: + angle = _to_exiftool_number(angle) + + return telemetry.GPSPoint( + time=epoch_time, + lat=lat, + lon=lon, + alt=None, + angle=angle, + epoch_time=epoch_time, + fix=None, + precision=None, + ground_speed=None, + ) + + +def _epoch_time( + year: int, month: int, day: int, hour: int, minute: int, second: int +) -> float | None: + """ + Convert the date and time to Unix time in the same way as parsing the GPSDateTime + extracted by ExifTool: invalid dates are rejected, while the time overflows into the date + + >>> _epoch_time(2023, 1, 6, 15, 5, 58) + 1673017558.0 + >>> _epoch_time(2023, 1, 6, 23, 59, 60) + 1673049600.0 + >>> _epoch_time(2023, 2, 30, 1, 2, 3) is None + True + """ + try: + dt = datetime.datetime(year, month, day, tzinfo=datetime.timezone.utc) + except ValueError: + return None + try: + dt = dt + datetime.timedelta(hours=hour, minutes=minute, seconds=second) + except OverflowError as ex: + raise _UnsupportedBlockError("Invalid time") from ex + return dt.timestamp() + + +def _ddmm_to_degrees(value: float) -> float: + """ + Convert DDDMM.MMMM to degrees in the same way as ExifTool (ConvertLatLon) + + >>> _ddmm_to_degrees(4721.35197) + 47.355866166666665 + """ + degrees = int(value / 100) + return degrees + (value - degrees * 100) / 60 + + +def _to_exiftool_number(value: float) -> float: + """ + ExifTool prints numbers with 15 significant digits (Perl's default), + so round the same way to get identical points + + >>> _to_exiftool_number(0.1 + 0.2) + 0.3 + """ + return float(f"{value:.15g}") + + +def _point_key(point: telemetry.GPSPoint) -> tuple: + # Same as how the ExifTool extractor tells duplicate points + return (point.time, point.lon, point.lat, point.epoch_time, point.angle) diff --git a/tests/unit/test_novatek_parser.py b/tests/unit/test_novatek_parser.py new file mode 100644 index 00000000..fbab848d --- /dev/null +++ b/tests/unit/test_novatek_parser.py @@ -0,0 +1,253 @@ +# 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. + +from __future__ import annotations + +import io +import struct +from pathlib import Path + +import pytest + +from mapillary_tools import exceptions, novatek_parser, telemetry, types +from mapillary_tools.geotag.video_extractors.native import NativeVideoExtractor + + +def _box(box_type: bytes, data: bytes) -> bytes: + return struct.pack(">I", 8 + len(data)) + box_type + data + + +def _build_video(blocks: list[bytes]) -> bytes: + """ + Build a video with the blocks in mdat, indexed by moov/"gps " as Novatek does + """ + ftyp = _box(b"ftyp", b"isom\x00\x00\x02\x00isomiso2mp41") + index = b"" + offset = len(ftyp) + 8 + for block in blocks: + index += struct.pack(">II", offset, len(block)) + offset += len(block) + gps = struct.pack(">II", 0x101, len(blocks)) + index + return ftyp + _box(b"mdat", b"".join(blocks)) + _box(b"moov", _box(b"gps ", gps)) + + +def _freegps_block(fields: dict[int, bytes], size: int = 0x100) -> bytes: + block = bytearray(size) + block[0:12] = struct.pack(">I", size) + b"freeGPS " + for offset, data in fields.items(): + block[offset : offset + len(data)] = data + return bytes(block) + + +def _type1_block( + date_time: bytes = b"20180924224928", + lat: bytes = b"N40464350", + lon: bytes = b"W007040308", + tail: bytes = b"00000007", +) -> bytes: + # AZDome GS63H: text XOR encrypted from byte 18 + plain = ( + b"\x00\x00XKZD\xfe\xfe" + date_time + b"\x0c5567GP \x00\x00\x00\x00\x00\x03" + ) + plain += lat + lon + tail + return _freegps_block( + { + 12: b"\x05\x01\x00\x00\x01\x03", + 18: bytes(b ^ 0xAA for b in plain.ljust(0x100 - 18, b"\x00")), + } + ) + + +def _type3_block( + hms: tuple[int, int, int] = (5, 47, 3), + ymd: tuple[int, int, int] = (19, 9, 27), + lat: tuple[float, bytes] = (4922.143, b"N"), + lon: tuple[float, bytes] = (12305.985, b"W"), + track: float = 16.49, +) -> bytes: + # Viofo A129: float32 values + return _freegps_block( + { + 16: struct.pack("<6I", *hms, *ymd), + 0x28: b"A" + lat[1] + lon[1] + b"\x00", + 0x2C: struct.pack("<4f", lat[0], lon[0], 26.3, track), + } + ) + + +def _type15_block( + hms: tuple[int, int, int] = (13, 22, 30), + ymd: tuple[int, int, int] = (22, 12, 14), + lat: tuple[float, bytes] = (4721.35197, b"N"), + lon: tuple[float, bytes] = (830.80859, b"E"), + track: float = 199.88, +) -> bytes: + # Vantrue N4: float64 values + return _freegps_block( + { + 16: struct.pack("<3I", *hms), + 28: b"A", + 32: struct.pack(" list[telemetry.GPSPoint] | None: + return novatek_parser.extract_points(io.BytesIO(_build_video(blocks))) + + +# The expected points below are the same as extracted by ExifTool 13.40 from the same videos +def _point(time, lat, lon, angle, epoch_time) -> telemetry.GPSPoint: + return telemetry.GPSPoint( + time=time, + lat=lat, + lon=lon, + alt=None, + angle=angle, + epoch_time=epoch_time, + fix=None, + precision=None, + ground_speed=None, + ) + + +def test_type15(): + next_block = _type15_block( + hms=(13, 22, 31), lat=(4721.3524, b"N"), lon=(830.8082, b"E"), track=198.5 + ) + points = _extract( + [ + NO_FIX_BLOCK, + next_block, + # Out of order, in the southern and western hemispheres + _type15_block(lat=(4721.35197, b"S"), lon=(830.80859, b"W")), + # Duplicate + next_block, + ] + ) + assert points == [ + _point(0.0, -47.3558661666667, -8.5134765, 199.88, 1671024150.0), + _point(1.0, 47.3558733333333, 8.51347, 198.5, 1671024151.0), + ] + + +def test_type3(): + points = _extract( + [ + NO_FIX_BLOCK, + _type3_block(), + _type3_block( + hms=(5, 47, 4), + lat=(4922.155, b"N"), + lon=(12305.995, b"W"), + track=16.5, + ), + ] + ) + assert points == [ + _point( + 0.0, 49.3690511067708, -123.099755859375, 16.4899997711182, 1569563223.0 + ), + _point(1.0, 49.3692464192708, -123.099918619792, 16.5, 1569563224.0), + ] + + +def test_type3_local_time(): + # ExifTool converts the time to UTC by the time zone of the machine it runs on + assert _extract([_type3_block(ymd=(2019, 9, 27))]) is None + + +def test_type1(): + points = _extract( + [ + _type1_block(), + _type1_block( + date_time=b"20180924224929", lat=b"N40464355", lon=b"W007040310" + ), + ] + ) + assert points == [ + _point(0.0, 40.7739166666667, -7.06718, None, 1537829368.0), + _point(1.0, 40.773925, -7.06718333333333, None, 1537829369.0), + ] + + +def test_type1_date_without_gps(): + # AZDome stores the date without GPS, and ExifTool extracts it + # when the accelerometer data is at the AZDome location + block = _type1_block( + lat=b" " * 9, lon=b" " * 10, tail=b"\x00" * (173 - 57) + b"+001-002+098" + ) + assert _extract([block]) is None + + +def test_invalid_date_skipped(): + points = _extract( + [ + _type15_block(hms=(13, 22, 30), ymd=(22, 2, 30)), + _type15_block(hms=(13, 22, 31), ymd=(22, 13, 14)), + _type15_block(hms=(13, 22, 32), ymd=(22, 12, 14)), + ] + ) + assert points is not None + assert [p.epoch_time for p in points] == [1671024152.0] + + +@pytest.mark.parametrize( + "blocks", + [ + [], + [NO_FIX_BLOCK], + # Unsupported layout (type 13) + [_type15_block(), _freegps_block({16: b"ANE\x00"})], + # Valid Nextbase records (type 20) + [ + _type15_block(), + _freegps_block({0x36: struct.pack(">HBBBBH", 2018, 10, 8, 6, 42, 465)}), + ], + # Mixed layouts + [_type15_block(), _type3_block()], + # Invalid numbers + [_type15_block(lat=(float("nan"), b"N"))], + ], +) +def test_no_points(blocks: list[bytes]): + assert _extract(blocks) is None + + +def test_invalid_index(): + block = _type15_block() + data = _build_video([block]) + # Point the index past the end of the file + data = data[:-8] + struct.pack(">II", len(data) + 100, len(block)) + assert novatek_parser.extract_points(io.BytesIO(data)) is None + assert novatek_parser.extract_points(io.BytesIO(b"")) is None + assert novatek_parser.extract_points(io.BytesIO(b"\x00\x00\x00\x09moov")) is None + + +def test_native_video_extractor(tmp_path: Path): + video_path = tmp_path / "novatek.mp4" + video_path.write_bytes(_build_video([_type15_block()])) + video_metadata = NativeVideoExtractor(video_path).extract() + assert video_metadata.filetype == types.FileType.VIDEO + assert video_metadata.make is None + assert video_metadata.model is None + assert video_metadata.points == [ + _point(0.0, 47.3558661666667, 8.5134765, 199.88, 1671024150.0) + ] + + # Novatek is not one of the native file types + with pytest.raises(exceptions.MapillaryVideoGPSNotFoundError): + NativeVideoExtractor(video_path, filetypes={types.FileType.CAMM}).extract() + + video_path.write_bytes(_build_video([NO_FIX_BLOCK])) + with pytest.raises(exceptions.MapillaryVideoGPSNotFoundError): + NativeVideoExtractor(video_path).extract() From e580d1545976db386ab645cf90257103db31f3eb Mon Sep 17 00:00:00 2001 From: Caglar Pir Date: Wed, 23 Sep 2026 14:47:53 +0200 Subject: [PATCH 2/2] Time Novatek points by their position in the video The first version rebased point times on the first fix, as the ExifTool fallback does. Distance sampling interpolates the track at frame times, so that shifts the track whenever the camera gets its first fix after it starts recording. In timelapse videos, where GPS time runs faster than the video, it also leaves most of the track past the end of the video. The camera writes one freeGPS block per second of video, right after the frames of that second, whether it has a fix or not. So the position of the block in the index is its time in the video. Use that as the point time, and keep the GPS time in epoch_time as before. Since the timing relies on one block per second, leave the video to ExifTool if the block count doesn't match the duration in mvhd. Positions and GPS times still match ExifTool's on all 27 Novatek videos it reads. Times change on the four of them that get their first fix 17 to 35 seconds in, or are timelapse. --- mapillary_tools/novatek_parser.py | 90 ++++++++++++++++++++--------- tests/unit/test_novatek_parser.py | 94 ++++++++++++++++++++++++++----- 2 files changed, 145 insertions(+), 39 deletions(-) diff --git a/mapillary_tools/novatek_parser.py b/mapillary_tools/novatek_parser.py index 47f3c777..a41e0d55 100644 --- a/mapillary_tools/novatek_parser.py +++ b/mapillary_tools/novatek_parser.py @@ -15,9 +15,15 @@ - Type 3: float32 values (Viofo A129, Viofo A139, Anker Roav C1 Pro) - Type 15: float64 values (Vantrue N4, Rove R2-4K Pro) -The decoded points match what the ExifTool fallback extracts from the same video. -If a video contains any block that can't be decoded that way, extract_points returns -None and the video is left to ExifTool. +The decoded positions and GPS times match what the ExifTool fallback extracts from +the same video. If a video contains any block that can't be decoded that way, +extract_points returns None and the video is left to ExifTool. + +Point times differ from ExifTool's on purpose. The camera writes one block per second +of video, right after the frames of that second, whether it has a fix or not (and in +timelapse too), so the position of the block in the index is its time in the video. +Counting from the first fix instead, as ExifTool's times do, would shift the track +whenever the camera gets its first fix after it starts recording. """ from __future__ import annotations @@ -38,6 +44,9 @@ # Enough to probe and decode every layout _MAX_BLOCK_READ_SIZE = 0x1000 +# The last, partial second of the video may have no block +_MAX_BLOCK_COUNT_DIFF = 1.5 + _FREEGPS_SIGNATURE = b"freeGPS " _TYPE1_KEY = b"\xaa\xaa\xf2\xe1\xf0\xee\x54\x54" @@ -80,22 +89,31 @@ class _UnsupportedBlockError(Exception): def extract_points(fp: T.BinaryIO) -> list[telemetry.GPSPoint] | None: """ Return the GPS points sorted by time, or None if the video has no "gps " box, - no points, or any block this parser does not support. - Like other parsers, point time is relative to the first point, - and epoch_time is the Unix time. + no points, or anything this parser does not support. + Point time is the time in the video, and epoch_time is the Unix time of the fix. """ try: + fp.seek(0) gps_box = sparser.parse_mp4_data_first(fp, [b"moov", b"gps "]) + fp.seek(0) + mvhd = sparser.parse_mp4_data_first(fp, [b"moov", b"mvhd"]) except sparser.ParsingError: return None - if gps_box is None: + if gps_box is None or mvhd is None: + return None + + index = _parse_gps_box(gps_box) + + # Block times assume one block per second of video + duration = _parse_duration(mvhd) + if duration is None or _MAX_BLOCK_COUNT_DIFF < abs(len(index) - duration): return None gps_types: set[int] = set() points: list[telemetry.GPSPoint] = [] - for block in _read_blocks(fp, _parse_gps_box(gps_box)): + for position, block in _read_blocks(fp, index): gps_type = _probe_block(block) gps_types.add(gps_type) @@ -118,6 +136,9 @@ def extract_points(fp: T.BinaryIO) -> list[telemetry.GPSPoint] | None: if point is None: continue + point.time = float(position) + + # Without a new fix, the camera may repeat the last one if not points or _point_key(points[-1]) != _point_key(point): points.append(point) @@ -129,18 +150,7 @@ def extract_points(fp: T.BinaryIO) -> list[telemetry.GPSPoint] | None: if not points: return None - points.sort(key=lambda p: p.time) - - deduplicated: list[telemetry.GPSPoint] = [] - for point in points: - if not deduplicated or _point_key(deduplicated[-1]) != _point_key(point): - deduplicated.append(point) - - first_point_time = deduplicated[0].time - for point in deduplicated: - point.time = point.time - first_point_time - - return deduplicated + return points def _parse_gps_box(data: bytes) -> list[tuple[int, int]]: @@ -163,10 +173,36 @@ def _parse_gps_box(data: bytes) -> list[tuple[int, int]]: return [struct.unpack_from(">II", data, 8 + i * 8) for i in range(count)] +def _parse_duration(mvhd: bytes) -> float | None: + """ + Return the duration in seconds from the "mvhd" box + + >>> _parse_duration(bytes.fromhex("00000000 00000000 00000000 000003e8 0000ea60")) + 60.0 + >>> _parse_duration(bytes.fromhex("01000000" + "00" * 16 + "000003e8 000000000000ea60")) + 60.0 + >>> _parse_duration(bytes.fromhex("00000000 00000000 00000000 00000000 0000ea60")) is None + True + """ + try: + if mvhd[0] == 1: + timescale, duration = struct.unpack_from(">IQ", mvhd, 20) + else: + timescale, duration = struct.unpack_from(">II", mvhd, 12) + except (IndexError, struct.error): + return None + if not timescale: + return None + return duration / timescale + + def _read_blocks( fp: T.BinaryIO, index: list[tuple[int, int]] -) -> T.Generator[bytes, None, None]: - for offset, size in index: +) -> T.Generator[tuple[int, bytes], None, None]: + """ + Yield the valid blocks with their positions in the index + """ + for position, (offset, size) in enumerate(index): fp.seek(offset, 0) # The layouts can be told apart by the first bytes and a few size thresholds, # all smaller than _MAX_BLOCK_READ_SIZE, so reading more doesn't change the result @@ -175,7 +211,7 @@ def _read_blocks( continue if len(block) < _MIN_BLOCK_SIZE: continue - yield block + yield position, block def _probe_block(block: bytes) -> int: @@ -355,7 +391,8 @@ def _build_point( angle = _to_exiftool_number(angle) return telemetry.GPSPoint( - time=epoch_time, + # Set by extract_points from the position of the block + time=0.0, lat=lat, lon=lon, alt=None, @@ -415,5 +452,6 @@ def _to_exiftool_number(value: float) -> float: def _point_key(point: telemetry.GPSPoint) -> tuple: - # Same as how the ExifTool extractor tells duplicate points - return (point.time, point.lon, point.lat, point.epoch_time, point.angle) + # The fix, regardless of the block it is in. The ExifTool extractor tells + # duplicate points the same way, as its point time is the fix time. + return (point.lon, point.lat, point.epoch_time, point.angle) diff --git a/tests/unit/test_novatek_parser.py b/tests/unit/test_novatek_parser.py index fbab848d..2792f7dc 100644 --- a/tests/unit/test_novatek_parser.py +++ b/tests/unit/test_novatek_parser.py @@ -12,6 +12,7 @@ import pytest from mapillary_tools import exceptions, novatek_parser, telemetry, types +from mapillary_tools.geotag.video_extractors.gpx import GPXVideoExtractor from mapillary_tools.geotag.video_extractors.native import NativeVideoExtractor @@ -19,9 +20,10 @@ def _box(box_type: bytes, data: bytes) -> bytes: return struct.pack(">I", 8 + len(data)) + box_type + data -def _build_video(blocks: list[bytes]) -> bytes: +def _build_video(blocks: list[bytes], duration: float | None = None) -> bytes: """ - Build a video with the blocks in mdat, indexed by moov/"gps " as Novatek does + Build a video with the blocks in mdat, indexed by moov/"gps " as Novatek does. + By default the video is as long as the blocks, one per second. """ ftyp = _box(b"ftyp", b"isom\x00\x00\x02\x00isomiso2mp41") index = b"" @@ -29,8 +31,12 @@ def _build_video(blocks: list[bytes]) -> bytes: for block in blocks: index += struct.pack(">II", offset, len(block)) offset += len(block) + if duration is None: + duration = len(blocks) + mvhd = struct.pack(">4xIIII", 0, 0, 1000, int(duration * 1000)).ljust(100, b"\x00") gps = struct.pack(">II", 0x101, len(blocks)) + index - return ftyp + _box(b"mdat", b"".join(blocks)) + _box(b"moov", _box(b"gps ", gps)) + moov = _box(b"moov", _box(b"mvhd", mvhd) + _box(b"gps ", gps)) + return ftyp + _box(b"mdat", b"".join(blocks)) + moov def _freegps_block(fields: dict[int, bytes], size: int = 0x100) -> bytes: @@ -104,7 +110,8 @@ def _extract(blocks: list[bytes]) -> list[telemetry.GPSPoint] | None: return novatek_parser.extract_points(io.BytesIO(_build_video(blocks))) -# The expected points below are the same as extracted by ExifTool 13.40 from the same videos +# The expected positions and epoch times below are the same as extracted by +# ExifTool 13.40 from the same videos. The times are the positions of the blocks. def _point(time, lat, lon, angle, epoch_time) -> telemetry.GPSPoint: return telemetry.GPSPoint( time=time, @@ -120,25 +127,59 @@ def _point(time, lat, lon, angle, epoch_time) -> telemetry.GPSPoint: def test_type15(): + # In the southern and western hemispheres next_block = _type15_block( - hms=(13, 22, 31), lat=(4721.3524, b"N"), lon=(830.8082, b"E"), track=198.5 + hms=(13, 22, 31), lat=(4721.3524, b"S"), lon=(830.8082, b"W"), track=198.5 ) points = _extract( [ + # The camera has no fix yet when it starts recording NO_FIX_BLOCK, + _type15_block(), next_block, - # Out of order, in the southern and western hemispheres - _type15_block(lat=(4721.35197, b"S"), lon=(830.80859, b"W")), - # Duplicate + # The camera repeats the last fix next_block, + NO_FIX_BLOCK, ] ) assert points == [ - _point(0.0, -47.3558661666667, -8.5134765, 199.88, 1671024150.0), - _point(1.0, 47.3558733333333, 8.51347, 198.5, 1671024151.0), + _point(1.0, 47.3558661666667, 8.5134765, 199.88, 1671024150.0), + _point(2.0, -47.3558733333333, -8.51347, 198.5, 1671024151.0), ] +def test_timelapse(): + # A block per second of video, 15 seconds apart in GPS time + points = _extract( + [ + _type15_block(hms=(13, 22, 30)), + _type15_block(hms=(13, 22, 45)), + _type15_block(hms=(13, 23, 0)), + ] + ) + assert points is not None + assert [(p.time, p.epoch_time) for p in points] == [ + (0.0, 1671024150.0), + (1.0, 1671024165.0), + (2.0, 1671024180.0), + ] + + +def test_skipped_blocks(): + # Skipped blocks still take their second of the video + points = _extract( + [ + # Too small for ExifTool + _type15_block()[:81], + # Not a freeGPS block + b"\x00\x00\x01\x00free " + _type15_block()[12:], + _type15_block(), + ] + ) + assert points is not None + assert [p.time for p in points] == [2.0] + + def test_type3(): points = _extract( [ @@ -154,9 +195,9 @@ def test_type3(): ) assert points == [ _point( - 0.0, 49.3690511067708, -123.099755859375, 16.4899997711182, 1569563223.0 + 1.0, 49.3690511067708, -123.099755859375, 16.4899997711182, 1569563223.0 ), - _point(1.0, 49.3692464192708, -123.099918619792, 16.5, 1569563224.0), + _point(2.0, 49.3692464192708, -123.099918619792, 16.5, 1569563224.0), ] @@ -198,7 +239,7 @@ def test_invalid_date_skipped(): ] ) assert points is not None - assert [p.epoch_time for p in points] == [1671024152.0] + assert [(p.time, p.epoch_time) for p in points] == [(2.0, 1671024152.0)] @pytest.mark.parametrize( @@ -217,12 +258,21 @@ def test_invalid_date_skipped(): [_type15_block(), _type3_block()], # Invalid numbers [_type15_block(lat=(float("nan"), b"N"))], + # Truncated + [_type15_block()[:84]], ], ) def test_no_points(blocks: list[bytes]): assert _extract(blocks) is None +@pytest.mark.parametrize("duration", [1.0, 5.0, 0.0]) +def test_unexpected_block_count(duration: float): + # Not a block per second of video, so the block times are unknown + video = _build_video([_type15_block()] * 3, duration=duration) + assert novatek_parser.extract_points(io.BytesIO(video)) is None + + def test_invalid_index(): block = _type15_block() data = _build_video([block]) @@ -251,3 +301,21 @@ def test_native_video_extractor(tmp_path: Path): video_path.write_bytes(_build_video([NO_FIX_BLOCK])) with pytest.raises(exceptions.MapillaryVideoGPSNotFoundError): NativeVideoExtractor(video_path).extract() + + +def test_gpx_sync(tmp_path: Path): + video_path = tmp_path / "novatek.mp4" + video_path.write_bytes( + _build_video([_type15_block(), _type15_block(hms=(13, 22, 31))]) + ) + gpx_path = tmp_path / "track.gpx" + gpx_path.write_text( + '' + '' + '' + '' + "" + ) + # The GPX track is synced by the GPS time of the first fix + video_metadata = GPXVideoExtractor(video_path, gpx_path).extract() + assert [p.time for p in video_metadata.points] == [-10.0, 0.0, 10.0]