Client

The s7 package is the recommended entry point for communicating with any supported Siemens S7 PLC. It provides a unified client that works with all PLC models – S7-300, S7-400, S7-1200 and S7-1500 – and automatically selects the best protocol (S7CommPlus or legacy S7).

Synchronous client

from s7 import Client

client = Client()
client.connect("192.168.1.10", 0, 1)
data = client.db_read(1, 0, 4)
print(client.protocol)   # Protocol.S7COMMPLUS or Protocol.LEGACY
client.disconnect()

Asynchronous client

import asyncio
from s7 import AsyncClient

async def main():
    client = AsyncClient()
    await client.connect("192.168.1.10", 0, 1)
    data = await client.db_read(1, 0, 4)
    await client.disconnect()

asyncio.run(main())

V2 connection with TLS

S7-1500 PLCs with firmware 2.x use S7CommPlus V2, which requires TLS. Pass use_tls=True to the connect() method:

from s7 import Client

client = Client()
client.connect("192.168.1.10", 0, 1, use_tls=True)
data = client.db_read(1, 0, 4)
client.disconnect()

For PLCs with custom certificates, provide the certificate paths:

client.connect(
    "192.168.1.10", 0, 1,
    use_tls=True,
    tls_cert="/path/to/client.pem",
    tls_key="/path/to/client.key",
    tls_ca="/path/to/ca.pem",
)

Password authentication

Password-protected PLCs require the password keyword argument:

from s7 import Client

client = Client()
client.connect("192.168.1.10", 0, 1, use_tls=True, password="my_plc_password")
data = client.db_read(1, 0, 4)
client.disconnect()

Protocol selection

By default the client uses Protocol.AUTO which tries S7CommPlus first. You can force a specific protocol:

from s7 import Client, Protocol

# Force legacy S7 only
client = Client()
client.connect("192.168.1.10", 0, 1, protocol=Protocol.LEGACY)

# Force S7CommPlus (raises on failure)
client.connect("192.168.1.10", 0, 1, protocol=Protocol.S7COMMPLUS)

Concurrent async reads

An internal asyncio.Lock serialises each send/receive cycle so that multiple coroutines can safely share a single connection:

results = await asyncio.gather(
    client.db_read(1, 0, 4),
    client.db_read(1, 10, 4),
)

s7.Client

Unified S7 client with protocol auto-discovery.

Provides a single client that automatically selects the best protocol (S7CommPlus or legacy S7) for communicating with Siemens S7 PLCs.

Usage:

from s7 import Client

client = Client()
client.connect("192.168.1.10", 0, 1)
data = client.db_read(1, 0, 4)

class s7.client.Client[source]

Unified S7 client with protocol auto-discovery.

Automatically selects the best protocol for the target PLC: - S7CommPlus for S7-1200/1500 PLCs with full data operations - Legacy S7 for S7-300/400 or when S7CommPlus is unavailable

Methods not explicitly defined are delegated to the underlying legacy client via __getattr__.

Example:

from s7 import Client

client = Client()
client.connect("192.168.1.10", 0, 1)
data = client.db_read(1, 0, 4)
print(client.protocol)

__getattr__(name: str) → Any[source]: Delegate unknown methods to the legacy client.

browse() → list[dict[str, Any]][source]

Browse the PLC symbol table.

Warning

This method is experimental and may change.

Returns a flat list of variable info dicts. Can be converted to Tag objects:

from snap7 import Tag
variables = client.browse()
tags = {v["name"]: Tag(Area.DB, v["db_number"], v["byte_offset"], v["data_type"]) for v in variables}

Requires S7CommPlus connection.

connect(address: str, rack: int = 0, slot: int = 1, tcp_port: int = 102, *, protocol: Protocol = Protocol.AUTO, use_tls: bool = False, tls_cert: str | None = None, tls_key: str | None = None, tls_ca: str | None = None, password: str | None = None) → Client[source]

Connect to an S7 PLC.

Parameters:

address – PLC IP address or hostname.
rack – PLC rack number.
slot – PLC slot number.
tcp_port – TCP port (default 102).
protocol – Protocol selection. AUTO tries S7CommPlus first, then falls back to legacy S7.
use_tls – Whether to activate TLS (required for V2+).
tls_cert – Path to client TLS certificate (PEM).
tls_key – Path to client private key (PEM).
tls_ca – Path to CA certificate for PLC verification (PEM).
password – PLC password for legitimation (V2+ with TLS).

Returns:

self, for method chaining.

property connected: bool: Whether the client is connected to a PLC.

create_subscription(items: list[tuple[int, int, int]], cycle_ms: int = 0) → int[source]

Create a data change subscription (S7CommPlus only).

Warning

This method is experimental and may change.

Parameters:

items – List of (db_number, start_offset, size) tuples.
cycle_ms – Cycle time in milliseconds (0 = on change).

Returns:

Subscription ID.

db_read(db_number: int, start: int, size: int) → bytearray[source]

Read raw bytes from a data block.

Uses S7CommPlus when available, otherwise legacy S7.

db_read_multi(items: list[tuple[int, int, int]]) → list[bytearray][source]

Read multiple data block regions in a single request.

Uses S7CommPlus native multi-read when available.

db_write(db_number: int, start: int, data: bytearray) → int[source]

Write raw bytes to a data block.

Uses S7CommPlus when available, otherwise legacy S7.

Returns:: 0 on success (matches snap7.Client).

delete_subscription(subscription_id: int) → None[source]: Delete a data change subscription (S7CommPlus only).

Warning

This method is experimental and may change.

disconnect() → int[source]

Disconnect from PLC.

Returns:: 0 on success (matches snap7.Client).

download_block(block_type: int, block_number: int, data: bytes) → None[source]

Download (write) a program block to the PLC.

Warning

This method is experimental and may change.

Uses S7CommPlus when available, otherwise falls back to legacy download.

explore(explore_id: int = 0) → bytes[source]

Browse the PLC object tree (S7CommPlus only).

Parameters:: explore_id – Object to explore (0 = root).
Raises:: RuntimeError – If not connected via S7CommPlus.

list_datablocks() → list[dict[str, Any]][source]

List all datablocks on the PLC.

Warning

This method is experimental and may change.

Uses S7CommPlus EXPLORE when available, otherwise falls back to legacy list_blocks_of_type.

Returns:: List of dicts with keys name, number, rid.

property protocol: Protocol: The protocol currently in use for DB operations.

read_diagnostic_buffer() → list[dict[str, Any]][source]

Read the PLC diagnostic buffer.

Warning

This method is experimental and may change.

Uses the legacy S7 protocol (SZL read).

read_tag(tag: Any) → Any[source]

Read a typed value by Tag or address string.

For symbolic tags (with access_sequence set), routes to S7CommPlus LID-based access. For classic tags (byte-offset), delegates to the legacy client.

Parameters:: tag – A Tag instance or address string.
Returns:: The typed value.

read_tags(tags: list[Any]) → list[Any][source]: Read multiple tags, routing each to the appropriate protocol.

upload_block(block_type: int, block_number: int) → bytes[source]

Upload (read) a program block from the PLC.

Warning

This method is experimental and may change.

Uses S7CommPlus when available, otherwise falls back to legacy full_upload.

write_tag(tag: Any, value: Any) → int[source]: Write a typed value by Tag or address string.

s7.AsyncClient

Unified async S7 client with protocol auto-discovery.

Provides a single async client that automatically selects the best protocol (S7CommPlus or legacy S7) for communicating with Siemens S7 PLCs.

Usage:

from s7 import AsyncClient

async with AsyncClient() as client:
    await client.connect("192.168.1.10", 0, 1)
    data = await client.db_read(1, 0, 4)

class s7.async_client.AsyncClient[source]

Unified async S7 client with protocol auto-discovery.

Async counterpart of s7.Client. Automatically selects the best protocol for the target PLC using asyncio for non-blocking I/O.

Methods not explicitly defined are delegated to the underlying legacy async client via __getattr__.

Example:

from s7 import AsyncClient

async with AsyncClient() as client:
    await client.connect("192.168.1.10", 0, 1)
    data = await client.db_read(1, 0, 4)
    print(client.protocol)

__getattr__(name: str) → Any[source]: Delegate unknown methods to the legacy client.

async connect(address: str, rack: int = 0, slot: int = 1, tcp_port: int = 102, *, protocol: Protocol = Protocol.AUTO, use_tls: bool = False, tls_cert: str | None = None, tls_key: str | None = None, tls_ca: str | None = None) → AsyncClient[source]

Connect to an S7 PLC.

Parameters:

address – PLC IP address or hostname.
rack – PLC rack number.
slot – PLC slot number.
tcp_port – TCP port (default 102).
protocol – Protocol selection. AUTO tries S7CommPlus first, then falls back to legacy S7.
use_tls – Whether to activate TLS after InitSSL.
tls_cert – Path to client TLS certificate (PEM).
tls_key – Path to client private key (PEM).
tls_ca – Path to CA certificate for PLC verification (PEM).

Returns:

self, for method chaining.

property connected: bool: Whether the client is connected to a PLC.

async db_read(db_number: int, start: int, size: int) → bytearray[source]: Read raw bytes from a data block.

async db_read_multi(items: list[tuple[int, int, int]]) → list[bytearray][source]: Read multiple data block regions in a single request.

async db_write(db_number: int, start: int, data: bytearray) → int[source]

Write raw bytes to a data block.

Returns:: 0 on success (matches snap7.AsyncClient).

async disconnect() → int[source]

Disconnect from PLC.

Returns:: 0 on success (matches snap7.AsyncClient).

async explore() → bytes[source]

Browse the PLC object tree (S7CommPlus only).

Raises:: RuntimeError – If not connected via S7CommPlus.

property protocol: Protocol: The protocol currently in use for DB operations.

snap7.Client (legacy)

The snap7.Client is the legacy S7 protocol client. For new projects, use s7.Client above instead.

Legacy S7 client implementation.

Pure Python implementation of the classic S7 protocol. For new projects, use s7.Client instead, which supports all PLC models and automatically selects the best protocol.

class snap7.client.Client(lib_location: str | None = None, *, auto_reconnect: bool = False, max_retries: int = 3, retry_delay: float = 1.0, backoff_factor: float = 2.0, max_delay: float = 30.0, heartbeat_interval: float = 0, on_disconnect: Callable[[], None] | None = None, on_reconnect: Callable[[], None] | None = None, **kwargs: Any)[source]

Legacy S7 client for classic PUT/GET communication.

Supports S7-300, S7-400, S7-1200 and S7-1500 PLCs via the classic S7 protocol. For new projects, use s7.Client instead, which automatically selects the best protocol for any supported PLC.

Examples

>>> from s7 import Client
>>> client = Client()
>>> client.connect("192.168.1.10", 0, 1)
>>> data = client.db_read(1, 0, 4)
>>> client.disconnect()

__enter__() → Client[source]: Context manager entry.

__exit__(exc_type: Any, exc_val: Any, exc_tb: Any) → None[source]: Context manager exit.

__init__(lib_location: str | None = None, *, auto_reconnect: bool = False, max_retries: int = 3, retry_delay: float = 1.0, backoff_factor: float = 2.0, max_delay: float = 30.0, heartbeat_interval: float = 0, on_disconnect: Callable[[], None] | None = None, on_reconnect: Callable[[], None] | None = None, **kwargs: Any)[source]

Initialize S7 client.

Parameters:

lib_location – Ignored. Kept for backwards compatibility.
auto_reconnect – Enable automatic reconnection on connection loss.
max_retries – Maximum number of reconnection attempts.
retry_delay – Initial delay between reconnection attempts in seconds.
backoff_factor – Multiplier for exponential backoff between retries.
max_delay – Maximum delay between reconnection attempts in seconds.
heartbeat_interval – Interval in seconds for heartbeat probes (0=disabled).
on_disconnect – Optional callback invoked when connection is lost.
on_reconnect – Optional callback invoked after successful reconnection.
**kwargs – Ignored. Kept for backwards compatibility.

ab_read(start: int, size: int) → bytearray[source]

Read from process output area (PA).

Parameters:

start – Start byte offset
size – Number of bytes to read

Returns:

Data read from output area

ab_write(start: int, data: bytearray) → int[source]

Write to process output area (PA).

Parameters:

start – Start byte offset
data – Data to write

Returns:

0 on success

as_ab_read(start: int, size: int, data: Array | Array | Array | Array | Array | Array) → int[source]: Async read from process output area.

as_ab_write(start: int, data: bytearray) → int[source]: Async write to process output area.

as_compress(timeout: int) → int[source]: Async compress PLC memory.

as_copy_ram_to_rom(timeout: int = 0) → int[source]: Async copy RAM to ROM.

as_ct_read(start: int, size: int, data: Array | Array | Array | Array | Array | Array) → int[source]: Async read from counter area.

as_ct_write(start: int, size: int, data: bytearray) → int[source]: Async write to counter area.

as_db_fill(db_number: int, filler: int) → int[source]: Async fill DB.

as_db_get(db_number: int, data: Array | Array | Array | Array | Array | Array, size: int) → int[source]: Async get entire DB.

as_db_read(db_number: int, start: int, size: int, data: Array | Array | Array | Array | Array | Array) → int[source]: Async read from DB.

as_db_write(db_number: int, start: int, size: int, data: Array | Array | Array | Array | Array | Array) → int[source]: Async write to DB.

as_download(data: bytearray, block_num: int = -1) → int[source]: Async download block.

as_eb_read(start: int, size: int, data: Array | Array | Array | Array | Array | Array) → int[source]: Async read from input area.

as_eb_write(start: int, size: int, data: bytearray) → int[source]: Async write to input area.

as_full_upload(block_type: Block, block_num: int) → int[source]: Async full upload of block.

as_list_blocks_of_type(block_type: Block, data: Array | Array | Array | Array | Array | Array, count: int) → int[source]: Async list blocks of type.

as_mb_read(start: int, size: int, data: Array | Array | Array | Array | Array | Array) → int[source]: Async read from marker area.

as_mb_write(start: int, size: int, data: bytearray) → int[source]: Async write to marker area.

as_read_area(area: Area, db_number: int, start: int, size: int, wordlen: WordLen, data: Array | Array | Array | Array | Array | Array) → int[source]: Async read from memory area.

as_read_szl(ssl_id: int, index: int, szl: S7SZL, size: int) → int[source]: Async read SZL.

