upgrade test runner to use more yaml and less hardcoding

tools/odrive/tests/calibration_test.py

@@ -6,7 +6,7 @@ from math import pi
 import os
 
 from fibre.utils import Logger
-from test_runner import AxisTestContext, MotorTestContext, EncoderTestContext, test_assert_eq, test_assert_no_error, request_state, program_teensy
+from test_runner import *
 from odrive.enums import *
 
 def modpm(val, range):
@@ -19,14 +19,19 @@ class TestMotorCalibration():
     and checks if the measurements match the expectation.
     """
 
-    def is_compatible(self, axis_ctx: AxisTestContext, motor_ctx: MotorTestContext):
-        return axis_ctx.yaml == motor_ctx.yaml['name'] # check if connected
+    def get_test_cases(self, testrig: TestRig):
+        """Returns all axes that are connected to a motor, along with the corresponding motor(s)"""
+        for odrive in testrig.get_components(ODriveComponent):
+            for axis in odrive.axes:
+                for motor in testrig.get_connected_components(axis, MotorComponent):
+                    yield (axis, motor)
 
-    def run_test(self, axis_ctx: AxisTestContext, motor_ctx: MotorTestContext, logger: Logger):
+    def run_test(self, axis_ctx: ODriveAxisComponent, motor_ctx: MotorComponent, logger: Logger):
         # reset old calibration values
         axis_ctx.handle.motor.config.phase_resistance = 0.0
         axis_ctx.handle.motor.config.phase_inductance = 0.0
         axis_ctx.handle.motor.config.pre_calibrated = False
+        axis_ctx.handle.config.enable_watchdog = False
 
         axis_ctx.handle.clear_errors()
 
@@ -47,10 +52,14 @@ class TestDisconnectedMotorCalibration():
     Tests if the motor calibration fails as expected if the phases are floating.
     """
 
-    def is_compatible(self, axis_ctx: AxisTestContext):
-        return axis_ctx.yaml == 'floating'
+    def get_test_cases(self, testrig: TestRig):
+        """Returns all axes that are disconnected"""
+        for odrive in testrig.get_components(ODriveComponent):
+            for axis in odrive.axes:
+                if axis.yaml == 'floating':
+                    yield (axis,)
 
-    def run_test(self, axis_ctx: AxisTestContext, logger: Logger):
+    def run_test(self, axis_ctx: ODriveAxisComponent, logger: Logger):
         axis = axis_ctx.handle
 
         # reset old calibration values
@@ -73,14 +82,21 @@ class TestEncoderDirFind():
     Runs the encoder index search.
     """
 
-    def is_compatible(self, axis_ctx: AxisTestContext, motor_ctx: MotorTestContext, enc_ctx: EncoderTestContext):
-        return (axis_ctx.yaml == motor_ctx.yaml['name']) and (axis_ctx.num == enc_ctx.num) # check if connected
+    def get_test_cases(self, testrig: TestRig):
+        for odrive in testrig.get_components(ODriveComponent):
+            for num in range(2):
+                encoders = testrig.get_connected_components({
+                    'a': (odrive.encoders[num].a, False),
+                    'b': (odrive.encoders[num].b, False)
+                }, EncoderComponent)
+                motors = testrig.get_connected_components(odrive.axes[num], MotorComponent)
 
-    def run_test(self, axis_ctx: AxisTestContext, motor_ctx: MotorTestContext, enc_ctx: EncoderTestContext, logger: Logger):
+                for motor, encoder in itertools.product(motors, encoders):
+                    if encoder.impl in testrig.get_connected_components(motor):
+                        yield (odrive.axes[num], motor, encoder)
+
+    def run_test(self, axis_ctx: ODriveAxisComponent, motor_ctx: MotorComponent, enc_ctx: EncoderComponent, logger: Logger):
         axis = axis_ctx.handle
-        # TODO: read teensy config from YAML file
-        hexfile = 'encoder_pass_through.ino.hex'
-        program_teensy(os.path.join(os.path.dirname(__file__), hexfile), 26, logger)
         time.sleep(1.0) # wait for PLLs to stabilize
 
         # Set motor calibration values
@@ -110,14 +126,21 @@ class TestEncoderOffsetCalibration():
     Runs the encoder index search.
     """
 
