Simplify video annotation examples and request fixtures

Author: digaobarbosa

docs/core/project.md

@@ -7,25 +7,16 @@ 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")
-video_id = final_upload_status["videoId"]  # status is "uploaded"
-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,
-    }],
-}
-result = project.annotate_video_segments(video_id, document)
+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`

roboflow/core/project.py

@@ -602,22 +602,13 @@ def annotate_video_segments(
         split: Optional[str] = None,
         add_to_dataset: Optional[bool] = None,
     ) -> Dict:
-        """Annotate a canonical video Source in an Action Recognition project.
+        """Send a complete roboflow-video-coco document without converting timestamps.
 
-        ``document`` is a complete ``roboflow-video-coco`` document. Its native PTS,
-        frame indices, and rational time base are forwarded without conversion.
-        Use the final ``videoId`` from an uploaded video's status, which may differ
-        from the initial upload ID after content reuse.
-
-        The API adds the Source to the Dataset by default. Set ``add_to_dataset=False``
-        to keep its existing membership, ``split`` to assign train/valid/test, or
-        ``overwrite=True`` to replace different existing segments. An identical
-        retry succeeds without overwrite. Server rejections, including HTTP 409
-        preservation, raise ``AnnotationSaveError`` with ``status_code``.
-
-        Returns:
-            The API response, including ``success``, ``inDataset``, and
-            ``createdClasses``.
+        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,

tests/test_video_segment_annotation.py

@@ -9,29 +9,14 @@
 
 from roboflow.adapters.rfapi import AnnotationSaveError
 from roboflow.config import API_URL
-from roboflow.core.project import Project
+from tests import PROJECT_NAME, ROBOFLOW_API_KEY, WORKSPACE_NAME, RoboflowTest
 
 
-class TestVideoSegmentAnnotation(unittest.TestCase):
+class TestVideoSegmentAnnotation(RoboflowTest):
     def setUp(self):
-        self.project = Project(
-            "test-key",
-            {
-                "annotation": "actions",
-                "classes": {},
-                "colors": {},
-                "created": 0,
-                "id": "my-workspace/actions",
-                "images": 0,
-                "name": "Actions",
-                "public": False,
-                "splits": {},
-                "type": "action-recognition",
-                "unannotated": 0,
-                "updated": 0,
-            },
-        )
-        self.url = f"{API_URL}/my-workspace/actions/annotate/final-source-id"
+        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": [
@@ -59,56 +44,46 @@ def setUp(self):
             ],
         }
 
-    def test_acceptance_and_identical_retry_forward_the_canonical_document(self):
+    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)
-            self.assertEqual(self.project.annotate_video_segments("final-source-id", self.document), first)
-            self.assertEqual(self.project.annotate_video_segments("final-source-id", self.document), 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(len(http.calls), 2)
-            for call in http.calls:
+            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": ["test-key"], "name": ["annotations.json"]})
+                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):
-        def annotate(request):
-            query = parse_qs(urlparse(request.url).query)
-            if "overwrite" not in query:
-                return (
-                    409,
-                    {"Content-Type": "application/json"},
-                    json.dumps(
-                        {
-                            "error": {
-                                "message": "This video already has annotations. Send overwrite=true to replace them."
-                            }
-                        }
-                    ),
-                )
-            self.assertEqual(query["overwrite"], ["true"])
-            self.assertEqual(query["split"], ["valid"])
-            self.assertEqual(query["addToDataset"], ["false"])
-            return (
-                200,
-                {"Content-Type": "application/json"},
-                json.dumps({"success": True, "inDataset": False, "createdClasses": []}),
-            )
-
+        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_callback(responses.POST, self.url, callback=annotate)
+            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.assertIn("already has annotations", str(error.exception))
+            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, {"success": True, "inDataset": False, "createdClasses": []})
+            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: