Added Events support

Author: jtc42

examples/simple_thing.py

@@ -3,6 +3,8 @@
 from labthings import ActionView, PropertyView, create_app, fields, find_component, op
 from labthings.example_components import PretendSpectrometer
 
+import logging
+
 """
 Class for our lab component functionality. This could include serial communication,
 equipment API calls, network requests, or a "virtual" device as seen here.
@@ -56,6 +58,7 @@ def get(self):
         my_component = find_component("org.labthings.example.mycomponent")
         return my_component.data
 
+
 """
 Create a view to start an averaged measurement, and register is as a Thing action
 """
@@ -66,7 +69,9 @@ class MeasurementAction(ActionView):
     # Pass to post function as dictionary argument.
     args = {
         "averages": fields.Integer(
-            missing=20, example=20, description="Number of data sets to average over",
+            missing=20,
+            example=20,
+            description="Number of data sets to average over",
         )
     }
     # Marshal the response as a list of numbers
@@ -76,13 +81,16 @@ class MeasurementAction(ActionView):
     @op.invokeaction
     def post(self, args):
         """Start an averaged measurement"""
+        logging.warning("Starting a measurement")
 
         # Find our attached component
         my_component = find_component("org.labthings.example.mycomponent")
 
         # Get arguments and start a background task
         n_averages = args.get("averages")
 
+        logging.warning("Finished a measurement")
+
         # Return the task information
         return my_component.average_data(n_averages)
 

src/labthings/__init__.py

@@ -30,7 +30,7 @@
 from .sync import ClientEvent, CompositeLock, StrictLock
 
 # Views
-from .views import ActionView, PropertyView, op
+from .views import ActionView, PropertyView, EventView, op
 
 # Suggested WSGI server class
 from .wsgi import Server
@@ -58,5 +58,5 @@
     "json",
     "PropertyView",
     "ActionView",
-    "op"
+    "op",
 ]

src/labthings/actions/thread.py

@@ -8,6 +8,8 @@
 from flask import copy_current_request_context, has_request_context
 
 from ..utilities import TimeoutTracker
+from ..schema import LogRecordSchema
+from ..deque import Deque
 
 _LOG = logging.getLogger(__name__)
 
@@ -29,6 +31,7 @@ def __init__(
         kwargs=None,
         daemon=True,
         default_stop_timeout: int = 5,
+        log_len: int = 100,
     ):
         threading.Thread.__init__(
             self,
@@ -76,10 +79,14 @@ def __init__(
         self._end_time = None  # Task end time
 
         # Public state properties
-        self.input: dict = {}  # Input arguments. TODO: Automate this. Currently manual via Action.dispatch_request
+        self.input: dict = (
+            {}
+        )  # Input arguments. TODO: Automate this. Currently manual via Action.dispatch_request
         self.progress: int = None  # Percent progress of the task
         self.data = {}  # Dictionary of custom data added during the task
-        self.log = []  # The log will hold dictionary objects with log information
+        self.log = Deque(
+            None, log_len
+        )  # The log will hold dictionary objects with log information
 
         # Stuff for handling termination
         self._running_lock = (
@@ -106,7 +113,7 @@ def status(self):
         Current running status of the thread.
 
         ==============  =============================================
-        Status          Meaning     
+        Status          Meaning
         ==============  =============================================
         ``pending``     Not yet started
         ``running``     Currently in-progress
@@ -143,7 +150,7 @@ def update_progress(self, progress: int):
     def update_data(self, data: dict):
         """
 
-        :param data: dict: 
+        :param data: dict:
 
         """
         # Store data to be used before task finishes (eg for real-time plotting)
@@ -165,15 +172,15 @@ def _thread_proc(self, f):
         """Wraps the target function to handle recording `status` and `return` to `state`.
         Happens inside the task thread.
 
-        :param f: 
+        :param f:
 
         """
 
         def wrapped(*args, **kwargs):
             """
 
-            :param *args: 
-            :param **kwargs: 
+            :param *args:
+            :param **kwargs:
 
             """
             nonlocal self
@@ -222,7 +229,7 @@ def get(self, block=True, timeout=None):
     def _async_raise(self, exc_type):
         """
 
-        :param exc_type: 
+        :param exc_type:
 
         """
         # Should only be called on a started thread, so raise otherwise.
@@ -319,18 +326,20 @@ def stop(self, timeout=None, exception=ActionKilledException):
 
 
 class ThreadLogHandler(logging.Handler):
-    def __init__(self, thread=None, dest=None, level=logging.INFO):
+    def __init__(
+        self, thread=None, dest=None, level=logging.INFO, default_log_len: int = 100
+    ):
         """Set up a log handler that appends messages to a list.
-    
+
         This log handler will first filter by ``thread``, if one is
         supplied.  This should be a ``threading.Thread`` object.
         Only log entries from the specified thread will be
         saved.
-    
+
         ``dest`` should specify a list, to which we will append
         each log entry as it comes in.  If none is specified, a
         new list will be created.
-    
+
         NB this log handler does not currently rotate or truncate
         the list - so if you use it on a thread that produces a
         lot of log messages, you may run into memory problems.
@@ -340,34 +349,28 @@ def __init__(self, thread=None, dest=None, level=logging.INFO):
         logging.Handler.__init__(self)
         self.setLevel(level)
         self.thread = thread
-        self.dest = dest if dest is not None else []
+        self.dest = dest if dest is not None else Deque(None, default_log_len)
         self.addFilter(self.check_thread)
 
     def check_thread(self, record):
         """Determine if a thread matches the desired record
 
-        :param record: 
+        :param record:
 
         """
         if self.thread is None:
             return 1
-
         if threading.get_ident() == self.thread.ident:
             return 1
         return 0
 
     def emit(self, record):
         """Do something with a logged message
 
-        :param record: 
+        :param record:
 
         """
-        record_dict = {"message": record.getMessage()}
-        for k in ["created", "levelname", "levelno", "lineno", "filename"]:
-            record_dict[k] = getattr(record, k)
-        self.dest.append(record_dict)
-        # FIXME: make sure this doesn't become a memory disaster!
-        # We probably need to check the size of the list...
+        self.dest.append(LogRecordSchema().dump(record))
         # TODO: think about whether any of the keys are security flaws
         # (this is why I don't dump the whole logrecord)
 

src/labthings/apispec/plugins.py

@@ -7,9 +7,9 @@
 
 from .. import fields
 from ..json.schemas import schema_to_json
-from ..schema import build_action_schema
+from ..schema import build_action_schema, EventSchema
 from ..utilities import get_docstring, get_summary, merge
-from ..views import ActionView, PropertyView
+from ..views import ActionView, PropertyView, EventView
 
 
 class ExtendedOpenAPIConverter(OpenAPIConverter):
@@ -19,16 +19,16 @@ class ExtendedOpenAPIConverter(OpenAPIConverter):
 
     def init_attribute_functions(self, *args, **kwargs):
         """
-        :param *args: 
-        :param **kwargs: 
+        :param *args:
+        :param **kwargs:
         """
         OpenAPIConverter.init_attribute_functions(self, *args, **kwargs)
         self.attribute_functions.append(self.jsonschema_type_mapping)
 
     def jsonschema_type_mapping(self, field, **kwargs):
         """
-        :param field: 
-        :param **kwargs: 
+        :param field:
+        :param **kwargs:
         """
         ret = {}
         if hasattr(field, "_jsonschema_type_mapping"):
@@ -215,6 +215,37 @@ def spec_for_action(cls, action):
         d["post"]["responses"].update(action.responses)
         return d
 
