From 919f68c93445e59fb5bf9d371bc832f1a2c877b2 Mon Sep 17 00:00:00 2001 From: Ryan Summers Date: Wed, 5 May 2021 14:30:32 +0200 Subject: [PATCH 1/5] Updating miniconf utility after testing --- miniconf.py | 21 ++++++++++----------- 1 file changed, 10 insertions(+), 11 deletions(-) diff --git a/miniconf.py b/miniconf.py index 40f163e..36ea0db 100644 --- a/miniconf.py +++ b/miniconf.py @@ -14,8 +14,6 @@ import uuid from gmqtt import Client as MqttClient -logger = logging.getLogger(__name__) - class Miniconf: """An asynchronous API for controlling Miniconf devices using MQTT.""" @@ -34,13 +32,14 @@ class Miniconf: client: A connected MQTT5 client. prefix: The MQTT toptic prefix of the device to control. """ - self.uuid = uuid.uuid1(prefix) + 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.logger = logging.getLogger(__name__) def _handle_response(self, _client, _topic, payload, _qos, properties): """Callback function for when messages are received over MQTT. @@ -54,23 +53,23 @@ class Miniconf: """ # Extract corrleation data from the properties try: - correlation_data = json.loads(properties['correlation_data']) + correlation_data = json.loads(properties['correlation_data'][0].decode('ascii')) except (json.decoder.JSONDecodeError, KeyError): - logger.warning('Ignoring message with invalid correlation data') + self.logger.warning('Ignoring message with invalid correlation data') return # Validate the correlation data. try: if correlation_data['id'] != self.uuid.hex: - logger.info('Ignoring correlation data for different ID') + self.logger.info('Ignoring correlation data for different ID') return pid = correlation_data['pid'] except KeyError: - logger.warning('Ignoring unknown correlation data: %s', correlation_data) + self.logger.warning('Ignoring unknown correlation data: %s', correlation_data) return if pid not in self.inflight: - logger.warning('Unexpected pid: %s', pid) + self.logger.warning('Unexpected pid: %s', pid) return try: @@ -78,7 +77,7 @@ class Miniconf: self.inflight[pid].set_result((response['code'], response['msg'])) del self.inflight[pid] except json.decoder.JSONDecodeError: - logger.warning('Invalid response format: %s', payload) + self.logger.warning('Invalid response format: %s', payload) async def command(self, path, value): @@ -103,10 +102,10 @@ class Miniconf: correlation_data = json.dumps({ 'id': self.uuid.hex, 'pid': pid, - }) + }).encode('ascii') value = json.dumps(value) - logger.info('Sending %s to "%s"', value, setting_topic) + self.logger.info('Sending %s to "%s"', value, setting_topic) fut = asyncio.get_running_loop().create_future() self.inflight[pid] = fut From f9b1b8df13cbda6e979729c4b63f1809c7dd820c Mon Sep 17 00:00:00 2001 From: Ryan Summers Date: Wed, 5 May 2021 14:33:34 +0200 Subject: [PATCH 2/5] Adding comment about subscription failures --- src/net/mqtt_interface.rs | 2 ++ 1 file changed, 2 insertions(+) diff --git a/src/net/mqtt_interface.rs b/src/net/mqtt_interface.rs index cef26dd..42ab41e 100644 --- a/src/net/mqtt_interface.rs +++ b/src/net/mqtt_interface.rs @@ -115,6 +115,8 @@ where String::from(self.settings_prefix.as_str()); settings_topic.push_str("/#").unwrap(); + // We do not currently handle or process potential subscription failures. Instead, this + // failure will be logged through the stabilizer logging interface. self.mqtt.subscribe(&settings_topic, &[]).unwrap(); self.subscribed = true; } From e6180de147d70aaf3943baa359b05291ff7b9bb2 Mon Sep 17 00:00:00 2001 From: Ryan Summers Date: Wed, 5 May 2021 14:38:43 +0200 Subject: [PATCH 3/5] Fixing style --- src/net/mqtt_interface.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/net/mqtt_interface.rs b/src/net/mqtt_interface.rs index 42ab41e..2a55b90 100644 --- a/src/net/mqtt_interface.rs +++ b/src/net/mqtt_interface.rs @@ -88,7 +88,7 @@ where self.network_was_reset = true; self.mqtt.network_stack.handle_link_reset(); } - _ => {}, + _ => {} }; let mqtt_connected = match self.mqtt.is_connected() { From 1ad3f1d1a80720fa839a09220e4c9ccec14fe3f3 Mon Sep 17 00:00:00 2001 From: Ryan Summers Date: Wed, 5 May 2021 17:09:51 +0200 Subject: [PATCH 4/5] Cleaning up python script --- miniconf.py | 48 ++++++++++++++---------------------------------- 1 file changed, 14 insertions(+), 34 deletions(-) diff --git a/miniconf.py b/miniconf.py index 36ea0db..4878419 100644 --- a/miniconf.py +++ b/miniconf.py @@ -38,7 +38,7 @@ class Miniconf: self.prefix = prefix self.inflight = {} self.client.on_message = self._handle_response - self.client.subscribe(f'{prefix}/response/#') + self.client.subscribe(f'{prefix}/response/{self.uuid.hex}') self.logger = logging.getLogger(__name__) def _handle_response(self, _client, _topic, payload, _qos, properties): @@ -52,32 +52,13 @@ class Miniconf: properties: A dictionary of properties associated with the message. """ # Extract corrleation data from the properties - try: - correlation_data = json.loads(properties['correlation_data'][0].decode('ascii')) - except (json.decoder.JSONDecodeError, KeyError): - self.logger.warning('Ignoring message with invalid correlation data') - return + correlation_data = json.loads(properties['correlation_data'][0].decode('ascii')) - # Validate the correlation data. - try: - if correlation_data['id'] != self.uuid.hex: - self.logger.info('Ignoring correlation data for different ID') - return - pid = correlation_data['pid'] - except KeyError: - self.logger.warning('Ignoring unknown correlation data: %s', correlation_data) - return + # Get the request ID from the correlation data + request_id = correlation_data['request_id'] - if pid not in self.inflight: - self.logger.warning('Unexpected pid: %s', pid) - return - - try: - response = json.loads(payload) - self.inflight[pid].set_result((response['code'], response['msg'])) - del self.inflight[pid] - except json.decoder.JSONDecodeError: - self.logger.warning('Invalid response format: %s', payload) + self.inflight[request_id].set_result(json.loads(payload)) + del self.inflight[request_id] async def command(self, path, value): @@ -92,23 +73,22 @@ class Miniconf: a use-readable message indicating further information. """ setting_topic = f'{self.prefix}/settings/{path}' - response_topic = f'{self.prefix}/response/{path}' + response_topic = f'{self.prefix}/response/{self.uuid.hex}' # Assign a unique identifier to this update request. - pid = self.request_id + request_id = self.request_id self.request_id += 1 - assert pid not in self.inflight, 'Invalid PID encountered' + assert request_id not in self.inflight, 'Invalid ID encountered' correlation_data = json.dumps({ - 'id': self.uuid.hex, - 'pid': pid, + 'request_id': request_id, }).encode('ascii') value = json.dumps(value) self.logger.info('Sending %s to "%s"', value, setting_topic) fut = asyncio.get_running_loop().create_future() - self.inflight[pid] = fut + self.inflight[request_id] = fut self.client.publish(setting_topic, payload=value, qos=0, retain=True, response_topic=response_topic, correlation_data=correlation_data) @@ -145,10 +125,10 @@ def main(): interface = await Miniconf.create(args.prefix, args.broker) for key_value in args.settings: path, value = key_value.split("=", 1) - code, response = await interface.command(path, json.loads(value)) + response = await interface.command(path, json.loads(value)) print(f'{path}: {response}') - if code != 0: - return code + if response['code'] != 0: + return response['code'] return 0 sys.exit(loop.run_until_complete(configure_settings())) From 05dc80709e93669e016c6e2f9b32f0f886759e10 Mon Sep 17 00:00:00 2001 From: Ryan Summers Date: Wed, 5 May 2021 17:32:07 +0200 Subject: [PATCH 5/5] Updating docs and logger --- miniconf.py | 7 +++---- 1 file changed, 3 insertions(+), 4 deletions(-) diff --git a/miniconf.py b/miniconf.py index 4878419..7f88f65 100644 --- a/miniconf.py +++ b/miniconf.py @@ -14,6 +14,7 @@ import uuid from gmqtt import Client as MqttClient +LOGGER = logging.getLogger(__name__) class Miniconf: """An asynchronous API for controlling Miniconf devices using MQTT.""" @@ -39,7 +40,6 @@ class Miniconf: self.inflight = {} self.client.on_message = self._handle_response self.client.subscribe(f'{prefix}/response/{self.uuid.hex}') - self.logger = logging.getLogger(__name__) def _handle_response(self, _client, _topic, payload, _qos, properties): """Callback function for when messages are received over MQTT. @@ -69,8 +69,7 @@ class Miniconf: value: The value to write to the path. Returns: - (code, msg) tuple as a response to the command. `code` is zero for success and `msg` is - a use-readable message indicating further information. + The response to the command as a dictionary. """ setting_topic = f'{self.prefix}/settings/{path}' response_topic = f'{self.prefix}/response/{self.uuid.hex}' @@ -85,7 +84,7 @@ class Miniconf: }).encode('ascii') value = json.dumps(value) - self.logger.info('Sending %s to "%s"', value, setting_topic) + LOGGER.info('Sending %s to "%s"', value, setting_topic) fut = asyncio.get_running_loop().create_future() self.inflight[request_id] = fut