mapillary-mapillary-tools-779

script
tests
./.git/COMMIT_EDITMSG
./.git/config
./.git/description
./.git/HEAD
./.git/hooks/applypatch-msg.sample
./.git/hooks/commit-msg.sample
./.git/hooks/fsmonitor-watchman.sample
./.git/hooks/post-update.sample
./.git/hooks/pre-applypatch.sample
./.git/hooks/pre-commit.sample
./.git/hooks/pre-merge-commit.sample
./.git/hooks/pre-push.sample
./.git/hooks/pre-rebase.sample
./.git/hooks/pre-receive.sample
./.git/hooks/prepare-commit-msg.sample
./.git/hooks/push-to-checkout.sample
./.git/hooks/sendemail-validate.sample
./.git/hooks/update.sample
./.git/index
./.git/info/exclude
./.git/logs/HEAD
./.github/workflows/codeql-analysis.yml
./.github/workflows/python-package.yml
./.github/workflows/release.yml
./.gitignore

…[3286 chars omitted]…
ts/unit/test_persistent_cache.py:190:    cache = PersistentCache(cache_file)
./tests/unit/test_persistent_cache.py:203:    cache = PersistentCache(cache_file)
./tests/unit/test_persistent_cache.py:228:    cache = PersistentCache(cache_file)
./tests/unit/test_persistent_cache.py:272:    cache = PersistentCache(cache_file)
./tests/unit/test_persistent_cache.py:286:    cache = PersistentCache(cache_file)

# dbm modules are dynamically imported, so here we explicitly import dbm.sqlite3 to make sure pyinstaller include it
# Otherwise you will see: ImportError: no dbm clone found; tried ['dbm.sqlite3', 'dbm.gnu', 'dbm.ndbm', 'dbm.dumb']
try:
    import dbm.sqlite3  # type: ignore
except ImportError:
    pass


from . import constants, types
from .serializer.description import DescriptionJSONSerializer

JSONDict = T.Dict[str, T.Union[str, int, float, None]]

LOG = logging.getLogger(__name__)


def _validate_hexdigits(md5sum: str):
    try:
        assert set(md5sum).issubset(string.hexdigits)
        assert 4 <= len(md5sum)
        _ = int(md5sum, 16)
    except Exception:
        raise ValueError(f"Invalid md5su
…[4326 chars omitted]…
, dict):
            LOG.warning(f"Invalid cache value format: {payload}")
            return {}

        return payload
