fix models and add logic supporting datastream retrieval

Author: tipatterson-dev

external_models/__init__.py

@@ -4,8 +4,10 @@
 #   Author:  Ian Patterson
 #   Contact Email:  ian@botts-inc.com
 #   ==============================================================================
+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
 
@@ -45,6 +47,23 @@ def check_iso_date(cls, v) -> str:
         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]}
+
+    def __repr__(self):
+        return f'{[self.start, self.end]}'
+
+
+
 class SecurityConstraints:
     constraints: list
 

external_models/object_models.py

@@ -15,7 +15,7 @@
 from external_models import BoundingBox, Capabilities, Characteristics, ConfigurationSettings, Contact, DateTime, \
     Documentation, \
     FeatureOfInterest, HistoryEvent, Input, LegalConstraints, \
-    Mode, Output, Parameter, ProcessMethod, SecurityConstraints
+    Mode, Output, Parameter, ProcessMethod, SecurityConstraints, TimePeriod
 
 
 class System(BaseModel):
@@ -53,21 +53,23 @@ class System(BaseModel):
     method: ProcessMethod = Field(None)
 
 
-class Datastream(BaseModel):
-    ds_id: str = Field(..., serialization_alias="id")
+class DatastreamResource(BaseModel):
+    # model_config = ConfigDict(populate_by_name=True)
+
+    ds_id: str = Field(..., alias="id")
     name: str = Field(...)
-    description: str
-    valid_time: DateTime = Field(..., serialization_alias="validTime")
-    output_name: str = Field(None, serialization_alias="outputName")
-    procedure_link: Link = Field(None, serialization_alias="procedureLink@link")
-    deployment_link: Link = Field(None, serialization_alias="deploymentLink@link")
-    ultimate_feature_of_interest_link: Link = Field(None, serialization_alias="ultimateFeatureOfInterest@link")
-    sampling_feature_link: Link = Field(None, serialization_alias="samplingFeature@link")
+    description: str = Field(None)
+    valid_time: TimePeriod = Field(..., alias="validTime")
+    output_name: str = Field(None, alias="outputName")
+    procedure_link: Link = Field(None, alias="procedureLink@link")
+    deployment_link: Link = Field(None, alias="deploymentLink@link")
+    ultimate_feature_of_interest_link: Link = Field(None, alias="ultimateFeatureOfInterest@link")
+    sampling_feature_link: Link = Field(None, alias="samplingFeature@link")
     parameters: dict = Field(None)
-    phenomenon_time: DateTime = Field(None, serialization_alias="phenomenonTimeInterval")
-    result_time: DateTime = Field(None, serialization_alias="resultTimeInterval")
-    ds_type: str = Field(None, serialization_alias="type")
-    result_type: str = Field(None, serialization_alias="resultType")
+    phenomenon_time: TimePeriod = Field(None, alias="phenomenonTimeInterval")
+    result_time: TimePeriod = Field(None, alias="resultTimeInterval")
+    ds_type: str = Field(None, alias="type")
+    result_type: str = Field(None, alias="resultType")
     links: List[Link] = Field(None)
     schema: DatastreamSchema = Field(None)
 

oshconnect/datamodels/datamodels.py

@@ -7,13 +7,14 @@
 from __future__ import annotations
 
 import base64
+import json
 import uuid
 from dataclasses import dataclass, field
 
-from conSys4Py import APIResourceTypes
+from conSys4Py import APIResourceTypes, DatastreamBodyJSON
 from conSys4Py.core.default_api_helpers import APIHelper
 
-from external_models.object_models import System as SystemResource
+from external_models.object_models import DatastreamResource, System as SystemResource
 from oshconnect import Endpoints
 
 
@@ -82,6 +83,9 @@ def discover_systems(self):
     def add_new_system(self, system: System):
         self._system_ids.append(system.uid)
 
+    def get_api_helper(self) -> APIHelper:
+        return self._api_helper
+
 
 class System:
     uid: uuid.UUID
@@ -102,6 +106,18 @@ def __init__(self, name: str, label: str, **kwargs):
     def update_parent_node(self, node: Node):
         self._parent_node = 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
     def from_system_resource(system_resource: SystemResource):
         other_props = system_resource.dict()

oshconnect/oshconnect.py

@@ -84,8 +84,11 @@ def visualize_streams(self, streams: list):
     def get_visualization_recommendations(self, streams: list):
         pass
 
-    def discover_datastreams(self, streams: list):
-        pass
+    def discover_datastreams(self):
+        for system in self._systems:
+            res_datastreams = system.discover_datastreams()
+            print(f'Datastreams found: {res_datastreams}')
+            self._datafeeds.extend(res_datastreams)
 
     def discover_systems(self, nodes: list[str] = None):
         search_nodes = self._nodes

tests/test_oshconnect.py

@@ -35,3 +35,13 @@ def test_find_systems(self):
         print(f'Found systems: {app._systems}')
         # assert len(systems) == 1
         # assert systems[0] == node.get_api_endpoint()
+
+    def test_oshconnect_find_datastreams(self):
+        app = OSHConnect(name="Test OSH Connect")
+        node = Node(address="http://localhost", port=8585, username="admin", password="admin")
+        app.add_node(node)
+        app.discover_systems()
+
+        app.discover_datastreams()
+        assert len(app._datafeeds) > 0
+