WIP command updates

Author: tipatterson-dev

oshconnect/control.py

@@ -4,6 +4,8 @@
 #   Author:  Ian Patterson
 #   Contact Email:  ian@botts-inc.com
 #   ==============================================================================
+from __future__ import annotations
+
 import websockets
 from consys4py.comm.mqtt import MQTTCommClient
 from consys4py.datamodels.commands import CommandJSON
@@ -18,16 +20,17 @@ class ControlSchema:
 
 class ControlStream:
     name: str = None
-    _parent_systems: System = None
+    _parent_system: System = None
     _strategy: str = "mqtt"
     _resource_endpoint = None
     # _auth: str = None
     _websocket: websockets.WebSocketServerProtocol = None
     _schema: ControlStreamJSONSchema = None
     _mqtt_client: MQTTCommClient = None
+    _id: str = None
 
     def __init__(self, parent_system: System, resource_endpoint: str, name=None, strategy="mqtt"):
-        self._parent_systems = parent_system
+        self._parent_system = parent_system
         self.name = name
         self._strategy = strategy
         self._resource_endpoint = resource_endpoint
@@ -46,7 +49,15 @@ def subscribe(self):
         elif self._strategy == "websocket":
             pass
 
-    def publish(self, payload: CommandJSON):
+    def publish_status(self, payload: CommandJSON):
+        if self._strategy == "mqtt" and self._mqtt_client is not None:
+            self._mqtt_client.publish(f'{self._resource_endpoint}/status', payload=payload, qos=1)
+        elif self._strategy == "mqtt" and self._mqtt_client is None:
+            raise ValueError("No MQTT Client found.")
+        elif self._strategy == "websocket":
+            pass
+
+    def publish_command(self, payload: CommandJSON):
         if self._strategy == "mqtt" and self._mqtt_client is not None:
             self._mqtt_client.publish(f'{self._resource_endpoint}/status', payload=payload, qos=1)
         elif self._strategy == "mqtt" and self._mqtt_client is None:
@@ -60,6 +71,45 @@ def disconnect(self):
     def unsubscribe(self):
         self._mqtt_client.unsubscribe(f'{self._resource_endpoint}/commands')
 
+    def get_schema(self):
+        return self._schema
+
+    def set_id(self, id: str):
+        self._id = id
+
+    def get_id(self):
+        return self._id
+
+    def add_status_listener(self, listener: callable):
+        """
+        Adds a callback function which will be called when a status message is received via MQTT.
+        :param listener: callback function that executes when the status message is received.
+        :return: None
+        """
+        self._mqtt_client.subscribe(f'{self._resource_endpoint}/status', msg_callback=listener)
+
+    def add_command_listener_and_callback(self, callback: callable):
+        """
+        Creates a listener for the command stream and adds a callback function which will be called when a command message is received.
+        """
+        self._mqtt_client.subscribe(f'{self._resource_endpoint}/commands', msg_callback=callback)
+
+    def send_command(self, command: CommandJSON | Command):
+        if isinstance(command, CommandJSON):
+            self.publish_command(command)
+        elif isinstance(command, Command):
+            self.publish_status(command.record)
+
 
 class Command:
-    pass
+    _id: str = None
+    record: CommandJSON = None
+    status: str = None
+    _parent_stream: ControlStream = None
+
+    def send(self):
+        self._parent_stream.send_command(self)
+
+    def update_status(self, status_val: str):
+        self._parent_stream.publish_status(status_val)
+

oshconnect/osh_connect_datamodels.py

@@ -16,6 +16,7 @@
 from consys4py.datamodels.encoding import JSONEncoding
 from consys4py.datamodels.swe_components import DataRecordSchema
 
+from .control import ControlStream
 from .core_datamodels import DatastreamResource, ObservationResource, SystemResource
 from .timemanagement import TimeInstant, TimePeriod, TimeUtils
 
@@ -139,7 +140,7 @@ class System:
     label: str
     # datastreams: list[Datastream]
     datastreams: list[DatastreamResource]
-    control_channels: list[ControlChannel]
+    control_channels: list[ControlStream]
     description: str
     urn: str
     _parent_node: Node
@@ -271,6 +272,26 @@ def retrieve_resource(self):
             print(f'System Resource: {system_resource}')
             self._sys_resource = system_resource
 
+    def add_insert_control_stream(self, control_stream: ControlStream):
+        """
+        Adds a control stream to the system while also inserting it into the system's parent node via HTTP POST.
+        :param control_stream: ControlStreamSchema to be used to define the control stream
+        :return:
+        """
+        print(f'Adding control stream: {control_stream.get_schema().model_dump_json(exclude_none=True)}')
+        api = self._parent_node.get_api_helper()
+        res = api.create_resource(APIResourceTypes.CONTROL_CHANNEL,
+                            control_stream.get_schema().model_dump_json(exclude_none=True, by_alias=True), req_headers={
+                'Content-Type': 'application/json'
+            }, parent_res_id=self.resource_id)
+        if res.ok:
+            control_stream_id = res.headers['Location'].split('/')[-1]
+            control_stream.set_id(control_stream_id)
+            self.control_channels.append(control_stream)
+        else:
+            raise Exception(f'Failed to create control stream: {control_stream.name}')
+        return control_stream
+
 
 class Datastream:
     should_poll: bool
@@ -324,11 +345,14 @@ def insert_observation_dict(self, obs_data: dict):
     #     return new_ds
 
 
-class ControlChannel:
-    # _cc_resource: ControlStream
-
-    def __init__(self):
-        pass
+# class ControlChannel:
+#     """
+#     * deprecated *
+#     """
+#     # _cc_resource: ControlStream
+#
+#     def __init__(self):
+#         pass
 
 
 class Observation:

pyproject.toml

@@ -10,7 +10,7 @@ python = "^3.12"
 pydantic = "^2.7.4"
 shapely = "^2.0.4"
 websockets = "^12.0"
-consys4py = "^0.0.1a1"
+consys4py = "^0.0.1a7"
 swecommondm = "^0.0.1a0"
 
 [tool.poetry.group.dev.dependencies]