tests/integration/test_process_and_upload.py:152:def test_process_and_upload_images_only(
tests/unit/test_uploader.py:27:def test_upload_images(setup_unittest_data: py.path.local, setup_upload: py.path.local):
tests/unit/test_uploader.py:68:def test_upload_images_multiple_sequences(

            except FileNotFoundError:
                pass


class ImageSequenceUploader:
    def __init__(self, upload_options: UploadOptions, emitter: EventEmitter):
        self.upload_options = upload_options
        self.emitter = emitter

    def upload_images(
        self, image_metadatas: T.Sequence[types.ImageMetadata]
    ) -> T.Generator[tuple[str, UploadResult], None, None]:
        sequences = types.group_and_sort_images(image_metadatas)

        for sequence_idx, (sequence_uuid, sequence) in enumerate(sequences
…[8628 chars omitted]…
rt len(results) == 2
    actual_descs = sum(extract_all_uploaded_descs(Path(setup_upload)), [])
    assert 2 == len(actual_descs)


def test_upload_zip(
    setup_unittest_data: py.path.local, setup_upload: py.path.local, emitter=None
):
    test_exif = setup_unittest_data.join("test_exif.jpg")
    setup_unittest_data.join("another_directory").mkdir()
197:class EventEmitter:
541:    def upload_images(

                executor.submit(
                    self._upload_images_from_queue,
                    image_queue,
                    lock,
                    upload_interrupted,
                    sequence_progress,
                )
                for _ in range(max_workers)
            ]

            indexed_image_file_handles = []

            try:
                for future in futures:
                 
…[6177 chars omitted]…
except Exception as ex:
                yield video_metadata, UploadResult(error=ex)
                continue

            assert isinstance(video_metadata.md5sum, str), "md5sum should be updated"

            progress: SequenceProgress = {
mapillary_tools/upload.py:559:    image_uploader = uploader.ImageSequenceUploader(
150:UPLOAD_CACHE_DIR: str = os.getenv(
151:    _ENV_PREFIX + "UPLOAD_CACHE_DIR",

 
     def __post_init__(self):
         if self.num_upload_workers <= 0:
@@ -367,6 +368,10 @@ class ZipUploader:
     def zip_images_and_upload(
         cls, uploader: Uploader, image_metadatas: T.Sequence[types.ImageMetadata]
     ) -> T.Generator[tuple[str, UploadResult], None, None]:
+        # The options may be replaced by callers between invocations, but cache
+        # creation 
…[5925 chars omitted]…
token"])
+                )
             )
-        )
-        cache_path_dir.mkdir(parents=True, exist_ok=True)
-        cache_path = cache_path_dir.joinpath("cached_file_handles")
+            cache_path_dir.mkdir(parents=True, exist_ok=True)
+            cache_path = cache_path_dir.joinpath("cached_file_handles")
 
         # Sanitize sensitive segments for logging
         sanitized_cache_path = (

            )
            with cls._wip_file_context(wip_zip_filename) as wip_path:
                with wip_path.open("wb") as wip_fp:
                    cls._zip_sequence_fp(sequence, wip_fp)

    @classmethod
    def zip_images_and_upload(
        cls, uploader: Uploader, image_metadatas: T.Sequence[types.ImageMetadata]
    ) -> T.Generator[tuple
…[3409 chars omitted]…
                                     ^^^^
E       NameError: name 'self' is not defined

mapillary_tools/uploader.py:374: NameError
=========================== short test summary info ============================
FAILED tests/unit/test_uploader.py::test_upload_images - NameError: name 'sel...
!!!!!!!!!!!!!!!!!!!!!!!!!! stopping after 1 failures !!!!!!!!!!!!!!!!!!!!!!!!!!!
1 failed, 60 passed in 40.51s

             self.upload_options.user_items, upload_options
         )
 
     def upload(
-        self, image_metadata: types.ImageMetadata, image_progress: dict[str, T.Any]
+        self,
+        image_metadata:
…[3403 chars omitted]…
token"])
+                )
             )
-        )
-        cache_path_dir.mkdir(parents=True, exist_ok=True)
-        cache_path = cache_path_dir.joinpath("cached_file_handles")
+            cache_path_dir.mkdir(parents=True, exist_ok=True)
+            cache_path = cache_path_dir.joinpath("cached_file_handles")
 
         # Sanitize sensitive segments for logging
         sanitized_cache_path = (

@@ -534,13 +535,22 @@ class ZipUploader:
 
 
 class ImageSequenceUploader:
-    def __init__(self, upload_options: UploadOptions, emitter: EventEmitter):
+    def __init__(
+        self, upload_options: UploadOptions, emitter: EventEmitter | None = None
+    ):
         self.upload_options = upload_options
-        self.emitter = emitter
+        self.emitter = emitter if emitter is not None else EventEmitter()
+        # A single cache owner is important: dbm implementations are not
…[1967 chars omitted]…
 )
 
     def upload(
-        self, image_metadata: types.ImageMetadata, image_progress: dict[str, T.Any]
+        self,
+        image_metadata: types.ImageMetadata,
+        image_progress: dict[str, T.Any],
+        user_session: requests.Session | None = None,
     ) -> str:
         image_bytes = self.dump_image_bytes(image_metadata)
 
 M mapillary_tools/history.py
 M mapillary_tools/uploader.py

        utils.find_images(import_paths, skip_subfolders=skip_subfolders),
    )
    image_uploader = uploader.ImageSequenceUploader(
        mly_uploader.upload_options, emitter=mly_uploader.emitter
    )
    yield from image_uploader.upload_images(image_metadatas)

    # Upload videos
    video_metadatas = _find_metadata_with_filename_existed_in(
        (m for m in metadatas if isinstance(m, types.VideoMetadata)),
        utils.find_videos(import_paths, skip_subfolders=skip_subfo
…[1223 chars omitted]…
lf) -> list[str]:
+        """Return the keys currently stored in the cache."""
+        with self._lock:
+            with dbm.open(self._file, flag="c") as db:
+                return [
+                    key.decode("utf-8") if isinstance(key, bytes) else key
+                    for key in db.keys()
+                ]
+
     def clear_expired(self) -> list[str]:
         s = time.perf_counter()

            session_key = utils.md5sum_fp(fp).hexdigest()

        filetype = progress.get("file_type")
        if filetype is not None:
            session_key = _suffix_session_key(session_key, types.FileType(filetype))

        return session_key


def _validate_metadatas(metadatas: T.Sequence[types.ImageMetadata]):
    for metadata in metadatas:
        validate_image_desc(Des
…[29 chars omitted]…
c(metadata))
        if not metadata.filename.is_file():
            raise FileNotFoundError(f"No such file {metadata.filename}")


def _is_immediate_retriable_exception(ex: BaseException) -> bool:
    if (
    prefixed = f"uuid_{uuid.uuid4().hex}"
    assert _is_uuid(prefixed)
    return prefixed


def _is_uuid(key: str) -> bool:
    return key.startswith("uuid_") or key.startswith("mly_tools_uuid_")

script
tests
./mapillary_tools/history.py:88:class PersistentCache:
./mapillary_tools/uploader.py:55:class UploadOptions:
./mapillary_tools/uploader.py:536:class ImageSequenceUploader:

    "upload_failed",
    "upload_finished",
]


class EventEmitter:
    events: dict[EventName, list]

    def __init__(self):
        self.events = {}

    def on(self, event: EventName):
        def _wrap(callback):
            self.events.setdefault(event, []).append(callback)
            return callback

        return _wrap

    def emit(self, event: EventName, *args, **kwargs):
        for callback in self.events.get(event, []):
            callback(*args, **kwargs)


@dataclasses.dataclass
class UploadResult:
    result: str | None = None
    error: Exception | None = None


class VideoUploader:
    @clas
…[8630 chars omitted]…
apillary_tools/uploader.py:841:        return self.cache.get(key)
mapillary_tools/uploader.py:843:    def _set_file_handle_cache(self, key: str, value: str) -> None:
mapillary_tools/uploader.py:844:        if self.cache is None:
mapillary_tools/uploader.py:850:        self.cache.set(key, value)
mapillary_tools/constants.py:152:    os.path.join(tempfile.gettempdir(), "mapillary_tools", "upload_cache"),

            sequence_md5sum = types.update_sequence_md5sum(sequence)

            sequence_progress: SequenceProgress = {
                "sequence_idx": sequence_idx,
                "total_sequence_count": len(sequences),
                "sequence_image_count": len(sequence),
                "sequence_uuid": sequence_uuid,
                
…[8629 chars omitted]…
None,
            "filetype": "image",
            "MAPSequenceUUID": "sequence_2",
        },
    ]
    mly_uploader = uploader.Uploader(
        uploader.UploadOptions(
            {
                "user_upload_token": "YOUR_USER_ACCESS_TOKEN",
                # will call the API for real
                # "MAPOrganizationKey": "3011753992432185",
            },
            dry_run=True,
        ),

class SingleImageUploader:
    def __init__(
        self,
        upload_options: UploadOptions,
        user_session: requests.Session | None = None,
    ):
        self.upload_options = upload_options
        self.user_session = user_session
        self.cache = self._maybe_create_persistent_cache_instance(
            self.upload_options.user_items, upload_options
        )

    def upload(
        self, image_metadata: types.ImageMetadata, image_progress: dict[str, T.Any]
    ) -> str:
        image_bytes = 
…[6569 chars omitted]…
se(resp)
            # TODO: Validate cluster_id
            cluster_id = body.get("cluster_id")

        progress["cluster_id"] = cluster_id
        self.emitter.emit("upload_finished", progress)
mapillary_tools/upload.py:89:        upload_options = uploader.UploadOptions(
223:    def upload_videos(
320:    def upload_zipfiles(
541:    def upload_images(
746:    def upload(
868:    def upload_stream(

+        self.file = file
+
+    def keys(self) -> list[str]:
+        """Return the currently stored cache keys."""
+        with self._lock:
+            with dbm.open(self._file, flag="c") as db:
+                return [key.decode() if isinstance(key, bytes) else key for key in db.keys()]
 
     def get(self, key: str) -> str | None:
         s = time.perf_c
…[6162 chars omitted]…
token"])
+                )
             )
-        )
-        cache_path_dir.mkdir(parents=True, exist_ok=True)
-        cache_path = cache_path_dir.joinpath("cached_file_handles")
+            cache_path_dir.mkdir(parents=True, exist_ok=True)
+            cache_path = cache_path_dir.joinpath("cached_file_handles")
 
         # Sanitize sensitive segments for logging
         sanitized_cache_path = (

        )
        test_exif = setup_unittest_data.join("test_exif.jpg")
        descs: T.List[description.DescriptionOrError] = [
            {
                "MAPLatitude": 58.5927694,
                "MAPLongitude": 16.1840944,
                "MAPCaptureTime": "2021_02_13_13_24_41
…[1768 chars omitted]…
                                                 ^^^^
E       NameError: name 'self' is not defined

mapillary_tools/uploader.py:375: NameError
=========================== short test summary info ============================
FAILED tests/unit/test_uploader.py::test_upload_images - NameError: name 'sel...
!!!!!!!!!!!!!!!!!!!!!!!!!! stopping after 1 failures !!!!!!!!!!!!!!!!!!!!!!!!!!!
1 failed in 0.73s

        # Keep the operational options current while deliberately retaining the
        # cache created at construction time (including when callers replace the
        # options with a dry-run copy).
        self.cached_image_uploader.upload_options = self.upload_options

        for sequence_idx, (sequ
…[1028 chars omitted]…
replace the
        # options with a dry-run copy).
        self.cached_image_uploader.upload_options = self.upload_options

        for sequence_idx, (sequence_uuid, sequence) in enumerate(sequences.items()):
            LOG.debug(f"Checksum for image sequence {sequence_uuid}...")
            sequence_md5sum = types.update_sequence_md5sum(sequence)

            sequence_progress: SequenceProgress = {

script
tests
./mapillary_tools/history.py:88:class PersistentCache:
./mapillary_tools/uploader.py:55:class UploadOptions:
./mapillary_tools/uploader.py:536:class ImageSequenceUploader:

641:        with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor:
679:    def _upload_images_from_queue(
746:    def upload(
858:        emitter: EventEmitter | None = None,
864:            self.emitter = EventEmitter()
868:    def upload_stream(
920:    def finish_upload(
954:    def _create_upload_service(
975:    def _handle_upload_exception(
1005:    def _upload_name(cls, progress: UploaderProgress):
1036:    def _upload_stream_retryable(

    def on(self, event: EventName):
        def _wrap(callback):
            self.events.setdefault(event, []).append(callback)
            return callback

        return _wrap

    def emit(self, event: EventName, *args, **kwargs):
        for callback in self.events.get(event, []):
            callback(*args, **kwargs)


@dataclasses.dataclass
class UploadResult:
    result: str | None = None
    error: Exception | None = None


class VideoUploader:
    @classmethod
    def upload_videos(
        cls, mly_uploader: Uploader, video_metadatas: T.Sequence[types.VideoMetadata]
    ) -> T.Generator[tuple[types.VideoMetadata, UploadResult], None, None]:
        # If upload in a random order, then 
…[8630 chars omitted]…
      if self.cache is None:
            return None

        if _is_uuid(key):
            return None

mapillary_tools/uploader.py:742:        self.cache = self._maybe_create_persistent_cache_instance(
mapillary_tools/uploader.py:790:    def _maybe_create_persistent_cache_instance(
tests/integration/fixtures.py:69:    os.environ["MAPILLARY_TOOLS_UPLOAD_CACHE_DIR"] = str(tmpdir.mkdir("upload_cache"))

            sequence_md5sum = types.update_sequence_md5sum(sequence)

            sequence_progress: SequenceProgress = {
                "sequence_idx": sequence_idx,
                "total_sequence_count": len(sequences),
                "sequence_image_count": len(sequence),
                "sequence_uuid": sequence_uuid,
                
…[4329 chars omitted]…
mapillary_tools/upload.py:562:    yield from image_uploader.upload_images(image_metadatas)
mapillary_tools/uploader.py:541:    def upload_images(
tests/unit/test_uploader.py:27:def test_upload_images(setup_unittest_data: py.path.local, setup_upload: py.path.local):
mapillary_tools/constants.py:150:UPLOAD_CACHE_DIR: str = os.getenv(
mapillary_tools/constants.py:151:    _ENV_PREFIX + "UPLOAD_CACHE_DIR",

class Uploader:
    def __init__(
        self,
        upload_options: UploadOptions,
        user_session: requests.Session | None = None,
        emitter: EventEmitter | None = None,
    ):
        self.upload_options = upload_options
        self.user_session = user_session
        if emitter is None:
            # An empty event emitter that does nothing
            self.emitter = EventEmitter()
        else:
            self.emitter = emitter

    def upload_stream(
        self,
        fp: T.IO[bytes],
        session_key: str | None = None,
        progress: dict[str, T.Any] | None = None,
    ) -> str:
        if progress is None:
         
…[7954 chars omitted]…

    try:
        upload_options = uploader.UploadOptions(
            user_items,
            dry_run=dry_run,
            nofinish=nofinish,
            noresume=noresume,
            num_upload_workers=num_upload_workers,
        )
    except ValueError as ex:
        raise exceptions.MapillaryBadParameterError(str(ex)) from ex

    mly_uploader = uploader.Uploader(upload_options, emitter=emitter)

    results = list(
        uploader.ZipUploader.zip_images_and_upload(
            mly_uploader,
            [
                description.DescriptionJSONSerializer.from_desc(T.cast(T.Any, desc))
                for desc in descs
            ],
        )
    )
    assert len(results) == 2
    actual_descs = sum(extract_all_uploaded_descs(Path(setup_upload)), [])
    assert 2 == len(actual_descs)


def test_upload_zip(
    setup_unittest_data: py.path.local, setup_upload: py.path.local, emitter=None
):
    test_exif = setup_unittest_data.join("test_exif.jpg")
    
…[5754 chars omitted]…
s(mly_uploader, video_metadatas)

    # Upload zip files
    zip_paths = utils.find_zipfiles(import_paths, skip_subfolders=skip_subfolders)
    yield from uploader.ZipUploader.upload_zipfiles(mly_uploader, zip_paths)


def _normalize_import_paths(import_path: Path | T.Sequence[Path]) -> list[Path]:
    import_paths: list[Path]

    if isinstance(import_path, Path):
        import_paths = [import_path]

        self, image_metadata: types.ImageMetadata, image_progress: dict[str, T.Any]
    ) -> str:
        image_bytes = self.dump_image_bytes(image_metadata)

        uploader = Uploader(self.upload_options, user_session=self.user_session)

        session_key = uploader._gen_session_key(io.BytesIO(image_bytes), image_progress)

        file_handle = self._get_cached_file_handle(session_key)

        if file_handle is None:
            # image_progress will be updated during uploading
            file_handle = uploader.u
…[340 chars omitted]…
  try:
            edit = exif_write.ExifEdit(metadata.filename)
        except struct.error as ex:
            raise ExifError(f"Failed to load EXIF: {ex}", metadata.filename) from ex

        # The cast is to fix the type checker error
        edit.add_image_description(
            T.cast(
                T.Dict, desc_file_to_exif(DescriptionJSONSerializer.as_desc(metadata))
            )
        )

                Path(constants.UPLOAD_CACHE_DIR).joinpath(version)
                .joinpath(api_v4.MAPILLARY_CLIENT_TOKEN.replace("|", "_"))
                .joinpath(user_items.get("MAPSettingsUserKey", user_items["user_upload_token"]))
            )
            cache_path_dir.mkdir(parents=True, exist_ok=True)
…[502 chars omitted]…
_expired()

        return cache

    def _get_cached_file_handle(self, key: str) -> str | None:
        if self.cache is None:
            return None

        if _is_uuid(key):
            return None

        return self.cache.get(key)

    def _set_file_handle_cache(self, key: str, value: str) -> None:
        if self.cache is None:
            return

        if _is_uuid(key):
            return

            raise ExifError(
                f"Failed to dump EXIF bytes: {ex}", metadata.filename
            ) from ex

    @classmethod
    def _maybe_create_persistent_cache_instance(
        cls, user_items: config.UserItem, upload_options: UploadOptions
    ) -> history.PersistentCache | None:
        if upload_options.upload_cache_path is not None:
            cache_path = upload_options.upload_cache_path
            cache_path.parent.mkdir(parents=True, exist_ok=True)
        else:
     
…[232 chars omitted]…
ts.UPLOAD_CACHE_DIR).joinpath(version)
                .joinpath(api_v4.MAPILLARY_CLIENT_TOKEN.replace("|", "_"))
                .joinpath(user_items.get("MAPSettingsUserKey", user_items["user_upload_token"]))
 mapillary_tools/history.py  |  5 ++++
 mapillary_tools/uploader.py | 69 +++++++++++++++++++++++----------------------
 2 files changed, 40 insertions(+), 34 deletions(-)

no tests ran in 0.02s

            file_handle = self._get_cached_file_handle(session_key)
            if file_handle is None:
                file_handle = uploader.upload_stream(
                    io.BytesIO(image_bytes), session_key=session_key, progress=image_progress
                )
                self._set_file_handle_cache(session_key, file_handle)
            return file_handle

    @classmethod
    def dump_
…[801 chars omitted]…
ions: UploadOptions
    ) -> history.PersistentCache | None:
        if upload_options.upload_cache_path is not None:
            cache_path = upload_options.upload_cache_path
            cache_path.parent.mkdir(parents=True, exist_ok=True)
        else:
            if not constants.UPLOAD_CACHE_DIR:
                LOG.debug("Upload cache directory is set empty, skipping caching upload file handles")

     def __post_init__(self):
         if self.num_upload_workers <= 0:
@@ -537,6 +538,7 @@ class ImageSequenceUploader:
     def __init__(self, upload_options: UploadOptions, emitter: EventEmitter):
         self.upload_options = upload_options
         self.emitter = emitter
+        self.cached_image_uploader = SingleImageUploader(upload_options)
 
     def upload_images(
         self, image_metadatas: T.Sequence[types.ImageMetadata]
@@ -691,6 +693,11 @
…[4186 chars omitted]…
tems["user_upload_token"]))
             )
-        )
-        cache_path_dir.mkdir(parents=True, exist_ok=True)
-        cache_path = cache_path_dir.joinpath("cached_file_handles")
+            cache_path_dir.mkdir(parents=True, exist_ok=True)
+            cache_path = cache_path_dir.joinpath("cached_file_handles")
 
         # Sanitize sensitive segments for logging
         sanitized_cache_path = (

                # Create a new mutatble progress to keep the sequence_progress immutable
                image_progress = {
                    **sequence_progress,
                    "import_path": str(image_metadata.filename),
                }

                # image_progress will be updated during uploading
                file_handle = single_image_uploader.upload(
                    image_metadata, 
…[123 chars omitted]…
ss["chunk_size"] = image_metadata.filesize

                # Main thread will handle the interruption
                if upload_interrupted.is_set():
                    break

                with lock:
                    self.emitter.emit("upload_progress", image_progress)

                indexed_file_handles.append((idx, file_handle))

 M mapillary_tools/history.py
 M mapillary_tools/uploader.py

script
tests
./.git/COMMIT_EDITMSG
./.git/config
./.git/description
./.git/HEAD
./.git/hooks/applypatch-msg.sample
./.git/hooks/commit-msg.sample
./.git/hooks/fsmonitor-watchman.sample
./.git/hooks/post-update.sample
./.git/hooks/pre-applypatch.sample
./.git/hooks/pre-commit.sample
./.git/hooks/pre-merge-commit.sample
./.git/hooks/pre-push.sample
./.git/hooks/pre-rebase.sample
./.git/hooks/pre-receive.sample
./.git/hooks/prepare-commit-msg.sample
./.git/hooks/push-to-checkout.sample
./.git/hooks/sendemail-validate.sample
./.git/hooks/update.sample
./.git/index
./.git/info/exclude
./.git/logs/HEAD
./.github/workflows/codeql-analysis.yml
./.github/workflows/python-package.yml
./.github/workflows/release.yml
./.gitignore

…[3286 chars omitted]…
ts/unit/test_persistent_cache.py:190:    cache = PersistentCache(cache_file)
./tests/unit/test_persistent_cache.py:203:    cache = PersistentCache(cache_file)
./tests/unit/test_persistent_cache.py:228:    cache = PersistentCache(cache_file)
./tests/unit/test_persistent_cache.py:272:    cache = PersistentCache(cache_file)
./tests/unit/test_persistent_cache.py:286:    cache = PersistentCache(cache_file)

import threading
import time
import typing as T
import uuid
import zipfile
from contextlib import contextmanager
from pathlib import Path

if sys.version_info >= (3, 11):
    from typing import Required
else:
    from typing_extensions import Required

import requests

from . import (
    api_v4,
    config,
    constants,
    exif_write,
    geo,
    history,
    telemetry,
    types,
    upload_api_v4,
    utils,
    VERSION,
)
from .camm import camm_builder, camm_parser
from .gpmf import gpmf_parser
from .mp4 import simple_mp4_builder
from .serializer.description import (
    desc_file_to_exif,
    DescriptionJSONSerializer,
    validate_image_desc,
)


LOG = logging.getLogger(__name__)


@dataclasses.d
…[8630 chars omitted]…
, dict):
            LOG.warning(f"Invalid cache value format: {payload}")
            return {}

        return payload
tests/integration/test_process_and_upload.py:152:def test_process_and_upload_images_only(
tests/unit/test_uploader.py:27:def test_upload_images(setup_unittest_data: py.path.local, setup_upload: py.path.local):
tests/unit/test_uploader.py:68:def test_upload_images_multiple_sequences(

        except FileNotFoundError:
            pass
        try:
            yield wip_path

            with wip_path.open("rb") as fp:
                upload_md5sum = utils.md5sum_fp(fp).hexdigest()

            done_path = wip_path.parent.joinpath(
                _suffix_session_key(upload_md5sum, api_v4.ClusterFileType.ZIP)
            )

            try:
                os.remove(done_path)
            except FileNotFoundError:
                pass
            wip_path.rename(done_path)
        finally:
         
…[8629 chars omitted]…
istentCache(str(cache_path.resolve()))
830:        cache.clear_expired()
832:        return cache
834:    def _get_cached_file_handle(self, key: str) -> str | None:
835:        if self.cache is None:
841:        return self.cache.get(key)
843:    def _set_file_handle_cache(self, key: str, value: str) -> None:
844:        if self.cache is None:
850:        self.cache.set(key, value)
853:class Uploader:

        # Push all images into the queue
        image_queue: queue.Queue[tuple[int, types.ImageMetadata]] = queue.Queue()
        for idx, image_metadata in enumerate(sequence):
            image_queue.put((idx, image_metadata))

        upload_interrupted = threading.Event()

        with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor:
            futures = [
                executor.submit(
                    self._upload_images_from_queue,
                    image_queue,
       
…[8629 chars omitted]…
roup_and_sort_images(metadatas)
        os.makedirs(zip_dir, exist_ok=True)

        for sequence_uuid, sequence in sequences.items():
            _validate_metadatas(sequence)
            # For atomicity we write into a WIP file and then rename to the final file
            wip_zip_filename = zip_dir.joinpath(
                f".mly_zip_{uuid.uuid4()}_{sequence_uuid}_{os.getpid()}_{int(time.time())}"

            self.upload_options.user_items, upload_options
        )

    def upload(
        self, image_metadata: types.ImageMetadata, image_progress: dict[str, T.Any]
    ) -> str:
        image_bytes = self.dump_image_bytes(image_metadata)

        uploader = Uploader(self.upload_options, user_session=self.user_session)

        session_key = uploader._gen_session_key(io.BytesIO(image_bytes), image_progress)

        file_handle = self._get_cached_file_handle(session_key)

        if file_handle is None:
            # image_progress wil
…[7828 chars omitted]…
me, should_exist
):
    """Test expiration with different timing combinations."""
    cache_file = os.path.join(
        tmpdir, f"cache_param_exp_{dbm_backend}_{expire_time}_{sleep_time}"
    )
    cache = PersistentCache(cache_file)

    key = f"key_expires_in_{expire_time}_sleeps_{sleep_time}"
    cache.set(key, "test_value", expires_in=expire_time)

    time.sleep(sleep_time)

    if should_exist:

        if emitter is None:
            # An empty event emitter that does nothing
            self.emitter = EventEmitter()
        else:
            self.emitter = emitter

    def upload_stream(
        self,
        fp: T.IO[bytes],
        session_key: str | None = None,
        progress: dict[str, T.Any] | None = None,
    ) -> str:
        if progress is None:
            progress = {}

        if session_key is None:
            session_key = self._gen_session_key(fp, progress)

        fp.seek(0, io.SEEK_END)
        entity_size = fp.tell()

        progress["entity
…[6276 chars omitted]…
TE: Sequences themselves are uploaded sequentially, not in parallel.
MAX_IMAGE_UPLOAD_WORKERS: int = int(
    os.getenv(_ENV_PREFIX + "MAX_IMAGE_UPLOAD_WORKERS", 4)
)
# The chunk size in MB (see chunked transfer encoding https://en.wikipedia.org/wiki/Chunked_transfer_encoding)
# for uploading data to MLY upload service.
mapillary_tools/upload.py:559:    image_uploader = uploader.ImageSequenceUploader(

    metadatas: T.Iterable[_M], paths: T.Iterable[Path]
) -> list[_M]:
    resolved_image_paths = set(p.resolve() for p in paths)
    return [d for d in metadatas if d.filename.resolve() in resolved_image_paths]


def _gen_upload_everything(
    mly_uploader: uploader.Uploader,
    metadatas: T.Sequence[types.Metadata],
    import_paths: T.Sequence[Path],
    skip_subfolders: bool,
):
    # Upload images
    image_metadatas = _find_metadata_with_filename_existed_in(
        (m for m in metadatas if isinstance(m, types.ImageM
…[3310 chars omitted]…
ent_cache.py:290:    assert cache._is_expired({"expires_at": time.time() + 10}) is False
tests/unit/test_persistent_cache.py:291:    assert cache._is_expired({}) is False
tests/unit/test_persistent_cache.py:292:    assert cache._is_expired({"expires_at": "not a number"}) is False
tests/unit/test_persistent_cache.py:293:    assert cache._is_expired({"expires_at": None}) is False
197:class EventEmitter:

+        """Return the keys currently stored in the cache."""
+        with self._lock:
+            with dbm.open(self._file, flag="c") as db:
+                return [
+                    key.decode("utf-8") if isinstance(key, bytes) else key
+                    for key in db.keys()
+                ]
+
     def clear_expired(self) -> list[str]:
         s = time.perf_c
…[3953 chars omitted]…
cache_path
+            cache_path.parent.mkdir(parents=True, exist_ok=True)
+            cache = history.PersistentCache(str(cache_path.resolve()))
+            cache.clear_expired()
+            return cache
+
         # Different python/CLI versions use different cache (dbm) formats.
         # Separate them to avoid conflicts
         py_version_parts = [str(part) for part in sys.version_info[:3]]

/repo/mapillary_tools/commands/zip.py
/repo/mapillary_tools/commands/__init__.py
/repo/mapillary_tools/commands/__main__.py
/repo/mapillary_tools/config.py
/repo/mapillary_tools/constants.py
/repo/mapillary_tools/exceptions.py
/repo/mapillary_tools/exiftool_read.py
/repo/mapillary_tools/exiftool_read_video.py
/repo/mapillary_tools/exiftool_runner.py
/repo/mapill
…[1167 chars omitted]…
ctors/native.py
/repo/mapillary_tools/geotag/__init__.py
/repo/mapillary_tools/gpmf/gpmf_gps_filter.py
/repo/mapillary_tools/gpmf/gpmf_parser.py
/repo/mapillary_tools/gpmf/gps_filter.py
/repo/mapillary_tools/history.py
---
CHANGELOG.md
CODE_OF_CONDUCT.md
CONTRIBUTING.md
LICENSE
README.md
README_osx_package.txt
docs
mapillary_tools
mapillary_tools.egg-info
pyinstaller
pyproject.toml
schema
script
tests

exiftool_read_video.py
exiftool_runner.py
ffmpeg.py
geo.py
geotag
gpmf
history.py
http.py
ipc.py
mp4
process_geotag_properties.py
process_sequence_properties.py
sample_video.py
serializer
telemetry.py
types.py
upload.py
upload_api_v4.py
uploader.py
utils.py
---
/repo/mapillary_tools/upload.py:559:    image_uploader = uploader.ImageSequenceUploader(
/repo/mapillary_tools/uploader.py:536:class ImageSequenceUploader:

import threading
import time
import typing as T
import uuid
import zipfile
from contextlib import contextmanager
from pathlib import Path

if sys.version_info >= (3, 11):
    from typing import Required
else:
    from typing_extensions import Required

import requests

from . import (
    api_v4,
    config,
    constants,
    exif_write,
    geo,
    history,
    telemetry,
    types,
    upload_api_v4,
    utils,
    VERSION,
)
from .camm import camm_builder, camm_parser
from .gpmf import gpmf_parser
from .mp4 import simple_mp4_builder
from .serializer.description import (
    desc_file_to_exif,
    DescriptionJSONSerializer,
    validate_image_desc,
)


LOG = logging.getLogger(__name__)


@dataclasses.d
…[8629 chars omitted]…
= Uploader(self.upload_options, user_session=self.user_session)

        session_key = uploader._gen_session_key(io.BytesIO(image_bytes), image_progress)

        file_handle = self._get_cached_file_handle(session_key)

        if file_handle is None:
            # image_progress will be updated during uploading
            file_handle = uploader.upload_stream(
                io.BytesIO(image_bytes),

        except struct.error as ex:
            raise ExifError(f"Failed to load EXIF: {ex}", metadata.filename) from ex

        # The cast is to fix the type checker error
        edit.add_image_description(
            T.cast(
                T.Dict, desc_file_to_exif(DescriptionJSONSerializer.as_desc(metadata))
            )
        )

        try:
            return edit.dump_image_bytes()
        except struct.error as ex:
            raise ExifError(
                f"Failed to dump EXIF bytes: 
…[6860 chars omitted]…
retries in the caller
            retries += 1
            if _is_immediate_retriable_exception(ex):
                sleep_for = 0
            else:
                sleep_for = min(2**retries, 16)
            LOG.info(
                f"Retrying in {sleep_for} seconds ({retries}/{constants.MAX_UPLOAD_RETRIES})"
            )
            if sleep_for:
                time.sleep(sleep_for)
        else:

            sequence_md5sum = types.update_sequence_md5sum(sequence)

            sequence_progress: SequenceProgress = {
                "sequence_idx": sequence_idx,
                "total_sequence_count": len(sequences),
                "sequence_image_count": len(sequence),
                "sequence_uuid": sequence_uuid,
                
…[5434 chars omitted]…
  summary: JSONDict,
    metadatas: T.Sequence[types.Metadata] | None = None,
) -> None:
    if not constants.MAPILLARY_UPLOAD_HISTORY_PATH:
        return
    path = history_desc_path(md5sum)
    LOG.debug("Writing upload history: %s", path)
    path.resolve().parent.mkdir(parents=True, exist_ok=True)
    history: dict[str, T.Any] = {"params": params, "summary": summary}
    if metadatas is not None:

100-        s = time.perf_counter()
101-
102-        with self._lock:
103-            with dbm.open(self._file, flag="c") as db:
104-                value: bytes | None = db.get(key)
105-
106-        if value is None:
107-            return None
108-
109-        payload = self._decode(value)
110-
111-        if self._is_expired(payload):
112-            return None
113-
114-        file_handle = payload.get("file_handle")
115-
116-        LOG.debug(
117-  
…[797 chars omitted]…
lf) -> list[str]:
141-        s = time.perf_counter()
142-
143-        expired_keys: list[str] = []
144-
145-        with self._lock:
146-            with dbm.open(self._file, flag="c") as db:
147-                if hasattr(db, "items"):
148-                    items: T.Iterable[tuple[str | bytes, bytes]] = db.items()
===
150:UPLOAD_CACHE_DIR: str = os.getenv(
151:    _ENV_PREFIX + "UPLOAD_CACHE_DIR",

                f"Cleared {len(expired_keys)} expired entries from the cache ({(time.perf_counter() - s) * 1000:.0f} ms)"
            )

        return expired_keys

    def _is_expired(self, payload: JSONDict) -> bool:
        expires_at = payload.get("expires_at")
        if isinstance(expires_at, (int, float)):
            return expires_at is None or expires_at <= time.time()
        return False

    def _decode(self, value: bytes) -
…[1691 chars omitted]…
ersistent_cache.py:48:    cache = PersistentCache(cache_file)
/repo/tests/unit/test_persistent_cache.py:83:    cache = PersistentCache(cache_file)
/repo/tests/unit/test_persistent_cache.py:100:    cache = PersistentCache(cache_file)
/repo/tests/unit/test_persistent_cache.py:123:    cache = PersistentCache(cache_file)
/repo/tests/unit/test_persistent_cache.py:149:    cache = PersistentCache(cache_file)

@pytest.mark.parametrize("dbm_backend", DBM_BACKENDS)
def test_basic_operations_with_backend(tmpdir, dbm_backend):
    """Test basic operations with different DBM backends.

    Note: This is a demonstration of pytest's parametrize feature.
    The actual PersistentCache class might not support specifying backends.
    """
    cache_file = os.path.join(tmpdir, dbm_backend)
    # Here you would use the backend if the cache implementation supported it
    cache = PersistentCache(cache_file)

    # Perform basic operations
    cache.set("test_key", "test_value")
    assert cache.get("test_key") == "test_value"

    # Add specific test logic for different backends if needed
  
…[8628 chars omitted]…
amm_parser.py
test_config.py
test_description.py
test_exceptions.py
test_exifedit.py
test_exifread.py
test_ffmpeg.py
test_geo.py
test_gpmf_parser.py
test_gps_filter.py
test_io_utils.py
test_mp4_sample_parser.py
test_persistent_cache.py
test_sample_video.py
test_sequence_processing.py
test_simple_mp4_builder.py
test_simple_mp4_parser.py
test_types.py
test_upload_api_v4.py
test_uploader.py
test_utils.py

IMPORT_PATH = "tests/unit/data"


@pytest.fixture
def setup_unittest_data(tmpdir: py.path.local):
    data_path = tmpdir.mkdir("data")
    source = py.path.local(IMPORT_PATH)
    source.copy(data_path)
    yield data_path
    if tmpdir.check():
        tmpdir.remove(ignore_errors=True)


def test_upload_images(setup_unittest_data: py.path.local, setup_upload: py.path.local):
    mly_uploader = uploader.Uploader(
        uploader.UploadOptions(
            {"user_upload_token": "YOUR_USER_ACCESS_TOKEN"}, dry_run=True
        )
    )
    test_exif = setup_unittest_data.join("test_exif.jpg")
    descs: T.List[description.DescriptionOrError] 
…[5134 chars omitted]…
 == len(actual_descs)


def test_upload_blackvue(tmpdir: py.path.local, setup_upload: py.path.local):
    mly_uploader = uploader.Uploader(
        uploader.UploadOptions(
            {
                "user_upload_token": "YOUR_USER_ACCESS_TOKEN",
                # will call the API for real
                # "MAPOrganizationKey": "3011753992432185",
            },
            dry_run=True,
        )

    )
    assert cluster_id == "0"
    assert setup_upload.join("this_is_a_blackvue.mp4").exists()
    with open(setup_upload.join("this_is_a_blackvue.mp4"), "rb") as fp:
        assert fp.read() == b"this is a fake video"


def test_upload_zip_with_emitter(
    setup_unittest_data: py.path.local, setup_upload: py.path.local
):
    emitter = uploader.EventEmitter()

    stats = {}

    @emitter.on("upload_start")
    def _upload_start(payload):
        assert payload["entit
…[3990 chars omitted]…
      return False
    return True


IS_FFMPEG_INSTALLED = _ffmpeg_installed()


def _exiftool_installed():
    try:
        subprocess.run(
            [EXIFTOOL_EXECUTABLE, "-ver"],
            stderr=subprocess.PIPE,
            stdout=subprocess.PIPE,
            check=True,
        )
    except FileNotFoundError:
        return False
    return True


IS_EXIFTOOL_INSTALLED = _exiftool_installed()

533-                pass
534-
535-
536-class ImageSequenceUploader:
537-    def __init__(self, upload_options: UploadOptions, emitter: EventEmitter):
538-        self.upload_options = upload_options
539-        self.emitter = emitter
540-
541-    def upload_images(
--
618:                manifest_fp, session_key=f"{_prefixed_uuid4()}.json"
619-            )
620-
621-    def _upload_images_parallel(
622-        self,
623-        sequence: T.Sequence[types.ImageMetadata],
624-        sequence
…[1391 chars omitted]…
(f"Failed to load EXIF: {ex}", metadata.filename) from ex
--
838:        if _is_uuid(key):
839-            return None
840-
841-        return self.cache.get(key)
842-
843-    def _set_file_handle_cache(self, key: str, value: str) -> None:
844-        if self.cache is None:
845-            return
846-
847:        if _is_uuid(key):
848-            return
849-
850-        self.cache.set(key, value)
851-

1092-
1093-        return session_key
1094-
1095-
1096-def _validate_metadatas(metadatas: T.Sequence[types.ImageMetadata]):
1097-    for metadata in metadatas:
1098-        validate_image_desc(DescriptionJSONSerializer.as_desc(metadata))
1099-        if not me
…[412 chars omitted]…
56-    key = f"mly_tools_{key}{_SUFFIX_MAP[filetype]}"
1157-
1158-    assert _is_uuid(key) is is_uuid_before
1159-
1160-    return key
1161-
--
1163:def _prefixed_uuid4():
1164-    prefixed = f"uuid_{uuid.uuid4().hex}"
1165-    assert _is_uuid(prefixed)
1166-    return prefixed
1167-
1168-
1169:def _is_uuid(key: str) -> bool:
1170-    return key.startswith("uuid_") or key.startswith("mly_tools_uuid_")

            )
        except Exception:
            LOG.warning(f"Error from logging action {action}", exc_info=True)


_M = T.TypeVar("_M", bound=types.Metadata)


def _find_metadata_with_filename_existed_in(
    metadatas: T.Iterable[_M], paths: T.Iterable[Path]
) -> list[_M]:
    resolved_image_paths = set(p.resolve() for p in paths)
    return [d for d in metadatas if d.fil
…[1793 chars omitted]…
eption:
    """
    Wrap the exception, or re-raise if it is a fatal error (i.e. there is no point to continue)
    """
/repo/mapillary_tools/uploader.py:474:                zipf.writestr(zipinfo, SingleImageUploader.dump_image_bytes(metadata))
/repo/mapillary_tools/uploader.py:691:            single_image_uploader = SingleImageUploader(
/repo/mapillary_tools/uploader.py:734:class SingleImageUploader:

    def upload(
        self, image_metadata: types.ImageMetadata, image_progress: dict[str, T.Any]
    ) -> str:
        image_bytes = self.dump_image_bytes(image_metadata)

        uploader = Uploader(self.upload_options, user_session=self.user_session)

        session_key = uploader._gen_session_key(io.BytesIO(image_bytes), image_progress)

...
            cache_path = cache_path_dir.joinpath("cached_file_handles")

            # Sanitize sensitive segments for logging
       
…[440 chars omitted]…
 def _get_cached_file_handle(self, key: str) -> str | None:
        if self.cache is None:
            return None

        if _is_uuid(key):
            return None

        return self.cache.get(key)

    def _set_file_handle_cache(self, key: str, value: str) -> None:
        if self.cache is None:
            return

        if _is_uuid(key):
            return

        self.cache.set(key, value)

185-        transient_error_ratio: float = 0.0,
186-        **kwargs,
187-    ):
188-        super().__init__(*args, **kwargs)
189-        if upload_path is None:
190-            upload_path = Path(tempfile.gettempdir()).joinpath(
191-                "mapillary_public_uploads"
192-            )
193-        self._upload_path = upload_path
194-        self._transient_error_ratio = transient_error_ratio
195-
196-    @override
197-    def upload_shifted_chunks(
198-        self
…[529 chars omitted]…
with filename.open("ab") as fp:
212-            for chunk in shifted_chunks:
213-                self._randomly_raise_transient_error()
27:class UploadService:
39:    def fetch_offset(self) -> int:
113:    def upload_byte_stream(
128:    def upload_chunks(
141:    def upload_shifted_chunks(
197:    def upload_shifted_chunks(
227:    def fetch_offset(self) -> int:
237:    def upload_path(self) -> Path:

    @override
    def fetch_offset(self) -> int:
        self._randomly_raise_transient_error()
        filename = self._upload_path.joinpath(self.session_key)
        if not filename.exists():
            return 0
        with open(filename, "rb") as fp:
            fp.seek(0, io.SEEK_END)
            return fp.tell()

    @property
    def upload_path
…[200 chars omitted]…
is for testing purposes only.
        """
        if random.random() <= self._transient_error_ratio:
            raise requests.ConnectionError(
                f"[TEST ONLY]: Transient error with ratio {self._transient_error_ratio}"
            )
40:def create_user_session(user_access_token: str) -> requests.Session:
149:def finish_upload(
155:    url = f"{MAPILLARY_GRAPH_API_ENDPOINT}/finish_upload"

    """

    def __init__(
        self,
        upload_options: UploadOptions,
        user_session: requests.Session | None = None,
        cache: history.PersistentCache | None = None,
    ):
        self.upload_options = upload_options
        self.user_session = user_session
        if cache is None:
            self.cache = self._maybe_create_persistent_cache_instance(upload_options)
        else:
            self.c
…[1036 chars omitted]…
         cache_path_dir = (
                Path(constants.UPLOAD_CACHE_DIR)
                .joinpath(version)
                .joinpath(api_v4.MAPILLARY_CLIENT_TOKEN.replace("|", "_"))
                .joinpath(
                    upload_options.user_items.get(
                        "MAPSettingsUserKey",
                        upload_options.user_items["user_upload_token"],
                    )

                 ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
AttributeError: 'CachedUploader' object has no attribute '_maybe_create_persistent_cache_instance'

            LOG.debug("Dry-run mode enabled, skipping caching upload file handles")
            return None

        if upload_options.upload_cache_path is not None:
            cache_path = upload_options.upload_cache_path
            cache_path.parent.mkdir(parents=True, exist_ok=True)
        else:
            if not constants.UPLOAD_CACHE_DIR:
                LOG.debug(
                    "Upload cache directory is set empty, skipping caching upload file handles"
                )
                ret
…[1338 chars omitted]…
 return cache


class SingleImageUploader(CachedUploader):
    def upload(
        self, image_metadata: types.ImageMetadata, image_progress: dict[str, T.Any]
    ) -> str:
        image_bytes = self.dump_image_bytes(image_metadata)

        uploader = Uploader(self.upload_options, user_session=self.user_session)

        session_key = uploader._gen_session_key(io.BytesIO(image_bytes), image_progress)

743:class CachedUploader:
820:class SingleImageUploader(CachedUploader):
864:def _validate_metadatas(metadatas: T.Sequence[types.ImageMetadata]):
871:def _is_immediate_retriable_exception(ex: BaseException) -> bool:
887:def _is_retriable_exception(ex: BaseException) -> bool:
919:def _suffix_session_key(
931:def _prefixed_uuid4():
937:def _is_uuid(key: str) -> bool:

734:class SingleImageUploader:
853:class Uploader:
1096:def _validate_metadatas(metadatas: T.Sequence[types.ImageMetadata]):
1103:def _is_immediate_retriable_exception(ex: BaseException) -> bool:
1119:def _is_retriable_exception(ex: BaseException) -> bool:
1151:def _suffix_session_key(
1163:def _prefixed_uuid4():
1169:def _is_uuid(key: str) -> bool:

        self,
        upload_options: UploadOptions,
        user_session: requests.Session | None = None,
        cache: history.PersistentCache | None = None,
    ):
        self.upload_options = upload_options
        self.user_session = user_session
        if cache is None:
            self.cache = self._maybe_create_persistent_cache_instance(upload_options)
        else:
            self.cache = cache

    @classmethod
    def
…[3198 chars omitted]…
it.add_image_description(
            T.cast(
                T.Dict, desc_file_to_exif(DescriptionJSONSerializer.as_desc(metadata))
            )
        )

        try:
            return edit.dump_image_bytes()
        except struct.error as ex:
            raise ExifError(
                f"Failed to dump EXIF bytes: {ex}", metadata.filename
            ) from ex

class Uploader:
    def __init__(

    def fetch_offset(self) -> int:
        url = f"{MAPILLARY_UPLOAD_ENDPOINT}/{self.session_key}"

        resp = self.user_session.get(url, timeout=REQUESTS_TIMEOUT)
        resp.raise_for_status()

        data = jsonify_response(resp)
        try:
            return data["offset"]
        except KeyError:
            raise HTTPContentError("Offset not found in the response", resp)

    @classmethod
    def chunkize_byte_stream(
        cls, stream: T.IO[bytes], chunk_size: int
    ) -> T.Generator[bytes, None, None]:
        """
        Chunkize a byte stream into chunks of the specified size.

        >>> list(UploadService.chunkize_byte_stream(io.BytesIO(b"foo"), 1))
        [b'f', b'o', b'o']

LOG = logging.getLogger(__name__)


class ExifEdit:
    _filename_or_bytes: str | bytes

    def __init__(self, filename_or_bytes: Path | bytes) -> None:
        """Initialize the object"""
        if isinstance(filename_or_bytes, Path):
            # make sure filename is resolved to avoid to be interpretted as bytes in piexif
            # see https://github.com/hMatoba/Piexif/issues/124
            self._filename_or_bytes = str(filename_or_bytes.resolve())
        else:
            self._filename_or_bytes = filename_or_bytes
        self._ef: dict = piexif.load(self._filename_or_bytes)

    @staticmethod
    def decimal_to_dms(
        value: float, precision: int
    ) -> tuple[tuple[float, int], tuple[fl
…[1840 chars omitted]…
["Exif"]:
                    del self._ef["Exif"][piexif.ExifIFD.OffsetTimeOriginal]
        else:
            if piexif.ExifIFD.OffsetTimeOriginal in self._ef["Exif"]:
                del self._ef["Exif"][piexif.ExifIFD.OffsetTimeOriginal]

    def add_gps_datetime(self, dt: datetime.datetime) -> None:
        """Add GPSDateStamp and GPSTimeStamp."""
        dt = dt.astimezone(datetime.timezone.utc)

190-
191-    # deduplicate and sort metadatas per uuid
192-    sorted_sequences_by_uuid = {}
193-    for sequence_uuid, sequence in sequences_by_uuid.items():
194-        dedups = {metadata.filename.resolve(): metadata for metadata in sequence}
195-        sorted_sequences_by_uuid[sequence_uuid] = sorted(
196-            dedups.values(),
19
…[405 chars omitted]…
data.md5sum.encode("utf-8"))
208-    return md5.hexdigest()
202:def update_sequence_md5sum(sequence: T.Iterable[ImageMetadata]) -> str:
203-    md5 = hashlib.md5()
204-    for metadata in sequence:
205-        metadata.update_md5sum()
206-        assert isinstance(metadata.md5sum, str), "md5sum should be calculated"
207-        md5.update(metadata.md5sum.encode("utf-8"))
208-    return md5.hexdigest()

205:        metadata.update_md5sum()
206:        assert isinstance(metadata.md5sum, str), "md5sum should be calculated"
207:        md5.update(metadata.md5sum.encode("utf-8"))

    @classmethod
    @abc.abstractmethod
    def deserialize(cls, data: bytes) -> list[Metadata]:
        raise NotImplem
…[1562 chars omitted]…
tadata] = []

    for metadata in metadatas:
        if isinstance(metadata, ErrorMetadata):
            bad.append(metadata)
        else:
            good.append(metadata)

    return good, bad


def describe_error_metadata(
    exc: Exception, filename: Path, filetype: FileType
) -> ErrorMetadata:
    return ErrorMetadata(filename=filename, filetype=filetype, error=exc)


def group_and_sort_images(

    MAPGPSAccuracyMeters: float | None = None
    MAPCameraUUID: str | None = None
    MAPOrientation: int | None = None
    MAPMetaTags: dict | None = None
    MAPFilename: str | None = None

    def update_md5sum(self, image_data: T.BinaryIO | None = None) -> None:
        if self.md5sum is None:
            if image_data is None:
                with self.filename.open("rb") as fp:
                    self.md5sum = utils.md5sum_fp(fp).hexdigest()
            else:
                self.md5s
…[539 chars omitted]…
images(image_metadatas)
mapillary_tools/uploader.py:354:        sequences = types.group_and_sort_images(metadatas)
mapillary_tools/uploader.py:371:        sequences = types.group_and_sort_images(image_metadatas)
mapillary_tools/uploader.py:461:        sequence_groups = types.group_and_sort_images(sequence)
mapillary_tools/uploader.py:549:        sequences = types.group_and_sort_images(image_metadatas)

+
+        with self._lock:
+            with dbm.open(self._file, flag="c") as db:
+                for key in db.keys():
+                    if isinstance(key, bytes):
+                        key = key.decode("utf-8", errors="replace")
+                    keys.append(T.cast(str, key))
+
+        return keys
+
     def clear_expired(self) -> list[str]:
         s = time.perf_counter()
 
diff --git a/mapillary_tools/uploader.py b/mapillary_tools/up
…[7700 chars omitted]…
ile_handle(self, key: str) -> str | None:
-        if self.cache is None:
-            return None
-
-        if _is_uuid(key):
-            return None
-
-        return self.cache.get(key)
-
-    def _set_file_handle_cache(self, key: str, value: str) -> None:
-        if self.cache is None:
-            return
-
-        if _is_uuid(key):
-            return
-
-        self.cache.set(key, value)
-
-