15474433755

Committed 05 Jun 2025 06:21PM UTC coverage: 70.972% (+0.07%) from 70.901%

Build # 15474433755

Build Type

Pull #1949

github

Committed by

web-flow

Commit Message

Merge e6599c908 into f43bedbc6

Pull Request Pull Request #1949: Add public functions for creating bus command lines options

Run Details

137 of 152 new or added lines in 4 files covered. (90.13%)

1 existing line in 1 file now uncovered.

7743 of 10910 relevant lines covered (70.97%)

13.54 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

96.36

/can/viewer.py

# Copyright (C) 2018 Kristian Sloth Lauszus.
#
# This program is free software; you can redistribute it and/or
# modify it under the terms of the GNU Lesser General Public
# License as published by the Free Software Foundation; either
# version 3 of the License, or (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
# Lesser General Public License for more details.
#
# You should have received a copy of the GNU Lesser General Public License
# along with this program; if not, write to the Free Software Foundation,
# Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301, USA.
#
# Contact information
# -------------------
# Kristian Sloth Lauszus
# Web      :  http://www.lauszus.com
# e-mail   :  lauszus@gmail.com

import argparse
import errno
import logging
import os
import struct
import sys
import time

from can import __version__
from can.cli import (
    _set_logging_level_from_namespace,
    add_bus_arguments,
    create_bus_from_namespace,
)
from can.typechecking import TDataStructs

logger = logging.getLogger("can.viewer")

try:
    import curses
    from curses.ascii import ESC as KEY_ESC  # type: ignore[attr-defined,unused-ignore]
    from curses.ascii import SP as KEY_SPACE  # type: ignore[attr-defined,unused-ignore]
except ImportError:
    # Probably on Windows while windows-curses is not installed (e.g. in PyPy)
    logger.warning(
        "You won't be able to use the viewer program without curses installed!"
    )
    curses = None  # type: ignore[assignment]


