pounder_test/stabilizer.py

#!/usr/bin/python
"""
Author: Vertigo Designs, Ryan Summers

Description: Provides an API for controlling Stabilizer over Miniconf (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 the device. """
        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 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 stabilizer.
    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)

    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(description='Stabilizer settings modification utility')
    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()