update control channels helpers for parity

Author: tipatterson-dev

conSys/datamodels/control_streams.py

@@ -0,0 +1,38 @@
+from __future__ import annotations
+
+from typing import Union
+
+from pydantic import BaseModel, Field, SerializeAsAny
+
+from conSys import AnyComponentSchema
+from conSys.datamodels.encoding import Encoding
+
+
+class ControlStreamJSONSchema(BaseModel):
+    """
+    A class to represent the schema of a control stream
+    """
+    id: str = Field(None)
+    name: str = Field(...)
+    description: str = Field(None)
+    deployment_link: str = Field(None, serialization_alias='deployment@link')
+    ultimate_feature_of_interest_link: str = Field(None, serialization_alias='ultimateFeatureOfInterest@link')
+    sampling_feature_link: str = Field(None, alias='samplingFeature@link')
+    valid_time: list = Field(None, serialization_alias='validTime')
+    input_name: str = Field(None, serialization_alias='inputName')
+    links: list = Field(None)
+    schema: SerializeAsAny[Union[SWEControlChannelSchema, JSONControlChannelSchema]] = Field(...)
+
+
+class SWEControlChannelSchema(BaseModel):
+    """
+    A class to represent the schema of a control channel
+    """
+    command_format: str = Field("application/swe+json", serialization_alias='commandFormat')
+    encoding: SerializeAsAny[Encoding] = Field(...)
+    record_schema: SerializeAsAny[AnyComponentSchema] = Field(..., serialization_alias='recordSchema')
+
+
+class JSONControlChannelSchema(BaseModel):
+    command_format: str = Field("application/cmd+json", serialization_alias='commandFormat')
+    params_schema: SerializeAsAny[AnyComponentSchema] = Field(..., serialization_alias='paramsSchema')

conSys/part_2/control_channels.py

@@ -1,11 +1,13 @@
+from typing import Union
+
 import requests
 from pydantic import HttpUrl
 
 from conSys.con_sys_api import ConnectedSystemsRequestBuilder
 from conSys.constants import APITerms
 
 
-def list_all_constrol_streams(server_addr: HttpUrl, api_root: str = APITerms.API.value):
+def list_all_control_streams(server_addr: HttpUrl, api_root: str = APITerms.API.value, headers: dict = None):
     """
     Lists all control streams
     :return:
@@ -15,12 +17,14 @@ def list_all_constrol_streams(server_addr: HttpUrl, api_root: str = APITerms.API
                    .with_api_root(api_root)
                    .for_resource_type(APITerms.CONTROL_STREAMS.value)
                    .build_url_from_base()
+                   .with_headers(headers)
+                   .with_request_method('GET')
                    .build())
-    resp = requests.get(api_request.url, params=api_request.body, headers=api_request.headers)
-    return resp.json()
+    return api_request.make_request()
 
 
-def list_control_streams_of_system(server_addr: HttpUrl, system_id: str, api_root: str = APITerms.API.value):
+def list_control_streams_of_system(server_addr: HttpUrl, system_id: str, api_root: str = APITerms.API.value,
+                                   headers=None):
     """
     Lists all control streams of a system
     :return:
@@ -32,13 +36,14 @@ def list_control_streams_of_system(server_addr: HttpUrl, system_id: str, api_roo
                    .with_resource_id(system_id)
                    .for_resource_type(APITerms.CONTROL_STREAMS.value)
                    .build_url_from_base()
+                   .with_headers(headers)
+                   .with_request_method('GET')
                    .build())
-    resp = requests.get(api_request.url, params=api_request.body, headers=api_request.headers)
-    return resp.json()
+    return api_request.make_request()
 
 
-def add_control_streams_to_system(server_addr: HttpUrl, system_id: str, request_body: dict,
-                                  api_root: str = APITerms.API.value):
+def add_control_streams_to_system(server_addr: HttpUrl, system_id: str, request_body: Union[str, dict],
+                                  api_root: str = APITerms.API.value, headers=None):
     """
     Adds a control stream to a system by its id
     :return:
@@ -48,16 +53,17 @@ def add_control_streams_to_system(server_addr: HttpUrl, system_id: str, request_
                    .with_api_root(api_root)
                    .for_resource_type(APITerms.SYSTEMS.value)
                    .with_resource_id(system_id)
-                   .for_resource_type(APITerms.CONTROL_STREAMS.value)
+                   .for_sub_resource_type(APITerms.CONTROL_STREAMS.value)
                    .with_request_body(request_body)
                    .build_url_from_base()
+                   .with_headers(headers)
+                   .with_request_method('POST')
                    .build())
-    resp = requests.post(api_request.url, params=api_request.body, headers=api_request.headers)
-    return resp.json()
+    return api_request.make_request()
 
 
 def retrieve_control_stream_description_by_id(server_addr: HttpUrl, control_stream_id: str,
-                                              api_root: str = APITerms.API.value):
+                                              api_root: str = APITerms.API.value, headers: dict = None):
     """
     Retrieves a control stream by its id
     :return:
@@ -68,12 +74,15 @@ def retrieve_control_stream_description_by_id(server_addr: HttpUrl, control_stre
                    .for_resource_type(APITerms.CONTROL_STREAMS.value)
                    .with_resource_id(control_stream_id)
                    .build_url_from_base()
+                   .with_headers(headers)
+                   .with_request_method('GET')
                    .build())
-    resp = requests.get(api_request.url, params=api_request.body, headers=api_request.headers)
-    return resp.json()
+
+    return api_request.make_request()
 
 
-def update_control_stream_description_by_id(server_addr: HttpUrl, control_stream_id: str, request_body: dict,
+def update_control_stream_description_by_id(server_addr: HttpUrl, control_stream_id: str,
+                                            request_body: Union[str, dict],
                                             api_root: str = APITerms.API.value):
     """
     Updates a control stream by its id
@@ -91,7 +100,8 @@ def update_control_stream_description_by_id(server_addr: HttpUrl, control_stream
     return resp.json()
 
 
-def delete_control_stream_by_id(server_addr: HttpUrl, control_stream_id: str, api_root: str = APITerms.API.value):
+def delete_control_stream_by_id(server_addr: HttpUrl, control_stream_id: str, api_root: str = APITerms.API.value,
+                                headers=None):
     """
     Deletes a control stream by its id
     :return:
@@ -102,13 +112,15 @@ def delete_control_stream_by_id(server_addr: HttpUrl, control_stream_id: str, ap
                    .for_resource_type(APITerms.CONTROL_STREAMS.value)
                    .with_resource_id(control_stream_id)
                    .build_url_from_base()
+                   .with_headers(headers)
+                   .with_request_method('DELETE')
                    .build())
-    resp = requests.delete(api_request.url, params=api_request.body, headers=api_request.headers)
-    return resp.json()
+
+    return api_request.make_request()
 
 
 def retrieve_control_stream_schema_by_id(server_addr: HttpUrl, control_stream_id: str,
-                                         api_root: str = APITerms.API.value):
+                                         api_root: str = APITerms.API.value, headers: dict = None):
     """
     Retrieves a control stream schema by its id
     :return:
@@ -120,13 +132,15 @@ def retrieve_control_stream_schema_by_id(server_addr: HttpUrl, control_stream_id
                    .with_resource_id(control_stream_id)
                    .for_resource_type(APITerms.SCHEMA.value)
                    .build_url_from_base()
+                   .with_headers(headers)
+                   .with_request_method('GET')
                    .build())
-    resp = requests.get(api_request.url, params=api_request.body, headers=api_request.headers)
-    return resp.json()
+
+    return api_request.make_request()
 
 
 def update_control_stream_schema_by_id(server_addr: HttpUrl, control_stream_id: str, request_body: dict,
-                                       api_root: str = APITerms.API.value):
+                                       api_root: str = APITerms.API.value, headers: dict = None):
     """
     Updates a control stream schema by its id
     :return:
@@ -139,6 +153,7 @@ def update_control_stream_schema_by_id(server_addr: HttpUrl, control_stream_id:
                    .for_resource_type(APITerms.SCHEMA.value)
                    .with_request_body(request_body)
                    .build_url_from_base()
+                   .with_headers(headers)
+                   .with_request_method('PUT')
                    .build())
-    resp = requests.put(api_request.url, params=api_request.body, headers=api_request.headers)
-    return resp.json()
+    return api_request.make_request()

tests/test_script_full_suite.py

@@ -1,10 +1,11 @@
 import random
 
 from conSys import Systems, SamplingFeatures, Datastreams, SmlJSONBody, GeoJSONBody, model_utils, \
-    DatastreamBodyJSON, ObservationFormat, URI, Procedures, Geometry, Deployments
+    DatastreamBodyJSON, ObservationFormat, URI, Procedures, Geometry, Deployments, ControlChannels
+from conSys.datamodels.control_streams import ControlStreamJSONSchema, SWEControlChannelSchema, JSONControlChannelSchema
 from conSys.datamodels.datastreams import SWEDatastreamSchema
 from conSys.datamodels.encoding import JSONEncoding
-from conSys.datamodels.swe_components import BooleanSchema, TimeSchema, DataRecordSchema
+from conSys.datamodels.swe_components import BooleanSchema, TimeSchema, DataRecordSchema, CountSchema
 
 server_url = "http://localhost:8282/sensorhub"
 geo_json_headers = {"Content-Type": "application/geo+json"}
@@ -281,20 +282,37 @@ def test_create_datastreams():
                                                  headers=json_headers)
     print(resp)
 
+
 """
 Command and Control Channel Section
 """
+
+
 def test_create_control_channel():
-    geo_json_body = GeoJSONBody(type='Feature', id=str(random.randint(1000, 9999)),
-                                description="Test Insertion of Control Channel via GEOJSON",
-                                properties={
-                                    "featureType": "http://www.w3.org/ns/ssn/ControlChannel",
-                                    "name": f'Test Control Channel - GeoJSON',
-                                    "uid": f'urn:test:client:geo-cc',
-                                    "description": "A Test Control Channel inserted from the Python CSAPI Client",
-                                })
-    resp = ControlChannels.create_new_control_channels(server_url, geo_json_body.model_dump_json(exclude_none=True, by_alias=True),
-                                                         headers=geo_json_headers)
+    systems_list = Systems.list_all_systems(server_url)
+    system_id = systems_list["items"][0]["id"]
+
+    time_schema = TimeSchema(label="Test Control Channel Time", definition="http://test.com/Time",
+                             uom=URI(href="http://test.com/TimeUOM"))
+    count_schema = CountSchema(label="Test Control Channel Count", definition="http://test.com/Count")
+
+    control_schema = JSONControlChannelSchema(command_format=ObservationFormat.SWE_JSON.value,
+                                              params_schema=DataRecordSchema(label="Test Control Channel Record",
+                                                                             definition="http://test.com/Record",
+                                                                             fields=[time_schema, count_schema]))
+    request_body = ControlStreamJSONSchema(name="Test Control Channel", input_name="TestControlInput1",
+                                           schema=control_schema)
+    print(f'Request Body for Control Stream: {request_body.model_dump_json(exclude_none=True, by_alias=True)}')
+    resp = ControlChannels.add_control_streams_to_system(server_url, system_id,
+                                                         request_body.model_dump_json(exclude_none=True,
+                                                                                      by_alias=True),
+                                                         headers=json_headers)
+    print(resp)
+
+
+def test_list_control_streams():
+    control_streams = ControlChannels.list_all_control_streams(server_url)
+    print(control_streams)
 
 
 """