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
25 changes: 23 additions & 2 deletions mapillary_tools/geotag/factory.py
Original file line number Diff line number Diff line change
Expand Up @@ -10,6 +10,8 @@
import typing as T
from pathlib import Path

import jsonschema

from .. import exceptions, types, utils
from . import (
base,
Expand Down Expand Up @@ -50,6 +52,9 @@ def parse_source_option(source: str) -> list[SourceOption]:
"gpx,exif" -> [SourceOption(SourceType.GPX), SourceOption(SourceType.EXIF)]
"exif,gpx" -> [SourceOption(SourceType.EXIF), SourceOption(SourceType.GPX)]
'{"source": "gpx"}' -> [SourceOption(SourceType.GPX)]

Raise MapillaryBadParameterError for an unknown source type, and for a
JSON object that does not match SourceOptionSchema.
"""

try:
Expand All @@ -64,11 +69,27 @@ def parse_source_option(source: str) -> list[SourceOption]:
except json.JSONDecodeError:
pass
else:
return [SourceOption.from_dict(payload)]
try:
return [SourceOption.from_dict(payload)]
except jsonschema.ValidationError as ex:
raise exceptions.MapillaryBadParameterError(
f"Invalid geotag source {source}: {ex.message}"
) from ex

sources = source.split(",")

return [SourceOption(SourceType(SOURCE_TYPE_ALIAS.get(s, s))) for s in sources]
return [SourceOption(_parse_source_type(s)) for s in sources]


def _parse_source_type(source: str) -> SourceType:
try:
return SourceType(SOURCE_TYPE_ALIAS.get(source, source))
except ValueError:
raise exceptions.MapillaryBadParameterError(
f"Invalid geotag source {source!r}. Supported source types: "
f"{', '.join(s.value for s in SourceType)} "
f"(aliases: {', '.join(SOURCE_TYPE_ALIAS)})"
) from None


def parse_source_options(
Expand Down
68 changes: 68 additions & 0 deletions tests/unit/test_parse_source_option.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,68 @@
# 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

from pathlib import Path

import pytest
from mapillary_tools import exceptions, sample_video
from mapillary_tools.geotag.factory import parse_source_option, parse_source_options
from mapillary_tools.geotag.options import SourceType


class TestParseSourceOption:
def test_source_types(self):
assert [o.source for o in parse_source_option("gpx")] == [SourceType.GPX]
assert [o.source for o in parse_source_option("gopro_videos")] == [
SourceType.GOPRO
]
assert [o.source for o in parse_source_option("exif,native")] == [
SourceType.EXIF,
SourceType.NATIVE,
]
assert [o.source for o in parse_source_option('{"source": "nmea"}')] == [
SourceType.NMEA
]

@pytest.mark.parametrize("source", ["foo", "gpx,foo", "foo,gpx"])
def test_unknown_source_type_names_the_supported_ones(self, source):
with pytest.raises(exceptions.MapillaryBadParameterError) as excinfo:
parse_source_option(source)

message = str(excinfo.value)
assert "'foo'" in message
assert ", ".join(s.value for s in SourceType) in message
assert "(aliases: blackvue_videos, gopro_videos, exiftool)" in message

def test_unknown_key_is_named(self):
with pytest.raises(exceptions.MapillaryBadParameterError) as excinfo:
parse_source_option('{"source": "gpx", "path": "track.gpx"}')

assert "'path' was unexpected" in str(excinfo.value)

@pytest.mark.parametrize(
"source",
[
'{"source": "foo"}',
'{"path": "track.gpx"}',
'{"source": "gpx", "filetypes": ["mp4"]}',
'{"source": "gpx", "interpolation_offset_time": "1s"}',
"123",
],
)
def test_invalid_json_source(self, source):
with pytest.raises(exceptions.MapillaryBadParameterError):
parse_source_option(source)

def test_video_geotag_source(self):
with pytest.raises(exceptions.MapillaryBadParameterError):
parse_source_options(
geotag_source=[], video_geotag_source=["foo"], geotag_source_path=None
)

def test_video_process(self, tmp_path: Path):
with pytest.raises(exceptions.MapillaryBadParameterError):
sample_video._parse_video_sources(["foo"], None, None, tmp_path)
Loading