class CanViewer:  # pylint: disable=too-many-instance-attributes
    def __init__(self, stdscr, bus, data_structs, testing=False):
        self.stdscr = stdscr
        self.bus = bus
        self.data_structs = data_structs

        # Initialise the ID dictionary, Previous values dict, start timestamp,
        # scroll and variables for pausing the viewer and enabling byte highlighting
        self.ids = {}
        self.start_time = None
        self.scroll = 0
        self.paused = False
        self.highlight_changed_bytes = False
        self.previous_values = {}

        # 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)
        # Used to color changed bytes
        curses.init_pair(2, curses.COLOR_CYAN, curses.COLOR_BLUE)

        if not testing:  # pragma: no cover
            self.run()

    def run(self):
        # Clear the terminal and draw the header
        self.draw_header()

        while True:
            # 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.0 / 1000.0)
                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.0 / 1000.0)

            # 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'
            if key == ord("c"):
                self.ids = {}
                self.start_time = None
                self.scroll = 0
                self.draw_header()

            # Toggle byte change highlighting pressing 'h'
            elif key == ord("h"):
                self.highlight_changed_bytes = not self.highlight_changed_bytes
                if not self.highlight_changed_bytes:
                    # empty the previous values dict when leaving higlighting mode
                    self.previous_values.clear()
                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: int, cmd_to_struct: dict, data: bytes) -> list[float]:
        if not cmd_to_struct or not data:
            # These messages do not contain a data package
            return []

        for key, value in cmd_to_struct.items():
            if cmd == key if isinstance(key, int) else cmd in key:
                if isinstance(value, tuple):
                    # The struct is given as the fist argument
                    struct_t: struct.Struct = value[0]

                    # 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
                    as_struct_t: struct.Struct = value
                    values = list(as_struct_t.unpack(data))

                return values

        raise ValueError(f"Unknown command: 0x{cmd:02X}")

    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

                # Clear the old data bytes when the length of the new message is shorter
                if msg.dlc < self.ids[key]["msg"].dlc:
                    self.draw_line(
                        self.ids[key]["row"],
                        # Start drawing at the last byte that is not in the new message
                        52 + msg.dlc * 3,
                        # Draw spaces where the old bytes were drawn
                        " " * ((self.ids[key]["msg"].dlc - msg.dlc) * 3 - 1),
                    )

            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(  # pylint: disable=consider-using-f-string
                msg.arbitration_id,
                8 if msg.is_extended_id else 3,
            )
        )

        # 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,
            f"{self.ids[key]['msg'].timestamp - self.start_time:.6f}",
            color,
        )
        self.draw_line(self.ids[key]["row"], 23, f"{self.ids[key]['dt']:.6f}", 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)

        try:
            previous_byte_values = self.previous_values[key]
        except KeyError:  # no row of previous values exists for the current message ID
            # initialise a row to store the values for comparison next time
            self.previous_values[key] = {}
            previous_byte_values = self.previous_values[key]
        for i, b in enumerate(msg.data):
            col = 52 + i * 3
            if col > self.x - 2:
                # Data does not fit
                self.draw_line(self.ids[key]["row"], col - 4, "...", color)
                break
            if self.highlight_changed_bytes:
                try:
                    if b != previous_byte_values[i]:
                        # set colour to highlight a changed value
                        data_color = curses.color_pair(2)
                    else:
                        data_color = color
                except KeyError:
                    # previous entry for byte didn't exist - default to rest of line colour
                    data_color = color
                finally:
                    # write the new value to the previous values dict for next time
                    previous_byte_values[i] = b
            else:
                data_color = color
            text = f"{b:02X}"
            self.draw_line(self.ids[key]["row"], col, text, data_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(f"{x:.6f}")
                    else:
                        values_list.append(str(x))
                values_string = " ".join(values_list)
                self.ids[key]["values_string_length"] = len(values_string)
                values_string += " " * (self.x - len(values_string))

                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 past 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)

        # Indicate that byte change highlighting is enabled
        if self.highlight_changed_bytes:
            self.draw_line(0, 57, "(changed)", curses.color_pair(2))
        # Only draw if the dictionary is not empty
        if self.data_structs:
            self.draw_line(0, 77, "Parsed values", curses.A_BOLD)

    def redraw_screen(self):
        # Trigger a complete redraw
        self.draw_header()
        for ids in self.ids.values():
            self.draw_can_bus_message(ids["msg"])


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()._format_usage(usage, actions, groups, "Usage: ")

    def _format_args(self, action, default_metavar):
        if action.nargs not in (argparse.REMAINDER, argparse.ONE_OR_MORE):
            return super()._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 str(get_metavar(1))

    def _format_action_invocation(self, action):
        if not action.option_strings or action.nargs == 0:
            return super()._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(f"{option_string} {args_string}")
                else:
                    parts.append(str(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()._split_lines(text, width)

    def _fill_text(self, text, width, indent):
        if text.startswith("R|"):
            return "".join(indent + line + "\n" for line in text[2:].splitlines())
        else:
            return super()._fill_text(text, width, indent)


def _parse_viewer_args(
    args: list[str],
) -> tuple[argparse.Namespace, TDataStructs]:
    # 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        | h       | Toggle highlight byte changes |"
        "\n        | SPACE   | Pause the viewer              |"
        "\n        | UP/DOWN | Scroll the viewer             |"
        "\n        +---------+-------------------------------+",
        formatter_class=SmartFormatter,
        add_help=False,
        allow_abbrev=False,
    )

    # add bus options group
    add_bus_arguments(parser, filter_arg=True, group_title="Bus arguments")

    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=f"%(prog)s (version {__version__})",
    )

    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:<BHL"'
        "\nNote that the IDs are always interpreted as hex values."
        "\nAn optional conversion from integers to real units can be given"
        "\nas additional arguments. In order to convert from raw integer"
        "\nvalues the values are divided with the corresponding scaling value,"
        "\nsimilarly the values are multiplied by the scaling value in order"
        "\nto convert from real units to raw integer values."
        "\nFx lets say the uint8_t needs no conversion, but the uint16 and the uint32_t"
        "\nneeds to be divided by 10 and 100 respectively:"
        '\n  $ python -m can.viewer -d "101:<BHL:1:10.0:100.0"'
        "\nBe aware that integer division is performed if the scaling value is an integer."
        "\nMultiple arguments are separated by spaces:"
        '\n  $ python -m can.viewer -d "100:<BHL" "101:<BHL:1:10.0:100.0"'
        "\nAlternatively a file containing the conversion strings separated by new lines"
        "\ncan be given as input:"
        "\n  $ cat file.txt"
        "\n      100:<BHL"
        "\n      101:<BHL:1:10.0:100.0"
        "\n  $ python -m can.viewer -d file.txt",
        metavar="{<id>:<format>,<id>:<format>:<scaling1>:...:<scalingN>,file.txt}",
        nargs=argparse.ONE_OR_MORE,
        default="",
    )

    optional.add_argument(
        "-v",
        action="count",
        dest="verbosity",
        help="""How much information do you want to see at the command line?
                        You can add several of these e.g., -vv is DEBUG""",
        default=2,
    )

    # Print help message when no arguments are given
    if not args:
        parser.print_help(sys.stderr)
        raise SystemExit(errno.EINVAL)

    parsed_args, unknown_args = parser.parse_known_args(args)
    if unknown_args:
        print("Unknown arguments:", unknown_args)

    # 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: TDataStructs = {}
    if parsed_args.decode:
        if os.path.isfile(parsed_args.decode[0]):
            with open(parsed_args.decode[0], encoding="utf-8") 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: list[float] = []
            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), *scaling)
            else:
                data_structs[key] = struct.Struct(fmt)

    return parsed_args, data_structs