+    @classmethod
+    def spec_for_event(cls, event):
+        class_json_schema = schema_to_json(event.schema) if event.schema else None
+        queue_json_schema = schema_to_json(EventSchema(many=True))
+        if class_json_schema:
+            queue_json_schema["properties"]["data"] = class_json_schema
+
+        d = cls.spec_for_interaction(event)
+
+        # Add in Action spec
+        d = merge(
+            d,
+            {
+                "get": {
+                    "responses": {
+                        200: {
+                            "description": "Event queue",
+                            "content": {
+                                "application/json": (
+                                    {"schema": queue_json_schema}
+                                    if queue_json_schema
+                                    else {}
+                                )
+                            },
+                        }
+                    },
+                },
+            },
+        )
+        return d
+
     def operation_helper(self, path, operations, **kwargs):
         """Path helper that allows passing a Flask view function."""
         # rule = self._rule_for_view(interaction.dispatch_request, app=app)
@@ -224,4 +255,6 @@ def operation_helper(self, path, operations, **kwargs):
             ops = self.spec_for_property(interaction)
         elif issubclass(interaction, ActionView):
             ops = self.spec_for_action(interaction)
+        elif issubclass(interaction, EventView):
+            ops = self.spec_for_event(interaction)
         operations.update(ops)

src/labthings/default_views/actions.py

@@ -11,26 +11,38 @@ class ActionQueueView(View):
     """List of all actions from the session"""
 
     def get(self):
-        """ """
+        """Action queue
+        ---
+        description: Queue of most recent actions in the session
+        summary: Queue of most recent actions in the session
+        responses:
+            200:
+                content:
+                    application/json:
+                        schema: ActionSchema
+        """
         return ActionSchema(many=True).dump(current_labthing().actions.threads)
 
 
 class ActionObjectView(View):
     """Manage a particular action.
-    
+
     GET will safely return the current action progress.
     DELETE will cancel the action, if pending or running.
 
 
     """
 
     def get(self, task_id):
-        """Show status of a session task
-        
-        Includes progress and intermediate data.
-
-        :param task_id: 
-
+        """Show the status of an Action
+        ---
+        description: Status of an Action
+        summary: Status of an Action
+        responses:
+            200:
+                content:
+                    application/json:
+                        schema: ActionSchema
         """
         task_dict = current_labthing().actions.to_dict()
 
@@ -43,13 +55,15 @@ def get(self, task_id):
 
     @use_args({"timeout": fields.Int()})
     def delete(self, args, task_id):
-        """Terminate a running task.
-        
-        If the task is finished, deletes its entry.
-
-        :param args: 
-        :param task_id: 
-
+        """Cancel a running Action
+        ---
+        description: Cancel an Action
+        summary: Cancel an Action
+        responses:
+            200:
+                content:
+                    application/json:
+                        schema: ActionSchema
         """
         timeout = args.get("timeout", None)
         task_dict = current_labthing().actions.to_dict()

src/labthings/default_views/events.py

@@ -0,0 +1,9 @@
+from .. import fields
+from ..schema import LogRecordSchema
+from ..views import EventView
+
+
+class LoggingEventView(EventView):
+    """List of latest logging events from the session"""
+
+    schema = LogRecordSchema()

src/labthings/default_views/extensions.py

@@ -11,10 +11,16 @@ class ExtensionList(View):
 
     def get(self):
         """List enabled extensions.
-        
+
         Returns a list of Extension representations, including basic documentation.
         Describes server methods, web views, and other relevant Lab Things metadata.
-
-
+        ---
+        description: Extensions list
+        summary: Extensions list
+        responses:
+            200:
+                content:
+                    application/json:
+                        schema: ExtensionSchema
         """
         return ExtensionSchema(many=True).dump(registered_extensions().values() or [])

src/labthings/default_views/root.py

@@ -6,5 +6,9 @@ class RootView(View):
     """W3C Thing Description"""
 
     def get(self):
-        """ """
+        """Thing Description
+        ---
+        description: Thing Description
+        summary: Thing Description
+        """
         return current_labthing().thing_description.to_dict()