as_read_szl_list(szl_list: S7SZLList, items_count: int) → int[source]: Async read SZL list.

as_tm_read(start: int, size: int, data: Array | Array | Array | Array | Array | Array) → int[source]: Async read from timer area.

as_tm_write(start: int, size: int, data: bytearray) → int[source]: Async write to timer area.

as_upload(block_num: int, data: Array | Array | Array | Array | Array | Array, size: int) → int[source]: Async upload block.

as_write_area(area: Area, db_number: int, start: int, size: int, wordlen: WordLen, data: Array | Array | Array | Array | Array | Array) → int[source]: Async write to memory area.

check_as_completion(status: c_int) → int[source]: Check async completion status.

compress(timeout: int) → int[source]

Compress PLC memory.

Sends real S7 PLC_CONTROL protocol with PI service “_MSZL”.

Parameters:: timeout – Timeout in milliseconds (used for receive timeout)
Returns:: 0 on success

connect(address: str, rack: int, slot: int, tcp_port: int = 102) → Client[source]

Connect to S7 PLC.

Parameters:

address – PLC IP address
rack – Rack number
slot – Slot number
tcp_port – TCP port (default 102)

Returns:

Self for method chaining

connect_routed(host: str, router_rack: int, router_slot: int, subnet: int, dest_rack: int, dest_slot: int, port: int = 102, timeout: float = 5.0) → Client[source]

Connect to an S7 PLC via a routing gateway on another subnet.

The gateway PLC (identified by host, router_rack, router_slot) forwards the connection to the target PLC (identified by subnet, dest_rack, dest_slot) through S7 routing parameters embedded in the COTP Connection Request.

Warning

This method is experimental and may change in future versions.

Parameters:

host – IP address of the routing gateway PLC
router_rack – Rack number of the gateway PLC
router_slot – Slot number of the gateway PLC
subnet – Subnet ID of the target network (0x0000-0xFFFF)
dest_rack – Rack number of the destination PLC
dest_slot – Slot number of the destination PLC
port – TCP port (default 102)
timeout – Connection timeout in seconds

Returns:

Self for method chaining

copy_ram_to_rom(timeout: int = 0) → int[source]

Copy RAM to ROM.

Sends real S7 PLC_CONTROL protocol with PI service “_MSZL” and file ID “P”.

Parameters:: timeout – Timeout in milliseconds (used for receive timeout)
Returns:: 0 on success

create() → None[source]: Create client instance (no-op for compatibility).

ct_read(start: int, size: int) → bytearray[source]

Read from counter area (CT).

Parameters:

start – Start offset
size – Number of counters to read

Returns:

Counter data

ct_write(start: int, size: int, data: bytearray) → int[source]

Write to counter area (CT).

Parameters:

start – Start offset
size – Number of counters to write
data – Counter data to write

Returns:

0 on success

db_fill(db_number: int, filler: int, size: int = 0) → int[source]

Fill a DB with a filler byte.

Uses get_block_info() to determine the DB size automatically. If the PLC does not support get_block_info() or reports an incorrect size (common on S7-1200/1500), pass the size parameter explicitly.

Parameters:

db_number – DB number to fill
filler – Byte value to fill with
size – DB size in bytes. If 0, the size is determined automatically via get_block_info().

Returns:

0 on success

db_get(db_number: int, size: int = 0) → bytearray[source]

Get entire DB.

Uses get_block_info() to determine the DB size automatically. If the PLC does not support get_block_info() or reports an incorrect size (common on S7-1200/1500), pass the size parameter explicitly.

Parameters:

db_number – DB number to read
size – DB size in bytes. If 0, the size is determined automatically via get_block_info().

Returns:

Entire DB contents

db_read(db_number: int, start: int, size: int) → bytearray[source]

Read data from DB.

Parameters:

db_number – DB number to read from
start – Start byte offset
size – Number of bytes to read

Returns:

Data read from DB

db_read_array(db_number: int, start: int, count: int, fmt: str = '>f') → list[Any][source]

Read an array of typed values from a DB.

Reads count consecutive values of the given struct format starting at start byte offset in DB db_number.

Parameters:

db_number – DB number to read from.
start – Start byte offset.
count – Number of values to read.
fmt – struct format for a single value (default ">f" = REAL).

Returns:

List of unpacked values.

Examples

Read 10 REAL values starting at DB1.0:

values = client.db_read_array(1, 0, 10, ">f")

Read 20 INT values starting at DB1.100:

values = client.db_read_array(1, 100, 20, ">h")

db_read_bool(db_number: int, byte_offset: int, bit_offset: int) → bool[source]

Read a single bit from a DB.

Parameters:

db_number – DB number
byte_offset – Byte offset within the DB
bit_offset – Bit offset within the byte (0-7)

Returns:

Boolean value

db_read_byte(db_number: int, offset: int) → int[source]: Read a BYTE (8-bit unsigned) from a DB.

db_read_dint(db_number: int, offset: int) → int[source]: Read a DINT (32-bit signed) from a DB.

db_read_dword(db_number: int, offset: int) → int[source]: Read a DWORD (32-bit unsigned) from a DB.

db_read_int(db_number: int, offset: int) → int[source]: Read an INT (16-bit signed) from a DB.

db_read_lreal(db_number: int, offset: int) → float[source]: Read a LREAL (64-bit float) from a DB.

db_read_real(db_number: int, offset: int) → float[source]: Read a REAL (32-bit float) from a DB.

db_read_string(db_number: int, offset: int) → str[source]

Read an S7 STRING from a DB.

Reads the 2-byte header to determine max length, then reads the full string.

db_read_udint(db_number: int, offset: int) → int[source]: Read a UDINT (32-bit unsigned) from a DB.

db_read_uint(db_number: int, offset: int) → int[source]: Read a UINT (16-bit unsigned) from a DB.

db_read_word(db_number: int, offset: int) → int[source]: Read a WORD (16-bit unsigned) from a DB.

db_read_wstring(db_number: int, offset: int) → str[source]

Read an S7 WSTRING from a DB.

Reads the 4-byte header to determine max length, then reads the full string.

db_write(db_number: int, start: int, data: bytearray) → int[source]

Write data to DB.

Parameters:

db_number – DB number to write to
start – Start byte offset
data – Data to write

Returns:

0 on success

db_write_array(db_number: int, start: int, values: list[Any], fmt: str = '>f') → int[source]

Write an array of typed values to a DB.

Packs values using the given struct format and writes them starting at start byte offset in DB db_number.

Parameters:

db_number – DB number to write to.
start – Start byte offset.
values – List of values to write.
fmt – struct format for a single value (default ">f" = REAL).

Returns:

0 on success.

Examples

Write 10 REAL values starting at DB1.0:

client.db_write_array(1, 0, [1.0, 2.0, 3.0], ">f")

db_write_bool(db_number: int, byte_offset: int, bit_offset: int, value: bool) → None[source]

Write a single bit to a DB (preserving other bits in the byte).

Parameters:

db_number – DB number
byte_offset – Byte offset within the DB
bit_offset – Bit offset within the byte (0-7)
value – Boolean value to write

db_write_byte(db_number: int, offset: int, value: int) → None[source]: Write a BYTE (8-bit unsigned) to a DB.

db_write_dint(db_number: int, offset: int, value: int) → None[source]: Write a DINT (32-bit signed) to a DB.

db_write_dword(db_number: int, offset: int, value: int) → None[source]: Write a DWORD (32-bit unsigned) to a DB.

db_write_int(db_number: int, offset: int, value: int) → None[source]: Write an INT (16-bit signed) to a DB.

db_write_lreal(db_number: int, offset: int, value: float) → None[source]: Write a LREAL (64-bit float) to a DB.

db_write_real(db_number: int, offset: int, value: float) → None[source]: Write a REAL (32-bit float) to a DB.

db_write_string(db_number: int, offset: int, value: str, max_length: int = 254) → None[source]

Write an S7 STRING to a DB.

Parameters:

db_number – DB number
offset – Byte offset
value – String to write
max_length – Maximum string length (default 254)

db_write_udint(db_number: int, offset: int, value: int) → None[source]: Write a UDINT (32-bit unsigned) to a DB.

db_write_uint(db_number: int, offset: int, value: int) → None[source]: Write a UINT (16-bit unsigned) to a DB.

db_write_word(db_number: int, offset: int, value: int) → None[source]: Write a WORD (16-bit unsigned) to a DB.

db_write_wstring(db_number: int, offset: int, value: str, max_length: int = 254) → None[source]

Write an S7 WSTRING to a DB.

Parameters:

db_number – DB number
offset – Byte offset
value – String to write
max_length – Maximum string length in characters (default 254)

delete(block_type: Block, block_num: int) → int[source]

Delete a block from PLC.

Sends real S7 PLC_CONTROL protocol with PI service “_DELE”.

Parameters:

block_type – Type of block (DB, OB, FB, FC, etc.)
block_num – Block number to delete

Returns:

0 on success

destroy() → None[source]: Destroy client instance.

disconnect() → int[source]

Disconnect from S7 PLC.

Returns:: 0 on success

download(data: bytearray, block_num: int = -1) → int[source]

Download block to PLC.

Sends real S7 protocol requests: REQUEST_DOWNLOAD, DOWNLOAD_BLOCK, DOWNLOAD_ENDED.

Parameters:

data – Block data to download
block_num – Block number (-1 to extract from data)

Returns:

0 on success

eb_read(start: int, size: int) → bytearray[source]

Read from process input area (PE).

Parameters:

start – Start byte offset
size – Number of bytes to read

Returns:

Data read from input area

