Restrict write_registers etc to list[int].

examples/client_async_calls.py

@@ -165,10 +165,9 @@ async def async_write_registers_mypy(client: ModbusBaseClient) -> None:
     rr = await client.read_holding_registers(1, count=len(regs1), slave=SLAVE)
     assert not rr.isError()  # test that call was OK
 
-    regs2: list[bytes] = [b'\x01\x02', b'\x03\x04']
-    await client.write_registers(1, regs2, slave=SLAVE)
-    rr = await client.read_holding_registers(1, count=len(regs2), slave=SLAVE)
-    assert not rr.isError()  # test that call was OK
+    # regs2: list[bytes] = [b'\x01\x02', b'\x03\x04']
+    # await client.write_registers(1, regs2, slave=SLAVE)
+    # NOT ALLOWED
 
 
 async def async_handle_input_registers(client):

pymodbus/client/mixin.py

@@ -2,7 +2,6 @@
 from __future__ import annotations
 
 import struct
-from collections.abc import Sequence
 from enum import Enum
 from typing import Generic, TypeVar
 
@@ -159,7 +158,7 @@ def write_coil(self, address: int, value: bool, *, slave: int = 1, no_response_e
         pdu = pdu_bit.WriteSingleCoilRequest(address=address, bits=[value], slave_id=slave)
         return self.execute(no_response_expected, pdu)
 
-    def write_register(self, address: int, value: bytes | int, *, slave: int = 1, no_response_expected: bool = False) -> T:
+    def write_register(self, address: int, value: int, *, slave: int = 1, no_response_expected: bool = False) -> T:
         """Write register (code 0x06).
 
         :param address: Address to write to
@@ -502,7 +501,7 @@ def write_coils(
     def write_registers(
         self,
         address: int,
-        values: Sequence[int | bytes],
+        values: list[int],
         *,
         slave: int = 1,
         no_response_expected: bool = False
@@ -515,11 +514,6 @@ def write_registers(
         :param no_response_expected: (optional) The client will not expect a response to the request
         :raises ModbusException:
 
-        .. tip::
-            values= parameter:
-            entries defined as bytes are silently converted to int !
-            only list[int], list[bytes] or list[bytes | int] are expected (others may work unsupported)
-
         This function is used to write a block of contiguous registers
         (1 to approx. 120 registers) in a remote device.
         """
@@ -629,11 +623,6 @@ def readwrite_registers(
         :param no_response_expected: (optional) The client will not expect a response to the request
         :raises ModbusException:
 
-        .. tip::
-            values= parameter:
-            entries defined as bytes are silently converted to int !
-            only list[int], list[bytes] or list[bytes | int] are expected (others may work unsupported)
-
         This function performs a combination of one read operation and one
         write operation in a single MODBUS transaction. The write
         operation is performed before the read.

pymodbus/pdu/pdu.py

@@ -4,8 +4,6 @@
 import asyncio
 import struct
 from abc import abstractmethod
-from collections.abc import Sequence
-from typing import cast
 
 from pymodbus.exceptions import NotImplementedException
 from pymodbus.logging import Log
@@ -25,21 +23,16 @@ def __init__(self,
             address: int = 0,
             count: int = 0,
             bits: list[bool] | None = None,
-            registers: Sequence[bytes | int] | None = None,
+            registers: list[int] | None = None,
             status: int = 1,
         ) -> None:
         """Initialize the base data for a modbus request."""
         self.slave_id: int = slave_id
         self.transaction_id: int = transaction_id
         self.address: int = address
         self.bits: list[bool] = bits or []
-        if not registers:
-            registers = []
-        self.registers: list[int] = cast(list[int], registers)
-        for i, value in enumerate(registers):
-            if isinstance(value, bytes):
-                self.registers[i] = int.from_bytes(value, byteorder="big")
-        self.count: int = count or len(registers)
+        self.registers: list[int] = registers or []
+        self.count: int = count or len(self.registers)
         self.status: int = status
         self.fut: asyncio.Future
 

test/pdu/test_pdu.py

@@ -183,10 +183,11 @@ def test_pdu_instance_extras(self, pdutype, kwargs):
     def test_pdu_register_as_byte(self):
         """Test validate functions."""
         registers =[b'ab', b'cd']
+        # NOT ALLOWED, NO conversion.
         req = reg_msg.ReadHoldingRegistersRequest(address=117, registers=registers, count=3)
         assert len(req.registers) == 2
-        assert req.registers[0] == 24930
-        assert req.registers[1] == 25444
+        assert req.registers[0] != 24930
+        assert req.registers[1] != 25444
 
     def test_pdu_validate_address(self):
         """Test validate functions."""