feat(s7commplus): add multi-write APIs

s7commplus/async_client.py

@@ -12,6 +12,7 @@
 from . import typeinfo
 from .blob_decompressor import find_and_decompress
 from .client import (
+    DBWriteItem,
     _build_area_read_payload,
     _build_area_write_payload,
     _build_explore_payload,
@@ -495,12 +496,20 @@ async def db_read(self, db_number: int, start: int, size: int) -> bytes:
             raise RuntimeError("Read failed: PLC returned error for item")
         return results[0]
 
-    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)], self._protocol_version)
+    async def db_write(self, db_number: int, start: int, data: bytes, datatype: DataType = DataType.BLOB) -> None:
+        """Write raw bytes to a data block with an optional explicit PValue datatype."""
+        await self.db_write_multi([(db_number, start, data, datatype)])
+
+    async def db_write_multi(self, items: list[DBWriteItem]) -> None:
+        """Write multiple regions, optionally adding a DataType as each tuple's fourth item."""
+        payload = _build_write_payload(items, self._protocol_version)
         response = await self._send_request(FunctionCode.SET_MULTI_VARIABLES, payload)
         _parse_write_response(response)
 
+    async def write_multi(self, items: list[DBWriteItem]) -> 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, self._protocol_version)
@@ -694,10 +703,7 @@ async def read_symbolic(self, access_area: int, lids: list[int], symbol_crc: int
 
         .. warning:: This method is **experimental** and may change.
         """
-        # TODO: Send the correct integrity id once available
-        payload = _build_symbolic_read_payload(
-            access_area, lids, symbol_crc, False, self._integrity_id_read, self._protocol_version
-        )
+        payload = _build_symbolic_read_payload(access_area, lids, symbol_crc, self._protocol_version)
         response = await self._send_request(FunctionCode.GET_MULTI_VARIABLES, payload)
         results = _parse_read_response(response)
         if not results or results[0] is None:

s7commplus/client.py

@@ -6,7 +6,7 @@
 import logging
 import struct
 from collections.abc import Callable, Sequence
-from typing import Any, Optional, TypeVar
+from typing import Any, Optional, TypeAlias, TypeVar
 
 from snap7.error import S7ConnectionError
 
@@ -27,6 +27,7 @@
     encode_item_address,
     encode_object_qualifier,
     encode_pvalue_blob,
+    encode_pvalue_typed,
     parse_create_object_session_id,
 )
 from .connection import S7CommPlusConnection
@@ -43,6 +44,15 @@
 logger = logging.getLogger(__name__)
 
 _T = TypeVar("_T")
+DBWriteItem: TypeAlias = tuple[int, int, bytes] | tuple[int, int, bytes, DataType]
+
+
+def _normalize_write_item(item: DBWriteItem) -> tuple[int, int, bytes, DataType]:
+    if len(item) == 3:
+        db_number, start, data = item
+        return db_number, start, data, DataType.BLOB
+    db_number, start, data, datatype = item
+    return db_number, start, data, DataType(datatype)
 
 
 class S7CommPlusClient:
@@ -220,30 +230,46 @@ def _db_read_substreamed(self, db_number: int, start: int, size: int) -> bytes:
         response = self._connection.send_request(FunctionCode.GET_VAR_SUBSTREAMED, payload)
         return _parse_substreamed_read_response(response)
 
-    def db_write(self, db_number: int, start: int, data: bytes) -> None:
+    def db_write(self, db_number: int, start: int, data: bytes, datatype: DataType = DataType.BLOB) -> None:
         """Write raw bytes to a data block.
 
         Args:
             db_number: Data block number
             start: Start byte offset
             data: Bytes to write
+            datatype: S7CommPlus PValue datatype for ``data``. Defaults to BLOB.
+        """
+        self.db_write_multi([(db_number, start, data, datatype)])
+
+    def db_write_multi(self, items: list[DBWriteItem]) -> None:
+        """Write multiple data block regions in a single request.
+
+        Args:
+            items: ``(db_number, start_offset, data)`` tuples for BLOB writes,
+                or four-tuples adding an explicit :class:`DataType`.
         """
         if self._connection is None:
             raise RuntimeError("Not connected")
 
         if self._connection.requires_substreamed:
-            self._db_write_substreamed(db_number, start, data)
+            for item in items:
+                db_number, start, data, datatype = _normalize_write_item(item)
+                self._db_write_substreamed(db_number, start, data, datatype)
             return
 
-        payload = _build_write_payload([(db_number, start, data)], self._connection.protocol_version)
+        payload = _build_write_payload(items, self._connection.protocol_version)
         response = self._connection.send_request(FunctionCode.SET_MULTI_VARIABLES, payload)
         _parse_write_response(response)
 
-    def _db_write_substreamed(self, db_number: int, start: int, data: bytes) -> None:
+    def write_multi(self, items: list[DBWriteItem]) -> None:
+        """Alias for :meth:`db_write_multi`."""
+        self.db_write_multi(items)
+
+    def _db_write_substreamed(self, db_number: int, start: int, data: bytes, datatype: DataType = DataType.BLOB) -> None:
         assert self._connection is not None
         access_area = Ids.DB_ACCESS_AREA_BASE + (db_number & 0xFFFF)
         payload = _build_substreamed_write_payload(
-            self._connection.session_id, access_area, Ids.DB_VALUE_ACTUAL, [start + 1, len(data)], data
+            self._connection.session_id, access_area, Ids.DB_VALUE_ACTUAL, [start + 1, len(data)], data, datatype
         )
         self._connection.send_request(FunctionCode.SET_VAR_SUBSTREAMED, payload)
 
@@ -356,9 +382,7 @@ def read_symbolic(self, access_area: int, lids: list[int], symbol_crc: int = 0)
             raise RuntimeError("Not connected")
 
         # TODO: Send the correct integrity id once available
-        payload = _build_symbolic_read_payload(
-            access_area, lids, symbol_crc, False, protocol_version=self._connection.protocol_version
-        )
+        payload = _build_symbolic_read_payload(access_area, lids, symbol_crc, self._connection.protocol_version)
         response = self._connection.send_request(FunctionCode.GET_MULTI_VARIABLES, payload)
         results = _parse_read_response(response)
         if not results or results[0] is None:
@@ -830,7 +854,6 @@ def _build_read_payload(items: list[tuple[int, int, int]], protocol_version: int
     for addr in addresses:
         payload += addr
     payload += encode_object_qualifier(protocol_version=protocol_version)
-    payload += encode_uint32_vlq(1)
     payload += struct.pack(">I", 0)
 
     return bytes(payload)
@@ -891,7 +914,7 @@ def _parse_read_response(response: bytes) -> list[Optional[bytes]]:
     return results
 
 
-def _build_write_payload(items: list[tuple[int, int, bytes]], protocol_version: int = ProtocolVersion.V2) -> bytes:
+def _build_write_payload(items: list[DBWriteItem], protocol_version: int = ProtocolVersion.V2) -> bytes:
     """Build a SetMultiVariables request payload.
 
     Args:
@@ -902,7 +925,8 @@ def _build_write_payload(items: list[tuple[int, int, bytes]], protocol_version:
     """
     addresses: list[bytes] = []
     total_field_count = 0
-    for db_number, start, data in items:
+    normalized = [_normalize_write_item(item) for item in items]
+    for db_number, start, data, _ in normalized:
         access_area = Ids.DB_ACCESS_AREA_BASE + (db_number & 0xFFFF)
         addr_bytes, field_count = encode_item_address(
             access_area=access_area,
@@ -918,12 +942,11 @@ def _build_write_payload(items: list[tuple[int, int, bytes]], protocol_version:
     payload += encode_uint32_vlq(total_field_count)
     for addr in addresses:
         payload += addr
-    for i, (_, _, data) in enumerate(items, 1):
+    for i, (_, _, data, datatype) in enumerate(normalized, 1):
         payload += encode_uint32_vlq(i)
-        payload += encode_pvalue_blob(data)
+        payload += encode_pvalue_typed(datatype, data)
     payload += bytes([0x00])
     payload += encode_object_qualifier(protocol_version=protocol_version)
-    payload += encode_uint32_vlq(1)
     payload += struct.pack(">I", 0)
 
     return bytes(payload)
@@ -981,7 +1004,12 @@ def _build_substreamed_read_payload(session_id: int, access_area: int, access_su
 
 
 def _build_substreamed_write_payload(
-    session_id: int, access_area: int, access_sub_area: int, lids: list[int], data: bytes
+    session_id: int,
+    access_area: int,
+    access_sub_area: int,
+    lids: list[int],
+    data: bytes,
+    datatype: DataType = DataType.BLOB,
 ) -> bytes:
     """Build a SET_VAR_SUBSTREAMED payload for data access on V1-initial PLCs."""
     oq = encode_object_qualifier(protocol_version=ProtocolVersion.V1)
@@ -999,7 +1027,7 @@ def _build_substreamed_write_payload(
     payload += oq
     payload += bytes([0x00])
     payload += encode_uint32_vlq(1)
-    payload += encode_pvalue_blob(data)
+    payload += encode_pvalue_typed(datatype, data)
     payload += encode_uint32_vlq(1)
     payload += struct.pack(">I", 0)
     return bytes(payload)
@@ -1040,7 +1068,6 @@ def _build_area_read_payload(area_rid: int, start: int, size: int, protocol_vers
     payload += encode_uint32_vlq(field_count)
     payload += addr_bytes
     payload += encode_object_qualifier(protocol_version=protocol_version)
-    payload += encode_uint32_vlq(1)
     payload += struct.pack(">I", 0)
     return bytes(payload)
 
@@ -1062,7 +1089,6 @@ def _build_area_write_payload(area_rid: int, start: int, data: bytes, protocol_v
     payload += encode_pvalue_blob(data)
     payload += bytes([0x00])
     payload += encode_object_qualifier(protocol_version=protocol_version)
-    payload += encode_uint32_vlq(1)
     payload += struct.pack(">I", 0)
     return bytes(payload)
 
@@ -1071,8 +1097,6 @@ def _build_symbolic_read_payload(
     access_area: int,
     lids: list[int],
     symbol_crc: int = 0,
-    with_integrity: bool = True,
-    integrity_id: int = 1,
     protocol_version: int = ProtocolVersion.V2,
 ) -> bytes:
     """Build a GetMultiVariables payload for symbolic (LID-based) access.
@@ -1102,8 +1126,6 @@ def _build_symbolic_read_payload(
     payload += encode_uint32_vlq(field_count)
     payload += addr_bytes
     payload += encode_object_qualifier(protocol_version=protocol_version)
-    if with_integrity:
-        payload += encode_uint32_vlq(integrity_id)
     payload += struct.pack(">I", 0)
     return bytes(payload)
 
@@ -1137,7 +1159,6 @@ def _build_symbolic_write_payload(
     payload += encode_pvalue_blob(data)
     payload += bytes([0x00])
     payload += encode_object_qualifier(protocol_version=protocol_version)
-    payload += encode_uint32_vlq(1)
     payload += struct.pack(">I", 0)
     return bytes(payload)
 

s7commplus/codec.py

@@ -379,6 +379,57 @@ def encode_pvalue_blob(data: bytes) -> bytes:
     return bytes(result)
 
 
+def encode_pvalue_typed(datatype: DataType, data: bytes) -> bytes:
+    """Encode already-serialized value bytes with an explicit PValue type.
+
+    Raw DB offsets do not carry enough metadata to infer the PLC datatype.
+    Callers writing anything other than a byte-array/BLOB must therefore
+    provide the type reported by the PLC's symbol information.
+    """
+    datatype = DataType(datatype)
+    fixed_sizes = {
+        DataType.BOOL: 1,
+        DataType.USINT: 1,
+        DataType.BYTE: 1,
+        DataType.SINT: 1,
+        DataType.UINT: 2,
+        DataType.WORD: 2,
+        DataType.INT: 2,
+        DataType.UDINT: 4,
+        DataType.DWORD: 4,
+        DataType.DINT: 4,
+        DataType.REAL: 4,
+        DataType.ULINT: 8,
+        DataType.LWORD: 8,
+        DataType.LINT: 8,
+        DataType.LREAL: 8,
+        DataType.TIMESTAMP: 8,
+        DataType.TIMESPAN: 8,
+        DataType.RID: 4,
+        DataType.AID: 4,
+    }
+    expected_size = fixed_sizes.get(datatype)
+    if expected_size is not None and len(data) != expected_size:
+        raise ValueError(f"{datatype.name} requires {expected_size} encoded bytes, got {len(data)}")
+    if datatype == DataType.NULL and data:
+        raise ValueError("NULL requires an empty value")
+
+    result = bytearray((0x00, datatype))
+    if datatype in (DataType.BLOB, DataType.WSTRING, DataType.S7STRING):
+        result += encode_uint32_vlq(len(data))
+    if datatype in (DataType.UDINT, DataType.AID):
+        result += encode_uint32_vlq(int.from_bytes(data, "big"))
+    elif datatype == DataType.ULINT:
+        result += encode_uint64_vlq(int.from_bytes(data, "big"))
+    elif datatype == DataType.DINT:
+        result += encode_int32_vlq(int.from_bytes(data, "big", signed=True))
+    elif datatype in (DataType.LINT, DataType.TIMESPAN):
+        result += encode_int64_vlq(int.from_bytes(data, "big", signed=True))
+    else:
+        result += data
+    return bytes(result)
+
+
 def decode_pvalue_to_bytes(data: bytes, offset: int) -> tuple[bytes, int]:
     """Decode a PValue from S7CommPlus response to raw bytes.
 

tests/test_s7_codec.py

@@ -29,6 +29,7 @@
     encode_item_address,
     encode_object_qualifier,
     encode_pvalue_blob,
+    encode_pvalue_typed,
     encode_request_header,
     encode_typed_value,
     encode_uint8,
@@ -409,6 +410,32 @@ def test_roundtrip_with_decode(self) -> None:
         assert consumed == len(encoded)
 
 
+class TestPValueTyped:
+    def test_word_uses_raw_big_endian_bytes(self) -> None:
+        assert encode_pvalue_typed(DataType.WORD, b"\x02\x00") == bytes((0x00, DataType.WORD, 0x02, 0x00))
+
+    @pytest.mark.parametrize(
+        ("datatype", "data", "encoded_value"),
+        [
+            (DataType.UDINT, struct.pack(">I", 300), b"\x82\x2c"),
+            (DataType.DINT, struct.pack(">i", -5), b"\x7b"),
+            (DataType.ULINT, struct.pack(">Q", 300), b"\x82\x2c"),
+            (DataType.LINT, struct.pack(">q", -5), b"\x7b"),
+            (DataType.TIMESPAN, struct.pack(">q", -5), b"\x7b"),
+            (DataType.AID, struct.pack(">I", 300), b"\x82\x2c"),
+        ],
+    )
+    def test_variable_length_integer_types_use_vlq(self, datatype: DataType, data: bytes, encoded_value: bytes) -> None:
+        assert encode_pvalue_typed(datatype, data) == bytes((0x00, datatype)) + encoded_value
+
+    def test_rejects_wrong_fixed_width(self) -> None:
+        with pytest.raises(ValueError, match="REAL requires 4 encoded bytes"):
+            encode_pvalue_typed(DataType.REAL, b"\x00\x00")
+
+        with pytest.raises(ValueError, match="UDINT requires 4 encoded bytes"):
+            encode_pvalue_typed(DataType.UDINT, b"\x00\x00")
+
+
 class TestDecodePValue:
     """Test decode_pvalue_to_bytes for all scalar and array type branches."""
 

tests/test_s7_server.py

@@ -208,6 +208,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)
@@ -305,6 +323,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)