eb_write(start: int, size: int, data: bytearray) → int[source]

Write to process input area (PE).

Parameters:

start – Start byte offset
size – Number of bytes to write (must match len(data))
data – Data to write

Returns:

0 on success

full_upload(block_type: Block, block_num: int) → Tuple[bytearray, int][source]

Upload a block from PLC with header and footer info.

The whole block (including header and footer) is copied into the user buffer.

Sends real S7 protocol requests: START_UPLOAD, UPLOAD, END_UPLOAD.

Parameters:

block_type – Type of block (DB, OB, FB, FC, etc.)
block_num – Block number to upload

Returns:

Tuple of (buffer, size) where buffer contains the complete block with headers and size is the actual data length.

get_block_info(block_type: Block, db_number: int) → TS7BlockInfo[source]

Get block information.

Sends real S7 USER_DATA protocol request to server.

Parameters:

block_type – Type of block
db_number – Block number

Returns:

Block information structure

get_connected() → bool[source]

Check if client is connected to PLC.

Performs an active check on the underlying TCP socket to detect broken connections, rather than just checking a cached flag.

get_cp_info() → S7CpInfo[source]: Get communication processor info (SZL 0x0131).

get_cpu_info() → S7CpuInfo[source]: Get CPU component identification (SZL 0x001C).

get_cpu_state() → str[source]

Get CPU state (running/stopped).

Returns:: CPU state string

get_order_code() → S7OrderCode[source]: Get module order code and firmware version (SZL 0x0011).

get_plc_datetime() → datetime[source]

Get PLC date/time.

Sends real S7 USER_DATA protocol request to server.

Returns:: PLC date and time

get_protection() → S7Protection[source]: Get protection settings (SZL 0x0232).

property is_alive: bool

Whether the connection is alive according to the last heartbeat probe.

Returns True if heartbeat is disabled but the client is connected, or if the last heartbeat probe succeeded.

iso_exchange_buffer(data: bytearray) → bytearray[source]

Exchange raw ISO PDU.

Parameters:: data – Raw PDU data
Returns:: Response PDU data

list_blocks() → BlocksList[source]

List blocks available in PLC.

Sends real S7 USER_DATA protocol request to server.

Returns:: Block list structure with counts for each block type

list_blocks_of_type(block_type: Block, max_count: int) → List[int][source]

List blocks of a specific type.

Sends real S7 USER_DATA protocol request to server. Supports multi-packet responses when the block list doesn’t fit in one PDU.

Parameters:

block_type – Type of blocks to list
max_count – Maximum number of blocks to return

Returns:

List of block numbers

mb_read(start: int, size: int) → bytearray[source]

Read from marker/flag area (MK).

Parameters:

start – Start byte offset
size – Number of bytes to read

Returns:

Data read from marker area

mb_write(start: int, size: int, data: bytearray) → int[source]

Write to marker/flag area (MK).

Parameters:

start – Start byte offset
size – Number of bytes to write (must match len(data))
data – Data to write

Returns:

0 on success

plc_cold_start() → int[source]

Cold start PLC CPU.

Returns:: 0 on success

plc_hot_start() → int[source]

Hot start PLC CPU.

Returns:: 0 on success

plc_stop() → int[source]

Stop PLC CPU.

Returns:: 0 on success

read_area(area: Area, db_number: int, start: int, size: int, word_len: WordLen | None = None) → bytearray[source]

Read data from memory area.

Automatically splits into multiple requests if size exceeds PDU capacity.

Parameters:

area – Memory area to read from
db_number – DB number (for DB area only)
start – Start address
size – Number of items to read (for TM/CT: timers/counters, for others: bytes)
word_len – Optional word length override. If None, defaults to area-based logic (TIMER for TM, COUNTER for CT, BYTE for others).

Returns:

Data read from area

read_diagnostic_buffer() → list[dict[str, Any]][source]

Read the PLC diagnostic buffer.

Warning

This method is experimental and may change.

Returns a list of diagnostic entries, newest first. Each entry is a dict with keys event_id, timestamp, and description.

Returns:: List of diagnostic buffer entries.

read_multi_vars(items: List[dict[str, Any]] | Array) → Tuple[int, Any][source]

Read multiple variables in a single request.

When given a list of dicts with two or more items, uses the multi-variable read optimizer to merge adjacent reads and pack them into minimal PDU exchanges. This significantly reduces the number of round-trips compared to reading each variable individually.

Warning

The read optimizer is experimental and may change in future versions. Disable it with client.use_optimizer = False if you encounter issues.

Parameters:: items – List of item specifications (dicts with area, start, size, and optionally db_number) or a ctypes Array[S7DataItem].
Returns:: Tuple of (result_code, data) where data is either the updated ctypes array or a list of bytearrays in the original item order.
Raises:: ValueError – If more than MAX_VARS items are requested.

read_szl(ssl_id: int, index: int = 0) → S7SZL[source]

