[ar-api] Add CLI video segment annotation (VID-59)

CLI-COMMANDS.md

@@ -471,7 +471,7 @@ roboflow workspace stats --start-date 2026-01-01 --end-date 2026-03-31
 roboflow universe search "hard hats" --type dataset --limit 5
 ```
 
-### Native video upload and status
+### Native video upload and segment annotation
 
 Action Recognition projects take whole videos as Sources. `video upload` streams the
 original MP4/MOV bytes without re-encoding, then reports the **canonical video ID** to use
@@ -495,8 +495,25 @@ roboflow video upload-status aBcD1234 -p my-ar-project --json
 roboflow video upload-status aBcD1234 -p my-ar-project --wait --poll-timeout 120
 ```
 
+```bash
+# 3. Annotate segments from a complete roboflow-video-coco document.
+#    The file is forwarded unchanged, so native frame indices, PTS and
+#    rational time bases are preserved exactly as authored.
+roboflow video annotate -p my-ar-project -i aBcD1234 -a segments.json --json
+# { "success": true, "inDataset": true, "createdClasses": ["walking"] }
+```
+
+In `segments.json`, `segments` belongs at the document top level and
+`videos[0].time_base` is a rational object such as
+`{"numerator": 1, "denominator": 15360}`. `images` and `annotations` may be
+omitted; if supplied, each must be an empty array. Use the original video's
+probed PTS values rather than deriving them from frame indices or nominal FPS.
+
 Upload accepts `-b/--batch`, `-t/--tag` (comma-separated), `--metadata` (JSON object) and
-`-s/--split`.
+`-s/--split`. Annotate defaults to the API behaviour of adding the Source to the Dataset;
+override with `--no-add-to-dataset`, set the split with `-s/--split`, and pass `--overwrite`
+to replace segments that already differ (otherwise a conflicting save is rejected and an
+identical re-submit succeeds).
 
 Exit codes follow the CLI contract: `0` success, `1` error, `2` auth, `3` not found. A
 `failed` ingestion state and a `--wait` timeout both exit nonzero; the timeout message names
@@ -600,7 +617,7 @@ Version numbers are always numeric — that's how `x/y` is disambiguated between
 | `asynctasks` | Inspect async background tasks (e.g. project forks) |
 | `trash` | List items in Trash |
 | `universe` | Search Roboflow Universe |
-| `video` | Native video upload/status and video inference |
+| `video` | Native video upload/status/annotation, and video inference |
 | `batch` | Batch processing jobs *(coming soon)* |
 | `completion` | Install or generate shell completion scripts (bash, zsh, fish) |
 

roboflow/cli/handlers/video.py

@@ -1,4 +1,4 @@
-"""Video commands: native video Source ingestion and legacy video inference."""
+"""Video commands: native video Source ingestion/annotation and legacy video inference."""
 
 from __future__ import annotations
 
@@ -10,7 +10,7 @@
 
 video_app = typer.Typer(
     cls=SortedGroup,
-    help="Native video upload and video inference operations",
+    help="Native video upload/annotation and video inference operations",
     no_args_is_help=True,
 )
 
