Observation Insertion (#14)

* observation insert working

* improve ability to add systems from the top level of the app

* update docs

Author: tipatterson-dev

docs/source/conf.py

@@ -24,7 +24,7 @@ def setup(app):
 project = 'OSHConnect-Python'
 copyright = '2024, Botts Innovative Research, Inc.'
 author = 'Ian Patterson'
-release = '0.1'
+release = '0.2'
 
 # -- General configuration ---------------------------------------------------
 # https://www.sphinx-doc.org/en/master/usage/configuration.html#general-configuration

docs/source/insertion_tutorial.rst

@@ -91,3 +91,89 @@ observations.
     for message in messages:
         for observation in message.observations:
             do_something_with(observation)
+
+
+Resource Insertion
+=========================================
+Other use cases of the OSH Connect library may involve inserting new resources into OpenSensorHub or another Connected Systems API server.
+
+Adding and Inserting a New System
+-----------------------------------------
+The first major step in a common workflow is to add a new system to the OSH Connect instance.
+There are a couple of ways to do this, but the recommended method is as follows:
+
+.. note::
+
+    The `insert_system` method requires a `Node` object to be passed in as the second argument.
+    Creating one is covered in an earlier section.
+
+.. code-block:: python
+
+    from oshconnect.osh_connect_datamodels import System
+
+    new_system = app.insert_system(
+        System(name="Test System", description="Test System Description", label="Test System",
+               urn="urn:system:test"), node)
+
+Adding and Inserting a New Datastream
+-----------------------------------------
+Once you have a `System` object, you can add a new datastream to it. This is one of the more complex operations
+in the library as the schema is very flexible by design. Luckily, the schemas are validated by the underlying data
+models, so you can be sure that your datastream is valid before inserting it.
+
+.. caution::
+
+    Some implementations of the Connected Systems API may require additional fields to be filled in.
+    OSH Connect is primarily focused on the OpenSensorHub implementation, but does not some of the fields that
+    are required by and OpenSensorHub node.
+
+In this example, we will add a new datastream to the `new_system` object that we created in the previous example.
+You'll note the creation of a `DataRecordSchema` object, in OSH's implementation, a DataRecord is the root of all
+datastream schemas.
+
+.. code-block:: python
+
+    from oshconnect.osh_connect_datamodels import Datastream
+
+    datarecord_schema = DataRecordSchema(label='Example Data Record', description='Example Data Record Description',
+                                         definition='www.test.org/records/example-datarecord', fields=[])
+    time_schema = TimeSchema(label="Timestamp", definition="http://test.com/Time", name="timestamp",
+                             uom=URI(href="http://test.com/TimeUOM"))
+    continuous_value_field = QuantitySchema(name='continuous-value-distance', label='Continuous Value Distance',
+                                            description='Continuous Value Description',
+                                            definition='www.test.org/fields/continuous-value',
+                                            uom=UCUMCode(code='m', label='meters'))
+    example_text_field = TextSchema(name='example-text-field', label='Example Text Field', definition='www.test.org/fields/example-text-field')
+    # add the fields to the datarecord schema, these can also be added added to the datarecord when it is created
+    datarecord_schema.fields.append(time_schema)   # TimeSchema is required to be the first field in the datarecord for OSH
+    datarecord_schema.fields.append(continuous_value_field)
+    datarecord_schema.fields.append(example_text_field)
+    # Add the datastream to the system
+    datastream = new_system.add_insert_datastream(datarecord_schema)
+
+.. note::
+
+    A TimeSchema is required to be the first field in the DataRecordSchema for OSH.
+
+Inserting an Observation into and OpenSensorHub Node
+-----------------------------------------
+Upon successfully adding a new datastream to a system, it is now possible to send observation data to the node.
+
+.. code-block:: python
+
+    datastream.insert_observation_dict({
+        "resultTime": TimeInstant.now_as_time_instant().get_iso_time(),     # resultTime is required for OSH
+        "phenomenonTime": TimeInstant.now_as_time_instant().get_iso_time(), # phenomenonTime is required for OSH
+        "result": {
+            "timestamp": TimeInstant.now_as_time_instant().epoch_time,
+            "continuous-value-distance": 1.0,
+            "example-text-field": "Here is some text"
+        }
+    })
+
+.. note::
+
+        The `resultTime` and `phenomenonTime` fields are required for OSH.
+        The `result` field is representative of the schemas included in the DataRecordSchema's fields.
+        You'll notice that they are referred to by their `name` field in the schema as it is the "machine" name
+        of the output.

docs/source/tutorial.rst

@@ -43,7 +43,8 @@ class Node:
     is_secure: bool
     _basic_auth: bytes = None
     _api_helper: APIHelper
-    _system_ids: list[uuid] = field(default_factory=list)
+    # _system_ids: list[uuid] = field(default_factory=list)
+    _systems: list[System] = field(default_factory=list)
 
     def __init__(self, protocol: str, address: str, port: int,
                  username: str = None, password: str = None,
@@ -62,7 +63,7 @@ def __init__(self, protocol: str, address: str, port: int,
             password=password)
         if self.is_secure:
             self._api_helper.user_auth = True
-        self._system_ids = []
+        self._systems = []
 
     def get_id(self):
         return self._id
@@ -97,19 +98,38 @@ def discover_systems(self):
                 system = SystemResource.model_validate(system_json)
                 sys_obj = System.from_system_resource(system)
                 sys_obj.update_parent_node(self)
-                self._system_ids.append(sys_obj.uid)
+                self._systems.append(sys_obj)
                 new_systems.append(sys_obj)
             return new_systems
         else:
             return None
 
     def add_new_system(self, system: System):
         system.update_parent_node(self)
-        self._system_ids.append(system.uid)
+        self._systems.append(system)
 
     def get_api_helper(self) -> APIHelper:
         return self._api_helper
 
+    # System Management
+
+    def add_system(self, system: System, target_node: Node, insert_resource: bool = False):
+        """
+        Add a system to the target node.
+        :param system: System object
+        :param target_node: Node object
+        :param insert_resource: Whether to insert the system into the target node's server, default is False
+        :return:
+        """
+        if insert_resource:
+            system.insert_self()
+        target_node.add_new_system(system)
+        self._systems.append(system)
+        return system
+
+    def systems(self) -> list[System]:
+        return self._systems
+
 
 class System:
     uid: uuid.UUID
@@ -171,7 +191,8 @@ def from_system_resource(system_resource: SystemResource):
                                 label=other_props['properties']['name'])
         else:
             new_system = System(name=system_resource.name,
-                                label=system_resource.label, urn=system_resource.urn, resource_id=system_resource.system_id)
+                                label=system_resource.label, urn=system_resource.urn,
+                                resource_id=system_resource.system_id)
         return new_system
 
     def to_system_resource(self) -> SystemResource:
@@ -193,19 +214,23 @@ def add_insert_datastream(self, datastream: DataRecordSchema):
         print(f'Adding datastream: {datastream.model_dump_json(exclude_none=True, by_alias=True)}')
         # Make the request to add the datastream
         # if successful, add the datastream to the system
-        datastream_schema = SWEDatastreamSchema(record_schema=datastream, obs_format='application/swe+json', encoding=JSONEncoding())
-        datastream_resource = DatastreamResource(ds_id="default", name=datastream.label, output_name=datastream.label, record_schema=datastream_schema,
+        datastream_schema = SWEDatastreamSchema(record_schema=datastream, obs_format='application/swe+json',
+                                                encoding=JSONEncoding())
+        datastream_resource = DatastreamResource(ds_id="default", name=datastream.label, output_name=datastream.label,
+                                                 record_schema=datastream_schema,
                                                  valid_time=TimePeriod(start=TimeInstant.now_as_time_instant(),
-                                                                       end=TimeInstant(utc_time=TimeUtils.to_utc_time("2026-12-31T00:00:00Z"))))
-        # datasource = DataStream(name=datastream.label, datastream=datastream_resource, parent_system=self._sys_resource)
+                                                                       end=TimeInstant(utc_time=TimeUtils.to_utc_time(
+                                                                           "2026-12-31T00:00:00Z"))))
 
         api = self._parent_node.get_api_helper()
-        print(f'Attempting to create datastream: {datastream_resource.model_dump_json(by_alias=True, exclude_none=True)}')
+        print(
+            f'Attempting to create datastream: {datastream_resource.model_dump_json(by_alias=True, exclude_none=True)}')
         print(
             f'Attempting to create datastream: {datastream_resource.model_dump(by_alias=True, exclude_none=True)}')
-        res = api.create_resource(APIResourceTypes.DATASTREAM, datastream_resource.model_dump_json(by_alias=True, exclude_none=True),
+        res = api.create_resource(APIResourceTypes.DATASTREAM,
+                                  datastream_resource.model_dump_json(by_alias=True, exclude_none=True),
                                   req_headers={
-                                        'Content-Type': 'application/json'
+                                      'Content-Type': 'application/json'
                                   }, parent_res_id=self.resource_id)
 
         if res.ok:
@@ -216,10 +241,12 @@ def add_insert_datastream(self, datastream: DataRecordSchema):
             raise Exception(f'Failed to create datastream: {datastream_resource.name}')
 
         self.datastreams.append(datastream_resource)
+        return Datastream(datastream_id, self._parent_node, datastream_resource)
 
     def insert_self(self):
         res = self._parent_node.get_api_helper().create_resource(
-            APIResourceTypes.SYSTEM, self.to_system_resource().model_dump_json(by_alias=True, exclude_none=True), req_headers={
+            APIResourceTypes.SYSTEM, self.to_system_resource().model_dump_json(by_alias=True, exclude_none=True),
+            req_headers={
                 'Content-Type': 'application/sml+json'
             })
 
@@ -232,7 +259,8 @@ def insert_self(self):
     def retrieve_resource(self):
         if self.resource_id is None:
             return None
-        res = self._parent_node.get_api_helper().retrieve_resource(res_type=APIResourceTypes.SYSTEM, res_id=self.resource_id)
+        res = self._parent_node.get_api_helper().retrieve_resource(res_type=APIResourceTypes.SYSTEM,
+                                                                   res_id=self.resource_id)
         if res.ok:
             system_json = res.json()
             print(system_json)
@@ -241,31 +269,56 @@ def retrieve_resource(self):
             self._sys_resource = system_resource
 
 
-# class Datastream:
-#     should_poll: bool
-#     _datastream_resource: DatastreamResource
-#
-#     def __init__(self, datastream_resource: DatastreamResource):
-#         pass
-#
-#     def get_id(self):
-#         return self._datastream_resource.ds_id
-#
-#     def insert_observation(self, observation: Observation):
-#         pass
-#
-#     def to_resource(self) -> DatastreamResource:
-#         # if self._datastream_resource is None:
-#         #     self._datastream_resource = DatastreamResource(
-#         #         ds_id=uuid.uuid4(), name=self.name,
-#         #         valid_time=self.validTimeRange)
-#         return self._datastream_resource
-#
-#     # def create_from_record_schema(record_schema: DataRecordSchema, parent_system: System):
-#     #     new_ds = Datastream(name=record_schema.label, record_schema=record_schema)
-#     #     new_ds._datastream_resource = DatastreamResource(ds_id=uuid.uuid4(), name=new_ds.name)
-#     #     parent_system.datastreams.append(new_ds)
-#     #     return new_ds
+class Datastream:
+    should_poll: bool
+    _id: str
+    _datastream_resource: DatastreamResource
+    _parent_node: Node
+
+    def __init__(self, id: str = None, parent_node: Node = None, datastream_resource: DatastreamResource = None):
+        self._id = id
+        self._parent_node = parent_node
+        self._datastream_resource = datastream_resource
+
+    def get_id(self):
+        return self._datastream_resource.ds_id
+
+    def insert_observation(self, observation: Observation):
+        pass
+
+    def to_resource(self) -> DatastreamResource:
+        # if self._datastream_resource is None:
+        #     self._datastream_resource = DatastreamResource(
+        #         ds_id=uuid.uuid4(), name=self.name,
+        #         valid_time=self.validTimeRange)
+        return self._datastream_resource
+
+    def observation_template(self) -> Observation:
+        pass
+
+    def create_observation(self, obs_data: dict):
+        obs = ObservationResource(result=obs_data, result_time=TimeInstant.now_as_time_instant())
+        # Validate against the schema
+        if self._datastream_resource.record_schema is not None:
+            obs.validate_against_schema(self._datastream_resource.record_schema)
+        return obs
+
+    def insert_observation_dict(self, obs_data: dict):
+        res = self._parent_node.get_api_helper().create_resource(APIResourceTypes.OBSERVATION, obs_data,
+                                                                 parent_res_id=self._id,
+                                                                 req_headers={'Content-Type': 'application/json'})
+        if res.ok:
+            obs_id = res.headers['Location'].split('/')[-1]
+            print(f'Inserted observation: {obs_id}')
+            return id
+        else:
+            raise Exception(f'Failed to insert observation: {res.text}')
+
+    # def create_from_record_schema(record_schema: DataRecordSchema, parent_system: System):
+    #     new_ds = Datastream(name=record_schema.label, record_schema=record_schema)
+    #     new_ds._datastream_resource = DatastreamResource(ds_id=uuid.uuid4(), name=new_ds.name)
+    #     parent_system.datastreams.append(new_ds)
+    #     return new_ds
 
 
 class ControlChannel: