Skip to content

Improve can.Bus typing #1557

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 1 commit into from
May 15, 2023
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
79 changes: 38 additions & 41 deletions can/interface.py
Original file line number Diff line number Diff line change
Expand Up @@ -55,8 +55,20 @@ def _get_class_for_interface(interface: str) -> Type[BusABC]:
return cast(Type[BusABC], bus_class)


class Bus(BusABC): # pylint: disable=abstract-method
"""Bus wrapper with configuration loading.
@util.deprecated_args_alias(
deprecation_start="4.2.0",
deprecation_end="5.0.0",
bustype="interface",
context="config_context",
)
def Bus(
channel: Optional[Channel] = None,
interface: Optional[str] = None,
config_context: Optional[str] = None,
ignore_config: bool = False,
**kwargs: Any,
) -> BusABC:
"""Create a new bus instance with configuration loading.

Instantiates a CAN Bus of the given ``interface``, falls back to reading a
configuration file from default locations.
Expand Down Expand Up @@ -99,45 +111,30 @@ class Bus(BusABC): # pylint: disable=abstract-method
if the ``channel`` could not be determined
"""

@staticmethod
@util.deprecated_args_alias(
deprecation_start="4.2.0",
deprecation_end="5.0.0",
bustype="interface",
context="config_context",
)
def __new__( # type: ignore
cls: Any,
channel: Optional[Channel] = None,
interface: Optional[str] = None,
config_context: Optional[str] = None,
ignore_config: bool = False,
**kwargs: Any,
) -> BusABC:
# figure out the rest of the configuration; this might raise an error
if interface is not None:
kwargs["interface"] = interface
if channel is not None:
kwargs["channel"] = channel

if not ignore_config:
kwargs = util.load_config(config=kwargs, context=config_context)

# resolve the bus class to use for that interface
cls = _get_class_for_interface(kwargs["interface"])

# remove the "interface" key, so it doesn't get passed to the backend
del kwargs["interface"]

# make sure the bus can handle this config format
channel = kwargs.pop("channel", channel)
if channel is None:
# Use the default channel for the backend
bus = cls(**kwargs)
else:
bus = cls(channel, **kwargs)
# figure out the rest of the configuration; this might raise an error
if interface is not None:
kwargs["interface"] = interface
if channel is not None:
kwargs["channel"] = channel

if not ignore_config:
kwargs = util.load_config(config=kwargs, context=config_context)

# resolve the bus class to use for that interface
cls = _get_class_for_interface(kwargs["interface"])

# remove the "interface" key, so it doesn't get passed to the backend
del kwargs["interface"]

# make sure the bus can handle this config format
channel = kwargs.pop("channel", channel)
if channel is None:
# Use the default channel for the backend
bus = cls(**kwargs)
else:
bus = cls(channel, **kwargs)

return cast(BusABC, bus)
return bus


def detect_available_configs(
Expand All @@ -146,7 +143,7 @@ def detect_available_configs(
"""Detect all configurations/channels that the interfaces could
currently connect with.

This might be quite time consuming.
This might be quite time-consuming.

Automated configuration detection may not be implemented by
every interface on every platform. This method will not raise
Expand Down
4 changes: 2 additions & 2 deletions can/logger.py
Original file line number Diff line number Diff line change
Expand Up @@ -85,7 +85,7 @@ def _append_filter_argument(
)


def _create_bus(parsed_args: Any, **kwargs: Any) -> can.Bus:
def _create_bus(parsed_args: Any, **kwargs: Any) -> can.BusABC:
logging_level_names = ["critical", "error", "warning", "info", "debug", "subdebug"]
can.set_logging_level(logging_level_names[min(5, parsed_args.verbosity)])

Expand All @@ -99,7 +99,7 @@ def _create_bus(parsed_args: Any, **kwargs: Any) -> can.Bus:
if parsed_args.data_bitrate:
config["data_bitrate"] = parsed_args.data_bitrate

return Bus(parsed_args.channel, **config) # type: ignore
return Bus(parsed_args.channel, **config)


def _parse_filters(parsed_args: Any) -> CanFilters:
Expand Down
86 changes: 47 additions & 39 deletions can/util.py
Original file line number Diff line number Diff line change
Expand Up @@ -24,6 +24,8 @@
cast,
)

