python-s7comm is an unofficial Python implementation of the Siemens S7 communication protocol, enabling communication with Siemens S7 PLCs (Programmable Logic Controllers). The library provides both synchronous and asynchronous clients at two abstraction levels.
-
High-Level Clients (
python_s7comm/sync_client.py,async_client.py)Client- Synchronous high-level API with string-based addressesAsyncClient- Asynchronous high-level API with string-based addresses- Provides convenience methods:
connect(),disconnect(),close()read_area(),write_area()read_multi_vars(),write_multi_vars()get_cpu_state(),get_order_code()read_szl(),read_szl_list()plc_stop()
-
Core S7Comm (
python_s7comm/s7comm/client.py,async_client.py)S7Comm- Synchronous core implementationAsyncS7Comm- Asynchronous core implementationBaseS7Comm- Shared base class with common logic- Handles PDU reference management, packet validation, userdata continuation
-
Transport Layer (
python_s7comm/s7comm/transport/)COTP/AsyncCOTP- Connection-Oriented Transport ProtocolTPKT- ISO Transport Service on top of TCPTransport/AsyncTransport- Protocol interfaces
python_s7comm/- Main packagesync_client.py- High-level synchronousClientclassasync_client.py- High-level asynchronousAsyncClientclasss7comm/- Core S7 protocol implementationclient.py- CoreS7Commclassasync_client.py- CoreAsyncS7Commclassbase.py-BaseS7Commshared logicenums.py- Protocol enumerationsszl.py- System Status List (SZL) parsingexceptions.py- S7Comm exceptionserror_messages.py- Error code mappingspackets/- S7 protocol packet definitionstransport/- Transport layer (COTP, TPKT)
High-Level API: Client/AsyncClient (python_s7comm.sync_client/async_client)
↓
Core S7Comm: S7Comm/AsyncS7Comm (python_s7comm.s7comm.client/async_client)
↓
Transport Layer: COTP/AsyncCOTP (python_s7comm.s7comm.transport.cotp)
↓
Network Layer: TPKT (python_s7comm.s7comm.transport.cotp.tpkt)
↓
TCP/IP Socket
- Python 3.12+ required (tested on 3.12 and 3.13)
- Line length: 120 characters max (configured in ruff)
- Formatting: Use
ruff format - Linting: Use
rufffor linting - Type checking: Use
mypy(strict mode) - Type hints: Required for all function signatures
- Classes:
PascalCase(e.g.,S7Comm,VariableAddress,SetupCommunicationRequest) - Functions/methods:
snake_case(e.g.,read_area,serialize_parameter) - Constants:
UPPER_SNAKE_CASE(e.g.,MAX_VARS,PROTOCOL_ID) - Enums:
PascalCaseclass names,UPPER_SNAKE_CASEmembers
- Use
@dataclassfor data structures (seeVariableAddress) - Use
Protocolclasses for interfaces (seeTransport,S7Packet) - Use
structmodule for binary serialization/deserialization - Implement
serialize()andparse()methods for protocol packets - Use
@classmethodfor factory/parse methods
- Add docstrings for public API methods
- Keep inline comments minimal and meaningful
- All comments and documentation must be in English
- pytest with pytest-asyncio for async tests
- pytest-cov for coverage
asyncio_mode = "auto"configured inpyproject.toml
tests/
├── test_*.py # Unit tests (no hardware required)
├── s1200/ # Protocol unit tests (pcap-based, no hardware)
│ └── test_*.py # Tests using captured packet data
└── real_plc_s1200/ # Integration tests (requires PLC or simulator)
├── conftest.py # Fixtures with PLC connection
├── data_items.py # Test data definitions
└── test_*.py # Integration test files
# Unit tests only
uv run pytest tests/test_*.py
# All tests (requires PLC/simulator)
uv run pytest
# With coverage
uv run pytest --cov=python_s7commIntegration tests require environment variables or .env file:
PLC_IP=127.0.0.1
PLC_RACK=0
PLC_SLOT=1
PLC_PORT=1102
- Use
@pytest.mark.parametrizefor multiple test cases - Use fixtures from
conftest.pyfor client connections - Test both serialization and parsing (round-trip)
- Include edge cases and invalid input tests
# Install with dev dependencies
uv sync --group dev --group test
# Install pre-commit hooks
uv run pre-commit install
# Run pre-commit on all files
uv run pre-commit run --all-filesConfigured hooks (.pre-commit-config.yaml):
trailing-whitespace,end-of-file-fixercheck-ast,check-json,check-toml,check-yamlmypyfor type checkingrufffor lintingruff-formatfor formatting
P(0x80) - PeripheralI(0x81) - InputsQ(0x82) - OutputsM(0x83) - MarkersDB(0x84) - Data blocksC(0x1C) - CountersT(0x1D) - Timers
BOOL,BYTE,CHAR,WORD,INT,DWORD,DINT,REAL,COUNTER,TIMER
JobRequest(0x01) - Request sent by the masterAck(0x02) - Simple acknowledgement with no dataResponse(0x03) - Ack-Data responseUserdata(0x07) - Extended protocol for SZL, CPU functionsServerControl(0x08) - Server control request
SetupCommunication(0xF0)ReadVariable(0x04)WriteVariable(0x05)PLCControl(0x28)PLCStop(0x29)
S7CommException- Base exceptionReadVariableException- Read operation failurePacketLostError- PDU reference mismatch (lost packet)StalePacketError- Received old packet (retry receive)
DB area: DB<db_number>.<start>[.<bit>] <data_type> <amount>
Other areas: <area><start>[.<bit>] <data_type> <amount>
Examples:
DB1.100 BYTE 10 # Read 10 bytes from DB1 at offset 100
DB5.0.0 BOOL 1 # Read bit 0 from DB5 offset 0
M0 INT 5 # Read 5 INTs from Marker area
I0.0 BOOL 1 # Read input bit
- PDU Reference: Auto-incremented 16-bit counter for request/response matching
- PDU Length: Negotiated during connection setup (default 480 bytes, max depends on PLC model)
- Max Variables: 20 items per multi-read/write operation (
MAX_VARS) - Byte Order: Big-endian (
!in struct format) for protocol, some little-endian for PDU reference - Protocol ID: Always
0x32for S7comm
- Create packet class in
python_s7comm/s7comm/packets/ - Implement
S7Packetprotocol (serialize_parameter, serialize_data) - Add
parse()classmethod for deserialization - Register in
S7PacketParser.parse()method - Add unit tests
- Add to relevant enums in
enums.py - Update
VariableAddressregex patterns if needed - Add tests in
test_addresses.py
- Update version in
pyproject.toml - Run all checks:
uv run pre-commit run --all-files - Run tests:
uv run pytest tests/ --ignore=tests/real_plc_s1200/ - Create and push tag:
git tag -a v0.0.1 -m "v0.0.1" && git push origin v0.0.1 - GitHub Actions will automatically build and publish to PyPI