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

96.23

/can/logger.py

import argparse
import errno
import sys
from datetime import datetime
from typing import (
    TYPE_CHECKING,
    Union,
)

from can import BusState, Logger, SizedRotatingLogger
from can.cli import (
    _add_extra_args,
    _parse_additional_config,
    _set_logging_level_from_namespace,
    add_bus_arguments,
    create_bus_from_namespace,
)
from can.typechecking import TAdditionalCliArgs

if TYPE_CHECKING:
    from can.io import BaseRotatingLogger
    from can.io.generic import MessageWriter


def _parse_logger_args(
    args: list[str],
) -> tuple[argparse.Namespace, TAdditionalCliArgs]:
    """Parse command line arguments for logger script."""

    parser = argparse.ArgumentParser(
        description="Log CAN traffic, printing messages to stdout or to a "
        "given file.",
    )

    logger_group = parser.add_argument_group("logger arguments")

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

    logger_group.add_argument(
        "-a",
        "--append",
        dest="append",
        help="Append to the log file if it already exists.",
        action="store_true",
    )

    logger_group.add_argument(
        "-s",
        "--file_size",
        dest="file_size",
        type=int,
        help="Maximum file size in bytes. Rotate log file when size threshold "
        "is reached. (The resulting file sizes will be consistent, but are not "
        "guaranteed to be exactly what is specified here due to the rollover "
        "conditions being logger implementation specific.)",
        default=None,
    )

    logger_group.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,
    )

    state_group = logger_group.add_mutually_exclusive_group(required=False)
    state_group.add_argument(
        "--active",
        help="Start the bus as active, this is applied by default.",
        action="store_true",
    )
    state_group.add_argument(
        "--passive", help="Start the bus as passive.", action="store_true"
    )

    # handle remaining arguments
    _add_extra_args(logger_group)

    # add bus options
    add_bus_arguments(parser, filter_arg=True)

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

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


def main() -> None:
    results, additional_config = _parse_logger_args(sys.argv[1:])
    bus = create_bus_from_namespace(results)
    _set_logging_level_from_namespace(results)

    if results.active:
        bus.state = BusState.ACTIVE
    elif results.passive:
        bus.state = BusState.PASSIVE

    print(f"Connected to {bus.__class__.__name__}: {bus.channel_info}")
    print(f"Can Logger (Started on {datetime.now()})")

    logger: Union[MessageWriter, BaseRotatingLogger]
    if results.file_size:
        logger = SizedRotatingLogger(
            base_filename=results.log_file,
            max_bytes=results.file_size,
            append=results.append,
            **additional_config,
        )
    else:
        logger = Logger(
            filename=results.log_file,
            append=results.append,
            **additional_config,
        )

    try:
        while True:
            msg = bus.recv(1)
            if msg is not None:
                logger(msg)
    except KeyboardInterrupt:
        pass
    finally:
        bus.shutdown()
        logger.stop()


if __name__ == "__main__":
    main()

1	import argparse	21✔
2	import errno	21✔
3	import sys	21✔
4	from datetime import datetime	21✔
5	from typing import (	21✔
6	TYPE_CHECKING,
7	Union,
8	)
9
10	from can import BusState, Logger, SizedRotatingLogger	21✔
11	from can.cli import (	21✔
12	_add_extra_args,
13	_parse_additional_config,
14	_set_logging_level_from_namespace,
15	add_bus_arguments,
16	create_bus_from_namespace,
17	)
18	from can.typechecking import TAdditionalCliArgs	21✔
19
20	if TYPE_CHECKING:	21✔
UNCOV 21	from can.io import BaseRotatingLogger	×
UNCOV 22	from can.io.generic import MessageWriter	×
23
24
25	def _parse_logger_args(	21✔
26	args: list[str],
27	) -> tuple[argparse.Namespace, TAdditionalCliArgs]:
28	"""Parse command line arguments for logger script."""
29
30	parser = argparse.ArgumentParser(	21✔
31	description="Log CAN traffic, printing messages to stdout or to a "
32	"given file.",
33	)
34
35	logger_group = parser.add_argument_group("logger arguments")	21✔
36
37	logger_group.add_argument(	21✔
38	"-f",
39	"--file_name",
40	dest="log_file",
41	help="Path and base log filename, for supported types see can.Logger.",
42	default=None,
43	)
44
45	logger_group.add_argument(	21✔
46	"-a",
47	"--append",
48	dest="append",
49	help="Append to the log file if it already exists.",
50	action="store_true",
51	)
52
53	logger_group.add_argument(	21✔
54	"-s",
55	"--file_size",
56	dest="file_size",
57	type=int,
58	help="Maximum file size in bytes. Rotate log file when size threshold "
59	"is reached. (The resulting file sizes will be consistent, but are not "
60	"guaranteed to be exactly what is specified here due to the rollover "
61	"conditions being logger implementation specific.)",
62	default=None,
63	)
64
65	logger_group.add_argument(	21✔
66	"-v",
67	action="count",
68	dest="verbosity",
69	help="""How much information do you want to see at the command line?
70	You can add several of these e.g., -vv is DEBUG""",
71	default=2,
72	)
73
74	state_group = logger_group.add_mutually_exclusive_group(required=False)	21✔
75	state_group.add_argument(	21✔
76	"--active",
77	help="Start the bus as active, this is applied by default.",
78	action="store_true",
79	)
80	state_group.add_argument(	21✔
81	"--passive", help="Start the bus as passive.", action="store_true"
82	)
83
84	# handle remaining arguments
85	_add_extra_args(logger_group)	21✔
86
87	# add bus options
88	add_bus_arguments(parser, filter_arg=True)	21✔
89
90	# print help message when no arguments were given
91	if not args:	21✔
92	parser.print_help(sys.stderr)	21✔
93	raise SystemExit(errno.EINVAL)	21✔
94
95	results, unknown_args = parser.parse_known_args(args)	21✔
96	additional_config = _parse_additional_config([results.extra_args, unknown_args])	21✔
97	return results, additional_config	21✔
98
99
100	def main() -> None:	21✔
101	results, additional_config = _parse_logger_args(sys.argv[1:])	21✔
102	bus = create_bus_from_namespace(results)	21✔
103	_set_logging_level_from_namespace(results)	21✔
104
105	if results.active:	21✔
106	bus.state = BusState.ACTIVE	21✔
107	elif results.passive:	21✔
108	bus.state = BusState.PASSIVE	21✔
109
110	print(f"Connected to {bus.__class__.__name__}: {bus.channel_info}")	21✔
111	print(f"Can Logger (Started on {datetime.now()})")	21✔
112
113	logger: Union[MessageWriter, BaseRotatingLogger]
114	if results.file_size:	21✔
115	logger = SizedRotatingLogger(	21✔
116	base_filename=results.log_file,
117	max_bytes=results.file_size,
118	append=results.append,
119	**additional_config,
120	)
121	else:
122	logger = Logger(	21✔
123	filename=results.log_file,
124	append=results.append,
125	**additional_config,
126	)
127
128	try:	21✔
129	while True:	15✔
130	msg = bus.recv(1)	21✔
131	if msg is not None:	21✔
132	logger(msg)	21✔
133	except KeyboardInterrupt:	21✔
134	pass	21✔
135	finally:
136	bus.shutdown()	21✔
137	logger.stop()	21✔
138
139
140	if __name__ == "__main__":	21✔
141	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