Read SZL (System Status List).

Sends real S7 USER_DATA protocol request to server. Supports multi-packet responses where SZL data spans multiple PDUs.

Parameters:

ssl_id – SZL ID
index – SZL index

Returns:

SZL structure with header and data

read_szl_list() → bytes[source]

Read list of available SZL IDs.

Sends real S7 USER_DATA protocol request to server.

Returns:: SZL list data

read_tag(tag: Tag | str) → Any[source]

Read a typed value by Tag or address string.

Accepts a Tag or a PLC4X-style address string (e.g. "DB1.DBX0.0:BOOL", "DB1:10:INT", "M10.5:BOOL").

Parameters:: tag – A Tag instance or a parseable address string.
Returns:: The typed value (bool/int/float/datetime/str depending on type).

Example:

client.read_tag("DB1.DBX0.0:BOOL")            # bit
client.read_tag("DB1.DBD4:REAL")              # float
client.read_tag("DB1:20:STRING[30]")          # variable-length string
client.read_tag(Tag(Area.DB, 1, 0, "REAL"))   # from Tag instance

read_tags(tags: list[Tag | str]) → list[Any][source]

Read multiple tags in a single optimized request.

Uses the multi-variable read optimizer when available to batch reads into minimal PDU exchanges.

Parameters:: tags – List of Tag instances or address strings.
Returns:: List of decoded values in the same order as input.

set_as_callback(callback: Callable[[int, int], None]) → int[source]: Set async callback.

set_plc_datetime(dt: datetime) → int[source]

Set PLC date/time.

Sends real S7 USER_DATA protocol request to server.

Parameters:: dt – Date and time to set
Returns:: 0 on success

set_plc_system_datetime() → int[source]

Set PLC time to system time.

Returns:: 0 on success

tm_read(start: int, size: int) → bytearray[source]

Read from timer area (TM).

Parameters:

start – Start offset
size – Number of timers to read

Returns:

Timer data

tm_write(start: int, size: int, data: bytearray) → int[source]

Write to timer area (TM).

Parameters:

start – Start offset
size – Number of timers to write
data – Timer data to write

Returns:

0 on success

upload(block_num: int) → bytearray[source]

Upload block from PLC.

Sends real S7 protocol requests: START_UPLOAD, UPLOAD, END_UPLOAD.

Parameters:: block_num – Block number to upload
Returns:: Block data

wait_as_completion(timeout: int) → int[source]

Wait for async completion.

Raises:: RuntimeError – If no async operation is pending or timeout=0

write_area(area: Area, db_number: int, start: int, data: bytearray, word_len: WordLen | None = None) → int[source]

Write data to memory area.

Automatically splits into multiple requests if data exceeds PDU capacity.

Parameters:

area – Memory area to write to
db_number – DB number (for DB area only)
start – Start address
data – Data to write
word_len – Optional word length override. If None, defaults to area-based logic (TIMER for TM, COUNTER for CT, BYTE for others).

Returns:

0 on success

write_multi_vars(items: List[dict[str, Any]] | List[S7DataItem]) → int[source]

Write multiple variables in a single request.

Parameters:: items – List of item specifications with data
Returns:: 0 on success
Raises:: ValueError – If more than MAX_VARS items are requested

write_tag(tag: Tag | str, value: Any) → int[source]

Write a typed value by Tag or address string.

Parameters:

tag – A Tag instance or a parseable address string.
value – The value to write (type must match the tag’s datatype).

Returns:

0 on success.

snap7.AsyncClient (legacy)

Legacy async S7 client implementation.

Uses asyncio streams for non-blocking I/O with an asyncio.Lock() to serialize send/receive cycles, ensuring safe concurrent use via asyncio.gather().

For new projects, use s7.AsyncClient instead, which supports all PLC models and automatically selects the best protocol.

class snap7.async_client.AsyncClient[source]

Legacy async S7 client for classic PUT/GET communication.

Uses asyncio streams for non-blocking I/O. An internal asyncio.Lock serializes each send+receive cycle so that concurrent coroutines (e.g. via asyncio.gather) never interleave on the same TCP socket.

For new projects, use s7.AsyncClient instead.

Examples

>>> from s7 import AsyncClient
>>> async with AsyncClient() as client:
...     await client.connect("192.168.1.10", 0, 1)
...     data = await client.db_read(1, 0, 4)

async __aenter__() → AsyncClient[source]: Async context manager entry.

async __aexit__(exc_type: Any, exc_val: Any, exc_tb: Any) → None[source]: Async context manager exit.

async ab_read(start: int, size: int) → bytearray[source]: Read from process output area (PA).

async ab_write(start: int, data: bytearray) → int[source]: Write to process output area (PA).

async compress(timeout: int) → int[source]: Compress PLC memory.

async connect(address: str, rack: int, slot: int, tcp_port: int = 102) → AsyncClient[source]

