diff --git a/README.rst b/README.rst index ddc50f3d..d6e7f113 100644 --- a/README.rst +++ b/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. diff --git a/example/ppi.py b/example/ppi.py new file mode 100644 index 00000000..b20788ce --- /dev/null +++ b/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() diff --git a/pyproject.toml b/pyproject.toml index 46522101..c1be2c46 100644 --- a/pyproject.toml +++ b/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"] diff --git a/s7/__init__.py b/s7/__init__.py index a1fd2a7a..4c895d8b 100644 --- a/s7/__init__.py +++ b/s7/__init__.py @@ -30,6 +30,7 @@ "logo", "optimizer", "partner", + "ppi", "s7protocol", "server", "tags", diff --git a/snap7/__init__.py b/snap7/__init__.py index eb31ba78..1dbcfed8 100644 --- a/snap7/__init__.py +++ b/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", diff --git a/snap7/datatypes.py b/snap7/datatypes.py index f86a2ae7..5cf9ef42 100644 --- a/snap7/datatypes.py +++ b/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)) diff --git a/snap7/ppi.py b/snap7/ppi.py new file mode 100644 index 00000000..75b3b1b7 --- /dev/null +++ b/snap7/ppi.py @@ -0,0 +1,403 @@ +"""Siemens S7-200 PPI serial transport and client. + +PPI carries ordinary S7 PDUs inside PROFIBUS-style SD1/SD2 frames. The +implementation follows the request/acknowledgement exchange used by libnodave: +an SD2 request, an E5 acknowledgement, an SD1 request-data poll, and an SD2 +response. +""" + +import importlib +import logging +import threading +from dataclasses import dataclass +from enum import IntEnum +from types import TracebackType +from typing import Any, Protocol, cast + +from .datatypes import S7Area, S7WordLen +from .error import S7ConnectionError, S7ProtocolError +from .s7protocol import S7Protocol + +logger = logging.getLogger(__name__) + + +class PPIFrameType(IntEnum): + """PPI/PROFIBUS frame delimiters.""" + + SD1 = 0x10 + SD2 = 0x68 + SD3 = 0xA2 + SC = 0xE5 + + +class PPIArea(IntEnum): + """S7-200 memory area identifiers.""" + + S = 0x03 + SM = 0x05 + AI = 0x06 + AQ = 0x07 + I = 0x81 # noqa: E741 - Siemens names the input area I. + Q = 0x82 + M = 0x83 + V = 0x84 # V memory is addressed as DB1 on the wire. + C = 0x1E + T = 0x1F + + +@dataclass(frozen=True) +class PPIFrame: + """Decoded PPI frame.""" + + frame_type: PPIFrameType + destination: int | None = None + source: int | None = None + control: int | None = None + payload: bytes = b"" + + +class SerialPort(Protocol): + """Minimal serial-port interface used by :class:`PPITransport`.""" + + def read(self, size: int = 1) -> bytes: ... + + def write(self, data: bytes) -> int | None: ... + + def close(self) -> None: ... + + +class PPIExchangeTransport(Protocol): + """Transport interface consumed by :class:`PPIClient`.""" + + def open(self) -> None: ... + + def close(self) -> None: ... + + def exchange(self, pdu: bytes) -> bytes: ... + + +def _validate_station(address: int) -> None: + if not 0 <= address <= 126: + raise ValueError(f"PPI station address must be between 0 and 126, got {address}") + + +def _checksum(data: bytes) -> int: + return sum(data) & 0xFF + + +def encode_sd1(destination: int, source: int, control: int) -> bytes: + """Encode a fixed-length SD1 frame.""" + _validate_station(destination) + _validate_station(source) + body = bytes((destination, source, control)) + return bytes((PPIFrameType.SD1,)) + body + bytes((_checksum(body), 0x16)) + + +def encode_sd2(destination: int, source: int, control: int, payload: bytes) -> bytes: + """Encode a variable-length SD2 frame.""" + _validate_station(destination) + _validate_station(source) + body = bytes((destination, source, control)) + payload + if len(body) > 249: + raise ValueError(f"SD2 body exceeds the 249-byte limit: {len(body)}") + length = len(body) + return bytes((PPIFrameType.SD2, length, length, PPIFrameType.SD2)) + body + bytes((_checksum(body), 0x16)) + + +def encode_sd3(destination: int, source: int, control: int, payload: bytes) -> bytes: + """Encode an SD3 frame with its fixed eight-byte data field.""" + _validate_station(destination) + _validate_station(source) + if len(payload) != 8: + raise ValueError("SD3 payload must contain exactly 8 bytes") + body = bytes((destination, source, control)) + payload + return bytes((PPIFrameType.SD3,)) + body + bytes((_checksum(body), 0x16)) + + +def decode_frame(data: bytes) -> PPIFrame: + """Decode and validate one complete PPI frame.""" + if not data: + raise S7ProtocolError("Empty PPI frame") + + try: + frame_type = PPIFrameType(data[0]) + except ValueError as exc: + raise S7ProtocolError(f"Unknown PPI start delimiter: 0x{data[0]:02x}") from exc + + if frame_type == PPIFrameType.SC: + if data != bytes((PPIFrameType.SC,)): + raise S7ProtocolError("SC acknowledgement must be exactly one byte") + return PPIFrame(frame_type) + + if frame_type == PPIFrameType.SD1: + if len(data) != 6: + raise S7ProtocolError(f"SD1 frame must be 6 bytes, got {len(data)}") + body = data[1:4] + elif frame_type == PPIFrameType.SD2: + if len(data) < 9: + raise S7ProtocolError("SD2 frame is too short") + if data[1] != data[2] or data[3] != PPIFrameType.SD2: + raise S7ProtocolError("Invalid SD2 repeated length or delimiter") + if len(data) != data[1] + 6: + raise S7ProtocolError(f"SD2 length mismatch: header says {data[1]}, frame has {len(data)} bytes") + body = data[4:-2] + else: + if len(data) != 14: + raise S7ProtocolError(f"SD3 frame must be 14 bytes, got {len(data)}") + body = data[1:-2] + + if data[-1] != 0x16: + raise S7ProtocolError("Invalid PPI end delimiter") + if data[-2] != _checksum(body): + raise S7ProtocolError("Invalid PPI frame checksum") + + return PPIFrame(frame_type, body[0], body[1], body[2], bytes(body[3:])) + + +class PPITransport: + """Serial PPI master transport for a single S7-200 slave.""" + + def __init__( + self, + port: str, + *, + station: int = 2, + local_station: int = 0, + baudrate: int = 9600, + timeout: float = 0.15, + retries: int = 3, + serial_port: SerialPort | None = None, + ) -> None: + _validate_station(station) + _validate_station(local_station) + if baudrate <= 0: + raise ValueError("baudrate must be positive") + if timeout <= 0: + raise ValueError("timeout must be positive") + if retries < 1: + raise ValueError("retries must be at least 1") + + self.port = port + self.station = station + self.local_station = local_station + self.baudrate = baudrate + self.timeout = timeout + self.retries = retries + self._serial = serial_port + self._owns_serial = serial_port is None + self._lock = threading.Lock() + + @property + def is_open(self) -> bool: + return self._serial is not None + + def open(self) -> None: + """Open the configured serial port using PPI's 8E1 settings.""" + if self._serial is not None: + return + try: + serial = importlib.import_module("serial") + except ImportError as exc: + raise ImportError("PPI support requires pyserial; install python-snap7[ppi]") from exc + + try: + self._serial = cast( + SerialPort, + serial.Serial( + port=self.port, + baudrate=self.baudrate, + bytesize=serial.EIGHTBITS, + parity=serial.PARITY_EVEN, + stopbits=serial.STOPBITS_ONE, + timeout=self.timeout, + write_timeout=self.timeout, + ), + ) + except Exception as exc: + raise S7ConnectionError(f"Could not open PPI serial port {self.port}: {exc}") from exc + + def close(self) -> None: + """Close a serial port opened by this transport.""" + if self._serial is not None and self._owns_serial: + self._serial.close() + self._serial = None + + def _read_exact(self, size: int) -> bytes: + assert self._serial is not None + data = bytearray() + while len(data) < size: + chunk = self._serial.read(size - len(data)) + if not chunk: + raise S7ConnectionError(f"Timeout while reading PPI frame ({len(data)}/{size} bytes)") + data.extend(chunk) + return bytes(data) + + def _read_frame(self) -> PPIFrame: + start = self._read_exact(1) + delimiter = start[0] + if delimiter == PPIFrameType.SC: + return decode_frame(start) + if delimiter == PPIFrameType.SD1: + return decode_frame(start + self._read_exact(5)) + if delimiter == PPIFrameType.SD2: + prefix = self._read_exact(3) + if prefix[0] != prefix[1] or prefix[2] != PPIFrameType.SD2: + raise S7ProtocolError("Invalid SD2 repeated length or delimiter") + return decode_frame(start + prefix + self._read_exact(prefix[0] + 2)) + if delimiter == PPIFrameType.SD3: + return decode_frame(start + self._read_exact(13)) + raise S7ProtocolError(f"Unknown PPI start delimiter: 0x{delimiter:02x}") + + def _write_frame(self, frame: bytes) -> None: + assert self._serial is not None + written = self._serial.write(frame) + if written is not None and written != len(frame): + raise S7ConnectionError(f"Short PPI serial write ({written}/{len(frame)} bytes)") + + def exchange(self, pdu: bytes) -> bytes: + """Exchange one S7 PDU with the configured S7-200 station.""" + if self._serial is None: + raise S7ConnectionError("PPI serial port is not open") + + request = encode_sd2(self.station, self.local_station, 0x6C, pdu) + with self._lock: + for attempt in range(self.retries): + self._write_frame(request) + try: + acknowledgement = self._read_frame() + except S7ConnectionError: + if attempt + 1 == self.retries: + raise + continue + if acknowledgement.frame_type != PPIFrameType.SC: + raise S7ProtocolError("Expected E5 acknowledgement after PPI request") + break + + poll_control = 0x5C + self._write_frame(encode_sd1(self.station, self.local_station, poll_control)) + for _ in range(self.retries * 2): + response = self._read_frame() + if response.frame_type == PPIFrameType.SC: + poll_control = 0x7C if poll_control == 0x5C else 0x5C + self._write_frame(encode_sd1(self.station, self.local_station, poll_control)) + continue + if response.frame_type != PPIFrameType.SD2: + raise S7ProtocolError(f"Expected SD2 PPI response, got {response.frame_type.name}") + if response.destination != self.local_station or response.source != self.station: + raise S7ProtocolError(f"Unexpected PPI response addresses: {response.source} -> {response.destination}") + return response.payload + + raise S7ConnectionError("PPI response was not available after polling") + + +class PPIClient: + """Minimal S7-200 client using PPI over a serial port.""" + + def __init__( + self, + port: str, + *, + station: int = 2, + local_station: int = 0, + baudrate: int = 9600, + timeout: float = 0.15, + transport: PPIExchangeTransport | None = None, + ) -> None: + self.transport = transport or PPITransport( + port, + station=station, + local_station=local_station, + baudrate=baudrate, + timeout=timeout, + ) + self.protocol = S7Protocol() + self.connected = False + self.pdu_length = 240 + + def connect(self) -> "PPIClient": + """Open the serial port and negotiate the S7 PDU length.""" + self.transport.open() + try: + request = self.protocol.build_setup_communication_request(pdu_length=self.pdu_length) + response = self._exchange(request) + parameters = response.get("parameters") or {} + negotiated = int(parameters.get("pdu_length", self.pdu_length)) + # SD2's one-byte length field permits at most 249 body bytes; + # three of those are the PPI destination/source/control fields. + self.pdu_length = min(negotiated, 246) + self.connected = True + except Exception: + self.transport.close() + raise + return self + + def disconnect(self) -> None: + """Close the PPI transport.""" + self.transport.close() + self.connected = False + + def _exchange(self, request: bytes) -> dict[str, Any]: + response = self.protocol.parse_response(self.transport.exchange(request)) + self.protocol.validate_pdu_reference(int(response["sequence"])) + return response + + @staticmethod + def _area_spec(area: PPIArea) -> tuple[S7Area, int, S7WordLen]: + if area == PPIArea.V: + return S7Area.DB, 1, S7WordLen.BYTE + if area in (PPIArea.AI, PPIArea.AQ): + return cast(S7Area, area), 0, S7WordLen.WORD + if area == PPIArea.C: + return cast(S7Area, area), 0, S7WordLen.COUNTER_200 + if area == PPIArea.T: + return cast(S7Area, area), 0, S7WordLen.TIMER_200 + return cast(S7Area, area), 0, S7WordLen.BYTE + + def read_area(self, area: PPIArea, start: int, count: int) -> bytearray: + """Read items from an S7-200 memory area. + + ``count`` is bytes for S/SM/I/Q/M/V and 16-bit items for AI/AQ/C/T. + """ + if not self.connected: + raise S7ConnectionError("PPI client is not connected") + if count < 1: + raise ValueError("count must be at least 1") + wire_area, db_number, word_len = self._area_spec(area) + item_size = 2 if word_len in (S7WordLen.WORD, S7WordLen.COUNTER_200, S7WordLen.TIMER_200) else 1 + if count * item_size > self.pdu_length - 18: + raise ValueError("PPI read exceeds the negotiated PDU size") + request = self.protocol.build_read_request(wire_area, db_number, start, word_len, count) + response = self._exchange(request) + return bytearray(self.protocol.extract_read_data(response, word_len, count)) + + def write_area(self, area: PPIArea, start: int, data: bytes | bytearray) -> None: + """Write data to an S7-200 memory area.""" + if not self.connected: + raise S7ConnectionError("PPI client is not connected") + wire_area, db_number, word_len = self._area_spec(area) + item_size = 2 if word_len in (S7WordLen.WORD, S7WordLen.COUNTER_200, S7WordLen.TIMER_200) else 1 + if not data or len(data) % item_size: + raise ValueError(f"data length must be a non-zero multiple of {item_size}") + if len(data) > self.pdu_length - 35: + raise ValueError("PPI write exceeds the negotiated PDU size") + request = self.protocol.build_write_request(wire_area, db_number, start, word_len, bytes(data)) + self.protocol.check_write_response(self._exchange(request)) + + def v_read(self, start: int, size: int) -> bytearray: + """Read bytes from S7-200 V memory (wire-level DB1).""" + return self.read_area(PPIArea.V, start, size) + + def v_write(self, start: int, data: bytes | bytearray) -> None: + """Write bytes to S7-200 V memory (wire-level DB1).""" + self.write_area(PPIArea.V, start, data) + + def __enter__(self) -> "PPIClient": + return self.connect() + + def __exit__( + self, + exc_type: type[BaseException] | None, + exc_value: BaseException | None, + traceback: TracebackType | None, + ) -> None: + self.disconnect() diff --git a/tests/test_ppi.py b/tests/test_ppi.py new file mode 100644 index 00000000..461cfa17 --- /dev/null +++ b/tests/test_ppi.py @@ -0,0 +1,233 @@ +"""Tests for S7-200 PPI frames, serial exchange, and client operations.""" + +import struct + +import pytest + +from snap7.error import S7ConnectionError, S7ProtocolError +from snap7.ppi import ( + PPIArea, + PPIClient, + PPIFrame, + PPIFrameType, + PPITransport, + decode_frame, + encode_sd1, + encode_sd2, + encode_sd3, +) +from snap7.s7protocol import S7Function, S7PDUType + + +class FakeSerial: + def __init__(self, incoming: bytes, *, chunk_size: int = 2) -> None: + self.incoming = bytearray(incoming) + self.chunk_size = chunk_size + self.writes: list[bytes] = [] + self.closed = False + + def read(self, size: int = 1) -> bytes: + count = min(size, self.chunk_size, len(self.incoming)) + data = bytes(self.incoming[:count]) + del self.incoming[:count] + return data + + def write(self, data: bytes) -> int: + self.writes.append(data) + return len(data) + + def close(self) -> None: + self.closed = True + + +def _response_pdu(sequence: int, function: int, *, data: bytes = b"", setup_pdu_length: int | None = None) -> bytes: + if setup_pdu_length is None: + parameters = bytes((function, 1)) + else: + parameters = struct.pack(">BBHHH", function, 0, 1, 1, setup_pdu_length) + header = struct.pack( + ">BBHHHHBB", + 0x32, + S7PDUType.ACK_DATA, + 0, + sequence, + len(parameters), + len(data), + 0, + 0, + ) + return header + parameters + data + + +class StubTransport: + def __init__(self) -> None: + self.requests: list[bytes] = [] + self.opened = False + + def open(self) -> None: + self.opened = True + + def close(self) -> None: + self.opened = False + + def exchange(self, pdu: bytes) -> bytes: + self.requests.append(pdu) + sequence = struct.unpack_from(">H", pdu, 4)[0] + function = pdu[10] + if function == S7Function.SETUP_COMMUNICATION: + return _response_pdu(sequence, function, setup_pdu_length=240) + if function == S7Function.READ_AREA: + count = struct.unpack_from(">H", pdu, 16)[0] + word_len = pdu[15] + byte_count = count * (2 if word_len in (0x04, 0x1E, 0x1F) else 1) + values = bytes(range(byte_count)) + data = struct.pack(">BBH", 0xFF, 0x04, byte_count * 8) + values + return _response_pdu(sequence, function, data=data) + return _response_pdu(sequence, function) + + +@pytest.mark.parametrize( + "frame", + [ + encode_sd1(2, 0, 0x5C), + encode_sd2(2, 0, 0x6C, b"\x32\x01"), + encode_sd3(2, 0, 0x03, bytes(range(8))), + bytes((PPIFrameType.SC,)), + ], +) +def test_frame_roundtrip(frame: bytes) -> None: + decoded = decode_frame(frame) + assert decoded.frame_type == frame[0] + + +def test_decode_sd2_fields() -> None: + assert decode_frame(encode_sd2(2, 0, 0x6C, b"payload")) == PPIFrame( + PPIFrameType.SD2, + destination=2, + source=0, + control=0x6C, + payload=b"payload", + ) + + +@pytest.mark.parametrize( + "frame", + [ + b"", + b"\x99", + b"\xe5\x00", + b"\x10\x02", + b"\x68\x03\x04\x68\x02\x00\x6c\x6e\x16", + b"\x68\x03\x03\x68\x02\x00\x6c\x00\x16", + ], +) +def test_malformed_frames_rejected(frame: bytes) -> None: + with pytest.raises(S7ProtocolError): + decode_frame(frame) + + +def test_frame_encoder_validation() -> None: + with pytest.raises(ValueError, match="station address"): + encode_sd1(127, 0, 0x5C) + with pytest.raises(ValueError, match="249-byte"): + encode_sd2(2, 0, 0x6C, bytes(247)) + with pytest.raises(ValueError, match="exactly 8"): + encode_sd3(2, 0, 0x03, b"short") + + +def test_serial_exchange_uses_sd2_ack_sd1_sd2_flow() -> None: + response_pdu = b"\x32\x03 response" + incoming = bytes((PPIFrameType.SC,)) + encode_sd2(0, 2, 0x08, response_pdu) + serial = FakeSerial(incoming) + transport = PPITransport("test", serial_port=serial) + + assert transport.exchange(b"request") == response_pdu + assert serial.writes == [encode_sd2(2, 0, 0x6C, b"request"), encode_sd1(2, 0, 0x5C)] + + +def test_serial_exchange_alternates_poll_control_after_e5() -> None: + incoming = bytes((PPIFrameType.SC, PPIFrameType.SC)) + encode_sd2(0, 2, 0x08, b"response") + serial = FakeSerial(incoming) + transport = PPITransport("test", serial_port=serial) + + assert transport.exchange(b"request") == b"response" + assert serial.writes[-2:] == [encode_sd1(2, 0, 0x5C), encode_sd1(2, 0, 0x7C)] + + +def test_serial_timeout_is_reported() -> None: + transport = PPITransport("test", retries=1, serial_port=FakeSerial(b"")) + with pytest.raises(S7ConnectionError, match="Timeout"): + transport.exchange(b"request") + + +def test_client_negotiates_and_reads_writes_v_memory_as_db1() -> None: + transport = StubTransport() + client = PPIClient("test", transport=transport).connect() + + assert client.pdu_length == 240 + assert client.v_read(3, 4) == bytearray(range(4)) + client.v_write(5, b"\x01\x02") + + read_request = transport.requests[1] + assert read_request[15] == 0x02 # BYTE + assert read_request[18:20] == b"\x00\x01" # DB1 + assert read_request[20] == 0x84 # DB/V area + assert read_request[21:24] == b"\x00\x00\x18" # byte 3 as a bit address + + write_request = transport.requests[2] + assert write_request[18:20] == b"\x00\x01" + assert write_request[21:24] == b"\x00\x00\x28" + + +def test_client_encodes_analog_and_counter_item_addresses() -> None: + transport = StubTransport() + client = PPIClient("test", transport=transport).connect() + + assert client.read_area(PPIArea.AI, 1, 2) == bytearray(range(4)) + assert client.read_area(PPIArea.C, 3, 2) == bytearray(range(4)) + + analog_request = transport.requests[1] + assert analog_request[15] == 0x04 # WORD + assert analog_request[20] == PPIArea.AI + assert analog_request[21:24] == b"\x00\x00\x08" + + counter_request = transport.requests[2] + assert counter_request[15] == PPIArea.C + assert counter_request[20] == PPIArea.C + assert counter_request[21:24] == b"\x00\x00\x03" # item index, not bit address + + +@pytest.mark.parametrize( + ("area", "wire_area", "db_number", "word_len"), + [ + (PPIArea.S, 0x03, 0, 0x02), + (PPIArea.SM, 0x05, 0, 0x02), + (PPIArea.AI, 0x06, 0, 0x04), + (PPIArea.AQ, 0x07, 0, 0x04), + (PPIArea.I, 0x81, 0, 0x02), + (PPIArea.Q, 0x82, 0, 0x02), + (PPIArea.M, 0x83, 0, 0x02), + (PPIArea.V, 0x84, 1, 0x02), + (PPIArea.C, 0x1E, 0, 0x1E), + (PPIArea.T, 0x1F, 0, 0x1F), + ], +) +def test_s7_200_area_mapping(area: PPIArea, wire_area: int, db_number: int, word_len: int) -> None: + mapped_area, mapped_db, mapped_word_len = PPIClient._area_spec(area) + assert int(mapped_area) == wire_area + assert mapped_db == db_number + assert int(mapped_word_len) == word_len + + +def test_client_requires_connection_and_aligned_word_data() -> None: + client = PPIClient("test", transport=StubTransport()) + with pytest.raises(S7ConnectionError, match="not connected"): + client.v_read(0, 1) + + client.connect() + with pytest.raises(ValueError, match="multiple of 2"): + client.write_area(PPIArea.AQ, 0, b"\x01") + with pytest.raises(ValueError, match="negotiated PDU"): + client.v_read(0, client.pdu_length) + with pytest.raises(ValueError, match="negotiated PDU"): + client.v_write(0, bytes(client.pdu_length)) diff --git a/uv.lock b/uv.lock index c6c006eb..8bdffa52 100644 --- a/uv.lock +++ b/uv.lock @@ -1025,6 +1025,15 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/79/d7/29e1e5e882f79133631f7bcace42d23db493f616463c157a1ab614bf69dd/pyproject_api-1.10.1-py3-none-any.whl", hash = "sha256:fa9e6f66c35b5017e909825d8f2b5d5482ea699d7be809d21c03bd1f7317f36a", size = 12992, upload-time = "2026-05-28T14:22:12.711Z" }, ] +[[package]] +name = "pyserial" +version = "3.5" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/1e/7d/ae3f0a63f41e4d2f6cb66a5b57197850f919f59e558159a4dd3a818f5082/pyserial-3.5.tar.gz", hash = "sha256:3c77e014170dfffbd816e6ffc205e9842efb10be9f58ec16d3e8675b4925cddb", size = 159125, upload-time = "2020-11-23T03:59:15.045Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/07/bc/587a445451b253b285629263eb51c2d8e9bcea4fc97826266d186f96f558/pyserial-3.5-py2.py3-none-any.whl", hash = "sha256:c4451db6ba391ca6ca299fb3ec7bae67a5c55dde170964c7a14ceefec02f2cf0", size = 90585, upload-time = "2020-11-23T03:59:13.41Z" }, +] + [[package]] name = "pytest" version = "9.1.1" @@ -1134,6 +1143,9 @@ doc = [ { name = "sphinx", version = "9.1.0", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version >= '3.12'" }, { name = "sphinx-rtd-theme" }, ] +ppi = [ + { name = "pyserial" }, +] s7commplus = [ { name = "cryptography" }, ] @@ -1161,6 +1173,7 @@ requires-dist = [ { name = "mypy", marker = "extra == 'test'" }, { name = "pnio-dcp", marker = "extra == 'discovery'" }, { name = "psutil", marker = "extra == 'demo'" }, + { name = "pyserial", marker = "extra == 'ppi'", specifier = ">=3.5" }, { name = "pytest", marker = "extra == 'test'" }, { name = "pytest-asyncio", marker = "extra == 'test'" }, { name = "pytest-cov", marker = "extra == 'test'" }, @@ -1176,7 +1189,7 @@ requires-dist = [ { name = "types-setuptools", marker = "extra == 'test'" }, { name = "uv", marker = "extra == 'test'" }, ] -provides-extras = ["test", "s7commplus", "cli", "demo", "doc", "discovery"] +provides-extras = ["test", "s7commplus", "cli", "demo", "doc", "discovery", "ppi"] [[package]] name = "requests"