-    def is_compatible(self, axis_ctx: AxisTestContext, motor_ctx: MotorTestContext, enc_ctx: EncoderTestContext):
-        return (axis_ctx.yaml == motor_ctx.yaml['name']) and (axis_ctx.num == enc_ctx.num) # check if connected
+    def get_test_cases(self, testrig: TestRig):
+        for odrive in testrig.get_components(ODriveComponent):
+            for num in range(2):
+                encoders = testrig.get_connected_components({
+                    'a': (odrive.encoders[num].a, False),
+                    'b': (odrive.encoders[num].b, False)
+                }, EncoderComponent)
+                motors = testrig.get_connected_components(odrive.axes[num], MotorComponent)
 
-    def run_test(self, axis_ctx: AxisTestContext, motor_ctx: MotorTestContext, enc_ctx: EncoderTestContext, logger: Logger):
+                for motor, encoder in itertools.product(motors, encoders):
+                    if encoder.impl in testrig.get_connected_components(motor):
+                        yield (odrive.axes[num], motor, encoder)
+
+    def run_test(self, axis_ctx: ODriveAxisComponent, motor_ctx: MotorComponent, enc_ctx: EncoderComponent, logger: Logger):
         axis = axis_ctx.handle
-        # TODO: read teensy config from YAML file
-        hexfile = 'encoder_pass_through.ino.hex'
-        program_teensy(os.path.join(os.path.dirname(__file__), hexfile), 26, logger)
         time.sleep(1.0) # wait for PLLs to stabilize
 
         # Set motor calibration values
@@ -127,9 +150,9 @@ class TestEncoderOffsetCalibration():
 
         # Set calibration settings
         axis_ctx.handle.motor.config.direction = 0
-        enc_ctx.handle.config.use_index = False
-        enc_ctx.handle.config.calib_scan_omega = 12.566 # 2 electrical revolutions per second
-        enc_ctx.handle.config.calib_scan_distance = 50.265 # 8 revolutions
+        axis_ctx.handle.encoder.config.use_index = False
+        axis_ctx.handle.encoder.config.calib_scan_omega = 12.566 # 2 electrical revolutions per second
+        axis_ctx.handle.encoder.config.calib_scan_distance = 50.265 # 8 revolutions
 
         axis_ctx.handle.clear_errors()
 
@@ -141,7 +164,7 @@ class TestEncoderOffsetCalibration():
         test_assert_eq(axis_ctx.handle.current_state, AXIS_STATE_IDLE)
         test_assert_no_error(axis_ctx)
 
-        test_assert_eq(enc_ctx.handle.is_ready, True)
+        test_assert_eq(axis_ctx.handle.encoder.is_ready, True)
         test_assert_eq(axis_ctx.handle.motor.config.direction in [-1, 1], True)
 
 
