feat(ppi): add S7-200 serial transport

README.rst

@@ -47,6 +47,28 @@ Connect to any S7 PLC::
 
 No native libraries or platform-specific dependencies are required.
 
+Experimental S7-200 PPI support
+--------------------------------
+
+Serial PPI communication is available through an optional dependency::
+
+   $ pip install "python-snap7[ppi]"
+
+Use ``PPIClient`` with the serial device and S7-200 station address. V memory
+is translated to DB1, matching the PLC wire protocol::
+
+   from s7 import PPIClient
+
+   with PPIClient("/dev/ttyUSB0", station=2, baudrate=9600) as client:
+       value = client.v_read(0, 4)
+       client.v_write(10, b"\x01\x02")
+
+The generic ``read_area()`` and ``write_area()`` methods accept ``PPIArea``
+values for S, SM, AI, AQ, I, Q, M, V, counters, and timers. The initial
+implementation supports a PC master talking to one S7-200 slave using SD1/SD2
+request framing. Multimaster token passing and PPI-over-TCP are not yet
+implemented and require hardware or trace validation.
+
 .. note::
 
    The ``s7`` package is the recommended import for the legacy S7 protocol.

example/ppi.py

@@ -0,0 +1,13 @@
+"""Read and write S7-200 V memory over a serial PPI cable."""
+
+from s7 import PPIClient
+
+
+def main() -> None:
+    with PPIClient("/dev/ttyUSB0", station=2, baudrate=9600) as client:
+        print(f"VB0..VB3: {client.v_read(0, 4).hex(' ')}")
+        client.v_write(10, b"\x01\x02")
+
+
+if __name__ == "__main__":
+    main()

pyproject.toml

@@ -38,6 +38,7 @@ cli = ["rich", "click" ]
 demo = ["psutil", "rich", "click"]
 doc = ["sphinx", "sphinx_rtd_theme"]
 discovery = ["pnio-dcp"]
+ppi = ["pyserial>=3.5"]
 
 [tool.setuptools.package-data]
 snap7 = ["py.typed"]

s7/__init__.py

@@ -30,6 +30,7 @@
     "logo",
     "optimizer",
     "partner",
+    "ppi",
     "s7protocol",
     "server",
     "tags",

snap7/__init__.py

@@ -19,6 +19,7 @@
 from .async_client import AsyncClient
 from .server import Server
 from .partner import Partner
+from .ppi import PPIArea, PPIClient
 from .logo import Logo
 from .util.db import Row, DB
 from .tags import NodeS7Tag, PLC4XTag, Tag, from_browse, load_csv, load_json, load_tia_xml, parse_tag
@@ -29,6 +30,8 @@
     "AsyncClient",
     "Server",
     "Partner",
+    "PPIClient",
+    "PPIArea",
     "Logo",
     "Row",
     "DB",

snap7/datatypes.py

@@ -44,6 +44,8 @@ class S7WordLen(IntEnum):
     REAL = 0x08  # 32-bit IEEE float
     COUNTER = 0x1C  # Counter value
     TIMER = 0x1D  # Timer value
+    COUNTER_200 = 0x1E  # S7-200 counter value
+    TIMER_200 = 0x1F  # S7-200 timer value
 
 
 class S7DataTypes:
@@ -61,6 +63,8 @@ class S7DataTypes:
         S7WordLen.REAL: 4,  # 4 bytes
         S7WordLen.COUNTER: 2,  # 2 bytes
         S7WordLen.TIMER: 2,  # 2 bytes
+        S7WordLen.COUNTER_200: 2,  # 2 bytes
+        S7WordLen.TIMER_200: 2,  # 2 bytes
     }
 
     @staticmethod
@@ -94,6 +98,9 @@ def encode_address(area: S7Area, db_number: int, start: int, word_len: S7WordLen
             byte_addr = start // 8
             bit_addr = start % 8
             address = (byte_addr << 3) | bit_addr
+        elif word_len in (S7WordLen.COUNTER_200, S7WordLen.TIMER_200):
+            # S7-200 timer/counter addresses are item indexes, not bit addresses.
+            address = start
         else:
             # For word access: convert to bit address
             address = start * 8
@@ -134,7 +141,13 @@ def decode_s7_data(data: bytes, word_len: S7WordLen, count: int) -> List[Union[b
                 values.append(data[offset])
                 offset += 1
 
-            elif word_len == S7WordLen.WORD or word_len == S7WordLen.COUNTER or word_len == S7WordLen.TIMER:
+            elif (
+                word_len == S7WordLen.WORD
+                or word_len == S7WordLen.COUNTER
+                or word_len == S7WordLen.TIMER
+                or word_len == S7WordLen.COUNTER_200
+                or word_len == S7WordLen.TIMER_200
+            ):
                 # 16-bit unsigned values (big-endian)
                 value = struct.unpack(">H", data[offset : offset + 2])[0]
                 values.append(value)
@@ -187,7 +200,13 @@ def encode_s7_data(values: Sequence[Union[bool, int, float]], word_len: S7WordLe
                 # 8-bit values
                 data.append(int(value) & 0xFF)
 
-            elif word_len == S7WordLen.WORD or word_len == S7WordLen.COUNTER or word_len == S7WordLen.TIMER:
+            elif (
+                word_len == S7WordLen.WORD
+                or word_len == S7WordLen.COUNTER
+                or word_len == S7WordLen.TIMER
+                or word_len == S7WordLen.COUNTER_200
+                or word_len == S7WordLen.TIMER_200
+            ):
                 # 16-bit unsigned values (big-endian)
                 data.extend(struct.pack(">H", int(value) & 0xFFFF))