Add first pass for batched observations, to be iterated on

Author: tipatterson-dev

external_models/__init__.py

@@ -58,7 +58,11 @@ def valid_time_period(cls, data) -> Any:
         if isinstance(data, list):
             if data[0] > data[1]:
                 raise ValueError("Time period start must be before end.")
-        return {"start": data[0], "end": data[1]}
+            return {"start": data[0], "end": data[1]}
+        elif isinstance(data, dict):
+            if data['start'] > data['end']:
+                raise ValueError("Time period start must be before end.")
+            return data
 
     def __repr__(self):
         return f'{[self.start, self.end]}'

oshconnect/datasource/datasource.py

@@ -4,19 +4,20 @@
 #
 #  Contact Email: ian@botts-inc.com
 from __future__ import annotations
+
 import asyncio
 import json
 from uuid import uuid4
 
+import requests
 import websockets
+from conSys4Py import APIResourceTypes
 from conSys4Py.datamodels.observations import ObservationOMJSONInline
 
 from external_models import TimePeriod
 from external_models.object_models import DatastreamResource
-from oshconnect import Utilities
-from oshconnect.datamodels.datamodels import System
-from oshconnect.datasource import Mode
 from oshconnect import TemporalModes
+from oshconnect.datamodels.datamodels import System
 
 
 class DataSource:
@@ -55,10 +56,6 @@ def terminate_process(self):
     def subscribe(self):
         pass
 
-    def update_properties(self, properties: dict):
-        # TODO: need to stop in progress sub-processes and restart
-        self.properties = properties
-
     def set_mode(self, mode: TemporalModes):
         self._playback_mode = mode
         self.generate_url()
@@ -75,11 +72,13 @@ async def connect(self):
             self._status = "connected"
             return self._websocket
         elif self._playback_mode == TemporalModes.ARCHIVE:
+            self._websocket = await websockets.connect(self._url, extra_headers=self._extra_headers)
             self._status = "connected"
-            return "Playback mode is not yet implemented."
+            return self._websocket
         elif self._playback_mode == TemporalModes.BATCH:
+            self._websocket = await websockets.connect(self._url, extra_headers=self._extra_headers)
             self._status = "connected"
-            return "Live-batch mode is not yet implemented."
+            return self._websocket
 
     def disconnect(self):
         self._websocket.close()
@@ -91,13 +90,17 @@ def reset(self):
     def get_status(self):
         return self._status
 
+    def get_parent_system(self):
+        return self._parent_system
+
     def get_ws_client(self):
         return self._websocket
 
     def is_within_timeperiod(self, timeperiod: TimePeriod):
         return timeperiod.does_timeperiod_overlap(self._datastream.valid_time)
 
     def generate_url(self):
+        # TODO: need to specify secure vs insecure protocols
         if self._playback_mode == TemporalModes.REAL_TIME:
             self._url = (f'ws://{self._parent_system.get_parent_node().get_address()}:'
                          f'{self._parent_system.get_parent_node().get_port()}'
@@ -109,6 +112,13 @@ def generate_url(self):
                          f'/sensorhub/api/datastreams/{self._datastream.ds_id}'
                          f'/observations?f=application%2Fjson&resultTime={self._datastream.valid_time.start}/'
                          f'{self._datastream.valid_time.end}')
+        elif self._playback_mode == TemporalModes.BATCH:
+            # TODO: need to allow for batch counts selection through DS Handler or TimeManager
+            self._url = (f'wss://{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&resultTime={self._datastream.valid_time.start}/'
+                         f'{self._datastream.valid_time.end}')
         else:
             raise ValueError("Playback mode not set. Cannot generate URL for DataSource.")
 
@@ -160,10 +170,15 @@ async def connect_all(self, timeperiod: TimePeriod):
         else:
             ds_matches = self.datasource_map.values()
 
