Improve can.Bus typing (#1557)

Author: zariiii9003

can/interface.py

@@ -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.
@@ -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(
@@ -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

can/logger.py

@@ -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)])
 
@@ -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:

can/util.py

@@ -24,6 +24,8 @@
     cast,
 )
 
+from typing_extensions import ParamSpec
+
 import can
 
 from . import typechecking
@@ -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)
 
@@ -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,
@@ -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.
@@ -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
 

doc/bus.rst

@@ -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', ...)
 
@@ -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:
 

doc/conf.py

@@ -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"),

doc/internal-api.rst

@@ -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