from typing_extensions import ParamSpec

import can

from . import typechecking
Expand Down Expand Up @@ -325,9 +327,15 @@ def channel2int(channel: Optional[typechecking.Channel]) -> Optional[int]:
return None


def deprecated_args_alias( # type: ignore
deprecation_start: str, deprecation_end: Optional[str] = None, **aliases
):
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)

Expand Down Expand Up @@ -356,9 +364,9 @@ def library_function(new_arg):

"""

def deco(f):
def deco(f: Callable[P1, T1]) -> Callable[P1, T1]:
@functools.wraps(f)
def wrapper(*args, **kwargs):
def wrapper(*args: P1.args, **kwargs: P1.kwargs) -> T1:
_rename_kwargs(
func_name=f.__name__,
start=deprecation_start,
Expand All @@ -373,10 +381,42 @@ def wrapper(*args, **kwargs):
return deco


T = TypeVar("T", BitTiming, BitTimingFd)
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: T, valid_clocks: Iterable[int]) -> T:

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.
Expand Down Expand Up @@ -416,38 +456,6 @@ def check_or_adjust_timing_clock(timing: T, valid_clocks: Iterable[int]) -> T:
) from None


def _rename_kwargs(
func_name: str,
start: str,
end: Optional[str],
kwargs: Dict[str, str],
aliases: Dict[str, 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)


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

Expand Down
13 changes: 7 additions & 6 deletions doc/bus.rst
Original file line number Diff line number Diff line change
Expand Up @@ -3,10 +3,10 @@
Bus
---

The :class:`~can.Bus` provides a wrapper around a physical or virtual CAN Bus.
The :class:`~can.BusABC` class provides a wrapper around a physical or virtual CAN Bus.

An interface specific instance is created by instantiating the :class:`~can.Bus`
class with a particular ``interface``, for example::
An interface specific instance is created by calling the :func:`~can.Bus`
function with a particular ``interface``, for example::

vector_bus = can.Bus(interface='vector', ...)

Expand Down Expand Up @@ -77,13 +77,14 @@ See :meth:`~can.BusABC.set_filters` for the implementation.
Bus API
'''''''

.. autoclass:: can.Bus
.. autofunction:: can.Bus

.. autoclass:: can.BusABC
:class-doc-from: class
:show-inheritance:
:members:
:inherited-members:

.. autoclass:: can.bus.BusState
.. autoclass:: can.BusState
:members:
:undoc-members:

Expand Down
4 changes: 3 additions & 1 deletion doc/conf.py
Original file line number Diff line number Diff line change
Expand Up @@ -126,7 +126,9 @@
("py:class", "can.typechecking.CanFilter"),
("py:class", "can.typechecking.CanFilterExtended"),
("py:class", "can.typechecking.AutoDetectedConfig"),
("py:class", "can.util.T"),
("py:class", "can.util.T1"),
("py:class", "can.util.T2"),
("py:class", "~P1"),
# intersphinx fails to reference some builtins
("py:class", "asyncio.events.AbstractEventLoop"),
("py:class", "_thread.allocate_lock"),
Expand Down
25 changes: 17 additions & 8 deletions doc/internal-api.rst
Original file line number Diff line number Diff line change
Expand Up @@ -57,23 +57,32 @@ They **might** implement the following:
and thus might not provide message filtering:


Concrete instances are usually created by :class:`can.Bus` which takes the users
Concrete instances are usually created by :func:`can.Bus` which takes the users
configuration into account.


Bus Internals
~~~~~~~~~~~~~

Several methods are not documented in the main :class:`can.Bus`
Several methods are not documented in the main :class:`can.BusABC`
as they are primarily useful for library developers as opposed to
library users. This is the entire ABC bus class with all internal
methods:
library users.

.. autoclass:: can.BusABC
:members:
:private-members:
:special-members:
.. automethod:: can.BusABC.__init__

.. automethod:: can.BusABC.__iter__

.. automethod:: can.BusABC.__str__

.. autoattribute:: can.BusABC.__weakref__

.. automethod:: can.BusABC._recv_internal

.. automethod:: can.BusABC._apply_filters

.. automethod:: can.BusABC._send_periodic_internal

.. automethod:: can.BusABC._detect_available_configs


About the IO module
Expand Down