Merge pull request #965 from quartiq/sweptsine

Sweptsine

Cargo.toml

@@ -50,7 +50,7 @@ rtic = { version = "2.1", features = ["thumbv7-backend"] }
 rtic-monotonics = { version = "2.0", features = ["cortex-m-systick"] }
 num_enum = { version = "0.7.3", default-features = false }
 paste = "1"
-idsp = "0.15.1"
+idsp = "0.16.0"
 ad9959 = { path = "ad9959", version = "0.3.0" }
 serial-settings = { version = "0.2", path = "serial-settings" }
 mcp230xx = "1.0"

hitl/streaming.py

@@ -61,7 +61,7 @@ async def _main():
         try:
             logger.info("Testing stream reception")
             _transport, stream = await StabilizerStream.open(
-                args.ip, args.port, args.broker
+                args.port, args.ip, args.broker
             )
             loss = await measure(stream, args.duration)
             if loss > args.max_loss:

py/stabilizer/__init__.py

@@ -2,13 +2,13 @@
 """Stabilizer data conversion and streaming utilities"""
 
 # Sample period in seconds, default 100 MHz timer clock and a reload value of 128
-SAMPLE_PERIOD = 10e-9*128
+SAMPLE_PERIOD = 10e-9 * 128
 
 # The number of DAC LSB codes per volt on Stabilizer outputs.
 DAC_LSB_PER_VOLT = (1 << 16) / (4.096 * 5)
 
 # The number of volts per ADC LSB.
-ADC_VOLTS_PER_LSB = (5.0 / 2.0 * 4.096)  / (1 << 15)
+ADC_VOLTS_PER_LSB = (5.0 / 2.0 * 4.096) / (1 << 15)
 
 # The number of volts per DAC LSB.
 DAC_VOLTS_PER_LSB = 1 / DAC_LSB_PER_VOLT
@@ -17,6 +17,7 @@
 # DAC.
 DAC_FULL_SCALE = float(0x7FFF / DAC_LSB_PER_VOLT)
 
+
 def voltage_to_machine_units(voltage):
     """Convert a voltage to machine units."""
     code = int(round(voltage * DAC_LSB_PER_VOLT))

py/stabilizer/plot_iir_frequency_response.py

@@ -11,8 +11,7 @@
 from stabilizer.iir_coefficients import get_filters
 
 # disable warnings about short variable names and similar code
-#pylint: disable=invalid-name, duplicate-code, redefined-builtin
-
+# pylint: disable=invalid-name, duplicate-code, redefined-builtin
 
 
 def _main():
@@ -52,7 +51,7 @@ def _main():
     if forward_gain == 0 and args.x_offset != 0:
         print("Filter has no DC gain but x_offset is non-zero")
 
-    f = np.logspace(-8.5, 0, 1024, endpoint=False)*(.5/args.sample_period)
+    f = np.logspace(-8.5, 0, 1024, endpoint=False) * (0.5 / args.sample_period)
     f, h = signal.freqz(
         coefficients[:3],
         np.r_[1, [-c for c in coefficients[3:]]],

py/stabilizer/stream.py

@@ -1,4 +1,6 @@
 #!/usr/bin/python3
+# pylint: disable=too-few-public-methods
+
 """Stabilizer streaming receiver and parsers"""
 
 import argparse
@@ -21,7 +23,7 @@
 
 def wrap(wide):
     """Wrap to 32 bit integer"""
-    return wide & 0xffffffff
+    return wide & 0xFFFFFFFF
 
 
 def get_local_ip(remote):
@@ -37,6 +39,7 @@ def get_local_ip(remote):
 
 class AdcDac:
     """Stabilizer default striming data format"""
+
     format_id = 1
 
     def __init__(self, header, body):
@@ -69,15 +72,16 @@ def to_traces(self):
         """Convert the raw data to labelled Trace instances"""
         data = self.to_mu()
         return [
-            Trace(data[0], scale=DAC_VOLTS_PER_LSB, label='ADC0'),
-            Trace(data[1], scale=DAC_VOLTS_PER_LSB, label='ADC1'),
-            Trace(data[2], scale=DAC_VOLTS_PER_LSB, label='DAC0'),
-            Trace(data[3], scale=DAC_VOLTS_PER_LSB, label='DAC1')
+            Trace(data[0], scale=DAC_VOLTS_PER_LSB, label="ADC0"),
+            Trace(data[1], scale=DAC_VOLTS_PER_LSB, label="ADC1"),
+            Trace(data[2], scale=DAC_VOLTS_PER_LSB, label="DAC0"),
+            Trace(data[3], scale=DAC_VOLTS_PER_LSB, label="DAC1"),
         ]
 
 
-class StabilizerStream(asyncio.DatagramProtocol):
-    """Stabilizer streaming receiver protocol"""
+class Frame:
+    """Stream frame constisting of a header and multiple data batches"""
+
     # The magic header half-word at the start of each packet.
     magic = 0x057B
     header_fmt = struct.Struct("<HBBI")
@@ -87,7 +91,23 @@ class StabilizerStream(asyncio.DatagramProtocol):
     }
 
     @classmethod
