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

93.94

/can/io/csv.py

"""
This module contains handling for CSV (comma separated values) files.

TODO: CAN FD messages are not yet supported.

TODO: This module could use https://docs.python.org/2/library/csv.html#module-csv
      to allow different delimiters for writing, special escape chars to circumvent
      the base64 encoding and use csv.Sniffer to automatically deduce the delimiters
      of a CSV file.
"""

from base64 import b64decode, b64encode
from collections.abc import Generator
from typing import Any, TextIO, Union

from can.message import Message

from ..typechecking import StringPathLike
from .generic import TextIOMessageReader, TextIOMessageWriter


class CSVReader(TextIOMessageReader):
    """Iterator over CAN messages from a .csv file that was
    generated by :class:`~can.CSVWriter` or that uses the same
    format as described there. Assumes that there is a header
    and thus skips the first line.

    Any line separator is accepted.
    """

    file: TextIO

    def __init__(
        self,
        file: Union[StringPathLike, TextIO],
        **kwargs: Any,
    ) -> None:
        """
        :param file: a path-like object or as file-like object to read from
                     If this is a file-like object, is has to opened in text
                     read mode, not binary read mode.
        """
        super().__init__(file, mode="r")

    def __iter__(self) -> Generator[Message, None, None]:
        # skip the header line
        try:
            next(self.file)
        except StopIteration:
            # don't crash on a file with only a header
            return

        for line in self.file:
            timestamp, arbitration_id, extended, remote, error, dlc, data = line.split(
                ","
            )

            yield Message(
                timestamp=float(timestamp),
                is_remote_frame=(remote == "1"),
                is_extended_id=(extended == "1"),
                is_error_frame=(error == "1"),
                arbitration_id=int(arbitration_id, base=16),
                dlc=int(dlc),
                data=b64decode(data),
            )

        self.stop()


class CSVWriter(TextIOMessageWriter):
    """Writes a comma separated text file with a line for
    each message. Includes a header line.

    The columns are as follows:

    ================ ======================= ===============
    name of column   format description      example
    ================ ======================= ===============
    timestamp        decimal float           1483389946.197
    arbitration_id   hex                     0x00dadada
    extended         1 == True, 0 == False   1
    remote           1 == True, 0 == False   0
    error            1 == True, 0 == False   0
    dlc              int                     6
    data             base64 encoded          WzQyLCA5XQ==
    ================ ======================= ===============

    Each line is terminated with a platform specific line separator.
    """

    file: TextIO

    def __init__(
        self,
        file: Union[StringPathLike, TextIO],
        append: bool = False,
        **kwargs: Any,
    ) -> None:
        """
        :param file: a path-like object or a file-like object to write to.
                     If this is a file-like object, is has to open in text
                     write mode, not binary write mode.
        :param bool append: if set to `True` messages are appended to
                            the file and no header line is written, else
                            the file is truncated and starts with a newly
                            written header line
        """
        mode = "a" if append else "w"
        super().__init__(file, mode=mode)

        # Write a header row
        if not append:
            self.file.write("timestamp,arbitration_id,extended,remote,error,dlc,data\n")

    def on_message_received(self, msg: Message) -> None:
        row = ",".join(
            [
                repr(msg.timestamp),  # cannot use str() here because that is rounding
                hex(msg.arbitration_id),
                "1" if msg.is_extended_id else "0",
                "1" if msg.is_remote_frame else "0",
                "1" if msg.is_error_frame else "0",
                str(msg.dlc),
                b64encode(msg.data).decode("utf8"),
            ]
        )
        self.file.write(row)
        self.file.write("\n")

