feat(s7commplus): add multi-write APIs

s7commplus/async_client.py

@@ -451,10 +451,18 @@ async def db_read(self, db_number: int, start: int, size: int) -> bytes:
 
     async def db_write(self, db_number: int, start: int, data: bytes) -> None:
         """Write raw bytes to a data block."""
-        payload = _build_write_payload([(db_number, start, data)])
+        await self.db_write_multi([(db_number, start, data)])
+
+    async def db_write_multi(self, items: list[tuple[int, int, bytes]]) -> None:
+        """Write multiple data block regions in a single request."""
+        payload = _build_write_payload(items)
         response = await self._send_request(FunctionCode.SET_MULTI_VARIABLES, payload)
         _parse_write_response(response)
 
+    async def write_multi(self, items: list[tuple[int, int, bytes]]) -> None:
+        """Alias for :meth:`db_write_multi`."""
+        await self.db_write_multi(items)
+
     async def db_read_multi(self, items: list[tuple[int, int, int]]) -> list[bytes]:
         """Read multiple data block regions in a single request."""
         payload = _build_read_payload(items)

s7commplus/client.py

@@ -137,13 +137,25 @@ def db_write(self, db_number: int, start: int, data: bytes) -> None:
             start: Start byte offset
             data: Bytes to write
         """
+        self.db_write_multi([(db_number, start, data)])
+
+    def db_write_multi(self, items: list[tuple[int, int, bytes]]) -> None:
+        """Write multiple data block regions in a single request.
+
+        Args:
+            items: List of ``(db_number, start_offset, data)`` tuples.
+        """
         if self._connection is None:
             raise RuntimeError("Not connected")
 
-        payload = _build_write_payload([(db_number, start, data)])
+        payload = _build_write_payload(items)
         response = self._connection.send_request(FunctionCode.SET_MULTI_VARIABLES, payload)
         _parse_write_response(response)
 
+    def write_multi(self, items: list[tuple[int, int, bytes]]) -> None:
+        """Alias for :meth:`db_write_multi`."""
+        self.db_write_multi(items)
+
     def db_read_multi(self, items: list[tuple[int, int, int]]) -> list[bytes]:
         """Read multiple data block regions in a single request.
 

tests/test_s7_server.py

@@ -207,6 +207,24 @@ def test_multi_read(self, server: S7CommPlusServer) -> None:
         finally:
             client.disconnect()
 
+    def test_multi_write(self, server: S7CommPlusServer) -> None:
+        client = S7CommPlusClient()
+        client.connect("127.0.0.1", port=TEST_PORT)
+        try:
+            client.db_write_multi(
+                [
+                    (1, 0, b"first"),
+                    (1, 10, b"second"),
+                    (2, 20, b"third"),
+                ]
+            )
+
+            assert client.db_read(1, 0, 5) == b"first"
+            assert client.db_read(1, 10, 6) == b"second"
+            assert client.db_read(2, 20, 5) == b"third"
+        finally:
+            client.disconnect()
+
     def test_explore(self, server: S7CommPlusServer) -> None:
         client = S7CommPlusClient()
         client.connect("127.0.0.1", port=TEST_PORT)
@@ -304,6 +322,21 @@ async def test_multi_read(self, server: S7CommPlusServer) -> None:
             temp = struct.unpack(">f", results[0])[0]
             assert abs(temp - 23.5) < 0.1  # May be modified by earlier test
 
+    async def test_multi_write(self, server: S7CommPlusServer) -> None:
+        async with S7CommPlusAsyncClient() as client:
+            await client.connect("127.0.0.1", port=TEST_PORT)
+            await client.write_multi(
+                [
+                    (1, 0, b"alpha"),
+                    (1, 10, b"beta"),
+                    (2, 20, b"gamma"),
+                ]
+            )
+
+            assert await client.db_read(1, 0, 5) == b"alpha"
+            assert await client.db_read(1, 10, 4) == b"beta"
+            assert await client.db_read(2, 20, 5) == b"gamma"
+
     async def test_explore(self, server: S7CommPlusServer) -> None:
         async with S7CommPlusAsyncClient() as client:
             await client.connect("127.0.0.1", port=TEST_PORT)

tests/test_s7_unit.py

@@ -501,6 +501,16 @@ def test_db_read_multi_not_connected(self) -> None:
         with pytest.raises(RuntimeError, match="Not connected"):
             client.db_read_multi([(1, 0, 4)])
 
+    def test_db_write_multi_not_connected(self) -> None:
+        client = S7CommPlusClient()
+        with pytest.raises(RuntimeError, match="Not connected"):
+            client.db_write_multi([(1, 0, b"data")])
+
+    def test_write_multi_not_connected(self) -> None:
+        client = S7CommPlusClient()
+        with pytest.raises(RuntimeError, match="Not connected"):
+            client.write_multi([(1, 0, b"data")])
+
     def test_explore_not_connected(self) -> None:
         client = S7CommPlusClient()
         with pytest.raises(RuntimeError, match="Not connected"):