Fix multi-read optimizer: cache mutation, dead code, and usability issues

- Fix cache mutation bug: deep-copy cached ReadPackets before assigning
  buffers so repeated calls don't corrupt cached state
- Remove dead _map_area_int method
- Replace per-call set comprehension with module-level _VALID_AREA_VALUES
  frozenset for Area validation
- Fix O(n^2) byte concatenation in build_multi_read_request using list
  and b"".join()
- Clear optimizer cache on disconnect
- Add use_optimizer attribute (default True) to allow opting out

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: gijzelaerr

snap7/client.py

@@ -4,6 +4,7 @@
 Drop-in replacement for the ctypes-based client with native Python implementation.
 """
 
+import copy
 import logging
 import struct
 import time
@@ -39,6 +40,8 @@
     CDataArrayType,
 )
 
+_VALID_AREA_VALUES: frozenset[int] = frozenset(a.value for a in Area)
+
 logger = logging.getLogger(__name__)
 
 
@@ -113,6 +116,7 @@ def __init__(self, lib_location: Optional[str] = None, **kwargs: Any):
         # Multi-read optimizer state
         self._opt_plan: Optional[_OptimizationPlan] = None
         self.multi_read_max_gap: int = 5
+        self.use_optimizer: bool = True
 
         # Async operation state
         self._async_pending = False
@@ -224,6 +228,7 @@ def disconnect(self) -> int:
             self.connection = None
 
         self.connected = False
+        self._opt_plan = None
         logger.info(f"Disconnected from {self.host}:{self.port}")
         return 0
 
@@ -503,8 +508,8 @@ def read_multi_vars(self, items: Union[List[dict[str, Any]], "Array[S7DataItem]"
         # Dict list path -- use optimizer for 2+ items
         dict_items = cast(List[dict[str, Any]], items)
 
-        if len(dict_items) <= 1:
-            # Single item: no optimization needed
+        if len(dict_items) <= 1 or not self.use_optimizer:
+            # Single item or optimizer disabled: no optimization needed
             results: list[bytearray] = []
             for dict_item in dict_items:
                 area = dict_item["area"]
@@ -555,15 +560,18 @@ def _read_multi_vars_optimized(self, dict_items: List[dict[str, Any]]) -> Tuple[
             packets = packetize(blocks, self.pdu_length)
             self._opt_plan = _OptimizationPlan(cache_key, packets, read_items)
 
+        # Deep-copy blocks from cached packets so we don't mutate cached state
+        working_packets = copy.deepcopy(packets)
+
         # Execute each packet
-        for packet in packets:
+        for packet in working_packets:
             block_specs = [(blk.area, blk.db_number, blk.start_offset, blk.byte_length) for blk in packet.blocks]
 
             if len(block_specs) == 1:
                 # Single block: use regular read to avoid multi-read overhead
                 blk = packet.blocks[0]
                 data = self.read_area(
-                    Area(blk.area) if blk.area in {a.value for a in Area} else Area.DB,
+                    Area(blk.area) if blk.area in _VALID_AREA_VALUES else Area.DB,
                     blk.db_number,
                     blk.start_offset,
                     blk.byte_length,
@@ -578,13 +586,9 @@ def _read_multi_vars_optimized(self, dict_items: List[dict[str, Any]]) -> Tuple[
                     blk.buffer = buf
 
         # Extract per-item results in original order
-        results = extract_results(packets, len(dict_items))
+        results = extract_results(working_packets, len(dict_items))
         return (0, results)
 
-    def _map_area_int(self, area_int: int) -> S7Area:
-        """Map integer area value to S7Area enum."""
-        return S7Area(area_int)
-
     def write_multi_vars(self, items: Union[List[dict[str, Any]], List[S7DataItem]]) -> int:
         """
         Write multiple variables in a single request.

snap7/s7protocol.py

@@ -188,13 +188,14 @@ def build_multi_read_request(self, items: List[Tuple[int, int, int, int]]) -> by
         item_count = len(items)
 
         # Build N * 12-byte address specifications
-        addr_specs = b""
+        addr_spec_parts: list[bytes] = []
         for area_code, db_number, start_offset, byte_length in items:
-            addr_spec = S7DataTypes.encode_address(S7Area(area_code), db_number, start_offset, S7WordLen.BYTE, byte_length)
-            addr_specs += addr_spec
+            addr_spec_parts.append(
+                S7DataTypes.encode_address(S7Area(area_code), db_number, start_offset, S7WordLen.BYTE, byte_length)
+            )
 
         # Parameter: function_code(1) + item_count(1) + N * address_spec(12)
-        param_data = struct.pack(">BB", S7Function.READ_AREA, item_count) + addr_specs
+        param_data = struct.pack(">BB", S7Function.READ_AREA, item_count) + b"".join(addr_spec_parts)
         param_len = len(param_data)
 
         # S7 Header (12 bytes)