Reorganizing miniconf, adding loopback test

This commit is contained in:
Ryan Summers 2021-08-05 15:05:55 +02:00
parent f7c77bd860
commit 1d082c28c3
13 changed files with 247 additions and 51 deletions

148
hitl/loopback.py Normal file
View File

@ -0,0 +1,148 @@
#!/usr/bin/python3
"""
Author: Vertigo Designs, Ryan Summers
Description: Loop-back integration tests for Stabilizer hardware
"""
import argparse
import asyncio
import json
import sys
from gmqtt import Client as MqttClient
from miniconf import Miniconf
# The minimum allowably loopback voltage error (difference between output set point and input
# measured value).
MINIMUM_VOLTAGE_ERROR = 0.010
def _voltage_to_machine_units(voltage):
""" Convert a voltage to IIR machine units.
Args:
voltage: The voltage to convert
Returns:
The IIR machine-units associated with the voltage.
"""
dac_range = 4.096 * 2.5
assert abs(voltage) <= dac_range, 'Voltage out-of-range'
return voltage / dac_range * 0x7FFF
def static_iir_output(output_voltage):
""" Generate IIR configuration for a static output voltage.
Args:
output_voltage: The desired static IIR output voltage.
Returns
The IIR configuration to send over Miniconf.
"""
machine_units = _voltage_to_machine_units(output_voltage)
return {
'y_min': machine_units,
'y_max': machine_units,
'y_offset': 0,
'ba': [1, 0, 0, 0, 0],
}
class TelemetryReader:
""" Helper utility to read Stabilizer telemetry. """
@classmethod
async def create(cls, prefix, broker):
"""Create a connection to the broker and an MQTT device using it."""
client = MqttClient(client_id='')
await client.connect(broker)
return cls(client, prefix)
def __init__(self, client, prefix):
""" Constructor. """
self.client = client
self._telemetry = []
self.client.on_message = self.handle_telemetry
self._telemetry_topic = f'{prefix}/telemetry'
self.client.subscribe(self._telemetry_topic)
def handle_telemetry(self, _client, topic, payload, _qos, _properties):
""" Handle incoming telemetry messages over MQTT. """
assert topic == self._telemetry_topic
self._telemetry.append(json.loads(payload))
def get_latest(self):
""" Get the latest telemetry message received. """
return self._telemetry[-1]
async def test_loopback(miniconf, telemetry, set_point):
""" Test loopback operation of Stabilizer.
Note:
Loopback is tested by configuring DACs for static output and verifying telemetry reports the
ADCs are measuring those values. Output OUTx should be connected in a loopback configuration
to INx on the device.
Args:
miniconf: The miniconf configuration interface.
telemetry: a helper utility to read inbound telemetry.
set_point: The desired output voltage to test.
"""
print(f'Testing loopback for Vout = {set_point:.2f}')
print('---------------------------------')
# Configure the IIRs to output at the set point
assert (await miniconf.command('iir_ch/0/0', static_iir_output(set_point)))['code'] == 0
assert (await miniconf.command('iir_ch/1/0', static_iir_output(set_point)))['code'] == 0
assert (await miniconf.command('telemetry_period', 1))['code'] == 0
# Wait for telemetry values to update.
await asyncio.sleep(2.0)
# Verify the ADCs are receiving the setpoint voltage.
tolerance = max(0.05 * set_point, MINIMUM_VOLTAGE_ERROR)
latest_values = telemetry.get_latest()
print(f'Latest telemtry: {latest_values}')
assert abs(latest_values['adcs'][0] - set_point) < tolerance
assert abs(latest_values['adcs'][1] - set_point) < tolerance
print('PASS')
print('')
def main():
""" Main program entry point. """
parser = argparse.ArgumentParser(description='Loopback tests for Stabilizer HITL testing',)
parser.add_argument('prefix', type=str,
help='The MQTT topic prefix of the target')
parser.add_argument('--broker', '-b', default='mqtt', type=str,
help='The MQTT broker address')
args = parser.parse_args()
async def test():
""" The actual testing being completed. """
interface = await Miniconf.create(args.prefix, args.broker)
telemetry = await TelemetryReader.create(args.prefix, args.broker)
# Test loopback with a static 1V output of the DACs.
await test_loopback(interface, telemetry, 1.0)
# Repeat test with AFE = 2x
print('Configuring AFEs to 2x input')
assert (await interface.command('afe/0', "G2"))['code'] == 0
assert (await interface.command('afe/1', "G2"))['code'] == 0
await test_loopback(interface, telemetry, 1.0)
# Test with 0V output
await test_loopback(interface, telemetry, 0.0)
loop = asyncio.get_event_loop()
sys.exit(loop.run_until_complete(test()))
if __name__ == '__main__':
main()

