Implement pyflakes-simplify

examples/contrib/redis_datastore.py

@@ -1,11 +1,10 @@
 """Datastore using redis."""
 # pylint: disable=missing-type-doc
+from contextlib import suppress
 
 
-try:
+with suppress(ImportError):
     import redis
-except ImportError:
-    pass
 
 from pymodbus.datastore import ModbusBaseSlaveContext
 from pymodbus.logging import Log

pymodbus/client/serial.py

@@ -1,6 +1,7 @@
 """Modbus client async serial communication."""
 import asyncio
 import time
+from contextlib import suppress
 from functools import partial
 from typing import Any, Type
 
@@ -14,10 +15,8 @@
 from pymodbus.utilities import ModbusTransactionState
 
 
-try:
+with suppress(ImportError):
     import serial
-except ImportError:
-    pass
 
 
 class AsyncModbusSerialClient(ModbusBaseClient, asyncio.Protocol):

pymodbus/server/async_io.py

@@ -4,6 +4,7 @@
 import ssl
 import time
 import traceback
+from contextlib import suppress
 from typing import Union
 
 from pymodbus.client.serial_asyncio import create_serial_connection
@@ -22,10 +23,8 @@
 )
 
 
-try:
+with suppress(ImportError):
     import serial
-except ImportError:
-    pass
 
 
 def sslctx_provider(
@@ -1072,10 +1071,8 @@ async def run(cls, server, custom_functions):
         for func in custom_functions:
             server.decoder.register(func)
         cls.active_server = _serverList(server)
-        try:
+        with suppress(asyncio.exceptions.CancelledError):
             await server.serve_forever()
-        except asyncio.CancelledError:
-            pass
 
     @classmethod
     async def async_stop(cls):

pymodbus/server/simulator/http_server.py

@@ -598,7 +598,7 @@ def server_response_manipulator(self, response):
             self.call_list.append(tracer)
             self.call_monitor.trace_response = False
 
-        if not self.call_response.active == RESPONSE_INACTIVE:
+        if self.call_response.active != RESPONSE_INACTIVE:
             return response, False
 
         skip_encoding = False

pymodbus/transaction.py

@@ -13,6 +13,7 @@
 # pylint: disable=missing-type-doc
 import struct
 import time
+from contextlib import suppress
 from functools import partial
 from threading import RLock
 
@@ -374,15 +375,14 @@ def _recv(self, expected_response_length, full):  # pylint: disable=too-complex
                     elif expected_response_length is None and isinstance(
                         self.client.framer, ModbusRtuFramer
                     ):
-                        try:
+                        with suppress(
+                            IndexError  # response length indeterminate with available bytes
+                        ):
                             expected_response_length = (
                                 self.client.framer.get_expected_response_length(
                                     read_min
                                 )
                             )
-                        except IndexError:
-                            # Could not determine response length with available bytes
-                            pass
                     if expected_response_length is not None:
                         expected_response_length -= min_size
                         total = expected_response_length + min_size

ruff.toml

@@ -18,6 +18,13 @@ select = [
     "E",
     "F",
     "W",
+    "SIM105",
+    "SIM117",
+    "SIM118",
+    "SIM201",
+    "SIM212",
+    "SIM300",
+    "SIM401",
 ]
 [pydocstyle]
 convention = "pep257"

test/test_client.py

@@ -258,9 +258,9 @@ async def test_client_instanciate(
     assert client.delay_ms == initial_delay
 
     rc1 = client._get_address_family("127.0.0.1")  # pylint: disable=protected-access
-    assert socket.AF_INET == rc1
+    assert rc1 == socket.AF_INET
     rc2 = client._get_address_family("::1")  # pylint: disable=protected-access
-    assert socket.AF_INET6 == rc2
+    assert rc2 == socket.AF_INET6
 
     # a successful execute
     client.connect = lambda: True

test/test_repl_client.py

@@ -1,4 +1,6 @@
 """Test client sync."""
+from contextlib import suppress
+
 from pymodbus.repl.client.main import _process_args
 from pymodbus.server.reactive.default_config import DEFAULT_CONFIG
 
@@ -32,17 +34,11 @@ def test_repl_client_process_args():
     resp = _process_args(["address=0b11", "value=0x10"], False)
     assert resp == ({"address": 3, "value": 16}, True)
 
-    try:
+    with suppress(ValueError):
         resp = _process_args(["address=0xhj", "value=0x10"], False)
-    except ValueError:
-        pass
 
-    try:
+    with suppress(ValueError):
         resp = _process_args(["address=11ah", "value=0x10"], False)
-    except ValueError:
-        pass
 
-    try:
+    with suppress(ValueError):
         resp = _process_args(["address=0b12", "value=0x10"], False)
-    except ValueError:
-        pass

test/test_server_asyncio.py

@@ -3,6 +3,7 @@
 import logging
 import ssl
 from asyncio import CancelledError
+from contextlib import suppress
 from unittest import mock
 
 import pytest
@@ -127,19 +128,15 @@ async def setup_teardown(self):
             await asyncio.sleep(0.1)
             if not self.task.cancelled():
                 self.task.cancel()
-                try:
+                with suppress(CancelledError):
                     await self.task
-                except CancelledError:
-                    pass
                 self.task = None
         BasicClient.clear()
 
     def handle_task(self, result):
         """Handle task exit."""
-        try:
+        with suppress(CancelledError):
             result = result.result()
-        except CancelledError:
-            pass
 
     async def start_server(
         self, do_forever=True, do_defer=True, do_tls=False, do_udp=False, do_ident=False
@@ -297,11 +294,10 @@ async def test_async_tls_server_serve_forever(self):
         """Test StartAsyncTcpServer serve_forever() method"""
         with mock.patch(
             "asyncio.base_events.Server.serve_forever", new_callable=mock.AsyncMock
-        ) as serve:
-            with mock.patch.object(ssl.SSLContext, "load_cert_chain"):
-                await self.start_server(do_tls=True, do_forever=False)
-                await self.server.serve_forever()
-                serve.assert_awaited()
+        ) as serve, mock.patch.object(ssl.SSLContext, "load_cert_chain"):
+            await self.start_server(do_tls=True, do_forever=False)
+            await self.server.serve_forever()
+            serve.assert_awaited()
 
     async def test_async_tls_server_serve_forever_twice(self):
         """Call on serve_forever() twice should result in a runtime error"""