374: rj/miniconf r=jordens a=jordens

- miniconf.py: make retain an option
- miniconf: add some checks, simplify


Co-authored-by: Robert Jördens <rj@quartiq.de>
This commit is contained in:
bors[bot] 2021-06-04 09:06:30 +00:00 committed by GitHub
commit 387e9f12b8
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
1 changed files with 42 additions and 36 deletions

View File

@ -16,6 +16,7 @@ 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."""
@ -33,64 +34,64 @@ 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()
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.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. """Callback function for when messages are received over MQTT.
Args: Args:
_client: The MQTT client. _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. payload: The payload of the message.
_qos: The quality-of-service level of the received packet _qos: The quality-of-service level of the received packet
properties: A dictionary of properties associated with the message. properties: A dictionary of properties associated with the message.
""" """
# Extract corrleation data from the properties if topic == self.response_topic:
correlation_data = json.loads(properties['correlation_data'][0].decode('ascii')) # Extract request_id corrleation data from the properties
request_id = int.from_bytes(
properties['correlation_data'][0], 'big')
# Get the request ID from the correlation data self.inflight[request_id].set_result(json.loads(payload))
request_id = correlation_data['request_id'] del self.inflight[request_id]
else:
LOGGER.warn('Unexpected message on "%s"', topic)
self.inflight[request_id].set_result(json.loads(payload)) async def command(self, path, value, retain=True):
del self.inflight[request_id]
async def command(self, path, value):
"""Write the provided data to the specified path. """Write the provided data to the specified path.
Args: Args:
path: The path to write the message to. path: The path to write the message to.
value: The value to write to the path. value: The value to write to the path.
retain: Retain the MQTT message changing the setting
by the broker.
Returns: Returns:
The response to the command as a dictionary. The response to the command as a dictionary.
""" """
setting_topic = f'{self.prefix}/settings/{path}' topic = f'{self.prefix}/settings/{path}'
response_topic = f'{self.prefix}/response/{self.uuid.hex}'
# 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 = json.dumps({
'request_id': request_id,
}).encode('ascii')
value = json.dumps(value)
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[request_id] = fut # Assign unique correlation data for response dispatch
self.client.publish(setting_topic, payload=value, qos=0, retain=True, assert self.request_id not in self.inflight
response_topic=response_topic, self.inflight[self.request_id] = fut
correlation_data=correlation_data) 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 return await fut
@ -100,16 +101,20 @@ def main():
description='Miniconf command line interface.', description='Miniconf command line interface.',
formatter_class=argparse.RawDescriptionHelpFormatter, formatter_class=argparse.RawDescriptionHelpFormatter,
epilog='''Examples: epilog='''Examples:
%(prog)s dt/sinara/stabilizer afe/0='"G2"' iir_ch/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]}' '{"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, parser.add_argument('-v', '--verbose', action='count', default=0,
help='Increase logging verbosity') help='Increase logging verbosity')
parser.add_argument('--broker', '-b', default='mqtt', type=str, parser.add_argument('--broker', '-b', default='mqtt', type=str,
help='The MQTT broker address') 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, parser.add_argument('prefix', type=str,
help='The MQTT topic prefix of the target') 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.') help='JSON encoded values for settings path keys.')
args = parser.parse_args() args = parser.parse_args()
@ -122,9 +127,10 @@ def main():
async def configure_settings(): async def configure_settings():
interface = await Miniconf.create(args.prefix, args.broker) interface = await Miniconf.create(args.prefix, args.broker)
for key_value in args.settings: for setting in args.settings:
path, value = key_value.split("=", 1) path, value = setting.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}') print(f'{path}: {response}')
if response['code'] != 0: if response['code'] != 0:
return response['code'] return response['code']