Add framework for future event bus functionality

fix a couple of incorrect schema models

Author: tipatterson-dev

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "oshconnect"
-version = "0.3.0a6"
+version = "0.3.0a7"
 description = "Library for interfacing with OSH, helping guide visualization efforts, and providing a place to store configurations."
 readme = "README.md"
 authors = [

src/oshconnect/eventbus.py

@@ -4,28 +4,196 @@
 #  Author: Ian Patterson
 #  Contact Email: ian@botts-inc.com
 #  =============================================================================
-import collections
-from typing import Any
-from uuid import UUID
-from abc import ABC
 
+from __future__ import annotations
 
-class Event(ABC):
+import datetime
+from abc import ABC, abstractmethod
+from collections import deque
+from dataclasses import dataclass
+from enum import Enum
+from typing import Any, Union
+
+from pydantic import BaseModel, ConfigDict
+
+
+class Event(BaseModel):
+    model_config = ConfigDict(arbitrary_types_allowed=True)
+
+    timestamp: datetime.datetime
+    type: DefaultEventTypes
+    topic: str
+    data: Any
+    producer: Any
+
+    @classmethod
+    def blank_event(cls) -> Event:
+        return cls(
+            timestamp=datetime.datetime.now(),
+            type=DefaultEventTypes.NEW_OBSERVATION,
+            topic="",
+            data=None,
+            producer=None
+        )
+
+
+@dataclass
+class IEventListener(ABC):
     """
-    A base class for events in the event bus system.
+    Interface for event listeners. They may subscribe to specific topics and/or certain event types.
     """
-    id: UUID
-    topic: str
-    payload: Any
+    topics: list[str]
+    types: list[DefaultEventTypes]
+
+    @abstractmethod
+    def handle_events(self, event: Event):
+        pass
+
+
+class EventHandler(object):
+    """
+    Singleton event handler to manage event listeners and publish events.
+    """
+    listeners: list[IEventListener] = []
+    to_add: list[IEventListener] = []
+    to_remove: list[IEventListener] = []
+    event_queue: deque[Event] = deque()
+    publish_lock: bool = False
+
+    def __new__(cls):
+        if not hasattr(cls, "instance"):
+            cls.instance = super(EventHandler, cls).__new__(cls)
+        return cls.instance
+
+    def register_listener(self, listener: IEventListener):
+        if listener not in self.listeners:
+            if not self.publish_lock:
+                self.listeners.append(listener)
+            else:
+                self.to_add.append(listener)
+
+    def unregister_listener(self, listener: IEventListener):
+        if not self.publish_lock:
+            self.listeners.remove(listener)
+        else:
+            self.to_remove.append(listener)
+
+    def publish(self, evt: Event):
+        if self.publish_lock:
+            self.event_queue.append(evt)
+        else:
+            self.publish_lock = True
+
+            try:
+                for listener in self.listeners:
+                    listener.handle_events(evt)
+            except Exception as e:
+                # TODO: handle a more specific error
+                print(f"Error publishing event: {e}")
+            finally:
+                self.publish_lock = False
+                self.commit_changes()
 
-    def __init__(self, id: UUID, topic: str, payload: Any):
-        self.id = id
-        self.topic = topic
-        self.payload = payload
+    def commit_changes(self):
+        self.commit_removes()
+        self.commit_adds()
 
+        while len(self.event_queue) > 0:
+            self.publish(self.event_queue.popleft())
 
-class EventBus(ABC):
+    def commit_adds(self):
+        for listener in self.to_add:
+            self.listeners.append(listener)
+        self.to_add.clear()
+
+    def commit_removes(self):
+        for listener in self.to_remove:
+            self.listeners.remove(listener)
+        self.to_remove.clear()
+
+    def clear_listeners(self):
+        self.listeners.clear()
+        self.to_add.clear()
+        self.to_remove.clear()
+
+    def get_num_listeners(self) -> int:
+        return len(self.listeners)
+
+
+class DefaultEventTypes(Enum):
+    ADD_NODE: str = "add_node"
+    REMOVE_NODE: str = "remove_node"
+    ADD_SYSTEM: str = "add_system"
+    REMOVE_SYSTEM: str = "remove_system"
+    ADD_DATASTREAM: str = "add_datastream"
+    REMOVE_DATASTREAM: str = "remove_datastream"
+    ADD_CONTROLSTREAM: str = "add_controlstream"
+    REMOVE_CONTROLSTREAM: str = "remove_controlstream"
+    NEW_OBSERVATION: str = "new_observation"
+    NEW_COMMAND: str = "new_command"
+    NEW_COMMAND_STATUS: str = "new_command_status"
+
+
+class AtomicEventTypes(Enum):
     """
-    A base class for an event bus system.
+    Defines atomic event types
+
+    Attributes:
+        CREATE (str): Event type for creating a resource within OSHConnect (local, in-app).
+        POST (str): Event type for posting a resource to an external server.
+        GET (str): Event type for retrieving a resource from an external server.
+        MODIFY (str): Event type for modifying a resource within OSHConnect (local, in-app).
+        UPDATE (str): Event type for updating a resource on an external server.
+        REMOVE (str): Event type for removing a resource within OSHConnect (local, in-app).
+        DELETE (str): Event type for deleting a resource from an external server.
     """
-    _deque: collections.deque
+    #
+    CREATE: str = "create"
+    POST: str = "post"
+    GET: str = "get"
+    MODIFY: str = "modify"
+    UPDATE: str = "update"
+    REMOVE: str = "remove"
+    DELETE: str = "delete"
+
+
+class EventBuilder(ABC):
+    _event: Event
+
+    def __init__(self):
+        self._event: Event = Event.blank_event()
+
+    def with_type(self, event_type: DefaultEventTypes) -> EventBuilder:
+        self._event.type = event_type
+        return self
+
+    def with_topic(self, topic: str) -> EventBuilder:
+        self._event.topic = topic
+        return self
+
+    def with_data(self, data: Any) -> EventBuilder:
+        self._event.data = data
+        return self
+
+    def with_producer(self, producer: Any) -> EventBuilder:
+        self._event.producer = producer
+        return self
+
+    def with_timestamp(self, timestamp: datetime.datetime) -> EventBuilder:
+        self._event.timestamp = timestamp
+        return self
+
+    def build(self) -> Event:
+        built = self._event.model_copy(deep=True)
+        self.reset()
+        return built
+
+    def reset(self) -> None:
+        self._event = Event.blank_event()
+
+    @staticmethod
+    def create_topic(base_topic: DefaultEventTypes, resource_id: Union[str, None] = None) -> str:
+        if resource_id:
+            return f"{base_topic.value}/{resource_id}"
+        else:
+            return base_topic.value

src/oshconnect/oshconnectapi.py

@@ -8,6 +8,7 @@
 import json
 from uuid import UUID
 
+from eventbus import EventHandler
 from .csapi4py.default_api_helpers import APIHelper
 from .datastore import DataStore
 from .resource_datamodels import DatastreamResource
@@ -31,6 +32,7 @@ class OSHConnect:
     _tasks: list = []
     _playback_mode: TemporalModes = TemporalModes.REAL_TIME
     _session_manager: SessionManager = None
+    _event_bus: EventHandler = None
 
     def __init__(self, name: str, **kwargs):
         """
@@ -40,6 +42,7 @@ def __init__(self, name: str, **kwargs):
         self._name = name
         logging.info(f"OSHConnect instance {name} created")
         self._session_manager = SessionManager()
+        self._event_bus = EventHandler()
 
     def get_name(self):
         """

src/oshconnect/streamableresource.py

@@ -214,7 +214,7 @@ def get_api_helper(self) -> APIHelper:
 
     # System Management
 
-    def add_system(self, system: System, target_node: Node, insert_resource: bool = False):
+    def add_system(self, system: System, insert_resource: bool = False):
         """
         Add a system to the target node.
         :param system: System object
@@ -224,7 +224,7 @@ def add_system(self, system: System, target_node: Node, insert_resource: bool =
         """
         if insert_resource:
             system.insert_self()
-        target_node.add_new_system(system)
+        self.add_new_system(system)
         self._systems.append(system)
         return system
 

src/oshconnect/swe_components.py

@@ -60,7 +60,7 @@ class DataArraySchema(AnyComponentSchema):
     type: str = "DataArray"
     name: str = Field(...)
     element_count: dict | str | CountSchema = Field(..., serialization_alias='elementCount')  # Should type of Count
-    element_type: SerializeAsAny[list[AnyComponentSchema]] = Field(..., serialization_alias='elementType')
+    element_type: SerializeAsAny[AnyComponentSchema] = Field(..., serialization_alias='elementType')
     encoding: str = Field(...)  # TODO: implement an encodings class
     values: list = Field(None)
 
@@ -113,7 +113,7 @@ class AnySimpleComponentSchema(AnyComponentSchema):
     definition: str = Field(...)
     reference_frame: str = Field(None, serialization_alias='referenceFrame')
     axis_id: str = Field(None, serialization_alias='axisID')
-    quality: Union[list[QuantitySchema], list[QuantityRangeSchema], list[CategorySchema], list[TextSchema]] = Field(
+    quality: list[Union[QuantitySchema, QuantityRangeSchema, CategorySchema, TextSchema]] = Field(
         None)  # TODO: Union[Quantity, QuantityRange, Category, Text]
     nil_values: list = Field(None, serialization_alias='nilValues')
     constraint: Any = Field(None)