def main() -> None:
    parsed_args, data_structs = _parse_viewer_args(sys.argv[1:])
    bus = create_bus_from_namespace(parsed_args)
    _set_logging_level_from_namespace(parsed_args)
    curses.wrapper(CanViewer, bus, data_structs)  # type: ignore[attr-defined,unused-ignore]


if __name__ == "__main__":
    # Catch ctrl+c
    try:
        main()
    except KeyboardInterrupt:
        pass

1	# Copyright (C) 2018 Kristian Sloth Lauszus.
2	#
3	# This program is free software; you can redistribute it and/or
4	# modify it under the terms of the GNU Lesser General Public
5	# License as published by the Free Software Foundation; either
6	# version 3 of the License, or (at your option) any later version.
7	#
8	# This program is distributed in the hope that it will be useful,
9	# but WITHOUT ANY WARRANTY; without even the implied warranty of
10	# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
11	# Lesser General Public License for more details.
12	#
13	# You should have received a copy of the GNU Lesser General Public License
14	# along with this program; if not, write to the Free Software Foundation,
15	# Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.
16	#
17	# Contact information
18	# -------------------
19	# Kristian Sloth Lauszus
20	# Web : http://www.lauszus.com
21	# e-mail : lauszus@gmail.com
22
23	import argparse	21✔
24	import errno	21✔
25	import logging	21✔
26	import os	21✔
27	import struct	21✔
28	import sys	21✔
29	import time	21✔
30
31	from can import __version__	21✔
32	from can.cli import (	21✔
33	_set_logging_level_from_namespace,
34	add_bus_arguments,
35	create_bus_from_namespace,
36	)
37	from can.typechecking import TDataStructs	21✔
38
39	logger = logging.getLogger("can.viewer")	21✔
40
41	try:	21✔
42	import curses	21✔
43	from curses.ascii import ESC as KEY_ESC # type: ignore[attr-defined,unused-ignore]	14✔
44	from curses.ascii import SP as KEY_SPACE # type: ignore[attr-defined,unused-ignore]	14✔
45	except ImportError:	7✔
46	# Probably on Windows while windows-curses is not installed (e.g. in PyPy)
47	logger.warning(	7✔
48	"You won't be able to use the viewer program without curses installed!"
49	)
50	curses = None # type: ignore[assignment]	7✔
51
52
53	class CanViewer: # pylint: disable=too-many-instance-attributes	21✔
54	def __init__(self, stdscr, bus, data_structs, testing=False):	21✔
55	self.stdscr = stdscr	14✔
56	self.bus = bus	14✔
57	self.data_structs = data_structs	14✔
58
59	# Initialise the ID dictionary, Previous values dict, start timestamp,
60	# scroll and variables for pausing the viewer and enabling byte highlighting
61	self.ids = {}	14✔
62	self.start_time = None	14✔
63	self.scroll = 0	14✔
64	self.paused = False	14✔
65	self.highlight_changed_bytes = False	14✔
66	self.previous_values = {}	14✔
67
68	# Get the window dimensions - used for resizing the window
69	self.y, self.x = self.stdscr.getmaxyx()	14✔
70
71	# Do not wait for key inputs, disable the cursor and choose the background color
72	# automatically
73	self.stdscr.nodelay(True)	14✔
74	curses.curs_set(0)	14✔
75	curses.use_default_colors()	14✔
76
77	# Used to color error frames red
78	curses.init_pair(1, curses.COLOR_RED, -1)	14✔
79	# Used to color changed bytes
80	curses.init_pair(2, curses.COLOR_CYAN, curses.COLOR_BLUE)	14✔
81
82	if not testing: # pragma: no cover	14✔
83	self.run()
84
85	def run(self):	21✔
86	# Clear the terminal and draw the header
87	self.draw_header()	14✔
88
89	while True:	10✔
90	# Do not read the CAN-Bus when in paused mode
91	if not self.paused:	14✔
92	# Read the CAN-Bus and draw it in the terminal window
93	msg = self.bus.recv(timeout=1.0 / 1000.0)	14✔
94	if msg is not None:	14✔
95	self.draw_can_bus_message(msg)	14✔
96	else:
97	# Sleep 1 ms, so the application does not use 100 % of the CPU resources
98	time.sleep(1.0 / 1000.0)	14✔
99
100	# Read the terminal input
101	key = self.stdscr.getch()	14✔
102
103	# Stop program if the user presses ESC or 'q'
104	if key == KEY_ESC or key == ord("q"):	14✔
105	break	12✔
106
107	# Clear by pressing 'c'
108	if key == ord("c"):	14✔
109	self.ids = {}	14✔
110	self.start_time = None	14✔
111	self.scroll = 0	14✔
112	self.draw_header()	14✔
113
114	# Toggle byte change highlighting pressing 'h'
115	elif key == ord("h"):	14✔
116	self.highlight_changed_bytes = not self.highlight_changed_bytes	14✔
117	if not self.highlight_changed_bytes:	14✔
118	# empty the previous values dict when leaving higlighting mode
119	self.previous_values.clear()	14✔
120	self.draw_header()	14✔
121
122	# Sort by pressing 's'
123	elif key == ord("s"):	14✔
124	# Sort frames based on the CAN-Bus ID
125	self.draw_header()	14✔
126	for i, key in enumerate(sorted(self.ids.keys())):	14✔
127	# Set the new row index, but skip the header
128	self.ids[key]["row"] = i + 1	14✔
129
130	# Do a recursive call, so the frames are repositioned
131	self.draw_can_bus_message(self.ids[key]["msg"], sorting=True)	14✔
132
133	# Pause by pressing space
134	elif key == KEY_SPACE:	14✔
135	self.paused = not self.paused	14✔
136
137	# Scroll by pressing up/down
138	elif key == curses.KEY_UP:	14✔
139	# Limit scrolling, so the user do not scroll passed the header
140	if self.scroll > 0:	14✔
141	self.scroll -= 1	14✔
142	self.redraw_screen()	14✔
143	elif key == curses.KEY_DOWN:	14✔
144	# Limit scrolling, so the maximum scrolling position is one below the last line
145	if self.scroll <= len(self.ids) - self.y + 1:	14✔
146	self.scroll += 1	14✔
147	self.redraw_screen()	14✔
148
149	# Check if screen was resized
150	resized = curses.is_term_resized(self.y, self.x)	14✔
151	if resized is True:	14✔
152	self.y, self.x = self.stdscr.getmaxyx()	14✔
153	if hasattr(curses, "resizeterm"): # pragma: no cover	14✔
154	curses.resizeterm(self.y, self.x)	14✔
155	self.redraw_screen()	14✔
156
157	# Shutdown the CAN-Bus interface
158	self.bus.shutdown()	14✔
159
160	# Unpack the data and then convert it into SI-units
161	@staticmethod	21✔
162	def unpack_data(cmd: int, cmd_to_struct: dict, data: bytes) -> list[float]:	21✔
163	if not cmd_to_struct or not data:	14✔
164	# These messages do not contain a data package
165	return []	14✔
166
167	for key, value in cmd_to_struct.items():	14✔
168	if cmd == key if isinstance(key, int) else cmd in key:	14✔
169	if isinstance(value, tuple):	14✔
170	# The struct is given as the fist argument
171	struct_t: struct.Struct = value[0]	14✔
172
173	# The conversion from raw values to SI-units are given in the rest of the tuple
174	values = [	14✔
175	d // val if isinstance(val, int) else float(d) / val
176	for d, val in zip(struct_t.unpack(data), value[1:])
177	]
178	else:
179	# No conversion from SI-units is needed
180	as_struct_t: struct.Struct = value	14✔
181	values = list(as_struct_t.unpack(data))	14✔
182
183	return values	14✔
184
185	raise ValueError(f"Unknown command: 0x{cmd:02X}")	14✔
186
187	def draw_can_bus_message(self, msg, sorting=False):	21✔
188	# Use the CAN-Bus ID as the key in the dict
189	key = msg.arbitration_id	14✔
190
191	# Sort the extended IDs at the bottom by setting the 32-bit high
192	if msg.is_extended_id:	14✔
193	key \|= 1 << 32	14✔
194
195	new_id_added, length_changed = False, False	14✔
196	if not sorting:	14✔
197	# Check if it is a new message or if the length is not the same
198	if key not in self.ids:	14✔
199	new_id_added = True	14✔
200	# Set the start time when the first message has been received
201	if not self.start_time:	14✔
202	self.start_time = msg.timestamp	14✔
203	elif msg.dlc != self.ids[key]["msg"].dlc:	14✔
204	length_changed = True	14✔
205
206	# Clear the old data bytes when the length of the new message is shorter
207	if msg.dlc < self.ids[key]["msg"].dlc:	14✔
208	self.draw_line(	14✔
209	self.ids[key]["row"],
210	# Start drawing at the last byte that is not in the new message
211	52 + msg.dlc * 3,
212	# Draw spaces where the old bytes were drawn
213	" " * ((self.ids[key]["msg"].dlc - msg.dlc) * 3 - 1),
214	)
215
216	if new_id_added or length_changed:	14✔
217	# Increment the index if it was just added, but keep it if the length just changed
218	row = len(self.ids) + 1 if new_id_added else self.ids[key]["row"]	14✔
219
220	# It's a new message ID or the length has changed, so add it to the dict
221	# The first index is the row index, the second is the frame counter,
222	# the third is a copy of the CAN-Bus frame
223	# and the forth index is the time since the previous message
224	self.ids[key] = {"row": row, "count": 0, "msg": msg, "dt": 0}	14✔
225	else:
226	# Calculate the time since the last message and save the timestamp
227	self.ids[key]["dt"] = msg.timestamp - self.ids[key]["msg"].timestamp	14✔
228
229	# Copy the CAN-Bus frame - this is used for sorting
230	self.ids[key]["msg"] = msg	14✔
231
232	# Increment frame counter
233	self.ids[key]["count"] += 1	14✔
234
235	# Format the CAN-Bus ID as a hex value
236	arbitration_id_string = (	14✔
237	"0x{0:0{1}X}".format( # pylint: disable=consider-using-f-string
238	msg.arbitration_id,
239	8 if msg.is_extended_id else 3,
240	)
241	)
242
243	# Use red for error frames
244	if msg.is_error_frame:	14✔
245	color = curses.color_pair(1)	14✔
246	else:
247	color = curses.color_pair(0)	14✔
248
249	# Now draw the CAN-Bus message on the terminal window
250	self.draw_line(self.ids[key]["row"], 0, str(self.ids[key]["count"]), color)	14✔
251	self.draw_line(	14✔
252	self.ids[key]["row"],
253	8,
254	f"{self.ids[key]['msg'].timestamp - self.start_time:.6f}",
255	color,
256	)
257	self.draw_line(self.ids[key]["row"], 23, f"{self.ids[key]['dt']:.6f}", color)	14✔
258	self.draw_line(self.ids[key]["row"], 35, arbitration_id_string, color)	14✔
259	self.draw_line(self.ids[key]["row"], 47, str(msg.dlc), color)	14✔
260
261	try:	14✔
262	previous_byte_values = self.previous_values[key]	14✔
263	except KeyError: # no row of previous values exists for the current message ID	14✔
264	# initialise a row to store the values for comparison next time
265	self.previous_values[key] = {}	14✔
266	previous_byte_values = self.previous_values[key]	14✔
267	for i, b in enumerate(msg.data):	14✔
268	col = 52 + i * 3	14✔
269	if col > self.x - 2:	14✔
270	# Data does not fit
271	self.draw_line(self.ids[key]["row"], col - 4, "...", color)	×
272	break	×
273	if self.highlight_changed_bytes:	14✔
274	try:	14✔
275	if b != previous_byte_values[i]:	14✔
276	# set colour to highlight a changed value
277	data_color = curses.color_pair(2)	×
278	else:
279	data_color = color	×
280	except KeyError:	14✔
281	# previous entry for byte didn't exist - default to rest of line colour
282	data_color = color	14✔
283	finally:
284	# write the new value to the previous values dict for next time
285	previous_byte_values[i] = b	14✔
286	else:
287	data_color = color	14✔
288	text = f"{b:02X}"	14✔
289	self.draw_line(self.ids[key]["row"], col, text, data_color)	14✔
290
291	if self.data_structs:	14✔
292	try:	14✔
293	values_list = []	14✔
294	for x in self.unpack_data(	14✔
295	msg.arbitration_id, self.data_structs, msg.data
296	):
297	if isinstance(x, float):	14✔
298	values_list.append(f"{x:.6f}")	14✔
299	else:
300	values_list.append(str(x))	14✔
301	values_string = " ".join(values_list)	14✔
302	self.ids[key]["values_string_length"] = len(values_string)	14✔
303	values_string += " " * (self.x - len(values_string))	14✔
304
305	self.draw_line(self.ids[key]["row"], 77, values_string, color)	14✔
306	except (ValueError, struct.error):	14✔
307	pass	14✔
308
309	return self.ids[key]	14✔
310
311	def draw_line(self, row, col, txt, *args):	21✔
312	if row - self.scroll < 0:	14✔
313	# Skip if we have scrolled past the line
314	return	14✔
315	try:	14✔
316	self.stdscr.addstr(row - self.scroll, col, txt, *args)	14✔
317	except curses.error:	14✔
318	# Ignore if we are trying to write outside the window
319	# This happens if the terminal window is too small
320	pass	14✔
321
322	def draw_header(self):	21✔
323	self.stdscr.erase()	14✔
324	self.draw_line(0, 0, "Count", curses.A_BOLD)	14✔
325	self.draw_line(0, 8, "Time", curses.A_BOLD)	14✔
326	self.draw_line(0, 23, "dt", curses.A_BOLD)	14✔
327	self.draw_line(0, 35, "ID", curses.A_BOLD)	14✔
328	self.draw_line(0, 47, "DLC", curses.A_BOLD)	14✔
329	self.draw_line(0, 52, "Data", curses.A_BOLD)	14✔
330
331	# Indicate that byte change highlighting is enabled
332	if self.highlight_changed_bytes:	14✔
333	self.draw_line(0, 57, "(changed)", curses.color_pair(2))	14✔
334	# Only draw if the dictionary is not empty
335	if self.data_structs:	14✔
336	self.draw_line(0, 77, "Parsed values", curses.A_BOLD)	14✔
337
338	def redraw_screen(self):	21✔
339	# Trigger a complete redraw
340	self.draw_header()	14✔
341	for ids in self.ids.values():	14✔
342	self.draw_can_bus_message(ids["msg"])	14✔
343
344
345	class SmartFormatter(argparse.HelpFormatter):	21✔
346	def _get_default_metavar_for_optional(self, action):	21✔
347	return action.dest.upper()	14✔
348
349	def _format_usage(self, usage, actions, groups, prefix):	21✔
350	# Use uppercase for "Usage:" text
351	return super()._format_usage(usage, actions, groups, "Usage: ")	14✔
352
353	def _format_args(self, action, default_metavar):	21✔
354	if action.nargs not in (argparse.REMAINDER, argparse.ONE_OR_MORE):	14✔
355	return super()._format_args(action, default_metavar)	14✔
356
357	# Use the metavar if "REMAINDER" or "ONE_OR_MORE" is set
358	get_metavar = self._metavar_formatter(action, default_metavar)	14✔
359	return str(get_metavar(1))	14✔
360
361	def _format_action_invocation(self, action):	21✔
362	if not action.option_strings or action.nargs == 0:	14✔
363	return super()._format_action_invocation(action)	14✔
364
365	# Modified so "-s ARGS, --long ARGS" is replaced with "-s, --long ARGS"
366	else:
367	parts = []	14✔
368	default = self._get_default_metavar_for_optional(action)	14✔
369	args_string = self._format_args(action, default)	14✔
370	for i, option_string in enumerate(action.option_strings):	14✔
371	if i == len(action.option_strings) - 1:	14✔
372	parts.append(f"{option_string} {args_string}")	14✔
373	else:
374	parts.append(str(option_string))	14✔
375	return ", ".join(parts)	14✔
376
377	def _split_lines(self, text, width):	21✔
378	# Allow to manually split the lines
379	if text.startswith("R\|"):	14✔
380	return text[2:].splitlines()	14✔
381	return super()._split_lines(text, width)	14✔
382
383	def _fill_text(self, text, width, indent):	21✔
384	if text.startswith("R\|"):	14✔
385	return "".join(indent + line + "\n" for line in text[2:].splitlines())	14✔
386	else:
387	return super()._fill_text(text, width, indent)	14✔
388
389
390	def _parse_viewer_args(	21✔
391	args: list[str],
392	) -> tuple[argparse.Namespace, TDataStructs]:
393	# Parse command line arguments
394	parser = argparse.ArgumentParser(	14✔
395	"python -m can.viewer",
396	description="A simple CAN viewer terminal application written in Python",
397	epilog="R\|Shortcuts: "
398	"\n +---------+-------------------------------+"
399	"\n \| Key \| Description \|"
400	"\n +---------+-------------------------------+"
401	"\n \| ESQ/q \| Exit the viewer \|"
402	"\n \| c \| Clear the stored frames \|"
403	"\n \| s \| Sort the stored frames \|"
404	"\n \| h \| Toggle highlight byte changes \|"
405	"\n \| SPACE \| Pause the viewer \|"
406	"\n \| UP/DOWN \| Scroll the viewer \|"
407	"\n +---------+-------------------------------+",
408	formatter_class=SmartFormatter,
409	add_help=False,
410	allow_abbrev=False,
411	)
412
413	# add bus options group
414	add_bus_arguments(parser, filter_arg=True, group_title="Bus arguments")	14✔
415
416	optional = parser.add_argument_group("Optional arguments")	14✔
417
418	optional.add_argument(	14✔
419	"-h", "--help", action="help", help="Show this help message and exit"
420	)
421
422	optional.add_argument(	14✔
423	"--version",
424	action="version",
425	help="Show program's version number and exit",
426	version=f"%(prog)s (version {__version__})",
427	)
428
429	optional.add_argument(	14✔
430	"-d",
431	"--decode",
432	dest="decode",
433	help="R\|Specify how to convert the raw bytes into real values."
434	"\nThe ID of the frame is given as the first argument and the format as the second."
435	"\nThe Python struct package is used to unpack the received data"
436	"\nwhere the format characters have the following meaning:"
437	"\n < = little-endian, > = big-endian"
438	"\n x = pad byte"
439	"\n c = char"
440	"\n ? = bool"
441	"\n b = int8_t, B = uint8_t"
442	"\n h = int16, H = uint16"
443	"\n l = int32_t, L = uint32_t"
444	"\n q = int64_t, Q = uint64_t"
445	"\n f = float (32-bits), d = double (64-bits)"
446	"\nFx to convert six bytes with ID 0x100 into uint8_t, uint16 and uint32_t:"
447	'\n $ python -m can.viewer -d "100:<BHL"'
448	"\nNote that the IDs are always interpreted as hex values."
449	"\nAn optional conversion from integers to real units can be given"
450	"\nas additional arguments. In order to convert from raw integer"
451	"\nvalues the values are divided with the corresponding scaling value,"
452	"\nsimilarly the values are multiplied by the scaling value in order"
453	"\nto convert from real units to raw integer values."
454	"\nFx lets say the uint8_t needs no conversion, but the uint16 and the uint32_t"
455	"\nneeds to be divided by 10 and 100 respectively:"
456	'\n $ python -m can.viewer -d "101:<BHL:1:10.0:100.0"'
457	"\nBe aware that integer division is performed if the scaling value is an integer."
458	"\nMultiple arguments are separated by spaces:"
459	'\n $ python -m can.viewer -d "100:<BHL" "101:<BHL:1:10.0:100.0"'
460	"\nAlternatively a file containing the conversion strings separated by new lines"
461	"\ncan be given as input:"
462	"\n $ cat file.txt"
463	"\n 100:<BHL"
464	"\n 101:<BHL:1:10.0:100.0"
465	"\n $ python -m can.viewer -d file.txt",
466	metavar="{<id>:<format>,<id>:<format>:<scaling1>:...:<scalingN>,file.txt}",
467	nargs=argparse.ONE_OR_MORE,
468	default="",
469	)
470
471	optional.add_argument(	14✔
472	"-v",
473	action="count",
474	dest="verbosity",
475	help="""How much information do you want to see at the command line?
476	You can add several of these e.g., -vv is DEBUG""",
477	default=2,
478	)
479
480	# Print help message when no arguments are given
481	if not args:	14✔
482	parser.print_help(sys.stderr)	14✔
483	raise SystemExit(errno.EINVAL)	14✔
484
485	parsed_args, unknown_args = parser.parse_known_args(args)	14✔
486	if unknown_args:	14✔
NEW 487	print("Unknown arguments:", unknown_args)	×
488
489	# Dictionary used to convert between Python values and C structs represented as Python strings.
490	# If the value is 'None' then the message does not contain any data package.
491	#
492	# The struct package is used to unpack the received data.
493	# Note the data is assumed to be in little-endian byte order.
494	# < = little-endian, > = big-endian
495	# x = pad byte
496	# c = char
497	# ? = bool
498	# b = int8_t, B = uint8_t
499	# h = int16, H = uint16
500	# l = int32_t, L = uint32_t
501	# q = int64_t, Q = uint64_t
502	# f = float (32-bits), d = double (64-bits)
503	#
504	# An optional conversion from real units to integers can be given as additional arguments.
505	# In order to convert from raw integer value the real units are multiplied with the values and
506	# similarly the values
507	# are divided by the value in order to convert from real units to raw integer values.
508
509	data_structs: TDataStructs = {}	14✔
510	if parsed_args.decode:	14✔
511	if os.path.isfile(parsed_args.decode[0]):	14✔
512	with open(parsed_args.decode[0], encoding="utf-8") as f:	14✔
513	structs = f.readlines()	14✔
514	else:
515	structs = parsed_args.decode	14✔
516
517	for s in structs:	14✔
518	tmp = s.rstrip("\n").split(":")	14✔
519
520	# The ID is given as a hex value, the format needs no conversion
521	key, fmt = int(tmp[0], base=16), tmp[1]	14✔
522
523	# The scaling
524	scaling: list[float] = []	14✔
525	for t in tmp[2:]:	14✔
526	# First try to convert to int, if that fails, then convert to a float
527	try:	14✔
528	scaling.append(int(t))	14✔
529	except ValueError:	14✔
530	scaling.append(float(t))	14✔
531
532	if scaling:	14✔
533	data_structs[key] = (struct.Struct(fmt), *scaling)	14✔
534	else:
535	data_structs[key] = struct.Struct(fmt)	14✔
536
537	return parsed_args, data_structs	14✔
538
539
540	def main() -> None:	21✔
NEW 541	parsed_args, data_structs = _parse_viewer_args(sys.argv[1:])	×
NEW 542	bus = create_bus_from_namespace(parsed_args)	×
NEW 543	_set_logging_level_from_namespace(parsed_args)	×
UNCOV 544	curses.wrapper(CanViewer, bus, data_structs) # type: ignore[attr-defined,unused-ignore]	×
545
546
547	if __name__ == "__main__":	21✔
548	# Catch ctrl+c
549	try:
550	main()
551	except KeyboardInterrupt:
552	pass

hardbyte / python-can / 15474433755

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous