Fix CI/CD issues: mypy errors, partner port, sphinx docs

- Change partner default port from 102 to 1102 (non-privileged)
- Add missing type annotations across all snap7 modules
- Fix client.py read_multi_vars and write_multi_vars type handling
- Use cast() for proper type narrowing in union types
- Change encode_s7_data parameter type from List to Sequence
- Add missing return type annotations to test methods
- Fix callback type annotations (use SrvEvent instead of str)
- Update example files to use correct API signatures
- Update server.rst documentation for pure Python implementation

🤖 Generated with [Claude Code](https://claude.com/claude-code)

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

Author: gijzelaerr

doc/API/server.rst

@@ -1,34 +1,22 @@
 Server
 ======
 
-If you just need a quick server with some default values initalised, this package provides a default implementation.
-To use it you first need to install some aditional dependencies, using:
+The pure Python server implementation provides a simulated S7 server for testing.
 
-.. code:: bash
+To start a server programmatically:
 
-   pip install python-snap7[cli]
+.. code:: python
 
-Now you can start it using one of the following commands:
+   from snap7.server import Server, mainloop
 
-.. code:: bash
+   # Quick start with mainloop helper
+   mainloop(tcp_port=1102)
 
-   python -m snap7.server
-   # or, if your Python `Scripts/` folder is on PATH:
-   snap7-server
-
-You can optionally provide the port to be used as an argument, like this:
-
-.. code:: bash
-
-   python -m snap7.server --port 102
+   # Or create and configure manually
+   server = Server()
+   server.start(port=1102)
 
 ----
 
 .. automodule:: snap7.server
    :members:
-
-----
-
-.. automodule:: snap7.server.__main__
-
-   .. autofunction:: main(port, dll)

example/boolean.py

@@ -27,7 +27,7 @@
 
 reading = plc.db_read(31, 120, 1)  # read 1 byte from db 31 staring from byte 120
 set_bool(reading, 0, 5, True)  # set a value of fifth bit
-plc.db_write(reading, 31, 120, 1)  # write back the bytearray and now the boolean value is changed in the PLC.
+plc.db_write(31, 120, reading)  # write back the bytearray and now the boolean value is changed in the PLC.
 
 # NOTE you could also use the read_area and write_area functions.
 # then you can specify an area to read from:
@@ -41,6 +41,6 @@
 
 data = bytearray()
 set_int(data, 0, 127)
-plc.write_area(area=Area.MK, dbnumber=0, start=20, data=data)
+plc.write_area(area=Area.MK, db_number=0, start=20, data=data)
 # read the client source code!
 # and official snap7 documentation

example/example.py

@@ -9,6 +9,7 @@
 from db_layouts import tank_rc_if_db_layout
 
 from snap7 import Client, Row, DB
+from snap7.type import Area
 from util.db import print_row
 
 client = Client()
@@ -61,7 +62,8 @@ def set_row(x: int, row: Row) -> None:
     byte array representation of row in the PLC
     """
     row_size = 126
-    client.db_write(1, 4 + x * row_size, row_size, row._bytearray)
+    assert isinstance(row._bytearray, bytearray)
+    client.db_write(1, 4 + x * row_size, row._bytearray)
 
 
 def open_row(row: Row) -> None:
@@ -107,7 +109,7 @@ def open_and_close() -> None:
 
 def set_part_db(start: int, size: int, _bytearray: bytearray) -> None:
     data = _bytearray[start : start + size]
-    client.db_write(1, start, size, data)
+    client.db_write(1, start, data)
 
 
 # def write_data_db(dbnumber, all_data, size):
@@ -126,7 +128,7 @@ def open_and_close_db1() -> None:
         # set_part_db(4+x*126, 126, all_data)
 
     t = time.time()
-    client.write_area(1, all_data, 4 + 126 * 450)
+    client.write_area(Area.DB, 1, 4, all_data)
     print(f"opening all valves took: {time.time() - t}")
 
     print("sleep...")
@@ -138,7 +140,7 @@ def open_and_close_db1() -> None:
     print(time.time() - t)
 
     t = time.time()
-    client.write_area(1, all_data, 4 + 126 * 450)
+    client.write_area(Area.DB, 1, 4, all_data)
     print(f"closing all valves took: {time.time() - t}")
 
 

snap7/client.py

@@ -7,7 +7,7 @@
 import logging
 import struct
 import time
-from typing import List, Any, Optional, Tuple, Union, Callable
+from typing import List, Any, Optional, Tuple, Union, Callable, cast
 from datetime import datetime
 from ctypes import (
     c_int,
@@ -91,6 +91,10 @@ def __init__(self, lib_location: Optional[str] = None, **kwargs: Any):
         # Session password
         self.session_password: Optional[str] = None
 
+        # Execution time tracking
+        self._exec_time = 0
+        self.last_error = 0
+
         # Parameter storage
         self._params = {
             Parameter.LocalPort: 0,
@@ -116,6 +120,12 @@ def __init__(self, lib_location: Optional[str] = None, **kwargs: Any):
 
         logger.info("S7Client initialized (pure Python implementation)")
 
+    def _get_connection(self) -> ISOTCPConnection:
+        """Get connection, raising if not connected."""
+        if self.connection is None:
+            raise S7ConnectionError("Not connected to PLC")
+        return self.connection
+
     def connect(self, address: str, rack: int, slot: int, tcp_port: int = 102) -> "Client":
         """
         Connect to S7 PLC.
@@ -267,8 +277,7 @@ def read_area(self, area: Area, db_number: int, start: int, size: int) -> bytear
         Returns:
             Data read from area
         """
-        if not self.get_connected():
-            raise S7ConnectionError("Not connected to PLC")
+        conn = self._get_connection()
 
         start_time = time.time()
 
@@ -286,10 +295,10 @@ def read_area(self, area: Area, db_number: int, start: int, size: int) -> bytear
         # Build and send read request
         request = self.protocol.build_read_request(area=s7_area, db_number=db_number, start=start, word_len=word_len, count=size)
 
-        self.connection.send_data(request)
+        conn.send_data(request)
 
         # Receive and parse response
-        response_data = self.connection.receive_data()
+        response_data = conn.receive_data()
         response = self.protocol.parse_response(response_data)
 
         # Extract data from response - pass item count, not byte count
@@ -311,8 +320,7 @@ def write_area(self, area: Area, db_number: int, start: int, data: bytearray) ->
         Returns:
             0 on success
         """
-        if not self.get_connected():
-            raise S7ConnectionError("Not connected to PLC")
+        conn = self._get_connection()
 
         start_time = time.time()
 
@@ -332,18 +340,18 @@ def write_area(self, area: Area, db_number: int, start: int, data: bytearray) ->
             area=s7_area, db_number=db_number, start=start, word_len=word_len, data=bytes(data)
         )
 
-        self.connection.send_data(request)
+        conn.send_data(request)
 
         # Receive and parse response
-        response_data = self.connection.receive_data()
+        response_data = conn.receive_data()
         response = self.protocol.parse_response(response_data)
 
         # Check for write errors
         self.protocol.check_write_response(response)
         self._exec_time = int((time.time() - start_time) * 1000)
         return 0
 
-    def read_multi_vars(self, items: Union[List[dict], "Array[S7DataItem]"]) -> Tuple[int, Any]:
+    def read_multi_vars(self, items: Union[List[dict[str, Any]], "Array[S7DataItem]"]) -> Tuple[int, Any]:
         """
         Read multiple variables in a single request.
 
@@ -358,34 +366,36 @@ def read_multi_vars(self, items: Union[List[dict], "Array[S7DataItem]"]) -> Tupl
 
         # Handle S7DataItem array (ctypes)
         if hasattr(items, "_type_") and hasattr(items[0], "Area"):
-            # This is a ctypes array of S7DataItem
-            for item in items:
-                area = Area(item.Area)
-                db_number = item.DBNumber
-                start = item.Start
-                size = item.Amount
+            # This is a ctypes array of S7DataItem - use cast for type safety
+            s7_items = cast("Array[S7DataItem]", items)
+            for s7_item in s7_items:
+                area = Area(s7_item.Area)
+                db_number = s7_item.DBNumber
+                start = s7_item.Start
+                size = s7_item.Amount
                 data = self.read_area(area, db_number, start, size)
 
                 # Copy data to pData buffer
-                if item.pData:
+                if s7_item.pData:
                     for i, b in enumerate(data):
-                        item.pData[i] = b
+                        s7_item.pData[i] = b
 
             return (0, items)
 
         # Handle dict list
+        dict_items = cast(List[dict[str, Any]], items)
         results = []
-        for item in items:
-            area = item["area"]
-            db_number = item.get("db_number", 0)
-            start = item["start"]
-            size = item["size"]
+        for dict_item in dict_items:
+            area = dict_item["area"]
+            db_number = dict_item.get("db_number", 0)
+            start = dict_item["start"]
+            size = dict_item["size"]
             data = self.read_area(area, db_number, start, size)
             results.append(data)
 
         return (0, results)
 
-    def write_multi_vars(self, items: Union[List[dict], List[S7DataItem]]) -> int:
+    def write_multi_vars(self, items: Union[List[dict[str, Any]], List[S7DataItem]]) -> int:
         """
         Write multiple variables in a single request.
 