Connect to S7 PLC.

Parameters:

address – PLC IP address
rack – Rack number
slot – Slot number
tcp_port – TCP port (default 102)

Returns:

Self for method chaining

async copy_ram_to_rom(timeout: int = 0) → int[source]: Copy RAM to ROM.

async ct_read(start: int, size: int) → bytearray[source]: Read from counter area (CT).

async ct_write(start: int, size: int, data: bytearray) → int[source]: Write to counter area (CT).

async db_fill(db_number: int, filler: int, size: int = 0) → int[source]

Fill a DB with a filler byte.

Parameters:

db_number – DB number to fill
filler – Byte value to fill with
size – DB size in bytes. If 0, determined via get_block_info().

Returns:

0 on success

async db_get(db_number: int, size: int = 0) → bytearray[source]

Get entire DB.

Parameters:

db_number – DB number to read
size – DB size in bytes. If 0, determined via get_block_info().

Returns:

Entire DB contents

async db_read(db_number: int, start: int, size: int) → bytearray[source]

Read data from DB.

Parameters:

db_number – DB number to read from
start – Start byte offset
size – Number of bytes to read

Returns:

Data read from DB

async db_write(db_number: int, start: int, data: bytearray) → int[source]

Write data to DB.

Parameters:

db_number – DB number to write to
start – Start byte offset
data – Data to write

Returns:

0 on success

async delete(block_type: Block, block_num: int) → int[source]: Delete a block from PLC.

async disconnect() → int[source]

Disconnect from S7 PLC.

Returns:: 0 on success

async download(data: bytearray, block_num: int = -1) → int[source]: Download block to PLC.

async eb_read(start: int, size: int) → bytearray[source]: Read from process input area (PE).

async eb_write(start: int, size: int, data: bytearray) → int[source]: Write to process input area (PE).

async full_upload(block_type: Block, block_num: int) → Tuple[bytearray, int][source]: Upload a block from PLC with header and footer info.

async get_block_info(block_type: Block, db_number: int) → TS7BlockInfo[source]: Get block information.

get_connected() → bool[source]: Check if client is connected.

async get_cp_info() → S7CpInfo[source]: Get communication processor info (SZL 0x0131).

async get_cpu_info() → S7CpuInfo[source]: Get CPU component identification (SZL 0x001C).

async get_cpu_state() → str[source]: Get CPU state (running/stopped).

async get_order_code() → S7OrderCode[source]: Get module order code and firmware version (SZL 0x0011).

async get_plc_datetime() → datetime[source]: Get PLC date/time.

async get_protection() → S7Protection[source]: Get protection settings (SZL 0x0232).

async iso_exchange_buffer(data: bytearray) → bytearray[source]: Exchange raw ISO PDU.

async list_blocks() → BlocksList[source]: List blocks available in PLC.

async list_blocks_of_type(block_type: Block, max_count: int) → List[int][source]

List blocks of a specific type.

Supports multi-packet responses.

async mb_read(start: int, size: int) → bytearray[source]: Read from marker/flag area (MK).

async mb_write(start: int, size: int, data: bytearray) → int[source]: Write to marker/flag area (MK).

async plc_cold_start() → int[source]: Cold start PLC CPU.

async plc_hot_start() → int[source]: Hot start PLC CPU.

async plc_stop() → int[source]: Stop PLC CPU.

async read_area(area: Area, db_number: int, start: int, size: int) → bytearray[source]

Read data from memory area.

Automatically splits into multiple requests if size exceeds PDU capacity.

async read_multi_vars(items: List[dict[str, Any]]) → Tuple[int, list[bytearray]][source]

Read multiple variables (sequentially, one read_area per item).

Parameters:: items – List of item dicts with keys: area, db_number, start, size
Returns:: Tuple of (result_code, list_of_bytearrays)

async read_szl(ssl_id: int, index: int = 0) → S7SZL[source]

Read SZL (System Status List).

Supports multi-packet responses.

async read_szl_list() → bytes[source]: Read list of available SZL IDs.

async set_plc_datetime(dt: datetime) → int[source]: Set PLC date/time.

async set_plc_system_datetime() → int[source]: Set PLC time to system time.

async tm_read(start: int, size: int) → bytearray[source]: Read from timer area (TM).

async tm_write(start: int, size: int, data: bytearray) → int[source]: Write to timer area (TM).

async upload(block_num: int) → bytearray[source]: Upload block from PLC (3-step: START_UPLOAD, UPLOAD, END_UPLOAD).

async write_area(area: Area, db_number: int, start: int, data: bytearray) → int[source]

Write data to memory area.

Automatically splits into multiple requests if data exceeds PDU capacity.

async write_multi_vars(items: List[dict[str, Any]]) → int[source]

Write multiple variables (sequentially, one write_area per item).

Parameters:: items – List of item dicts with keys: area, db_number, start, data
Returns:: 0 on success