A few more type hints (#1556)

Co-authored-by: jan iversen <[email protected]>

Author: alexrudd2

API_changes.rst

@@ -10,6 +10,7 @@ Version 3.3.0
 - move SqlSlaveContext, RedisSlaveContext to examples/contrib (due to lack of maintenance)
 - :code:`BinaryPayloadBuilder.to_string` was renamed to :code:`BinaryPayloadBuilder.encode`
 - on_reconnect_callback for async clients works slightly different
+- utilities/unpack_bitstring now expects an argument named `data` not `string`
 
 -------------
 Version 3.2.0

pymodbus/device.py

@@ -464,7 +464,7 @@ class ModbusControlBlock:
     __mode = "ASCII"
     __diagnostic = [False] * 16
     __listen_only = False
-    __delimiter = "\r"
+    __delimiter = b"\r"
     __counters = ModbusCountersHandler()
     __identity = ModbusDeviceIdentification()
     __plus = ModbusPlusStatistics()

pymodbus/events.py

@@ -17,15 +17,15 @@ def encode(self):
 
         :raises NotImplementedException:
         """
-        raise NotImplementedException()
+        raise NotImplementedException
 
     def decode(self, event):
         """Decode the event message to its status bits.
 
         :param event: The event to decode
         :raises NotImplementedException:
         """
-        raise NotImplementedException()
+        raise NotImplementedException
 
 
 class RemoteReceiveEvent(ModbusEvent):
@@ -54,7 +54,7 @@ def __init__(self, **kwargs):
         self.listen = kwargs.get("listen", False)
         self.broadcast = kwargs.get("broadcast", False)
 
-    def encode(self):
+    def encode(self) -> bytes:
         """Encode the status bits to an event message.
 
         :returns: The encoded event message
@@ -64,7 +64,7 @@ def encode(self):
         packet = pack_bitstring(bits)
         return packet
 
-    def decode(self, event):
+    def decode(self, event: bytes) -> None:
         """Decode the event message to its status bits.
 
         :param event: The event to decode

pymodbus/framer/base.py

@@ -1,5 +1,8 @@
 """Framer start."""
 # pylint: disable=missing-type-doc
+from typing import Any, Dict, Union
+
+from pymodbus.factory import ClientDecoder, ServerDecoder
 
 
 # Unit ID, Function Code
@@ -18,15 +21,20 @@ class ModbusFramer:
 
     name = ""
 
-    def __init__(self, decoder, client=None):
+    def __init__(
+        self,
+        decoder: Union[ClientDecoder, ServerDecoder],
+        client=None,
+    ) -> None:
         """Initialize a new instance of the framer.
 
         :param decoder: The decoder implementation to use
         """
         self.decoder = decoder
         self.client = client
+        self._header: Dict[str, Any] = {}
 
-    def _validate_slave_id(self, slaves, single):
+    def _validate_slave_id(self, slaves: list, single: bool) -> bool:
         """Validate if the received data is valid for the client.
 
         :param slaves: list of slave id for which the transaction is valid
@@ -39,7 +47,7 @@ def _validate_slave_id(self, slaves, single):
             # Handle Modbus TCP slave identifier (0x00 0r 0xFF)
             # in asynchronous requests
             return True
-        return self._header["uid"] in slaves  # pylint: disable=no-member
+        return self._header["uid"] in slaves
 
     def sendPacket(self, message):
         """Send packets on the bus.

pymodbus/framer/binary_framer.py

@@ -65,7 +65,7 @@ def decode_data(self, data):
             return {"slave": uid, "fcode": fcode}
         return {}
 
-    def checkFrame(self):
+    def checkFrame(self) -> bool:
         """Check and decode the next frame.
 
         :returns: True if we are successful, False otherwise
@@ -84,7 +84,7 @@ def checkFrame(self):
             return checkCRC(data, self._header["crc"])
         return False
 
-    def advanceFrame(self):
+    def advanceFrame(self) -> None:
         """Skip over the current framed message.
 
         This allows us to skip over the current message after we have processed
@@ -94,7 +94,7 @@ def advanceFrame(self):
         self._buffer = self._buffer[self._header["len"] + 2 :]
         self._header = {"crc": 0x0000, "len": 0, "uid": 0x00}
 
-    def isFrameReady(self):
+    def isFrameReady(self) -> bool:
         """Check if we should continue decode logic.
 
         This is meant to be used in a while loop in the decoding phase to let

pymodbus/utilities.py

@@ -17,6 +17,7 @@
 
 # pylint: disable=missing-type-doc
 import struct
+from typing import List
 
 
 class ModbusTransactionState:  # pylint: disable=too-few-public-methods
@@ -106,10 +107,10 @@ def dict_property(store, index):
 # --------------------------------------------------------------------------- #
 # Bit packing functions
 # --------------------------------------------------------------------------- #
-def pack_bitstring(bits):
-    """Create a string out of an array of bits.
+def pack_bitstring(bits: List[bool]) -> bytes:
+    """Create a bytestring out of a list of bits.
 
-    :param bits: A bit array
+    :param bits: A list of bits
 
     example::
 
@@ -133,20 +134,20 @@ def pack_bitstring(bits):
     return ret
 
 
-def unpack_bitstring(string):
-    """Create bit array out of a string.
+def unpack_bitstring(data: bytes) -> List[bool]:
+    """Create bit list out of a bytestring.
 
-    :param string: The modbus data packet to decode
+    :param data: The modbus data packet to decode
 
     example::
 
         bytes  = "bytes to decode"
         result = unpack_bitstring(bytes)
     """
-    byte_count = len(string)
+    byte_count = len(data)
     bits = []
     for byte in range(byte_count):
-        value = int(int(string[byte]))
+        value = int(int(data[byte]))
         for _ in range(8):
             bits.append((value & 1) == 1)
             value >>= 1