15347366011

Committed 30 May 2025 01:01PM UTC coverage: 70.785%. First build

Build # 15347366011

Build Type

Pull #1945

github

Committed by

web-flow

Commit Message

Merge 0ca50a665 into c46d3ea7b

Pull Request Pull Request #1945: Use dependency groups for docs/lint/test

Run Details

7 of 10 new or added lines in 3 files covered. (70.0%)

7625 of 10772 relevant lines covered (70.79%)

13.49 hits per line

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

79.17

/can/interfaces/udp_multicast/utils.py

"""
Defines common functions.
"""

from typing import Any, Optional

from can import CanInterfaceNotImplementedError, Message
from can.typechecking import ReadableBytesLike

try:
    import msgpack
except ImportError:
    msgpack = None


def is_msgpack_installed(raise_exception: bool = True) -> bool:
    """Check whether the ``msgpack`` module is installed.

    :param raise_exception:
        If True, raise a :class:`can.CanInterfaceNotImplementedError` when ``msgpack`` is not installed.
        If False, return False instead.
    :return:
        True if ``msgpack`` is installed, False otherwise.
    :raises can.CanInterfaceNotImplementedError:
        If ``msgpack`` is not installed and ``raise_exception`` is True.
    """
    if msgpack is None:
        if raise_exception:
            raise CanInterfaceNotImplementedError("msgpack not installed")
        return False
    return True


def pack_message(message: Message) -> bytes:
    """
    Pack a can.Message into a msgpack byte blob.

    :param message: the message to be packed
    """
    is_msgpack_installed()
    as_dict = {
        "timestamp": message.timestamp,
        "arbitration_id": message.arbitration_id,
        "is_extended_id": message.is_extended_id,
        "is_remote_frame": message.is_remote_frame,
        "is_error_frame": message.is_error_frame,
        "channel": message.channel,
        "dlc": message.dlc,
        "data": message.data,
        "is_fd": message.is_fd,
        "bitrate_switch": message.bitrate_switch,
        "error_state_indicator": message.error_state_indicator,
    }
    return msgpack.packb(as_dict, use_bin_type=True)


def unpack_message(
    data: ReadableBytesLike,
    replace: Optional[dict[str, Any]] = None,
    check: bool = False,
) -> Message:
    """Unpack a can.Message from a msgpack byte blob.

    :param data: the raw data
    :param replace: a mapping from field names to values to be replaced after decoding the new message, or
                    `None` to disable this feature
    :param check: this is passed to :meth:`can.Message.__init__` to specify whether to validate the message

    :raise TypeError: if the data contains key that are not valid arguments for :meth:`can.Message.__init__`
    :raise ValueError: if `check` is true and the message metadata is invalid in some way
    :raise Exception: if there was another problem while unpacking
    """
    is_msgpack_installed()
    as_dict = msgpack.unpackb(data, raw=False)
    if replace is not None:
        as_dict.update(replace)
    return Message(check=check, **as_dict)

1	"""	21✔
2	Defines common functions.
3	"""
4
5	from typing import Any, Optional	21✔
6
7	from can import CanInterfaceNotImplementedError, Message	21✔
8	from can.typechecking import ReadableBytesLike	21✔
9
10	try:	21✔
11	import msgpack	21✔
12	except ImportError:	×
13	msgpack = None	×
14
15
16	def is_msgpack_installed(raise_exception: bool = True) -> bool:	21✔
17	"""Check whether the ``msgpack`` module is installed.
18
19	:param raise_exception:
20	If True, raise a :class:`can.CanInterfaceNotImplementedError` when ``msgpack`` is not installed.
21	If False, return False instead.
22	:return:
23	True if ``msgpack`` is installed, False otherwise.
24	:raises can.CanInterfaceNotImplementedError:
25	If ``msgpack`` is not installed and ``raise_exception`` is True.
26	"""
27	if msgpack is None:	21✔
NEW 28	if raise_exception:	×
NEW 29	raise CanInterfaceNotImplementedError("msgpack not installed")	×
NEW 30	return False	×
31	return True	21✔
32
33
34	def pack_message(message: Message) -> bytes:	21✔
35	"""
36	Pack a can.Message into a msgpack byte blob.
37
38	:param message: the message to be packed
39	"""
40	is_msgpack_installed()	14✔
41	as_dict = {	14✔
42	"timestamp": message.timestamp,
43	"arbitration_id": message.arbitration_id,
44	"is_extended_id": message.is_extended_id,
45	"is_remote_frame": message.is_remote_frame,
46	"is_error_frame": message.is_error_frame,
47	"channel": message.channel,
48	"dlc": message.dlc,
49	"data": message.data,
50	"is_fd": message.is_fd,
51	"bitrate_switch": message.bitrate_switch,
52	"error_state_indicator": message.error_state_indicator,
53	}
54	return msgpack.packb(as_dict, use_bin_type=True)	14✔
55
56
57	def unpack_message(	21✔
58	data: ReadableBytesLike,
59	replace: Optional[dict[str, Any]] = None,
60	check: bool = False,
61	) -> Message:
62	"""Unpack a can.Message from a msgpack byte blob.
63
64	:param data: the raw data
65	:param replace: a mapping from field names to values to be replaced after decoding the new message, or
66	`None` to disable this feature
67	:param check: this is passed to :meth:`can.Message.__init__` to specify whether to validate the message
68
69	:raise TypeError: if the data contains key that are not valid arguments for :meth:`can.Message.__init__`
70	:raise ValueError: if `check` is true and the message metadata is invalid in some way
71	:raise Exception: if there was another problem while unpacking
72	"""
73	is_msgpack_installed()	14✔
74	as_dict = msgpack.unpackb(data, raw=False)	14✔
75	if replace is not None:	14✔
76	as_dict.update(replace)	14✔
77	return Message(check=check, **as_dict)	14✔

hardbyte / python-can / 15347366011

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