diff --git a/.gitignore b/.gitignore index b0f71da4d..6b813427e 100644 --- a/.gitignore +++ b/.gitignore @@ -15,6 +15,7 @@ __pycache__/ # Distribution / packaging .Python env/ +venv/ build/ develop-eggs/ dist/ diff --git a/CONTRIBUTORS.txt b/CONTRIBUTORS.txt index 5e5ea882b..3ce2ff730 100644 --- a/CONTRIBUTORS.txt +++ b/CONTRIBUTORS.txt @@ -21,3 +21,4 @@ Eduard Bröcker Boris Wenzlaff Pierre-Luc Tessier Gagné Felix Divo +Kristian Sloth Lauszus diff --git a/can/viewer.py b/can/viewer.py new file mode 100644 index 000000000..c120df994 --- /dev/null +++ b/can/viewer.py @@ -0,0 +1,483 @@ +#!/usr/bin/python +# coding: utf-8 +# +# Copyright (C) 2018 Kristian Sloth Lauszus. All rights reserved. +# +# Contact information +# ------------------- +# Kristian Sloth Lauszus +# Web : http://www.lauszus.com +# e-mail : lauszus@gmail.com + +from __future__ import absolute_import, print_function + +import argparse +import can +import curses +import os +import struct +import sys +import time + +from curses.ascii import ESC as KEY_ESC, SP as KEY_SPACE +from typing import Dict, List, Tuple, Union + +from can import __version__ + + +class CanViewer: + + def __init__(self, stdscr, bus, data_structs, testing=False): + self.stdscr = stdscr + self.bus = bus + self.data_structs = data_structs + + # Initialise the ID dictionary, start timestamp, scroll and variable for pausing the viewer + self.ids = {} + self.start_time = None + self.scroll = 0 + self.paused = False + + # Get the window dimensions - used for resizing the window + self.y, self.x = self.stdscr.getmaxyx() + + # Do not wait for key inputs, disable the cursor and choose the background color automatically + self.stdscr.nodelay(True) + curses.curs_set(0) + curses.use_default_colors() + + # Used to color error frames red + curses.init_pair(1, curses.COLOR_RED, -1) + + if not testing: # pragma: no cover + self.run() + + def run(self): + # Clear the terminal and draw the header + self.draw_header() + + while 1: + # Do not read the CAN-Bus when in paused mode + if not self.paused: + # Read the CAN-Bus and draw it in the terminal window + msg = self.bus.recv(timeout=1. / 1000.) + if msg is not None: + self.draw_can_bus_message(msg) + else: + # Sleep 1 ms, so the application does not use 100 % of the CPU resources + time.sleep(1. / 1000.) + + # Read the terminal input + key = self.stdscr.getch() + + # Stop program if the user presses ESC or 'q' + if key == KEY_ESC or key == ord('q'): + break + + # Clear by pressing 'c' + elif key == ord('c'): + self.ids = {} + self.start_time = None + self.scroll = 0 + self.draw_header() + + # Sort by pressing 's' + elif key == ord('s'): + # Sort frames based on the CAN-Bus ID + self.draw_header() + for i, key in enumerate(sorted(self.ids.keys())): + # Set the new row index, but skip the header + self.ids[key]['row'] = i + 1 + + # Do a recursive call, so the frames are repositioned + self.draw_can_bus_message(self.ids[key]['msg'], sorting=True) + + # Pause by pressing space + elif key == KEY_SPACE: + self.paused = not self.paused + + # Scroll by pressing up/down + elif key == curses.KEY_UP: + # Limit scrolling, so the user do not scroll passed the header + if self.scroll > 0: + self.scroll -= 1 + self.redraw_screen() + elif key == curses.KEY_DOWN: + # Limit scrolling, so the maximum scrolling position is one below the last line + if self.scroll <= len(self.ids) - self.y + 1: + self.scroll += 1 + self.redraw_screen() + + # Check if screen was resized + resized = curses.is_term_resized(self.y, self.x) + if resized is True: + self.y, self.x = self.stdscr.getmaxyx() + if hasattr(curses, 'resizeterm'): # pragma: no cover + curses.resizeterm(self.y, self.x) + self.redraw_screen() + + # Shutdown the CAN-Bus interface + self.bus.shutdown() + + # Unpack the data and then convert it into SI-units + @staticmethod + def unpack_data(cmd, cmd_to_struct, data): # type: (int, Dict, bytes) -> List[Union[float, int]] + if not cmd_to_struct or len(data) == 0: + # These messages do not contain a data package + return [] + + for key in cmd_to_struct.keys(): + if cmd == key if isinstance(key, int) else cmd in key: + value = cmd_to_struct[key] + if isinstance(value, tuple): + # The struct is given as the fist argument + struct_t = value[0] # type: struct.Struct + + # The conversion from raw values to SI-units are given in the rest of the tuple + values = [d // val if isinstance(val, int) else float(d) / val + for d, val in zip(struct_t.unpack(data), value[1:])] + else: + # No conversion from SI-units is needed + struct_t = value # type: struct.Struct + values = list(struct_t.unpack(data)) + + return values + else: + raise ValueError('Unknown command: 0x{:02X}'.format(cmd)) + + def draw_can_bus_message(self, msg, sorting=False): + # Use the CAN-Bus ID as the key in the dict + key = msg.arbitration_id + + # Sort the extended IDs at the bottom by setting the 32-bit high + if msg.is_extended_id: + key |= (1 << 32) + + new_id_added, length_changed = False, False + if not sorting: + # Check if it is a new message or if the length is not the same + if key not in self.ids: + new_id_added = True + # Set the start time when the first message has been received + if not self.start_time: + self.start_time = msg.timestamp + elif msg.dlc != self.ids[key]['msg'].dlc: + length_changed = True + + if new_id_added or length_changed: + # Increment the index if it was just added, but keep it if the length just changed + row = len(self.ids) + 1 if new_id_added else self.ids[key]['row'] + + # It's a new message ID or the length has changed, so add it to the dict + # The first index is the row index, the second is the frame counter, + # the third is a copy of the CAN-Bus frame + # and the forth index is the time since the previous message + self.ids[key] = {'row': row, 'count': 0, 'msg': msg, 'dt': 0} + else: + # Calculate the time since the last message and save the timestamp + self.ids[key]['dt'] = msg.timestamp - self.ids[key]['msg'].timestamp + + # Copy the CAN-Bus frame - this is used for sorting + self.ids[key]['msg'] = msg + + # Increment frame counter + self.ids[key]['count'] += 1 + + # Format the CAN-Bus ID as a hex value + arbitration_id_string = '0x{0:0{1}X}'.format(msg.arbitration_id, 8 if msg.is_extended_id else 3) + + # Generate data string + data_string = '' + if msg.dlc > 0: + data_string = ' '.join('{:02X}'.format(x) for x in msg.data) + + # Use red for error frames + if msg.is_error_frame: + color = curses.color_pair(1) + else: + color = curses.color_pair(0) + + # Now draw the CAN-Bus message on the terminal window + self.draw_line(self.ids[key]['row'], 0, str(self.ids[key]['count']), color) + self.draw_line(self.ids[key]['row'], 8, '{0:.6f}'.format(self.ids[key]['msg'].timestamp - self.start_time), + color) + self.draw_line(self.ids[key]['row'], 23, '{0:.6f}'.format(self.ids[key]['dt']), color) + self.draw_line(self.ids[key]['row'], 35, arbitration_id_string, color) + self.draw_line(self.ids[key]['row'], 47, str(msg.dlc), color) + self.draw_line(self.ids[key]['row'], 52, data_string, color) + + if self.data_structs: + try: + values_list = [] + for x in self.unpack_data(msg.arbitration_id, self.data_structs, msg.data): + if isinstance(x, float): + values_list.append('{0:.6f}'.format(x)) + else: + values_list.append(str(x)) + values_string = ' '.join(values_list) + self.draw_line(self.ids[key]['row'], 77, values_string, color) + except (ValueError, struct.error): + pass + + return self.ids[key] + + def draw_line(self, row, col, txt, *args): + if row - self.scroll < 0: + # Skip if we have scrolled passed the line + return + try: + self.stdscr.addstr(row - self.scroll, col, txt, *args) + except curses.error: + # Ignore if we are trying to write outside the window + # This happens if the terminal window is too small + pass + + def draw_header(self): + self.stdscr.erase() + self.draw_line(0, 0, 'Count', curses.A_BOLD) + self.draw_line(0, 8, 'Time', curses.A_BOLD) + self.draw_line(0, 23, 'dt', curses.A_BOLD) + self.draw_line(0, 35, 'ID', curses.A_BOLD) + self.draw_line(0, 47, 'DLC', curses.A_BOLD) + self.draw_line(0, 52, 'Data', curses.A_BOLD) + if self.data_structs: # Only draw if the dictionary is not empty + self.draw_line(0, 77, 'Parsed values', curses.A_BOLD) + + def redraw_screen(self): + # Trigger a complete redraw + self.draw_header() + for key in self.ids.keys(): + self.draw_can_bus_message(self.ids[key]['msg']) + + +# noinspection PyProtectedMember +class SmartFormatter(argparse.HelpFormatter): + + def _get_default_metavar_for_optional(self, action): + return action.dest.upper() + + def _format_usage(self, usage, actions, groups, prefix): + # Use uppercase for "Usage:" text + return super(SmartFormatter, self)._format_usage(usage, actions, groups, 'Usage: ') + + def _format_args(self, action, default_metavar): + if action.nargs != argparse.REMAINDER and action.nargs != argparse.ONE_OR_MORE: + return super(SmartFormatter, self)._format_args(action, default_metavar) + + # Use the metavar if "REMAINDER" or "ONE_OR_MORE" is set + get_metavar = self._metavar_formatter(action, default_metavar) + return '%s' % get_metavar(1) + + def _format_action_invocation(self, action): + if not action.option_strings or action.nargs == 0: + return super(SmartFormatter, self)._format_action_invocation(action) + + # Modified so "-s ARGS, --long ARGS" is replaced with "-s, --long ARGS" + else: + parts = [] + default = self._get_default_metavar_for_optional(action) + args_string = self._format_args(action, default) + for i, option_string in enumerate(action.option_strings): + if i == len(action.option_strings) - 1: + parts.append('%s %s' % (option_string, args_string)) + else: + parts.append('%s' % option_string) + return ', '.join(parts) + + def _split_lines(self, text, width): + # Allow to manually split the lines + if text.startswith('R|'): + return text[2:].splitlines() + return super(SmartFormatter, self)._split_lines(text, width) + + def _fill_text(self, text, width, indent): + if text.startswith('R|'): + # noinspection PyTypeChecker + return ''.join(indent + line + '\n' for line in text[2:].splitlines()) + else: + return super(SmartFormatter, self)._fill_text(text, width, indent) + + +def parse_args(args): + # Python versions >= 3.5 + kwargs = {} + if sys.version_info[0] * 10 + sys.version_info[1] >= 35: # pragma: no cover + kwargs = {'allow_abbrev': False} + + # Parse command line arguments + parser = argparse.ArgumentParser('python -m can.viewer', + description='A simple CAN viewer terminal application written in Python', + epilog='R|Shortcuts: ' + '\n +---------+-------------------------+' + '\n | Key | Description |' + '\n +---------+-------------------------+' + '\n | ESQ/q | Exit the viewer |' + '\n | c | Clear the stored frames |' + '\n | s | Sort the stored frames |' + '\n | SPACE | Pause the viewer |' + '\n | UP/DOWN | Scroll the viewer |' + '\n +---------+-------------------------+', + formatter_class=SmartFormatter, add_help=False, **kwargs) + + optional = parser.add_argument_group('Optional arguments') + + optional.add_argument('-h', '--help', action='help', help='Show this help message and exit') + + optional.add_argument('--version', action='version', help="Show program's version number and exit", + version='%(prog)s (version {version})'.format(version=__version__)) + + # Copied from: https://github.com/hardbyte/python-can/blob/develop/can/logger.py + optional.add_argument('-b', '--bitrate', type=int, help='''Bitrate to use for the given CAN interface''') + + optional.add_argument('-c', '--channel', help='''Most backend interfaces require some sort of channel. + For example with the serial interface the channel might be a rfcomm device: "/dev/rfcomm0" + with the socketcan interfaces valid channel examples include: "can0", "vcan0". + (default: use default for the specified interface)''') + + optional.add_argument('-d', '--decode', dest='decode', + help='R|Specify how to convert the raw bytes into real values.' + '\nThe ID of the frame is given as the first argument and the format as the second.' + '\nThe Python struct package is used to unpack the received data' + '\nwhere the format characters have the following meaning:' + '\n < = little-endian, > = big-endian' + '\n x = pad byte' + '\n c = char' + '\n ? = bool' + '\n b = int8_t, B = uint8_t' + '\n h = int16, H = uint16' + '\n l = int32_t, L = uint32_t' + '\n q = int64_t, Q = uint64_t' + '\n f = float (32-bits), d = double (64-bits)' + '\nFx to convert six bytes with ID 0x100 into uint8_t, uint16 and uint32_t:' + '\n $ python -m can.viewer -d "100:: (matches when & mask == can_id & mask)' + '\n ~ (matches when & mask != can_id & mask)' + '\nFx to show only frames with ID 0x100 to 0x103:' + '\n python -m can.viewer -f 100:7FC' + '\nNote that the ID and mask are alway interpreted as hex values', + metavar='{:,~}', nargs=argparse.ONE_OR_MORE, default='') + + optional.add_argument('-i', '--interface', dest='interface', + help='R|Specify the backend CAN interface to use.', + choices=sorted(can.VALID_INTERFACES)) + + # Print help message when no arguments are given + if len(args) < 2: + parser.print_help(sys.stderr) + import errno + raise SystemExit(errno.EINVAL) + + parsed_args = parser.parse_args(args) + + can_filters = [] + if len(parsed_args.filter) > 0: + # print('Adding filter/s', parsed_args.filter) + for flt in parsed_args.filter: + # print(filter) + if ':' in flt: + _ = flt.split(':') + can_id, can_mask = int(_[0], base=16), int(_[1], base=16) + elif '~' in flt: + can_id, can_mask = flt.split('~') + can_id = int(can_id, base=16) | 0x20000000 # CAN_INV_FILTER + can_mask = int(can_mask, base=16) & 0x20000000 # socket.CAN_ERR_FLAG + else: + raise argparse.ArgumentError(None, 'Invalid filter argument') + can_filters.append({'can_id': can_id, 'can_mask': can_mask}) + + # Dictionary used to convert between Python values and C structs represented as Python strings. + # If the value is 'None' then the message does not contain any data package. + # + # The struct package is used to unpack the received data. + # Note the data is assumed to be in little-endian byte order. + # < = little-endian, > = big-endian + # x = pad byte + # c = char + # ? = bool + # b = int8_t, B = uint8_t + # h = int16, H = uint16 + # l = int32_t, L = uint32_t + # q = int64_t, Q = uint64_t + # f = float (32-bits), d = double (64-bits) + # + # An optional conversion from real units to integers can be given as additional arguments. + # In order to convert from raw integer value the real units are multiplied with the values and similarly the values + # are divided by the value in order to convert from real units to raw integer values. + data_structs = {} # type: Dict[Union[int, Tuple[int, ...]], Union[struct.Struct, Tuple, None]] + if len(parsed_args.decode) > 0: + if os.path.isfile(parsed_args.decode[0]): + with open(parsed_args.decode[0], 'r') as f: + structs = f.readlines() + else: + structs = parsed_args.decode + + for s in structs: + tmp = s.rstrip('\n').split(':') + + # The ID is given as a hex value, the format needs no conversion + key, fmt = int(tmp[0], base=16), tmp[1] + + # The scaling + scaling = [] # type: list + for t in tmp[2:]: + # First try to convert to int, if that fails, then convert to a float + try: + scaling.append(int(t)) + except ValueError: + scaling.append(float(t)) + + if scaling: + data_structs[key] = (struct.Struct(fmt),) + tuple(scaling) + else: + data_structs[key] = struct.Struct(fmt) + # print(data_structs[key]) + + return parsed_args, can_filters, data_structs + + +def main(): # pragma: no cover + parsed_args, can_filters, data_structs = parse_args(sys.argv[1:]) + + config = {'single_handle': True} + if can_filters: + config['can_filters'] = can_filters + if parsed_args.interface: + config['interface'] = parsed_args.interface + if parsed_args.bitrate: + config['bitrate'] = parsed_args.bitrate + + # Create a CAN-Bus interface + bus = can.Bus(parsed_args.channel, **config) + # print('Connected to {}: {}'.format(bus.__class__.__name__, bus.channel_info)) + + curses.wrapper(CanViewer, bus, data_structs) + + +if __name__ == '__main__': # pragma: no cover + # Catch ctrl+c + try: + main() + except KeyboardInterrupt: + pass diff --git a/doc/history.rst b/doc/history.rst index 70d0460c0..3ffc9bb3b 100644 --- a/doc/history.rst +++ b/doc/history.rst @@ -21,6 +21,7 @@ project in 2011. The socketcan interface was helped immensely by Phil Dixon who wrote a leaf-socketcan driver for Linux. The pcan interface was contributed by Albert Bloomfield in 2013. +Support for pcan on Mac was added by Kristian Sloth Lauszus in 2018. The usb2can interface was contributed by Joshua Villyard in 2015. @@ -34,13 +35,14 @@ a C++ library by Toby Lorenz. The slcan interface, ASCII listener and log logger and listener were contributed by Eduard Bröcker in 2017. -The NeoVi interface for ICS (Intrepid Control Systems) devices was contributed +The NeoVi interface for ICS (Intrepid Control Systems) devices was contributed by Pierre-Luc Tessier Gagné in 2017. Many improvements all over the library, cleanups, unifications as well as more comprehensive documentation and CI testing was contributed by Felix Divo in 2017 and 2018. +The CAN viewer terminal script was contributed by Kristian Sloth Lauszus in 2018. Support for CAN within Python ----------------------------- diff --git a/doc/images/viewer.png b/doc/images/viewer.png new file mode 100644 index 000000000..fb91701b2 Binary files /dev/null and b/doc/images/viewer.png differ diff --git a/doc/scripts.rst b/doc/scripts.rst index 76088eb7c..b58125ea5 100644 --- a/doc/scripts.rst +++ b/doc/scripts.rst @@ -5,7 +5,6 @@ The following modules are callable from python-can. They can be called for example by ``python -m can.logger`` or ``can_logger.py`` (if installed using pip). - can.logger ---------- @@ -84,3 +83,94 @@ Command line help, called with ``--help``:: minimum gap between frames) -g GAP, --gap GAP minimum time between replayed frames -s SKIP, --skip SKIP skip gaps greater than 's' seconds + +can.viewer +---------- + +A screenshot of the application can be seen below: + +.. image:: ../images/viewer.png + :width: 100% + +The first column is the number of times a frame with the particular ID that has been received, next is the timestamp of the frame relative to the first received message. The third column is the time between the current frame relative to the previous one. Next is the length of the frame, the data and then the decoded data converted according to the ``-d`` argument. The top red row indicates an error frame. + +Command line arguments +^^^^^^^^^^^^^^^^^^^^^^ + +By default it will be using the :doc:`/interfaces/socketcan` interface. All interfaces supported are supported and can be specified using the ``-i`` argument. + +The full usage page can be seen below:: + + Usage: python -m can.viewer [-h] [--version] [-b BITRATE] [-c CHANNEL] + [-d {:,:::...:,file.txt}] + [-f {:,~}] + [-i {iscan,ixxat,kvaser,neovi,nican,pcan,serial,slcan,socketcan,socketcan_ctypes,socketcan_native,usb2can,vector,virtual}] + + A simple CAN viewer terminal application written in Python + + Optional arguments: + -h, --help Show this help message and exit + --version Show program's version number and exit + -b, --bitrate BITRATE + Bitrate to use for the given CAN interface + -c, --channel CHANNEL + Most backend interfaces require some sort of channel. + For example with the serial interface the channel + might be a rfcomm device: "/dev/rfcomm0" with the + socketcan interfaces valid channel examples include: + "can0", "vcan0". (default: use default for the + specified interface) + -d, --decode {:,:::...:,file.txt} + Specify how to convert the raw bytes into real values. + The ID of the frame is given as the first argument and the format as the second. + The Python struct package is used to unpack the received data + where the format characters have the following meaning: + < = little-endian, > = big-endian + x = pad byte + c = char + ? = bool + b = int8_t, B = uint8_t + h = int16, H = uint16 + l = int32_t, L = uint32_t + q = int64_t, Q = uint64_t + f = float (32-bits), d = double (64-bits) + Fx to convert six bytes with ID 0x100 into uint8_t, uint16 and uint32_t: + $ python -m can.viewer -d "100::,~} + Comma separated CAN filters for the given CAN interface: + : (matches when & mask == can_id & mask) + ~ (matches when & mask != can_id & mask) + Fx to show only frames with ID 0x100 to 0x103: + python -m can.viewer -f 100:7FC + Note that the ID and mask are alway interpreted as hex values + -i, --interface {iscan,ixxat,kvaser,neovi,nican,pcan,serial,slcan,socketcan,socketcan_ctypes,socketcan_native,usb2can,vector,virtual} + Specify the backend CAN interface to use. + + Shortcuts: + +---------+-------------------------+ + | Key | Description | + +---------+-------------------------+ + | ESQ/q | Exit the viewer | + | c | Clear the stored frames | + | s | Sort the stored frames | + | SPACE | Pause the viewer | + | UP/DOWN | Scroll the viewer | + +---------+-------------------------+ diff --git a/scripts/can_viewer.py b/scripts/can_viewer.py new file mode 100644 index 000000000..3c9ba738c --- /dev/null +++ b/scripts/can_viewer.py @@ -0,0 +1,14 @@ +#!/usr/bin/env python +# coding: utf-8 + +""" +See :mod:`can.viewer`. +""" + +from __future__ import absolute_import + +from can.viewer import main + + +if __name__ == "__main__": + main() diff --git a/setup.py b/setup.py index a7f466941..4aed26e23 100644 --- a/setup.py +++ b/setup.py @@ -35,7 +35,9 @@ 'pytest ~= 3.6', 'pytest-timeout ~= 1.2', 'pytest-cov ~= 2.5', - 'codecov ~= 2.0' + 'codecov ~= 2.0', + 'future', + 'six' ] + extras_require['serial'] extras_require['test'] = tests_require @@ -98,7 +100,7 @@ # see https://www.python.org/dev/peps/pep-0345/#version-specifiers python_requires=">=2.7,!=3.0,!=3.1,!=3.2,!=3.3", install_requires=[ - 'wrapt ~= 1.10', + 'wrapt ~= 1.10', 'typing', 'windows-curses;platform_system=="Windows"', ], extras_require=extras_require, diff --git a/test/test_viewer.py b/test/test_viewer.py new file mode 100644 index 000000000..d978f12ed --- /dev/null +++ b/test/test_viewer.py @@ -0,0 +1,445 @@ +#!/usr/bin/python +# coding: utf-8 +# +# Copyright (C) 2018 Kristian Sloth Lauszus. All rights reserved. +# +# Contact information +# ------------------- +# Kristian Sloth Lauszus +# Web : http://www.lauszus.com +# e-mail : lauszus@gmail.com + +from __future__ import absolute_import + +import argparse +import can +import curses +import math +import pytest +import random +import struct +import time +import unittest +import os +import six + +from typing import Dict, Tuple, Union + +try: + # noinspection PyCompatibility + from unittest.mock import Mock, patch +except ImportError: + # noinspection PyPackageRequirements + from mock import Mock, patch + +from can.viewer import KEY_ESC, KEY_SPACE, CanViewer, parse_args + + +# noinspection SpellCheckingInspection,PyUnusedLocal +class StdscrDummy: + + def __init__(self): + self.key_counter = 0 + + @staticmethod + def clear(): + pass + + @staticmethod + def erase(): + pass + + @staticmethod + def getmaxyx(): + # Set y-value, so scrolling gets tested + return 1, 1 + + @staticmethod + def addstr(row, col, txt, *args): + assert row >= 0 + assert col >= 0 + assert txt is not None + # Raise an exception 50 % of the time, so we can make sure the code handles it + if random.random() < .5: + raise curses.error + + @staticmethod + def nodelay(_bool): + pass + + def getch(self): + self.key_counter += 1 + if self.key_counter == 1: + # Send invalid key + return -1 + elif self.key_counter == 2: + return ord('c') # Clear + elif self.key_counter == 3: + return KEY_SPACE # Pause + elif self.key_counter == 4: + return KEY_SPACE # Unpause + elif self.key_counter == 5: + return ord('s') # Sort + + # Keep scrolling until it exceeds the number of messages + elif self.key_counter <= 100: + return curses.KEY_DOWN + # Scroll until the header is back as the first line and then scroll over the limit + elif self.key_counter <= 200: + return curses.KEY_UP + + return KEY_ESC + + +class CanViewerTest(unittest.TestCase): + + @classmethod + def setUpClass(cls): + # Set seed, so the tests are not affected + random.seed(0) + + def setUp(self): + stdscr = StdscrDummy() + config = {'interface': 'virtual', 'receive_own_messages': True} + bus = can.Bus(**config) + data_structs = None + + patch_curs_set = patch('curses.curs_set') + patch_curs_set.start() + self.addCleanup(patch_curs_set.stop) + + patch_use_default_colors = patch('curses.use_default_colors') + patch_use_default_colors.start() + self.addCleanup(patch_use_default_colors.stop) + + patch_init_pair = patch('curses.init_pair') + patch_init_pair.start() + self.addCleanup(patch_init_pair.stop) + + patch_color_pair = patch('curses.color_pair') + patch_color_pair.start() + self.addCleanup(patch_color_pair.stop) + + patch_is_term_resized = patch('curses.is_term_resized') + mock_is_term_resized = patch_is_term_resized.start() + mock_is_term_resized.return_value = True if random.random() < .5 else False + self.addCleanup(patch_is_term_resized.stop) + + if hasattr(curses, 'resizeterm'): + patch_resizeterm = patch('curses.resizeterm') + patch_resizeterm.start() + self.addCleanup(patch_resizeterm.stop) + + self.can_viewer = CanViewer(stdscr, bus, data_structs, testing=True) + + def tearDown(self): + # Run the viewer after the test, this is done, so we can receive the CAN-Bus messages and make sure that they + # are parsed correctly + self.can_viewer.run() + + def test_send(self): + # CANopen EMCY + data = [1, 2, 3, 4, 5, 6, 7] # Wrong length + msg = can.Message(arbitration_id=0x080 + 1, data=data, extended_id=False) + self.can_viewer.bus.send(msg) + + data = [1, 2, 3, 4, 5, 6, 7, 8] + msg = can.Message(arbitration_id=0x080 + 1, data=data, extended_id=False) + self.can_viewer.bus.send(msg) + + # CANopen HEARTBEAT + data = [0x05] # Operational + msg = can.Message(arbitration_id=0x700 + 0x7F, data=data, extended_id=False) + self.can_viewer.bus.send(msg) + + # Send non-CANopen message + data = [1, 2, 3, 4, 5, 6, 7, 8] + msg = can.Message(arbitration_id=0x101, data=data, extended_id=False) + self.can_viewer.bus.send(msg) + + # Send the same command, but with another data length + data = [1, 2, 3, 4, 5, 6] + msg = can.Message(arbitration_id=0x101, data=data, extended_id=False) + self.can_viewer.bus.send(msg) + + # Message with extended id + data = [1, 2, 3, 4, 5, 6, 7, 8] + msg = can.Message(arbitration_id=0x123456, data=data, extended_id=True) + self.can_viewer.bus.send(msg) + # self.assertTupleEqual(self.can_viewer.parse_canopen_message(msg), (None, None)) + + # Send the same message again to make sure that resending works and dt is correct + time.sleep(.1) + self.can_viewer.bus.send(msg) + + # Send error message + msg = can.Message(is_error_frame=True) + self.can_viewer.bus.send(msg) + + def test_receive(self): + # Send the messages again, but this time the test code will receive it + self.test_send() + + data_structs = { + # For converting the EMCY and HEARTBEAT messages + 0x080 + 0x01: struct.Struct('ff'), + } + # Receive the messages we just sent in 'test_canopen' + while 1: + msg = self.can_viewer.bus.recv(timeout=0) + if msg is not None: + self.can_viewer.data_structs = data_structs if msg.arbitration_id != 0x101 else None + _id = self.can_viewer.draw_can_bus_message(msg) + if _id['msg'].arbitration_id == 0x101: + # Check if the counter is reset when the length has changed + self.assertEqual(_id['count'], 1) + elif _id['msg'].arbitration_id == 0x123456: + # Check if the counter is incremented + if _id['dt'] == 0: + self.assertEqual(_id['count'], 1) + else: + self.assertTrue(pytest.approx(_id['dt'], 0.1)) # dt should be ~0.1 s + self.assertEqual(_id['count'], 2) + else: + # Make sure dt is 0 + if _id['count'] == 1: + self.assertEqual(_id['dt'], 0) + else: + break + + # Convert it into raw integer values and then pack the data + @staticmethod + def pack_data(cmd, cmd_to_struct, *args): # type: (int, Dict, Union[*float, *int]) -> bytes + if not cmd_to_struct or len(args) == 0: + # If no arguments are given, then the message does not contain a data package + return b'' + + for key in cmd_to_struct.keys(): + if cmd == key if isinstance(key, int) else cmd in key: + value = cmd_to_struct[key] + if isinstance(value, tuple): + # The struct is given as the fist argument + struct_t = value[0] # type: struct.Struct + + # The conversion from SI-units to raw values are given in the rest of the tuple + fmt = struct_t.format + if isinstance(fmt, six.string_types): # pragma: no cover + # Needed for Python 3.7 + fmt = six.b(fmt) + + # Make sure the endian is given as the first argument + assert six.byte2int(fmt) == ord('<') or six.byte2int(fmt) == ord('>') + + # Disable rounding if the format is a float + data = [] + for c, arg, val in zip(six.iterbytes(fmt[1:]), args, value[1:]): + if c == ord('f'): + data.append(arg * val) + else: + data.append(round(arg * val)) + else: + # No conversion from SI-units is needed + struct_t = value # type: struct.Struct + data = args + + return struct_t.pack(*data) + else: + raise ValueError('Unknown command: 0x{:02X}'.format(cmd)) + + def test_pack_unpack(self): + CANOPEN_TPDO1 = 0x180 + CANOPEN_TPDO2 = 0x280 + CANOPEN_TPDO3 = 0x380 + CANOPEN_TPDO4 = 0x480 + + # Dictionary used to convert between Python values and C structs represented as Python strings. + # If the value is 'None' then the message does not contain any data package. + # + # The struct package is used to unpack the received data. + # Note the data is assumed to be in little-endian byte order. + # < = little-endian, > = big-endian + # x = pad byte + # c = char + # ? = bool + # b = int8_t, B = uint8_t + # h = int16, H = uint16 + # l = int32_t, L = uint32_t + # q = int64_t, Q = uint64_t + # f = float (32-bits), d = double (64-bits) + # + # An optional conversion from real units to integers can be given as additional arguments. + # In order to convert from raw integer value the SI-units are multiplied with the values and similarly the values + # are divided by the value in order to convert from real units to raw integer values. + data_structs = { + # CANopen node 1 + CANOPEN_TPDO1 + 1: struct.Struct('lL'), + (CANOPEN_TPDO3 + 2, CANOPEN_TPDO4 + 2): struct.Struct('>LL'), + } # type: Dict[Union[int, Tuple[int, ...]], Union[struct.Struct, Tuple, None]] + + raw_data = self.pack_data(CANOPEN_TPDO1 + 1, data_structs, -7, 13, -1024, 2048, 0xFFFF) + parsed_data = CanViewer.unpack_data(CANOPEN_TPDO1 + 1, data_structs, raw_data) + self.assertListEqual(parsed_data, [-7, 13, -1024, 2048, 0xFFFF]) + self.assertTrue(all(isinstance(d, int) for d in parsed_data)) + + raw_data = self.pack_data(CANOPEN_TPDO2 + 1, data_structs, 12.34, 4.5, 6) + parsed_data = CanViewer.unpack_data(CANOPEN_TPDO2 + 1, data_structs, raw_data) + self.assertTrue(pytest.approx(parsed_data, [12.34, 4.5, 6])) + self.assertTrue(isinstance(parsed_data[0], float) and isinstance(parsed_data[1], float) and + isinstance(parsed_data[2], int)) + + raw_data = self.pack_data(CANOPEN_TPDO3 + 1, data_structs, 123.45, 67.89) + parsed_data = CanViewer.unpack_data(CANOPEN_TPDO3 + 1, data_structs, raw_data) + self.assertTrue(pytest.approx(parsed_data, [123.45, 67.89])) + self.assertTrue(all(isinstance(d, float) for d in parsed_data)) + + raw_data = self.pack_data(CANOPEN_TPDO4 + 1, data_structs, math.pi / 2., math.pi) + parsed_data = CanViewer.unpack_data(CANOPEN_TPDO4 + 1, data_structs, raw_data) + self.assertTrue(pytest.approx(parsed_data, [math.pi / 2., math.pi])) + self.assertTrue(all(isinstance(d, float) for d in parsed_data)) + + raw_data = self.pack_data(CANOPEN_TPDO1 + 2, data_structs) + parsed_data = CanViewer.unpack_data(CANOPEN_TPDO1 + 2, data_structs, raw_data) + self.assertListEqual(parsed_data, []) + self.assertIsInstance(parsed_data, list) + + raw_data = self.pack_data(CANOPEN_TPDO2 + 2, data_structs, -2147483648, 0xFFFFFFFF) + parsed_data = CanViewer.unpack_data(CANOPEN_TPDO2 + 2, data_structs, raw_data) + self.assertListEqual(parsed_data, [-2147483648, 0xFFFFFFFF]) + + raw_data = self.pack_data(CANOPEN_TPDO3 + 2, data_structs, 0xFF, 0xFFFF) + parsed_data = CanViewer.unpack_data(CANOPEN_TPDO3 + 2, data_structs, raw_data) + self.assertListEqual(parsed_data, [0xFF, 0xFFFF]) + + raw_data = self.pack_data(CANOPEN_TPDO4 + 2, data_structs, 0xFFFFFF, 0xFFFFFFFF) + parsed_data = CanViewer.unpack_data(CANOPEN_TPDO4 + 2, data_structs, raw_data) + self.assertListEqual(parsed_data, [0xFFFFFF, 0xFFFFFFFF]) + + # See: http://python-future.org/compatible_idioms.html#long-integers + from past.builtins import long + self.assertTrue(all(isinstance(d, (int, long)) for d in parsed_data)) + + # Make sure that the ValueError exception is raised + with self.assertRaises(ValueError): + self.pack_data(0x101, data_structs, 1, 2, 3, 4) + + with self.assertRaises(ValueError): + CanViewer.unpack_data(0x102, data_structs, b'\x01\x02\x03\x04\x05\x06\x07\x08') + + def test_parse_args(self): + parsed_args, _, _ = parse_args(['-b', '250000']) + self.assertEqual(parsed_args.bitrate, 250000) + + parsed_args, _, _ = parse_args(['--bitrate', '500000']) + self.assertEqual(parsed_args.bitrate, 500000) + + parsed_args, _, _ = parse_args(['-c', 'can0']) + self.assertEqual(parsed_args.channel, 'can0') + + parsed_args, _, _ = parse_args(['--channel', 'PCAN_USBBUS1']) + self.assertEqual(parsed_args.channel, 'PCAN_USBBUS1') + + parsed_args, _, data_structs = parse_args(['-d', '100: