Added "standard" task creation (#605)

* added method to create task without any base annotation

* updated changelog

* applied PR feedback

Author: hotzenklotz

webknossos/Changelog.md

@@ -12,6 +12,7 @@ For upgrade instructions, please check the respective *Breaking Changes* section
 ### Breaking Changes
 
 ### Added
+- Added `Task.create()` method to create tasks by prodiving a dataset name, location, and rotation. [#605](https://github.com/scalableminds/webknossos-libs/pull/605)
 
 ### Changed
 

webknossos/webknossos/administration/task.py

@@ -12,7 +12,7 @@
     task_info,
 )
 from webknossos.client.context import _get_generated_client
-from webknossos.geometry import BoundingBox
+from webknossos.geometry import BoundingBox, Vec3Int
 
 logger = logging.getLogger(__name__)
 
@@ -64,9 +64,11 @@ def create_from_annotations(
         bounding_box: Optional[BoundingBox] = None,
     ) -> List["Task"]:
         """Submits tasks in webKnossos based on existing annotations, and returns the Task objects"""
+
         assert (
             len(base_annotations) > 0
         ), "Must supply at least one base annotation to create tasks"
+
         client = _get_generated_client()
         url = f"{client.base_url}/api/tasks/createFromFiles"
         task_parameters = {
@@ -86,6 +88,7 @@ def create_from_annotations(
         files: Mapping[str, Tuple[str, Union[bytes, BinaryIO]]] = {
             f"{a.name}.zip": (f"{a.name}.zip", a.binary()) for a in base_annotations
         }
+
         response = httpx.post(
             url=url,
             headers=client.get_headers(),
@@ -97,31 +100,56 @@ def create_from_annotations(
         assert (
             response.status_code == 200
         ), f"Failed to create tasks from files: {response.status_code}: {response.content.decode('utf-8')}"
-        result = json.loads(response.content)
-        if "warnings" in result:
-            warnings = result["warnings"]
-            if len(warnings) > 0:
-                logger.warning(
-                    f"There were {len(warnings)} warnings during task creation:"
-                )
-            for warning in warnings:
-                logger.warning(warning)
-        assert "tasks" in result, "Invalid result of task creation"
-        successes = []
-        errors = []
-        for t in result["tasks"]:
-            print(t)
-            if "success" in t:
-                successes.append(t["success"])
-            if "error" in t:
-                errors.append(t["error"])
-        if len(errors) > 0:
-            logger.error(f"{len(errors)} tasks could not be created:")
-            for error in errors:
-                logger.error(error)
-        if len(successes) > 0:
-            logger.info(f"{len(successes)} tasks were successfully created.")
-        return [cls._from_dict(t) for t in successes]
+
+        return cls._handle_task_creation_response(response)
+
+    @classmethod
+    def create(
+        cls,
+        task_type_id: str,
+        project_name: str,
+        dataset_name: str,
+        needed_experience_domain: str,
+        needed_experience_value: int,
+        starting_position: Vec3Int,
+        starting_rotation: Optional[Vec3Int] = Vec3Int(0, 0, 0),
+        instances: int = 1,
+        script_id: Optional[str] = None,
+        bounding_box: Optional[BoundingBox] = None,
+    ) -> List["Task"]:
+        """Submits tasks in webKnossos based on a dataset, starting position + rotation, and returns the Task objects"""
+
+        client = _get_generated_client()
+        url = f"{client.base_url}/api/tasks"
+        task_parameters = {
+            "taskTypeId": task_type_id,
+            "neededExperience": {
+                "domain": needed_experience_domain,
+                "value": needed_experience_value,
+            },
+            "openInstances": instances,
+            "projectName": project_name,
+            "scriptId": script_id,
+            "dataSet": dataset_name,
+            "editPosition": starting_position,
+            "editRotation": starting_rotation,
+            "boundingBox": bounding_box.to_wkw_dict()
+            if bounding_box is not None
+            else None,
+        }
+
+        response = httpx.post(
+            url=url,
+            headers=client.get_headers(),
+            cookies=client.get_cookies(),
+            timeout=client.get_timeout(),
+            json=[task_parameters],
+        )
+        assert (
+            response.status_code == 200
+        ), f"Failed to create tasks: {response.status_code}: {response.content.decode('utf-8')}"
+
+        return cls._handle_task_creation_response(response)
 
     @classmethod
     def _from_dict(cls, response_dict: Dict) -> "Task":
@@ -159,3 +187,32 @@ def get_annotation_infos(self) -> List[AnnotationInfo]:
     def get_project(self) -> Project:
         """Returns the project this task belongs to"""
         return Project.get_by_id(self.project_id)
+
+    @classmethod
+    def _handle_task_creation_response(cls, response: httpx.Response) -> List["Task"]:
+        result = json.loads(response.content)
+        if "warnings" in result:
+            warnings = result["warnings"]
+            if len(warnings) > 0:
+                logger.warning(
+                    f"There were {len(warnings)} warnings during task creation:"
+                )
+            for warning in warnings:
+                logger.warning(warning)
+        assert "tasks" in result, "Invalid result of task creation"
+
+        successes = []
+        errors = []
+        for t in result["tasks"]:
+            print(t)
+            if "success" in t:
+                successes.append(t["success"])
+            if "error" in t:
+                errors.append(t["error"])
+        if len(errors) > 0:
+            logger.error(f"{len(errors)} tasks could not be created:")
+            for error in errors:
+                logger.error(error)
+        if len(successes) > 0:
+            logger.info(f"{len(successes)} tasks were successfully created.")
+        return [cls._from_dict(t) for t in successes]