Implemented basic threaded tasks for long-running functionality

openflexure_microscope/api/app.py

@@ -115,10 +115,14 @@ app.register_blueprint(stage_blueprint, url_prefix=uri('/stage', 'v1'))
 camera_blueprint = blueprints.camera.construct_blueprint(api_microscope)
 app.register_blueprint(camera_blueprint, url_prefix=uri('/camera', 'v1'))
 
-# Pluginroutes
+# Plugin routes
 plugin_blueprint = blueprints.plugins.construct_blueprint(api_microscope)
 app.register_blueprint(plugin_blueprint, url_prefix=uri('/plugin', 'v1'))
 
+# Task routes
+task_blueprint = blueprints.task.construct_blueprint(api_microscope)
+app.register_blueprint(task_blueprint, url_prefix=uri('/task', 'v1'))
+
 # List all routes
 list_routes(app)
 

openflexure_microscope/api/v1/blueprints/__init__.py

@@ -1 +1 @@
-from . import camera, stage, base, plugins
+from . import camera, stage, base, plugins, task

openflexure_microscope/api/v1/blueprints/task.py

@@ -0,0 +1,150 @@
+from openflexure_microscope.api.v1.views import MicroscopeView
+from flask import jsonify, abort, Blueprint
+
+class TaskListAPI(MicroscopeView):
+
+    def get(self):
+        """
+        Get list of long-running tasks.
+
+        .. :quickref: Tasks; Get collection of tasks
+
+        **Example response**:
+
+        .. sourcecode:: http
+
+          HTTP/1.1 200 OK
+          Vary: Accept
+          Content-Type: application/json
+
+          [
+            {
+                "end_time": "2019-01-23 16-33-33", 
+                "id": "db13a66787e1419bb06b1504e4d80b0c", 
+                "return": [
+                0.848622546386467, 
+                0.6106785018091292, 
+                ], 
+                "start_time": "2019-01-23 16-33-13", 
+                "status": "success"
+            }, 
+            {
+                "end_time": null, 
+                "id": "df46558cc8844924821bd0181881871e", 
+                "return": null, 
+                "start_time": "2019-01-23 16-34-54", 
+                "status": "running"
+            }
+          ]
+
+        :>header Accept: application/json
+        :query include_unavailable: return json representations of captures that have been completely deleted
+
+        :>header Content-Type: application/json
+        """
+
+        data = self.microscope.task.state
+
+        return jsonify(data)
+
+    def delete(self):
+        """
+        Clean list of long-running tasks (running tasks persist).
+
+        .. :quickref: Tasks; Clean collection of tasks
+
+        :>header Accept: application/json
+
+        :>header Content-Type: application/json
+        """
+
+        self.microscope.task.clean()
+        data = self.microscope.task.state
+
+        return jsonify(data)
+
+class TaskAPI(MicroscopeView):
+
+    def get(self, task_id):
+        """
+        Get JSON representation of a task
+
+        .. :quickref: Tasks; Get task
+
+        **Example request**:
+
+        .. sourcecode:: http
+
+          GET /task/db13a66787e1419bb06b1504e4d80b0c/ HTTP/1.1
+          Accept: application/json
+
+        **Example response**:
+
+        .. sourcecode:: http
+
+          HTTP/1.1 200 OK
+          Vary: Accept
+          Content-Type: application/json
+
+          {
+            "end_time": "2019-01-23 16-33-33", 
+            "id": "db13a66787e1419bb06b1504e4d80b0c", 
+            "return": [
+            0.848622546386467, 
+            0.6106785018091292, 
+            ], 
+            "start_time": "2019-01-23 16-33-13", 
+            "status": "success"
+          }
+
+        """
+
+        task = self.microscope.task.task_from_id(task_id)
+
+        if not task_id:
+            return abort(404)  # 404 Not Found
+
+        # Return task state
+        return jsonify(task.state)
+
+    def delete(self, task_id):
+        """
+        Remove a particular task (fails if task is currently running).
+
+        .. :quickref: Tasks; Remove a task
+
+        :>header Accept: application/json
+
+        :>header Content-Type: application/json
+        """
+
+        success = self.microscope.task.delete(task_id)
+        
+        if success:
+            data = {
+                'status': 'success',
+                'message': 'task successfully deleted'
+            }
+        else:
+            data = {
+                'status': 'fail',
+                'message': 'task could not be deleted, as it is currently running or does not exist'
+            }
+
+        return jsonify(data)
+
+def construct_blueprint(microscope_obj):
+
+    blueprint = Blueprint('task_blueprint', __name__)
+
+    blueprint.add_url_rule(
+        '/',
+        view_func=TaskListAPI.as_view('task_list', microscope=microscope_obj)
+    )
+
+    blueprint.add_url_rule(
+        '/<task_id>/',
+        view_func=TaskAPI.as_view('task', microscope=microscope_obj)
+    )
+
+    return(blueprint)