-    async def open(cls, addr, port, broker, maxsize=1):
+    def parse(cls, data):
+        """Parse known length frame"""
+        header = cls.header._make(cls.header_fmt.unpack_from(data))
+        if header.magic != cls.magic:
+            raise ValueError(f"Bad frame magic: {header.magic:#04x}")
+        try:
+            parser = cls.parsers[header.format_id]
+        except KeyError as exc:
+            raise ValueError(f"No parser for format: {header.format_id}") from exc
+        return parser(header, data[cls.header_fmt.size :])
+
+
+class StabilizerStream(asyncio.DatagramProtocol):
+    """Stabilizer streaming receiver protocol"""
+
+    @classmethod
+    async def open(cls, port=9293, addr="0.0.0.0", broker=None, maxsize=1):
         """Open a UDP socket and start receiving frames"""
         loop = asyncio.get_running_loop()
         sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM, socket.IPPROTO_UDP)
@@ -104,15 +124,16 @@ async def open(cls, addr, port, broker, maxsize=1):
         # wrong one. Thus, use the broker address to figure out our local address for the interface
         # of interest.
         if ipaddress.ip_address(addr).is_multicast:
-            print('Subscribing to multicast')
             group = socket.inet_aton(addr)
-            iface = socket.inet_aton('.'.join([str(x) for x in get_local_ip(broker)]))
+            iface = socket.inet_aton(get_local_ip(broker))
             sock.setsockopt(socket.IPPROTO_IP, socket.IP_ADD_MEMBERSHIP, group + iface)
-            sock.bind(('', port))
+            sock.bind(("", port))
         else:
             sock.bind((addr, port))
 
-        transport, protocol = await loop.create_datagram_endpoint(lambda: cls(maxsize), sock=sock)
+        transport, protocol = await loop.create_datagram_endpoint(
+            lambda: cls(maxsize), sock=sock
+        )
         return transport, protocol
 
     def __init__(self, maxsize):
@@ -125,16 +146,11 @@ def connection_lost(self, _exc):
         logger.info("Connection lost")
 
     def datagram_received(self, data, _addr):
-        header = self.header._make(self.header_fmt.unpack_from(data))
-        if header.magic != self.magic:
-            logger.warning("Bad frame magic: %#04x, ignoring", header.magic)
-            return
         try:
-            parser = self.parsers[header.format_id]
-        except KeyError:
-            logger.warning("No parser for format %s, ignoring", header.format_id)
+            frame = Frame.parse(data)
+        except ValueError as e:
+            logger.warning("Parse error: %s", e)
             return
-        frame = parser(header, data[self.header_fmt.size:])
         if self.queue.full():
             old = self.queue.get_nowait()
             logger.debug("Dropping frame: %#08x", old.header.sequence)
@@ -143,12 +159,14 @@ def datagram_received(self, data, _addr):
 
 async def measure(stream, duration):
     """Measure throughput and loss of stream reception"""
+
     @dataclass
     class _Statistics:
         expect = None
         received = 0
         lost = 0
         bytes = 0
+
     stat = _Statistics()
 
     async def _record():
@@ -167,36 +185,35 @@ async def _record():
     except asyncio.TimeoutError:
         pass
 
-    logger.info("Received %g MB, %g MB/s", stat.bytes/1e6,
-                stat.bytes/1e6/duration)
+    logger.info(
+        "Received %g MB, %g MB/s", stat.bytes / 1e6, stat.bytes / 1e6 / duration
+    )
 
     sent = stat.received + stat.lost
     if sent:
-        loss = stat.lost/sent
+        loss = stat.lost / sent
     else:
         loss = 1
-    logger.info("Loss: %s/%s batches (%g %%)", stat.lost, sent, loss*1e2)
+    logger.info("Loss: %s/%s batches (%g %%)", stat.lost, sent, loss * 1e2)
     return loss
 
 
 async def main():
     """Test CLI"""
     parser = argparse.ArgumentParser(description="Stabilizer streaming demo")
-    parser.add_argument("--port", type=int, default=9293,
-                        help="Local port to listen on")
-    parser.add_argument("--host", default="0.0.0.0",
-                        help="Local address to listen on")
-    parser.add_argument("--broker", default="mqtt",
-                        help="The MQTT broker address")
-    parser.add_argument("--maxsize", type=int, default=1,
-                        help="Frame queue size")
-    parser.add_argument("--duration", type=float, default=1.,
-                        help="Test duration")
+    parser.add_argument(
+        "--port", type=int, default=9293, help="Local port to listen on"
+    )
+    parser.add_argument("--host", default="0.0.0.0", help="Local address to listen on")
+    parser.add_argument("--broker", default="mqtt", help="The MQTT broker address")
+    parser.add_argument("--maxsize", type=int, default=1, help="Frame queue size")
+    parser.add_argument("--duration", type=float, default=1.0, help="Test duration")
     args = parser.parse_args()
 
     logging.basicConfig(level=logging.INFO)
     _transport, stream = await StabilizerStream.open(
-        args.host, args.port, args.broker, args.maxsize)
+        args.port, args.host, args.broker, args.maxsize
+    )
     await measure(stream, args.duration)