@@ -152,14 +175,27 @@ class TestEncoderIndexSearch():
     host's GPIO.
     """
 
-    def is_compatible(self, axis_ctx: AxisTestContext, motor_ctx: MotorTestContext, enc_ctx: EncoderTestContext):
-        return (axis_ctx.yaml == motor_ctx.yaml['name']) and (axis_ctx.num == enc_ctx.num) # check if connected
+    def get_test_cases(self, testrig: TestRig):
+        for odrive in testrig.get_components(ODriveComponent):
+            for num in range(2):
+                encoders = testrig.get_connected_components({
+                    'a': (odrive.encoders[num].a, False),
+                    'b': (odrive.encoders[num].b, False)
+                }, EncoderComponent)
+                motors = testrig.get_connected_components(odrive.axes[num], MotorComponent)
+                z_gpio = list(testrig.get_connected_components((odrive.encoders[num].z, False), LinuxGpioComponent))
 
-    def run_test(self, axis_ctx: AxisTestContext, motor_ctx: MotorTestContext, enc_ctx: EncoderTestContext, logger: Logger):
+                for motor, encoder in itertools.product(motors, encoders):
+                    if encoder.impl in testrig.get_connected_components(motor):
+                        yield (odrive.axes[num], motor, encoder, z_gpio)
+
+    def run_test(self, axis_ctx: ODriveAxisComponent, motor_ctx: MotorComponent, enc_ctx: EncoderComponent, z_gpio: LinuxGpioComponent, logger: Logger):
         axis = axis_ctx.handle
-        # TODO: read teensy config from YAML file
-        hexfile = 'encoder_pass_through.ino.hex'
-        program_teensy(os.path.join(os.path.dirname(__file__), hexfile), 26, logger)
+        cpr = int(enc_ctx.yaml['cpr'])
+        
+        z_gpio.config(output=True)
+        z_gpio.write(False)
+
         time.sleep(1.0) # wait for PLLs to stabilize
 
         # Set motor calibration values
@@ -177,24 +213,20 @@ class TestEncoderIndexSearch():
 
         time.sleep(3)
 
-        test_assert_eq(enc_ctx.handle.index_found, False)
-        with open("/sys/class/gpio/gpio{}/direction".format(20), "w") as fp:
-            fp.write("out")
-        with open("/sys/class/gpio/gpio{}/value".format(20), "w") as gpio:
-            gpio.write("0")
+        test_assert_eq(axis_ctx.handle.encoder.index_found, False)
         time.sleep(0.1)
-        with open("/sys/class/gpio/gpio{}/value".format(20), "w") as gpio:
-            gpio.write("1")
-        test_assert_eq(enc_ctx.handle.index_found, True)
+        z_gpio.write(True)
+        test_assert_eq(axis_ctx.handle.encoder.index_found, True)
+        z_gpio.write(False)
         
         test_assert_eq(axis_ctx.handle.current_state, AXIS_STATE_IDLE)
         test_assert_no_error(axis_ctx)
 
-        test_assert_eq(enc_ctx.handle.shadow_count, 0.0, range=20)
-        test_assert_eq(enc_ctx.handle.count_in_cpr, 0.0, range=20)
-        test_assert_eq(enc_ctx.handle.pos_estimate, 0.0, range=20)
-        test_assert_eq(enc_ctx.handle.pos_cpr, 0.0, range=20)
-        test_assert_eq(enc_ctx.handle.pos_abs, 0.0, range=20)
+        test_assert_eq(axis_ctx.handle.encoder.shadow_count, 0.0, range=20)
+        test_assert_eq(modpm(axis_ctx.handle.encoder.count_in_cpr, cpr), 0.0, range=20)
+        test_assert_eq(axis_ctx.handle.encoder.pos_estimate, 0.0, range=20)
+        test_assert_eq(axis_ctx.handle.encoder.pos_cpr, 0.0, range=20)
+        test_assert_eq(axis_ctx.handle.encoder.pos_abs, 0.0, range=20)
 
 
 if __name__ == '__main__':

tools/odrive/tests/can_test.py

@@ -7,8 +7,8 @@ import asyncio
 import time
 
 from fibre.utils import Logger
-from odrive.enums import errors
-from test_runner import CANTestContext, ODriveTestContext, test_assert_eq
+from odrive.enums import *
+from test_runner import *
 
 # Each argument is described as tuple (name, format, scale).
 # Struct format codes: https://docs.python.org/2/library/struct.html
@@ -81,10 +81,12 @@ async def request(bus, node_id, cmd_name, timeout = 1.0):
 
 
 class TestSimpleCAN():
-    def is_compatible(self, canbus: CANTestContext, odrive: ODriveTestContext):
-        return canbus.yaml['bus'] == odrive.yaml['can'] # check if connected
+    def get_test_cases(self, testrig: TestRig):
+        for odrive in testrig.get_components(ODriveComponent):
+            can_interfaces = testrig.get_connected_components(odrive.can, CanInterfaceComponent)
+            yield (odrive, list(can_interfaces))
 
-    def run_test(self, canbus: CANTestContext, odrive: ODriveTestContext, logger: Logger):
+    def run_test(self, odrive: ODriveComponent, canbus: CanInterfaceComponent, logger: Logger):
         node_id = 0
         axis = odrive.handle.axis0
         axis.config.can_node_id = node_id
@@ -139,11 +141,13 @@ class TestSimpleCAN():
         test_assert_eq(axis.controller.input_vel, 2.0, range=0.01)
         test_assert_eq(axis.controller.input_current, 3.0, range=0.001)
 
+        axis.controller.config.control_mode = CTRL_MODE_VELOCITY_CONTROL
         my_cmd('set_input_vel', input_vel=-10.0, cur_ff=30.1234)
         fence()
         test_assert_eq(axis.controller.input_vel, -10.0, range=0.01)
         test_assert_eq(axis.controller.input_current, 30.1234, range=0.01)
 
+        axis.controller.config.control_mode = CTRL_MODE_CURRENT_CONTROL
         my_cmd('set_input_current', input_current=3.1415)
         fence()
         test_assert_eq(axis.controller.input_current, 3.1415, range=0.01)

tools/odrive/tests/encoder_test.py

@@ -6,15 +6,59 @@ from math import pi
 import os
 
 from fibre.utils import Logger
-from test_runner import EncoderTestContext, test_assert_eq, program_teensy
+from test_runner import *
+
+
+teensy_code_template = """
+void setup() {
+  pinMode({enc_a}, OUTPUT);
+  pinMode({enc_b}, OUTPUT);
+}
+
+int cpr = 8192;
+int rpm = 30;
+
+// the loop routine runs over and over again forever:
+void loop() {
+  int microseconds_per_count = (1000000 * 60 / cpr / rpm);
+
+  for (;;) {
+    digitalWrite({enc_a}, HIGH);
+    delayMicroseconds(microseconds_per_count);
+    digitalWrite({enc_b}, HIGH);
+    delayMicroseconds(microseconds_per_count);
+    digitalWrite({enc_a}, LOW);
+    delayMicroseconds(microseconds_per_count);
+    digitalWrite({enc_b}, LOW);
+    delayMicroseconds(microseconds_per_count);
+  }
+}
+
+"""
+
 
 def modpm(val, range):
     return ((val + (range / 2)) % range) - (range / 2)
 
 class TestIncrementalEncoder():
 
-    def is_compatible(self, enc_ctx: EncoderTestContext):
-        return True
+    def get_test_cases(self, testrig: TestRig):
+        for odrive in testrig.get_components(ODriveComponent):
+            for encoder in odrive.encoders:
+                # Find the Teensy that is connected to the encoder pins and the corresponding Teensy GPIOs
+
+                gpio_conns = [
+                    testrig.get_directly_connected_components(encoder.a),
+                    testrig.get_directly_connected_components(encoder.b),
+                ]
+
+                valid_combinations = [
+                    [combination[0].parent] + list(combination)
+                    for combination in itertools.product(*gpio_conns)
+                    if ((len(set(c.parent for c in combination)) == 1) and isinstance(combination[0].parent, TeensyComponent))
+                ]
+
+                yield (encoder, valid_combinations)
 
     def run_delta_test(self, encoder, true_cps, with_cpr):
         encoder.config.cpr = with_cpr
@@ -45,17 +89,15 @@ class TestIncrementalEncoder():
  
             time.sleep(0.01)
 
-    def run_test(self, enc_ctx: EncoderTestContext, logger: Logger):
+    def run_test(self, enc: EncoderComponent, teensy: TeensyComponent, teensy_gpio_a: int, teensy_gpio_b: int, logger: Logger):
         true_cps = 8192*-0.5 # counts per second generated by the virtual encoder
-        # TODO: read teensy config from YAML file
-        if enc_ctx.num == 0:
-            hexfile = 'enc0_sim_-4096cps.ino.hex'
-        else:
-            hexfile = 'enc1_sim_-4096cps.ino.hex'
-        program_teensy(os.path.join(os.path.dirname(__file__), hexfile), 26, logger)
+        
+        code = teensy_code_template.replace("{enc_a}", str(teensy_gpio_a.num)).replace("{enc_b}", str(teensy_gpio_b.num))
+        teensy.compile_and_program(code)
+
         time.sleep(1.0) # wait for PLLs to stabilize
 
-        encoder = enc_ctx.handle
+        encoder = enc.handle
 
         # The true encoder count and PLL output should be roughly the same.
         # At 8192 CPR and 0.5 RPM, the delta because of sequential reading is
@@ -67,8 +109,8 @@ class TestIncrementalEncoder():
         #encoder.config.cpr = configured_cpr
         #expected_delta = true_cps/1200
         #for _ in range(1000):
-        #    first = enc_ctx.handle.axis0.encoder.count_in_cpr
-        #    second = enc_ctx.handle.axis0.encoder.pos_cpr
+        #    first = enc.handle.axis0.encoder.count_in_cpr
+        #    second = enc.handle.axis0.encoder.pos_cpr
         #    test_assert_eq(modpm(second - first, configured_cpr), expected_delta, range=abs(true_cps/500))
         #    time.sleep(0.001)
 

tools/odrive/tests/nvm_test.py

@@ -7,17 +7,18 @@ import os
 
 import fibre
 from fibre.utils import Logger
-from test_runner import ODriveTestContext, test_assert_eq
+from test_runner import *
 
 class TestStoreAndReboot():
     """
     Stores the current configuration to NVM and reboots.
     """
 
-    def is_compatible(self, odrive: ODriveTestContext):
-        return True
+    def get_test_cases(self, testrig: TestRig):
+        for odrive in testrig.get_components(ODriveComponent):
+            yield (odrive,)
 
-    def run_with_values(self, values, odrive: ODriveTestContext, logger: Logger):
+    def run_with_values(self, odrive: ODriveComponent, values: list, logger: Logger):
         logger.debug("storing configuration and rebooting...")
 
         for value in values:
@@ -31,15 +32,15 @@ class TestStoreAndReboot():
         odrive.handle = None
         time.sleep(2)
 
-        odrive.make_available(logger)
+        odrive.prepare(logger)
 
         logger.debug("verifying configuration after reboot...")
         test_assert_eq(odrive.handle.config.brake_resistance, values[-1], accuracy=0.01)
 
-    def run_test(self, odrive: ODriveTestContext, logger: Logger):
-        self.run_with_values([0.5, 1.0, 1.5], odrive, logger)
-        self.run_with_values([2.5, 3.7], odrive, logger)
-        self.run_with_values([0.47], odrive, logger)
+    def run_test(self, odrive: ODriveComponent, logger: Logger):
+        self.run_with_values(odrive, [0.5, 1.0, 1.5], logger)
+        self.run_with_values(odrive, [2.5, 3.7], logger)
+        self.run_with_values(odrive, [0.47], logger)
 
 if __name__ == '__main__':
     test_runner.run(TestStoreAndReboot())

tools/odrive/tests/pwm_input_test.py

@@ -5,13 +5,38 @@ import time
 import math
 import os
 
-import fibre
-from fibre.utils import Logger
 from odrive.enums import errors
-from test_runner import ODriveTestContext, test_assert_eq, program_teensy
+from test_runner import *
+
+
+teensy_code_template = """
+float position = 0; // between 0 and 1
+float velocity = 1; // [position per second]
+
+void setup() {
+{setup_code}
+}
+
+// the loop routine runs over and over again forever:
+void loop() {
+  int high_microseconds = 1000 + (int)(position * 1000.0f);
+
+{set_high_code}
+  delayMicroseconds(high_microseconds);
+{set_low_code}
+
+  // Wait for a total of 20ms.
+  // delayMicroseconds() only works well for values <= 16383
+  delayMicroseconds(10000 - high_microseconds);
+  delayMicroseconds(10000);
+
+  position += velocity * 0.02;
+  while (position > 1.0)
+      position -= 1.0;
+}
+"""
+
 
-#def modpm(val, lower_bound, upper_bound):
-#    return ((val - lower_bound) % (upper_bound - lower_bound)) - lower_bound
 
 def modpm(val, range):
     return ((val + (range / 2)) % range) - (range / 2)
@@ -28,8 +53,24 @@ class TestPwmInput():
     Note: this test is currently only written for ODrive 3.6 (or similar GPIO layout).
     """
 
