From 824b6763e16b48d60da291323c736fa558883436 Mon Sep 17 00:00:00 2001
From: Ryan Summers <ryan.summers@vertigo-designs.com>
Date: Thu, 5 Aug 2021 16:21:55 +0200
Subject: [PATCH 1/6] Adding livestream HITL test

---
 hitl/run.sh       |   3 +
 hitl/streaming.py | 167 ++++++++++++++++++++++++++++++++++++++++++++++
 2 files changed, 170 insertions(+)
 create mode 100644 hitl/streaming.py

diff --git a/hitl/run.sh b/hitl/run.sh
index 582c0f0..45acb03 100755
--- a/hitl/run.sh
+++ b/hitl/run.sh
@@ -33,3 +33,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
diff --git a/hitl/streaming.py b/hitl/streaming.py
new file mode 100644
index 0000000..1675409
--- /dev/null
+++ b/hitl/streaming.py
@@ -0,0 +1,167 @@
+#!/usr/bin/python3
+"""
+Author: Vertigo Designs, Ryan Summers
+
+Description: Implements HITL testing of Stabilizer data livestream capabilities.
+"""
+import asyncio
+import sys
+import time
+import argparse
+import socket
+import struct
+import logging
+
+from miniconf import Miniconf
+
+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
+
+
+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):
+        """ Initialize the stream. """
+        self.socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
+        self.socket.bind(("", port))
+        self.socket.settimeout(0.5)
+
+
+    def clear(self, duration=5):
+        """ Clear the socket RX buffer by reading all available data.
+
+        Args:
+            duration: The maximum duration in seconds to read data for.
+        """
+        start = time.time()
+        while (time.time() - start) < duration:
+            try:
+                self.socket.recv(4096)
+            except socket.timeout:
+                return
+
+
+    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)
+
+        # 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)
+
+
+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)
+
+        # Configure the stream
+        print(f'Configuring stream to target {".".join(map(str, local_ip))}:{args.port}')
+        print('')
+        assert (await interface.command('stream_target',
+                                        {'ip': local_ip, 'port': args.port}))['code'] == 0
+
+        # Verify frame reception
+        print('Testing stream reception')
+        print('')
+        last_sequence = None
+        for _ in range(5000):
+            for (seqnum, _data) in stream.read_frame():
+                assert sequence_delta(last_sequence, seqnum) == 0
+                last_sequence = seqnum
+
+        # Disable the stream.
+        print('Closing stream')
+        print('')
+        assert (await interface.command('stream_target',
+                                        {'ip': [0, 0, 0, 0], 'port': 0}))['code'] == 0
+        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()

From e478dfc70fd3a6bab08c67c2a8663acb79f42cbb Mon Sep 17 00:00:00 2001
From: Ryan Summers <ryan.summers@vertigo-designs.com>
Date: Mon, 9 Aug 2021 14:18:09 +0200
Subject: [PATCH 2/6] Fixing miniconf API after updates

---
 hitl/streaming.py | 6 ++----
 1 file changed, 2 insertions(+), 4 deletions(-)

diff --git a/hitl/streaming.py b/hitl/streaming.py
index 1675409..6d884b6 100644
--- a/hitl/streaming.py
+++ b/hitl/streaming.py
@@ -131,8 +131,7 @@ def main():
         # Configure the stream
         print(f'Configuring stream to target {".".join(map(str, local_ip))}:{args.port}')
         print('')
-        assert (await interface.command('stream_target',
-                                        {'ip': local_ip, 'port': args.port}))['code'] == 0
+        await interface.command('stream_target', {'ip': local_ip, 'port': args.port})
 
         # Verify frame reception
         print('Testing stream reception')
@@ -146,8 +145,7 @@ def main():
         # Disable the stream.
         print('Closing stream')
         print('')
-        assert (await interface.command('stream_target',
-                                        {'ip': [0, 0, 0, 0], 'port': 0}))['code'] == 0
+        await interface.command('stream_target', {'ip': [0, 0, 0, 0], 'port': 0})
         stream.clear()
 
         print('Verifying no further data is received')

From 5d7cac22939ac55a690cc7dd90819e066f3db0e6 Mon Sep 17 00:00:00 2001
From: Ryan Summers <ryan.summers@vertigo-designs.com>
Date: Mon, 9 Aug 2021 16:23:18 +0200
Subject: [PATCH 3/6] Updating clear method

