- use MQTT as default communication protocol

- update some URl construction methods

Author: tipatterson-dev

src/oshconnect/csapi4py/default_api_helpers.py

@@ -16,6 +16,8 @@
 from .constants import APIResourceTypes, EncodingSchema, APITerms
 
 
+# TODO: rework to make the first resource in the endpoint the primary key for URL construction, currently, the implementation is a bit on the confusing side with what is being generated and why.
+
 def determine_parent_type(res_type: APIResourceTypes):
     match res_type:
         case APIResourceTypes.SYSTEM:
@@ -84,6 +86,7 @@ class APIHelper(ABC):
     server_url: str = None
     port: int = None
     protocol: str = "https"
+    server_root: str = "sensorhub"
     api_root: str = "api"
     username: str = None
     password: str = None
@@ -178,52 +181,54 @@ def delete_resource(self, res_type: APIResourceTypes, res_id: str, parent_res_id
         return api_request.make_request()
 
     # Helpers
-    def resource_url_resolver(self, res_type: APIResourceTypes, res_id: str = None, parent_res_id: str = None,
+    def resource_url_resolver(self, subresource_type: APIResourceTypes, subresource_id: str = None,
+                              resource_id: str = None,
                               from_collection: bool = False):
         """
         Helper to generate a URL endpoint for a given resource type and id by matching the resource type to an
         appropriate parent endpoint and inserting the resource ids as necessary.
-        :param res_type:
-        :param res_id:
-        :param parent_res_id:
+        :param subresource_type:
+        :param subresource_id:
+        :param resource_id:
         :param from_collection:
         :return:
         """
-        if res_type is None:
+        if subresource_type is None:
             raise ValueError('Resource type must contain a valid APIResourceType')
-        if res_type is APIResourceTypes.COLLECTION and from_collection:
+        if subresource_type is APIResourceTypes.COLLECTION and from_collection:
             raise ValueError('Collections are not sub-resources of other collections')
 
         parent_type = None
-        if parent_res_id and not from_collection:
-            parent_type = determine_parent_type(res_type)
-        elif parent_res_id and from_collection:
+        if resource_id and not from_collection:
+            parent_type = determine_parent_type(subresource_type)
+        elif resource_id and from_collection:
             parent_type = APIResourceTypes.COLLECTION
 
-        return self.construct_url(parent_type, res_id, res_type, parent_res_id)
+        return self.construct_url(parent_type, subresource_id, subresource_type, resource_id)
 
-    def construct_url(self, parent_type, res_id, res_type, parent_res_id, for_socket: bool = False):
+    def construct_url(self, resource_type: APIResourceTypes, subresource_id, subresource_type, resource_id,
+                      for_socket: bool = False):
         """
         Constructs an API endpoint url from the given parameters
-        :param parent_type:
-        :param res_id:
-        :param res_type:
-        :param parent_res_id:
+        :param resource_type:
+        :param subresource_id:
+        :param subresource_type:
+        :param resource_id:
         :param for_socket: If true, will construct a WebSocket URL (ws:// or wss://) instead of HTTP/HTTPS.
         :return:
         """
         # TODO: Test for less common cases to ensure that the URL is being constructed correctly
         base_url = self.get_api_root_url(socket=for_socket)
 
-        resource_endpoint = resource_type_to_endpoint(res_type, parent_type)
+        resource_endpoint = resource_type_to_endpoint(subresource_type, resource_type)
         url = f'{base_url}/{resource_endpoint}'
 
-        if parent_type:
-            parent_endpoint = resource_type_to_endpoint(parent_type)
-            url = f'{base_url}/{parent_endpoint}/{parent_res_id}/{resource_endpoint}'
+        if resource_type:
+            parent_endpoint = resource_type_to_endpoint(resource_type)
+            url = f'{base_url}/{parent_endpoint}/{resource_id}/{resource_endpoint}'
 
-        if res_id:
-            url = f'{url}/{res_id}'
+        if subresource_id:
+            url = f'{url}/{subresource_id}'
 
         return url
 
@@ -244,13 +249,27 @@ def get_api_root_url(self, socket: bool = False):
         :param socket: If true, will return a WebSocket URL (ws:// or wss://) instead of HTTP/HTTPS.
         :return:
         """
-        return f'{self.get_base_url(socket=socket)}/{self.api_root}'
+        return f'{self.get_base_url(socket=socket)}/{self.server_root}/{self.api_root}'
 
     def set_protocol(self, protocol: str):
         if protocol not in ['http', 'https', 'ws', 'wss']:
             raise ValueError('Protocol must be either "http" or "https"')
         self.protocol = protocol
 
+    def get_mqtt_topic(self, resource_type, subresource_type, resource_id: str,
+                       for_socket: bool = False):
+        """
+        Returns the MQTT topic for the resource type, if applicable.
+        :return:
+        """
+        resource_endpoint = f'/{resource_type_to_endpoint(subresource_type, resource_type)}'
+        parent_endpoint = "" if resource_type is None else f'/{resource_type_to_endpoint(resource_type)}'
+        parent_id = "" if resource_id is None else f'/{resource_id}'
+        topic_locator = f'/{self.api_root}{parent_endpoint}{parent_id}{resource_endpoint}'
+        print(f'MQTT Topic: {topic_locator}')
+
+        return topic_locator
+
 
 @dataclass(kw_only=True)
 class ResponseParserHelper:

src/oshconnect/csapi4py/mqtt.py

@@ -173,7 +173,7 @@ def start(self):
 
     def stop(self):
         """
-        Stop the MQTT client.\
+        Stop the MQTT client.
 
         :return:
         """

src/oshconnect/encoding.py

@@ -1,7 +1,8 @@
-from pydantic import BaseModel, Field
+from pydantic import BaseModel, Field, ConfigDict
 
 
 class Encoding(BaseModel):
+    model_config = ConfigDict(populate_by_name=True)
     id: str = Field(None)
     type: str = Field(...)
     vector_as_arrays: bool = Field(False, alias='vectorAsArrays')

src/oshconnect/eventbus.py

@@ -0,0 +1,15 @@
+#  =============================================================================
+#  Copyright (c) 2025 Botts Innovative Research Inc.
+#  Date: 2025/10/6
+#  Author: Ian Patterson
+#  Contact Email: ian@botts-inc.com
+#  =============================================================================
+import collections
+from abc import ABC
+
+
+class EventBus(ABC):
+    """
+    A base class for an event bus system.
+    """
+    _deque: collections.deque

src/oshconnect/oshconnectapi.py

@@ -6,11 +6,12 @@
 #   ==============================================================================
 import logging
 import shelve
+from uuid import UUID
 
 from .csapi4py.default_api_helpers import APIHelper
-from .resource_datamodels import DatastreamResource
-from .datasource import DataStream, MessageWrapper
+from .datasource import MessageWrapper
 from .datastore import DataStore
+from .resource_datamodels import DatastreamResource
 from .streamableresource import Node, System, SessionManager, Datastream
 from .styling import Styling
 from .timemanagement import TemporalModes, TimeManagement, TimePeriod
@@ -25,7 +26,7 @@ class OSHConnect:
     _systems: list[System] = []
     _cs_api_builder: APIHelper = None
     # _datasource_handler: DataStreamHandler = None
-    _datastreams: list[DataStream] = []
+    _datastreams: list[Datastream] = []
     _datataskers: list[DataStore] = []
     _datagroups: list = []
     _tasks: list = []
@@ -38,15 +39,6 @@ def __init__(self, name: str, **kwargs):
         :param kwargs:
         """
         self._name = name
-        # if 'nodes' in kwargs:
-        #     self._nodes = kwargs['nodes']
-        #     self._playback_mode = kwargs['playback_mode']
-        #     self._datasource_handler.set_playback_mode(self._playback_mode)
-        # self._datasource_handler = DataStreamHandler()
-        # if 'playback_mode' in kwargs:
-        #     self._playback_mode = kwargs['playback_mode']
-        #     self._datasource_handler.set_playback_mode(self._playback_mode)
-
         logging.info(f"OSHConnect instance {name} created")
         self._session_manager = SessionManager()
 
@@ -116,48 +108,20 @@ def select_temporal_mode(self, mode: str):
         """
         pass
 
-    async def playback_streams(self, stream_ids: list = None):
-        """
-        Begins playback of the datastreams that have been connected to the app. The method of playback is determined
-        by the temporal mode that has been set.
-        :param stream_ids:
-        :return:
-        """
-        if stream_ids is None:
-            await self._datasource_handler.connect_all(
-                self.timestream.get_time_range())
-        else:
-            for stream_id in stream_ids:
-                await self._datasource_handler.connect_ds(stream_id)
-
     def visualize_streams(self, streams: list):
         pass
 
     # Second Level Use Cases
     def get_visualization_recommendations(self, streams: list):
         pass
 
-    def dep_discover_datastreams(self):
-        """
-        Discover datastreams of the current systems of the OSHConnect instance and create objects for them that are
-        stored in the DataSourceHandler.
-        :return:
-        """
-        # NOTE: This will need to check to prevent dupes in the future
-        for system in self._systems:
-            res_datastreams = system.discover_datastreams()
-            # create DataSource(s)
-            new_datasource = [
-                DataStream(name=ds.name, datastream=ds, parent_system=system.get_system_resource())
-                for ds in
-                res_datastreams]
-            self._datastreams.extend(new_datasource)
-            list(map(self._datasource_handler.add_datasource, new_datasource))
-
     def discover_datastreams(self):
         for system in self._systems:
             res_datastreams = system.discover_datastreams()
-            datastreams = list(map(lambda ds: Datastream(parent_node=system.get_parent_node(), id=ds.ds_id, datastream_resource=ds), res_datastreams))
+            datastreams = list(
+                map(lambda ds: Datastream(parent_node=system.get_parent_node(), id=ds.ds_id, datastream_resource=ds),
+                    res_datastreams))
+            datastreams = [ds.set_parent_resource_id(system.get_underlying_resource().system_id) for ds in datastreams]
             self._datastreams.extend(datastreams)
 
     def discover_systems(self, nodes: list[str] = None):
@@ -279,13 +243,54 @@ def remove_system(self, system_id: str):
         pass
 
     # DataStream Helpers
-    def get_datastreams(self) -> list[DataStream]:
+    def get_datastreams(self) -> list[Datastream]:
         return self._datastreams
 
+    def get_datastream_ids(self) -> list[UUID]:
+        return [ds.get_internal_id() for ds in self._datastreams]
+
     def connect_session_streams(self, session_id: str):
         """
         Connects all datastreams that are associated with the given session ID.
         :param session_id:
         :return:
         """
         self._session_manager.start_session_streams(session_id)
+
+    def get_resource_group(self, resource_ids: list[UUID]) -> tuple[list[System], list[Datastream]]:
+        """
+        Get a group of resources by their IDs. Can be any mix of systems, datastreams, and controlstreams.
+        :param resource_ids: list of resource IDs (internal UUID)
+        """
+        systems = [system for system in self._systems if system.get_internal_id() in resource_ids]
+        datastreams = [ds for ds in self._datastreams if ds.get_internal_id() in resource_ids]
+        return systems, datastreams
+
+    def initialize_resource_groups(self, resource_ids: list = None):
+        """
+        Initializes the datastreams that are specified.
+        """
+        systems, datastreams = self.get_resource_group(resource_ids)
+
+        if systems:
+            for system in systems:
+                system.initialize()
+        if datastreams:
+            for ds in datastreams:
+                ds.initialize()
+
+    def start_datastreams(self, dsid_list: list = None):
+        """
+        Starts the datastreams that are specified.
+        """
+        datastreams = self.get_resource_group(dsid_list)[1]
+        for ds in datastreams:
+            ds.start()
+
+    def start_systems(self, sysid_list: list = None):
+        """
+        Starts the systems that are specified.
+        """
+        systems = self.get_resource_group(sysid_list)[0]
+        for system in systems:
+            system.start()

src/oshconnect/resource_datamodels.py

@@ -17,7 +17,7 @@
 
 
 class BoundingBox(BaseModel):
-    model_config = ConfigDict(arbitrary_types_allowed=True)
+    model_config = ConfigDict(arbitrary_types_allowed=True, populate_by_name=True)
 
     lower_left_corner: Point = Field(..., description="The lower left corner of the bounding box.")
     upper_right_corner: Point = Field(..., description="The upper right corner of the bounding box.")
@@ -88,7 +88,7 @@ class ProcessMethod:
 
 
 class BaseResource(BaseModel):
-    model_config = ConfigDict(arbitrary_types_allowed=True)
+    model_config = ConfigDict(arbitrary_types_allowed=True, populate_by_name=True)
 
     id: str = Field(..., alias="id")
     name: str = Field(...)
@@ -98,7 +98,7 @@ class BaseResource(BaseModel):
 
 
 class SystemResource(BaseModel):
-    model_config = ConfigDict(arbitrary_types_allowed=True)
+    model_config = ConfigDict(arbitrary_types_allowed=True, populate_by_name=True)
 
     feature_type: str = Field(None, alias="type")
     system_id: str = Field(None, alias="id")
@@ -139,7 +139,7 @@ class DatastreamResource(BaseModel):
     that, depending on the format of the  request, the fields needed may differ. There may be derived models in a later
     release that will have different sets of required fields to ease the validation process for users.
     """
-    # model_config = ConfigDict(populate_by_name=True)
+    model_config = ConfigDict(populate_by_name=True)
 
     ds_id: str = Field(..., alias="id")
     name: str = Field(...)
@@ -176,6 +176,7 @@ def handle_aliases(cls, values):
 
 
 class ObservationResource(BaseModel):
+    model_config = ConfigDict(populate_by_name=True)
     sampling_feature_id: str = Field(None, alias="samplingFeature@Id")
     procedure_link: Link = Field(None, alias="procedure@link")
     phenomenon_time: DateTimeSchema = Field(None, alias="phenomenonTime")
@@ -186,6 +187,7 @@ class ObservationResource(BaseModel):
 
 
 class ControlStreamResource(BaseModel):
+    model_config = ConfigDict(populate_by_name=True)
     name: str = Field(...)
     description: str = Field(None)
     valid_time: TimePeriod = Field(..., alias="validTime")