-        [(ds, await ds.connect()) for ds in ds_matches]
-        for ds in ds_matches:
-            task = asyncio.create_task(self._handle_datastream_client(ds))
-            # return task
+        if self._playback_mode == TemporalModes.REAL_TIME:
+            [(ds, await ds.connect()) for ds in ds_matches]
+            for ds in ds_matches:
+                task = asyncio.create_task(self._handle_datastream_client(ds))
+        elif self._playback_mode == TemporalModes.ARCHIVE:
+            pass
+        elif self._playback_mode == TemporalModes.BATCH:
+            for ds in ds_matches:
+                task = asyncio.create_task(self.handle_http_batching(ds))
 
     def disconnect_ds(self, datasource_id: str):
         ds = self.datasource_map.get(datasource_id)
@@ -183,6 +198,33 @@ async def _handle_datastream_client(self, datasource: DataSource):
         except Exception as e:
             print(f"An error occurred while reading from websocket: {e}")
 
+    async def handle_http_batching(self, datasource: DataSource, offset: int = None, query_params: dict = None,
+                                   next_link: str = None):
+        # access api_helper
+        api_helper = datasource.get_parent_system().get_parent_node().get_api_helper()
+        # needs to create a new call to make a request to the server if there is a link to a next page
+        resp = None
+        if next_link is None:
+            resp = api_helper.retrieve_resource(APIResourceTypes.OBSERVATION,
+                                                parent_res_id=datasource._datastream.ds_id,
+                                                req_headers={'Content-Type': 'application/json'})
+        elif next_link is not None:
+            resp = requests.get(next_link, auth=(datasource._parent_system.get_parent_node()._api_helper.username,
+                                                 datasource._parent_system.get_parent_node()._api_helper.password))
+        results = resp.json()
+        if 'links' in results:
+            for link in results['links']:
+                if link['rel'] == 'next':
+                    new_offset = link['href'].split('=')[-1]
+                    asyncio.create_task(self.handle_http_batching(datasource, next_link=link['href']))
+
+        # print(results)
+        for obs in results['items']:
+            obs_obj = ObservationOMJSONInline.model_validate(obs)
+            msg_wrapper = MessageWrapper(datasource=datasource, message=obs_obj)
+            self._message_list.add_message(msg_wrapper)
+        return resp.json()
+
 
 class MessageHandler:
     _message_list: list[MessageWrapper]

oshconnect/oshconnect.py

@@ -10,6 +10,7 @@
 from conSys4Py.core.default_api_helpers import APIHelper
 from conSys4Py.datamodels.observations import ObservationOMJSONInline
 
+from external_models import TimePeriod
 from oshconnect import TemporalModes
 from oshconnect.datamodels.datamodels import Node, System
 from oshconnect.datasource.datasource import DataSource, DataSourceHandler
@@ -46,6 +47,7 @@ def __init__(self, name: str, **kwargs):
         self._datasource_handler = DataSourceHandler()
         if 'playback_mode' in kwargs:
             self._playback_mode = kwargs['playback_mode']
+            self._datasource_handler.set_playback_mode(self._playback_mode)
 
     def get_name(self):
         return self._name
@@ -91,7 +93,7 @@ def select_temporal_mode(self, mode: str):
 
     async def playback_streams(self, stream_ids: list = None):
         if stream_ids is None:
-            await self._datasource_handler.connect_all(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)
@@ -134,3 +136,7 @@ def synchronize_streams(self, systems: list):
 
     def set_playback_mode(self, mode: TemporalModes):
         self._datasource_handler.set_playback_mode(mode)
+
+    def set_timeperiod(self, start_time: str, end_time: str):
+        tp = TimePeriod(start=start_time, end=end_time)
+        self.timestream = TimeManagement(time_range=tp)

oshconnect/timemanagement/timemanagement.py

@@ -4,13 +4,21 @@
 
 from swecommondm import component_implementations as swe_common_components
 
+from external_models import TimePeriod
 from oshconnect.timemanagement import Time, TemporalMode
 
 
 class TimeManagement:
-    time = swe_common_components.TimeComponent(label="Time", name="default_time_component")
+    time_range: TimePeriod
     time_controller: TimeController
 
+    def __init__(self, time_range: TimePeriod):
+        self.time_range = time_range
+        self.time_controller = TimeController()
+
+    def get_time_range(self):
+        return self.time_range
+
 
 class TimeController:
     _instance = None