From 28aef35d8779871524a4f25a6329688967900dec Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Robert=20J=C3=B6rdens?= Date: Mon, 31 May 2021 11:42:09 +0200 Subject: [PATCH 1/3] miniconf.py: make retain an option --- miniconf.py | 26 ++++++++++++++------------ 1 file changed, 14 insertions(+), 12 deletions(-) diff --git a/miniconf.py b/miniconf.py index 7f88f65..542f29b 100644 --- a/miniconf.py +++ b/miniconf.py @@ -16,6 +16,7 @@ from gmqtt import Client as MqttClient LOGGER = logging.getLogger(__name__) + class Miniconf: """An asynchronous API for controlling Miniconf devices using MQTT.""" @@ -51,22 +52,21 @@ class Miniconf: _qos: The quality-of-service level of the received packet properties: A dictionary of properties associated with the message. """ - # Extract corrleation data from the properties - correlation_data = json.loads(properties['correlation_data'][0].decode('ascii')) - - # Get the request ID from the correlation data - request_id = correlation_data['request_id'] + # Extract request_id corrleation data from the properties + request_id = int.from_bytes( + properties['correlation_data'][0], 'big') self.inflight[request_id].set_result(json.loads(payload)) del self.inflight[request_id] - - async def command(self, path, value): + async def command(self, path, value, retain=True): """Write the provided data to the specified path. Args: path: The path to write the message to. value: The value to write to the path. + retain: Retain the MQTT message changing the setting + by the broker. Returns: The response to the command as a dictionary. @@ -79,16 +79,14 @@ class Miniconf: self.request_id += 1 assert request_id not in self.inflight, 'Invalid ID encountered' - correlation_data = json.dumps({ - 'request_id': request_id, - }).encode('ascii') + correlation_data = request_id.to_bytes(4, 'big') value = json.dumps(value) LOGGER.info('Sending %s to "%s"', value, setting_topic) fut = asyncio.get_running_loop().create_future() self.inflight[request_id] = fut - self.client.publish(setting_topic, payload=value, qos=0, retain=True, + self.client.publish(setting_topic, payload=value, qos=0, retain=retain, response_topic=response_topic, correlation_data=correlation_data) return await fut @@ -107,6 +105,9 @@ def main(): help='Increase logging verbosity') parser.add_argument('--broker', '-b', default='mqtt', type=str, help='The MQTT broker address') + parser.add_argument('--no-retain', '-n', default=False, + action='store_true', + help='Do not retain the affected settings') parser.add_argument('prefix', type=str, help='The MQTT topic prefix of the target') parser.add_argument('settings', metavar="KEY=VALUE", nargs='+', @@ -124,7 +125,8 @@ def main(): interface = await Miniconf.create(args.prefix, args.broker) for key_value in args.settings: path, value = key_value.split("=", 1) - response = await interface.command(path, json.loads(value)) + response = await interface.command(path, json.loads(value), + not args.no_retain) print(f'{path}: {response}') if response['code'] != 0: return response['code'] From af874c2eef712089436d594b45ff85d0d7a6a360 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Robert=20J=C3=B6rdens?= Date: Tue, 1 Jun 2021 18:32:01 +0200 Subject: [PATCH 2/3] miniconf: add some checks, simplify --- miniconf.py | 59 ++++++++++++++++++++++++++++------------------------- 1 file changed, 31 insertions(+), 28 deletions(-) diff --git a/miniconf.py b/miniconf.py index 542f29b..ca47248 100644 --- a/miniconf.py +++ b/miniconf.py @@ -34,30 +34,33 @@ class Miniconf: client: A connected MQTT5 client. prefix: The MQTT toptic prefix of the device to control. """ - self.uuid = uuid.uuid1() self.request_id = 0 self.client = client self.prefix = prefix self.inflight = {} self.client.on_message = self._handle_response - self.client.subscribe(f'{prefix}/response/{self.uuid.hex}') + self.response_topic = f'{prefix}/response/{uuid.uuid1().hex}' + self.client.subscribe(self.response_topic) - def _handle_response(self, _client, _topic, payload, _qos, properties): + def _handle_response(self, _client, topic, payload, _qos, properties): """Callback function for when messages are received over MQTT. Args: _client: The MQTT client. - _topic: The topic that the message was received on. + topic: The topic that the message was received on. payload: The payload of the message. _qos: The quality-of-service level of the received packet properties: A dictionary of properties associated with the message. """ - # Extract request_id corrleation data from the properties - request_id = int.from_bytes( - properties['correlation_data'][0], 'big') + if topic == self.response_topic: + # Extract request_id corrleation data from the properties + request_id = int.from_bytes( + properties['correlation_data'][0], 'big') - self.inflight[request_id].set_result(json.loads(payload)) - del self.inflight[request_id] + self.inflight[request_id].set_result(json.loads(payload)) + del self.inflight[request_id] + else: + LOGGER.warn('Unexpected message on "%s"', topic) async def command(self, path, value, retain=True): """Write the provided data to the specified path. @@ -71,24 +74,24 @@ class Miniconf: Returns: The response to the command as a dictionary. """ - setting_topic = f'{self.prefix}/settings/{path}' - response_topic = f'{self.prefix}/response/{self.uuid.hex}' + topic = f'{self.prefix}/settings/{path}' - # Assign a unique identifier to this update request. - request_id = self.request_id - self.request_id += 1 - assert request_id not in self.inflight, 'Invalid ID encountered' - - correlation_data = request_id.to_bytes(4, 'big') - - value = json.dumps(value) - LOGGER.info('Sending %s to "%s"', value, setting_topic) fut = asyncio.get_running_loop().create_future() - self.inflight[request_id] = fut - self.client.publish(setting_topic, payload=value, qos=0, retain=retain, - response_topic=response_topic, - correlation_data=correlation_data) + # Assign unique correlation data for response dispatch + assert self.request_id not in self.inflight + self.inflight[self.request_id] = fut + correlation_data = self.request_id.to_bytes(4, 'big') + self.request_id += 1 + + payload = json.dumps(value) + LOGGER.info('Sending "%s" to "%s"', value, topic) + + self.client.publish( + topic, payload=payload, qos=0, retain=retain, + response_topic=self.response_topic, + correlation_data=correlation_data) + return await fut @@ -98,7 +101,7 @@ def main(): description='Miniconf command line interface.', formatter_class=argparse.RawDescriptionHelpFormatter, epilog='''Examples: -%(prog)s dt/sinara/stabilizer afe/0='"G2"' iir_ch/0/0=\ +%(prog)s dt/sinara/stabilizer/00-11-22-33-aa-bb afe/0='"G2"' iir_ch/0/0=\ '{"y_min": -32767, "y_max": 32767, "y_offset": 0, "ba": [1.0, 0, 0, 0, 0]}' ''') parser.add_argument('-v', '--verbose', action='count', default=0, @@ -110,7 +113,7 @@ def main(): help='Do not retain the affected settings') parser.add_argument('prefix', type=str, help='The MQTT topic prefix of the target') - parser.add_argument('settings', metavar="KEY=VALUE", nargs='+', + parser.add_argument('settings', metavar="PATH=VALUE", nargs='+', help='JSON encoded values for settings path keys.') args = parser.parse_args() @@ -123,8 +126,8 @@ def main(): async def configure_settings(): interface = await Miniconf.create(args.prefix, args.broker) - for key_value in args.settings: - path, value = key_value.split("=", 1) + for setting in args.settings: + path, value = setting.split("=", 1) response = await interface.command(path, json.loads(value), not args.no_retain) print(f'{path}: {response}') From 77ae4363fd3e30f5327e7cad2a159f52726897dc Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Robert=20J=C3=B6rdens?= Date: Fri, 4 Jun 2021 11:05:49 +0200 Subject: [PATCH 3/3] miniconf: update example usage --- miniconf.py | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/miniconf.py b/miniconf.py index ca47248..644e026 100644 --- a/miniconf.py +++ b/miniconf.py @@ -101,8 +101,9 @@ def main(): description='Miniconf command line interface.', formatter_class=argparse.RawDescriptionHelpFormatter, epilog='''Examples: -%(prog)s dt/sinara/stabilizer/00-11-22-33-aa-bb afe/0='"G2"' iir_ch/0/0=\ -'{"y_min": -32767, "y_max": 32767, "y_offset": 0, "ba": [1.0, 0, 0, 0, 0]}' +%(prog)s dt/sinara/dual-iir/00-11-22-33-aa-bb iir_ch/0/0=\ +'{"y_min":-32767,"y_max":32767,"y_offset":0,"ba":[1.0,0,0,0,0]}' +%(prog)s dt/sinara/lockin/00-11-22-33-aa-bb afe/0='"G2"'\ ''') parser.add_argument('-v', '--verbose', action='count', default=0, help='Increase logging verbosity')