update test scripts for commands, awaiting checks for MQTT subs on osh-core for further testing

Author: tipatterson-dev

conSys/comm/__init__.py

@@ -0,0 +1,194 @@
+import paho.mqtt.client as mqtt
+
+
+class MQTTClient:
+    def __init__(self, url, port=1883, username=None, password=None, path='mqtt', client_id="", transport='tcp'):
+        """
+    Wraps a paho mqtt client to provide a simple interface for interacting with the mqtt server that is customized
+    for this library.
+
+    :param url: url of the mqtt server
+    :param port: port the mqtt server is communicating over, default is 1883 or whichever port the main node is
+    using if in websocket mode
+    :param username: used if node is requiring authentication to access this service
+    :param password: used if node is requiring authentication to access this service
+    :param path: used for setting the path when using websockets (usually sensorhub/mqtt by default)
+    """
+        self.__url = url
+        self.__port = port
+        self.__path = path
+        self.__client_id = client_id
+        self.__transport = transport
+
+        self.__client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2)
+
+        if self.__transport == 'websockets':
+            self.__client.ws_set_options(path=self.__path)
+
+        if username is not None and password is not None:
+            self.__client.username_pw_set(username, password)
+            self.__client.tls_set()
+
+        self.__client.on_connect = self.on_connect
+        self.__client.on_subscribe = self.on_subscribe
+        self.__client.on_message = self.on_message
+        self.__client.on_publish = self.on_publish
+        self.__client.on_log = self.on_log
+        self.__client.on_disconnect = self.on_disconnect
+
+        self.__is_connected = False
+
+    @staticmethod
+    def on_connect(client, userdata, flags, rc, properties):
+        print(f'Connected with result code: {rc}')
+        print(f'{properties}')
+
+    @staticmethod
+    def on_subscribe(client, userdata, mid, granted_qos, properties):
+        print(f'Subscribed: {mid} {granted_qos}')
+
+    @staticmethod
+    def on_message(client, userdata, msg):
+        print(f'{msg.payload.decode("utf-8")}')
+
+    @staticmethod
+    def on_publish(client, userdata, mid):
+        print(f'Published: {mid}')
+
+    @staticmethod
+    def on_log(client, userdata, level, buf):
+        print(f'Log: {buf}')
+
+    @staticmethod
+    def on_disconnect(client, userdata, dc_flag, rc, properties):
+        print(f'Disconnected: {rc}')
+
+    def connect(self):
+        print(f'Connecting to {self.__url}:{self.__port}')
+        self.__client.connect(self.__url, self.__port)
+
+    def subscribe(self, topic, qos=0, msg_callback=None):
+        """
+        Subscribe to a topic, and optionally set a callback for when a message is received on that topic. To actually
+        retrieve any information you must set a callback.
+
+        :param topic: MQTT topic to subscribe to (example/topic)
+        :param qos: quality of service, 0, 1, or 2
+        :param msg_callback: callback with the form: callback(client, userdata, msg)
+        :return:
+        """
+        self.__client.subscribe(topic, qos)
+        if msg_callback is not None:
+            self.__client.message_callback_add(topic, msg_callback)
+
+    def publish(self, topic, payload=None, qos=0, retain=False):
+        self.__client.publish(topic, payload, qos, retain)
+
+    def disconnect(self):
+        self.__client.disconnect()
+
+    def set_on_connect(self, on_connect):
+        """
+        Set the on_connect callback for the MQTT client.
+
+        :param on_connect:
+        :return:
+        """
+        self.__client.on_connect = on_connect
+
+    def set_on_disconnect(self, on_disconnect):
+        """
+        Set the on_disconnect callback for the MQTT client.
+
+        :param on_disconnect:
+        :return:
+        """
+        self.__client.on_disconnect = on_disconnect
+
+    def set_on_subscribe(self, on_subscribe):
+        """
+        Set the on_subscribe callback for the MQTT client.
+
+        :param on_subscribe:
+        :return:
+        """
+        self.__client.on_subscribe = on_subscribe
+
+    def set_on_unsubscribe(self, on_unsubscribe):
+        """
+        Set the on_unsubscribe callback for the MQTT client.
+
+        :param on_unsubscribe:
+        :return:
+        """
+        self.__client.on_unsubscribe = on_unsubscribe
+
+    def set_on_publish(self, on_publish):
+        """
+        Set the on_publish callback for the MQTT client.
+
+        :param on_publish:
+        :return:
+        """
+        self.__client.on_publish = on_publish
+
+    def set_on_message(self, on_message):
+        """
+        Set the on_message callback for the MQTT client. It is recommended to set individual callbacks for each
+        subscribed topic.
+
+        :param on_message:
+        :return:
+        """
+        self.__client.on_message = on_message
+
+    def set_on_log(self, on_log):
+        """
+        Set the on_log callback for the MQTT client.
+
+        :param on_log:
+        :return:
+        """
+        self.__client.on_log = on_log
+
+    def set_on_message_callback(self, sub, on_message_callback):
+        """
+        Set the on_message callback for a specific topic.
+        :param sub:
+        :param on_message_callback:
+        :return:
+        """
+        self.__client.message_callback_add(sub, on_message_callback)
+
+    def start(self):
+        """
+        Start the MQTT client in a separate thread. This is required for the client to be able to receive messages.
+
+        :return:
+        """
+        self.__client.loop_start()
+
+    def stop(self):
+        """
+        Stop the MQTT client.\
+
+        :return:
+        """
+        self.__client.loop_stop()
+
+    def __toggle_is_connected(self):
+        self.__is_connected = not self.__is_connected
+
+    def is_connected(self):
+        return self.__is_connected
+
+    def publish(self, topic, msg):
+        self.__client.publish(topic, msg, 1)
+
+    @staticmethod
+    def publish_single(self, topic, msg):
+        self.__client.single(topic, msg, 0)
+
+    @staticmethod
+    def publish_multiple(self, topic, msgs):
+        self.__client.multiple(msgs, )