-    def is_compatible(self, odrive: ODriveTestContext):
-        return True
+    def get_test_cases(self, testrig: TestRig):
+        for odrive in testrig.get_components(ODriveComponent):
+            # Find the Teensy that is connected to gpios 1-4 of the ODrive and the corresponding Teensy GPIOs
+
+            gpio_conns = [
+                testrig.get_directly_connected_components(odrive.gpio1),
+                testrig.get_directly_connected_components(odrive.gpio2),
+                testrig.get_directly_connected_components(odrive.gpio3),
+                testrig.get_directly_connected_components(odrive.gpio4)
+            ]
+
+            valid_combinations = [
+                [combination[0].parent] + list(combination)
+                for combination in itertools.product(*gpio_conns)
+                if ((len(set(c.parent for c in combination)) == 1) and isinstance(combination[0].parent, TeensyComponent))
+            ]
+
+            yield (odrive, valid_combinations)
 
     def run_delta_test(self, attr, with_min, with_max, timeout = 5.0):
         rounds_per_s = 1.0
@@ -84,10 +125,16 @@ class TestPwmInput():
         test_assert_eq(min_val, with_min, range = step_size)
         test_assert_eq(max_val, with_max, range = step_size)
 
-    def run_test(self, odrive: ODriveTestContext, logger: Logger):
+    def run_test(self, odrive: ODriveComponent, teensy: TeensyComponent, teensy_gpio1: int, teensy_gpio2: int, teensy_gpio3: int, teensy_gpio4: int, logger: Logger):
         # TODO: test each GPIO separately
