[ar-api] Annotate Action Recognition video segments from Python

docs/core/project.md

@@ -1,3 +1,32 @@
+## Action Recognition video segments
+
+Pass the **final** `videoId` from the uploaded video status to
+`Project.annotate_video_segments`. An upload's initial ID can change when the
+server reuses an existing video Source. The annotation document uses the
+official `roboflow-video-coco` format; keep its PTS and rational time base from
+the video rather than converting them to seconds or rounding frames.
+
+```python
+import json
+import roboflow
+
+project = roboflow.Roboflow(api_key="MY_API_KEY").workspace("my-workspace").project("actions")
+with open("clip.video-coco.json") as annotations:
+    document = json.load(annotations)  # An official video-coco document for clip.mp4.
+status = project.upload_video("clip.mp4", wait=True)
+if status["status"] != "uploaded":
+    raise RuntimeError(status["message"])
+result = project.annotate_video_segments(status["videoId"], document)
+```
+
+The API adds the video to the Dataset by default. Use `add_to_dataset=False`
+to preserve membership, `split="valid"` to choose a split, or `overwrite=True`
+to replace different existing segments. An identical retry succeeds. A
+conflicting annotation raises `AnnotationSaveError` with `status_code == 409`;
+the server keeps the prior segments. This route requires the public video
+annotate API from [platform PR #16236](https://github.com/roboflow/roboflow/pull/16236)
+and its [atomic save prerequisite](https://github.com/roboflow/roboflow/pull/16456).
+
 :::roboflow.core.project
 
 ## Upload a native Action Recognition video

roboflow/adapters/rfapi.py

@@ -842,6 +842,48 @@ def save_annotation(
     return responsejson
 
 
+def annotate_video_segments(
+    api_key: str,
+    workspace_url: str,
+    project_url: str,
+    video_id: str,
+    document: Dict[str, Any],
+    *,
+    overwrite: bool = False,
+    split: Optional[str] = None,
+    add_to_dataset: Optional[bool] = None,
+) -> Dict[str, Any]:
+    """Send a roboflow-video-coco document to the public Source annotation route."""
+    params = {"api_key": api_key, "name": "annotations.json"}
+    if overwrite:
+        params["overwrite"] = "true"
+    if split is not None:
+        params["split"] = split
+    if add_to_dataset is not None:
+        params["addToDataset"] = "true" if add_to_dataset else "false"
+
+    url = f"{API_URL}/{quote(workspace_url, safe='')}/{quote(project_url, safe='')}/annotate/{quote(video_id, safe='')}"
+    try:
+        response = requests.post(
+            url,
+            params=params,
+            json={"annotationFile": json.dumps(document)},
+            timeout=(60, 60),
+        )
+    except RequestException as e:
+        raise AnnotationSaveError(str(e)) from e
+
+    try:
+        result = response.json()
+    except ValueError:
+        raise _save_annotation_error(response) from None
+    if not isinstance(result, dict):
+        raise AnnotationSaveError(response.text, status_code=response.status_code)
+    if response.status_code != 200 or result.get("success") is not True:
+        raise _save_annotation_error(response)
+    return result
+
+
 def _save_annotation_url(
     api_key, project_url, name, image_id, job_name, is_prediction, overwrite=False, add_to_dataset=None
 ):

roboflow/core/project.py

@@ -593,6 +593,34 @@ def save_annotation(
 
         return annotation, upload_time, upload_retry_attempts
 
+    def annotate_video_segments(
+        self,
+        video_id: str,
+        document: Dict,
+        *,
+        overwrite: bool = False,
+        split: Optional[str] = None,
+        add_to_dataset: Optional[bool] = None,
+    ) -> Dict:
+        """Send a complete roboflow-video-coco document without converting timestamps.
+
+        Use the final uploaded-status ``videoId``. The API adds the video to the
+        Dataset unless ``add_to_dataset=False``; ``split`` assigns train/valid/test.
+        Identical retries succeed; different segments require ``overwrite=True``.
+        Returns the API response (``success``, ``inDataset``, ``createdClasses``).
+        Rejections raise ``AnnotationSaveError`` with the server's HTTP status.
+        """
+        return rfapi.annotate_video_segments(
+            self.__api_key,
+            self.__workspace,
+            self.__project_name,
+            video_id,
+            document,
+            overwrite=overwrite,
+            split=split,
+            add_to_dataset=add_to_dataset,
+        )
+
     def single_upload(
         self,
         image_path=None,

tests/test_video_segment_annotation.py

@@ -0,0 +1,109 @@
+"""User-facing SDK calls against isolated public annotate request fixtures."""
+
+import json
+import unittest
+from urllib.parse import parse_qs, urlparse
+
+import requests
+import responses
+
+from roboflow.adapters.rfapi import AnnotationSaveError
+from roboflow.config import API_URL
+from tests import PROJECT_NAME, ROBOFLOW_API_KEY, WORKSPACE_NAME, RoboflowTest
+
+
+class TestVideoSegmentAnnotation(RoboflowTest):
+    def setUp(self):
+        super().setUp()
+        self.project.type = "action-recognition"
+        self.url = f"{API_URL}/{WORKSPACE_NAME}/{PROJECT_NAME}/annotate/final-source-id"
+        self.document = {
+            "info": {"format": "roboflow-video-coco"},
+            "videos": [
+                {
+                    "id": 1,
+                    "file_name": "clip.mp4",
+                    "width": 640,
+                    "height": 360,
+                    "duration": 10,
+                    "fps": 24,
+                    "time_base": {"numerator": 1, "denominator": 12288},
+                }
+            ],
+            "categories": [{"id": 1, "name": "jumping"}],
+            "segments": [
+                {
+                    "id": 1,
+                    "video_id": 1,
+                    "category_id": 1,
+                    "start_frame": 48,
+                    "end_frame": 95,
+                    "start_pts": 24576,
+                    "end_pts": 49152,
+                }
+            ],
+        }
+
+    def test_finalized_source_and_identical_retry_forward_the_canonical_document(self):
+        first = {"success": True, "inDataset": True, "createdClasses": ["jumping"]}
+        retry = {"success": True, "inDataset": True, "createdClasses": []}
+        with responses.RequestsMock() as http:
+            http.add(
+                responses.GET,
+                f"{API_URL}/{WORKSPACE_NAME}/upload/video/reservation-id",
+                json={"status": "uploaded", "videoId": "final-source-id"},
+            )
+            http.add(responses.POST, self.url, json=first)
+            http.add(responses.POST, self.url, json=retry)
+            status = self.project.wait_for_video_upload("reservation-id")
+            self.assertEqual(self.project.annotate_video_segments(status["videoId"], self.document), first)
+            self.assertEqual(self.project.annotate_video_segments(status["videoId"], self.document), retry)
+
+            self.assertEqual([call.request.method for call in http.calls], ["GET", "POST", "POST"])
+            for call in http.calls[1:]:
+                query = parse_qs(urlparse(call.request.url).query)
+                self.assertEqual(query, {"api_key": [ROBOFLOW_API_KEY], "name": ["annotations.json"]})
+                self.assertEqual(json.loads(call.request.body), {"annotationFile": json.dumps(self.document)})
+
+    def test_conflict_is_raised_and_explicit_overwrite_and_split_are_forwarded(self):
+        message = "This video already has annotations. Send overwrite=true to replace them."
+        accepted = {"success": True, "inDataset": False, "createdClasses": []}
+        with responses.RequestsMock() as http:
+            http.add(responses.POST, self.url, json={"error": {"message": message}}, status=409)
+            http.add(responses.POST, self.url, json=accepted)
+            with self.assertRaises(AnnotationSaveError) as error:
+                self.project.annotate_video_segments("final-source-id", self.document)
+            self.assertEqual(error.exception.status_code, 409)
+            self.assertEqual(str(error.exception), message)
+
+            result = self.project.annotate_video_segments(
+                "final-source-id", self.document, overwrite=True, split="valid", add_to_dataset=False
+            )
+            self.assertEqual(result, accepted)
+            query = parse_qs(urlparse(http.calls[-1].request.url).query)
+            self.assertEqual(query["overwrite"], ["true"])
+            self.assertEqual(query["split"], ["valid"])
+            self.assertEqual(query["addToDataset"], ["false"])
+
+    def test_server_validation_and_transport_errors_are_not_recast_as_success(self):
+        with responses.RequestsMock() as http:
+            http.add(
+                responses.POST,
+                self.url,
+                json={"error": {"message": "Invalid roboflow-video-coco document"}},
+                status=400,
+            )
+            with self.assertRaises(AnnotationSaveError) as error:
+                self.project.annotate_video_segments("final-source-id", self.document)
+            self.assertEqual(error.exception.status_code, 400)
+            self.assertEqual(str(error.exception), "Invalid roboflow-video-coco document")
+
+        with responses.RequestsMock() as http:
+            http.add(responses.POST, self.url, body=requests.ConnectionError("connection lost"))
+            with self.assertRaises(AnnotationSaveError) as error:
+                self.project.annotate_video_segments("final-source-id", self.document)
+            self.assertIn("connection lost", str(error.exception))
+
+
+if __name__ == "__main__":
+    unittest.main()