Add playback mode to OShConnect, DatasourceHandler, and DataSource

Author: tipatterson-dev

external_models/__init__.py

@@ -4,10 +4,11 @@
 #   Author:  Ian Patterson
 #   Contact Email:  ian@botts-inc.com
 #   ==============================================================================
+from __future__ import annotations
+
 from typing import Any
 
 from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator
-from pydantic._internal import _repr
 from shapely import Point
 from typing_extensions import Self
 
@@ -62,6 +63,12 @@ def valid_time_period(cls, data) -> Any:
     def __repr__(self):
         return f'{[self.start, self.end]}'
 
+    def does_timeperiod_overlap(self, checked_timeperiod: TimePeriod) -> bool:
+        if checked_timeperiod.start < self.end and checked_timeperiod.end > self.start:
+            return True
+        else:
+            return False
+
 
 
 class SecurityConstraints:

oshconnect/__init__.py

@@ -6,16 +6,8 @@
 #   ==============================================================================
 
 import base64
-import uuid
-from abc import ABC
-from dataclasses import dataclass, field
+from dataclasses import dataclass
 from enum import Enum
-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
 
 
 @dataclass(kw_only=True)
@@ -26,32 +18,15 @@ class Endpoints:
 
 
 class TemporalModes(Enum):
-    REAL_TIME = 0
-    ARCHIVE = 1
-    BATCH = 2
-    RT_SYNC = 3
-    ARCHIVE_SYNC = 4
+    REAL_TIME = "realtime"
+    ARCHIVE = "archive"
+    BATCH = "batch"
+    RT_SYNC = "realtimesync"
+    ARCHIVE_SYNC = "archivesync"
 
 
 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/datasource/datasource.py

@@ -11,32 +11,28 @@
 import websockets
 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
 
 
 class DataSource:
     """
     DataSource: represents the active connection of a datastream object
     """
 
-    def __init__(self, name: str, mode: str, properties: dict, datastream: DatastreamResource, parent_system: System):
+    def __init__(self, name: str, 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._playback_mode = None
         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:
@@ -63,8 +59,9 @@ def update_properties(self, properties: dict):
         # TODO: need to stop in progress sub-processes and restart
         self.properties = properties
 
-    def set_mode(self, mode: str):
-        self.mode = mode
+    def set_mode(self, mode: TemporalModes):
+        self._playback_mode = mode
+        self.generate_url()
 
     def initialize(self):
         if self._websocket.is_open():
@@ -73,14 +70,14 @@ def initialize(self):
         self._status = "initialized"
 
     async def connect(self):
-        if self.mode == "websocket":
+        if self._playback_mode == TemporalModes.REAL_TIME:
             self._websocket = await websockets.connect(self._url, extra_headers=self._extra_headers)
             self._status = "connected"
             return self._websocket
-        elif self.mode == "playback":
+        elif self._playback_mode == TemporalModes.ARCHIVE:
             self._status = "connected"
             return "Playback mode is not yet implemented."
-        elif self.mode == "live-batch":
+        elif self._playback_mode == TemporalModes.BATCH:
             self._status = "connected"
             return "Live-batch mode is not yet implemented."
 
@@ -97,16 +94,40 @@ def get_status(self):
     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):
+        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()}'
+                         f'/sensorhub/api/datastreams/{self._datastream.ds_id}'
+                         f'/observations?f=application%2Fjson')
+        elif self._playback_mode == TemporalModes.ARCHIVE:
+            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&resultTime={self._datastream.valid_time.start}/'
+                         f'{self._datastream.valid_time.end}')
+        else:
+            raise ValueError("Playback mode not set. Cannot generate URL for DataSource.")
+
 
 class DataSourceHandler:
     datasource_map: dict[str, DataSource]
     _message_list: MessageHandler
+    _playback_mode: TemporalModes
 
-    def __init__(self):
+    def __init__(self, playback_mode: TemporalModes = TemporalModes.REAL_TIME):
         self.datasource_map = {}
         self._message_list = MessageHandler()
+        self._playback_mode = playback_mode
+
+    def set_playback_mode(self, mode: TemporalModes):
+        self._playback_mode = mode
 
     def add_datasource(self, datasource: DataSource):
+        datasource.set_mode(self._playback_mode)
         self.datasource_map[datasource.get_id()] = datasource
 
     def remove_datasource(self, datasource_id: str):
@@ -120,17 +141,27 @@ def initialize_all(self):
         # list comp is faster than for loop
         [ds.initialize() for ds in self.datasource_map.values()]
 
-    def set_ds_mode(self, mode: str):
-        var = (ds.set_mode(mode) for ds in self.datasource_map.values())
+    def set_ds_mode(self):
+        var = (ds.set_mode(self._playback_mode) for ds in self.datasource_map.values())
 
     async def connect_ds(self, datasource_id: str):
         ds = self.datasource_map.get(datasource_id)
         await ds.connect()
 
-    async def connect_all(self):
-        # call connect for all datasources
-        [(ds, await ds.connect()) for ds in self.datasource_map.values()]
-        for ds in self.datasource_map.values():
+    async def connect_all(self, timeperiod: TimePeriod):
+        """
+        Connects all datasources, optionally within a provided TimePeriod
+        :param timeperiod:
+        :return:
+        """
+        # search for datasources that fall within the timeperiod
+        if timeperiod is not None:
+            ds_matches = [ds for ds in self.datasource_map.values() if ds.is_within_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
 

oshconnect/oshconnect.py

@@ -10,6 +10,7 @@
 from conSys4Py.core.default_api_helpers import APIHelper
 from conSys4Py.datamodels.observations import ObservationOMJSONInline
 
+from oshconnect import TemporalModes
 from oshconnect.datamodels.datamodels import Node, System
 from oshconnect.datasource.datasource import DataSource, DataSourceHandler
 from oshconnect.datastore.datastore import DataStore
@@ -31,12 +32,20 @@ class OSHConnect:
     _datataskers: list[DataStore] = []
     _datagroups: list = []
     _tasks: list = []
+    _playback_mode: TemporalModes = TemporalModes.REAL_TIME
 
     def __init__(self, name: str, **kwargs):
+        """
+        :param name: name of the OSHConnect instance, in the event that
+        :param kwargs:
+            - 'playback_mode': TemporalModes
+        """
         self._name = name
         if 'nodes' in kwargs:
             self._nodes = kwargs['nodes']
         self._datasource_handler = DataSourceHandler()
+        if 'playback_mode' in kwargs:
+            self._playback_mode = kwargs['playback_mode']
 
     def get_name(self):
         return self._name
@@ -82,7 +91,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()
+            await self._datasource_handler.connect_all(None)
         else:
             for stream_id in stream_ids:
                 await self._datasource_handler.connect_ds(stream_id)
@@ -100,7 +109,7 @@ def discover_datastreams(self):
             res_datastreams = system.discover_datastreams()
             # create DataSource(s)
             new_datasource = [
-                DataSource(name=ds.name, mode="websocket", properties={}, datastream=ds, parent_system=system) for ds in
+                DataSource(name=ds.name, datastream=ds, parent_system=system) for ds in
                 res_datastreams]
             self._datafeeds.extend(new_datasource)
             list(map(self._datasource_handler.add_datasource, new_datasource))
@@ -122,3 +131,6 @@ def authenticate_user(self, user: dict):
 
     def synchronize_streams(self, systems: list):
         pass
+
+    def set_playback_mode(self, mode: TemporalModes):
+        self._datasource_handler.set_playback_mode(mode)