miniconf: add some checks, simplify
This commit is contained in:
parent
28aef35d87
commit
af874c2eef
59
miniconf.py
59
miniconf.py
@ -34,30 +34,33 @@ 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 request_id corrleation data from the properties
|
if topic == self.response_topic:
|
||||||
request_id = int.from_bytes(
|
# Extract request_id corrleation data from the properties
|
||||||
properties['correlation_data'][0], 'big')
|
request_id = int.from_bytes(
|
||||||
|
properties['correlation_data'][0], 'big')
|
||||||
|
|
||||||
self.inflight[request_id].set_result(json.loads(payload))
|
self.inflight[request_id].set_result(json.loads(payload))
|
||||||
del self.inflight[request_id]
|
del self.inflight[request_id]
|
||||||
|
else:
|
||||||
|
LOGGER.warn('Unexpected message on "%s"', topic)
|
||||||
|
|
||||||
async def command(self, path, value, retain=True):
|
async def command(self, path, value, retain=True):
|
||||||
"""Write the provided data to the specified path.
|
"""Write the provided data to the specified path.
|
||||||
@ -71,24 +74,24 @@ class Miniconf:
|
|||||||
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 = 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()
|
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=retain,
|
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
|
||||||
|
|
||||||
|
|
||||||
@ -98,7 +101,7 @@ 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/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]}'
|
'{"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,
|
parser.add_argument('-v', '--verbose', action='count', default=0,
|
||||||
@ -110,7 +113,7 @@ def main():
|
|||||||
help='Do not retain the affected settings')
|
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()
|
||||||
@ -123,8 +126,8 @@ 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)
|
not args.no_retain)
|
||||||
print(f'{path}: {response}')
|
print(f'{path}: {response}')
|
||||||
|
Loading…
Reference in New Issue
Block a user