conSys/comm/mqtt.py

@@ -1 +1,10 @@
-import pytest
+import pytest
+
+sever_url = None
+
+
+@pytest.fixture(autouse=True)
+def server_url():
+    print('Setting up server url')
+    global server_url
+    server_url = 'http://localhost:8282/sensorhub'

tests/conftest.py

@@ -0,0 +1,111 @@
+import json
+import time
+from datetime import datetime
+
+from conSys import GeoJSONBody, Systems, ControlChannels, ObservationFormat, Commands
+from conSys.datamodels.commands import CommandJSON
+from conSys.datamodels.control_streams import ControlStreamJSONSchema, JSONControlChannelSchema
+from conSys.datamodels.swe_components import DataRecordSchema, TimeSchema, CountSchema, URI
+from conSys.comm.mqtt import MQTTClient
+
+server_url = "http://localhost:8282/sensorhub"
+geo_json_headers = {"Content-Type": "application/geo+json"}
+sml_json_headers = {"Content-Type": "application/sml+json"}
+json_headers = {"Content-Type": "application/json"}
+systems_list = []
+
+
+def test_setup():
+    geo_temp = GeoJSONBody(type='Feature', id='12345-commanded',
+                           description="Test Insertion of System via GEOJSON",
+                           properties={
+                               "featureType": "http://www.w3.org/ns/ssn/System",
+                               "name": f'Test System - GeoJSON',
+                               "uid": f'urn:test:client:geo-single',
+                               "description": "A Test System inserted from the Python Connected Systems API Client",
+                           })
+    resp = Systems.create_new_systems(server_url, geo_temp.model_dump_json(exclude_none=True, by_alias=True),
+                                      uname="admin",
+                                      pword="admin",
+                                      headers=geo_json_headers)
+    print(resp)
+
+    systems_list = Systems.list_all_systems(server_url, headers=json_headers).json()
+    system_id = systems_list["items"][0]["id"]
+
+    time_schema = TimeSchema(label="Test Control Channel Time", definition="http://test.com/Time", name="timestamp",
+                             uom=URI(href="http://test.com/TimeUOM"))
+    count_schema = CountSchema(label="Test Control Channel Count", definition="http://test.com/Count", name="testcount")
+
+    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)
+    ctl_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(ctl_resp)
+
+    control_streams = ControlChannels.list_all_control_streams(server_url).json()
+    command_json = CommandJSON(control_id=control_streams["items"][0]["id"],
+                               issue_time=datetime.now().isoformat() + 'Z',
+                               params={"timestamp": datetime.now().timestamp() * 1000, "testcount": 1})
+
+    print(f'Issuing Command: {command_json.model_dump_json(exclude_none=True, by_alias=True)}')
+    cmd_resp = Commands.send_commands_to_specific_control_stream(server_url, control_streams["items"][0]["id"],
+                                                                 command_json.model_dump_json(exclude_none=True,
+                                                                                              by_alias=True),
+                                                                 headers=json_headers)
+    print(cmd_resp)
+
+
+def test_subscribe_and_command():
+    mqtt_client = MQTTClient(url='localhost')
+
+    control_streams = ControlChannels.list_all_control_streams(server_url).json()
+    control_id = control_streams["items"][0]["id"]
+
+    mqtt_client.connect()
+
+    def on_message_command(client, userdata, msg):
+        print("Received Command")
+        print(f'{msg.payload.decode("utf-8")}')
+        control_stream_id = control_id
+        resp = {
+            'id': '*******',
+            'command@id': control_stream_id,
+            'statusCode': 'COMPLETED'
+        }
+        # client.publish(f'/api/controls/{control_stream_id}/status', payload=json.dumps(resp), qos=1)
+
+    def on_message_all(client, userdata, msg):
+        print(f'\nReceived Message:{msg}')
+        print(f'{msg.payload.decode("utf-8")}')
+        print(f'Topic: {msg.topic}\n')
+
+    mqtt_client.set_on_message_callback(f'/api/controls/{control_id}/commands', on_message_command)
+    mqtt_client.set_on_message_callback('#', on_message_all)
+    mqtt_client.subscribe('#')
+    mqtt_client.subscribe(f'/api/controls/{control_id}/commands')
+    mqtt_client.start()
+
+    time.sleep(2)
+    command_json = CommandJSON(control_id=control_streams["items"][0]["id"],
+                               issue_time=datetime.now().isoformat() + 'Z',
+                               params={"timestamp": datetime.now().timestamp() * 1000, "testcount": 1})
+
+    print(f'Issuing Command: {command_json.model_dump_json(exclude_none=True, by_alias=True)}')
+    cmd_resp = Commands.send_commands_to_specific_control_stream(server_url, control_streams["items"][0]["id"],
+                                                                 command_json.model_dump_json(exclude_none=True,
+                                                                                              by_alias=True),
+                                                                 headers=json_headers)
+    print(f'\n*****Command Response: {cmd_resp}*****')
+    status_resp = {
+        'id': '*******',
+        'command@id': "unknown",
+        'statusCode': 'COMPLETED'
+    }
+    Commands.add_command_status_reports(server_url, "0", json.dumps(status_resp))