convert results to Observation objects

Author: tipatterson-dev

oshconnect/__init__.py

@@ -13,6 +13,7 @@
 import swecommondm as swe_common
 from conSys4Py import APIResourceTypes
 from conSys4Py.core.default_api_helpers import APIHelper
+from conSys4Py.datamodels.observations import ObservationOMJSONInline
 
 from external_models.object_models import System as SystemResource
 
@@ -24,11 +25,33 @@ class Endpoints:
     connected_systems: str = f"{root}/api"
 
 
-
-
 class TemporalModes(Enum):
     REAL_TIME = 0
     ARCHIVE = 1
     BATCH = 2
     RT_SYNC = 3
     ARCHIVE_SYNC = 4
+
+
+class Utilities:
+
+    @staticmethod
+    def convert_auth_to_base64(username: str, password: str) -> str:
+        return base64.b64encode(f"{username}:{password}".encode()).decode()
+
+class StreamQueue:
+    _queue: list[ObservationOMJSONInline]
+    def __init__(self):
+        self._queue = []
+
+    def push(self, item):
+        self._queue.append(item)
+
+    def pop(self):
+        return self._queue.pop(0)
+
+    def peek(self):
+        return self._queue[0]
+
+    def is_empty(self):
+        return len(self._queue) == 0

oshconnect/datamodels/datamodels.py

@@ -21,6 +21,7 @@
 @dataclass(kw_only=True)
 class Node:
     _id: str
+    protocol: str
     address: str
     port: int
     endpoints: Endpoints
@@ -29,15 +30,16 @@ class Node:
     _api_helper: APIHelper
     _system_ids: list[uuid] = field(default_factory=list)
 
-    def __init__(self, address: str, port: int, username: str = None, password: str = None, **kwargs: dict):
+    def __init__(self, protocol: str, address: str, port: int, username: str = None, password: str = None, **kwargs: dict):
         self._id = f'node-{uuid.uuid4()}'
+        self.protocol = protocol
         self.address = address
         self.port = port
         self.is_secure = username is not None and password is not None
         if self.is_secure:
             self.add_basicauth(username, password)
         self.endpoints = Endpoints()
