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..a41e0d55 --- /dev/null +++ b/mapillary_tools/novatek_parser.py @@ -0,0 +1,457 @@ +# 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 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 + +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 + +# 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" +_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 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 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 position, block in _read_blocks(fp, index): + 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 + + 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) + + # 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 + + return points + + +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 _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[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 + block = fp.read(min(size, _MAX_BLOCK_READ_SIZE)) + if block[4:12] != _FREEGPS_SIGNATURE: + continue + if len(block) < _MIN_BLOCK_SIZE: + continue + yield position, 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( + # Set by extract_points from the position of the block + time=0.0, + 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: + # 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 new file mode 100644 index 00000000..2792f7dc --- /dev/null +++ b/tests/unit/test_novatek_parser.py @@ -0,0 +1,321 @@ +# 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.gpx import GPXVideoExtractor +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], duration: float | None = None) -> bytes: + """ + 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"" + offset = len(ftyp) + 8 + 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 + 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: + 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 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, + lat=lat, + lon=lon, + alt=None, + angle=angle, + epoch_time=epoch_time, + fix=None, + precision=None, + ground_speed=None, + ) + + +def test_type15(): + # In the southern and western hemispheres + next_block = _type15_block( + 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, + # The camera repeats the last fix + next_block, + NO_FIX_BLOCK, + ] + ) + assert points == [ + _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( + [ + 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( + 1.0, 49.3690511067708, -123.099755859375, 16.4899997711182, 1569563223.0 + ), + _point(2.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.time, p.epoch_time) for p in points] == [(2.0, 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"))], + # 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]) + # 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() + + +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]