Raise minimum python version to 3.9

.github/workflows/ci.yml

@@ -18,7 +18,6 @@ jobs:
         os: [ubuntu-latest, macos-latest, windows-latest]
         experimental: [false]
         python-version: [
-          "3.8",
           "3.9",
           "3.10",
           "3.11",
@@ -81,9 +80,6 @@ jobs:
       run: |
         python -m pip install --upgrade pip
         pip install -e .[lint]
-    - name: mypy 3.8
-      run: |
-        mypy --python-version 3.8 .
     - name: mypy 3.9
       run: |
         mypy --python-version 3.9 .

README.rst

@@ -3,7 +3,7 @@ python-can
 
 |pypi| |conda| |python_implementation| |downloads| |downloads_monthly|
 
-|docs| |github-actions| |coverage| |mergify| |formatter|
+|docs| |github-actions| |coverage| |formatter|
 
 .. |pypi| image:: https://img.shields.io/pypi/v/python-can.svg
    :target: https://pypi.python.org/pypi/python-can/
@@ -41,10 +41,6 @@ python-can
    :target: https://coveralls.io/github/hardbyte/python-can?branch=develop
    :alt: Test coverage reports on Coveralls.io
 
-.. |mergify| image:: https://img.shields.io/endpoint.svg?url=https://api.mergify.com/v1/badges/hardbyte/python-can&style=flat
-   :target: https://mergify.io
-   :alt: Mergify Status
-
 The **C**\ ontroller **A**\ rea **N**\ etwork is a bus standard designed
 to allow microcontrollers and devices to communicate with each other. It
 has priority based bus arbitration and reliable deterministic
@@ -64,6 +60,7 @@ Library Version                 Python
   3.x                           2.7+, 3.5+
   4.0+                          3.7+
   4.3+                          3.8+
+  4.6+                          3.9+
 ==============================  ===========
 
 

can/__init__.py

@@ -8,7 +8,7 @@
 import contextlib
 import logging
 from importlib.metadata import PackageNotFoundError, version
-from typing import Any, Dict
+from typing import Any
 
 __all__ = [
     "VALID_INTERFACES",
@@ -130,4 +130,4 @@
 
 log = logging.getLogger("can")
 
-rc: Dict[str, Any] = {}
+rc: dict[str, Any] = {}

can/_entry_points.py

@@ -2,7 +2,7 @@
 import sys
 from dataclasses import dataclass
 from importlib.metadata import entry_points
-from typing import Any, List
+from typing import Any
 
 
 @dataclass
@@ -20,14 +20,14 @@ def load(self) -> Any:
 # "Compatibility Note".
 if sys.version_info >= (3, 10):
 
-    def read_entry_points(group: str) -> List[_EntryPoint]:
+    def read_entry_points(group: str) -> list[_EntryPoint]:
         return [
             _EntryPoint(ep.name, ep.module, ep.attr) for ep in entry_points(group=group)
         ]
 
 else:
 
-    def read_entry_points(group: str) -> List[_EntryPoint]:
+    def read_entry_points(group: str) -> list[_EntryPoint]:
         return [
             _EntryPoint(ep.name, *ep.value.split(":", maxsplit=1))
             for ep in entry_points().get(group, [])

can/bit_timing.py

@@ -1,6 +1,7 @@
 # pylint: disable=too-many-lines
 import math
-from typing import TYPE_CHECKING, Iterator, List, Mapping, cast
+from collections.abc import Iterator, Mapping
+from typing import TYPE_CHECKING, cast
 
 if TYPE_CHECKING:
     from can.typechecking import BitTimingDict, BitTimingFdDict
@@ -286,7 +287,7 @@ def from_sample_point(
         if sample_point < 50.0:
             raise ValueError(f"sample_point (={sample_point}) must not be below 50%.")
 
-        possible_solutions: List[BitTiming] = list(
+        possible_solutions: list[BitTiming] = list(
             cls.iterate_from_sample_point(f_clock, bitrate, sample_point)
         )
 
@@ -874,7 +875,7 @@ def from_sample_point(
                 f"data_sample_point (={data_sample_point}) must not be below 50%."
             )
 
-        possible_solutions: List[BitTimingFd] = list(
+        possible_solutions: list[BitTimingFd] = list(
             cls.iterate_from_sample_point(
                 f_clock,
                 nom_bitrate,

can/broadcastmanager.py

@@ -12,13 +12,12 @@
 import threading
 import time
 import warnings
+from collections.abc import Sequence
 from typing import (
     TYPE_CHECKING,
     Callable,
     Final,
     Optional,
-    Sequence,
-    Tuple,
     Union,
     cast,
 )
@@ -127,7 +126,7 @@ def __init__(
     @staticmethod
     def _check_and_convert_messages(
         messages: Union[Sequence[Message], Message],
-    ) -> Tuple[Message, ...]:
+    ) -> tuple[Message, ...]:
         """Helper function to convert a Message or Sequence of messages into a
         tuple, and raises an error when the given value is invalid.
 
@@ -194,7 +193,7 @@ def start(self) -> None:
 
 
 class ModifiableCyclicTaskABC(CyclicSendTaskABC, abc.ABC):
-    def _check_modified_messages(self, messages: Tuple[Message, ...]) -> None:
+    def _check_modified_messages(self, messages: tuple[Message, ...]) -> None:
         """Helper function to perform error checking when modifying the data in
         the cyclic task.
 

can/bus.py

@@ -6,18 +6,14 @@
 import logging
 import threading
 from abc import ABC, ABCMeta, abstractmethod
+from collections.abc import Iterator, Sequence
 from enum import Enum, auto
 from time import time
 from types import TracebackType
 from typing import (
     Any,
     Callable,
-    Iterator,
-    List,
     Optional,
-    Sequence,
-    Tuple,
-    Type,
     Union,
     cast,
 )
@@ -97,7 +93,7 @@ def __init__(
         :raises ~can.exceptions.CanInitializationError:
             If the bus cannot be initialized
         """
-        self._periodic_tasks: List[_SelfRemovingCyclicTask] = []
+        self._periodic_tasks: list[_SelfRemovingCyclicTask] = []
         self.set_filters(can_filters)
         # Flip the class default value when the constructor finishes.  That
         # usually means the derived class constructor was also successful,
@@ -147,7 +143,7 @@ def recv(self, timeout: Optional[float] = None) -> Optional[Message]:
 
     def _recv_internal(
         self, timeout: Optional[float]
-    ) -> Tuple[Optional[Message], bool]:
+    ) -> tuple[Optional[Message], bool]:
         """
         Read a message from the bus and tell whether it was filtered.
         This methods may be called by :meth:`~can.BusABC.recv`
@@ -491,7 +487,7 @@ def __enter__(self) -> Self:
 
     def __exit__(
         self,
-        exc_type: Optional[Type[BaseException]],
+        exc_type: Optional[type[BaseException]],
         exc_value: Optional[BaseException],
         traceback: Optional[TracebackType],
     ) -> None:
@@ -529,7 +525,7 @@ def protocol(self) -> CanProtocol:
         return self._can_protocol
 
     @staticmethod
-    def _detect_available_configs() -> List[can.typechecking.AutoDetectedConfig]:
+    def _detect_available_configs() -> list[can.typechecking.AutoDetectedConfig]:
         """Detect all configurations/channels that this interface could
         currently connect with.
 

can/ctypesutil.py

@@ -5,7 +5,7 @@
 import ctypes
 import logging
 import sys
-from typing import Any, Callable, Optional, Tuple, Union
+from typing import Any, Callable, Optional, Union
 
 log = logging.getLogger("can.ctypesutil")
 
@@ -32,7 +32,7 @@ def map_symbol(
         self,
         func_name: str,
         restype: Any = None,
-        argtypes: Tuple[Any, ...] = (),
+        argtypes: tuple[Any, ...] = (),
         errcheck: Optional[Callable[..., Any]] = None,
     ) -> Any:
         """

can/exceptions.py

@@ -15,14 +15,9 @@
 :class:`ValueError`. This should always be documented for the function at hand.
 """
 
-import sys
+from collections.abc import Generator
 from contextlib import contextmanager
-from typing import Optional, Type
-
-if sys.version_info >= (3, 9):
-    from collections.abc import Generator
-else:
-    from typing import Generator
+from typing import Optional
 
 
 class CanError(Exception):
@@ -114,7 +109,7 @@ class CanTimeoutError(CanError, TimeoutError):
 @contextmanager
 def error_check(
     error_message: Optional[str] = None,
-    exception_type: Type[CanError] = CanOperationError,
+    exception_type: type[CanError] = CanOperationError,
 ) -> Generator[None, None, None]:
     """Catches any exceptions and turns them into the new type while preserving the stack trace."""
     try:

can/interface.py

@@ -6,7 +6,8 @@
 
 import importlib
 import logging
-from typing import Any, Iterable, List, Optional, Type, Union, cast
+from collections.abc import Iterable
+from typing import Any, Optional, Union, cast
 
 from . import util
 from .bus import BusABC
@@ -18,7 +19,7 @@
 log_autodetect = log.getChild("detect_available_configs")
 
 
-def _get_class_for_interface(interface: str) -> Type[BusABC]:
+def _get_class_for_interface(interface: str) -> type[BusABC]:
     """
     Returns the main bus class for the given interface.
 
@@ -52,7 +53,7 @@ def _get_class_for_interface(interface: str) -> Type[BusABC]:
             f"'{interface}': {e}"
         ) from None
 
-    return cast("Type[BusABC]", bus_class)
+    return cast("type[BusABC]", bus_class)
 
 
 @util.deprecated_args_alias(
@@ -139,7 +140,7 @@ def Bus(  # noqa: N802
 
 def detect_available_configs(
     interfaces: Union[None, str, Iterable[str]] = None,
-) -> List[AutoDetectedConfig]:
+) -> list[AutoDetectedConfig]:
     """Detect all configurations/channels that the interfaces could
     currently connect with.
 

can/interfaces/__init__.py

@@ -2,8 +2,6 @@
 Interfaces contain low level implementations that interact with CAN hardware.
 """
 
-from typing import Dict, Tuple
-
 from can._entry_points import read_entry_points
 
 __all__ = [
@@ -35,7 +33,7 @@
 ]
 
 # interface_name => (module, classname)
-BACKENDS: Dict[str, Tuple[str, str]] = {
+BACKENDS: dict[str, tuple[str, str]] = {
     "kvaser": ("can.interfaces.kvaser", "KvaserBus"),
     "socketcan": ("can.interfaces.socketcan", "SocketcanBus"),
     "serial": ("can.interfaces.serial.serial_can", "SerialBus"),

can/interfaces/canalystii.py

@@ -1,8 +1,9 @@
 import logging
 import time
 from collections import deque
+from collections.abc import Sequence
 from ctypes import c_ubyte
-from typing import Any, Deque, Dict, Optional, Sequence, Tuple, Union
+from typing import Any, Optional, Union
 
 import canalystii as driver
 
@@ -26,7 +27,7 @@ def __init__(
         timing: Optional[Union[BitTiming, BitTimingFd]] = None,
         can_filters: Optional[CanFilters] = None,
         rx_queue_size: Optional[int] = None,
-        **kwargs: Dict[str, Any],
+        **kwargs: dict[str, Any],
     ):
         """
 
@@ -68,7 +69,7 @@ def __init__(
             self.channels = list(channel)
 
         self.channel_info = f"CANalyst-II: device {device}, channels {self.channels}"
-        self.rx_queue: Deque[Tuple[int, driver.Message]] = deque(maxlen=rx_queue_size)
+        self.rx_queue: deque[tuple[int, driver.Message]] = deque(maxlen=rx_queue_size)
         self.device = driver.CanalystDevice(device_index=device)
         self._can_protocol = CanProtocol.CAN_20
 
@@ -129,7 +130,7 @@ def send(self, msg: Message, timeout: Optional[float] = None) -> None:
         if timeout is not None and not send_result:
             raise CanTimeoutError(f"Send timed out after {timeout} seconds")
 
-    def _recv_from_queue(self) -> Tuple[Message, bool]:
+    def _recv_from_queue(self) -> tuple[Message, bool]:
         """Return a message from the internal receive queue"""
         channel, raw_msg = self.rx_queue.popleft()
 
@@ -166,7 +167,7 @@ def poll_received_messages(self) -> None:
 
     def _recv_internal(
         self, timeout: Optional[float] = None
-    ) -> Tuple[Optional[Message], bool]:
+    ) -> tuple[Optional[Message], bool]:
         """
 
         :param timeout: float in seconds

can/interfaces/etas/__init__.py

@@ -1,5 +1,5 @@
 import time
-from typing import Dict, List, Optional, Tuple
+from typing import Optional
 
 import can
 from can.exceptions import CanInitializationError
@@ -16,7 +16,7 @@ def __init__(
         bitrate: int = 1000000,
         fd: bool = True,
         data_bitrate: int = 2000000,
-        **kwargs: Dict[str, any],
+        **kwargs: dict[str, any],
     ):
         self.receive_own_messages = receive_own_messages
         self._can_protocol = can.CanProtocol.CAN_FD if fd else can.CanProtocol.CAN_20
@@ -122,7 +122,7 @@ def __init__(
 
     def _recv_internal(
         self, timeout: Optional[float]
-    ) -> Tuple[Optional[can.Message], bool]:
+    ) -> tuple[Optional[can.Message], bool]:
         ociMsgs = (ctypes.POINTER(OCI_CANMessageEx) * 1)()
         ociMsg = OCI_CANMessageEx()
         ociMsgs[0] = ctypes.pointer(ociMsg)
@@ -295,12 +295,12 @@ def state(self, new_state: can.BusState) -> None:
         raise NotImplementedError("Setting state is not implemented.")
 
     @staticmethod
-    def _detect_available_configs() -> List[can.typechecking.AutoDetectedConfig]:
+    def _detect_available_configs() -> list[can.typechecking.AutoDetectedConfig]:
         nodeRange = CSI_NodeRange(CSI_NODE_MIN, CSI_NODE_MAX)
         tree = ctypes.POINTER(CSI_Tree)()
         CSI_CreateProtocolTree(ctypes.c_char_p(b""), nodeRange, ctypes.byref(tree))
 
-        nodes: List[Dict[str, str]] = []
+        nodes: list[dict[str, str]] = []
 
         def _findNodes(tree, prefix):
             uri = f"{prefix}/{tree.contents.item.uriName.decode()}"