-        hexfile = 'pwm_sim.ino.hex'
-        program_teensy(os.path.join(os.path.dirname(__file__), hexfile), 26, logger)
+        
+        setup_code = "\n".join("  pinMode(" + str(gpio.num) + ", OUTPUT);" for gpio in [teensy_gpio1, teensy_gpio2, teensy_gpio3, teensy_gpio4])
+        set_high_code = "\n".join("  digitalWrite(" + str(gpio.num) + ", HIGH);" for gpio in [teensy_gpio1, teensy_gpio2, teensy_gpio3, teensy_gpio4])
+        set_low_code = "\n".join("  digitalWrite(" + str(gpio.num) + ", LOW);" for gpio in [teensy_gpio1, teensy_gpio2, teensy_gpio3, teensy_gpio4])
+
+        code = teensy_code_template.replace("{setup_code}", setup_code).replace("{set_high_code}", set_high_code).replace("{set_low_code}", set_low_code)
+        teensy.compile_and_program(code)
+
         time.sleep(1.0) # wait for PLLs to stabilize
 
         logger.debug("Set up PWM input...")
@@ -106,15 +153,7 @@ class TestPwmInput():
         odrive.handle.config.gpio4_pwm_mapping.min = -20000
         odrive.handle.config.gpio4_pwm_mapping.max = 20000
         