@@ -59,7 +59,7 @@ def upload(
 ) -> None:
     """Upload original video bytes as a native video Source.
 
-    Streams the file unchanged and reports the canonical video ID.
+    Streams the file unchanged and reports the canonical video ID to annotate.
     """
     args = ctx_to_args(
         ctx,
@@ -101,6 +101,43 @@ def upload_status(
     _video_upload_status(args)
 
 
+@video_app.command("annotate")
+def annotate(
+    ctx: typer.Context,
+    annotation_file: Annotated[
+        str, typer.Option("-a", "--annotation-file", help="Path to a complete roboflow-video-coco JSON file")
+    ],
+    project: Annotated[str, typer.Option("-p", "--project", help="Project ID, or workspace/project")],
+    video_id: Annotated[str, typer.Option("-i", "--video-id", help="Canonical video ID from 'video upload'")],
+    add_to_dataset: Annotated[
+        Optional[bool],
+        typer.Option(
+            "--add-to-dataset/--no-add-to-dataset",
+            help="Override the API default of adding the Source to the Dataset",
+        ),
+    ] = None,
+    overwrite: Annotated[
+        bool, typer.Option("--overwrite", help="Replace different existing segments on this video")
+    ] = False,
+    split: Annotated[Optional[str], typer.Option("-s", "--split", help="Dataset split: train, valid or test")] = None,
+) -> None:
+    """Annotate a native video Source's segments from a video-coco file.
+
+    The file is read whole and forwarded unchanged, so native frame indices,
+    PTS and rational time bases survive exactly as authored.
+    """
+    args = ctx_to_args(
+        ctx,
+        annotation_file=annotation_file,
+        project=project,
+        video_id=video_id,
+        add_to_dataset=add_to_dataset,
+        overwrite=overwrite,
+        split=split,
+    )
+    _video_annotate(args)
+
+
 # ---------------------------------------------------------------------------
 # Business logic (unchanged from argparse version)
 # ---------------------------------------------------------------------------
@@ -371,3 +408,74 @@ def _video_upload_status(args) -> None:  # noqa: ANN001
             return
 
     _emit_upload_status(args, status)
+
+
+def _video_annotate(args) -> None:  # noqa: ANN001
+    import json as json_mod
+
+    from roboflow.adapters.rfapi import AnnotationSaveError
+    from roboflow.cli._output import output, output_api_error, output_error
+
+    try:
+        # Explicit UTF-8: the locale default (cp1252 on Windows) would silently corrupt non-ASCII class names.
+        with open(args.annotation_file, encoding="utf-8") as handle:
+            document = json_mod.load(handle)
+    except OSError as exc:
+        output_error(args, f"Cannot read annotation file: {exc}", hint="Pass the path to a video-coco JSON file.")
+        return
+    except UnicodeDecodeError as exc:
+        output_error(
+            args,
+            f"{args.annotation_file} is not UTF-8: {exc}",
+            hint="Save the video-coco document as UTF-8 JSON.",
+        )
+        return
+    except json_mod.JSONDecodeError as exc:
+        output_error(
+            args,
+            f"Invalid JSON in {args.annotation_file}: {exc}",
+            hint="The file must be one complete roboflow-video-coco document.",
+        )
+        return
+
+    if not isinstance(document, dict):
+        output_error(
+            args,
+            f"{args.annotation_file} must contain a JSON object.",
+            hint="The file must be one complete roboflow-video-coco document.",
+        )
+        return
+
+    project = _load_project(args)
+    if project is None:
+        return
+
+    try:
+        # `document` is forwarded as parsed: no re-encoding of frames, PTS or time bases.
+        result = project.annotate_video_segments(
+            args.video_id,
+            document,
+            overwrite=args.overwrite,
+            split=args.split,
+            add_to_dataset=args.add_to_dataset,
+        )
+    except AnnotationSaveError as exc:
+        hints = {
+            400: "Check that the document is a complete video-coco with at least one segment.",
+            409: "Different segments already exist on this video. Re-run with --overwrite to replace them.",
+        }
+        output_api_error(
+            args,
+            exc,
+            hint=hints.get(exc.status_code),
+            not_found_hint="Check the canonical video ID from 'roboflow video upload'.",
+        )
+        return
+
+    lines = [f"Annotated video {args.video_id}."]
+    if "inDataset" in result:
+        lines.append(f"In dataset: {'yes' if result.get('inDataset') else 'no'}")
+    created = result.get("createdClasses")
+    if created:
+        lines.append(f"Created classes: {', '.join(map(str, created))}")
+    output(args, result, text="\n".join(lines))

tests/cli/test_video_handler.py

@@ -78,6 +78,52 @@ def test_status_passes_job_id_to_api(self, _mock_key, mock_api) -> None:
     }
 }
 
