Mirrors_Framework/ODrive
1
0
Fork 0
mirror of https://github.com/odriverobotics/ODrive.git synced 2026-09-21 15:34:33 +08:00
Code Issues Actions Packages Projects Releases Wiki Activity

add test for extended CAN IDs

Browse Source
This commit is contained in:
Samuel Sadok
2020-05-18 18:52:24 +02:00
parent b0e7dcc6fd
commit 16059a9e9b
2 changed files with 29 additions and 20 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tools/odrive/tests/can_test.py
+27 -18
View File
@@ -17,8 +17,8 @@ command_set = {
'estop': (0x002, []), # tested
'get_motor_error': (0x003, [('motor_error', 'I', 1)]), # untested
'get_encoder_error': (0x004, [('encoder_error', 'I', 1)]), # untested
'get_sensorless_error': (0x004, [('sensorless_error', 'I', 1)]), # untested
'set_node_id': (0x006, [('node_id', 'H', 1)]), # tested
'get_sensorless_error': (0x005, [('sensorless_error', 'I', 1)]), # untested
'set_node_id': (0x006, [('node_id', 'I', 1)]), # tested
'set_requested_state': (0x007, [('requested_state', 'I', 1)]), # tested
# 0x008 not yet implemented
'get_encoder_estimates': (0x009, [('encoder_pos_estimate', 'f', 1), ('encoder_vel_estimate', 'f', 1)]), # partially tested
@@ -39,7 +39,7 @@ command_set = {
'clear_errors': (0x018, []), # partially tested
}
def command(bus, node_id_, cmd_name, **kwargs):
def command(bus, node_id_, extended_id, cmd_name, **kwargs):
cmd_spec = command_set[cmd_name]
cmd_id = cmd_spec[0]
fmt = '<' + ''.join([f for (n, f, s) in cmd_spec[1]]) # all little endian
@@ -49,10 +49,10 @@ def command(bus, node_id_, cmd_name, **kwargs):
fields = [((kwargs[n] / s) if f == 'f' else int(kwargs[n] / s)) for (n, f, s) in cmd_spec[1]]
data = struct.pack(fmt, *fields)
msg = can.Message(arbitration_id=((node_id_ << 5) | cmd_id), data=data)
msg = can.Message(arbitration_id=((node_id_ << 5) | cmd_id), extended_id=extended_id, data=data)
bus.send(msg)
async def record_messages(bus, node_id, cmd_name, timeout = 5.0):
async def record_messages(bus, node_id, extended_id, cmd_name, timeout = 5.0):
"""
Returns an async generator that yields a dictionary for each CAN message that
is received, provided that the CAN ID matches the expected value.
@@ -71,7 +71,7 @@ async def record_messages(bus, node_id, cmd_name, timeout = 5.0):
start = time.monotonic()
while True:
msg = await reader.get_message()
if ((msg.arbitration_id == ((node_id << 5) | cmd_id)) and not msg.is_remote_frame):
if ((msg.arbitration_id == ((node_id << 5) | cmd_id)) and (msg.is_extended_id == extended_id) and not msg.is_remote_frame):
fields = struct.unpack(fmt, msg.data[:(struct.calcsize(fmt))])
res = {n: (fields[i] * s) for (i, (n, f, s)) in enumerate(cmd_spec[1])}
res['t'] = time.monotonic()
@@ -81,13 +81,13 @@ async def record_messages(bus, node_id, cmd_name, timeout = 5.0):
finally:
notifier.stop()
async def request(bus, node_id, cmd_name, timeout = 1.0):
async def request(bus, node_id, extended_id, cmd_name, timeout = 1.0):
cmd_spec = command_set[cmd_name]
cmd_id = cmd_spec[0]
msg_generator = record_messages(bus, node_id, cmd_name, timeout)
msg_generator = record_messages(bus, node_id, extended_id, cmd_name, timeout)
msg = can.Message(arbitration_id=((node_id << 5) | cmd_id), data=[], is_remote_frame=True)
msg = can.Message(arbitration_id=((node_id << 5) | cmd_id), extended_id=extended_id, data=[], is_remote_frame=True)
bus.send(msg)
async for msg in msg_generator:
@@ -102,34 +102,43 @@ async def get_all(async_iterator):
class TestSimpleCAN():
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))
can_interfaces = list(testrig.get_connected_components(odrive.can, CanInterfaceComponent))
yield (odrive, can_interfaces, 0, False) # standard ID
yield (odrive, can_interfaces, 0xfedcba, True) # extended ID
def run_test(self, odrive: ODriveComponent, canbus: CanInterfaceComponent, logger: Logger):
def run_test(self, odrive: ODriveComponent, canbus: CanInterfaceComponent, node_id: int, extended_id: bool, logger: Logger):
# make sure no gpio input is overwriting our values
odrive.unuse_gpios()
node_id = 0
axis = odrive.handle.axis0
axis.clear_errors()
axis.config.can_node_id = node_id
axis.config.can_node_id_extended = extended_id
time.sleep(0.1)
def my_cmd(cmd_name, **kwargs): command(canbus.handle, node_id, cmd_name, **kwargs)
def my_req(cmd_name, **kwargs): return asyncio.run(request(canbus.handle, node_id, cmd_name, **kwargs))
def my_cmd(cmd_name, **kwargs): command(canbus.handle, node_id, extended_id, cmd_name, **kwargs)
def my_req(cmd_name, **kwargs): return asyncio.run(request(canbus.handle, node_id, extended_id, cmd_name, **kwargs))
def fence(): my_req('get_vbus_voltage') # fence to ensure the CAN command was sent
test_assert_eq(my_req('get_vbus_voltage')['vbus_voltage'], odrive.handle.vbus_voltage, accuracy=0.01)
my_cmd('set_node_id', node_id=node_id+20)
asyncio.run(request(canbus.handle, node_id+20, 'get_vbus_voltage'))
asyncio.run(request(canbus.handle, node_id+20, extended_id, 'get_vbus_voltage'))
test_assert_eq(axis.config.can_node_id, node_id+20)
# Reset node ID to default value
command(canbus.handle, node_id+20, 'set_node_id', node_id=node_id)
command(canbus.handle, node_id+20, extended_id, 'set_node_id', node_id=node_id)
fence()
test_assert_eq(axis.config.can_node_id, node_id)
# Check that extended node IDs are not carelessly projected to 6-bit IDs
extended_id = not extended_id
my_cmd('estop') # should not be accepted
extended_id = not extended_id
fence()
test_assert_eq(axis.error, errors.axis.ERROR_NONE)
axis.encoder.set_linear_count(123)
test_assert_eq(my_req('get_encoder_estimates')['encoder_pos_estimate'], 123.0, accuracy=0.01)
test_assert_eq(my_req('get_encoder_count')['encoder_shadow_count'], 123.0, accuracy=0.01)
@@ -205,7 +214,7 @@ class TestSimpleCAN():
logger.debug('testing heartbeat...')
# note that this will include the heartbeats that were received during the
# watchdog test (which takes 4.8s).
heartbeats = asyncio.run(get_all(record_messages(canbus.handle, node_id, 'heartbeat', timeout = 1.0)))
heartbeats = asyncio.run(get_all(record_messages(canbus.handle, node_id, extended_id, 'heartbeat', timeout = 1.0)))
test_assert_eq(len(heartbeats), 5.8 / 0.1, accuracy=0.05)
test_assert_eq([msg['error'] for msg in heartbeats[0:35]], [0] * 35) # before watchdog expiry
test_assert_eq([msg['error'] for msg in heartbeats[-10:]], [errors.axis.ERROR_WATCHDOG_TIMER_EXPIRED] * 10) # after watchdog expiry
tools/odrive/tests/test_runner.py
+2 -2
View File
@@ -679,7 +679,7 @@ def select_params(param_options):
# Select parameters from the resource list
# (this could be arbitrarily complex to improve parallelization of the tests)
for combination in get_combinations(param_options):
if all_unique(combination):
if all_unique([x for x in combination if isinstance(x, Component)]):
return list(combination)
return None
@@ -708,7 +708,7 @@ def run(tests):
test_cases = list(test.get_test_cases(testrig))
if len(test_cases) == 0:
logger.warn('no resources are available to conduct the test {}'.format(type(test).__name__))
logger.warn('no test cases are available to conduct the test {}'.format(type(test).__name__))
continue
for test_case in test_cases: