Merge pull request #80 from labthings/better-events

Better events

Author: jtc42

.coveragerc

@@ -1,7 +1,7 @@
 [run]
 branch = True
-source = ./labthings
-omit = .venv/*, labthings/server/wsgi/*, , labthings/server/monkey.py
+source = ./src/labthings
+omit = .venv/*, ./src/labthings/server/wsgi/*, ./src/labthings/server/monkey.py
 concurrency = greenlet
 
 [report]

src/labthings/server/decorators.py

@@ -14,6 +14,7 @@
 from .view import View
 from .find import current_labthing
 from .utilities import unpack
+from .event import PropertyStatusEvent, ActionStatusEvent
 
 from labthings.core.tasks.pool import TaskThread
 from labthings.core.utilities import merge
@@ -122,6 +123,7 @@ def ThingAction(viewcls: View):
     Returns:
         View: View class with Action spec tags
     """
+    # TODO: Handle actionStatus messages
     # Update Views API spec
     tag_spec(viewcls, "actions")
     return viewcls
@@ -180,10 +182,20 @@ def wrapped(*args, **kwargs):
             # Call the update function first to update property value
             original_response = func(*args, **kwargs)
 
-            # Once updated, then notify all subscribers
-            subscribers = getattr(current_labthing(), "subscribers", [])
-            for sub in subscribers:
-                sub.property_notify(viewcls)
+            if hasattr(viewcls, "get_value") and callable(viewcls.get_value):
+                property_value = viewcls().get_value()
+            else:
+                property_value = None
+
+            property_name = getattr(viewcls, "endpoint", None) or getattr(
+                viewcls, "__name__", "unknown"
+            )
+
+            if current_labthing():
+                current_labthing().message(
+                    PropertyStatusEvent(property_name), property_value,
+                )
+
             return original_response
 
         return wrapped

src/labthings/server/event.py

@@ -0,0 +1,44 @@
+import datetime
+
+
+class Event:
+    def __init__(self, name, schema=None):
+        self.name = name
+        self.schema = schema
+
+        self.events = []  # TODO: Make rotating
+
+    def emit(self, data):
+        response = {
+            "messageType": "event",
+            "timestamp": datetime.datetime.utcnow().strftime("%Y-%m-%dT%H:%M:%S.%fZ"),
+            "data": {self.name: data},
+        }  # TODO: Format data with schema
+        self.events.append(response)
+        return response
+
+
+class PropertyStatusEvent:
+    def __init__(self, property_name, schema=None):
+        self.name = property_name
+
+    def emit(self, data):
+        response = {
+            "messageType": "propertyStatus",
+            "timestamp": datetime.datetime.utcnow().strftime("%Y-%m-%dT%H:%M:%S.%fZ"),
+            "data": {self.name: data},
+        }
+        return response
+
+
+class ActionStatusEvent:
+    def __init__(self, action_name, schema=None):
+        self.name = action_name
+
+    def emit(self, data):
+        response = {
+            "messageType": "actionStatus",
+            "timestamp": datetime.datetime.utcnow().strftime("%Y-%m-%dT%H:%M:%S.%fZ"),
+            "data": {self.name: data},
+        }
+        return response

src/labthings/server/labthing.py

@@ -20,6 +20,7 @@
 from .spec.td import ThingDescription
 from .decorators import tag
 from .sockets import Sockets
+from .event import Event
 
 from .view.builder import property_of, action_from
 
@@ -55,6 +56,8 @@ def __init__(
 
         self.extensions = {}
 
+        self.events = {}
+
         self.views = []
         self._property_views = {}
         self._action_views = {}
@@ -163,6 +166,9 @@ def init_app(self, app):
         self.sockets = Sockets(app)
         self._create_base_sockets()
 
+        # Create base events
+        self.add_event("logging")
+
     def _create_base_routes(self):
         # Add root representation
         self.add_view(RootView, "/", endpoint="root")
@@ -319,6 +325,28 @@ def _register_view(self, app, view, *urls, endpoint=None, **kwargs):
             self.thing_description.property(flask_rules, view)
             self._property_views[view.endpoint] = view
 
+    # Event stuff
+    def add_event(self, name, schema=None):
+        # TODO: Handle schema
+        # TODO: Add view for event, returning list of Event.events
+        self.events[name] = Event(name, schema=schema)
+        self.thing_description.event(self.events[name])
+
+    def emit(self, event_type: str, data: dict):
+        """
+        Find a matching event type if one exists, and emit some data to it
+        """
+        event_object = self.events[event_type]
+        self.message(event_object, data)
+
+    def message(self, event: Event, data: dict):
+        """
+        Emit an event object to all subscribers
+        """
+        event_response = event.emit(data)
+        for sub in self.subscribers:
+            sub.emit(event_response)
+
     # Utilities
 
     def url_for(self, view, **values):

src/labthings/server/logging.py

@@ -12,15 +12,8 @@ def emit(self, record):
         log_event = self.rest_format_record(record)
 
         # Broadcast to subscribers
-        subscribers = getattr(current_labthing(), "subscribers", [])
-        for sub in subscribers:
-            sub.event_notify(log_event)
+        if current_labthing():
+            current_labthing().emit("logging", log_event)
 
     def rest_format_record(self, record):
-        data = {
-            "data": str(record.msg),
-            "timestamp": datetime.datetime.utcnow().strftime("%Y-%m-%dT%H:%M:%S.%fZ"),
-        }
-        level_string = record.levelname.lower()
-
-        return {level_string: data}
+        return {"message": str(record.msg), "level": record.levelname.lower()}

src/labthings/server/sockets/base.py

@@ -10,25 +10,9 @@ class SocketSubscriber:
     def __init__(self, ws):
         self.ws = ws
 
-    def property_notify(self, viewcls):
-        if hasattr(viewcls, "get_value") and callable(viewcls.get_value):
-            property_value = viewcls().get_value()
-        else:
-            property_value = None
-
-        property_name = getattr(viewcls, "endpoint", None) or getattr(
-            viewcls, "__name__", "unknown"
-        )
-
-        response = encode_json(
-            {"messageType": "propertyStatus", "data": {property_name: property_value}}
-        )
-
-        self.ws.send(response)
-
-    def event_notify(self, event_dict: dict):
-        response = encode_json({"messageType": "event", "data": event_dict})
-
+    def emit(self, event: dict):
+        response = encode_json(event)
+        # TODO: Logic surrounding if this subscriber is subscribed to the requested event type
         self.ws.send(response)
 
 

src/labthings/server/spec/td.py

@@ -3,6 +3,7 @@
 import weakref
 
 from ..view import View
+from ..event import Event
 
 from .utilities import get_spec, convert_schema, schema_to_json, get_topmost_spec_attr
 from .paths import rule_to_params, rule_to_path
@@ -92,12 +93,17 @@ def to_dict(self):
             "description": current_labthing().description,
             "properties": self.properties,
             "actions": self.actions,
+            # "events": self.events,  # TODO: Enable once properly populated
             "links": self.links,
             # TODO: Add proper security schemes
             "securityDefinitions": {"nosec_sc": {"scheme": "nosec"}},
             "security": ["nosec_sc"],
         }
 
+    def event_to_thing_event(self, event: Event):
+        # TODO: Include event schema
+        return {}
+
     def view_to_thing_property(self, rules: list, view: View):
         prop_urls = [rule_to_path(rule) for rule in rules]
 
@@ -237,3 +243,6 @@ def build_forms_for_view(self, rules: list, view: View, op: list):
             forms.append({"op": op, "href": url, "contentType": content_type})
 
         return forms
+
+    def event(self, event: Event):
+        self.events[event.name] = self.event_to_thing_event(event)

tests/test_server_default_views_socket_handler.py

@@ -7,4 +7,4 @@ def test_socket_handler(thing_ctx, fake_websocket):
         socket_handler(ws)
         # Only responses should be announcing new subscribers
         for response in ws.responses:
-            assert '"data": "Added subscriber' in response
+            assert '"message": "Added subscriber' in response

tests/test_server_event.py

@@ -0,0 +1,29 @@
+from labthings.server import event
+
+
+def test_event():
+    e = event.Event("eventName")
+    event_data = {"key": "value"}
+
+    response = e.emit(event_data)
+    assert response.get("messageType") == "event"
+    assert response.get("data") == {"eventName": event_data}
+    assert response in e.events
+
+
+def test_property_status_event():
+    e = event.PropertyStatusEvent("propertyName")
+    event_data = {"key": "value"}
+
+    response = e.emit(event_data)
+    assert response.get("messageType") == "propertyStatus"
+    assert response.get("data") == {"propertyName": event_data}
+
+
+def test_action_status_event():
+    e = event.ActionStatusEvent("actionName")
+    event_data = {"key": "value"}
+
+    response = e.emit(event_data)
+    assert response.get("messageType") == "actionStatus"
+    assert response.get("data") == {"actionName": event_data}

tests/test_server_sockets.py

@@ -4,39 +4,6 @@
 from flask import Blueprint
 
 
-def test_socket_subscriber_property_notify(view_cls, fake_websocket):
-    setattr(view_cls, "endpoint", "index")
-    ws = fake_websocket("", recieve_once=True)
-    sub = base.SocketSubscriber(ws)
-
-    sub.property_notify(view_cls)
-    assert json.loads(ws.response) == {
-        "messageType": "propertyStatus",
-        "data": {"index": "GET"},
-    }
-
-
-def test_socket_subscriber_property_notify_empty_view(flask_view_cls, fake_websocket):
-    ws = fake_websocket("", recieve_once=True)
-    sub = base.SocketSubscriber(ws)
-
-    sub.property_notify(flask_view_cls)
-    assert json.loads(ws.response) == {
-        "messageType": "propertyStatus",
-        "data": {flask_view_cls.__name__: None},
-    }
-
-
-def test_socket_subscriber_event_notify(fake_websocket):
-    ws = fake_websocket("", recieve_once=True)
-    sub = base.SocketSubscriber(ws)
-
-    data = {"key": "value"}
-
-    sub.event_notify(data)
-    assert json.loads(ws.response) == {"messageType": "event", "data": data}
-
-
 def test_sockets_flask_init(app):
     original_wsgi_app = app.wsgi_app
     socket = gsocket.Sockets(app)