kai/pounder_test
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Merge #425

Browse Source 
425: Feature/loopback hitl r=ryan-summers a=ryan-summers

This PR fixes #354 by implementing loop-back HITL tests for Stabilizer.

Co-authored-by: Ryan Summers <ryan.summers@vertigo-designs.com>
This commit is contained in:
bors[bot] 2021-08-09 11:50:06 +00:00 committed by GitHub
commit 555f1e2d1b
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
5 changed files with 185 additions and 153 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

19
docs/pages/usage.md
View File

@ -25,15 +25,22 @@ to another. Disambiguation of devices is done by using Stabilizer's MAC address.
Settings are specific to an application. If two identical settings exist for two different
applications, each application maintains its own independent value.
### Setup
### Installation
Install the Miniconf configuration utilities:
```
python -m pip install git+https://github.com/quartiq/miniconf#subdirectory=miniconf-py
```
In order to use `miniconf.py`, run the following command:
To use `miniconf`, execute it as follows:
```
python -m pip install gmqtt
python -m miniconf --help
```
Miniconf also exposes a programmatic Python API, so it's possible to write automation scripting of
Stabilizer as well.
### Usage
The `miniconf.py` script utilizes a unique "device prefix". The device prefix is always of the
The Miniconf Python utility utilizes a unique "device prefix". The device prefix is always of the
form `dt/sinara/<app>/<mac-address>`, where `<app>` is the name of the application and
`<mac-address>` is the MAC address of the device, formatted with delimiting dashes.
@ -46,7 +53,7 @@ used:
* `value` = `{"ip": [192, 168, 0, 1], "port": 4000}`
```
python miniconf.py --broker 10.34.16.10 dt/sinara/dual-iir/00-11-22-33-44-55 stream_target='{"ip": [10, 34, 16, 123], "port": 4000}'
python -m miniconf --broker 10.34.16.10 dt/sinara/dual-iir/00-11-22-33-44-55 stream_target='{"ip": [10, 34, 16, 123], "port": 4000}'
Where `10.34.16.10` is the MQTT broker address that matches the one configured in the source code and `10.34.16.123` and `4000` are the desire stream target IP and port.
```
@ -61,7 +68,7 @@ The rules for constructing `path` values are documented in [`miniconf`'s
documentation](https://github.com/quartiq/miniconf#settings-paths)
Refer to the documentation for [Miniconf]({{site.baseurl}}/firmware/miniconf/enum.Error.html) for a
description of the possible error codes that `miniconf.py` may return if the settings update was
description of the possible error codes that Miniconf may return if the settings update was
unsuccessful.
## Telemetry

163
hitl/loopback.py Normal file
View File

@ -0,0 +1,163 @@
#!/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 allowable 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, queue):
"""Create a connection to the broker and an MQTT device using it."""
client = MqttClient(client_id='')
await client.connect(broker)
return cls(client, prefix, queue)
def __init__(self, client, prefix, queue):
""" 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)
self.queue = queue
def handle_telemetry(self, _client, topic, payload, _qos, _properties):
""" Handle incoming telemetry messages over MQTT. """
assert topic == self._telemetry_topic
self.queue.put_nowait(json.loads(payload))
async def test_loopback(miniconf, telemetry_queue, set_point, gain=1, channel=0):
""" 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.
channel: The loopback channel to test on. Either 0 or 1.
gain: The desired AFE gain.
"""
print(f'Testing loopback for Vout = {set_point:.2f}, Gain = x{gain}')
print('---------------------------------')
# Configure the AFE and IIRs to output at the set point
await miniconf.command(f'afe/{channel}', f'G{gain}', retain=False)
await miniconf.command(f'iir_ch/{channel}/0', static_iir_output(set_point), retain=False)
# Configure signal generators to not affect the test.
await miniconf.command('signal_generator/0/amplitude', 0, retain=False)
# Wait for telemetry to update.
await asyncio.sleep(5.0)
# Verify the ADCs are receiving the setpoint voltage.
tolerance = max(0.05 * set_point, MINIMUM_VOLTAGE_ERROR)
latest_values = await telemetry_queue.get()
print(f'Latest telemtry: {latest_values}')
assert abs(latest_values['adcs'][channel] - 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()
telemetry_queue = asyncio.LifoQueue()
async def telemetry():
await TelemetryReader.create(args.prefix, args.broker, telemetry_queue)
try:
while True:
await asyncio.sleep(1)
except asyncio.CancelledError:
pass
telemetry_task = asyncio.Task(telemetry())
async def test():
""" The actual testing being completed. """
interface = await Miniconf.create(args.prefix, args.broker)
# Disable IIR holds and configure the telemetry rate.
await interface.command('allow_hold', False, retain=False)
await interface.command('force_hold', False, retain=False)
await interface.command('telemetry_period', 1, retain=False)
# Test loopback with a static 1V output of the DACs.
await test_loopback(interface, telemetry_queue, 1.0)
# Repeat test with AFE = 2x
await test_loopback(interface, telemetry_queue, 1.0, gain=2)
# Test with 0V output
await test_loopback(interface, telemetry_queue, 0.0)
telemetry_task.cancel()
loop = asyncio.get_event_loop()
sys.exit(loop.run_until_complete(test()))
if __name__ == '__main__':
main()

12
hitl/run.sh
View File

@ -13,7 +13,10 @@ set -eux
# Set up python for testing
python3 -m venv --system-site-packages py
. py/bin/activate
python3 -m pip install -r requirements.txt
# Install Miniconf utilities for configuring stabilizer.
python3 -m pip install git+https://github.com/quartiq/miniconf#subdirectory=py/miniconf-mqtt
python3 -m pip install gmqtt
cargo flash --chip STM32H743ZITx --elf target/thumbv7em-none-eabihf/release/dual-iir
@ -27,6 +30,9 @@ sleep 30
ping -c 5 -w 20 stabilizer-hitl
# Test the MQTT interface.
python3 miniconf.py dt/sinara/dual-iir/04-91-62-d9-7e-5f afe/0='"G2"'
python3 miniconf.py dt/sinara/dual-iir/04-91-62-d9-7e-5f afe/0='"G1"' iir_ch/0/0=\
python3 -m miniconf dt/sinara/dual-iir/04-91-62-d9-7e-5f afe/0='"G2"'
python3 -m miniconf dt/sinara/dual-iir/04-91-62-d9-7e-5f afe/0='"G1"' iir_ch/0/0=\
'{"y_min": -32767, "y_max": 32767, "y_offset": 0, "ba": [1.0, 0, 0, 0, 0]}'
# Test the ADC/DACs connected via loopback.
python3 hitl/loopback.py dt/sinara/dual-iir/04-91-62-d9-7e-5f

143
miniconf.py
View File

@ -1,143 +0,0 @@
#!/usr/bin/python
"""
Author: Vertigo Designs, Ryan Summers
Robert Jördens
Description: Provides an API for controlling Miniconf devices over MQTT.
"""
import argparse
import asyncio
import json
import logging
import sys
import uuid
from gmqtt import Client as MqttClient
LOGGER = logging.getLogger(__name__)
class Miniconf:
"""An asynchronous API for controlling Miniconf devices using MQTT."""
@classmethod
async def create(cls, prefix, broker):
"""Create a connection to the broker and a Miniconf device using it."""
client = MqttClient(client_id='')
await client.connect(broker)
return cls(client, prefix)
def __init__(self, client, prefix):
"""Constructor.
Args:
client: A connected MQTT5 client.
prefix: The MQTT toptic prefix of the device to control.
"""
self.request_id = 0
self.client = client
self.prefix = prefix
self.inflight = {}
self.client.on_message = self._handle_response
self.response_topic = f'{prefix}/response/{uuid.uuid1().hex}'
self.client.subscribe(self.response_topic)
def _handle_response(self, _client, topic, payload, _qos, properties):
"""Callback function for when messages are received over MQTT.
Args:
_client: The MQTT client.
topic: The topic that the message was received on.
payload: The payload of the message.
_qos: The quality-of-service level of the received packet
properties: A dictionary of properties associated with the message.
"""
if topic == self.response_topic:
# Extract request_id corrleation data from the properties
request_id = int.from_bytes(
properties['correlation_data'][0], 'big')
self.inflight[request_id].set_result(json.loads(payload))
del self.inflight[request_id]
else:
LOGGER.warn('Unexpected message on "%s"', topic)
async def command(self, path, value, retain=True):
"""Write the provided data to the specified path.
Args:
path: The path to write the message to.
value: The value to write to the path.
retain: Retain the MQTT message changing the setting
by the broker.
Returns:
The response to the command as a dictionary.
"""
topic = f'{self.prefix}/settings/{path}'
fut = asyncio.get_running_loop().create_future()
# Assign unique correlation data for response dispatch
assert self.request_id not in self.inflight
self.inflight[self.request_id] = fut
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
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()

1
requirements.txt
View File

@ -1 +0,0 @@
gmqtt