---
 hitl/run.sh                  |  3 +-
 hitl/streaming.py            | 82 +++-----------------------------
 py/.gitignore                |  2 +
 py/README.md                 |  8 ++++
 py/setup.py                  |  7 +++
 py/stabilizer/__init__.py    |  0
 py/stabilizer/stream.py      | 92 ++++++++++++++++++++++++++++++++++++
 scripts/stream_throughput.py | 72 ++++++----------------------
 8 files changed, 132 insertions(+), 134 deletions(-)
 create mode 100644 py/.gitignore
 create mode 100644 py/README.md
 create mode 100644 py/setup.py
 create mode 100644 py/stabilizer/__init__.py
 create mode 100644 py/stabilizer/stream.py

diff --git a/hitl/run.sh b/hitl/run.sh
index 7c7aa27..c4f8821 100755
--- a/hitl/run.sh
+++ b/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
 
 # 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
 
diff --git a/hitl/streaming.py b/hitl/streaming.py
index 6d884b6..2e6fa3b 100644
--- a/hitl/streaming.py
+++ b/hitl/streaming.py
@@ -6,13 +6,11 @@ Description: Implements HITL testing of Stabilizer data livestream capabilities.
 """
 import asyncio
 import sys
-import time
 import argparse
 import socket
-import struct
-import logging
 
 from miniconf import Miniconf
+from stabilizer.stream import StabilizerStream
 
 def _get_ip(broker):
     """ Get the IP of the local device.
@@ -42,74 +40,6 @@ def sequence_delta(previous_sequence, next_sequence):
     return delta & 0xFFFFFFFF
 
 
-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):
-        """ Initialize the stream. """
-        self.socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
-        self.socket.bind(("", port))
-        self.socket.settimeout(0.5)
-
-
-    def clear(self, duration=5):
-        """ Clear the socket RX buffer by reading all available data.
-
-        Args:
-            duration: The maximum duration in seconds to read data for.
-        """
-        start = time.time()
-        while (time.time() - start) < duration:
-            try:
-                self.socket.recv(4096)
-            except socket.timeout:
-                return
-
-
-    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)
-
-        # 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)
-
-
 def main():
     """ Main program entry point. """
     parser = argparse.ArgumentParser(description='Loopback tests for Stabilizer HITL testing',)
@@ -126,12 +56,13 @@ def main():
         """ The actual testing being completed. """
         local_ip = _get_ip(args.broker)
         interface = await Miniconf.create(args.prefix, args.broker)
-        stream = StabilizerStream(args.port)
+        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})
+        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')
@@ -139,13 +70,14 @@ def main():
         last_sequence = None
         for _ in range(5000):
             for (seqnum, _data) in stream.read_frame():
-                assert sequence_delta(last_sequence, seqnum) == 0
+                assert sequence_delta(last_sequence, seqnum) == 0, \
+                        f'Frame drop detected: 0x{last_sequence:08X} -> 0x{seqnum:08X}'
                 last_sequence = seqnum
 
         # Disable the stream.
         print('Closing stream')
         print('')
-        await interface.command('stream_target', {'ip': [0, 0, 0, 0], 'port': 0})
+        await interface.command('stream_target', {'ip': [0, 0, 0, 0], 'port': 0}, retain=False)
         stream.clear()
 
         print('Verifying no further data is received')
