Metadata-Version: 2.4
Name: python-s7comm
Version: 0.1.1
Summary: Unofficial Python implementation of Siemens S7 communication protocol
Author: nikteliy
License-Expression: MIT
Project-URL: Homepage, https://github.com/nikteliy/python-s7comm
Project-URL: Repository, https://github.com/nikteliy/python-s7comm
Project-URL: Issues, https://github.com/nikteliy/python-s7comm/issues
Keywords: siemens,s7,plc,s7comm,automation,industrial
Classifier: Development Status :: 4 - Beta
Classifier: Intended Audience :: Developers
Classifier: Operating System :: OS Independent
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Topic :: System :: Hardware
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Requires-Python: >=3.12
Description-Content-Type: text/markdown
License-File: LICENSE
Dynamic: license-file

# python-s7comm

Unofficial Python implementation of the Siemens S7 communication protocol for interacting with Siemens S7 PLCs

> **Note:** I only have an S7-1200 to test with, so I can't guarantee it works with other models (S7-300, S7-400, S7-1500).

> **Disclaimer:**
> This project is provided "as is", without warranty of any kind, express or implied.
> The author assumes no liability for any damage, loss, downtime, or safety issues resulting from the use of this software, including but not limited to damage to PLCs, industrial equipment, production systems, or data.
>
> This library is an unofficial, experimental implementation of the S7 protocol and is not intended for production or safety-critical use. It was created primarily for educational and research purposes.
>
> The software has not been certified, validated, or tested for industrial deployment. Use at your own risk.
>
> Always test thoroughly in a safe, isolated environment before connecting to live equipment.
>
> This project is not affiliated with or endorsed by Siemens AG.

## Features

- **Synchronous and Asynchronous clients** - Choose between `Client` and `AsyncClient`
- **High-level API** - Simple string-based addressing (`"DB1.0 INT 1"`)
- **Low-level API** - Direct access to `S7Comm` / `AsyncS7Comm` for advanced use cases
- **Read/Write operations** - Single and multi-variable read/write support
- **SZL (System Status List)** - Read CPU state, module identification, and other diagnostic information
- **PLC Control** - Stop PLC execution
- **Type-safe** - Full type hints with mypy strict mode

## Installation

```bash
pip install python-s7comm
```