-        # Save and reboot
-        odrive.handle.save_configuration()
-        try:
-            odrive.handle.reboot()
-        except fibre.ChannelBrokenException:
-            pass # this is expected
-        odrive.handle = None
-        time.sleep(2)
-        odrive.make_available(logger)
+        odrive.save_config_and_reboot()
 
         logger.debug("Check if PWM on GPIO1 works...")
         self.run_delta_test(odrive.handle.axis0.controller._remote_attributes['input_pos'], -50, 200)

tools/odrive/tests/uart_ascii_test.py

@@ -5,13 +5,12 @@ import struct
 import time
 import os
 import io
-import serial
 import functools
 import operator
 
 from fibre.utils import Logger
 from odrive.enums import *
-from test_runner import ODriveTestContext, test_assert_eq, test_assert_no_error, program_teensy
+from test_runner import *
 
 
 def append_checksum(command):
@@ -32,28 +31,29 @@ def reset_state(ser):
     ser.flushInput() # discard response
 
 class TestUartAscii():
-    def is_compatible(self, odrive: ODriveTestContext):
-        return True
+    def get_test_cases(self, testrig: TestRig):
+        for odrive in testrig.get_components(ODriveComponent):
+            ports = list(testrig.get_connected_components({
+                'rx': (odrive.gpio1, True),
+                'tx': (odrive.gpio2, False)
+            }, SerialPortComponent))
+            yield (odrive, ports)
 
-    def run_test(self, odrive: ODriveTestContext, logger: Logger):
+    def run_test(self, odrive: ODriveComponent, port: SerialPortComponent, logger: Logger):
         """
         Tests the most important functions of the ASCII protocol.
         """
 
-        # Disable noise
-        with open("/sys/class/gpio/gpio{}/direction".format(20), "w") as fp:
-            fp.write("out")
-        with open("/sys/class/gpio/gpio{}/value".format(20), "w") as gpio:
-            gpio.write("0")
-
-        hexfile = 'uart_pass_through.ino.hex'
-        program_teensy(os.path.join(os.path.dirname(__file__), hexfile), 26, logger)
-        time.sleep(1.0)
+        if (odrive.handle.config.gpio1_pwm_mapping.endpoint != (0,0)) or (odrive.handle.config.gpio2_pwm_mapping.endpoint != (0,0)):
+            logger.debug('UART pins in use. Reconfiguring...')
+            odrive.handle.config.gpio1_pwm_mapping.endpoint = None
+            odrive.handle.config.gpio2_pwm_mapping.endpoint = None
+            odrive.save_config_and_reboot()
 
         odrive.handle.axis0.config.enable_step_dir = False
         odrive.handle.config.enable_uart = True
 
-        with serial.Serial('/dev/ttyS0', 115200, timeout=1) as ser:
+        with port.open(115200) as ser:
             # reset port to known state
             reset_state(ser)
 
@@ -159,84 +159,24 @@ class TestUartAscii():
             # TODO: test cases for 't', 'ss', 'se', 'sr' commands
 
 
