Merge #426

426: Feature/streaming hitl r=ryan-summers a=ryan-summers

This PR fixes #386 by implementing HITL tests for the data livestream functionalities of Stabilizer.

- [x] This should be merged after https://github.com/quartiq/stabilizer/pull/425

Co-authored-by: Ryan Summers <ryan.summers@vertigo-designs.com>

hitl/run.sh

@@ -11,10 +11,11 @@
 set -eux
 
 # Set up python for testing
-python3 -m venv --system-site-packages py
+python3 -m venv --system-site-packages py-venv
-. py/bin/activate
+. py-venv/bin/activate
 
 # Install Miniconf utilities for configuring stabilizer.
+python3 -m pip install -e py/
 python3 -m pip install git+https://github.com/quartiq/miniconf#subdirectory=py/miniconf-mqtt
 python3 -m pip install gmqtt
 
@@ -36,3 +37,6 @@ python3 -m miniconf dt/sinara/dual-iir/04-91-62-d9-7e-5f afe/0='"G1"' iir_ch/0/0
 
 # Test the ADC/DACs connected via loopback.
 python3 hitl/loopback.py dt/sinara/dual-iir/04-91-62-d9-7e-5f
+
+# Test the livestream capabilities
+python3 hitl/streaming.py dt/sinara/dual-iir/04-91-62-d9-7e-5f

hitl/streaming.py

@@ -0,0 +1,128 @@
+#!/usr/bin/python3
+"""
+Author: Vertigo Designs, Ryan Summers
+
+Description: Implements HITL testing of Stabilizer data livestream capabilities.
+"""
+import asyncio
+import sys
+import argparse
+import logging
+import socket
+import time
+
+from miniconf import Miniconf
+from stabilizer.stream import StabilizerStream
+
+# The duration to receive frames for.
+STREAM_TEST_DURATION_SECS = 5.0
+
+# The minimum efficiency of the stream in frame transfer to pass testing. Represented as
+# (received_frames / transmitted_frames).
+MIN_STREAM_EFFICIENCY = 0.95
+
+def _get_ip(broker):
+    """ Get the IP of the local device.
+
+    Args:
+        broker: The broker IP of the test. Used to select an interface to get the IP of.
+
+    Returns:
+        The IP as an array of integers.
+    """
+    sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
+    try:
+        sock.connect((broker, 1883))
+        address = sock.getsockname()[0]
+    finally:
+        sock.close()
+
+    return list(map(int, address.split('.')))
+
+
+def sequence_delta(previous_sequence, next_sequence):
+    """ Check the number of items between two sequence numbers. """
+    if previous_sequence is None:
+        return 0
+
+    delta = next_sequence - (previous_sequence + 1)
+    return delta & 0xFFFFFFFF
+
+
+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')
+    parser.add_argument('--port', '-p', default=2000, type=int,
+                        help='The UDP port to use for streaming')
+
+    args = parser.parse_args()
+
+    async def test():
+        """ The actual testing being completed. """
+        local_ip = _get_ip(args.broker)
+        interface = await Miniconf.create(args.prefix, args.broker)
+        stream = StabilizerStream(args.port, timeout=0.5)
+
+        # Configure the stream
+        print(f'Configuring stream to target {".".join(map(str, local_ip))}:{args.port}')
+        print('')
+        await interface.command('stream_target', {'ip': local_ip, 'port': args.port}, retain=False)
+        await interface.command('telemetry_period', 10, retain=False)
+
+        # Verify frame reception
+        print('Testing stream reception')
+        print('')
+        last_sequence = None
+
+        # Sample frames over a set time period and verify that no drops are encountered.
+        stop = time.time() + STREAM_TEST_DURATION_SECS
+        dropped_batches = 0
+        total_batches = 0
+
+        while time.time() < stop:
+            for (seqnum, _data) in stream.read_frame():
+                num_dropped = sequence_delta(last_sequence, seqnum)
+                total_batches += 1 + num_dropped
+
+                if num_dropped:
+                    dropped_batches += num_dropped
+                    logging.warning('Frame drop detected: 0x%08X -> 0x%08X (%d batches)',
+                                    last_sequence, seqnum, num_dropped)
+
+                last_sequence = seqnum
+
+        assert total_batches, 'Stream did not receive any frames'
+        stream_efficiency = 1.0 - (dropped_batches / total_batches)
+
+        print(f'Stream Reception Rate: {stream_efficiency * 100:.2f} %')
+        print(f'Received {total_batches} ')
+        print(f'Lost {dropped_batches} batches')
+
+        assert stream_efficiency > MIN_STREAM_EFFICIENCY, \
+            f'Stream dropped too many packets.  Reception rate: {stream_efficiency * 100:.2f} %'
+
+        # Disable the stream.
+        print('Closing stream')
+        print('')
+        await interface.command('stream_target', {'ip': [0, 0, 0, 0], 'port': 0}, retain=False)
+        stream.clear()
+
+        print('Verifying no further data is received')
+        try:
+            for _ in stream.read_frame():
+                raise Exception('Unexpected data encountered on stream')
+        except socket.timeout:
+            pass
+        print('PASS')
+
+
+    loop = asyncio.get_event_loop()
+    sys.exit(loop.run_until_complete(test()))
+
+
+if __name__ == '__main__':
+    main()

