Initial commit

Author: jtc42

README.md

@@ -1,2 +1,2 @@
-# flask-labthings
-Python implementation of LabThings, based on the Flask microframework
+# python-labthings
+Python implementation of the LabThings API structure, based on the Flask microframework.

labthings/__init__.py

@@ -0,0 +1 @@
+from .core import tasks, lock

labthings/consumer/__init__.py

@@ -0,0 +1,23 @@
+from threading import ThreadError
+
+
+class LockError(ThreadError):
+    ERROR_CODES = {
+        "ACQUIRE_ERROR": "Unable to acquire. Lock in use by another thread.",
+        "IN_USE_ERROR": "Lock in use by another thread.",
+    }
+
+    def __init__(self, code, lock):
+        self.code = code
+        if code in LockError.ERROR_CODES:
+            self.message = LockError.ERROR_CODES[code]
+        else:
+            self.message = "Unknown error."
+
+        self.string = "{}: {}".format(self.code, self.message)
+        print(self.string)
+
+        ThreadError.__init__(self)
+
+    def __str__(self):
+        return self.string

labthings/core/__init__.py

@@ -0,0 +1,76 @@
+from threading import RLock
+
+from .exceptions import LockError
+
+
+class StrictLock(object):
+    """
+    Class that behaves like a Python RLock, but with stricter timeout conditions and custom exceptions.
+
+    Args:
+        timeout (int): Time, in seconds, lock acquisition will wait before raising an exception
+
+    Attributes:
+        _lock (:py:class:`threading.RLock`): Parent RLock object
+        timeout (int): Time, in seconds, lock acquisition will wait before raising an exception
+    """
+
+    def __init__(self, timeout=1):
+        self._lock = RLock()
+        self.timeout = timeout
+
+    def locked(self):
+        return self._lock.locked()
+
+    def acquire(self, blocking=True):
+        return self._lock.acquire(blocking, timeout=self.timeout)
+
+    def __enter__(self):
+        result = self._lock.acquire(blocking=True, timeout=self.timeout)
+        if result:
+            return result
+        else:
+            raise LockError("ACQUIRE_ERROR", self)
+
+    def __exit__(self, *args):
+        self._lock.release()
+
+    def release(self):
+        self._lock.release()
+
+
+class CompositeLock(object):
+    """
+    Class that behaves like a :py:class:`labthings.core.lock.StrictLock`,
+    but allows multiple locks to be acquired and released.
+
+    Args:
+        locks (list): List of parent RLock objects
+        timeout (int): Time, in seconds, lock acquisition will wait before raising an exception
+
+    Attributes:
+        locks (list): List of parent RLock objects
+        timeout (int): Time, in seconds, lock acquisition will wait before raising an exception
+    """
+
+    def __init__(self, locks, timeout=1):
+        self.locks = locks
+        self.timeout = timeout
+
+    def acquire(self, blocking=True):
+        return (lock.acquire(blocking=blocking) for lock in self.locks)
+
+    def __enter__(self):
+        result = (lock.acquire(blocking=True) for lock in self.locks)
+        if all(result):
+            return result
+        else:
+            raise LockError("ACQUIRE_ERROR", self)
+
+    def __exit__(self, *args):
+        for lock in self.locks:
+            lock.release()
+
+    def release(self):
+        for lock in self.locks:
+            lock.release()

labthings/core/exceptions.py

@@ -0,0 +1,25 @@
+__all__ = [
+    "taskify",
+    "tasks",
+    "dict",
+    "states",
+    "current_task",
+    "update_task_progress",
+    "cleanup_tasks",
+    "remove_task",
+    "update_task_data",
+    "ThreadTerminationError",
+]
+
+from .pool import (
+    tasks,
+    dict,
+    states,
+    current_task,
+    update_task_progress,
+    cleanup_tasks,
+    remove_task,
+    update_task_data,
+    taskify,
+)
+from .thread import ThreadTerminationError

labthings/core/lock.py

@@ -0,0 +1,145 @@
+import threading
+import logging
+from functools import wraps
+
+from .thread import TaskThread
+
+from flask import copy_current_request_context
+
+
+class TaskMaster:
+    def __init__(self, *args, **kwargs):
+        self._tasks = []
+
+    @property
+    def tasks(self):
+        """
+        Returns:
+            list: List of TaskThread objects. 
+        """
+        return self._tasks
+
+    @property
+    def dict(self):
+        """
+        Returns:
+            dict: Dictionary of TaskThread objects. Key is TaskThread ID.
+        """
+        return {str(t.id): t for t in self._tasks}
+
+    @property
+    def states(self):
+        """
+        Returns:
+            dict: Dictionary of TaskThread.state dictionaries. Key is TaskThread ID.
+        """
+        return {str(t.id): t.state for t in self._tasks}
+
+    def new(self, f, *args, **kwargs):
+        # copy_current_request_context allows threads to access flask current_app
+        task = TaskThread(
+            target=copy_current_request_context(f), args=args, kwargs=kwargs
+        )
+        self._tasks.append(task)
+        return task
+
+    def remove(self, task_id):
+        for task in self._tasks:
+            if (task.id == task_id) and not task.isAlive():
+                del task
+
+    def cleanup(self):
+        for task in self._tasks:
+            if not task.isAlive():
+                del task
+
+
+# Task management
+
+
+def tasks():
+    """
+    List of tasks in default taskmaster
+    Returns:
+        list: List of tasks in default taskmaster
+    """
+    global _default_task_master
+    return _default_task_master.tasks
+
+
+def dict():
+    """
+    Dictionary of tasks in default taskmaster
+    Returns:
+        dict: Dictionary of tasks in default taskmaster
+    """
+    global _default_task_master
+    return _default_task_master.dict
+
+
+def states():
+    """
+    Dictionary of TaskThread.state dictionaries. Key is TaskThread ID.
+    Returns:
+        dict: Dictionary of task states in default taskmaster
+    """
+    global _default_task_master
+    return _default_task_master.states
+
+
+def cleanup_tasks():
+    global _default_task_master
+    return _default_task_master.cleanup()
+
+
+def remove_task(task_id: str):
+    global _default_task_master
+    return _default_task_master.remove(task_id)
+
+
+# Operations on the current task
+
+
+def current_task():
+    current_task_thread = threading.current_thread()
+    if not isinstance(current_task_thread, TaskThread):
+        return None
+    return current_task_thread
+
+
+def update_task_progress(progress: int):
+    if current_task():
+        current_task().update_progress(progress)
+    else:
+        logging.info("Cannot update task progress of __main__ thread. Skipping.")
+
+
+def update_task_data(data: dict):
+    if current_task():
+        current_task().update_data(data)
+    else:
+        logging.info("Cannot update task data of __main__ thread. Skipping.")
+
+
+# Main "taskify" functions
+
+
+def taskify(f):
+    """
+    A decorator that wraps the passed in function
+    and surpresses exceptions should one occur
+    """
+
+    @wraps(f)
+    def wrapped(*args, **kwargs):
+        task = _default_task_master.new(
+            f, *args, **kwargs
+        )  # Append to parent object's task list
+        task.start()  # Start the function
+        return task
+
+    return wrapped
+
+
+# Create our default, protected, module-level task pool
+_default_task_master = TaskMaster()