-        self._api_helper = APIHelper(server_url=f'{self.address}:{self.port}',
+        self._api_helper = APIHelper(server_url=f'{self.protocol}://{self.address}:{self.port}',
                                      api_root=self.endpoints.connected_systems, username=username, password=password)
         if self.is_secure:
             self._api_helper.user_auth = True
@@ -106,16 +108,17 @@ def __init__(self, name: str, label: str, **kwargs):
     def update_parent_node(self, node: Node):
         self._parent_node = node
 
+    def get_parent_node(self) -> Node:
+        return self._parent_node
+
     def discover_datastreams(self):
         res = self._parent_node.get_api_helper().retrieve_resource(APIResourceTypes.DATASTREAM, req_headers={})
         datastream_json = res.json()['items']
-        print(f'Result of datastream discovery: {datastream_json}')
         datastreams = []
         for ds in datastream_json:
             datastream_objs = DatastreamResource.model_validate(ds)
             datastreams.append(datastream_objs)
 
-        print(f'Found datastreams: {datastreams}')
         return datastreams
 
     @staticmethod
@@ -132,10 +135,15 @@ def from_system_resource(system_resource: SystemResource):
 
 
 class Datastream:
+    should_poll: bool
+    _datastream_resource: DatastreamResource
 
-    def __init__(self):
+    def __init__(self, datastream_resource: DatastreamResource):
         pass
 
+    def get_id(self):
+        return self._datastream_resource.ds_id
+
 
 class ControlChannel:
 

oshconnect/datasource/datasource.py

@@ -3,19 +3,42 @@
 #  Author: Ian Patterson <ian@botts-inc.com>
 #
 #  Contact Email: ian@botts-inc.com
-
+import asyncio
 from uuid import uuid4
 
+import websockets
+
+from external_models.object_models import DatastreamResource
+from oshconnect import Utilities
+from oshconnect.datamodels.datamodels import System
 from oshconnect.datasource import Mode
 
 
 class DataSource:
+    """
+    DataSource: represents the active connection of a datastream object
+    """
 
-    def __init__(self, name: str, mode: Mode, properties: dict):
+    def __init__(self, name: str, mode: str, properties: dict, datastream: DatastreamResource, parent_system: System):
+        self._status = None
         self._id = f'datasource-{uuid4()}'
         self.name = name
         self.mode = mode
         self.properties = properties
+        self._datastream = datastream
+        self._websocket = None
+        self._parent_system = parent_system
+        self._url = None
+        if mode == "websocket":
+            self._url = (f'ws://{self._parent_system.get_parent_node().get_address()}:'
+                         f'{self._parent_system.get_parent_node().get_port()}'
+                         f'/sensorhub/api/datastreams/{self._datastream.ds_id}'
+                         f'/observations?f=application%2Fjson')
+        self._auth = None
+        self._extra_headers = None
+        if self._parent_system.get_parent_node().is_secure:
+            self._auth = self._parent_system.get_parent_node().get_decoded_auth()
+            self._extra_headers = {'Authorization': f'Basic {self._auth}'}
 
     def get_id(self) -> str:
         return self._id
@@ -40,20 +63,26 @@ def update_properties(self, properties: dict):
     def initialize(self):
         pass
 
-    def connect(self):
-        pass
+    async def connect(self):
+        if self.mode == "websocket":
+            self._websocket = await websockets.connect(self._url, extra_headers=self._extra_headers)
+            self._status = "connected"
+            return self._websocket
 
-    def disconnect(self):
-        pass
 
-    def reset(self):
-        pass
+def disconnect(self):
+    pass
+
+
+def reset(self):
+    pass
+
 
-    def get_status(self):
-        return self.status
+def get_status(self):
+    return self.status
 
 
-class DatasourceHandler:
+class DataSourceHandler:
     datasource_map: dict[str, DataSource]
 
     def __init__(self):
@@ -70,12 +99,13 @@ def initialize_all(self):
         # list comp is faster than for loop
         [ds.initialize() for ds in self.datasource_map.values()]
 
-    def connect_ds(self, datasource_id: str):
+    async def connect_ds(self, datasource_id: str):
         ds = self.datasource_map.get(datasource_id)
-        ds.connect()
+        await ds.connect()
 
-    def connect_all(self):
-        [ds.connect() for ds in self.datasource_map.values()]
+    async def connect_all(self):
+        results = await asyncio.gather(*(ds.connect() for ds in self.datasource_map.values()))
+        return results
 
     def disconnect_ds(self, datasource_id: str):
         ds = self.datasource_map.get(datasource_id)

oshconnect/oshconnect.py

@@ -4,33 +4,39 @@
 #  Author:  Ian Patterson
 #  Contact email:  ian@botts-inc.com
 #   ==============================================================================
+import asyncio
+import json
 
 from conSys4Py.core.default_api_helpers import APIHelper
+from conSys4Py.datamodels.observations import ObservationOMJSONInline
 
 from oshconnect.datamodels.datamodels import Node, System
-from oshconnect.datasource.datasource import DataSource
+from oshconnect.datasource.datasource import DataSource, DataSourceHandler
 from oshconnect.datastore.datastore import DataStore
 from oshconnect.styling.styling import Styling
 from oshconnect.timemanagement.timemanagement import TimeManagement
 
 
 class OSHConnect:
     _name: str = None
-    datasource: DataSource = None
+    # datasource: DataSource = None
     datastore: DataStore = None
     styling: Styling = None
     timestream: TimeManagement = None
     _nodes: list[Node] = []
     _systems: list[System] = []
     _cs_api_builder: APIHelper = None
+    _datasource_handler: DataSourceHandler = None
     _datafeeds: list[DataSource] = []
     _datataskers: list[DataStore] = []
     _datagroups: list = []
+    _tasks: list = []
 
     def __init__(self, name: str, **kwargs):
         self._name = name
         if 'nodes' in kwargs:
             self._nodes = kwargs['nodes']
+        self._datasource_handler = DataSourceHandler()
 
     def get_name(self):
         return self._name
@@ -74,8 +80,18 @@ def select_temporal_mode(self, mode: str):
         """
         pass
 
-    def playback_streams(self, streams: list):
-        pass
+    async def playback_streams(self, stream_ids: list = None):
+        if stream_ids is None:
+            clients = await self._datasource_handler.connect_all()
+            for client in clients:
+                task = asyncio.create_task(self._handle_datastream_client(client))
+                self._tasks.append(task)
+        else:
+            for stream_id in stream_ids:
+                clients = await self._datasource_handler.connect_ds(stream_id)
+                for client in clients:
+                    msg = await client.recv()
+                    print(msg)
 
     def visualize_streams(self, streams: list):
         pass
@@ -85,10 +101,15 @@ def get_visualization_recommendations(self, streams: list):
         pass
 
     def discover_datastreams(self):
+        # NOTE: This will need to check to prevent dupes in the future
         for system in self._systems:
             res_datastreams = system.discover_datastreams()
-            print(f'Datastreams found: {res_datastreams}')
-            self._datafeeds.extend(res_datastreams)
+            # create DataSource(s)
+            new_datasource = [
+                DataSource(name=ds.name, mode="websocket", properties={}, datastream=ds, parent_system=system) for ds in
+                res_datastreams]
+            self._datafeeds.extend(new_datasource)
+            list(map(self._datasource_handler.add_datasource, new_datasource))
 
     def discover_systems(self, nodes: list[str] = None):
         search_nodes = self._nodes
@@ -107,3 +128,12 @@ def authenticate_user(self, user: dict):
 
     def synchronize_streams(self, systems: list):
         pass
+
+    async def _handle_datastream_client(self, client):
+        try:
+            async for msg in client:
+                msg_dict = json.loads(msg.decode('utf-8'))
+                obs = ObservationOMJSONInline.model_validate(msg_dict)
+
+        except Exception as e:
+            print(f"An error occurred while reading from websocket: {e}")

pyproject.toml

@@ -11,6 +11,7 @@ consys4py = { git = "https://github.com/Botts-Innovative-Research/CSAPI4Py.git",
 swecommondm = { git = "https://github.com/ChainReaction31/SWECommonDMPython.git", branch = "master" }
 pydantic = "^2.7.4"
 shapely = "^2.0.4"
+websockets = "^12.0"
 
 [tool.poetry.dev-dependencies]
 pytest = "^8.2.2"

tests/test_oshconnect.py

@@ -6,6 +6,8 @@
 #   ==============================================================================
 
 import pytest
+import requests
+import websockets
 
 from oshconnect.oshconnect import OSHConnect
 from oshconnect.datamodels.datamodels import Node, System
@@ -45,3 +47,13 @@ def test_oshconnect_find_datastreams(self):
         app.discover_datastreams()
         assert len(app._datafeeds) > 0
 
+    async def test_obs_ws_stream(self):
+        ds_url = ("ws://localhost:8585/sensorhub/api/datastreams/e07n5sbjqvalm/observations?f=application%2Fjson"
+                  "&resultTime=latest/2025-06-18T15:46:32Z")
+
+        # stream = requests.get(ds_url, stream=True, auth=('admin', 'admin'))
+        async with websockets.connect(ds_url, extra_headers={'Authorization': 'Basic YWRtaW46YWRtaW4='}) as stream:
+            async for message in stream:
+                print(message)
+
+