make mypy happy

Author: zariiii9003

can/ctypesutil.py

@@ -1,7 +1,7 @@
+# type: ignore
 """
 This module contains common `ctypes` utils.
 """
-
 import ctypes
 import logging
 import sys
@@ -14,14 +14,14 @@
 
 
 try:
-    _LibBase = ctypes.WinDLL  # type: ignore
-    _FUNCTION_TYPE = ctypes.WINFUNCTYPE  # type: ignore
+    _LibBase = ctypes.WinDLL
+    _FUNCTION_TYPE = ctypes.WINFUNCTYPE
 except AttributeError:
     _LibBase = ctypes.CDLL
     _FUNCTION_TYPE = ctypes.CFUNCTYPE
 
 
-class CLibrary(_LibBase):  # type: ignore
+class CLibrary(_LibBase):
     def __init__(self, library_or_path: Union[str, ctypes.CDLL]) -> None:
         self.func_name: Any
 

can/exceptions.py

@@ -14,13 +14,17 @@
 For example, validating typical arguments and parameters might result in a
 :class:`ValueError`. This should always be documented for the function at hand.
 """
-
-
+import sys
 from contextlib import contextmanager
 
 from typing import Optional
 from typing import Type
 
+if sys.version_info >= (3, 9):
+    from collections.abc import Generator
+else:
+    from typing import Generator
+
 
 class CanError(Exception):
     """Base class for all CAN related exceptions.
@@ -106,7 +110,7 @@ class CanTimeoutError(CanError, TimeoutError):
 def error_check(
     error_message: Optional[str] = None,
     exception_type: Type[CanError] = CanOperationError,
-) -> None:
+) -> Generator[None, None, None]:
     """Catches any exceptions and turns them into the new type while preserving the stack trace."""
     try:
         yield

can/io/asc.py

@@ -6,7 +6,7 @@
     - under `test/data/logfile.asc`
 """
 import gzip
-from typing import cast, Any, Generator, IO, List, Optional, Dict, Union
+from typing import cast, Any, Generator, IO, List, Optional, Dict, Union, TextIO
 
 from datetime import datetime
 import time
@@ -34,6 +34,8 @@ class ASCReader(BaseIOHandler):
     bus statistics, J1939 Transport Protocol messages) is ignored.
     """
 
+    file: TextIO
+
     FORMAT_START_OF_FILE_DATE = "%a %b %d %I:%M:%S.%f %p %Y"
 
     def __init__(
@@ -205,19 +207,18 @@ def _process_fd_can_frame(self, line: str, msg_kwargs: Dict[str, Any]) -> Messag
         return Message(**msg_kwargs)
 
     def __iter__(self) -> Generator[Message, None, None]:
-        # This is guaranteed to not be None since we raise ValueError in __init__
-        self.file = cast(IO[Any], self.file)
         self._extract_header()
 
         for line in self.file:
             temp = line.strip()
             if not temp or not temp[0].isdigit():
                 # Could be a comment
                 continue
-            msg_kwargs = {}
+
+            msg_kwargs: Dict[str, Union[float, bool, int]] = {}
             try:
-                timestamp, channel, rest_of_message = temp.split(None, 2)
-                timestamp = float(timestamp) + self.start_time
+                _timestamp, channel, rest_of_message = temp.split(None, 2)
+                timestamp = float(_timestamp) + self.start_time
                 msg_kwargs["timestamp"] = timestamp
                 if channel == "CANFD":
                     msg_kwargs["is_fd"] = True
@@ -250,6 +251,8 @@ class ASCWriter(FileIOMessageWriter, Listener):
     It the first message does not have a timestamp, it is set to zero.
     """
 
+    file: TextIO
+
     FORMAT_MESSAGE = "{channel}  {id:<15} {dir:<4} {dtype} {data}"
     FORMAT_MESSAGE_FD = " ".join(
         [
@@ -319,8 +322,6 @@ def log_event(self, message: str, timestamp: Optional[float] = None) -> None:
         if not message:  # if empty or None
             logger.debug("ASCWriter: ignoring empty message")
             return
-        # This is guaranteed to not be None since we raise ValueError in __init__
-        self.file = cast(IO[Any], self.file)
 
         # this is the case for the very first message:
         if not self.header_written:

can/io/blf.py

@@ -17,7 +17,7 @@
 import datetime
 import time
 import logging
-from typing import List
+from typing import List, BinaryIO
 
 from ..message import Message
 from ..listener import Listener
@@ -139,6 +139,8 @@ class BLFReader(BaseIOHandler):
     silently ignored.
     """
 
+    file: BinaryIO
+
     def __init__(self, file: AcceptedIOType) -> None:
         """
         :param file: a path-like object or as file-like object to read from
@@ -352,6 +354,8 @@ class BLFWriter(FileIOMessageWriter, Listener):
     Logs CAN data to a Binary Logging File compatible with Vector's tools.
     """
 
+    file: BinaryIO
+
     #: Max log container size of uncompressed data
     max_container_size = 128 * 1024
 

can/io/canutils.py

@@ -114,7 +114,9 @@ class CanutilsLogWriter(FileIOMessageWriter, Listener):
     It the first message does not have a timestamp, it is set to zero.
     """
 
-    def __init__(self, file: AcceptedIOType, channel="vcan0", append=False):
+    def __init__(
+        self, file: AcceptedIOType, channel: str = "vcan0", append: bool = False
+    ):
         """
         :param file: a path-like object or as file-like object to write to
                      If this is a file-like object, is has to opened in text

can/io/csv.py

@@ -10,6 +10,7 @@
 """
 
 from base64 import b64encode, b64decode
+from typing import TextIO
 
 from can.message import Message
 from can.listener import Listener
@@ -82,6 +83,8 @@ class CSVWriter(FileIOMessageWriter, Listener):
     Each line is terminated with a platform specific line separator.
     """
 
+    file: TextIO
+
     def __init__(self, file: AcceptedIOType, append: bool = False) -> None:
         """
         :param file: a path-like object or a file-like object to write to.

can/io/generic.py

@@ -45,7 +45,10 @@ def __init__(
         else:
             # pylint: disable=consider-using-with
             # file is some path-like object
-            self.file = open(cast(can.typechecking.StringPathLike, file), mode)
+            self.file = cast(
+                can.typechecking.FileLike,
+                open(cast(can.typechecking.StringPathLike, file), mode),
+            )
 
         # for multiple inheritance
         super().__init__()
@@ -80,9 +83,7 @@ class FileIOMessageWriter(MessageWriter, metaclass=ABCMeta):
 
     file: Union[TextIO, BinaryIO]
 
-    def __init__(
-        self, file: Union[can.typechecking.FileLike, TextIO, BinaryIO], mode: str = "rt"
-    ) -> None:
+    def __init__(self, file: can.typechecking.AcceptedIOType, mode: str = "rt") -> None:
         # Not possible with the type signature, but be verbose for user friendliness
         if file is None:
             raise ValueError("The given file cannot be None")

can/io/logger.py

@@ -15,7 +15,7 @@
 )
 from types import TracebackType
 
-
+from typing_extensions import Literal
 from pkg_resources import iter_entry_points
 
 from ..message import Message
@@ -95,6 +95,9 @@ def __new__(  # type: ignore
                 f'No write support for this unknown log format "{suffix}"'
             ) from None
 
+    def on_message_received(self, msg: Message) -> None:
+        pass
+
 
 class BaseRotatingLogger(Listener, BaseIOHandler, ABC):
     """
@@ -232,7 +235,7 @@ def __exit__(
         exc_type: Optional[Type[BaseException]],
         exc_val: Optional[BaseException],
         exc_tb: Optional[TracebackType],
-    ) -> bool:
+    ) -> Literal[False]:
         return self._writer.__exit__(exc_type, exc_val, exc_tb)
 
     @abstractmethod

can/io/printer.py

@@ -4,7 +4,7 @@
 
 import logging
 
-from typing import Optional
+from typing import Optional, cast, TextIO
 
 from ..message import Message
 from ..listener import Listener
@@ -24,6 +24,8 @@ class Printer(BaseIOHandler, Listener):
                          standard out
     """
 
+    file: Optional[TextIO]
+
     def __init__(
         self, file: Optional[AcceptedIOType] = None, append: bool = False
     ) -> None:
@@ -35,12 +37,11 @@ def __init__(
         :param append: If set to `True` messages, are appended to the file,
                        else the file is truncated
         """
-        self.write_to_file = file is not None
         mode = "a" if append else "w"
         super().__init__(file, mode=mode)
 
     def on_message_received(self, msg: Message) -> None:
-        if self.write_to_file:
+        if self.file is not None:
             self.file.write(str(msg) + "\n")
         else:
             print(msg)

can/typechecking.py

@@ -27,7 +27,7 @@
 Channel = typing.Union[ChannelInt, ChannelStr]
 
 # Used by the IO module
-FileLike = typing.IO[typing.Any]
+FileLike = typing.Union[typing.TextIO, typing.BinaryIO]
 StringPathLike = typing.Union[str, "os.PathLike[str]"]
 AcceptedIOType = typing.Union[FileLike, StringPathLike]