@@ -400,27 +410,29 @@ def write_multi_vars(self, items: Union[List[dict], List[S7DataItem]]) -> int:
 
         # Handle S7DataItem list (ctypes)
         if hasattr(items[0], "Area"):
-            for item in items:
-                area = Area(item.Area)
-                db_number = item.DBNumber
-                start = item.Start
-                size = item.Amount
+            s7_items = cast(List[S7DataItem], items)
+            for s7_item in s7_items:
+                area = Area(s7_item.Area)
+                db_number = s7_item.DBNumber
+                start = s7_item.Start
+                size = s7_item.Amount
 
                 # Extract data from pData
                 data = bytearray(size)
-                if item.pData:
+                if s7_item.pData:
                     for i in range(size):
-                        data[i] = item.pData[i]
+                        data[i] = s7_item.pData[i]
 
                 self.write_area(area, db_number, start, data)
             return 0
 
         # Handle dict list
-        for item in items:
-            area = item["area"]
-            db_number = item.get("db_number", 0)
-            start = item["start"]
-            data = item["data"]
+        dict_items = cast(List[dict[str, Any]], items)
+        for dict_item in dict_items:
+            area = dict_item["area"]
+            db_number = dict_item.get("db_number", 0)
+            start = dict_item["start"]
+            data = dict_item["data"]
             self.write_area(area, db_number, start, data)
 
         return 0
@@ -491,13 +503,12 @@ def get_cpu_state(self) -> str:
         Returns:
             CPU state string
         """
-        if not self.get_connected():
-            raise S7ConnectionError("Not connected to PLC")
+        conn = self._get_connection()
 
         request = self.protocol.build_cpu_state_request()
-        self.connection.send_data(request)
+        conn.send_data(request)
 
-        response_data = self.connection.receive_data()
+        response_data = conn.receive_data()
         response = self.protocol.parse_response(response_data)
 
         return self.protocol.extract_cpu_state(response)
@@ -674,13 +685,12 @@ def plc_stop(self) -> int:
         Returns:
             0 on success
         """
-        if not self.get_connected():
-            raise S7ConnectionError("Not connected to PLC")
+        conn = self._get_connection()
 
         request = self.protocol.build_plc_control_request("stop")
-        self.connection.send_data(request)
+        conn.send_data(request)
 
-        response_data = self.connection.receive_data()
+        response_data = conn.receive_data()
         response = self.protocol.parse_response(response_data)
 
         self.protocol.check_control_response(response)
@@ -692,13 +702,12 @@ def plc_hot_start(self) -> int:
         Returns:
             0 on success
         """
-        if not self.get_connected():
-            raise S7ConnectionError("Not connected to PLC")
+        conn = self._get_connection()
 
         request = self.protocol.build_plc_control_request("hot_start")
-        self.connection.send_data(request)
+        conn.send_data(request)
 
-        response_data = self.connection.receive_data()
+        response_data = conn.receive_data()
         response = self.protocol.parse_response(response_data)
 
         self.protocol.check_control_response(response)
@@ -710,13 +719,12 @@ def plc_cold_start(self) -> int:
         Returns:
             0 on success
         """
-        if not self.get_connected():
-            raise S7ConnectionError("Not connected to PLC")
+        conn = self._get_connection()
 
         request = self.protocol.build_plc_control_request("cold_start")
-        self.connection.send_data(request)
+        conn.send_data(request)
 
-        response_data = self.connection.receive_data()
+        response_data = conn.receive_data()
         response = self.protocol.parse_response(response_data)
 
         self.protocol.check_control_response(response)
@@ -944,11 +952,10 @@ def iso_exchange_buffer(self, data: bytearray) -> bytearray:
         Returns:
             Response PDU data
         """
-        if not self.get_connected():
-            raise S7ConnectionError("Not connected to PLC")
+        conn = self._get_connection()
 
-        self.connection.send_data(bytes(data))
-        response = self.connection.receive_data()
+        conn.send_data(bytes(data))
+        response = conn.receive_data()
         return bytearray(response)
 
     # Convenience methods for specific memory areas
@@ -1409,11 +1416,12 @@ def set_param(self, param: Parameter, value: int) -> int:
 
     def _setup_communication(self) -> None:
         """Setup communication and negotiate PDU length."""
+        conn = self._get_connection()
         request = self.protocol.build_setup_communication_request(max_amq_caller=1, max_amq_callee=1, pdu_length=self.pdu_length)
 
-        self.connection.send_data(request)
+        conn.send_data(request)
 
-        response_data = self.connection.receive_data()
+        response_data = conn.receive_data()
         response = self.protocol.parse_response(response_data)
 
         if response.get("parameters"):