diff --git a/can/broadcastmanager.py b/can/broadcastmanager.py index 90cf0ce60..c15186e2c 100644 --- a/can/broadcastmanager.py +++ b/can/broadcastmanager.py @@ -19,7 +19,7 @@ import threading import time -# try to import win32event for event-based cyclic send task(needs pywin32 package) +# try to import win32event for event-based cyclic send task (needs the pywin32 package) try: import win32event @@ -260,7 +260,7 @@ def stop(self) -> None: def start(self) -> None: self.stopped = False if self.thread is None or not self.thread.is_alive(): - name = "Cyclic send task for 0x%X" % (self.messages[0].arbitration_id) + name = f"Cyclic send task for 0x{self.messages[0].arbitration_id:X}" self.thread = threading.Thread(target=self._run, name=name) self.thread.daemon = True diff --git a/can/exceptions.py b/can/exceptions.py index aec0dfd1d..5a7aa0b7c 100644 --- a/can/exceptions.py +++ b/can/exceptions.py @@ -14,6 +14,7 @@ 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 @@ -114,7 +115,7 @@ def error_check( """Catches any exceptions and turns them into the new type while preserving the stack trace.""" try: yield - except Exception as error: + except Exception as error: # pylint: disable=broad-except if error_message is None: raise exception_type(str(error)) from error else: diff --git a/can/interface.py b/can/interface.py index 5282d77bf..e217f2fb6 100644 --- a/can/interface.py +++ b/can/interface.py @@ -32,7 +32,7 @@ def _get_class_for_interface(interface: str) -> Type[BusABC]: module_name, class_name = BACKENDS[interface] except KeyError: raise NotImplementedError( - "CAN interface '{}' not supported".format(interface) + f"CAN interface '{interface}' not supported" ) from None # Import the correct interface module @@ -40,9 +40,7 @@ def _get_class_for_interface(interface: str) -> Type[BusABC]: module = importlib.import_module(module_name) except Exception as e: raise CanInterfaceNotImplementedError( - "Cannot import module {} for CAN interface '{}': {}".format( - module_name, interface, e - ) + f"Cannot import module {module_name} for CAN interface '{interface}': {e}" ) from None # Get the correct class @@ -50,9 +48,8 @@ def _get_class_for_interface(interface: str) -> Type[BusABC]: bus_class = getattr(module, class_name) except Exception as e: raise CanInterfaceNotImplementedError( - "Cannot import class {} from module {} for CAN interface '{}': {}".format( - class_name, module_name, interface, e - ) + f"Cannot import class {class_name} from module {module_name} for CAN interface " + f"'{interface}': {e}" ) from None return cast(Type[BusABC], bus_class) diff --git a/can/listener.py b/can/listener.py index 8b90fc79e..12836a83c 100644 --- a/can/listener.py +++ b/can/listener.py @@ -58,7 +58,7 @@ def stop(self) -> None: """ -class RedirectReader(Listener): +class RedirectReader(Listener): # pylint: disable=abstract-method """ A RedirectReader sends all received messages to another Bus. """ @@ -71,7 +71,7 @@ def on_message_received(self, msg: Message) -> None: self.bus.send(msg) -class BufferedReader(Listener): +class BufferedReader(Listener): # pylint: disable=abstract-method """ A BufferedReader is a subclass of :class:`~can.Listener` which implements a **message buffer**: that is, when the :class:`can.BufferedReader` instance is @@ -126,7 +126,7 @@ def stop(self) -> None: self.is_stopped = True -class AsyncBufferedReader(Listener): +class AsyncBufferedReader(Listener): # pylint: disable=abstract-method """A message buffer for use with :mod:`asyncio`. See :ref:`asyncio` for how to use with :class:`can.Notifier`. diff --git a/can/logconvert.py b/can/logconvert.py index 6a2f52341..730e82304 100644 --- a/can/logconvert.py +++ b/can/logconvert.py @@ -12,7 +12,7 @@ class ArgumentParser(argparse.ArgumentParser): def error(self, message): self.print_help(sys.stderr) - self.exit(errno.EINVAL, "%s: error: %s\n" % (self.prog, message)) + self.exit(errno.EINVAL, f"{self.prog}: error: {message}\n") def main(): diff --git a/can/viewer.py b/can/viewer.py index b74e954a0..9cd5246fb 100644 --- a/can/viewer.py +++ b/can/viewer.py @@ -29,7 +29,6 @@ import time from typing import Dict, List, Tuple, Union -import can from can import __version__ from .logger import ( _create_bus, @@ -53,7 +52,7 @@ curses = None # type: ignore -class CanViewer: +class CanViewer: # pylint: disable=too-many-instance-attributes def __init__(self, stdscr, bus, data_structs, testing=False): self.stdscr = stdscr self.bus = bus @@ -89,7 +88,7 @@ def run(self): # Clear the terminal and draw the header self.draw_header() - while 1: + while True: # Do not read the CAN-Bus when in paused mode if not self.paused: # Read the CAN-Bus and draw it in the terminal window @@ -237,8 +236,11 @@ def draw_can_bus_message(self, msg, sorting=False): self.ids[key]["count"] += 1 # Format the CAN-Bus ID as a hex value - arbitration_id_string = "0x{0:0{1}X}".format( - msg.arbitration_id, 8 if msg.is_extended_id else 3 + arbitration_id_string = ( + "0x{0:0{1}X}".format( # pylint: disable=consider-using-f-string + msg.arbitration_id, + 8 if msg.is_extended_id else 3, + ) ) # Use red for error frames @@ -263,7 +265,7 @@ def draw_can_bus_message(self, msg, sorting=False): previous_byte_values = self.previous_values[key] except KeyError: # no row of previous values exists for the current message ID # initialise a row to store the values for comparison next time - self.previous_values[key] = dict() + self.previous_values[key] = {} previous_byte_values = self.previous_values[key] for i, b in enumerate(msg.data): col = 52 + i * 3 @@ -279,7 +281,7 @@ def draw_can_bus_message(self, msg, sorting=False): else: data_color = color except KeyError: - # previous entry for byte didnt exist - default to rest of line colour + # previous entry for byte didn't exist - default to rest of line colour data_color = color finally: # write the new value to the previous values dict for next time @@ -336,7 +338,7 @@ def draw_header(self): def redraw_screen(self): # Trigger a complete redraw self.draw_header() - for key, ids in self.ids.items(): + for ids in self.ids.values(): self.draw_can_bus_message(ids["msg"]) @@ -545,7 +547,6 @@ def main() -> None: if can_filters: additional_config.update({"can_filters": can_filters}) bus = _create_bus(parsed_args, **additional_config) - # print(f"Connected to {bus.__class__.__name__}: {bus.channel_info}") curses.wrapper(CanViewer, bus, data_structs)