chore: upgrade types (#741)

* git rm -r mapillary_tools/video_data_extraction/

* upgrade types with prompt:

Update types with the following rules:

1. Relace T.Optional with | and T.List with list, and so on
2. Keep compatible with Python>=3.8
3. Do no use `from typing import ?`. Instead, always use `import typing as T`

* fix ruff check

* fix tests for 3.8

* integration tests for python3.8

Author: ptpt

mapillary_tools/api_v4.py

@@ -135,7 +135,7 @@ def _log_debug_response(resp: requests.Response):
     if logging.getLogger().getEffectiveLevel() <= logging.DEBUG:
         return
 
-    data: T.Union[str, bytes]
+    data: str | bytes
     try:
         data = _truncate(dumps(_sanitize(resp.json())))
     except Exception:
@@ -148,7 +148,7 @@ def readable_http_error(ex: requests.HTTPError) -> str:
     req = ex.request
     resp = ex.response
 
-    data: T.Union[str, bytes]
+    data: str | bytes
     try:
         data = _truncate(dumps(_sanitize(resp.json())))
     except Exception:
@@ -284,7 +284,7 @@ def get_upload_token(email: str, password: str) -> requests.Response:
 
 
 def fetch_organization(
-    user_access_token: str, organization_id: T.Union[int, str]
+    user_access_token: str, organization_id: int | str
 ) -> requests.Response:
     resp = request_get(
         f"{MAPILLARY_GRAPH_API_ENDPOINT}/{organization_id}",
@@ -329,7 +329,7 @@ def fetch_user_or_me(
 ]
 
 
-def log_event(action_type: ActionType, properties: T.Dict) -> requests.Response:
+def log_event(action_type: ActionType, properties: dict) -> requests.Response:
     resp = request_post(
         f"{MAPILLARY_GRAPH_API_ENDPOINT}/logging",
         json={

mapillary_tools/camm/camm_parser.py

@@ -373,7 +373,7 @@ def serialize(cls, data: telemetry.MagnetometerData) -> bytes:
 assert len(SAMPLE_ENTRY_CLS_BY_CAMM_TYPE) == 5, SAMPLE_ENTRY_CLS_BY_CAMM_TYPE.keys()
 
 
-_SWITCH: T.Dict[int, C.Struct] = {
+_SWITCH: dict[int, C.Struct] = {
     # Angle_axis
     CAMMType.ANGLE_AXIS.value: _Float[3],  # type: ignore
     # Exposure time
@@ -436,7 +436,7 @@ def _parse_telemetry_from_sample(
 
 def _filter_telemetry_by_elst_segments(
     measurements: T.Iterable[TelemetryMeasurement],
-    elst: T.Sequence[T.Tuple[float, float]],
+    elst: T.Sequence[tuple[float, float]],
 ) -> T.Generator[TelemetryMeasurement, None, None]:
     empty_elst = [entry for entry in elst if entry[0] == -1]
     if empty_elst:
@@ -466,8 +466,8 @@ def _filter_telemetry_by_elst_segments(
 
 
 def elst_entry_to_seconds(
-    entry: T.Dict, movie_timescale: int, media_timescale: int
-) -> T.Tuple[float, float]:
+    entry: dict, movie_timescale: int, media_timescale: int
+) -> tuple[float, float]:
     assert movie_timescale > 0, "expected positive movie_timescale"
     assert media_timescale > 0, "expected positive media_timescale"
     media_time, duration = entry["media_time"], entry["segment_duration"]
@@ -477,7 +477,7 @@ def elst_entry_to_seconds(
     return (media_time, duration)
 
 
-def _is_camm_description(description: T.Dict) -> bool:
+def _is_camm_description(description: dict) -> bool:
     return description["format"] == b"camm"
 
 

mapillary_tools/commands/__main__.py

@@ -2,7 +2,6 @@
 import enum
 import logging
 import sys
-import typing as T
 from pathlib import Path
 
 import requests
@@ -86,7 +85,7 @@ def configure_logger(logger: logging.Logger, stream=None) -> None:
     logger.addHandler(handler)
 
 
-def _log_params(argvars: T.Dict) -> None:
+def _log_params(argvars: dict) -> None:
     MAX_ENTRIES = 5
 
     def _stringify(x) -> str:

mapillary_tools/config.py

@@ -1,3 +1,5 @@
+from __future__ import annotations
+
 import configparser
 import os
 import typing as T
@@ -35,8 +37,8 @@ def _load_config(config_path: str) -> configparser.ConfigParser:
 
 
 def load_user(
-    profile_name: str, config_path: T.Optional[str] = None
-) -> T.Optional[types.UserItem]:
+    profile_name: str, config_path: str | None = None
+) -> types.UserItem | None:
     if config_path is None:
         config_path = MAPILLARY_CONFIG_PATH
     config = _load_config(config_path)
@@ -46,7 +48,7 @@ def load_user(
     return T.cast(types.UserItem, user_items)
 
 
-def list_all_users(config_path: T.Optional[str] = None) -> T.Dict[str, types.UserItem]:
+def list_all_users(config_path: str | None = None) -> dict[str, types.UserItem]:
     if config_path is None:
         config_path = MAPILLARY_CONFIG_PATH
     cp = _load_config(config_path)
@@ -58,7 +60,7 @@ def list_all_users(config_path: T.Optional[str] = None) -> T.Dict[str, types.Use
 
 
 def update_config(
-    profile_name: str, user_items: types.UserItem, config_path: T.Optional[str] = None
+    profile_name: str, user_items: types.UserItem, config_path: str | None = None
 ) -> None:
     if config_path is None:
         config_path = MAPILLARY_CONFIG_PATH
@@ -72,7 +74,7 @@ def update_config(
         config.write(fp)
 
 
-def remove_config(profile_name: str, config_path: T.Optional[str] = None) -> None:
+def remove_config(profile_name: str, config_path: str | None = None) -> None:
     if config_path is None:
         config_path = MAPILLARY_CONFIG_PATH
 

mapillary_tools/constants.py

@@ -1,7 +1,6 @@
 from __future__ import annotations
 
 import os
-import typing as T
 
 import appdirs
 
@@ -52,7 +51,7 @@ def _yes_or_no(val: str) -> bool:
 # It is used to filter out noisy points
 GOPRO_MAX_DOP100 = int(os.getenv(_ENV_PREFIX + "GOPRO_MAX_DOP100", 1000))
 # Within the GPS stream: 0 - no lock, 2 or 3 - 2D or 3D Lock
-GOPRO_GPS_FIXES: T.Set[int] = set(
+GOPRO_GPS_FIXES: set[int] = set(
     int(fix) for fix in os.getenv(_ENV_PREFIX + "GOPRO_GPS_FIXES", "2,3").split(",")
 )
 MAX_UPLOAD_RETRIES: int = int(os.getenv(_ENV_PREFIX + "MAX_UPLOAD_RETRIES", 200))