Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 2 additions & 2 deletions mapillary_tools/geotag/geotag_from_generic.py
Original file line number Diff line number Diff line change
Expand Up @@ -34,7 +34,7 @@ class GeotagImagesFromGeneric(abc.ABC, T.Generic[TImageExtractor]):
"""

def __init__(
self, image_paths: T.Sequence[Path], num_processes: int | None
self, image_paths: T.Sequence[Path], num_processes: int | None = None
) -> None:
self.image_paths = image_paths
self.num_processes = num_processes
Expand Down Expand Up @@ -109,7 +109,7 @@ class GeotagVideosFromGeneric(abc.ABC, T.Generic[TVideoExtractor]):
"""

def __init__(
self, video_paths: T.Sequence[Path], num_processes: int | None
self, video_paths: T.Sequence[Path], num_processes: int | None = None
) -> None:
self.video_paths = video_paths
self.num_processes = num_processes
Expand Down
24 changes: 20 additions & 4 deletions mapillary_tools/gpmf/gpmf_parser.py
Original file line number Diff line number Diff line change
Expand Up @@ -166,7 +166,7 @@ def extract_gopro_info(
gyros_by_dvid = None
magns_by_dvid = None

_load_telemetry_from_samples(
device_found = _load_telemetry_from_samples(
fp,
gpmd_samples,
points_by_dvid=points_by_dvid,
Expand All @@ -175,6 +175,10 @@ def extract_gopro_info(
magns_by_dvid=magns_by_dvid,
dvnm_by_dvid=dvnm_by_dvid,
)
# If no device found, it's likely some other cameras using
# the "gpmd" container format, e.g. VANTRUE N2S 4K Dashcam
if not device_found:
return None

gopro_info = GoProInfo()

Expand Down Expand Up @@ -215,7 +219,11 @@ def extract_camera_model(fp: T.BinaryIO) -> str:
if _contains_gpmd_description(track):
gpmd_samples = _filter_gpmd_samples(track)
dvnm_by_dvid: dict[int, bytes] = {}
_load_telemetry_from_samples(fp, gpmd_samples, dvnm_by_dvid=dvnm_by_dvid)
device_found = _load_telemetry_from_samples(
fp, gpmd_samples, dvnm_by_dvid=dvnm_by_dvid
)
if not device_found:
return ""
return _extract_camera_model_from_devices(dvnm_by_dvid)

return ""
Expand Down Expand Up @@ -595,13 +603,19 @@ def _load_telemetry_from_samples(
gyros_by_dvid: dict[int, list[telemetry.GyroscopeData]] | None = None,
magns_by_dvid: dict[int, list[telemetry.MagnetometerData]] | None = None,
dvnm_by_dvid: dict[int, bytes] | None = None,
) -> None:
) -> bool:
device_found: bool = False

for sample, sample_data in _iterate_read_sample_data(fp, samples):
gpmf_sample_data = T.cast(T.Dict, GPMFSampleData.parse(sample_data))
try:
gpmf_sample_data = T.cast(T.Dict, GPMFSampleData.parse(sample_data))
except C.ConstructError:
continue

# iterate devices
devices = (klv for klv in gpmf_sample_data if klv["key"] == b"DEVC")
for device in devices:
device_found = True
device_id = _find_first_device_id(device["data"])

if dvnm_by_dvid is not None:
Expand Down Expand Up @@ -667,6 +681,8 @@ def _load_telemetry_from_samples(
for idx, (z, x, y, *_) in enumerate(sample_magns)
)

return device_found


def _is_gpmd_description(description: T.Dict) -> bool:
return description["format"] == b"gpmd"
Expand Down
Loading