Or with [uv](https://github.com/astral-sh/uv):

```bash
uv add python-s7comm
```

## Quick Start

### Synchronous Client

```python
from python_s7comm import Client

# Connect to PLC
client = Client()
client.connect(address="192.168.0.1", rack=0, slot=1)

# Read data
data = client.read_area("DB1.0 INT 1")  # Read 1 INT from DB1 at offset 0
print(int.from_bytes(data, "big", signed=True))

# Write data
client.write_area("DB1.0 INT 1", (42).to_bytes(2, "big", signed=True))

# Read multiple variables
results = client.read_multi_vars([
    "DB1.0 INT 1",
    "DB1.2 REAL 1",
    "M0 BYTE 4",
])

# Get CPU state
cpu_state = client.get_cpu_state()
print(f"CPU is in {cpu_state.name} mode")

# Disconnect
client.disconnect()
```

### Using Context Manager

```python
from python_s7comm import Client

with Client() as client:
    client.connect(address="192.168.0.1", rack=0, slot=1)
    data = client.read_area("DB1.0 BYTE 10")
```

### Asynchronous Client

```python
import asyncio
from python_s7comm import AsyncClient

async def main():
    client = AsyncClient()
    await client.connect(address="192.168.0.1", rack=0, slot=1)

    # Read data
    data = await client.read_area("DB1.0 DINT 2")

    # Write data
    await client.write_area("DB1.100 BYTE 4", b"\x01\x02\x03\x04")

    # Read SZL
    order_code = await client.get_order_code()
    print(f"PLC Order Code: {order_code}")

    await client.disconnect()

asyncio.run(main())
```

> **Note:** `AsyncClient` uses an internal lock to send requests sequentially. The S7 protocol does not support concurrent requests on a single connection. But if you are brave enough, you can create multiple `AsyncClient` instances (e.g., one for reading, another for writing) and use them within the same event loop:

```python
import asyncio
import struct
from python_s7comm import AsyncClient


async def periodic_reader(client: AsyncClient):
    """Reads data every 1 second - runs independently."""
    data = None
    while True:
        response = await client.read_area("DB2.0 INT 1")
        new_value = int.from_bytes(response, "big", signed=True)
        if new_value != data:
            print(f"Data changed from {data} to {new_value}")
            data = new_value

        await asyncio.sleep(1)


async def writer(client: AsyncClient):
    """Performs writes - doesn't block the reader."""
    for i in range(5):
        await client.write_area("DB2.0 INT 1", struct.pack("!H", i))
        print(f"Write: {i}")
        await asyncio.sleep(3)


async def main():
    reader = AsyncClient()
    writer_client = AsyncClient()

    await reader.connect(address="192.168.0.1", rack=0, slot=1)
    await writer_client.connect(address="192.168.0.1", rack=0, slot=1)

    # Reader and writer run concurrently without blocking each other
    read_task = asyncio.create_task(periodic_reader(reader))
    write_task = asyncio.create_task(writer(writer_client))

    await write_task  # Wait for writes to complete
    read_task.cancel()  # Stop the periodic reader

    await reader.disconnect()
    await writer_client.disconnect()

asyncio.run(main())

```

## Address Format

```
DB area:     DB<number>.<offset>[.<bit>] <type> <count>
Other areas: <area><offset>[.<bit>] <type> <count>
```

### Examples

| Address | Description |
|---------|-------------|
| `DB1.0 INT 1` | 1 INT from DB1 at byte 0 |
| `DB1.100 BYTE 10` | 10 BYTEs from DB1 at byte 100 |
| `DB5.0.0 BOOL 1` | Bit 0 from DB5 at byte 0 |
| `M0 BYTE 4` | 4 BYTEs from Marker area |
| `I0.0 BOOL 1` | Input bit 0.0 |
| `Q0 BYTE 1` | 1 BYTE from Output area |

### Supported Areas

| Area | Code | Description |
|------|------|-------------|
| `DB` | 0x84 | Data Blocks |
| `M` | 0x83 | Markers (Flags) |
| `I` | 0x81 | Inputs |
| `Q` | 0x82 | Outputs |
| `C` | 0x1C | Counters |
| `T` | 0x1D | Timers |
| `P` | 0x80 | Peripheral I/O |

### Supported Data Types

`BOOL`, `BYTE`, `CHAR`, `WORD`, `INT`, `DWORD`, `DINT`, `REAL`, `COUNTER`, `TIMER`

## Connection Parameters

```python
client = Client(
    tpdu_size=1024,      # COTP packet size
    pdu_length=480,      # S7 PDU length (negotiated)
    source_tsap=0x0100,  # Source TSAP
    dest_tsap=0x0101,    # Destination TSAP
)

client.connect(
    address="192.168.0.1",
    rack=0,              # Rack number
    slot=1,              # Slot number (CPU slot)
    port=102,            # ISO-on-TCP port (default: 102)
)
```

## Requirements

- Python 3.12 or newer

## Tested On

- Siemens S7-1200 (6ES7 214-1HG40-0XB0)

## Sources

- [snap7](https://snap7.sourceforge.net/)
- [python-snap7](https://github.com/gijzelaerr/python-snap7)
- [tpkt wireshark](https://wiki.wireshark.org/TPKT)
- [rfc905](https://datatracker.ietf.org/doc/html/rfc905)
- [rfc1006](https://www.ietf.org/rfc/rfc1006.txt)
- [s7-pcaps](https://github.com/gymgit/s7-pcaps)
- [@gymgit article](http://gmiru.com/article/s7comm/)
- [@gymgit article](http://gmiru.com/article/s7comm-part2/)
- [plc4x s7 protocol overview](https://plc4x.incubator.apache.org/)
- [plc4x s7 implementation](https://github.com/apache/plc4x/tree/develop/plc4c/driverss7)