6
miniconf-py/README.md Normal file
View File

@ -0,0 +1,6 @@
# Miniconf Python Utility
This directory contains a Python package for interacting with Miniconf utilities.
## Installation
Run `pip install .` from this directory to install the `miniconf` package.

View File

@ -0,0 +1,10 @@
Metadata-Version: 1.0
Name: miniconf
Version: 1.0.0
Summary: Utilities for configuring Miniconf-configurable devices
Home-page: https://github.com/quartiq/miniconf
Author: Ryan Summers, Robert Jördens
Author-email: UNKNOWN
License: UNKNOWN
Description: UNKNOWN
Platform: UNKNOWN

View File

@ -0,0 +1,8 @@
setup.py
miniconf/__init__.py
miniconf/miniconf.py
miniconf.egg-info/PKG-INFO
miniconf.egg-info/SOURCES.txt
miniconf.egg-info/dependency_links.txt
miniconf.egg-info/requires.txt
miniconf.egg-info/top_level.txt

View File

@ -0,0 +1 @@

View File

@ -0,0 +1 @@
gmqtt

View File

@ -0,0 +1 @@
miniconf

View File

@ -0,0 +1,4 @@
#!/usr/bin/python3
from .miniconf import Miniconf
__version__ = '1.0.0'

View File

@ -0,0 +1,53 @@
#!/usr/bin/python3
import sys
import argparse
from .miniconf import Miniconf
def main():
""" Main program entry point. """
parser = argparse.ArgumentParser(
description='Miniconf command line interface.',
formatter_class=argparse.RawDescriptionHelpFormatter,
epilog='''Examples:
%(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')
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="PATH=VALUE", nargs='+',
help='JSON encoded values for settings path keys.')
args = parser.parse_args()
logging.basicConfig(
format='%(asctime)s [%(levelname)s] %(name)s: %(message)s',
level=logging.WARN - 10*args.verbose)
loop = asyncio.get_event_loop()
async def configure_settings():
interface = await Miniconf.create(args.prefix, args.broker)
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}')
if response['code'] != 0:
return response['code']
return 0
sys.exit(loop.run_until_complete(configure_settings()))
if __name__ == '__main__':
main()

View File

@ -5,11 +5,9 @@ Author: Vertigo Designs, Ryan Summers
Description: Provides an API for controlling Miniconf devices over MQTT. Description: Provides an API for controlling Miniconf devices over MQTT.
""" """
import argparse
import asyncio import asyncio
import json import json
import logging import logging
import sys
import uuid import uuid
from gmqtt import Client as MqttClient from gmqtt import Client as MqttClient
@ -60,7 +58,7 @@ class Miniconf:
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: else:
LOGGER.warn('Unexpected message on "%s"', topic) LOGGER.warning('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.
@ -93,51 +91,3 @@ class Miniconf:
correlation_data=correlation_data) correlation_data=correlation_data)
return await fut return await fut
def main():
""" Main program entry point. """
parser = argparse.ArgumentParser(
description='Miniconf command line interface.',
formatter_class=argparse.RawDescriptionHelpFormatter,
epilog='''Examples:
%(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')
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="PATH=VALUE", nargs='+',
help='JSON encoded values for settings path keys.')
args = parser.parse_args()
logging.basicConfig(
format='%(asctime)s [%(levelname)s] %(name)s: %(message)s',
level=logging.WARN - 10*args.verbose)
loop = asyncio.get_event_loop()
async def configure_settings():
interface = await Miniconf.create(args.prefix, args.broker)
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}')
if response['code'] != 0:
return response['code']
return 0
sys.exit(loop.run_until_complete(configure_settings()))
if __name__ == '__main__':
main()

14
miniconf-py/setup.py Normal file
View File

@ -0,0 +1,14 @@
from setuptools import setup, find_packages
import miniconf
setup(name='miniconf',
version=miniconf.__version__,
author='Ryan Summers, Robert Jördens',
description='Utilities for configuring Miniconf-configurable devices',
url='https://github.com/quartiq/miniconf',
packages=find_packages(),
install_requires=[
'gmqtt'
],
)