openflexure_microscope/camera/base.py

@@ -18,6 +18,7 @@ except ImportError:
 
 from .capture import CaptureObject, capture_from_dict, BASE_CAPTURE_PATH
 from openflexure_microscope.config import USER_CONFIG_DIR
+from openflexure_microscope.utilities import entry_by_id
 
 
 def last_entry(object_list: list):
@@ -28,14 +29,6 @@ def last_entry(object_list: list):
         return None
 
 
-def entry_by_id(id: str, object_list: list):
-    """Return an object from a list, if <object>.id matches id argument."""
-    found = None
-    for o in object_list:
-        if o.id == id:
-            found = o
-    return found
-
 
 def generate_basename():
     """Return a default filename based on the capture datetime"""

openflexure_microscope/exceptions.py

@@ -0,0 +1,2 @@
+class TaskDeniedException(Exception):
+    pass

openflexure_microscope/microscope.py

@@ -13,6 +13,7 @@ from .camera.pi import StreamingCamera
 from .plugins import PluginMount
 from .config import load_config, save_config, convert_config
 from .utilities import axes_to_array
+from .task import TaskOrchestrator
 
 
 class Microscope(object):
@@ -35,6 +36,9 @@ class Microscope(object):
         # Store entire runtime-config
         self._config = config
 
+        # Create a task orchestrator
+        self.task = TaskOrchestrator()
+
         # Create plugin mountpoint
         self.plugin = PluginMount(self)  #: :py:class:`openflexure_microscope.plugins.PluginMount`: Mounting point for all microscope plugins
 
@@ -49,6 +53,7 @@ class Microscope(object):
         if not 'fov' in self._config:
             self._config['fov'] = [0, 0]  # Assumes pi camera 2, and 40x objective
 
+
     def __enter__(self):
         """Create microscope on context enter."""
         return self

openflexure_microscope/plugins/example/api.py

@@ -1,7 +1,8 @@
 from openflexure_microscope.api.utilities import JsonPayload
 from openflexure_microscope.api.v1.views import MicroscopeViewPlugin
+from openflexure_microscope.exceptions import TaskDeniedException
 
-from flask import request, Response, escape
+from flask import request, Response, escape, jsonify, abort
 
 import logging
 
@@ -51,3 +52,24 @@ class HelloWorldAPI(MicroscopeViewPlugin):
             self.microscope.plugin_string = new_plugin_string
 
         return Response(self.microscope.plugin_string)
+
+class LongRunningAPI(MicroscopeViewPlugin):
+    """
+    An example API plugin that uses a long-running plugin method.
+    """
+    def post(self):
+        """
+        Method to call when an HTTP POST request is made.
+        """
+        # Get payload JSON
+        payload = JsonPayload(request)
+
+        # Extract a values from the JSON payload.
+        time_to_run = payload.param('time', default=10, convert=int)
+
+        try:
+            task = self.microscope.task.start(self.plugin.long_running, time_to_run)
+            return jsonify(task.state), 202
+
+        except TaskDeniedException:
+            return abort(409)

openflexure_microscope/plugins/example/plugin.py

@@ -1,6 +1,8 @@
+import random
+import time
 from openflexure_microscope.plugins import MicroscopePlugin
 
-from .api import IdentifyAPI, HelloWorldAPI
+from .api import IdentifyAPI, HelloWorldAPI, LongRunningAPI
 
 
 class Plugin(MicroscopePlugin):
@@ -11,6 +13,7 @@ class Plugin(MicroscopePlugin):
     api_views = {
         '/identify': IdentifyAPI,
         '/hello': HelloWorldAPI,
+        '/long_running': LongRunningAPI,
     }
 
     def identify(self):
