Bugfixes into main (#5)

* fix duplicated TimePeriod model

* Fix bug around TimePeriod comparison and add some more utility methods to that effect

* add quick way to access messages

* remove unused portions of docs for the time being

* satisfy flake8, remove unused imports

Author: tipatterson-dev

docs/source/index.rst

@@ -13,9 +13,6 @@ certain streaming features made possible by OpenSensorHub.
    :caption: Contents
 
    api
-   datasources
-   external
-
 
 
 

oshconnect/core_datamodels.py

@@ -4,15 +4,17 @@
 #   Author:  Ian Patterson
 #   Contact Email:  ian@botts-inc.com
 #   ==============================================================================
+from __future__ import annotations
+
 import uuid
-from typing import Any, List, Self
+from typing import List
 
 from conSys4Py import DatastreamSchema, Geometry
 from conSys4Py.datamodels.api_utils import Link
-from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator
+from pydantic import BaseModel, ConfigDict, Field
 from shapely import Point
 
-from .timemanagement import TimePeriod
+from oshconnect.timemanagement import DateTimeSchema, TimePeriod
 
 
 class BoundingBox(BaseModel):
@@ -30,52 +32,6 @@ class BoundingBox(BaseModel):
     #     return self
 
 
-class DateTime(BaseModel):
-    is_instant: bool = Field(True, description="Whether the date time is an instant or a period.")
-    iso_date: str = Field(None, description="The ISO formatted date time.")
-    time_period: tuple = Field(None, description="The time period of the date time.")
-
-    @model_validator(mode='before')
-    def valid_datetime_type(self) -> Self:
-        if self.is_instant:
-            if self.iso_date is None:
-                raise ValueError("Instant date time must have a valid ISO8601 date.")
-        return self
-
-    @field_validator('iso_date')
-    @classmethod
-    def check_iso_date(cls, v) -> str:
-        if not v:
-            raise ValueError("Instant date time must have a valid ISO8601 date.")
-        return v
-
-
-class TimePeriod(BaseModel):
-    start: str = Field(...)
-    end: str = Field(...)
-
-    @model_validator(mode='before')
-    @classmethod
-    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]}
-        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]}'
-
-    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:
     constraints: list
 
@@ -148,7 +104,7 @@ class SystemResource(BaseModel):
     keywords: List[str] = Field(None)
     identifiers: List[str] = Field(None)
     classifiers: List[str] = Field(None)
-    valid_time: DateTime = Field(None, serialization_alias="validTime")
+    valid_time: DateTimeSchema = Field(None, serialization_alias="validTime")
     security_constraints: List[SecurityConstraints] = Field(None, serialization_alias="securityConstraints")
     legal_constraints: List[LegalConstraints] = Field(None, serialization_alias="legalConstraints")
     characteristics: List[Characteristics] = Field(None)
@@ -191,8 +147,8 @@ class DatastreamResource(BaseModel):
 class Observation(BaseModel):
     sampling_feature_id: str = Field(None, serialization_alias="samplingFeature@Id")
     procedure_link: Link = Field(None, serialization_alias="procedure@link")
-    phenomenon_time: DateTime = Field(None, serialization_alias="phenomenonTime")
-    result_time: DateTime = Field(..., serialization_alias="resultTime")
+    phenomenon_time: DateTimeSchema = Field(None, serialization_alias="phenomenonTime")
+    result_time: DateTimeSchema = Field(..., serialization_alias="resultTime")
     parameters: dict = Field(None)
     result: dict = Field(...)
     result_link: Link = Field(None, serialization_alias="result@link")

oshconnect/datasource.py

@@ -18,11 +18,15 @@
 import websockets
 from conSys4Py import APIResourceTypes
 from conSys4Py.datamodels.observations import ObservationOMJSONInline
+from conSys4Py.datamodels.swe_components import DataRecordSchema
 
 from .core_datamodels import DatastreamResource, SystemResource, TimePeriod
 from .osh_connect_datamodels import TemporalModes
 
 
+# from swecommondm.component_implementations import DataRecord
+
+
 class DataSource:
     """
     DataSource: represents the active connection of a datastream object.
@@ -43,6 +47,7 @@ class DataSource:
     _auth: str = None
     _websocket: websockets.WebSocketClientProtocol = None
     _extra_headers: dict = None
+    _result_schema: DataRecordSchema = None
 
     def __init__(self, name: str, datastream: DatastreamResource,
                  parent_system: SystemResource):
@@ -64,6 +69,15 @@ def __init__(self, name: str, datastream: DatastreamResource,
         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}'}
+            # get result schema
+
+            # t_url = f'http://{self._parent_system.get_parent_node().get_address()}:{self._parent_system.get_parent_node().get_port()}'
+            #
+            # res = conSys4Py.part_2.datastreams.retrieve_datastream_schema(t_url,
+            #                                                               datastream_id=self._datastream.ds_id,
+            #                                                               api_root=self._parent_system.get_parent_node()._api_helper.api_root,
+            #                                                               headers=self._extra_headers)
+            # print(res.json())
 
     def get_id(self) -> str:
         """
@@ -421,6 +435,22 @@ async def handle_http_batching(self, datasource: DataSource,
             self._message_list.add_message(msg_wrapper)
         return resp.json()
 
+    def get_message_handler(self) -> MessageHandler:
+        """
+        Get the MessageHandler object from the DataSourceHandler
+
+        :return: MessageHandler object
+        """
+        return self._message_list
+
+    def get_messages(self) -> list[MessageWrapper]:
+        """
+        Get the list of MessageWrapper objects from the MessageHandler
+
+        :return: List of MessageWrapper objects
+        """
+        return self._message_list.get_messages()
+
 
 class MessageHandler:
     """
@@ -441,7 +471,7 @@ def add_message(self, message: MessageWrapper):
         :return:
         """
         self._message_list.append(message)
-        print(self._message_list)
+        # print(self._message_list)
 
     def get_messages(self) -> list[MessageWrapper]:
         """
@@ -493,7 +523,7 @@ def get_message_as_dict(self) -> dict:
         """
         Get the observation data from the MessageWrapper as a dictionary
 
-        :return: disct of the observation result data
+        :return: dict of the observation result data
         """
         return self._message.model_dump()
 

oshconnect/oshconnectapi.py

@@ -8,7 +8,7 @@
 from conSys4Py.core.default_api_helpers import APIHelper
 
 from .core_datamodels import TimePeriod
-from .datasource import DataSource, DataSourceHandler
+from .datasource import DataSource, DataSourceHandler, MessageWrapper
 from .datastore import DataStore
 from .osh_connect_datamodels import Node, System, TemporalModes
 from .styling import Styling
@@ -180,3 +180,10 @@ def set_timeperiod(self, start_time: str, end_time: str):
         """
         tp = TimePeriod(start=start_time, end=end_time)
         self.timestream = TimeManagement(time_range=tp)
+
+    def get_message_list(self) -> list[MessageWrapper]:
+        """
+        Get the list of messages that have been received by the OSHConnect instance.
+        :return: list of MessageWrapper objects
+        """
+        return self._datasource_handler.get_messages()