pyS7 is a lightweight, pure Python library that implements the Siemens S7 communication protocol over ISO-on-TCP (RFC1006). It enables direct communication with Siemens S7-200, S7-300, S7-400, S7-1200, and S7-1500 PLCs from Python applications.
Production/Stable • 2x Performance • Fully Tested
⚠️ Neither this project nor its authors are affiliated with Siemens. S7-200, S7-300, S7-400, S7-1200, and S7-1500 are registered trademarks of Siemens AG.
- Pure Python – No external dependencies, easy installation across platforms
- Intuitive API – Clean, readable code with full typing support for IDE assistance
- High Performance – Optimized hot paths, 2x faster request preparation (v2.5.0)
- Graceful error handling – read_detailed() and write_detailed() provide per-tag success/failure info
- Transactional writes – Batch write with automatic rollback on read verification failure
- Optimized multi-variable reads – Automatically groups contiguous tags to reduce network requests
- Automatic chunking – Transparently splits large STRING/WSTRING reads exceeding PDU size
- CPU diagnostics – Read PLC status (RUN/STOP) and information (model, firmware) via SZL protocol
- Broad compatibility – Supports S7-200/300/400/1200/1500 series
- Production Ready – 253 tests, 85% coverage, strict type checking
Industrial safety must always remain your top priority. By using pyS7 you accept full responsibility for any damage, data loss, downtime, or unintended effects. Understand your system and the implications of each operation before interacting with live equipment.
Requires Python 3.8 or later.
pip install pys7Or install from GitHub:
pip install git+https://github.com/xtimmy86x/pyS7from pyS7 import S7Client
with S7Client(address="192.168.5.100", rack=0, slot=1) as client:
tags = [
"DB1,X0.0", # Bit 0 of DB1
"DB1,I30", # INT at byte 30 of DB1
"M54.4", # Bit 4 of marker memory
"IW22", # WORD at byte 22 of input area
"QR24", # REAL at byte 24 of output area
"DB1,S10.5" # String of 5 characters at byte 10 of DB1
]
data = client.read(tags)
print(data) # [True, 123, True, 10, 3.14, 'Hello']from pyS7 import S7Client
with S7Client(address="192.168.5.100", rack=0, slot=1) as client:
tags = ["DB1,X0.0", "DB1,I30", "DB1,R40", "DB1,S10.5"]
values = [True, 25000, 1.2345, "Hello"]
client.write(tags, values)from pyS7 import S7Client
with S7Client(address="192.168.5.100", rack=0, slot=1) as client:
# read_detailed() continues on errors, returns per-tag results
results = client.read_detailed(["DB1,I0", "DB99,I0", "DB1,R4"])
for result in results:
if result.success:
print(f"{result.tag}: {result.value}")
else:
print(f"{result.tag} failed: {result.error}")
# write_detailed() provides per-tag success/failure info
tags = ["DB1,I0", "DB1,I2", "DB99,I0"]
values = [100, 200, 300]
write_results = client.write_detailed(tags, values)
for result in write_results:
if result.success:
print(f"✓ {result.tag}: Written")
else:
print(f"✗ {result.tag}: {result.error}")from pyS7 import S7Client
with S7Client(address="192.168.5.100", rack=0, slot=1) as client:
# Batch write with automatic rollback on verification failure
with client.batch_write() as batch:
batch.add("DB1,I0", 100)
batch.add("DB1,I2", 200)
batch.add("DB1,R4", 3.14)
# Auto-commits on exit, rolls back on error
# Or with explicit control
batch = client.batch_write(auto_commit=False)
batch.add("DB1,I0", 100)
batch.add("DB1,I2", 200)
try:
batch.commit() # Write and verify
except Exception as e:
batch.rollback() # Restore original values
print(f"Write failed: {e}")from pyS7 import S7Client
with S7Client(address="192.168.5.100", rack=0, slot=1) as client:
# Get CPU status
status = client.get_cpu_status()
print(f"CPU Status: {status}") # "RUN" or "STOP"
# Get CPU information
info = client.get_cpu_info()
print(f"Model: {info['module_type_name']}")
print(f"Firmware: {info['firmware_version']}")
print(f"Hardware: {info['hardware_version']}")See docs/CPU_STATUS_READING.md for details.
pyS7 supports both ASCII and Unicode strings:
# STRING (ASCII) - All S7 models
tags = ["DB1,S10.20"] # STRING at byte 10, max 20 chars
data = client.read(tags)
print(data[0]) # "Hello World"
# WSTRING (Unicode) - S7-1200/1500 only
tags = ["DB1,WS100.30"] # WSTRING at byte 100, max 30 chars
data = client.read(tags)
print(data[0]) # "Hello 世界! 🌍"
# Large strings automatically chunked if exceeding PDU size
tags = ["DB1,S100.254"] # STRING[254] - handled transparently
data = client.read(tags) # Complete string returned- API Reference – Data types, address formats, supported operations
- Advanced Usage – TSAP connections, PDU tuning, chunking, multi-threading
- Best Practices – Connection management, error handling, production deployment
- Troubleshooting – Common issues and solutions
- CPU Status Reading – CPU diagnostics details
- CPU Info Technical Notes – CPU information internals
- Bit Read Troubleshooting – Bit operation guidance
- Supported address formats – Complete address mapping
- TSAP connection – Alternative connection method
- PDU optimization – Performance tuning
- Error handling – Robust error patterns
- Connection issues – Can't connect?
Example scripts in the examples/ directory demonstrate:
read_data.py– Basic reading operationswrite_data.py– Basic writing operationsread_detailed_demo.py– Graceful error handling for readswrite_detailed_demo.py– Graceful error handling for writesbatch_write_demo.py– Transactional batch writes with rollbackget_cpu_status.py– CPU status monitoringget_cpu_info.py– CPU information retrievalread_data_tsap.py– TSAP connection examplebit_read_workaround.py– Bit operationsmanage_reconnection.py– Connection handlingconnection_state_demo.py– Connection state management
This project is distributed under the MIT License. See the LICENSE file for details.
Special thanks to filocara for the original project that inspired this work.