diff --git a/mapillary_tools/authenticate.py b/mapillary_tools/authenticate.py index f58a1f220..8fb22ff87 100644 --- a/mapillary_tools/authenticate.py +++ b/mapillary_tools/authenticate.py @@ -11,7 +11,7 @@ import requests -from . import api_v4, config, constants, exceptions, types +from . import api_v4, config, constants, exceptions LOG = logging.getLogger(__name__) @@ -64,7 +64,7 @@ def authenticate( LOG.info('Creating new profile: "%s"', profile_name) if jwt: - user_items: types.UserItem = {"user_upload_token": jwt} + user_items: config.UserItem = {"user_upload_token": jwt} user_items = _verify_user_auth(_validate_profile(user_items)) else: user_items = _prompt_login( @@ -89,7 +89,7 @@ def authenticate( def fetch_user_items( user_name: str | None = None, organization_key: str | None = None, -) -> types.UserItem: +) -> config.UserItem: """ Read user information from the config file, or prompt the user to authenticate if the specified profile does not exist @@ -155,9 +155,9 @@ def _prompt(message: str) -> str: return input() -def _validate_profile(user_items: types.UserItem) -> types.UserItem: +def _validate_profile(user_items: config.UserItem) -> config.UserItem: try: - jsonschema.validate(user_items, types.UserItemSchema) + jsonschema.validate(user_items, config.UserItemSchema) except jsonschema.ValidationError as ex: raise exceptions.MapillaryBadParameterError( f"Invalid profile format: {ex.message}" @@ -165,7 +165,7 @@ def _validate_profile(user_items: types.UserItem) -> types.UserItem: return user_items -def _verify_user_auth(user_items: types.UserItem) -> types.UserItem: +def _verify_user_auth(user_items: config.UserItem) -> config.UserItem: """ Verify that the user access token is valid """ @@ -205,7 +205,7 @@ def _validate_profile_name(profile_name: str): ) -def _list_all_profiles(profiles: dict[str, types.UserItem]) -> None: +def _list_all_profiles(profiles: dict[str, config.UserItem]) -> None: _echo("Existing Mapillary profiles:") # Header @@ -256,7 +256,7 @@ def _is_login_retryable(ex: requests.HTTPError) -> bool: def _prompt_login( user_email: str | None = None, user_password: str | None = None, -) -> types.UserItem: +) -> config.UserItem: _enabled = _prompt_enabled() if user_email is None: @@ -288,7 +288,7 @@ def _prompt_login( data = resp.json() - user_items: types.UserItem = { + user_items: config.UserItem = { "user_upload_token": str(data["access_token"]), "MAPSettingsUserKey": str(data["user_id"]), } diff --git a/mapillary_tools/config.py b/mapillary_tools/config.py index 20d477d64..206e03771 100644 --- a/mapillary_tools/config.py +++ b/mapillary_tools/config.py @@ -2,31 +2,54 @@ import configparser import os +import sys import typing as T +from typing import TypedDict -from . import api_v4, types +if sys.version_info >= (3, 11): + from typing import Required +else: + from typing_extensions import Required +from . import api_v4 -_CLIENT_ID = api_v4.MAPILLARY_CLIENT_TOKEN -# Windows is not happy with | so we convert MLY|ID|TOKEN to MLY_ID_TOKEN -_CLIENT_ID = _CLIENT_ID.replace("|", "_", 2) - -DEFAULT_MAPILLARY_FOLDER = os.path.join( - os.path.expanduser("~"), - ".config", - "mapillary", -) +DEFAULT_MAPILLARY_FOLDER = os.path.join(os.path.expanduser("~"), ".config", "mapillary") MAPILLARY_CONFIG_PATH = os.getenv( "MAPILLARY_CONFIG_PATH", os.path.join( DEFAULT_MAPILLARY_FOLDER, "configs", - _CLIENT_ID, + # Windows is not happy with | so we convert MLY|ID|TOKEN to MLY_ID_TOKEN + api_v4.MAPILLARY_CLIENT_TOKEN.replace("|", "_"), ), ) +class UserItem(TypedDict, total=False): + MAPOrganizationKey: int | str + # Username + MAPSettingsUsername: str + # User ID + MAPSettingsUserKey: str + # User access token + user_upload_token: Required[str] + + +UserItemSchema = { + "type": "object", + "properties": { + "MAPOrganizationKey": {"type": ["integer", "string"]}, + # Not in use. Keep here for back-compatibility + "MAPSettingsUsername": {"type": "string"}, + "MAPSettingsUserKey": {"type": "string"}, + "user_upload_token": {"type": "string"}, + }, + "required": ["user_upload_token"], + "additionalProperties": True, +} + + def _load_config(config_path: str) -> configparser.ConfigParser: config = configparser.ConfigParser() # Override to not change option names (by default it will lower them) @@ -36,19 +59,17 @@ def _load_config(config_path: str) -> configparser.ConfigParser: return config -def load_user( - profile_name: str, config_path: str | None = None -) -> types.UserItem | None: +def load_user(profile_name: str, config_path: str | None = None) -> UserItem | None: if config_path is None: config_path = MAPILLARY_CONFIG_PATH config = _load_config(config_path) if not config.has_section(profile_name): return None user_items = dict(config.items(profile_name)) - return T.cast(types.UserItem, user_items) + return T.cast(UserItem, user_items) -def list_all_users(config_path: str | None = None) -> dict[str, types.UserItem]: +def list_all_users(config_path: str | None = None) -> dict[str, UserItem]: if config_path is None: config_path = MAPILLARY_CONFIG_PATH cp = _load_config(config_path) @@ -60,7 +81,7 @@ def list_all_users(config_path: str | None = None) -> dict[str, types.UserItem]: def update_config( - profile_name: str, user_items: types.UserItem, config_path: str | None = None + profile_name: str, user_items: UserItem, config_path: str | None = None ) -> None: if config_path is None: config_path = MAPILLARY_CONFIG_PATH diff --git a/mapillary_tools/geotag/geotag_images_from_gpx.py b/mapillary_tools/geotag/geotag_images_from_gpx.py index 56f4779ec..2ec0b9d2a 100644 --- a/mapillary_tools/geotag/geotag_images_from_gpx.py +++ b/mapillary_tools/geotag/geotag_images_from_gpx.py @@ -12,6 +12,7 @@ from typing_extensions import override from .. import exceptions, geo, types +from ..serializer.description import build_capture_time from .base import GeotagImagesFromGeneric from .geotag_images_from_exif import ImageEXIFExtractor @@ -43,26 +44,26 @@ def _interpolate_image_metadata_along( if image_metadata.time < sorted_points[0].time: delta = sorted_points[0].time - image_metadata.time - gpx_start_time = types.datetime_to_map_capture_time(sorted_points[0].time) - gpx_end_time = types.datetime_to_map_capture_time(sorted_points[-1].time) + gpx_start_time = build_capture_time(sorted_points[0].time) + gpx_end_time = build_capture_time(sorted_points[-1].time) # with the tolerance of 1ms if 0.001 < delta: raise exceptions.MapillaryOutsideGPXTrackError( f"The image date time is {round(delta, 3)} seconds behind the GPX start point", - image_time=types.datetime_to_map_capture_time(image_metadata.time), + image_time=build_capture_time(image_metadata.time), gpx_start_time=gpx_start_time, gpx_end_time=gpx_end_time, ) if sorted_points[-1].time < image_metadata.time: delta = image_metadata.time - sorted_points[-1].time - gpx_start_time = types.datetime_to_map_capture_time(sorted_points[0].time) - gpx_end_time = types.datetime_to_map_capture_time(sorted_points[-1].time) + gpx_start_time = build_capture_time(sorted_points[0].time) + gpx_end_time = build_capture_time(sorted_points[-1].time) # with the tolerance of 1ms if 0.001 < delta: raise exceptions.MapillaryOutsideGPXTrackError( f"The image time is {round(delta, 3)} seconds beyond the GPX end point", - image_time=types.datetime_to_map_capture_time(image_metadata.time), + image_time=build_capture_time(image_metadata.time), gpx_start_time=gpx_start_time, gpx_end_time=gpx_end_time, ) diff --git a/mapillary_tools/history.py b/mapillary_tools/history.py index 90d51b277..62b028ef0 100644 --- a/mapillary_tools/history.py +++ b/mapillary_tools/history.py @@ -7,6 +7,7 @@ from pathlib import Path from . import constants, types +from .serializer.description import DescriptionJSONSerializer JSONDict = T.Dict[str, T.Union[str, int, float, None]] @@ -57,6 +58,8 @@ def write_history( "summary": summary, } if metadatas is not None: - history["descs"] = [types.as_desc(metadata) for metadata in metadatas] + history["descs"] = [ + DescriptionJSONSerializer.as_desc(metadata) for metadata in metadatas + ] with open(path, "w") as fp: fp.write(json.dumps(history)) diff --git a/mapillary_tools/process_geotag_properties.py b/mapillary_tools/process_geotag_properties.py index 7ac842539..e7942078e 100644 --- a/mapillary_tools/process_geotag_properties.py +++ b/mapillary_tools/process_geotag_properties.py @@ -2,7 +2,6 @@ import collections import datetime -import json import logging import typing as T from pathlib import Path @@ -17,6 +16,11 @@ SourcePathOption, SourceType, ) +from .serializer.description import ( + DescriptionJSONSerializer, + validate_and_fail_metadata, +) +from .serializer.gpx import GPXSerializer LOG = logging.getLogger(__name__) DEFAULT_GEOTAG_SOURCE_OPTIONS = [ @@ -200,12 +204,16 @@ def _write_metadatas( desc_path: str, ) -> None: if desc_path == "-": - descs = [types.as_desc(metadata) for metadata in metadatas] - print(json.dumps(descs, indent=2)) + descs = DescriptionJSONSerializer.serialize(metadatas) + print(descs.decode("utf-8")) else: - descs = [types.as_desc(metadata) for metadata in metadatas] - with open(desc_path, "w") as fp: - json.dump(descs, fp) + normalized_suffix = Path(desc_path).suffix.strip().lower() + if normalized_suffix in [".gpx"]: + descs = GPXSerializer.serialize(metadatas) + else: + descs = DescriptionJSONSerializer.serialize(metadatas) + with open(desc_path, "wb") as fp: + fp.write(descs) LOG.info("Check the description file for details: %s", desc_path) @@ -293,7 +301,7 @@ def _validate_metadatas( # See https://stackoverflow.com/a/61432070 good_metadatas, error_metadatas = types.separate_errors(metadatas) map_results = utils.mp_map_maybe( - types.validate_and_fail_metadata, + validate_and_fail_metadata, T.cast(T.Iterable[types.Metadata], good_metadatas), num_processes=num_processes, ) diff --git a/mapillary_tools/process_sequence_properties.py b/mapillary_tools/process_sequence_properties.py index 9c325614f..c6ba7643b 100644 --- a/mapillary_tools/process_sequence_properties.py +++ b/mapillary_tools/process_sequence_properties.py @@ -7,6 +7,7 @@ import typing as T from . import constants, exceptions, geo, types, utils +from .serializer.description import DescriptionJSONSerializer LOG = logging.getLogger(__name__) @@ -106,7 +107,7 @@ def duplication_check( dup = types.describe_error_metadata( exceptions.MapillaryDuplicationError( msg, - types.as_desc(cur), + DescriptionJSONSerializer.as_desc(cur), distance=distance, angle_diff=angle_diff, ), diff --git a/mapillary_tools/sample_video.py b/mapillary_tools/sample_video.py index 527d1f581..414f60a69 100644 --- a/mapillary_tools/sample_video.py +++ b/mapillary_tools/sample_video.py @@ -13,6 +13,7 @@ from .exif_write import ExifEdit from .geotag import geotag_videos_from_video from .mp4 import mp4_sample_parser +from .serializer.description import parse_capture_time LOG = logging.getLogger(__name__) @@ -65,7 +66,7 @@ def sample_video( video_start_time_dt: datetime.datetime | None = None if video_start_time is not None: try: - video_start_time_dt = types.map_capture_time_to_datetime(video_start_time) + video_start_time_dt = parse_capture_time(video_start_time) except ValueError as ex: raise exceptions.MapillaryBadParameterError(str(ex)) diff --git a/mapillary_tools/serializer/description.py b/mapillary_tools/serializer/description.py new file mode 100644 index 000000000..198e7c108 --- /dev/null +++ b/mapillary_tools/serializer/description.py @@ -0,0 +1,587 @@ +from __future__ import annotations + +import dataclasses +import datetime +import json +import sys +import typing as T +from pathlib import Path +from typing import TypedDict + +if sys.version_info >= (3, 11): + from typing import Required +else: + from typing_extensions import Required + +if sys.version_info >= (3, 12): + from typing import override +else: + from typing_extensions import override + +import jsonschema + +from .. import exceptions, geo +from ..types import ( + BaseSerializer, + describe_error_metadata, + ErrorMetadata, + FileType, + ImageMetadata, + Metadata, + MetadataOrError, + VideoMetadata, +) + + +# http://wiki.gis.com/wiki/index.php/Decimal_degrees +# decimal places degrees distance +# 0 1.0 111 km +# 1 0.1 11.1 km +# 2 0.01 1.11 km +# 3 0.001 111 m +# 4 0.0001 11.1 m +# 5 0.00001 1.11 m +# 6 0.000001 0.111 m +# 7 0.0000001 1.11 cm +# 8 0.00000001 1.11 mm +_COORDINATES_PRECISION = 7 +_ALTITUDE_PRECISION = 3 +_ANGLE_PRECISION = 3 + + +class _CompassHeading(TypedDict, total=True): + TrueHeading: float + MagneticHeading: float + + +class _SharedDescription(TypedDict, total=False): + filename: Required[str] + filetype: Required[str] + + # if None or absent, it will be calculated + md5sum: str | None + filesize: int | None + + +class ImageDescription(_SharedDescription, total=False): + MAPLatitude: Required[float] + MAPLongitude: Required[float] + MAPAltitude: float + MAPCaptureTime: Required[str] + MAPCompassHeading: _CompassHeading + + MAPDeviceMake: str + MAPDeviceModel: str + MAPGPSAccuracyMeters: float + MAPCameraUUID: str + MAPOrientation: int + + # For grouping images in a sequence + MAPSequenceUUID: str + + +class VideoDescription(_SharedDescription, total=False): + MAPGPSTrack: Required[list[T.Sequence[float | int | None]]] + MAPDeviceMake: str + MAPDeviceModel: str + + +class _ErrorDescription(TypedDict, total=False): + # type and message are required + type: Required[str] + message: str + # vars is optional + vars: dict + + +class ImageDescriptionError(TypedDict, total=False): + filename: Required[str] + error: Required[_ErrorDescription] + filetype: str + + +Description = T.Union[ImageDescription, VideoDescription] +DescriptionOrError = T.Union[ImageDescription, VideoDescription, ImageDescriptionError] + + +ImageDescriptionEXIFSchema = { + "type": "object", + "properties": { + "MAPLatitude": { + "type": "number", + "description": "Latitude of the image", + "minimum": -90, + "maximum": 90, + }, + "MAPLongitude": { + "type": "number", + "description": "Longitude of the image", + "minimum": -180, + "maximum": 180, + }, + "MAPAltitude": { + "type": "number", + "description": "Altitude of the image, in meters", + }, + "MAPCaptureTime": { + "type": "string", + "description": "Capture time of the image", + "pattern": "[0-9]{4}_[0-9]{2}_[0-9]{2}_[0-9]{2}_[0-9]{2}_[0-9]{2}_[0-9]+", + }, + "MAPCompassHeading": { + "type": "object", + "properties": { + "TrueHeading": {"type": "number"}, + "MagneticHeading": {"type": "number"}, + }, + "required": ["TrueHeading", "MagneticHeading"], + "additionalProperties": False, + "description": "Camera angle of the image, in degrees. If null, the angle will be interpolated", + }, + "MAPSequenceUUID": { + "type": "string", + "description": "Arbitrary key for grouping images", + "pattern": "[a-zA-Z0-9_-]+", + }, + # deprecated since v0.10.0; keep here for compatibility + "MAPMetaTags": {"type": "object"}, + "MAPDeviceMake": {"type": "string"}, + "MAPDeviceModel": {"type": "string"}, + "MAPGPSAccuracyMeters": {"type": "number"}, + "MAPCameraUUID": {"type": "string"}, + "MAPFilename": { + "type": "string", + "description": "The base filename of the image", + }, + "MAPOrientation": {"type": "integer"}, + }, + "required": [ + "MAPLatitude", + "MAPLongitude", + "MAPCaptureTime", + ], + "additionalProperties": False, +} + +VideoDescriptionSchema = { + "type": "object", + "properties": { + "MAPGPSTrack": { + "type": "array", + "items": { + "type": "array", + "description": "track point", + "prefixItems": [ + { + "type": "number", + "description": "Time offset of the track point, in milliseconds, relative to the beginning of the video", + }, + { + "type": "number", + "description": "Longitude of the track point", + }, + { + "type": "number", + "description": "Latitude of the track point", + }, + { + "type": ["number", "null"], + "description": "Altitude of the track point in meters", + }, + { + "type": ["number", "null"], + "description": "Camera angle of the track point, in degrees. If null, the angle will be interpolated", + }, + ], + }, + }, + "MAPDeviceMake": { + "type": "string", + "description": "Device make, e.g. GoPro, BlackVue, Insta360", + }, + "MAPDeviceModel": { + "type": "string", + "description": "Device model, e.g. HERO10 Black, DR900S-1CH, Insta360 Titan", + }, + }, + "required": [ + "MAPGPSTrack", + ], + "additionalProperties": False, +} + + +def _merge_schema(*schemas: dict) -> dict: + for s in schemas: + assert s.get("type") == "object", "must be all object schemas" + properties = {} + all_required = [] + additional_properties = True + for s in schemas: + properties.update(s.get("properties", {})) + all_required += s.get("required", []) + if "additionalProperties" in s: + additional_properties = s["additionalProperties"] + return { + "type": "object", + "properties": properties, + "required": sorted(set(all_required)), + "additionalProperties": additional_properties, + } + + +ImageDescriptionFileSchema = _merge_schema( + ImageDescriptionEXIFSchema, + { + "type": "object", + "properties": { + "filename": { + "type": "string", + "description": "Absolute path of the image", + }, + "md5sum": { + "type": ["string", "null"], + "description": "MD5 checksum of the image content. If not provided, the uploader will compute it", + }, + "filesize": { + "type": ["number", "null"], + "description": "File size", + }, + "filetype": { + "type": "string", + "enum": [FileType.IMAGE.value], + "description": "The image file type", + }, + }, + "required": [ + "filename", + "filetype", + ], + }, +) + + +VideoDescriptionFileSchema = _merge_schema( + VideoDescriptionSchema, + { + "type": "object", + "properties": { + "filename": { + "type": "string", + "description": "Absolute path of the video", + }, + "md5sum": { + "type": ["string", "null"], + "description": "MD5 checksum of the video content. If not provided, the uploader will compute it", + }, + "filesize": { + "type": ["number", "null"], + "description": "File size", + }, + "filetype": { + "type": "string", + "enum": [ + FileType.CAMM.value, + FileType.GOPRO.value, + FileType.BLACKVUE.value, + FileType.VIDEO.value, + ], + "description": "The video file type", + }, + }, + "required": [ + "filename", + "filetype", + ], + }, +) + + +ImageVideoDescriptionFileSchema = { + "oneOf": [VideoDescriptionFileSchema, ImageDescriptionFileSchema] +} + + +class DescriptionJSONSerializer(BaseSerializer): + @override + @classmethod + def serialize(cls, metadatas: T.Sequence[MetadataOrError]) -> bytes: + descs = [cls.as_desc(m) for m in metadatas] + return json.dumps(descs).encode("utf-8") + + @override + @classmethod + def deserialize(cls, data: bytes) -> list[Metadata]: + descs = json.loads(data) + return [cls.from_desc(desc) for desc in descs if "error" not in desc] + + @override + @classmethod + def deserialize_stream(cls, data: T.IO[bytes]) -> list[Metadata]: + descs = json.load(data) + return [cls.from_desc(desc) for desc in descs if "error" not in desc] + + @T.overload + @classmethod + def as_desc(cls, metadata: ImageMetadata) -> ImageDescription: ... + + @T.overload + @classmethod + def as_desc(cls, metadata: ErrorMetadata) -> ImageDescriptionError: ... + + @T.overload + @classmethod + def as_desc(cls, metadata: VideoMetadata) -> VideoDescription: ... + + @classmethod + def as_desc(cls, metadata): + if isinstance(metadata, ErrorMetadata): + return _describe_error_desc( + metadata.error, metadata.filename, metadata.filetype + ) + + elif isinstance(metadata, VideoMetadata): + return cls._as_video_desc(metadata) + + else: + assert isinstance(metadata, ImageMetadata) + return cls._as_image_desc(metadata) + + @classmethod + def _as_video_desc(cls, metadata: VideoMetadata) -> VideoDescription: + desc: VideoDescription = { + "filename": str(metadata.filename.resolve()), + "md5sum": metadata.md5sum, + "filetype": metadata.filetype.value, + "filesize": metadata.filesize, + "MAPGPSTrack": [cls._encode_point(p) for p in metadata.points], + } + if metadata.make: + desc["MAPDeviceMake"] = metadata.make + if metadata.model: + desc["MAPDeviceModel"] = metadata.model + return desc + + @classmethod + def _as_image_desc(cls, metadata: ImageMetadata) -> ImageDescription: + desc: ImageDescription = { + "filename": str(metadata.filename.resolve()), + "md5sum": metadata.md5sum, + "filesize": metadata.filesize, + "filetype": FileType.IMAGE.value, + "MAPLatitude": round(metadata.lat, _COORDINATES_PRECISION), + "MAPLongitude": round(metadata.lon, _COORDINATES_PRECISION), + "MAPCaptureTime": build_capture_time(metadata.time), + } + if metadata.alt is not None: + desc["MAPAltitude"] = round(metadata.alt, _ALTITUDE_PRECISION) + if metadata.angle is not None: + desc["MAPCompassHeading"] = { + "TrueHeading": round(metadata.angle, _ANGLE_PRECISION), + "MagneticHeading": round(metadata.angle, _ANGLE_PRECISION), + } + fields = dataclasses.fields(metadata) + for field in fields: + if field.name.startswith("MAP"): + value = getattr(metadata, field.name) + if value is not None: + # ignore error: TypedDict key must be a string literal; + # expected one of ("MAPMetaTags", "MAPDeviceMake", "MAPDeviceModel", "MAPGPSAccuracyMeters", "MAPCameraUUID", ...) + desc[field.name] = value # type: ignore + return desc + + @T.overload + @classmethod + def from_desc(cls, desc: ImageDescription) -> ImageMetadata: ... + + @T.overload + @classmethod + def from_desc(cls, desc: VideoDescription) -> VideoMetadata: ... + + @classmethod + def from_desc(cls, desc): + if "error" in desc: + raise ValueError("Cannot deserialize error description") + + if desc["filetype"] == FileType.IMAGE.value: + return cls._from_image_desc(desc) + else: + return cls._from_video_desc(desc) + + @classmethod + def _from_image_desc(cls, desc) -> ImageMetadata: + validate_image_desc(desc) + + kwargs: dict = {} + for k, v in desc.items(): + if k not in [ + "filename", + "md5sum", + "filesize", + "filetype", + "MAPLatitude", + "MAPLongitude", + "MAPAltitude", + "MAPCaptureTime", + "MAPCompassHeading", + ]: + kwargs[k] = v + + return ImageMetadata( + filename=Path(desc["filename"]), + md5sum=desc.get("md5sum"), + filesize=desc.get("filesize"), + lat=desc["MAPLatitude"], + lon=desc["MAPLongitude"], + alt=desc.get("MAPAltitude"), + time=geo.as_unix_time(parse_capture_time(desc["MAPCaptureTime"])), + angle=desc.get("MAPCompassHeading", {}).get("TrueHeading"), + width=None, + height=None, + **kwargs, + ) + + @classmethod + def _encode_point(cls, p: geo.Point) -> T.Sequence[float | int | None]: + entry = [ + int(p.time * 1000), + round(p.lon, _COORDINATES_PRECISION), + round(p.lat, _COORDINATES_PRECISION), + round(p.alt, _ALTITUDE_PRECISION) if p.alt is not None else None, + round(p.angle, _ANGLE_PRECISION) if p.angle is not None else None, + ] + return entry + + @classmethod + def _decode_point(cls, entry: T.Sequence[T.Any]) -> geo.Point: + time_ms, lon, lat, alt, angle = entry + return geo.Point(time=time_ms / 1000, lon=lon, lat=lat, alt=alt, angle=angle) + + @classmethod + def _from_video_desc(cls, desc: VideoDescription) -> VideoMetadata: + validate_video_desc(desc) + + return VideoMetadata( + filename=Path(desc["filename"]), + md5sum=desc.get("md5sum"), + filesize=desc.get("filesize"), + filetype=FileType(desc["filetype"]), + points=[cls._decode_point(entry) for entry in desc["MAPGPSTrack"]], + make=desc.get("MAPDeviceMake"), + model=desc.get("MAPDeviceModel"), + ) + + +def build_capture_time(time: datetime.datetime | int | float) -> str: + if isinstance(time, (float, int)): + dt = datetime.datetime.fromtimestamp(time, datetime.timezone.utc) + # otherwise it will be assumed to be in local time + dt = dt.replace(tzinfo=datetime.timezone.utc) + else: + # otherwise it will be assumed to be in local time + dt = time.astimezone(datetime.timezone.utc) + return datetime.datetime.strftime(dt, "%Y_%m_%d_%H_%M_%S_%f")[:-3] + + +def parse_capture_time(time: str) -> datetime.datetime: + dt = datetime.datetime.strptime(time, "%Y_%m_%d_%H_%M_%S_%f") + dt = dt.replace(tzinfo=datetime.timezone.utc) + return dt + + +def validate_image_desc(desc: T.Any) -> None: + try: + jsonschema.validate(instance=desc, schema=ImageDescriptionFileSchema) + except jsonschema.ValidationError as ex: + # do not use str(ex) which is more verbose + raise exceptions.MapillaryMetadataValidationError(ex.message) from ex + + try: + parse_capture_time(desc["MAPCaptureTime"]) + except ValueError as ex: + raise exceptions.MapillaryMetadataValidationError(str(ex)) from ex + + +def validate_video_desc(desc: T.Any) -> None: + try: + jsonschema.validate(instance=desc, schema=VideoDescriptionFileSchema) + except jsonschema.ValidationError as ex: + # do not use str(ex) which is more verbose + raise exceptions.MapillaryMetadataValidationError(ex.message) from ex + + +def validate_and_fail_metadata(metadata: MetadataOrError) -> MetadataOrError: + if isinstance(metadata, ErrorMetadata): + return metadata + + if isinstance(metadata, ImageMetadata): + filetype = FileType.IMAGE + validate = validate_image_desc + else: + assert isinstance(metadata, VideoMetadata) + filetype = metadata.filetype + validate = validate_video_desc + + try: + validate(DescriptionJSONSerializer.as_desc(metadata)) + except exceptions.MapillaryMetadataValidationError as ex: + # rethrow because the original error is too verbose + return describe_error_metadata( + ex, + metadata.filename, + filetype=filetype, + ) + + if not metadata.filename.is_file(): + return describe_error_metadata( + exceptions.MapillaryMetadataValidationError( + f"No such file {metadata.filename}" + ), + metadata.filename, + filetype=filetype, + ) + + return metadata + + +def desc_file_to_exif(desc: ImageDescription) -> ImageDescription: + not_needed = ["MAPSequenceUUID"] + removed = { + key: value + for key, value in desc.items() + if key.startswith("MAP") and key not in not_needed + } + return T.cast(ImageDescription, removed) + + +def _describe_error_desc( + exc: Exception, filename: Path, filetype: FileType | None +) -> ImageDescriptionError: + err: _ErrorDescription = { + "type": exc.__class__.__name__, + "message": str(exc), + } + + exc_vars = vars(exc) + + if exc_vars: + # handle unserializable exceptions + try: + vars_json = json.dumps(exc_vars) + except Exception: + vars_json = "" + if vars_json: + err["vars"] = json.loads(vars_json) + + desc: ImageDescriptionError = { + "error": err, + "filename": str(filename.resolve()), + } + if filetype is not None: + desc["filetype"] = filetype.value + + return desc + + +if __name__ == "__main__": + print(json.dumps(ImageVideoDescriptionFileSchema, indent=4)) diff --git a/mapillary_tools/serializer/gpx.py b/mapillary_tools/serializer/gpx.py new file mode 100644 index 000000000..d44e69235 --- /dev/null +++ b/mapillary_tools/serializer/gpx.py @@ -0,0 +1,132 @@ +import datetime +import json +import sys +import typing as T + +if sys.version_info >= (3, 12): + from typing import override +else: + from typing_extensions import override + +import gpxpy +import gpxpy.gpx + +from .. import geo, types + +from ..telemetry import CAMMGPSPoint, GPSPoint +from ..types import ( + BaseSerializer, + ErrorMetadata, + ImageMetadata, + MetadataOrError, + VideoMetadata, +) +from .description import DescriptionJSONSerializer + + +class GPXSerializer(BaseSerializer): + @override + @classmethod + def serialize(cls, metadatas: T.Sequence[MetadataOrError]) -> bytes: + gpx = cls.as_gpx(metadatas) + return gpx.to_xml().encode("utf-8") + + @classmethod + def as_gpx(cls, metadatas: T.Sequence[MetadataOrError]) -> gpxpy.gpx.GPX: + gpx = gpxpy.gpx.GPX() + + error_metadatas = [] + image_metadatas = [] + video_metadatas = [] + + for metadata in metadatas: + if isinstance(metadata, ErrorMetadata): + error_metadatas.append(metadata) + elif isinstance(metadata, ImageMetadata): + image_metadatas.append(metadata) + elif isinstance(metadata, VideoMetadata): + video_metadatas.append(metadata) + + for metadata in error_metadatas: + gpx_track = gpxpy.gpx.GPXTrack() + gpx_track.name = str(metadata.filename) + gpx_track.description = cls._build_gpx_description(metadata, ["filename"]) + gpx.tracks.append(gpx_track) + + sequences = types.group_and_sort_images(image_metadatas) + for sequence_uuid, sequence in sequences.items(): + gpx.tracks.append(cls.image_sequence_as_gpx_track(sequence_uuid, sequence)) + + for metadata in video_metadatas: + gpx.tracks.append(cls.as_gpx_track(metadata)) + + return gpx + + @classmethod + def as_gpx_point(cls, point: geo.Point) -> gpxpy.gpx.GPXTrackPoint: + gpx_point = gpxpy.gpx.GPXTrackPoint( + latitude=point.lat, + longitude=point.lon, + elevation=point.alt, + time=datetime.datetime.fromtimestamp(point.time, datetime.timezone.utc), + ) + + if isinstance(point, types.ImageMetadata): + gpx_point.name = point.filename.name + elif isinstance(point, CAMMGPSPoint): + gpx_point.time = datetime.datetime.fromtimestamp( + point.time_gps_epoch, datetime.timezone.utc + ) + elif isinstance(point, GPSPoint): + if point.epoch_time is not None: + gpx_point.time = datetime.datetime.fromtimestamp( + point.epoch_time, datetime.timezone.utc + ) + + return gpx_point + + @classmethod + def as_gpx_track(cls, metadata: VideoMetadata) -> gpxpy.gpx.GPXTrack: + gpx_segment = gpxpy.gpx.GPXTrackSegment() + for point in metadata.points: + gpx_point = cls.as_gpx_point(point) + gpx_segment.points.append(gpx_point) + gpx_track = gpxpy.gpx.GPXTrack() + gpx_track.name = str(metadata.filename) + gpx_track.description = cls._build_gpx_description( + metadata, ["filename", "MAPGPSTrack"] + ) + gpx_track.segments.append(gpx_segment) + return gpx_track + + @classmethod + def image_sequence_as_gpx_track( + cls, sequence_uuid: str, sequence: T.Sequence[ImageMetadata] + ) -> gpxpy.gpx.GPXTrack: + gpx_segment = gpxpy.gpx.GPXTrackSegment() + for metadata in sequence: + gpx_point = cls.as_gpx_point(metadata) + gpx_segment.points.append(gpx_point) + gpx_track = gpxpy.gpx.GPXTrack() + gpx_track.name = sequence_uuid + gpx_track.description = cls._build_gpx_description( + metadata, + [ + "filename", + "MAPLongitude", + "MAPLatitude", + "MAPCaptureTime", + "MAPAltitude", + ], + ) + gpx_track.segments.append(gpx_segment) + return gpx_track + + @classmethod + def _build_gpx_description( + cls, metadata: MetadataOrError, excluded_properties: T.Sequence[str] + ) -> str: + desc = T.cast(T.Dict, DescriptionJSONSerializer.as_desc(metadata)) + for prop in excluded_properties: + desc.pop(prop, None) + return json.dumps(desc) diff --git a/mapillary_tools/types.py b/mapillary_tools/types.py index b4fe31615..9e98f4150 100644 --- a/mapillary_tools/types.py +++ b/mapillary_tools/types.py @@ -1,41 +1,14 @@ from __future__ import annotations +import abc import dataclasses -import datetime import enum import hashlib -import json -import os -import sys import typing as T import uuid from pathlib import Path -from typing import TypedDict -if sys.version_info >= (3, 11): - from typing import Required -else: - from typing_extensions import Required - -import jsonschema - -from . import exceptions, geo, utils - - -# http://wiki.gis.com/wiki/index.php/Decimal_degrees -# decimal places degrees distance -# 0 1.0 111 km -# 1 0.1 11.1 km -# 2 0.01 1.11 km -# 3 0.001 111 m -# 4 0.0001 11.1 m -# 5 0.00001 1.11 m -# 6 0.000001 0.111 m -# 7 0.0000001 1.11 cm -# 8 0.00000001 1.11 mm -_COORDINATES_PRECISION = 7 -_ALTITUDE_PRECISION = 3 -_ANGLE_PRECISION = 3 +from . import geo, utils class FileType(enum.Enum): @@ -59,9 +32,13 @@ class FileType(enum.Enum): @dataclasses.dataclass class ImageMetadata(geo.Point): filename: Path + # filetype should be always FileType.IMAGE md5sum: str | None = None width: int | None = None height: int | None = None + filesize: int | None = None + + # Fields starting with MAP* will be written to the image EXIF MAPSequenceUUID: str | None = None MAPDeviceMake: str | None = None MAPDeviceModel: str | None = None @@ -70,7 +47,6 @@ class ImageMetadata(geo.Point): MAPOrientation: int | None = None MAPMetaTags: dict | None = None MAPFilename: str | None = None - filesize: int | None = None def update_md5sum(self, image_data: T.BinaryIO | None = None) -> None: if self.md5sum is None: @@ -116,17 +92,47 @@ class ErrorMetadata: MetadataOrError = T.Union[Metadata, ErrorMetadata] -# Assume {GOPRO, VIDEO} are the NATIVE_VIDEO_FILETYPES: -# a | b = result -# {CAMM} | {GOPRO} = {} -# {CAMM} | {GOPRO, VIDEO} = {CAMM} -# {GOPRO} | {GOPRO, VIDEO} = {GOPRO} -# {GOPRO} | {VIDEO} = {GOPRO} -# {CAMM, GOPRO} | {VIDEO} = {CAMM, GOPRO} -# {VIDEO} | {VIDEO} = {CAMM, GOPRO, VIDEO} +class BaseSerializer(abc.ABC): + @classmethod + @abc.abstractmethod + def serialize(cls, metadatas: T.Sequence[MetadataOrError]) -> bytes: + raise NotImplementedError() + + @classmethod + @abc.abstractmethod + def deserialize(cls, data: bytes) -> list[Metadata]: + raise NotImplementedError() + + @classmethod + def deserialize_stream(cls, data: T.IO[bytes]) -> list[Metadata]: + return cls.deserialize(data.read()) + + def combine_filetype_filters( a: set[FileType] | None, b: set[FileType] | None ) -> set[FileType] | None: + """ + >>> combine_filetype_filters({FileType.CAMM}, {FileType.GOPRO}) + set() + + >>> combine_filetype_filters({FileType.CAMM}, {FileType.GOPRO, FileType.VIDEO}) + {} + + >>> combine_filetype_filters({FileType.GOPRO}, {FileType.GOPRO, FileType.VIDEO}) + {} + + >>> combine_filetype_filters({FileType.GOPRO}, {FileType.VIDEO}) + {} + + >>> expected = {FileType.CAMM, FileType.GOPRO} + >>> combine_filetype_filters({FileType.CAMM, FileType.GOPRO}, {FileType.VIDEO}) == expected + True + + >>> expected = {FileType.CAMM, FileType.GOPRO, FileType.BLACKVUE, FileType.VIDEO} + >>> combine_filetype_filters({FileType.VIDEO}, {FileType.VIDEO}) == expected + True + """ + if a is None: return b @@ -145,65 +151,6 @@ def combine_filetype_filters( return a.intersection(b) -class UserItem(TypedDict, total=False): - MAPOrganizationKey: int | str - # Not in use. Keep here for back-compatibility - MAPSettingsUsername: str - MAPSettingsUserKey: str - user_upload_token: Required[str] - - -class _CompassHeading(TypedDict, total=True): - TrueHeading: float - MagneticHeading: float - - -class _SharedDescription(TypedDict, total=False): - filename: Required[str] - filetype: Required[str] - - # if None or absent, it will be calculated - md5sum: str | None - filesize: int | None - - -class ImageDescription(_SharedDescription, total=False): - MAPLatitude: Required[float] - MAPLongitude: Required[float] - MAPAltitude: float - MAPCaptureTime: Required[str] - MAPCompassHeading: _CompassHeading - - MAPDeviceMake: str - MAPDeviceModel: str - MAPGPSAccuracyMeters: float - MAPCameraUUID: str - MAPOrientation: int - - # For grouping images in a sequence - MAPSequenceUUID: str - - -class VideoDescription(_SharedDescription, total=False): - MAPGPSTrack: Required[list[T.Sequence[float | int | None]]] - MAPDeviceMake: str - MAPDeviceModel: str - - -class _ErrorDescription(TypedDict, total=False): - # type and message are required - type: Required[str] - message: str - # vars is optional - vars: dict - - -class ImageDescriptionError(TypedDict, total=False): - filename: Required[str] - error: Required[_ErrorDescription] - filetype: str - - M = T.TypeVar("M") @@ -222,511 +169,12 @@ def separate_errors( return good, bad -def _describe_error_desc( - exc: Exception, filename: Path, filetype: FileType | None -) -> ImageDescriptionError: - err: _ErrorDescription = { - "type": exc.__class__.__name__, - "message": str(exc), - } - - exc_vars = vars(exc) - - if exc_vars: - # handle unserializable exceptions - try: - vars_json = json.dumps(exc_vars) - except Exception: - vars_json = "" - if vars_json: - err["vars"] = json.loads(vars_json) - - desc: ImageDescriptionError = { - "error": err, - "filename": str(filename.resolve()), - } - if filetype is not None: - desc["filetype"] = filetype.value - - return desc - - def describe_error_metadata( exc: Exception, filename: Path, filetype: FileType ) -> ErrorMetadata: return ErrorMetadata(filename=filename, filetype=filetype, error=exc) -Description = T.Union[ImageDescription, VideoDescription] -DescriptionOrError = T.Union[ImageDescription, VideoDescription, ImageDescriptionError] - - -UserItemSchema = { - "type": "object", - "properties": { - "MAPOrganizationKey": {"type": ["integer", "string"]}, - # Not in use. Keep here for back-compatibility - "MAPSettingsUsername": {"type": "string"}, - "MAPSettingsUserKey": {"type": "string"}, - "user_upload_token": {"type": "string"}, - }, - "required": ["user_upload_token"], - "additionalProperties": True, -} - - -ImageDescriptionEXIFSchema = { - "type": "object", - "properties": { - "MAPLatitude": { - "type": "number", - "description": "Latitude of the image", - "minimum": -90, - "maximum": 90, - }, - "MAPLongitude": { - "type": "number", - "description": "Longitude of the image", - "minimum": -180, - "maximum": 180, - }, - "MAPAltitude": { - "type": "number", - "description": "Altitude of the image, in meters", - }, - "MAPCaptureTime": { - "type": "string", - "description": "Capture time of the image", - "pattern": "[0-9]{4}_[0-9]{2}_[0-9]{2}_[0-9]{2}_[0-9]{2}_[0-9]{2}_[0-9]+", - }, - "MAPCompassHeading": { - "type": "object", - "properties": { - "TrueHeading": {"type": "number"}, - "MagneticHeading": {"type": "number"}, - }, - "required": ["TrueHeading", "MagneticHeading"], - "additionalProperties": False, - "description": "Camera angle of the image, in degrees. If null, the angle will be interpolated", - }, - "MAPSequenceUUID": { - "type": "string", - "description": "Arbitrary key for grouping images", - "pattern": "[a-zA-Z0-9_-]+", - }, - # deprecated since v0.10.0; keep here for compatibility - "MAPMetaTags": {"type": "object"}, - "MAPDeviceMake": {"type": "string"}, - "MAPDeviceModel": {"type": "string"}, - "MAPGPSAccuracyMeters": {"type": "number"}, - "MAPCameraUUID": {"type": "string"}, - "MAPFilename": { - "type": "string", - "description": "The base filename of the image", - }, - "MAPOrientation": {"type": "integer"}, - }, - "required": [ - "MAPLatitude", - "MAPLongitude", - "MAPCaptureTime", - ], - "additionalProperties": False, -} - -VideoDescriptionSchema = { - "type": "object", - "properties": { - "MAPGPSTrack": { - "type": "array", - "items": { - "type": "array", - "description": "track point", - "prefixItems": [ - { - "type": "number", - "description": "Time offset of the track point, in milliseconds, relative to the beginning of the video", - }, - { - "type": "number", - "description": "Longitude of the track point", - }, - { - "type": "number", - "description": "Latitude of the track point", - }, - { - "type": ["number", "null"], - "description": "Altitude of the track point in meters", - }, - { - "type": ["number", "null"], - "description": "Camera angle of the track point, in degrees. If null, the angle will be interpolated", - }, - ], - }, - }, - "MAPDeviceMake": { - "type": "string", - "description": "Device make, e.g. GoPro, BlackVue, Insta360", - }, - "MAPDeviceModel": { - "type": "string", - "description": "Device model, e.g. HERO10 Black, DR900S-1CH, Insta360 Titan", - }, - }, - "required": [ - "MAPGPSTrack", - ], - "additionalProperties": False, -} - - -def merge_schema(*schemas: dict) -> dict: - for s in schemas: - assert s.get("type") == "object", "must be all object schemas" - properties = {} - all_required = [] - additional_properties = True - for s in schemas: - properties.update(s.get("properties", {})) - all_required += s.get("required", []) - if "additionalProperties" in s: - additional_properties = s["additionalProperties"] - return { - "type": "object", - "properties": properties, - "required": sorted(set(all_required)), - "additionalProperties": additional_properties, - } - - -ImageDescriptionFileSchema = merge_schema( - ImageDescriptionEXIFSchema, - { - "type": "object", - "properties": { - "filename": { - "type": "string", - "description": "Absolute path of the image", - }, - "md5sum": { - "type": ["string", "null"], - "description": "MD5 checksum of the image content. If not provided, the uploader will compute it", - }, - "filesize": { - "type": ["number", "null"], - "description": "File size", - }, - "filetype": { - "type": "string", - "enum": [FileType.IMAGE.value], - "description": "The image file type", - }, - }, - "required": [ - "filename", - "filetype", - ], - }, -) - -VideoDescriptionFileSchema = merge_schema( - VideoDescriptionSchema, - { - "type": "object", - "properties": { - "filename": { - "type": "string", - "description": "Absolute path of the video", - }, - "md5sum": { - "type": ["string", "null"], - "description": "MD5 checksum of the video content. If not provided, the uploader will compute it", - }, - "filesize": { - "type": ["number", "null"], - "description": "File size", - }, - "filetype": { - "type": "string", - "enum": [ - FileType.CAMM.value, - FileType.GOPRO.value, - FileType.BLACKVUE.value, - FileType.VIDEO.value, - ], - "description": "The video file type", - }, - }, - "required": [ - "filename", - "filetype", - ], - }, -) - - -ImageVideoDescriptionFileSchema = { - "oneOf": [VideoDescriptionFileSchema, ImageDescriptionFileSchema] -} - - -def validate_image_desc(desc: T.Any) -> None: - try: - jsonschema.validate(instance=desc, schema=ImageDescriptionFileSchema) - except jsonschema.ValidationError as ex: - # do not use str(ex) which is more verbose - raise exceptions.MapillaryMetadataValidationError(ex.message) from ex - try: - map_capture_time_to_datetime(desc["MAPCaptureTime"]) - except ValueError as ex: - raise exceptions.MapillaryMetadataValidationError(str(ex)) from ex - - -def validate_video_desc(desc: T.Any) -> None: - try: - jsonschema.validate(instance=desc, schema=VideoDescriptionFileSchema) - except jsonschema.ValidationError as ex: - # do not use str(ex) which is more verbose - raise exceptions.MapillaryMetadataValidationError(ex.message) from ex - - -def datetime_to_map_capture_time(time: datetime.datetime | int | float) -> str: - if isinstance(time, (float, int)): - dt = datetime.datetime.fromtimestamp(time, datetime.timezone.utc) - # otherwise it will be assumed to be in local time - dt = dt.replace(tzinfo=datetime.timezone.utc) - else: - # otherwise it will be assumed to be in local time - dt = time.astimezone(datetime.timezone.utc) - return datetime.datetime.strftime(dt, "%Y_%m_%d_%H_%M_%S_%f")[:-3] - - -def map_capture_time_to_datetime(time: str) -> datetime.datetime: - dt = datetime.datetime.strptime(time, "%Y_%m_%d_%H_%M_%S_%f") - dt = dt.replace(tzinfo=datetime.timezone.utc) - return dt - - -@T.overload -def as_desc(metadata: ImageMetadata) -> ImageDescription: ... - - -@T.overload -def as_desc(metadata: ErrorMetadata) -> ImageDescriptionError: ... - - -@T.overload -def as_desc(metadata: VideoMetadata) -> VideoDescription: ... - - -def as_desc(metadata): - if isinstance(metadata, ErrorMetadata): - return _describe_error_desc( - metadata.error, metadata.filename, metadata.filetype - ) - elif isinstance(metadata, VideoMetadata): - return _as_video_desc(metadata) - else: - assert isinstance(metadata, ImageMetadata) - return _as_image_desc(metadata) - - -def _as_video_desc(metadata: VideoMetadata) -> VideoDescription: - desc: VideoDescription = { - "filename": str(metadata.filename.resolve()), - "md5sum": metadata.md5sum, - "filetype": metadata.filetype.value, - "filesize": metadata.filesize, - "MAPGPSTrack": [_encode_point(p) for p in metadata.points], - } - if metadata.make: - desc["MAPDeviceMake"] = metadata.make - if metadata.model: - desc["MAPDeviceModel"] = metadata.model - return desc - - -def _as_image_desc(metadata: ImageMetadata) -> ImageDescription: - desc: ImageDescription = { - "filename": str(metadata.filename.resolve()), - "md5sum": metadata.md5sum, - "filesize": metadata.filesize, - "filetype": FileType.IMAGE.value, - "MAPLatitude": round(metadata.lat, _COORDINATES_PRECISION), - "MAPLongitude": round(metadata.lon, _COORDINATES_PRECISION), - "MAPCaptureTime": datetime_to_map_capture_time(metadata.time), - } - if metadata.alt is not None: - desc["MAPAltitude"] = round(metadata.alt, _ALTITUDE_PRECISION) - if metadata.angle is not None: - desc["MAPCompassHeading"] = { - "TrueHeading": round(metadata.angle, _ANGLE_PRECISION), - "MagneticHeading": round(metadata.angle, _ANGLE_PRECISION), - } - fields = dataclasses.fields(metadata) - for field in fields: - if field.name.startswith("MAP"): - value = getattr(metadata, field.name) - if value is not None: - # ignore error: TypedDict key must be a string literal; - # expected one of ("MAPMetaTags", "MAPDeviceMake", "MAPDeviceModel", "MAPGPSAccuracyMeters", "MAPCameraUUID", ...) - desc[field.name] = value # type: ignore - return desc - - -@T.overload -def from_desc(metadata: ImageDescription) -> ImageMetadata: ... - - -@T.overload -def from_desc(metadata: VideoDescription) -> VideoMetadata: ... - - -def from_desc(desc): - assert "error" not in desc - if desc["filetype"] == FileType.IMAGE.value: - return _from_image_desc(desc) - else: - return _from_video_desc(desc) - - -def _from_image_desc(desc) -> ImageMetadata: - kwargs: dict = {} - for k, v in desc.items(): - if k not in [ - "filename", - "md5sum", - "filesize", - "filetype", - "MAPLatitude", - "MAPLongitude", - "MAPAltitude", - "MAPCaptureTime", - "MAPCompassHeading", - ]: - kwargs[k] = v - - return ImageMetadata( - filename=Path(desc["filename"]), - md5sum=desc.get("md5sum"), - filesize=desc.get("filesize"), - lat=desc["MAPLatitude"], - lon=desc["MAPLongitude"], - alt=desc.get("MAPAltitude"), - time=geo.as_unix_time(map_capture_time_to_datetime(desc["MAPCaptureTime"])), - angle=desc.get("MAPCompassHeading", {}).get("TrueHeading"), - width=None, - height=None, - **kwargs, - ) - - -def _encode_point(p: geo.Point) -> T.Sequence[float | int | None]: - entry = [ - int(p.time * 1000), - round(p.lon, _COORDINATES_PRECISION), - round(p.lat, _COORDINATES_PRECISION), - round(p.alt, _ALTITUDE_PRECISION) if p.alt is not None else None, - round(p.angle, _ANGLE_PRECISION) if p.angle is not None else None, - ] - return entry - - -def _decode_point(entry: T.Sequence[T.Any]) -> geo.Point: - time_ms, lon, lat, alt, angle = entry - return geo.Point(time=time_ms / 1000, lon=lon, lat=lat, alt=alt, angle=angle) - - -def _from_video_desc(desc: VideoDescription) -> VideoMetadata: - return VideoMetadata( - filename=Path(desc["filename"]), - md5sum=desc["md5sum"], - filesize=desc["filesize"], - filetype=FileType(desc["filetype"]), - points=[_decode_point(entry) for entry in desc["MAPGPSTrack"]], - make=desc.get("MAPDeviceMake"), - model=desc.get("MAPDeviceModel"), - ) - - -def validate_and_fail_desc(desc: DescriptionOrError) -> DescriptionOrError: - if "error" in desc: - return desc - - filetype = desc.get("filetype") - try: - if filetype == FileType.IMAGE.value: - validate_image_desc(desc) - else: - validate_video_desc(desc) - except exceptions.MapillaryMetadataValidationError as ex: - return _describe_error_desc( - ex, - Path(desc["filename"]), - filetype=FileType(filetype) if filetype else None, - ) - - if not os.path.isfile(desc["filename"]): - return _describe_error_desc( - exceptions.MapillaryMetadataValidationError( - f"No such file {desc['filename']}" - ), - Path(desc["filename"]), - filetype=FileType(filetype) if filetype else None, - ) - - return desc - - -# Same as validate_and_fail_desc but for the metadata dataclass -def validate_and_fail_metadata(metadata: MetadataOrError) -> MetadataOrError: - if isinstance(metadata, ErrorMetadata): - return metadata - - if isinstance(metadata, ImageMetadata): - filetype = FileType.IMAGE - validate = validate_image_desc - else: - assert isinstance(metadata, VideoMetadata) - filetype = metadata.filetype - validate = validate_video_desc - - try: - validate(as_desc(metadata)) - except exceptions.MapillaryMetadataValidationError as ex: - # rethrow because the original error is too verbose - return describe_error_metadata( - ex, - metadata.filename, - filetype=filetype, - ) - - if not metadata.filename.is_file(): - return describe_error_metadata( - exceptions.MapillaryMetadataValidationError( - f"No such file {metadata.filename}" - ), - metadata.filename, - filetype=filetype, - ) - - return metadata - - -def desc_file_to_exif( - desc: ImageDescription, -) -> ImageDescription: - not_needed = ["MAPSequenceUUID"] - removed = { - key: value - for key, value in desc.items() - if key.startswith("MAP") and key not in not_needed - } - return T.cast(ImageDescription, removed) - - def group_and_sort_images( metadatas: T.Iterable[ImageMetadata], ) -> dict[str, list[ImageMetadata]]: @@ -758,7 +206,3 @@ def update_sequence_md5sum(sequence: T.Iterable[ImageMetadata]) -> str: assert isinstance(metadata.md5sum, str), "md5sum should be calculated" md5.update(metadata.md5sum.encode("utf-8")) return md5.hexdigest() - - -if __name__ == "__main__": - print(json.dumps(ImageVideoDescriptionFileSchema, indent=4)) diff --git a/mapillary_tools/upload.py b/mapillary_tools/upload.py index 6387d77db..3a65d2ab4 100644 --- a/mapillary_tools/upload.py +++ b/mapillary_tools/upload.py @@ -1,5 +1,6 @@ from __future__ import annotations +import contextlib import json import logging import os @@ -15,6 +16,7 @@ from . import ( api_v4, + config, constants, exceptions, geo, @@ -29,6 +31,7 @@ from .camm import camm_builder, camm_parser from .gpmf import gpmf_parser from .mp4 import simple_mp4_builder +from .serializer.description import DescriptionJSONSerializer from .types import FileType JSONDict = T.Dict[str, T.Union[str, int, float, None]] @@ -40,87 +43,94 @@ class UploadedAlreadyError(uploader.SequenceError): pass -def _load_validate_metadatas_from_desc_path( - desc_path: str | None, import_paths: T.Sequence[Path] -) -> list[types.Metadata]: - is_default_desc_path = False - if desc_path is None: - is_default_desc_path = True - if len(import_paths) == 1 and import_paths[0].is_dir(): - desc_path = str( - import_paths[0].joinpath(constants.IMAGE_DESCRIPTION_FILENAME) - ) - else: - if 1 < len(import_paths): - raise exceptions.MapillaryBadParameterError( - "The description path must be specified (with --desc_path) when uploading multiple paths", - ) +def upload( + import_path: Path | T.Sequence[Path], + user_items: config.UserItem, + desc_path: str | None = None, + _metadatas_from_process: T.Sequence[types.MetadataOrError] | None = None, + dry_run=False, + skip_subfolders=False, +) -> None: + import_paths = _normalize_import_paths(import_path) + + metadatas = _load_descs(_metadatas_from_process, import_paths, desc_path) + + jsonschema.validate(instance=user_items, schema=config.UserItemSchema) + + # Setup the emitter -- the order matters here + + emitter = uploader.EventEmitter() + + # When dry_run mode is on, we disable history by default. + # But we need dry_run for tests, so we added MAPILLARY__ENABLE_UPLOAD_HISTORY_FOR_DRY_RUN + # and when it is on, we enable history regardless of dry_run + enable_history = constants.MAPILLARY_UPLOAD_HISTORY_PATH and ( + not dry_run or constants.MAPILLARY__ENABLE_UPLOAD_HISTORY_FOR_DRY_RUN + ) + + # Put it first one to check duplications first + if enable_history: + upload_run_params: JSONDict = { + # Null if multiple paths provided + "import_path": str(import_path) if isinstance(import_path, Path) else None, + "organization_key": user_items.get("MAPOrganizationKey"), + "user_key": user_items.get("MAPSettingsUserKey"), + "version": VERSION, + "run_at": time.time(), + } + _setup_history(emitter, upload_run_params, metadatas) + + # Set up tdqm + _setup_tdqm(emitter) + + # Now stats is empty but it will collect during ALL uploads + stats = _setup_api_stats(emitter) + + # Send the progress via IPC, and log the progress in debug mode + _setup_ipc(emitter) + + mly_uploader = uploader.Uploader(user_items, emitter=emitter, dry_run=dry_run) + + results = _gen_upload_everything( + mly_uploader, metadatas, import_paths, skip_subfolders + ) + + upload_successes = 0 + upload_errors: list[Exception] = [] + + # The real upload happens sequentially here + try: + for _, result in results: + if result.error is not None: + upload_errors.append(_continue_or_fail(result.error)) else: - raise exceptions.MapillaryBadParameterError( - "The description path must be specified (with --desc_path) when uploading a single file", - ) + upload_successes += 1 - descs: list[types.DescriptionOrError] = [] + except Exception as ex: + # Fatal error: log and raise + if not dry_run: + _api_logging_failed(_summarize(stats), ex) + raise ex - if desc_path == "-": - try: - descs = json.load(sys.stdin) - except json.JSONDecodeError as ex: - raise exceptions.MapillaryInvalidDescriptionFile( - f"Invalid JSON stream from stdin: {ex}" - ) from ex else: - if not os.path.isfile(desc_path): - if is_default_desc_path: - raise exceptions.MapillaryFileNotFoundError( - f"Description file {desc_path} not found. Has the directory been processed yet?" - ) - else: - raise exceptions.MapillaryFileNotFoundError( - f"Description file {desc_path} not found" - ) - with open(desc_path) as fp: - try: - descs = json.load(fp) - except json.JSONDecodeError as ex: - raise exceptions.MapillaryInvalidDescriptionFile( - f"Invalid JSON file {desc_path}: {ex}" - ) from ex - - # the descs load from stdin or json file may contain invalid entries - validated_descs = [ - types.validate_and_fail_desc(desc) - for desc in descs - # skip error descriptions - if "error" not in desc - ] + if not dry_run: + _api_logging_finished(_summarize(stats)) - # throw if we found any invalid descs - invalid_descs = [desc for desc in validated_descs if "error" in desc] - if invalid_descs: - for desc in invalid_descs: - LOG.error("Invalid description entry: %s", json.dumps(desc)) - raise exceptions.MapillaryInvalidDescriptionFile( - f"Found {len(invalid_descs)} invalid descriptions" + finally: + # We collected stats after every upload is finished + assert upload_successes == len(stats), ( + f"Expect {upload_successes} success but got {stats}" ) - - # validated_descs should contain no errors - return [ - types.from_desc(T.cast(types.Description, desc)) for desc in validated_descs - ] + _show_upload_summary(stats, upload_errors) -def zip_images( - import_path: Path, - zip_dir: Path, - desc_path: str | None = None, -): +def zip_images(import_path: Path, zip_dir: Path, desc_path: str | None = None): if not import_path.is_dir(): raise exceptions.MapillaryFileNotFoundError( f"Import directory not found: {import_path}" ) - metadatas = _load_validate_metadatas_from_desc_path(desc_path, [import_path]) + metadatas = _load_valid_metadatas_from_desc_path([import_path], desc_path) if not metadatas: LOG.warning("No images or videos found in %s", desc_path) @@ -422,34 +432,6 @@ def _api_logging_failed(payload: dict, exc: Exception): LOG.warning("Error from API Logging for action %s", action, exc_info=True) -def _load_descs( - _metadatas_from_process: T.Sequence[types.MetadataOrError] | None, - desc_path: str | None, - import_paths: T.Sequence[Path], -) -> list[types.Metadata]: - metadatas: list[types.Metadata] - - if _metadatas_from_process is not None: - metadatas, _ = types.separate_errors(_metadatas_from_process) - else: - metadatas = _load_validate_metadatas_from_desc_path(desc_path, import_paths) - - # Make sure all metadatas have sequence uuid assigned - # It is used to find the right sequence when writing upload history - missing_sequence_uuid = str(uuid.uuid4()) - for metadata in metadatas: - if isinstance(metadata, types.ImageMetadata): - if metadata.MAPSequenceUUID is None: - metadata.MAPSequenceUUID = missing_sequence_uuid - - for metadata in metadatas: - assert isinstance(metadata, (types.ImageMetadata, types.VideoMetadata)) - if isinstance(metadata, types.ImageMetadata): - assert metadata.MAPSequenceUUID is not None - - return metadatas - - _M = T.TypeVar("_M", bound=types.Metadata) @@ -634,87 +616,6 @@ def _continue_or_fail(ex: Exception) -> Exception: raise ex -def upload( - import_path: Path | T.Sequence[Path], - user_items: types.UserItem, - desc_path: str | None = None, - _metadatas_from_process: T.Sequence[types.MetadataOrError] | None = None, - dry_run=False, - skip_subfolders=False, -) -> None: - import_paths = _normalize_import_paths(import_path) - - metadatas = _load_descs(_metadatas_from_process, desc_path, import_paths) - - jsonschema.validate(instance=user_items, schema=types.UserItemSchema) - - # Setup the emitter -- the order matters here - - emitter = uploader.EventEmitter() - - # When dry_run mode is on, we disable history by default. - # But we need dry_run for tests, so we added MAPILLARY__ENABLE_UPLOAD_HISTORY_FOR_DRY_RUN - # and when it is on, we enable history regardless of dry_run - enable_history = constants.MAPILLARY_UPLOAD_HISTORY_PATH and ( - not dry_run or constants.MAPILLARY__ENABLE_UPLOAD_HISTORY_FOR_DRY_RUN - ) - - # Put it first one to check duplications first - if enable_history: - upload_run_params: JSONDict = { - # Null if multiple paths provided - "import_path": str(import_path) if isinstance(import_path, Path) else None, - "organization_key": user_items.get("MAPOrganizationKey"), - "user_key": user_items.get("MAPSettingsUserKey"), - "version": VERSION, - "run_at": time.time(), - } - _setup_history(emitter, upload_run_params, metadatas) - - # Set up tdqm - _setup_tdqm(emitter) - - # Now stats is empty but it will collect during ALL uploads - stats = _setup_api_stats(emitter) - - # Send the progress via IPC, and log the progress in debug mode - _setup_ipc(emitter) - - mly_uploader = uploader.Uploader(user_items, emitter=emitter, dry_run=dry_run) - - results = _gen_upload_everything( - mly_uploader, metadatas, import_paths, skip_subfolders - ) - - upload_successes = 0 - upload_errors: list[Exception] = [] - - # The real upload happens sequentially here - try: - for _, result in results: - if result.error is not None: - upload_errors.append(_continue_or_fail(result.error)) - else: - upload_successes += 1 - - except Exception as ex: - # Fatal error: log and raise - if not dry_run: - _api_logging_failed(_summarize(stats), ex) - raise ex - - else: - if not dry_run: - _api_logging_finished(_summarize(stats)) - - finally: - # We collected stats after every upload is finished - assert upload_successes == len(stats), ( - f"Expect {upload_successes} success but got {stats}" - ) - _show_upload_summary(stats, upload_errors) - - def _gen_upload_zipfiles( mly_uploader: uploader.Uploader, zip_paths: T.Sequence[Path], @@ -733,3 +634,66 @@ def _gen_upload_zipfiles( yield zip_path, uploader.UploadResult(error=ex) else: yield zip_path, uploader.UploadResult(result=cluster_id) + + +def _load_descs( + _metadatas_from_process: T.Sequence[types.MetadataOrError] | None, + import_paths: T.Sequence[Path], + desc_path: str | None, +) -> list[types.Metadata]: + metadatas: list[types.Metadata] + + if _metadatas_from_process is not None: + metadatas, _ = types.separate_errors(_metadatas_from_process) + else: + metadatas = _load_valid_metadatas_from_desc_path(import_paths, desc_path) + + # Make sure all metadatas have sequence uuid assigned + # It is used to find the right sequence when writing upload history + missing_sequence_uuid = str(uuid.uuid4()) + for metadata in metadatas: + if isinstance(metadata, types.ImageMetadata): + if metadata.MAPSequenceUUID is None: + metadata.MAPSequenceUUID = missing_sequence_uuid + + for metadata in metadatas: + assert isinstance(metadata, (types.ImageMetadata, types.VideoMetadata)) + if isinstance(metadata, types.ImageMetadata): + assert metadata.MAPSequenceUUID is not None + + return metadatas + + +def _load_valid_metadatas_from_desc_path( + import_paths: T.Sequence[Path], desc_path: str | None +) -> list[types.Metadata]: + if desc_path is None: + desc_path = _find_desc_path(import_paths) + + with ( + contextlib.nullcontext(sys.stdin.buffer) + if desc_path == "-" + else open(desc_path, "rb") + ) as fp: + try: + metadatas = DescriptionJSONSerializer.deserialize_stream(fp) + except json.JSONDecodeError as ex: + raise exceptions.MapillaryInvalidDescriptionFile( + f"Invalid JSON stream from {desc_path}: {ex}" + ) from ex + + return metadatas + + +def _find_desc_path(import_paths: T.Sequence[Path]) -> str: + if len(import_paths) == 1 and import_paths[0].is_dir(): + return str(import_paths[0].joinpath(constants.IMAGE_DESCRIPTION_FILENAME)) + + if 1 < len(import_paths): + raise exceptions.MapillaryBadParameterError( + "The description path must be specified (with --desc_path) when uploading multiple paths", + ) + else: + raise exceptions.MapillaryBadParameterError( + "The description path must be specified (with --desc_path) when uploading a single file", + ) diff --git a/mapillary_tools/uploader.py b/mapillary_tools/uploader.py index ca0b941cf..368aaff65 100644 --- a/mapillary_tools/uploader.py +++ b/mapillary_tools/uploader.py @@ -19,7 +19,12 @@ import requests -from . import api_v4, constants, exif_write, types, upload_api_v4, utils +from . import api_v4, config, constants, exif_write, types, upload_api_v4, utils +from .serializer.description import ( + desc_file_to_exif, + DescriptionJSONSerializer, + validate_image_desc, +) LOG = logging.getLogger(__name__) @@ -255,7 +260,10 @@ def _dump_image_bytes(cls, metadata: types.ImageMetadata) -> bytes: # The cast is to fix the type checker error edit.add_image_description( - T.cast(T.Dict, types.desc_file_to_exif(types.as_desc(metadata))) + T.cast( + T.Dict, + desc_file_to_exif(DescriptionJSONSerializer.as_desc(metadata)), + ) ) try: @@ -463,7 +471,7 @@ def upload_images( class Uploader: def __init__( self, - user_items: types.UserItem, + user_items: config.UserItem, emitter: EventEmitter | None = None, chunk_size: int = int(constants.UPLOAD_CHUNK_SIZE_MB * 1024 * 1024), dry_run=False, @@ -657,7 +665,7 @@ def _maybe_emit( def _validate_metadatas(metadatas: T.Sequence[types.ImageMetadata]): for metadata in metadatas: - types.validate_image_desc(types.as_desc(metadata)) + validate_image_desc(DescriptionJSONSerializer.as_desc(metadata)) if not metadata.filename.is_file(): raise FileNotFoundError(f"No such file {metadata.filename}") diff --git a/tests/integration/test_upload.py b/tests/integration/test_upload.py index 04a258b29..31a33d4c1 100644 --- a/tests/integration/test_upload.py +++ b/tests/integration/test_upload.py @@ -92,7 +92,7 @@ def test_upload_wrong_descs(setup_data: py.path.local, setup_upload: py.path.loc "filename": str(setup_data.join("not_found")), "filetype": "image", "MAPLatitude": 1, - "MAPLongitude": 1, + "MAPLongitude": 181, "MAPCaptureTime": "1970_01_01_00_00_02_000", "MAPCompassHeading": {"TrueHeading": 17.0, "MagneticHeading": 17.0}, }, @@ -104,4 +104,44 @@ def test_upload_wrong_descs(setup_data: py.path.local, setup_upload: py.path.loc f"{EXECUTABLE} upload {UPLOAD_FLAGS} --desc_path={desc_path} {setup_data} {setup_data} {setup_data}/images/DSC00001.JPG", shell=True, ) - assert x.returncode == 4, x.stderr + assert x.returncode == 15, x.stderr + + +@pytest.mark.usefixtures("setup_config") +def test_upload_read_descs_from_stdin( + setup_data: py.path.local, setup_upload: py.path.local +): + subprocess.run( + f"{EXECUTABLE} process {PROCESS_FLAGS} --file_types=image {setup_data}", + shell=True, + check=True, + ) + + descs = [ + { + "filename": "foo.jpg", + "filetype": "image", + "MAPLatitude": 1.0, + "MAPLongitude": 2.0, + "MAPCaptureTime": "2020_01_02_11_12_13_123456", + }, + ] + descs_json = json.dumps(descs) + + process = subprocess.Popen( + f"{EXECUTABLE} process_and_upload {UPLOAD_FLAGS} --file_types=image {setup_data}", + stdin=subprocess.PIPE, + text=True, + shell=True, + ) + + stdout, stderr = process.communicate(input=descs_json) + assert process.returncode == 0, stderr + + uploaded_descs: list[dict] = sum(extract_all_uploaded_descs(Path(setup_upload)), []) + assert len(uploaded_descs) > 0, "No images were uploaded" + + assert_contains_image_descs( + Path(setup_data.join("mapillary_image_description.json")), + uploaded_descs, + ) diff --git a/tests/unit/test_description.py b/tests/unit/test_description.py index 2e4097df7..f6aad7a8a 100644 --- a/tests/unit/test_description.py +++ b/tests/unit/test_description.py @@ -1,9 +1,10 @@ import json +from pathlib import Path from mapillary_tools.exceptions import MapillaryMetadataValidationError -from mapillary_tools.types import ( +from mapillary_tools.serializer.description import ( + DescriptionJSONSerializer, ImageVideoDescriptionFileSchema, - validate_and_fail_desc, validate_image_desc, ) @@ -43,6 +44,7 @@ def test_validate_descs_not_ok(): "MAPLongitude": 2, "filename": "foo", "filetype": "image", + "expected_error_message": "'MAPCaptureTime' is a required property", }, { "MAPLatitude": -90.1, @@ -50,6 +52,7 @@ def test_validate_descs_not_ok(): "MAPCaptureTime": "1020_01_02_11_33_13_123", "filename": "foo", "filetype": "image", + "expected_error_message": "-90.1 is less than the minimum of -90", }, { "MAPLatitude": 1, @@ -57,6 +60,7 @@ def test_validate_descs_not_ok(): "MAPCaptureTime": "3020_01_02_11_12_13_000", "filename": "foo", "filetype": "image", + "expected_error_message": "-180.2 is less than the minimum of -180", }, { "MAPLatitude": -90, @@ -64,29 +68,19 @@ def test_validate_descs_not_ok(): "MAPCaptureTime": "2000_12_00_10_20_10_000", "filename": "foo", "filetype": "image", + "expected_error_message": "time data '2000_12_00_10_20_10_000' does not match format '%Y_%m_%d_%H_%M_%S_%f'", }, ] errors = 0 for desc in descs: + expected_error_message = desc.pop("expected_error_message", None) try: validate_image_desc(desc) - except MapillaryMetadataValidationError: + except MapillaryMetadataValidationError as ex: + assert expected_error_message == str(ex) errors += 1 assert errors == len(descs) - validated = [validate_and_fail_desc(desc) for desc in descs] - actual_errors = [ - desc - for desc in validated - if desc["error"]["type"] == "MapillaryMetadataValidationError" - ] - assert { - "'MAPCaptureTime' is a required property", - "-90.1 is less than the minimum of -90", - "-180.2 is less than the minimum of -180", - "time data '2000_12_00_10_20_10_000' does not match format '%Y_%m_%d_%H_%M_%S_%f'", - } == set(e["error"]["message"] for e in actual_errors) - def test_validate_image_description_schema(): with open("./schema/image_description_schema.json") as fp: @@ -94,3 +88,27 @@ def test_validate_image_description_schema(): assert json.dumps(schema, sort_keys=True) == json.dumps( ImageVideoDescriptionFileSchema, sort_keys=True ) + + +def test_serialize_empty(): + assert b"[]" == DescriptionJSONSerializer.serialize([]) + + +def test_serialize_image_description_ok(): + desc = [ + { + "MAPLatitude": 1, + "MAPLongitude": 2, + "MAPCaptureTime": "2020_01_02_11_12_13_1", + "filename": "foo你好", + "filetype": "image", + } + ] + metadatas = DescriptionJSONSerializer.deserialize(json.dumps(desc).encode("utf-8")) + actual = json.loads(DescriptionJSONSerializer.serialize(metadatas)) + assert len(actual) == 1, actual + actual = actual[0] + assert "2020_01_02_11_12_13_100" == actual["MAPCaptureTime"] + assert "foo你好" == Path(actual["filename"]).name + assert 1 == actual["MAPLatitude"] + assert 2 == actual["MAPLongitude"] diff --git a/tests/unit/test_exceptions.py b/tests/unit/test_exceptions.py index 4b2e5220d..d70c6e1b2 100644 --- a/tests/unit/test_exceptions.py +++ b/tests/unit/test_exceptions.py @@ -2,6 +2,7 @@ from pathlib import Path from mapillary_tools import exceptions, types +from mapillary_tools.serializer import description def test_all(): @@ -24,5 +25,5 @@ def test_all(): e = exc("hello") # should not raise json.dumps( - types._describe_error_desc(e, Path("test.jpg"), types.FileType.IMAGE) + description._describe_error_desc(e, Path("test.jpg"), types.FileType.IMAGE) ) diff --git a/tests/unit/test_sample_video.py b/tests/unit/test_sample_video.py index 23c2fadeb..6e0d416d1 100644 --- a/tests/unit/test_sample_video.py +++ b/tests/unit/test_sample_video.py @@ -8,7 +8,8 @@ import py.path import pytest -from mapillary_tools import exif_read, ffmpeg, sample_video, types +from mapillary_tools import exif_read, ffmpeg, sample_video +from mapillary_tools.serializer import description _PWD = Path(os.path.dirname(os.path.abspath(__file__))) @@ -67,7 +68,7 @@ def test_sample_video(tmpdir: py.path.local, setup_mock): rerun=True, ) samples = sample_dir.join("hello.mp4").listdir() - video_start_time = types.map_capture_time_to_datetime("2021_08_10_14_37_05_023") + video_start_time = description.parse_capture_time("2021_08_10_14_37_05_023") _validate_interval([Path(s) for s in samples], video_start_time) @@ -83,7 +84,7 @@ def test_sample_single_video(tmpdir: py.path.local, setup_mock): rerun=True, ) samples = sample_dir.join("hello.mp4").listdir() - video_start_time = types.map_capture_time_to_datetime("2021_08_10_14_37_05_023") + video_start_time = description.parse_capture_time("2021_08_10_14_37_05_023") _validate_interval([Path(s) for s in samples], video_start_time) @@ -92,7 +93,7 @@ def test_sample_video_with_start_time(tmpdir: py.path.local, setup_mock): video_dir = root.joinpath("videos") sample_dir = tmpdir.mkdir("sampled_video_frames") video_start_time_str = "2020_08_10_14_37_05_023" - video_start_time = types.map_capture_time_to_datetime(video_start_time_str) + video_start_time = description.parse_capture_time(video_start_time_str) sample_video.sample_video( video_dir, Path(sample_dir), diff --git a/tests/unit/test_sequence_processing.py b/tests/unit/test_sequence_processing.py index 2064238c0..5ff306393 100644 --- a/tests/unit/test_sequence_processing.py +++ b/tests/unit/test_sequence_processing.py @@ -13,6 +13,7 @@ process_sequence_properties as psp, types, ) +from mapillary_tools.serializer import description def _make_image_metadata( @@ -481,7 +482,9 @@ def test_process_finalize(setup_data): # "filetype": "image", # }, ] - assert expected == [types.as_desc(d) for d in actual] + assert expected == [ + description.DescriptionJSONSerializer.as_desc(d) for d in actual + ] def test_cut_by_pixels(tmpdir: py.path.local): diff --git a/tests/unit/test_types.py b/tests/unit/test_types.py index 6b0fdbdd4..9115bc466 100644 --- a/tests/unit/test_types.py +++ b/tests/unit/test_types.py @@ -2,6 +2,7 @@ from pathlib import Path from mapillary_tools import geo, types +from mapillary_tools.serializer import description def test_desc(): @@ -39,9 +40,9 @@ def test_desc(): ), ] for metadata in metadatas: - desc = types.as_desc(metadata) - types.validate_image_desc(desc) - actual = types.from_desc(desc) + desc = description.DescriptionJSONSerializer.as_desc(metadata) + description.validate_image_desc(desc) + actual = description.DescriptionJSONSerializer.from_desc(desc) assert metadata == actual @@ -69,29 +70,27 @@ def test_desc_video(): ), ] for metadata in ds: - desc = types._as_video_desc(metadata) - types.validate_video_desc(desc) - actual = types._from_video_desc(desc) + desc = description.DescriptionJSONSerializer._as_video_desc(metadata) + description.validate_video_desc(desc) + actual = description.DescriptionJSONSerializer._from_video_desc(desc) assert metadata == actual def test_datetimes(): - ct = types.datetime_to_map_capture_time(0) + ct = description.build_capture_time(0) assert ct == "1970_01_01_00_00_00_000" - ct = types.datetime_to_map_capture_time(0.123456) + ct = description.build_capture_time(0.123456) assert ct == "1970_01_01_00_00_00_123" - ct = types.datetime_to_map_capture_time(0.000456) + ct = description.build_capture_time(0.000456) assert ct == "1970_01_01_00_00_00_000" - dt = types.map_capture_time_to_datetime(ct) + dt = description.parse_capture_time(ct) assert dt == datetime.datetime(1970, 1, 1, 0, 0, tzinfo=datetime.timezone.utc) x = datetime.datetime.fromisoformat("2020-01-01T00:00:12.123567+08:00") - assert "2019_12_31_16_00_12_123" == types.datetime_to_map_capture_time(x) + assert "2019_12_31_16_00_12_123" == description.build_capture_time(x) assert ( abs( geo.as_unix_time( - types.map_capture_time_to_datetime( - types.datetime_to_map_capture_time(x) - ) + description.parse_capture_time(description.build_capture_time(x)) ) - geo.as_unix_time(x) ) @@ -101,9 +100,7 @@ def test_datetimes(): assert ( abs( geo.as_unix_time( - types.map_capture_time_to_datetime( - types.datetime_to_map_capture_time(x) - ) + description.parse_capture_time(description.build_capture_time(x)) ) - geo.as_unix_time(x) ) @@ -113,9 +110,7 @@ def test_datetimes(): assert ( abs( geo.as_unix_time( - types.map_capture_time_to_datetime( - types.datetime_to_map_capture_time(x) - ) + description.parse_capture_time(description.build_capture_time(x)) ) - geo.as_unix_time(x) ) diff --git a/tests/unit/test_uploader.py b/tests/unit/test_uploader.py index f9367d900..97eaa11c9 100644 --- a/tests/unit/test_uploader.py +++ b/tests/unit/test_uploader.py @@ -5,7 +5,8 @@ import pytest -from mapillary_tools import api_v4, types, uploader +from mapillary_tools import api_v4, uploader +from mapillary_tools.serializer import description from ..integration.fixtures import extract_all_uploaded_descs, setup_upload @@ -28,7 +29,7 @@ def test_upload_images(setup_unittest_data: py.path.local, setup_upload: py.path {"user_upload_token": "YOUR_USER_ACCESS_TOKEN"}, dry_run=True ) test_exif = setup_unittest_data.join("test_exif.jpg") - descs: T.List[types.DescriptionOrError] = [ + descs: T.List[description.DescriptionOrError] = [ { "MAPLatitude": 58.5927694, "MAPLongitude": 16.1840944, @@ -49,7 +50,10 @@ def test_upload_images(setup_unittest_data: py.path.local, setup_upload: py.path results = list( uploader.ZipImageSequence.zip_images_and_upload( mly_uploader, - [types.from_desc(T.cast(T.Any, desc)) for desc in descs], + [ + description.DescriptionJSONSerializer.from_desc(T.cast(T.Any, desc)) + for desc in descs + ], ) ) assert len(results) == 1 @@ -64,7 +68,7 @@ def test_upload_images_multiple_sequences( ): test_exif = setup_unittest_data.join("test_exif.jpg") fixed_exif = setup_unittest_data.join("fixed_exif.jpg") - descs: T.List[types.DescriptionOrError] = [ + descs: T.List[description.DescriptionOrError] = [ { "MAPLatitude": 58.5927694, "MAPLongitude": 16.1840944, @@ -104,7 +108,10 @@ def test_upload_images_multiple_sequences( results = list( uploader.ZipImageSequence.zip_images_and_upload( mly_uploader, - [types.from_desc(T.cast(T.Any, desc)) for desc in descs], + [ + description.DescriptionJSONSerializer.from_desc(T.cast(T.Any, desc)) + for desc in descs + ], ) ) assert len(results) == 2 @@ -120,7 +127,7 @@ def test_upload_zip( test_exif2 = setup_unittest_data.join("another_directory").join("test_exif.jpg") test_exif.copy(test_exif2) - descs: T.List[types.DescriptionOrError] = [ + descs: T.List[description.DescriptionOrError] = [ { "MAPLatitude": 58.5927694, "MAPLongitude": 16.1840944, @@ -151,7 +158,11 @@ def test_upload_zip( ] zip_dir = setup_unittest_data.mkdir("zip_dir") uploader.ZipImageSequence.zip_images( - [types.from_desc(T.cast(T.Any, desc)) for desc in descs], Path(zip_dir) + [ + description.DescriptionJSONSerializer.from_desc(T.cast(T.Any, desc)) + for desc in descs + ], + Path(zip_dir), ) assert len(zip_dir.listdir()) == 2, list(zip_dir.listdir())