-
-class TestUartNoise():
-    def is_compatible(self, odrive: ODriveTestContext):
-        return True
-
-    def run_test(self, odrive: ODriveTestContext, logger: Logger):
-        """
-        Tests if the UART can handle invalid signals.
-        """
-
-        # Disable noise
-        with open("/sys/class/gpio/gpio{}/direction".format(20), "w") as fp:
-            fp.write("out")
-        with open("/sys/class/gpio/gpio{}/value".format(20), "w") as gpio:
-            gpio.write("0")
-
-        hexfile = 'uart_pass_through.ino.hex'
-        program_teensy(os.path.join(os.path.dirname(__file__), hexfile), 26, logger)
-        time.sleep(1.0)
-
-        odrive.handle.axis0.config.enable_step_dir = False
-        odrive.handle.config.enable_uart = True
-
-        with serial.Serial('/dev/ttyS0', 115200, timeout=1) as ser:
-            # reset port to known state
-            reset_state(ser)
-
-            # Enable square wave of ~1.6MHz on the ODrive's RX line
-            with open("/sys/class/gpio/gpio{}/value".format(20), "w") as gpio:
-                gpio.write("1")
-
-            time.sleep(0.1)
-            reset_state(ser)
-
-            # Read an attribute (should fail because the command is not passed through)
-            ser.write(b'r vbus_voltage\n')
-            test_assert_eq(ser.readline(), b'')
-
-            # Disable square wave
-            with open("/sys/class/gpio/gpio{}/value".format(20), "w") as gpio:
-                gpio.write("0")
-
-            # Give receiver some time to recover
-            time.sleep(0.1)
-
-            # reset port to known state
-            reset_state(ser)
-
-            # Try again
-            ser.write(b'r vbus_voltage\n')
-            response = float(ser.readline().strip())
-            test_assert_eq(response, odrive.handle.vbus_voltage, accuracy=0.1)
-
-
-
 class TestUartBurnIn():
-    def is_compatible(self, odrive: ODriveTestContext):
-        return True
+    def get_test_cases(self, testrig: TestRig):
+        for odrive in testrig.get_components(ODriveComponent):
+            ports = list(testrig.get_connected_components({
+                'rx': (odrive.gpio1, True),
+                'tx': (odrive.gpio2, False)
+            }, SerialPortComponent))
+            yield (odrive, ports)
 
-    def run_test(self, odrive: ODriveTestContext, logger: Logger):
+    def run_test(self, odrive: ODriveComponent, port: SerialPortComponent, logger: Logger):
         """
         Tests if the ASCII protocol can handle 64kB of random data being thrown at it.
         """
 
-        # Disable noise
-        with open("/sys/class/gpio/gpio{}/direction".format(20), "w") as fp:
-            fp.write("out")
-        with open("/sys/class/gpio/gpio{}/value".format(20), "w") as gpio:
-            gpio.write("0")
-
-        hexfile = 'uart_pass_through.ino.hex'
-        program_teensy(os.path.join(os.path.dirname(__file__), hexfile), 26, logger)
-        time.sleep(1.0)
-
         odrive.handle.axis0.config.enable_step_dir = False
         odrive.handle.config.enable_uart = True
 
-        with serial.Serial('/dev/ttyS0', 115200, timeout=1) as ser:
+        with port.open(115200) as ser:
             with open('/dev/random', 'rb') as rand:
                 buf = rand.read(65536)
             ser.write(buf)
@@ -250,9 +190,81 @@ class TestUartBurnIn():
             test_assert_eq(response, odrive.handle.vbus_voltage, accuracy=0.1)
 
 
