How to add sequence_delete_interval sub-command to delete part of a sequence?

[bruno0647]

Hi,

Request to add a new sequence_delete_interval subcommand.

Background
I have noticed two somewhat frequently recurring issues that case some of the images in a sequence to be "2bad":

1. The video stream is started before the GPS-position has stabilized.
2. The GPS-receiver is bumped, causing the gps position to go off true significantly.

With the Rioch ThetaX, the first 1~2 GPS positions are usually always inaccurate; in most cases it just annoying as it isn
t visible on the map, however in some cases it adds quite a few phantom images. (as in cfJCbTLBPqgEAvY0W3GuwS it adds 20 phantom images. Likely mostly due to handling the to start the recording so Issue #2. Other handling mistakes also tend result in 12 GPS positions that are really inaccurate, and interpolation can lead to quite a few phantom images in the middle of a sequence (that are nearly identical).
With the Dashcam seems provide more consistent position data overall, but it is not uncommon that the first half of the first video file (after power-on) has really inaccurate position data.

Cleaning up the resulting faulty images in the Web-GUI is really tedious and time-consuming; especially as one ether has to wait a long time for the image delete to be completed, or try to continue with the next one and risk that completion message interrupts the ongoing request before confirmation and thus corrupts the Web-GUI state.

Attepted solution

Add a new sequence_delete_interval subcommand to mapillary_tools, to automate the removal of a interval in a sequence, using two image key form the Web-GUI to mark the interval to be removed.
(Attempting to Vibe-code this may not have been the best ida, but it is a starting point for further work - too late to do more today. Retrieving the keys to delete seems to workable, but the authentication for delete proved problematic gotr Gemini.)

mapillary_tools/commands/sequence_delete_interval.py

import requests
from .. import authenticate

GRAPH_URL = "https://graph.mapillary.com"


class Command:
    name = "sequence_delete_interval"
    help = "Delete a range of images in a sequence between two keyframes"

    def add_basic_arguments(self, parser):
        parser.add_argument("start_key", type=str, help="Starting image key/ID")
        parser.add_argument("end_key", type=str, help="Ending image key/ID")
        parser.add_argument(
            "--yes",
            "-y",
            action="store_true",
            help="Skip interactive deletion confirmation prompt",
        )

    def run(self, vars_args: dict):
        start_key = vars_args["start_key"]
        end_key = vars_args["end_key"]
        skip_confirm = vars_args.get("yes", False)

        # 1. Fetch user items dynamically via internal helper
        user_items = authenticate.fetch_user_items()
        
        user_token = None
        if user_items:
            # Handle list or dict return types from fetch_user_items
            item = user_items[0] if isinstance(user_items, list) and len(user_items) > 0 else user_items
            if isinstance(item, dict):
                user_token = (
                    item.get("user_upload_token")
                    or item.get("MAPGlobalUserAccessToken")
                    or item.get("user_access_token")
                )

        if not user_token:
            raise RuntimeError(
                "No saved Mapillary credentials found. "
                "Please run `mapillary_tools authenticate` first."
            )

        headers = {"Authorization": f"OAuth {user_token}"}

        print(f"Fetching sequence information for start image '{start_key}' and end image '{end_key}'...")

        # 2. Fetch sequence IDs for keyframes
        res_start = requests.get(
            f"{GRAPH_URL}/{start_key}",
            headers=headers,
            params={"fields": "id,sequence", "access_token": user_token},
        ).json()

        res_end = requests.get(
            f"{GRAPH_URL}/{end_key}",
            headers=headers,
            params={"fields": "id,sequence", "access_token": user_token},
        ).json()

        if "error" in res_start:
            raise RuntimeError(f"Start key error: {res_start['error']['message']}")
        if "error" in res_end:
            raise RuntimeError(f"End key error: {res_end['error']['message']}")

        seq_start = res_start.get("sequence")
        seq_end = res_end.get("sequence")

        if not seq_start or not seq_end:
            raise ValueError("One or both images do not belong to a valid sequence.")
        if seq_start != seq_end:
            raise ValueError(f"Images belong to different sequences ({seq_start} vs {seq_end}).")

        # 3. Retrieve sequence image list
        seq_res = requests.get(
            f"{GRAPH_URL}/image_ids",
            headers=headers,
            params={"sequence_id": seq_start, "access_token": user_token},
        ).json()

        if "error" in seq_res:
            raise RuntimeError(f"Sequence query error: {seq_res['error']['message']}")

        images_in_seq = seq_res.get("data", [])

        detailed = []
        for img in images_in_seq:
            img_id = str(img["id"])
            item = requests.get(
                f"{GRAPH_URL}/{img_id}",
                headers=headers,
                params={"fields": "id,captured_at", "access_token": user_token},
            ).json()
            if "id" in item:
                detailed.append(item)

        sorted_imgs = sorted(detailed, key=lambda x: x.get("captured_at", 0))
        ordered_ids = [str(x["id"]) for x in sorted_imgs]

        idx_start = ordered_ids.index(start_key)
        idx_end = ordered_ids.index(end_key)

        if idx_start <= idx_end:
            target_ids = ordered_ids[idx_start : idx_end + 1]
        else:
            target_ids = ordered_ids[idx_end : idx_start + 1][::-1]

        print(f"\nFound {len(target_ids)} image(s) to delete:")
        for idx, img_id in enumerate(target_ids, 1):
            print(f"  {idx}. {img_id}")

        if not skip_confirm:
            confirm = input(f"\nAre you sure you want to delete these {len(target_ids)} images? [y/N]: ")
            if confirm.lower() not in ("y", "yes"):
                print("Aborted.")
                return

        # 4. Perform Deletion
        for idx, img_id in enumerate(target_ids, 1):
            del_res = requests.delete(
                f"{GRAPH_URL}/{img_id}",
                headers=headers,
                params={"access_token": user_token},
            )
            if del_res.status_code == 200:
                print(f"[{idx}/{len(target_ids)}] Deleted: {img_id}")
            else:
                print(f"[{idx}/{len(target_ids)}] Failed to delete {img_id}: {del_res.text}")

Changes to mapillary_tools/commands/main.py

...
from . import (
    authenticate,
    process,
    process_and_upload,
    sample_video,
    upload,
     video_process,
     video_process_and_upload,
     zip,
+    sequence_delete_interval,
 )

 mapillary_tools_commands = [
     process,
     upload,
     sample_video,
     video_process,
     authenticate,
     process_and_upload,
     video_process_and_upload,
     zip,
+    sequence_delete_interval,
 ]
...