feat: Robotics mcap インポートパイプライン追加

README.md

@@ -2230,6 +2230,20 @@ history = client.import_robotics_contents_file(
 )
 ```
 
+#### Import mcap
+
+Import an mcap zip file into an existing robotics task.
+
+This method starts an import pipeline, uploads the zip file via a signed URL, and triggers the batch import.
+
+```python
+result = client.import_robotics_mcap(
+    project="YOUR_PROJECT_SLUG",
+    task_id="YOUR_TASK_ID",
+    file_path="ZIP_FILE_PATH",  # Supported extension is .zip
+)
+```
+
 #### Import LeRobot Dataset
 
 Import a [LeRobot](https://github.com/huggingface/lerobot) dataset (v3) into a FastLabel robotics project.

examples/import_robotics_mcap.py

@@ -0,0 +1,19 @@
+"""
+Import an mcap zip file into an existing FastLabel robotics task.
+
+This starts an import pipeline, uploads the zip via a signed URL,
+and triggers the batch import.
+"""
+
+from pprint import pprint
+
+import fastlabel
+
+client = fastlabel.Client()
+
+result = client.import_robotics_mcap(
+    project="YOUR_PROJECT_SLUG",
+    task_id="YOUR_TASK_ID",
+    file_path="ZIP_FILE_PATH",  # Supported extension is .zip
+)
+pprint(result)

fastlabel/__init__.py

@@ -2451,6 +2451,55 @@ def import_robotics_contents_file(
 
         return self.api.post_request(endpoint, payload=payload)
 
+    def import_robotics_mcap(self, project: str, task_id: str, file_path: str) -> dict:
+        """
+        Import robotics mcap zip file.
+        project is slug of your project (Required).
+        task_id is a id of the robotics task to import contents (Required).
+        file_path is a path to data. Supported extensions are zip (Required).
+        """
+        if not file_path.endswith(".zip"):
+            raise FastLabelInvalidException(
+                "Supported extensions are zip (Required).", 422
+            )
+        history_id = self._start_pipeline(project=project, task_id=task_id)
+        self._upload_zip_with_signed_url(
+            history_id=history_id,
+            file_path=file_path,
+        )
+        return self._trigger_batch_import(history_id=history_id)
+
+    def _start_pipeline(
+        self,
+        project: str,
+        task_id: str,
+    ) -> str:
+        history: dict = self.api.post_request(
+            "data-lake/robotics/import-pipeline/start",
+            payload={
+                "project": project,
+                "taskId": task_id,
+            },
+        )
+        logger.info(f"Started data lake pipeline with history ID: {history['id']}")
+        return history["id"]
+
+    def _upload_zip_with_signed_url(self, history_id: str, file_path: str) -> None:
+        signed_url_data = self.api.get_request(
+            "data-lake/robotics/import-pipeline/signed-url",
+            params={"historyId": history_id},
+        )
+        res = self.api.upload_zipfile(url=signed_url_data["url"], file_path=file_path)
+        res.raise_for_status()
+
+    def _trigger_batch_import(self, history_id: str) -> dict:
+        return self.api.post_request(
+            "data-lake/robotics/import-pipeline/run",
+            payload={
+                "historyId": history_id,
+            },
+        )
+
     # Task Update
 
     def update_task(