Merge branch 'master' into feature/telemetry

This commit is contained in:
Ryan Summers 2021-05-05 17:47:34 +02:00
commit e9c3e24863
2 changed files with 21 additions and 41 deletions

View File

@ -14,8 +14,7 @@ import uuid
from gmqtt import Client as MqttClient from gmqtt import Client as MqttClient
logger = logging.getLogger(__name__) LOGGER = logging.getLogger(__name__)
class Miniconf: class Miniconf:
"""An asynchronous API for controlling Miniconf devices using MQTT.""" """An asynchronous API for controlling Miniconf devices using MQTT."""
@ -34,13 +33,13 @@ class Miniconf:
client: A connected MQTT5 client. client: A connected MQTT5 client.
prefix: The MQTT toptic prefix of the device to control. prefix: The MQTT toptic prefix of the device to control.
""" """
self.uuid = uuid.uuid1(prefix) self.uuid = uuid.uuid1()
self.request_id = 0 self.request_id = 0
self.client = client self.client = client
self.prefix = prefix self.prefix = prefix
self.inflight = {} self.inflight = {}
self.client.on_message = self._handle_response self.client.on_message = self._handle_response
self.client.subscribe(f'{prefix}/response/#') self.client.subscribe(f'{prefix}/response/{self.uuid.hex}')
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. """Callback function for when messages are received over MQTT.
@ -53,32 +52,13 @@ class Miniconf:
properties: A dictionary of properties associated with the message. properties: A dictionary of properties associated with the message.
""" """
# Extract corrleation data from the properties # Extract corrleation data from the properties
try: correlation_data = json.loads(properties['correlation_data'][0].decode('ascii'))
correlation_data = json.loads(properties['correlation_data'])
except (json.decoder.JSONDecodeError, KeyError):
logger.warning('Ignoring message with invalid correlation data')
return
# Validate the correlation data. # Get the request ID from the correlation data
try: request_id = correlation_data['request_id']
if correlation_data['id'] != self.uuid.hex:
logger.info('Ignoring correlation data for different ID')
return
pid = correlation_data['pid']
except KeyError:
logger.warning('Ignoring unknown correlation data: %s', correlation_data)
return
if pid not in self.inflight: self.inflight[request_id].set_result(json.loads(payload))
logger.warning('Unexpected pid: %s', pid) del self.inflight[request_id]
return
try:
response = json.loads(payload)
self.inflight[pid].set_result((response['code'], response['msg']))
del self.inflight[pid]
except json.decoder.JSONDecodeError:
logger.warning('Invalid response format: %s', payload)
async def command(self, path, value): async def command(self, path, value):
@ -89,27 +69,25 @@ class Miniconf:
value: The value to write to the path. value: The value to write to the path.
Returns: Returns:
(code, msg) tuple as a response to the command. `code` is zero for success and `msg` is The response to the command as a dictionary.
a use-readable message indicating further information.
""" """
setting_topic = f'{self.prefix}/settings/{path}' 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. # Assign a unique identifier to this update request.
pid = self.request_id request_id = self.request_id
self.request_id += 1 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({ correlation_data = json.dumps({
'id': self.uuid.hex, 'request_id': request_id,
'pid': pid, }).encode('ascii')
})
value = json.dumps(value) value = json.dumps(value)
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() 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, self.client.publish(setting_topic, payload=value, qos=0, retain=True,
response_topic=response_topic, response_topic=response_topic,
correlation_data=correlation_data) correlation_data=correlation_data)
@ -146,10 +124,10 @@ def main():
interface = await Miniconf.create(args.prefix, args.broker) interface = await Miniconf.create(args.prefix, args.broker)
for key_value in args.settings: for key_value in args.settings:
path, value = key_value.split("=", 1) 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}') print(f'{path}: {response}')
if code != 0: if response['code'] != 0:
return code return response['code']
return 0 return 0
sys.exit(loop.run_until_complete(configure_settings())) sys.exit(loop.run_until_complete(configure_settings()))

View File

@ -75,6 +75,8 @@ where
String::from(self.settings_prefix.as_str()); String::from(self.settings_prefix.as_str());
settings_topic.push_str("/#").unwrap(); 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.mqtt.subscribe(&settings_topic, &[]).unwrap();
self.subscribed = true; self.subscribed = true;
} }