1	"""	21✔
2	This module contains handling for CSV (comma separated values) files.
3
4	TODO: CAN FD messages are not yet supported.
5
6	TODO: This module could use https://docs.python.org/2/library/csv.html#module-csv
7	to allow different delimiters for writing, special escape chars to circumvent
8	the base64 encoding and use csv.Sniffer to automatically deduce the delimiters
9	of a CSV file.
10	"""
11
12	from base64 import b64decode, b64encode	21✔
13	from collections.abc import Generator	21✔
14	from typing import Any, TextIO, Union	21✔
15
16	from can.message import Message	21✔
17
18	from ..typechecking import StringPathLike	21✔
19	from .generic import TextIOMessageReader, TextIOMessageWriter	21✔
20
21
22	class CSVReader(TextIOMessageReader):	21✔
23	"""Iterator over CAN messages from a .csv file that was	21✔
24	generated by :class:`~can.CSVWriter` or that uses the same
25	format as described there. Assumes that there is a header
26	and thus skips the first line.
27
28	Any line separator is accepted.
29	"""
30
31	file: TextIO	21✔
32
33	def __init__(	21✔
34	self,
35	file: Union[StringPathLike, TextIO],
36	**kwargs: Any,
37	) -> None:
38	"""
39	:param file: a path-like object or as file-like object to read from
40	If this is a file-like object, is has to opened in text
41	read mode, not binary read mode.
42	"""
43	super().__init__(file, mode="r")	21✔
44
45	def __iter__(self) -> Generator[Message, None, None]:	21✔
46	# skip the header line
47	try:	21✔
48	next(self.file)	21✔
UNCOV 49	except StopIteration:	×
50	# don't crash on a file with only a header
UNCOV 51	return	×
52
53	for line in self.file:	21✔
54	timestamp, arbitration_id, extended, remote, error, dlc, data = line.split(	21✔
55	","
56	)
57
58	yield Message(	21✔
59	timestamp=float(timestamp),
60	is_remote_frame=(remote == "1"),
61	is_extended_id=(extended == "1"),
62	is_error_frame=(error == "1"),
63	arbitration_id=int(arbitration_id, base=16),
64	dlc=int(dlc),
65	data=b64decode(data),
66	)
67
68	self.stop()	21✔
69
70
71	class CSVWriter(TextIOMessageWriter):	21✔
72	"""Writes a comma separated text file with a line for	21✔
73	each message. Includes a header line.
74
75	The columns are as follows:
76
77	================ ======================= ===============
78	name of column format description example
79	================ ======================= ===============
80	timestamp decimal float 1483389946.197
81	arbitration_id hex 0x00dadada
82	extended 1 == True, 0 == False 1
83	remote 1 == True, 0 == False 0
84	error 1 == True, 0 == False 0
85	dlc int 6
86	data base64 encoded WzQyLCA5XQ==
87	================ ======================= ===============
88
89	Each line is terminated with a platform specific line separator.
90	"""
91
92	file: TextIO	21✔
93
94	def __init__(	21✔
95	self,
96	file: Union[StringPathLike, TextIO],
97	append: bool = False,
98	**kwargs: Any,
99	) -> None:
100	"""
101	:param file: a path-like object or a file-like object to write to.
102	If this is a file-like object, is has to open in text
103	write mode, not binary write mode.
104	:param bool append: if set to `True` messages are appended to
105	the file and no header line is written, else
106	the file is truncated and starts with a newly
107	written header line
108	"""
109	mode = "a" if append else "w"	21✔
110	super().__init__(file, mode=mode)	21✔
111
112	# Write a header row
113	if not append:	21✔
114	self.file.write("timestamp,arbitration_id,extended,remote,error,dlc,data\n")	21✔
115
116	def on_message_received(self, msg: Message) -> None:	21✔
117	row = ",".join(	21✔
118	[
119	repr(msg.timestamp), # cannot use str() here because that is rounding
120	hex(msg.arbitration_id),
121	"1" if msg.is_extended_id else "0",
122	"1" if msg.is_remote_frame else "0",
123	"1" if msg.is_error_frame else "0",
124	str(msg.dlc),
125	b64encode(msg.data).decode("utf8"),
126	]
127	)
128	self.file.write(row)	21✔
129	self.file.write("\n")	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