diff --git a/py/.gitignore b/py/.gitignore
new file mode 100644
index 0000000..d9dc407
--- /dev/null
+++ b/py/.gitignore
@@ -0,0 +1,2 @@
+/*.egg-info/
+__pycache__/
diff --git a/py/README.md b/py/README.md
new file mode 100644
index 0000000..b768ea1
--- /dev/null
+++ b/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 .
+```
diff --git a/py/setup.py b/py/setup.py
new file mode 100644
index 0000000..5b48bd3
--- /dev/null
+++ b/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')
diff --git a/py/stabilizer/__init__.py b/py/stabilizer/__init__.py
new file mode 100644
index 0000000..e69de29
diff --git a/py/stabilizer/stream.py b/py/stabilizer/stream.py
new file mode 100644
index 0000000..e6144db
--- /dev/null
+++ b/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)
diff --git a/scripts/stream_throughput.py b/scripts/stream_throughput.py
index f398dae..1abe500 100644
--- a/scripts/stream_throughput.py
+++ b/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.
-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)
-
-
+from stabilizer.stream import StabilizerStream
 
 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)
-        if data and not timer.is_started():
-            timer.start()
+        for (seqnum, _) in stream.read_frame():
+            if not timer.is_started():
+                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)
-            last_index = packet.index
+            drop_count += sequence_delta(last_index, seqnum)
+            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()
 
 

From 199ee380589587606ab78133cb17d6969be6f401 Mon Sep 17 00:00:00 2001
From: Ryan Summers <ryan.summers@vertigo-designs.com>
Date: Mon, 9 Aug 2021 17:00:42 +0200
Subject: [PATCH 4/6] Update hitl/run.sh
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Co-authored-by: Robert Jördens <rj@quartiq.de>
---
 hitl/run.sh | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/hitl/run.sh b/hitl/run.sh
index c4f8821..515429d 100755
--- a/hitl/run.sh
+++ b/hitl/run.sh
@@ -12,7 +12,7 @@ set -eux
 
 # Set up python for testing
 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/

From b8910f9b348418910df8fbf92b6bc3d20e6cadb9 Mon Sep 17 00:00:00 2001
From: Ryan Summers <ryan.summers@vertigo-designs.com>
Date: Tue, 10 Aug 2021 13:24:06 +0200
Subject: [PATCH 5/6] Updating stream test to allow for dropped frames

---
 hitl/streaming.py | 37 ++++++++++++++++++++++++++++++++++---
 1 file changed, 34 insertions(+), 3 deletions(-)

diff --git a/hitl/streaming.py b/hitl/streaming.py
index 2e6fa3b..d56bc57 100644
--- a/hitl/streaming.py
+++ b/hitl/streaming.py
@@ -7,11 +7,20 @@ 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.
 
@@ -68,12 +77,34 @@ def main():
         print('Testing stream reception')
         print('')
         last_sequence = None
-        for _ in range(5000):
+
+        # Sample frames over a set time period and verify that no drops are encountered.
+        stop = time.time() + STREAM_TEST_DURATION_SECS
+        dropped_frames = 0
+        total_frames = 0
+
+        while time.time() < stop:
             for (seqnum, _data) in stream.read_frame():
-                assert sequence_delta(last_sequence, seqnum) == 0, \
-                        f'Frame drop detected: 0x{last_sequence:08X} -> 0x{seqnum:08X}'
+                num_dropped = sequence_delta(last_sequence, seqnum)
+                total_frames += 1 + num_dropped
+
+                if num_dropped:
+                    dropped_frames += num_dropped
+                    logging.warning('Frame drop detected: 0x%08X -> 0x%08X (%d frames)',
+                                    last_sequence, seqnum, num_dropped)
+
                 last_sequence = seqnum
 
+        assert total_frames, 'Stream did not receive any frames'
+        stream_efficiency = 1.0 - (dropped_frames / total_frames)
+
+        print(f'Stream Reception Rate: {stream_efficiency * 100:.2f} %')
+        print(f'Received {total_frames} frames')
+        print(f'Lost {dropped_frames} frames')
+
+        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('')

From 19490209dd7e0a33fb12417db31f4fe3e60b954c Mon Sep 17 00:00:00 2001
From: Ryan Summers <ryan.summers@vertigo-designs.com>
Date: Tue, 10 Aug 2021 14:13:02 +0200
Subject: [PATCH 6/6] Fixing verbage

---
 hitl/streaming.py | 18 +++++++++---------
 1 file changed, 9 insertions(+), 9 deletions(-)

diff --git a/hitl/streaming.py b/hitl/streaming.py
index d56bc57..e9bfe66 100644
--- a/hitl/streaming.py
+++ b/hitl/streaming.py
@@ -80,27 +80,27 @@ def main():
 
         # Sample frames over a set time period and verify that no drops are encountered.
         stop = time.time() + STREAM_TEST_DURATION_SECS
-        dropped_frames = 0
-        total_frames = 0
+        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_frames += 1 + num_dropped
+                total_batches += 1 + num_dropped
 
                 if num_dropped:
-                    dropped_frames += num_dropped
-                    logging.warning('Frame drop detected: 0x%08X -> 0x%08X (%d frames)',
+                    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_frames, 'Stream did not receive any frames'
-        stream_efficiency = 1.0 - (dropped_frames / total_frames)
+        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_frames} frames')
-        print(f'Lost {dropped_frames} frames')
+        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} %'