From a40cb9b9f60f5c0fb97a73c5b4d4ce55b0208196 Mon Sep 17 00:00:00 2001 From: Joel Collins Date: Wed, 23 Jan 2019 16:48:49 +0000 Subject: [PATCH] Implemented basic threaded tasks for long-running functionality --- openflexure_microscope/api/app.py | 6 +- .../api/v1/blueprints/__init__.py | 2 +- .../api/v1/blueprints/task.py | 150 ++++++++++++++++++ openflexure_microscope/camera/base.py | 9 +- openflexure_microscope/exceptions.py | 2 + openflexure_microscope/microscope.py | 5 + openflexure_microscope/plugins/example/api.py | 24 ++- .../plugins/example/plugin.py | 22 ++- openflexure_microscope/task.py | 112 +++++++++++++ openflexure_microscope/utilities.py | 10 +- 10 files changed, 328 insertions(+), 14 deletions(-) create mode 100644 openflexure_microscope/api/v1/blueprints/task.py create mode 100644 openflexure_microscope/exceptions.py create mode 100644 openflexure_microscope/task.py diff --git a/openflexure_microscope/api/app.py b/openflexure_microscope/api/app.py index e133e445..dc655e4f 100644 --- a/openflexure_microscope/api/app.py +++ b/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) diff --git a/openflexure_microscope/api/v1/blueprints/__init__.py b/openflexure_microscope/api/v1/blueprints/__init__.py index f987ef03..eea7c36c 100644 --- a/openflexure_microscope/api/v1/blueprints/__init__.py +++ b/openflexure_microscope/api/v1/blueprints/__init__.py @@ -1 +1 @@ -from . import camera, stage, base, plugins +from . import camera, stage, base, plugins, task diff --git a/openflexure_microscope/api/v1/blueprints/task.py b/openflexure_microscope/api/v1/blueprints/task.py new file mode 100644 index 00000000..6e08c5ac --- /dev/null +++ b/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( + '//', + view_func=TaskAPI.as_view('task', microscope=microscope_obj) + ) + + return(blueprint) diff --git a/openflexure_microscope/camera/base.py b/openflexure_microscope/camera/base.py index 35a0afd8..cb8bf69b 100644 --- a/openflexure_microscope/camera/base.py +++ b/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 .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""" diff --git a/openflexure_microscope/exceptions.py b/openflexure_microscope/exceptions.py new file mode 100644 index 00000000..250eae65 --- /dev/null +++ b/openflexure_microscope/exceptions.py @@ -0,0 +1,2 @@ +class TaskDeniedException(Exception): + pass \ No newline at end of file diff --git a/openflexure_microscope/microscope.py b/openflexure_microscope/microscope.py index 0bad079d..e50604c9 100644 --- a/openflexure_microscope/microscope.py +++ b/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 diff --git a/openflexure_microscope/plugins/example/api.py b/openflexure_microscope/plugins/example/api.py index f73377be..a55254ed 100644 --- a/openflexure_microscope/plugins/example/api.py +++ b/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) \ No newline at end of file diff --git a/openflexure_microscope/plugins/example/plugin.py b/openflexure_microscope/plugins/example/plugin.py index 39261c6e..203aca19 100644 --- a/openflexure_microscope/plugins/example/plugin.py +++ b/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!" \ No newline at end of file + 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 \ No newline at end of file diff --git a/openflexure_microscope/task.py b/openflexure_microscope/task.py new file mode 100644 index 00000000..83b32103 --- /dev/null +++ b/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 \ No newline at end of file diff --git a/openflexure_microscope/utilities.py b/openflexure_microscope/utilities.py index 3c903dee..ecf38203 100644 --- a/openflexure_microscope/utilities.py +++ b/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 \ No newline at end of file + return out + +def entry_by_id(id: str, object_list: list): + """Return an object from a list, if .id matches id argument.""" + found = None + for o in object_list: + if o.id == id: + found = o + return found \ No newline at end of file