py/.gitignore

@@ -0,0 +1,2 @@
+/*.egg-info/
+__pycache__/

py/README.md

@@ -0,0 +1,8 @@
+# Stabilizer Python Utilities
+
+This directory contains common Python utilities for Stabilizer, such as livestream data receivers.
+
+To install this module locally (in editable mode):
+```
+python -m pip install -e .
+```

py/setup.py

@@ -0,0 +1,7 @@
+from setuptools import setup
+
+setup(name='stabilizer',
+      version='0.1',
+      description='Stabilizer Utilities',
+      author='QUARTIQ GmbH',
+      license='MIT')

py/stabilizer/stream.py

@@ -0,0 +1,92 @@
+#!/usr/bin/python3
+"""
+Author: Vertigo Designs, Ryan Summers
+
+Description: Provides a means of accessing Stabilizer livestream data.
+"""
+import socket
+import time
+import logging
+import struct
+
+class StabilizerStream:
+    """ Provides access to Stabilizer's livestreamed data. """
+
+    # The magic header half-word at the start of each packet.
+    MAGIC_HEADER = 0x057B
+
+    # The struct format of the header.
+    HEADER_FORMAT = '<HBBI'
+
+    # All supported formats by this reception script.
+    #
+    # The items in this dict are functions that will be provided the sample batch size and will
+    # return the struct deserialization code to unpack a single batch.
+    FORMAT = {
+        1: lambda batch_size: f'<{batch_size}H{batch_size}H{batch_size}H{batch_size}H'
+    }
+
+    def __init__(self, port, timeout=None):
+        """ Initialize the stream.
+
+        Args:
+            port: The UDP port to receive the stream from.
+            timeout: The timeout to set on the UDP socket.
+        """
+        self.socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
+        self.socket.bind(("", port))
+        self.total_bytes = 0
+
+        if timeout is not None:
+            self.socket.settimeout(timeout)
+
+
+    def clear(self, duration=1):
+        """ Clear the socket RX buffer by reading all available data.
+
+        Args:
+            duration: The maximum duration in seconds to read data for.
+        """
+        time.sleep(duration)
+
+        try:
+            while self.socket.recv(4096):
+                pass
+        except socket.timeout:
+            pass
+
+
+    def get_rx_bytes(self):
+        """ Get the number of bytes read from the stream. """
+        return self.total_bytes
+
+
+    def read_frame(self):
+        """ Read a single frame from the stream.
+
+        Returns:
+            Yields the (seqnum, data) of the batches available in the frame.
+        """
+        buf = self.socket.recv(4096)
+        self.total_bytes += len(buf)
+
+        # Attempt to parse a block from the buffer.
+        if len(buf) < struct.calcsize(self.HEADER_FORMAT):
+            return
+
+        # Parse out the packet header
+        magic, format_id, batch_size, sequence_number = struct.unpack_from(self.HEADER_FORMAT, buf)
+        buf = buf[struct.calcsize(self.HEADER_FORMAT):]
+
+        if magic != self.MAGIC_HEADER:
+            logging.warning('Encountered bad magic header: %s', hex(magic))
+            return
+
+        frame_format = self.FORMAT[format_id](batch_size)
+
+        batch_count = int(len(buf) / struct.calcsize(frame_format))
+
+        for offset in range(batch_count):
+            data = struct.unpack_from(frame_format, buf)
+            buf = buf[struct.calcsize(frame_format):]
+            yield (sequence_number + offset, data)

scripts/stream_throughput.py

@@ -5,53 +5,11 @@ Author: Ryan Summers
 Description: Provides a mechanism for measuring Stabilizer stream data throughput.
 """
 import argparse
-import socket
-import collections
-import struct
-import time
 import logging
+import sys
+import time
 
-# Representation of a single data batch transmitted by Stabilizer.
+from stabilizer.stream import StabilizerStream
-Packet = collections.namedtuple('Packet', ['index', 'data'])
-
-# The magic header half-word at the start of each packet.
-MAGIC_HEADER = 0x057B
-
-# The struct format of the header.
-HEADER_FORMAT = '<HBBI'
-
-# All supported formats by this reception script.
-#
-# The items in this dict are functions that will be provided the sample batch size and will return
-# the struct deserialization code to unpack a single batch.
-FORMAT = {
-    1: lambda batch_size: f'<{batch_size}H{batch_size}H{batch_size}H{batch_size}H'
-}
-
-def parse_packet(buf):
-    """ Attempt to parse packets from the received buffer. """
-    # Attempt to parse a block from the buffer.
-    if len(buf) < struct.calcsize(HEADER_FORMAT):
-        return
-
-    # Parse out the packet header
-    magic, format_id, batch_size, sequence_number = struct.unpack_from(HEADER_FORMAT, buf)
-    buf = buf[struct.calcsize(HEADER_FORMAT):]
-
-    if magic != MAGIC_HEADER:
-        logging.warning('Encountered bad magic header: %s', hex(magic))
-        return
-
-    frame_format = FORMAT[format_id](batch_size)
-
-    batch_count = int(len(buf) / struct.calcsize(frame_format))
-
-    for offset in range(batch_count):
-        data = struct.unpack_from(frame_format, buf)
-        buf = buf[struct.calcsize(frame_format):]
-        yield Packet(sequence_number + offset, data)
-
-
 
 class Timer:
     """ A basic timer for measuring elapsed time periods. """
@@ -104,13 +62,11 @@ def sequence_delta(previous_sequence, next_sequence):
 def main():
     """ Main program. """
     parser = argparse.ArgumentParser(description='Measure Stabilizer livestream quality')
-    parser.add_argument('--port', type=int, default=1111, help='The port that stabilizer is streaming to')
+    parser.add_argument('--port', type=int, default=2000,
+                        help='The port that stabilizer is streaming to')
 
     args = parser.parse_args()
 
-    connection = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
-    connection.bind(("", args.port))
-
     logging.basicConfig(level=logging.INFO,
                         format='%(asctime)s.%(msecs)03d %(levelname)-8s %(message)s')
 
@@ -122,23 +78,22 @@ def main():
 
     timer = Timer()
 
+    stream = StabilizerStream(args.port)
+
     while True:
         # Receive any data over UDP and parse it.
-        data = connection.recv(4096)
+        for (seqnum, _) in stream.read_frame():
-        if data and not timer.is_started():
+            if not timer.is_started():
-            timer.start()
+                timer.start()
 
-        # Handle any received packets.
-        total_bytes += len(data)
-        for packet in parse_packet(data):
             # Handle any dropped packets.
-            drop_count += sequence_delta(last_index, packet.index)
+            drop_count += sequence_delta(last_index, seqnum)
-            last_index = packet.index
+            last_index = seqnum
             good_blocks += 1
 
         # Report the throughput periodically.
         if timer.is_triggered():
-            drate = total_bytes * 8 / 1e6 / timer.elapsed()
+            drate = stream.get_rx_bytes() * 8 / 1e6 / timer.elapsed()
 
             print(f'''
 Data Rate:       {drate:.3f} Mbps
@@ -148,6 +103,7 @@ Dropped blocks:  {drop_count}
 Metadata: {total_bytes / 1e6:.3f} MB in {timer.elapsed():.2f} s
 ----
 ''')
+            sys.stdout.flush()
             timer.arm()