Pipeline task management

Session persists a collection of tasks with unique identifiers; shared pipeline caller can optionally enqueue its call as a task instead of running it directly

model_server/base/pipelines/roiset_obmap.py

@@ -8,9 +8,10 @@ from .segment_zproj import segment_zproj_pipeline
 from .shared import call_pipeline
 from ..roiset import get_label_ids, RoiSet, RoiSetMetaParams, RoiSetExportParams
 
-from ..pipelines.shared import PipelineTrace, PipelineParams, PipelineRecord
+from ..pipelines.shared import PipelineQueueRecord, PipelineTrace, PipelineParams, PipelineRecord
 
 from ..models import Model, InstanceMaskSegmentationModel
+from ..session import session
 
 
 class RoiSetObjectMapParams(PipelineParams):
@@ -32,7 +33,7 @@ class RoiSetObjectMapParams(PipelineParams):
     )
     object_classifier_model_id: Union[str, None] = Field(
         None,
-        description='Object classifier used to classify segmented objectss'
+        description='Object classifier used to classify segmented objects'
     )
     patches_channel: int = Field(
         description='Channel of input image used in patches sent to object classifier'
@@ -64,6 +65,13 @@ def roiset_object_map(p: RoiSetObjectMapParams) -> RoiSetToObjectMapRecord:
     """
     return call_pipeline(roiset_object_map_pipeline, p)
 
+@router.put('/queue/roiset_to_obmap')
+def queue_roiset_object_map(p: RoiSetObjectMapParams) -> PipelineQueueRecord:
+    task_id = session.queue.add_task(
+        lambda x: call_pipeline(roiset_object_map_pipeline, x),
+        p
+    )
+    return {'task_id': task_id}
 
 def roiset_object_map_pipeline(
         accessors: Dict[str, GenericImageDataAccessor],