Adding general MQTT utility for stabilizer

stabilizer.py

@@ -0,0 +1,147 @@
+#!/usr/bin/python
+"""
+Author: Vertigo Designs, Ryan Summers
+
+Description: Provides an API for controlling Booster NGFW over MQTT.
+"""
+import argparse
+import asyncio
+import json
+
+from gmqtt  import Client as MqttClient
+
+
+def parse_value(value):
+    """ Parse a command-line value into the most appropriate associated python datatype. """
+    if value.isnumeric():
+        return int(value)
+    try:
+        return float(value)
+    except ValueError:
+        return value
+
+
+class MiniconfApi:
+    """ An asynchronous API for controlling Miniconf devices using the MQTT control interface. """
+
+    @classmethod
+    async def create(cls, identifier, broker):
+        """ Create a connection to MQTT for communication with booster. """
+        client = MqttClient(client_id='')
+        await client.connect(broker)
+        return cls(client, identifier)
+
+
+    def __init__(self, client, identifier):
+        """ Consructor.
+
+        Args:
+            client: A connected MQTT5 client.
+            identifier: The ID of the device to control.
+        """
+        self.response_topic = f'{identifier}/feedback'
+        self.client = client
+        self.identifier = identifier
+        self.command_complete = asyncio.Event()
+        self.client.on_message = self._handle_response
+        self.response = None
+
+        self.client.subscribe(self.response_topic)
+
+
+    def _handle_response(self, client, topic, payload, *_args, **_kwargs):
+        """ Callback function for when messages are received over MQTT.
+
+        Args:
+            client: The MQTT client.
+            topic: The topic that the message was received on.
+            payload: The payload of the message.
+        """
+        if topic != self.response_topic:
+            raise Exception(f'Unknown topic: {topic}')
+
+        # Indicate a response was received.
+        self.response = payload.decode('ascii')
+        self.command_complete.set()
+
+
+    async def _command(self, topic, message):
+        """ Send a command to a booster control topic.
+
+        Args:
+            topic: The topic to send the message to.
+            message: The message to send to the provided topic.
+
+        Returns:
+            The received response to the command.
+        """
+        self.command_complete.clear()
+        self.client.publish(topic, payload=message, qos=0, retain=False,
+                            response_topic=self.response_topic)
+        await self.command_complete.wait()
+
+        response = self.response
+        self.response = None
+
+        return response
+
+
+    async def set_setting(self, setting, message):
+        """ Change the provided setting with the provided data. """
+        return await self._command(f'{self.identifier}/settings/{setting}', message)
+
+
+    async def commit(self):
+        """ Commit staged settings to become active. """
+        return await self._command(f'{self.identifier}/commit', 'commit')
+
+
+async def configure_settings(args):
+    """ Configure an RF channel. """
+
+    # Establish a communication interface with Booster.
+    interface = await MiniconfApi.create(args.stabilizer, args.broker)
+
+    request = None
+
+    # In the exceptional case that this is a terminal value, there is no key available and only a
+    # single value.
+    if len(args.values) == 1 and '=' not in args.values[0]:
+        request = parse_value(args.values[0])
+    else:
+        # Convert all of the values into a key-value list.
+        request = dict()
+        for pair in args.values:
+            key, value = pair.split('=')
+            request[str(key)] = parse_value(value)
+        response = json.dumps(request)
+
+    # TODO: Should we escape quotes in the dumped request?
+    response = await interface.set_setting(args.setting, request)
+    print(f'+ {response}')
+
+    if args.commit:
+        response = await interface.commit()
+        print(f'+ {response}')
+
+
+def main():
+    """ Main program entry point. """
+    parser = argparse.ArgumentParser(
+        formatter_class=argparse.RawTextHelpFormatter,
+        description='Modify booster RF channel configuration')
+    parser.add_argument('--stabilizer', type=str, default='stabilizer',
+                        help='The identifier of the stabilizer to configure')
+    parser.add_argument('--setting', required=True, type=str, help='The setting path to configure')
+    parser.add_argument('--broker', default='10.34.16.1', type=str, help='The MQTT broker address')
+    parser.add_argument('values', nargs='+', type=str,
+                        help='The value of settings. key=value list or a single value is accepted.')
+    parser.add_argument('--commit', action='store_true',
+                        help='Specified true to commit after updating settings.')
+
+    loop = asyncio.get_event_loop()
+    loop.run_until_complete(configure_settings(parser.parse_args()))
+
+
+if __name__ == '__main__':
+    main()