+class TestUartNoise():
+    def get_test_cases(self, testrig: TestRig):
+        for odrive in testrig.get_components(ODriveComponent):
+            # For every ODrive, find a connected serial port which has a teensy
+            # in between, so that we can inject noise,
+
+            ports = list(testrig.get_connected_components({
+                'rx': (odrive.gpio1, True),
+                'tx': (odrive.gpio2, False)
+            }, SerialPortComponent))
+
+            # Hack the bus objects to enable noise_enable functionality on the TX line.
+
+            def get_noise_gpio(bus):
+                teensy = bus.gpio_tuples[1][0]
+                for teensy_gpio in teensy.gpios:
+                    for other_gpio in testrig.get_directly_connected_components(teensy_gpio):
+                        if isinstance(other_gpio, LinuxGpioComponent):
+                            return teensy_gpio, other_gpio
+                return None
+
+            for idx, bus in enumerate(ports):
+                noise_gpio_on_teensy, noise_gpio_on_rpi = get_noise_gpio(bus)
+                assert(noise_gpio_on_rpi)
+                t, i, o, _ = bus.gpio_tuples[1]
+                bus.gpio_tuples[1] = (t, i, o, noise_gpio_on_teensy)
+                ports[idx] = (bus, noise_gpio_on_rpi)
+                
+            yield (odrive, ports)
+
+    def run_test(self, odrive: ODriveComponent, port: SerialPortComponent, noise_enable: LinuxGpioComponent, logger: Logger):
+        """
+        Tests if the UART can handle invalid signals.
+        """
+        noise_enable.config(output=True)
+        noise_enable.write(False)
+        time.sleep(0.1)
+
+        odrive.handle.axis0.config.enable_step_dir = False
+        odrive.handle.config.enable_uart = True
+
+        with port.open(115200) as ser:
+            # reset port to known state
+            reset_state(ser)
+
+            # Enable square wave of ~1.6MHz on the ODrive's RX line
+            noise_enable.write(True)
+
+            time.sleep(0.1)
+            reset_state(ser)
+
+            time.sleep(1.0)
+
+            # Read an attribute (should fail because the command is not passed through)
+            ser.write(b'r vbus_voltage\n')
+            test_assert_eq(ser.readline(), b'')
+
+            # Disable square wave
+            noise_enable.write(False)
+
+            # Give receiver some time to recover
+            time.sleep(0.1)
+
+            # reset port to known state
+            reset_state(ser)
+
+            # Try again
+            ser.write(b'r vbus_voltage\n')
+            response = float(ser.readline().strip())
+            test_assert_eq(response, odrive.handle.vbus_voltage, accuracy=0.1)
+
+
 if __name__ == '__main__':
     test_runner.run([
         TestUartAscii(),
-        TestUartNoise(),
         TestUartBurnIn(),
+        TestUartNoise(),
     ])

tools/test-rig-rpi.yaml

@@ -8,15 +8,24 @@ components:
     name: rpi
     ssh: odrv
     net: homenet
-    can0: main_canbus
-    uart0: /dev/serial/by-id/[not-yet-used]
+    components:
+      - type: uart
+        name: uart0
+        port: /dev/ttyS0
+        connected-to: main_uart
+      - type: can
+        name: can0
+        interface: can0
+        connected-to: odrive.can
+      - {type: gpio, num: 20} # need to specify GPIOs explicitly for the generalpurpose type
+      - {type: gpio, num: 26}
 
   - type: programmer
     name: The Blue STLink/v2
     id: '\x53\x3f\x75\x06\x49\x3f\x49\x51\x44\x54\x19\x3f'
 
   - type: odrive
-    name: ODrive
+    name: odrive
     board-version: v3.6-58V
     serial-number: "20703595524B"
     brake-resistance: 0.47
@@ -40,15 +49,35 @@ components:
     max-voltage: 40
 
   - type: encoder
-    name: real_encoder_0
-    cpr: 8192
-    max-rpm: 7000
-
-  - type: encoder
-    name: real_encoder_1
+    name: real_encoder
     cpr: 8192
     max-rpm: 7000
 
   - type: teensy
     name: teensy
-    
+
+connections:
+  - ['odrive.can', 'rpi.can0']
+  - ['teensy.program', 'rpi.gpio26']
+  - ['teensy.gpio11', 'rpi.uart0.tx']
+  - ['teensy.gpio12', 'rpi.uart0.rx']
+  - ['teensy.gpio10', 'odrive.gpio1']
+  - ['teensy.gpio9', 'odrive.gpio2']
+  - ['teensy.gpio8', 'odrive.gpio3']
+  - ['teensy.gpio7', 'odrive.gpio4']
+  - ['teensy.gpio15', 'odrive.gpio5']
+  - ['teensy.gpio16', 'odrive.gpio6']
+  - ['teensy.gpio17', 'odrive.gpio7']
+  - ['teensy.gpio18', 'odrive.gpio8']
+  - ['teensy.gpio4', 'rpi.gpio20']
+  - ['teensy.gpio23', 'odrive.encoder0.z']
+  - ['teensy.gpio22', 'odrive.encoder0.a']
+  - ['teensy.gpio21', 'odrive.encoder0.b']
+  - ['teensy.gpio20', 'odrive.encoder1.z']
+  - ['teensy.gpio19', 'odrive.encoder1.a']
+  - ['teensy.gpio18', 'odrive.encoder1.b']
+  - ['teensy.gpio0', 'real_encoder.z']
+  - ['teensy.gpio1', 'real_encoder.a']
+  - ['teensy.gpio2', 'real_encoder.b']
+  - ['odrive.axis0', 'D5065-270KV_0']
+  - ['D5065-270KV_0', 'real_encoder']