Add tests

Author: projectgus

canalystii/device.py

@@ -29,7 +29,7 @@ def __init__(
                 )
             )
             if not devices:
-                raise ValueError("No Canalyst-II device was found")
+                raise ValueError("No Canalyst-II USB device found")
             if len(devices) <= device_index:
                 raise ValueError(
                     f"Can't open device_index {device_index}, only {len(devices)} devices found."

tests/CAN_Spammer.ino

@@ -0,0 +1,52 @@
+// Spam messages out CAN0 as fast as the Due will allow!
+//
+// Use together with the can_spammer_test.py script to verify peak Canalyst-II
+// receive performance.
+//
+// NOTE: Requires due_can library. Library v2.0.1 sometimes sends CAN frames out
+// of order (or drops them, possibly?), so will give incorrect results. The
+// master branch library version doesn't have this issue.
+//
+#include "variant.h"
+#include <due_can.h>
+
+// Leave defined if you use native port, comment if using programming port
+//#define Serial SerialUSB
+
+void setup()
+{
+  Serial.begin(115200);
+  Can0.begin(CAN_BPS_1000K);
+  Serial.println("Starting...");
+}
+
+void loop()
+{
+  CAN_FRAME outgoing;
+  outgoing.id = 0x400;
+  outgoing.length = 8;
+  outgoing.data.value = 0xDEADBEEF00000000ULL;
+
+  while(1) {
+    // Sending will fail unless a device on the bus ACKs the frame
+    if(Can0.sendFrame(outgoing)) {
+      outgoing.data.value++;
+    }
+
+    // Every second, print an estimate of the sending rate
+    static uint32_t last_seconds;
+    uint32_t seconds = millis() / 1000;
+    if (seconds != last_seconds) {
+      uint32_t counter = outgoing.data.value; // Log the lower word only, Arduino doesn't format 64-bit integers
+      static uint32_t last_counter;
+      Serial.print(seconds);
+      Serial.print("s ");
+      Serial.print((uint32_t)counter);
+      Serial.print(" msgs ");
+      Serial.print(((uint32_t)counter - last_counter) / (seconds - last_seconds));
+      Serial.println("msg/s");
+      last_counter = counter;
+      last_seconds = seconds;
+    }
+  }
+}

tests/README.md

@@ -0,0 +1,20 @@
+# Tests
+
+Some basic automated tests for Canalyst-II Behaviour.
+
+## loopback_test.py
+
+Expects channel 0 & 1 to be joined together (low to low and high to high), with termination set correctly.
+
+This is a pytest module, run as `pytest loopback_test.py`
+
+## can_spammer_test.py
+
+Expects an Arduino Due running the "CAN_Spammer.ino" sketch to be attached to either Channel 0, Chanel 1, or both.
+
+Tests throughput receiving messages from the spammer. Prints throughput rate for messages received, and any unexpected messages that may indicate dropped frames.
+
+This is an ordinary script, doesn't print any pass/fail result just a throughput rate and an error count.
+
+Note that when the firmware clears the RX buffer it may still keep one message somehow (firmware bug?), so 1 error per channel is possible on any run.
+

tests/can_spammer_test.py

@@ -0,0 +1,60 @@
+#!/usr/bin/env python
+import canalystii
+import struct
+import time
+
+CHANNELS = (0, 1)  # Which channels to test on?
+TEST_SECS = 10  # How many seconds to sample for?
+SLEEP_BETWEEN = 0.05  # How long to sleep between samples?
+
+
+def main():
+    print("Connecting to Canalyst-II device...")
+    dev = canalystii.CanalystDevice(bitrate=1000000)
+
+    print(f"Testing for {TEST_SECS} seconds...")
+
+    deadline = time.time() + TEST_SECS
+    last = [None, None]
+    messages = [0, 0]
+    errors = [0, 0]
+
+    dev.clear_rx_buffer(0)
+    dev.clear_rx_buffer(1)
+
+    while time.time() < deadline:
+        for ch in CHANNELS:
+            for msg in dev.receive(ch):
+                if last[ch] is None:
+                    last[ch] = read_counter(msg)
+                else:
+                    messages[ch] += 1
+                    (last[ch], result) = check_id(msg, last[ch])
+                    if not result:
+                        errors[ch] += 1
+        time.sleep(0.05)
+
+    for ch in CHANNELS:
+        print(
+            f"Channel {ch}: Received {messages[ch]} messages ({messages[ch]/TEST_SECS}/sec), {errors[ch]} errors"
+        )
+
+
+def check_id(msg, last_counter):
+    msg_count = read_counter(msg)
+    result = True
+    if msg_count < last_counter:
+        print(f"Backwards {msg_count:#x} < prev {last_counter:#x}")
+        result = False
+    elif msg_count != last_counter + 1:
+        print(f"Unexpected {msg_count:#x} expected {last_counter + 1:#x}")
+        result = False
+    return (msg_count, result)
+
+
+def read_counter(msg):
+    return struct.unpack("<Q", bytes(msg.data))[0]
+
+
+if __name__ == "__main__":
+    main()