+# A complete video-coco document in the shape the import schema accepts, taken
+# from a real MOV: `segments` is top level, `time_base` is a rational object,
+# and `images`/`annotations` stay empty. Tests assert it reaches the SDK with
+# every value identical.
+VIDEO_COCO_DOCUMENT = {
+    "info": {"format": "roboflow-video-coco"},
+    "videos": [
+        {
+            "id": 1,
+            "file_name": "clip.mov",
+            "width": 1620,
+            "height": 1080,
+            "duration": 4.566667,
+            "fps": 30,
+            "frame_count": 137,
+            "time_base": {"numerator": 1, "denominator": 15360},
+        }
+    ],
+    "categories": [{"id": 1, "name": "hand_gesture"}],
+    # Real MOV presentation timestamps are not frame_index * ticks_per_frame,
+    # so they must survive the read exactly rather than being recomputed.
+    "segments": [
+        {
+            "id": 1,
+            "video_id": 1,
+            "category_id": 1,
+            "start_frame": 5,
+            "end_frame": 64,
+            "start_pts": 3067,
+            "end_pts": 33275,
+        }
+    ],
+    "images": [],
+    "annotations": [],
+}
+
+
+def _write_json(directory, name, payload):
+    import json as json_mod
+    import os
+
+    path = os.path.join(directory, name)
+    with open(path, "w") as handle:
+        json_mod.dump(payload, handle)
+    return path
+
 
 class NativeVideoCliTest(unittest.TestCase):
     """Shared fixtures that let real command dispatch build a real Project."""
@@ -108,7 +154,7 @@ class TestNativeVideoRegistration(NativeVideoCliTest):
     """The real CLI exposes the native video commands alongside inference."""
 
     def test_native_commands_are_registered(self) -> None:
-        for command in ("upload", "upload-status"):
+        for command in ("upload", "upload-status", "annotate"):
             with self.subTest(command=command):
                 result = runner.invoke(app, ["video", command, "--help"])
                 self.assertEqual(result.exit_code, 0)
@@ -401,6 +447,137 @@ def test_failed_state_exits_nonzero(self, mock_status) -> None:
         self.assertNotEqual(result.exit_code, 0)
 
 