@@ -27,4 +30,19 @@ class Plugin(MicroscopePlugin):
         Demonstrate passive method
         """
 
-        return "Hello world!"
+        return "Hello world!"
+
+    def long_running(self, t_run):
+        """
+        Demonstrate a long-running method that requires microscope hardware
+        """
+        print("Starting a long-running task...")
+        n_array = []
+
+        for _ in range(t_run):
+            n_array.append(random.random())
+            time.sleep(1)
+
+        print("Long-running task finished!")
+
+        return n_array

openflexure_microscope/task.py

@@ -0,0 +1,112 @@
+from threading import Thread
+from functools import wraps
+import datetime
+import time
+import uuid
+
+from openflexure_microscope.exceptions import TaskDeniedException
+from openflexure_microscope.utilities import entry_by_id
+
+class TaskOrchestrator:
+    def __init__(self):
+        # List of tasks in the orchestrator
+        self.tasks = []
+
+    @property
+    def state(self):
+        return [task.state for task in self.tasks]
+
+    def task_from_id(self, task_id):
+        return entry_by_id(task_id, self.tasks)
+
+    def start(self, function, *args, **kwargs):
+        """
+        Attach and start a new long-running task, if allowed by `start_condition()`.
+        """
+        # Create a task object
+        task = Task(function, *args, **kwargs)
+        
+        # If the task isn't allowed to run
+        if not self.start_condition(task):
+            raise TaskDeniedException("Unable to start this task. Aborting.")
+        else:
+            self.tasks.append(task)
+            task.start()
+            return task
+
+    def clean(self):
+        """
+        Remove all inactive tasks from the task array.
+        This will remove tasks that either finished or never started.
+        """
+        self.tasks = [task for task in self.tasks if task._running]
+
+    def delete(self, task_id):
+        """
+        Delete a given task by ID, only if task is not currently running.
+        """
+        success = False
+
+        for task in self.tasks:
+            if task.id == task_id and not task._running:
+                self.tasks.remove(task)
+                success = True
+
+        return success
+
+    def start_condition(self, task_obj):
+        """
+        Method returning bool describing if a particular task is allowed to run.
+        In the future, this will depend on the state of hardware locks?
+        Currently just allows a single task at a time.
+        """
+        return not any([task._running for task in self.tasks])
+
+
+class Task:
+    def __init__(self, task, *args, **kwargs):
+
+        # The task long-running method
+        self.task = task
+        self.args = args
+        self.kwargs = kwargs
+
+        # If task is currently running
+        self._running = False
+        
+        self.id = uuid.uuid4().hex
+
+        self.state = {
+            'id': self.id,
+            'status': 'idle',
+            'return': None,
+            'start_time': None,
+            'end_time': None,
+        }
+
+    def process(self, f):
+        def wrapped(*args, **kwargs):
+            self.state['status'] = 'running'
+            try:
+                r = f(*args, **kwargs)
+                s = 'success'
+            except Exception as e:
+                r = e
+                s = 'error'
+            self.state['return'] = r
+            self.state['status'] = s
+        return wrapped
+
+    def run(self):
+        self.state['start_time'] = datetime.datetime.now().strftime("%Y-%m-%d %H-%M-%S")
+        self.process(self.task)(*self.args, **self.kwargs)
+        self._running = False
+        self.state['end_time'] = datetime.datetime.now().strftime("%Y-%m-%d %H-%M-%S")
+
+    def start(self):  # Start and draw
+        # If not already running
+        if not self._running:
+            self._running = True  # Set start command
+            thread = Thread(target=self.run)  # Define thread
+            thread.daemon = True  # Stop this thread when main thread closes
+            thread.start()  # Start thread

openflexure_microscope/utilities.py

@@ -26,4 +26,12 @@ def filter_dict(dictionary: dict, keys: list):
 	# Create new dictionary by running reduce on key, val pairs
 	out = reduce(lambda x, y: {y: x}, reversed(keys), val)
 	
-	return out
+	return out
+
+def entry_by_id(id: str, object_list: list):
+    """Return an object from a list, if <object>.id matches id argument."""
+    found = None
+    for o in object_list:
+        if o.id == id:
+            found = o
+    return found