Hide navigation sidebar
Hide table of contents sidebar
Back to top
Toggle table of contents sidebar

# Source code for can.util

    """
    Utilities and configuration file parsing.
    """
    import copy
    import functools
    import json
    import logging
    import os
    import os.path
    import platform
    import re
    import warnings
    from configparser import ConfigParser
    from time import get_clock_info, perf_counter, time
    from typing import (
        Any,
        Callable,
        Dict,
        Iterable,
        Optional,
        Tuple,
        TypeVar,
        Union,
        cast,
    )

    from typing_extensions import ParamSpec

    import can

    from . import typechecking
    from .bit_timing import BitTiming, BitTimingFd
    from .exceptions import CanInitializationError, CanInterfaceNotImplementedError
    from .interfaces import VALID_INTERFACES

    log = logging.getLogger("can.util")

    # List of valid data lengths for a CAN FD message
    CAN_FD_DLC = [0, 1, 2, 3, 4, 5, 6, 7, 8, 12, 16, 20, 24, 32, 48, 64]

    REQUIRED_KEYS = ["interface", "channel"]

    CONFIG_FILES = ["~/can.conf"]

    if platform.system() in ("Linux", "Darwin"):
        CONFIG_FILES.extend(["/etc/can.conf", "~/.can", "~/.canrc"])
    elif platform.system() == "Windows" or platform.python_implementation() == "IronPython":
        CONFIG_FILES.extend(["can.ini", os.path.join(os.getenv("APPDATA", ""), "can.ini")])

    def load_file_config(
        path: Optional[typechecking.AcceptedIOType] = None, section: str = "default"
    ) -> Dict[str, str]:
        """
        Loads configuration from file with following content::

            [default]
            interface = socketcan
            channel = can0

        :param path:
            path to config file. If not specified, several sensible
            default locations are tried depending on platform.
        :param section:
            name of the section to read configuration from.
        """
        config = ConfigParser()

        # make sure to not transform the entries such that capitalization is preserved
        config.optionxform = lambda entry: entry  # type: ignore

        if path is None:
            config.read([os.path.expanduser(path) for path in CONFIG_FILES])
        else:
            config.read(path)

        _config: Dict[str, str] = {}

        if config.has_section(section):
            _config.update(config.items(section))

        return _config

    def load_environment_config(context: Optional[str] = None) -> Dict[str, str]:
        """
        Loads config dict from environmental variables (if set):

        * CAN_INTERFACE
        * CAN_CHANNEL
        * CAN_BITRATE
        * CAN_CONFIG

        if context is supplied, "_{context}" is appended to the environment
        variable name we will look at. For example if context="ABC":

        * CAN_INTERFACE_ABC
        * CAN_CHANNEL_ABC
        * CAN_BITRATE_ABC
        * CAN_CONFIG_ABC

        """
        mapper = {
            "interface": "CAN_INTERFACE",
            "channel": "CAN_CHANNEL",
            "bitrate": "CAN_BITRATE",
        }

        context_suffix = f"_{context}" if context else ""
        can_config_key = f"CAN_CONFIG{context_suffix}"
        config: Dict[str, str] = json.loads(os.environ.get(can_config_key, "{}"))

        for key, val in mapper.items():
            config_option = os.environ.get(val + context_suffix, None)
            if config_option:
                config[key] = config_option

        return config

    def load_config(
        path: Optional[typechecking.AcceptedIOType] = None,
        config: Optional[Dict[str, Any]] = None,
        context: Optional[str] = None,
    ) -> typechecking.BusConfig:
        """
        Returns a dict with configuration details which is loaded from (in this order):

        - config
        - can.rc
        - Environment variables CAN_INTERFACE, CAN_CHANNEL, CAN_BITRATE
        - Config files ``/etc/can.conf`` or ``~/.can`` or ``~/.canrc``
          where the latter may add or replace values of the former.

        Interface can be any of the strings from ``can.VALID_INTERFACES`` for example:
        kvaser, socketcan, pcan, usb2can, ixxat, nican, virtual.

        .. note::

                The key ``bustype`` is copied to ``interface`` if that one is missing
                and does never appear in the result.

        :param path:
            Optional path to config file.

        :param config:
            A dict which may set the 'interface', and/or the 'channel', or neither.
            It may set other values that are passed through.

        :param context:
            Extra 'context' pass to config sources. This can be used to section
            other than 'default' in the configuration file.

        :return:
            A config dictionary that should contain 'interface' & 'channel'::

                {
                    'interface': 'python-can backend interface to use',
                    'channel': 'default channel to use',
                    # possibly more
                }

            Note ``None`` will be used if all the options are exhausted without
            finding a value.

            All unused values are passed from ``config`` over to this.

        :raises:
            CanInterfaceNotImplementedError if the ``interface`` name isn't recognized
        """

        # Start with an empty dict to apply filtering to all sources
        given_config = config or {}
        config = {}

        # Use the given dict for default values
        config_sources = cast(
            Iterable[Union[Dict[str, Any], Callable[[Any], Dict[str, Any]]]],
            [
                given_config,
                can.rc,
                lambda _context: load_environment_config(  # pylint: disable=unnecessary-lambda
                    _context
                ),
                lambda _context: load_environment_config(),
                lambda _context: load_file_config(path, _context),
                lambda _context: load_file_config(path),
            ],
        )

        # Slightly complex here to only search for the file config if required
        for _cfg in config_sources:
            cfg = _cfg(context) if callable(_cfg) else _cfg
            # remove legacy operator (and copy to interface if not already present)
            if "bustype" in cfg:
                if "interface" not in cfg or not cfg["interface"]:
                    cfg["interface"] = cfg["bustype"]
                del cfg["bustype"]
            # copy all new parameters
            for key, val in cfg.items():
                if key not in config:
                    if isinstance(val, str):
                        config[key] = cast_from_string(val)
                    else:
                        config[key] = cfg[key]

        bus_config = _create_bus_config(config)
        can.log.debug("can config: %s", bus_config)
        return bus_config

    def _create_bus_config(config: Dict[str, Any]) -> typechecking.BusConfig:
        """Validates some config values, performs compatibility mappings and creates specific
        structures (e.g. for bit timings).

        :param config: The raw config as specified by the user
        :return: A config that can be used by a :class:`~can.BusABC`
        :raises NotImplementedError: if the ``interface`` is unknown
        """
        # substitute None for all values not found
        for key in REQUIRED_KEYS:
            if key not in config:
                config[key] = None

        if config["interface"] not in VALID_INTERFACES:
            raise CanInterfaceNotImplementedError(
                f'Unknown interface type "{config["interface"]}"'
            )
        if "port" in config:
            # convert port to integer if necessary
            if isinstance(config["port"], int):
                port = config["port"]
            elif isinstance(config["port"], str):
                if config["port"].isnumeric():
                    config["port"] = port = int(config["port"])
                else:
                    raise ValueError("Port config must be a number!")
            else:
                raise TypeError("Port config must be string or integer!")

            if not 0 < port < 65535:
                raise ValueError("Port config must be inside 0-65535 range!")

        if config.get("timing", None) is None:
            try:
                if set(typechecking.BitTimingFdDict.__annotations__).issubset(config):
                    config["timing"] = can.BitTimingFd(
                        **{
                            key: int(config[key])
                            for key in typechecking.BitTimingFdDict.__annotations__
                        },
                        strict=False,
                    )
                elif set(typechecking.BitTimingDict.__annotations__).issubset(config):
                    config["timing"] = can.BitTiming(
                        **{
                            key: int(config[key])
                            for key in typechecking.BitTimingDict.__annotations__
                        },
                        strict=False,
                    )
            except (ValueError, TypeError):
                pass

        if "fd" in config:
            config["fd"] = config["fd"] not in (0, False)

        return cast(typechecking.BusConfig, config)

    def set_logging_level(level_name: str) -> None:
        """Set the logging level for the `"can"` logger.

        :param level_name:
            One of: `'critical'`, `'error'`, `'warning'`, `'info'`,
            `'debug'`, `'subdebug'`, or the value :obj:`None` (=default).
            Defaults to `'debug'`.
        """
        can_logger = logging.getLogger("can")

        try:
            can_logger.setLevel(getattr(logging, level_name.upper()))
        except AttributeError:
            can_logger.setLevel(logging.DEBUG)
        log.debug("Logging set to %s", level_name)

    def len2dlc(length: int) -> int:
        """Calculate the DLC from data length.

        :param length: Length in number of bytes (0-64)

        :returns: DLC (0-15)
        """
        if length <= 8:
            return length
        for dlc, nof_bytes in enumerate(CAN_FD_DLC):
            if nof_bytes >= length:
                return dlc
        return 15

    def dlc2len(dlc: int) -> int:
        """Calculate the data length from DLC.

        :param dlc: DLC (0-15)

        :returns: Data length in number of bytes (0-64)
        """
        return CAN_FD_DLC[dlc] if dlc <= 15 else 64

    def channel2int(channel: Optional[typechecking.Channel]) -> Optional[int]:
        """Try to convert the channel to an integer.

        :param channel:
            Channel string (e.g. `"can0"`, `"CAN1"`) or an integer

        :returns: Channel integer or ``None`` if unsuccessful
        """
        if isinstance(channel, int):
            return channel
        if isinstance(channel, str):
            match = re.match(r".*?(\d+)$", channel)
            if match:
                return int(match.group(1))
        return None

    P1 = ParamSpec("P1")
    T1 = TypeVar("T1")

    def deprecated_args_alias(
        deprecation_start: str,
        deprecation_end: Optional[str] = None,
        **aliases: Optional[str],
    ) -> Callable[[Callable[P1, T1]], Callable[P1, T1]]:
        """Allows to rename/deprecate a function kwarg(s) and optionally
        have the deprecated kwarg(s) set as alias(es)

        Example::

            @deprecated_args_alias("1.2.0", oldArg="new_arg", anotherOldArg="another_new_arg")
            def library_function(new_arg, another_new_arg):
                pass

            @deprecated_args_alias(
                deprecation_start="1.2.0",
                deprecation_end="3.0.0",
                oldArg="new_arg",
                obsoleteOldArg=None,
            )
            def library_function(new_arg):
                pass

        :param deprecation_start:
            The *python-can* version, that introduced the :class:`DeprecationWarning`.
        :param deprecation_end:
            The *python-can* version, that marks the end of the deprecation period.
        :param aliases:
            keyword arguments, that map the deprecated argument names
            to the new argument names or ``None``.

        """

        def deco(f: Callable[P1, T1]) -> Callable[P1, T1]:
            @functools.wraps(f)
            def wrapper(*args: P1.args, **kwargs: P1.kwargs) -> T1:
                _rename_kwargs(
                    func_name=f.__name__,
                    start=deprecation_start,
                    end=deprecation_end,
                    kwargs=kwargs,
                    aliases=aliases,
                )
                return f(*args, **kwargs)

            return wrapper

        return deco

    def _rename_kwargs(
        func_name: str,
        start: str,
        end: Optional[str],
        kwargs: P1.kwargs,
        aliases: Dict[str, Optional[str]],
    ) -> None:
        """Helper function for `deprecated_args_alias`"""
        for alias, new in aliases.items():
            if alias in kwargs:
                deprecation_notice = (
                    f"The '{alias}' argument is deprecated since python-can v{start}"
                )
                if end:
                    deprecation_notice += (
                        f", and scheduled for removal in python-can v{end}"
                    )
                deprecation_notice += "."

                value = kwargs.pop(alias)
                if new is not None:
                    deprecation_notice += f" Use '{new}' instead."

                    if new in kwargs:
                        raise TypeError(
                            f"{func_name} received both '{alias}' (deprecated) and '{new}'."
                        )
                    kwargs[new] = value

                warnings.warn(deprecation_notice, DeprecationWarning)

    T2 = TypeVar("T2", BitTiming, BitTimingFd)

    def check_or_adjust_timing_clock(timing: T2, valid_clocks: Iterable[int]) -> T2:
        """Adjusts the given timing instance to have an *f_clock* value that is within the
        allowed values specified by *valid_clocks*. If the *f_clock* value of timing is
        already within *valid_clocks*, then *timing* is returned unchanged.

        :param timing:
            The :class:`~can.BitTiming` or :class:`~can.BitTimingFd` instance to adjust.
        :param valid_clocks:
            An iterable of integers representing the valid *f_clock* values that the timing instance
            can be changed to. The order of the values in *valid_clocks* determines the priority in
            which they are tried, with earlier values being tried before later ones.
        :return:
            A new :class:`~can.BitTiming` or :class:`~can.BitTimingFd` instance with an
            *f_clock* value within *valid_clocks*.
        :raises ~can.exceptions.CanInitializationError:
            If no compatible *f_clock* value can be found within *valid_clocks*.
        """
        if timing.f_clock in valid_clocks:
            # create a copy so this function always returns a new instance
            return copy.deepcopy(timing)

        for clock in valid_clocks:
            try:
                # Try to use a different f_clock
                adjusted_timing = timing.recreate_with_f_clock(clock)
                warnings.warn(
                    f"Adjusted f_clock in {timing.__class__.__name__} from "
                    f"{timing.f_clock} to {adjusted_timing.f_clock}"
                )
                return adjusted_timing
            except ValueError:
                pass

        raise CanInitializationError(
            f"The specified timing.f_clock value {timing.f_clock} "
            f"doesn't match any of the allowed device f_clock values: "
            f"{', '.join([str(f) for f in valid_clocks])}"
        ) from None

    def time_perfcounter_correlation() -> Tuple[float, float]:
        """Get the `perf_counter` value nearest to when time.time() is updated

        Computed if the default timer used by `time.time` on this platform has a resolution
        higher than 10μs, otherwise the current time and perf_counter is directly returned.
        This was chosen as typical timer resolution on Linux/macOS is ~1μs, and the Windows
        platform can vary from ~500μs to 10ms.

        Note this value is based on when `time.time()` is observed to update from Python,
        it is not directly returned by the operating system.

        :returns:
            (t, performance_counter) time.time value and time.perf_counter value when the time.time
            is updated

        """

        # use this if the resolution is higher than 10us
        if get_clock_info("time").resolution > 1e-5:
            t0 = time()
            while True:
                t1, performance_counter = time(), perf_counter()
                if t1 != t0:
                    break
        else:
            return time(), perf_counter()
        return t1, performance_counter

    def cast_from_string(string_val: str) -> Union[str, int, float, bool]:
        """Perform trivial type conversion from :class:`str` values.

        :param string_val:
            the string, that shall be converted
        """
        if re.match(r"^[-+]?\d+$", string_val):
            # value is integer
            return int(string_val)

        if re.match(r"^[-+]?\d*\.\d+(?:e[-+]?\d+)?$", string_val):
            # value is float
            return float(string_val)

        if re.match(r"^(?:True|False)$", string_val, re.IGNORECASE):
            # value is bool
            return string_val.lower() == "true"

        # value is string
        return string_val

    if __name__ == "__main__":
        print("Searching for configuration named:")
        print("\n".join(CONFIG_FILES))
        print()
        print("Settings:")
        print(load_config())