+class TestVideoAnnotate(NativeVideoCliTest):
+    """`roboflow video annotate` forwards the video-coco document unchanged."""
+
+    def setUp(self) -> None:
+        super().setUp()
+        self.document_path = _write_json(self.tmp.name, "segments.json", VIDEO_COCO_DOCUMENT)
+
+    @patch("roboflow.core.project.Project.annotate_video_segments")
+    def test_document_and_defaults_are_forwarded_unchanged(self, mock_annotate) -> None:
+        mock_annotate.return_value = {"success": True, "inDataset": True, "createdClasses": ["walking"]}
+
+        result = runner.invoke(
+            app,
+            ["video", "annotate", "-p", self.project_ref, "-i", "source-9", "-a", self.document_path],
+        )
+
+        self.assertEqual(result.exit_code, 0, result.output)
+        mock_annotate.assert_called_once_with(
+            "source-9",
+            VIDEO_COCO_DOCUMENT,
+            overwrite=False,
+            split=None,
+            add_to_dataset=None,
+        )
+        self.assertIn("walking", result.output)
+
+    @patch("roboflow.core.project.Project.annotate_video_segments")
+    def test_explicit_overwrite_split_and_membership_are_forwarded(self, mock_annotate) -> None:
+        mock_annotate.return_value = {"success": True, "inDataset": False}
+
+        result = runner.invoke(
+            app,
+            [
+                "video",
+                "annotate",
+                "-p",
+                self.project_ref,
+                "-i",
+                "source-9",
+                "-a",
+                self.document_path,
+                "--overwrite",
+                "-s",
+                "test",
+                "--no-add-to-dataset",
+            ],
+        )
+
+        self.assertEqual(result.exit_code, 0, result.output)
+        mock_annotate.assert_called_once_with(
+            "source-9",
+            VIDEO_COCO_DOCUMENT,
+            overwrite=True,
+            split="test",
+            add_to_dataset=False,
+        )
+        self.assertIn("In dataset: no", result.output)
+
+        mock_annotate.reset_mock()
+        runner.invoke(
+            app,
+            ["video", "annotate", "-p", self.project_ref, "-i", "s1", "-a", self.document_path, "--add-to-dataset"],
+        )
+        self.assertIs(mock_annotate.call_args.kwargs["add_to_dataset"], True)
+
+    @patch("roboflow.core.project.Project.annotate_video_segments")
+    def test_json_output_is_the_server_response(self, mock_annotate) -> None:
+        mock_annotate.return_value = {"success": True, "inDataset": True, "createdClasses": []}
+
+        result = runner.invoke(
+            app,
+            ["--json", "video", "annotate", "-p", self.project_ref, "-i", "s1", "-a", self.document_path],
+        )
+
+        self.assertEqual(result.exit_code, 0, result.output)
+        self.assertEqual(json.loads(result.output), {"success": True, "inDataset": True, "createdClasses": []})
+
+    @patch("roboflow.core.project.Project.annotate_video_segments")
+    def test_api_rejections_follow_exit_code_contract(self, mock_annotate) -> None:
+        from roboflow.adapters.rfapi import AnnotationSaveError
+
+        cases = [
+            (409, 1, "--overwrite"),
+            (400, 1, "at least one segment"),
+            (404, 3, "canonical video ID"),
+            (401, 2, "ROBOFLOW_API_KEY"),
+            (None, 1, None),  # transport failure: no status, no document hint
+        ]
+        for status_code, exit_code, hint in cases:
+            with self.subTest(status_code=status_code):
+                mock_annotate.side_effect = AnnotationSaveError("server said no", status_code=status_code)
+
+                result = runner.invoke(
+                    app,
+                    ["--json", "video", "annotate", "-p", self.project_ref, "-i", "s1", "-a", self.document_path],
+                )
+
+                self.assertEqual(result.exit_code, exit_code)
+                error = json.loads(result.output)["error"]
+                self.assertEqual(error["message"], "server said no")
+                if hint is None:
+                    self.assertNotIn("hint", error)
+                else:
+                    self.assertIn(hint, error["hint"])
+
+    @patch("roboflow.core.project.Project.annotate_video_segments")
+    def test_unreadable_document_never_reaches_the_api(self, mock_annotate) -> None:
+        malformed = os.path.join(self.tmp.name, "bad.json")
+        with open(malformed, "w") as handle:
+            handle.write('{"segments": ')
+        utf16 = os.path.join(self.tmp.name, "utf16.json")
+        with open(utf16, "w", encoding="utf-16") as handle:  # what Windows PowerShell 5.1 redirection writes
+            json.dump(VIDEO_COCO_DOCUMENT, handle)
+        cases = [
+            (malformed, "Invalid JSON"),
+            (utf16, "is not UTF-8"),
+            (_write_json(self.tmp.name, "list.json", [1, 2, 3]), "must contain a JSON object"),
+            (os.path.join(self.tmp.name, "absent.json"), "Cannot read annotation file"),
+        ]
+        for path, message in cases:
+            with self.subTest(message=message):
+                result = runner.invoke(
+                    app, ["--json", "video", "annotate", "-p", self.project_ref, "-i", "s1", "-a", path]
+                )
+
+                self.assertEqual(result.exit_code, 1)
+                self.assertIn(message, json.loads(result.output)["error"]["message"])
+        mock_annotate.assert_not_called()
+        self.mock_get_project.assert_not_called()
+
+
 class TestLegacyVideoContractsIntact(unittest.TestCase):
     """The native commands must not disturb legacy video inference."""