16362801995

Committed 18 Jul 2025 05:17AM UTC coverage: 70.862% (+0.1%) from 70.763%

Build # 16362801995

Build Type

Pull #1920

github

Committed by

web-flow

Commit Message

Merge f9e8a3c29 into 958fc64ed

Pull Request Pull Request #1920: add FD support to slcan according to CANable 2.0 impementation

Run Details

6 of 45 new or added lines in 1 file covered. (13.33%)

838 existing lines in 35 files now uncovered.

7770 of 10965 relevant lines covered (70.86%)

13.53 hits per line

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

91.3

/can/player.py

"""
Replays CAN traffic saved with can.logger back
to a CAN bus.

Similar to canplayer in the can-utils package.
"""

import argparse
import errno
import sys
from datetime import datetime
from typing import TYPE_CHECKING, cast

from can import LogReader, MessageSync
from can.cli import (
    _add_extra_args,
    _parse_additional_config,
    _set_logging_level_from_namespace,
    add_bus_arguments,
    create_bus_from_namespace,
)

if TYPE_CHECKING:
    from collections.abc import Iterable

    from can import Message


def main() -> None:
    parser = argparse.ArgumentParser(description="Replay CAN traffic.")

    player_group = parser.add_argument_group("Player arguments")

    player_group.add_argument(
        "-f",
        "--file_name",
        dest="log_file",
        help="Path and base log filename, for supported types see can.LogReader.",
        default=None,
    )

    player_group.add_argument(
        "-v",
        action="count",
        dest="verbosity",
        help="""Also print can frames to stdout.
                        You can add several of these to enable debugging""",
        default=2,
    )

    player_group.add_argument(
        "--ignore-timestamps",
        dest="timestamps",
        help="""Ignore timestamps (send all frames immediately with minimum gap between frames)""",
        action="store_false",
    )

    player_group.add_argument(
        "--error-frames",
        help="Also send error frames to the interface.",
        action="store_true",
    )

    player_group.add_argument(
        "-g",
        "--gap",
        type=float,
        help="<s> minimum time between replayed frames",
        default=0.0001,
    )
    player_group.add_argument(
        "-s",
        "--skip",
        type=float,
        default=60 * 60 * 24,
        help="<s> skip gaps greater than 's' seconds",
    )

    player_group.add_argument(
        "infile",
        metavar="input-file",
        type=str,
        help="The file to replay. For supported types see can.LogReader.",
    )

    # handle remaining arguments
    _add_extra_args(player_group)

    # add bus options
    add_bus_arguments(parser)

    # print help message when no arguments were given
    if len(sys.argv) < 2:
        parser.print_help(sys.stderr)
        raise SystemExit(errno.EINVAL)

    results, unknown_args = parser.parse_known_args()
    additional_config = _parse_additional_config([*results.extra_args, *unknown_args])

    _set_logging_level_from_namespace(results)
    verbosity = results.verbosity

    error_frames = results.error_frames

    with create_bus_from_namespace(results) as bus:
        with LogReader(results.infile, **additional_config) as reader:
            in_sync = MessageSync(
                cast("Iterable[Message]", reader),
                timestamps=results.timestamps,
                gap=results.gap,
                skip=results.skip,
            )

            print(f"Can LogReader (Started on {datetime.now()})")

            try:
                for message in in_sync:
                    if message.is_error_frame and not error_frames:
                        continue
                    if verbosity >= 3:
                        print(message)
                    bus.send(message)
            except KeyboardInterrupt:
                pass


if __name__ == "__main__":
    main()

1	"""	21✔
2	Replays CAN traffic saved with can.logger back
3	to a CAN bus.
4
5	Similar to canplayer in the can-utils package.
6	"""
7
8	import argparse	21✔
9	import errno	21✔
10	import sys	21✔
11	from datetime import datetime	21✔
12	from typing import TYPE_CHECKING, cast	21✔
13
14	from can import LogReader, MessageSync	21✔
15	from can.cli import (	21✔
16	_add_extra_args,
17	_parse_additional_config,
18	_set_logging_level_from_namespace,
19	add_bus_arguments,
20	create_bus_from_namespace,
21	)
22
23	if TYPE_CHECKING:	21✔
UNCOV 24	from collections.abc import Iterable	×
25
UNCOV 26	from can import Message	×
27
28
29	def main() -> None:	21✔
30	parser = argparse.ArgumentParser(description="Replay CAN traffic.")	21✔
31
32	player_group = parser.add_argument_group("Player arguments")	21✔
33
34	player_group.add_argument(	21✔
35	"-f",
36	"--file_name",
37	dest="log_file",
38	help="Path and base log filename, for supported types see can.LogReader.",
39	default=None,
40	)
41
42	player_group.add_argument(	21✔
43	"-v",
44	action="count",
45	dest="verbosity",
46	help="""Also print can frames to stdout.
47	You can add several of these to enable debugging""",
48	default=2,
49	)
50
51	player_group.add_argument(	21✔
52	"--ignore-timestamps",
53	dest="timestamps",
54	help="""Ignore timestamps (send all frames immediately with minimum gap between frames)""",
55	action="store_false",
56	)
57
58	player_group.add_argument(	21✔
59	"--error-frames",
60	help="Also send error frames to the interface.",
61	action="store_true",
62	)
63
64	player_group.add_argument(	21✔
65	"-g",
66	"--gap",
67	type=float,
68	help="<s> minimum time between replayed frames",
69	default=0.0001,
70	)
71	player_group.add_argument(	21✔
72	"-s",
73	"--skip",
74	type=float,
75	default=60 * 60 * 24,
76	help="<s> skip gaps greater than 's' seconds",
77	)
78
79	player_group.add_argument(	21✔
80	"infile",
81	metavar="input-file",
82	type=str,
83	help="The file to replay. For supported types see can.LogReader.",
84	)
85
86	# handle remaining arguments
87	_add_extra_args(player_group)	21✔
88
89	# add bus options
90	add_bus_arguments(parser)	21✔
91
92	# print help message when no arguments were given
93	if len(sys.argv) < 2:	21✔
94	parser.print_help(sys.stderr)	21✔
95	raise SystemExit(errno.EINVAL)	21✔
96
97	results, unknown_args = parser.parse_known_args()	21✔
98	additional_config = _parse_additional_config([results.extra_args, unknown_args])	21✔
99
100	_set_logging_level_from_namespace(results)	21✔
101	verbosity = results.verbosity	21✔
102
103	error_frames = results.error_frames	21✔
104
105	with create_bus_from_namespace(results) as bus:	21✔
106	with LogReader(results.infile, **additional_config) as reader:	21✔
107	in_sync = MessageSync(	21✔
108	cast("Iterable[Message]", reader),
109	timestamps=results.timestamps,
110	gap=results.gap,
111	skip=results.skip,
112	)
113
114	print(f"Can LogReader (Started on {datetime.now()})")	21✔
115
116	try:	21✔
117	for message in in_sync:	21✔
118	if message.is_error_frame and not error_frames:	21✔
119	continue	21✔
120	if verbosity >= 3:	21✔
121	print(message)	21✔
122	bus.send(message)	21✔
UNCOV 123	except KeyboardInterrupt:	×
UNCOV 124	pass	×
125
126
127	if __name__ == "__main__":	21✔
128	main()	21✔

hardbyte / python-can / 16362801995

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