Add control stream discovery functionality to oshconnectapi and streamableresource

Author: tipatterson-dev

src/oshconnect/oshconnectapi.py

@@ -11,7 +11,7 @@
 from .csapi4py.default_api_helpers import APIHelper
 from .datastore import DataStore
 from .resource_datamodels import DatastreamResource
-from .streamableresource import Node, System, SessionManager, Datastream
+from .streamableresource import Node, System, SessionManager, Datastream, ControlStream
 from .styling import Styling
 from .timemanagement import TemporalModes, TimeManagement, TimePeriod
 
@@ -120,6 +120,7 @@ def discover_datastreams(self):
             datastreams = list(
                 map(lambda ds: Datastream(parent_node=system.get_parent_node(), id=ds.ds_id, datastream_resource=ds),
                     res_datastreams))
+
             for ds in datastreams:
                 ds.set_parent_resource_id(system.get_underlying_resource().system_id)
             # datastreams = [ds.set_parent_resource_id(system.get_underlying_resource().system_id) for ds in datastreams]
@@ -142,7 +143,14 @@ def discover_systems(self, nodes: list[str] = None):
             self._systems.extend(res_systems)
 
     def discover_controlstreams(self, streams: list):
-        pass
+        for system in self._systems:
+            res_controlstreams = system.discover_controlstreams()
+            controlstreams = list(
+                map(lambda cs: ControlStream(parent_node=system.get_parent_node(), id=cs.cs_id,
+                                             controlstream_resource=cs), res_controlstreams))
+            for cs in controlstreams:
+                cs.set_parent_resource_id(system.get_underlying_resource().system_id)
+            self._datataskers.extend(controlstreams)
 
     def authenticate_user(self, user: dict):
         pass

src/oshconnect/streamableresource.py

@@ -574,6 +574,18 @@ def discover_datastreams(self) -> list[DatastreamResource]:
 
         return ds_resources
 
+    def discover_controlstreams(self) -> list[ControlStreamResource]:
+        res = self._parent_node.get_api_helper().get_resource(APIResourceTypes.SYSTEM, self._resource_id,
+                                                              APIResourceTypes.CONTROL_CHANNEL)
+        controlstream_json = res.json()['items']
+        cs_resources = []
+
+        for cs in controlstream_json:
+            controlstream_objs = ControlStreamResource.model_validate(cs)
+            cs_resources.append(controlstream_objs)
+
+        return cs_resources
+
     @staticmethod
     def from_system_resource(system_resource: SystemResource, parent_node: Node) -> System:
         other_props = system_resource.model_dump()
@@ -818,14 +830,6 @@ def init_mqtt(self):
         super().init_mqtt()
         self._topic = self.get_mqtt_topic(subresource=APIResourceTypes.COMMAND)
 
-    # def subscribe_to_status(self, topic: str):
-    #     # TODO: This should probably be a flag to subscribe to status updates as the commands come in, trying to manage this manually would
-    #     # prove tedious
-    #     pass
-    #
-    # def publish_status(self, payload):
-    #     pass
-
     def get_mqtt_status_topic(self):
         return self.get_mqtt_topic(subresource=APIResourceTypes.STATUS)