Client

python-snap7 provides two client packages:

• s7commplus: S7CommPlus protocol for S7-1200/1500 PLCs

• s7: Classic S7 protocol for S7-300/400 and PUT/GET access on S7-1200/1500

s7commplus.Client

from s7commplus import Client

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

s7commplus.AsyncClient

import asyncio
from s7commplus import AsyncClient

async def main():
    client = AsyncClient()
    await client.connect("192.168.1.10")
    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 s7commplus import Client

client = Client()
client.connect("192.168.1.10", 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",
    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 s7commplus import Client

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

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),
)

---

S7CommPlus client for S7-1200/1500 PLCs.

Reference: thomas-v2/S7CommPlusDriver (C#, LGPL-3.0)

class s7commplus.client.S7CommPlusClient

S7CommPlus client for S7-1200/1500 PLCs.

Use from s7commplus import Client to instantiate.

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

Browse the full per-tag symbol tree via EXPLORE + the type-info container.

Warning

This method is experimental and may change.

Returns a flat list of variable dicts with keys name, access_sequence (the dot-separated hex LID path usable with read_tag()), data_type, and the optimized/non-optimized byte+bit offsets. Steps: enumerate DBs, resolve each DB’s type-info RID via a LID=1 read, explore the OMS type-info container, then recombine into the symbol tree.

Returns:

List of variable info dicts.

connect(host: str, port: int = 102, rack: int = 0, slot: int = 1, use_tls: bool = False, tls_cert: str | None = None, tls_key: str | None = None, tls_ca: str | None = None, password: str | None = None) → None

Connect to an S7-1200/1500 PLC using S7CommPlus.

Parameters:

• host – PLC IP address or hostname

• port – TCP port (default 102)

• rack – PLC rack number (unused, kept for API symmetry)

• slot – PLC slot number (unused, kept for API symmetry)

• 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)

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

Create a data change subscription.

Warning

This method is experimental and may change.

The PLC will push data updates for the specified variables. Use receive_notification() to receive the pushed data.

Parameters:

• items – List of (db_number, start_offset, size) tuples to monitor.

• cycle_ms – Cycle time in milliseconds (0 = on change).

Returns:

Subscription object ID assigned by the PLC.

db_read(db_number: int, start: int, size: int) → bytes

Read raw bytes from a data block.

Parameters:

• db_number – Data block number

• start – Start byte offset

• size – Number of bytes to read

Returns:

Raw bytes read from the data block

db_read_multi(items: list[tuple[int, int, int]]) → list[bytes]

Read multiple data block regions in a single request.

Parameters:

items – List of (db_number, start_offset, size) tuples

Returns:

List of raw bytes for each item

db_write(db_number: int, start: int, data: bytes) → None

Write raw bytes to a data block.

Parameters:

• db_number – Data block number

• start – Start byte offset

• data – Bytes to write

delete_subscription(subscription_id: int) → None

Delete a data change subscription.

Warning

This method is experimental and may change.

Parameters:

subscription_id – ID returned by create_subscription().

disconnect() → None

Disconnect from PLC.

download_block(block_type: int, block_number: int, data: bytes) → None

Download (write) a program block to the PLC.

Warning

This method is experimental and may change.

Parameters:

• block_type – Block type.

• block_number – Block number.

• data – Raw block data to write.

explore(explore_id: int = 0) → bytes

Browse the PLC object tree.

Parameters:

explore_id – Object to explore (0 = root).

Returns:

Raw response payload.

get_cpu_state() → str

Get PLC CPU operating state via S7CommPlus.

Warning

This method is experimental and may change.

Returns:

One of "RUN", "STOP", or "UNKNOWN".

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

List all datablocks on the PLC via EXPLORE.

Warning

This method is experimental and may change.

Returns:

List of dicts with keys name, number, rid.

property protocol_version: int

Protocol version negotiated with the PLC.

read_area(area_rid: int, start: int, size: int) → bytes

Read raw bytes from a controller memory area (M, I, Q, counters, timers).

Parameters:

• area_rid – Native object RID for the area, e.g. Ids.NATIVE_THE_M_AREA_RID (82) for Merker.

• start – Start byte offset.

• size – Number of bytes to read.

Returns:

Raw bytes read from the area.

read_symbolic(access_area: int, lids: list[int], symbol_crc: int = 0) → bytes

Read a variable using S7CommPlus symbolic (LID-based) access.

Warning

This method is experimental and may change.

For S7-1200/1500 DBs with “Optimized block access” enabled, byte offsets are unreliable — the PLC internally relocates variables between downloads. Symbolic access navigates the PLC’s symbol tree using LIDs (Local IDs) discovered via browse().

Note

I/Q/M area caveats (observed on S7-1200 FW V4.5):

• Physical PAQ LIDs return error when value is 0x00. Reading a physical Q-byte LID (e.g. QB0) when all outputs are OFF raises RuntimeError instead of returning b"\x00". Callers should catch the exception and treat it as zero.

• TCP RST after first I/Q/M read. The PLC sends a TCP RST after the first successful I/Q/M GetMultiVariables per connection. DB reads are unaffected. Workaround: reconnect before each I/Q/M read.

• Symbolic BOOL LIDs can be stale vs. physical PAQ. A symbolic BOOL tag written via SetMultiVariables may not reflect later forced writes to the physical address. For reliable output state, read the physical byte LID instead.

Parameters:

• access_area – Access area ID. For DBs this is 0x8A0E0000 + db_number.

• lids – LID path through the symbol tree.

• symbol_crc – Symbol CRC for layout validation (0 = skip check).

Returns:

Raw bytes of the variable value.

property session_id: int

Session ID assigned by the PLC.

property session_setup_ok: bool

Whether the S7CommPlus session setup succeeded for data operations.

set_plc_operating_state(state: int) → None

Set the PLC operating state (start/stop).

Uses INVOKE to call the PLC’s operating-state setter.

Parameters:

state – Target operating state. 1 = STOP, 2 = RUN, 3 = HOT_RESTART.

property tls_active: bool

Whether TLS is active on the connection.

upload_block(block_type: int, block_number: int) → bytes

Upload (read) a program block from the PLC.

Warning

This method is experimental and may change.

Parameters:

• block_type – Block type (e.g. 0x08 for DB, 0x0C for FC).

• block_number – Block number.

Returns:

Raw block data.

write_area(area_rid: int, start: int, data: bytes) → None

Write raw bytes to a controller memory area (M, I, Q, counters, timers).

Parameters:

• area_rid – Native object RID for the area.

• start – Start byte offset.

• data – Bytes to write.

write_symbolic(access_area: int, lids: list[int], data: bytes, symbol_crc: int = 0) → None

Write a variable using S7CommPlus symbolic (LID-based) access.

Warning

This method is experimental and may change.

See read_symbolic() for context on when to use symbolic access and I/Q/M area caveats.

Note

Writing a symbolic BOOL tag does not update the physical PAQ byte on some S7-1200 firmware versions. A subsequent read of the physical byte LID may show a different value than the symbolic BOOL LID.

Parameters:

• access_area – Access area ID.

• lids – LID path through the symbol tree.

• data – Raw bytes to write.

• symbol_crc – Symbol CRC for layout validation (0 = skip check).

Async S7CommPlus client for S7-1200/1500 PLCs.

Reference: thomas-v2/S7CommPlusDriver (C#, LGPL-3.0)

class s7commplus.async_client.S7CommPlusAsyncClient

Async S7CommPlus client for S7-1200/1500 PLCs.

Use from s7commplus import AsyncClient to instantiate.

async authenticate(password: str, username: str = '') → None

Perform PLC password authentication (legitimation).

Parameters:

• password – PLC password

• username – Username for new-style auth (optional)

Raises:

S7ConnectionError – If not connected, TLS not active, or auth fails

async browse() → list[dict[str, Any]]

Browse the full per-tag symbol tree via EXPLORE + the type-info container.

Warning

This method is experimental and may change.

Returns a flat list of variable dicts with keys name, access_sequence (the dot-separated hex LID path usable with read_tag()), data_type, and the optimized/non-optimized byte+bit offsets. Steps: enumerate DBs, resolve each DB’s type-info RID via a LID=1 read, explore the OMS type-info container, then recombine into the symbol tree.

Returns:

List of variable info dicts.

async connect(host: str, port: int = 102, rack: int = 0, slot: int = 1, *, use_tls: bool = False, tls_cert: str | None = None, tls_key: str | None = None, tls_ca: str | None = None) → None

Connect to an S7-1200/1500 PLC using S7CommPlus.

Parameters:

• host – PLC IP address or hostname

• port – TCP port (default 102)

• rack – PLC rack number (unused, kept for API symmetry)

• slot – PLC slot number (unused, kept for API symmetry)

• 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)

async db_read(db_number: int, start: int, size: int) → bytes

Read raw bytes from a data block.

async db_read_multi(items: list[tuple[int, int, int]]) → list[bytes]

Read multiple data block regions in a single request.

async db_write(db_number: int, start: int, data: bytes) → None

Write raw bytes to a data block.

async disconnect() → None

Disconnect from PLC.

async explore(explore_id: int = 0) → bytes

Browse the PLC object tree.

async list_datablocks() → list[dict[str, Any]]

List all datablocks on the PLC via EXPLORE.

Warning

This method is experimental and may change.

property oms_secret: bytes | None

OMS exporter secret from TLS session (None if TLS not active).

async read_area(area_rid: int, start: int, size: int) → bytes

Read raw bytes from a controller memory area (M, I, Q, counters, timers).

async read_symbolic(access_area: int, lids: list[int], symbol_crc: int = 0) → bytes

Read a variable using S7CommPlus symbolic (LID-based) access.

Warning

This method is experimental and may change.

property session_setup_ok: bool

Whether the S7CommPlus session setup succeeded for data operations.

async set_plc_operating_state(state: int) → None

Set the PLC operating state (start/stop).

property tls_active: bool

Whether TLS is active on the connection.

async write_area(area_rid: int, start: int, data: bytes) → None

Write raw bytes to a controller memory area (M, I, Q, counters, timers).

s7.Client (legacy)

The s7.Client implements the classic S7 protocol for S7-300/400 PLCs and PUT/GET access on S7-1200/1500.

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)

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

Context manager entry.

__exit__(exc_type: Any, exc_val: Any, exc_tb: Any) → None

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)

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

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

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

Async read from process output area.

as_ab_write(start: int, data: bytearray) → int

Async write to process output area.

as_compress(timeout: int) → int

Async compress PLC memory.

as_copy_ram_to_rom(timeout: int = 0) → int

Async copy RAM to ROM.

as_ct_read(start: int, size: int, data: Array | Array | Array | Array | Array | Array) → int

Async read from counter area.

as_ct_write(start: int, size: int, data: bytearray) → int

Async write to counter area.

as_db_fill(db_number: int, filler: int) → int

Async fill DB.

as_db_get(db_number: int, data: Array | Array | Array | Array | Array | Array, size: int) → int

Async get entire DB.

as_db_read(db_number: int, start: int, size: int, data: Array | Array | Array | Array | Array | Array) → int

Async read from DB.

as_db_write(db_number: int, start: int, size: int, data: Array | Array | Array | Array | Array | Array) → int

Async write to DB.

as_download(data: bytearray, block_num: int = -1) → int

Async download block.

as_eb_read(start: int, size: int, data: Array | Array | Array | Array | Array | Array) → int

Async read from input area.

as_eb_write(start: int, size: int, data: bytearray) → int

Async write to input area.

as_full_upload(block_type: Block, block_num: int) → int

Async full upload of block.

as_list_blocks_of_type(block_type: Block, data: Array | Array | Array | Array | Array | Array, count: int) → int

Async list blocks of type.

as_mb_read(start: int, size: int, data: Array | Array | Array | Array | Array | Array) → int

Async read from marker area.

as_mb_write(start: int, size: int, data: bytearray) → int

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

Async read from memory area.

as_read_szl(ssl_id: int, index: int, szl: S7SZL, size: int) → int

Async read SZL.

as_read_szl_list(szl_list: S7SZLList, items_count: int) → int

Async read SZL list.

as_tm_read(start: int, size: int, data: Array | Array | Array | Array | Array | Array) → int

Async read from timer area.

as_tm_write(start: int, size: int, data: bytearray) → int

Async write to timer area.

as_upload(block_num: int, data: Array | Array | Array | Array | Array | Array, size: int) → int

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

Async write to memory area.

check_as_completion(status: c_int) → int

Check async completion status.

compress(timeout: int) → int

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

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

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

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

Create client instance (no-op for compatibility).

ct_read(start: int, size: int) → bytearray

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

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

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

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

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]

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

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

Read a BYTE (8-bit unsigned) from a DB.

db_read_dint(db_number: int, offset: int) → int

Read a DINT (32-bit signed) from a DB.

db_read_dword(db_number: int, offset: int) → int

Read a DWORD (32-bit unsigned) from a DB.

db_read_int(db_number: int, offset: int) → int

Read an INT (16-bit signed) from a DB.

db_read_lreal(db_number: int, offset: int) → float

Read a LREAL (64-bit float) from a DB.

db_read_real(db_number: int, offset: int) → float

Read a REAL (32-bit float) from a DB.

db_read_string(db_number: int, offset: int) → str

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

Read a UDINT (32-bit unsigned) from a DB.

db_read_uint(db_number: int, offset: int) → int

Read a UINT (16-bit unsigned) from a DB.

db_read_word(db_number: int, offset: int) → int

Read a WORD (16-bit unsigned) from a DB.

db_read_wstring(db_number: int, offset: int) → str

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

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

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

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

Write a BYTE (8-bit unsigned) to a DB.

db_write_dint(db_number: int, offset: int, value: int) → None

Write a DINT (32-bit signed) to a DB.

db_write_dword(db_number: int, offset: int, value: int) → None

Write a DWORD (32-bit unsigned) to a DB.

db_write_int(db_number: int, offset: int, value: int) → None

Write an INT (16-bit signed) to a DB.

db_write_lreal(db_number: int, offset: int, value: float) → None

Write a LREAL (64-bit float) to a DB.

db_write_real(db_number: int, offset: int, value: float) → None

Write a REAL (32-bit float) to a DB.

db_write_string(db_number: int, offset: int, value: str, max_length: int = 254) → None

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

Write a UDINT (32-bit unsigned) to a DB.

db_write_uint(db_number: int, offset: int, value: int) → None

Write a UINT (16-bit unsigned) to a DB.

db_write_word(db_number: int, offset: int, value: int) → None

Write a WORD (16-bit unsigned) to a DB.

db_write_wstring(db_number: int, offset: int, value: str, max_length: int = 254) → None

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

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

Destroy client instance.

disconnect() → int

Disconnect from S7 PLC.

Returns:

0 on success

download(data: bytearray, block_num: int = -1) → int

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

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

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]

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

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

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

Get communication processor info (SZL 0x0131).

get_cpu_info() → S7CpuInfo

Get CPU component identification (SZL 0x001C).

get_cpu_state() → str

Get CPU state (running/stopped).

Returns:

CPU state string

get_order_code() → S7OrderCode

Get module order code and firmware version (SZL 0x0011).

get_plc_datetime() → datetime

Get PLC date/time.

Sends real S7 USER_DATA protocol request to server.

Returns:

PLC date and time

get_protection() → S7Protection

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

Exchange raw ISO PDU.

Parameters:

data – Raw PDU data

Returns:

Response PDU data

list_blocks() → BlocksList

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]

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

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

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

Cold start PLC CPU.

Returns:

0 on success

plc_hot_start() → int

Hot start PLC CPU.

Returns:

0 on success

plc_stop() → int

Stop PLC CPU.

Returns:

0 on success

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

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]]

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]

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

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

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

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]

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

Set async callback.

set_plc_datetime(dt: datetime) → int

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

Set PLC time to system time.

Returns:

0 on success

tm_read(start: int, size: int) → bytearray

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

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

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

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

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

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

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.

s7.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

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

Async context manager entry.

async __aexit__(exc_type: Any, exc_val: Any, exc_tb: Any) → None

Async context manager exit.

async ab_read(start: int, size: int) → bytearray

Read from process output area (PA).

async ab_write(start: int, data: bytearray) → int

Write to process output area (PA).

async compress(timeout: int) → int

Compress PLC memory.

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

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

Copy RAM to ROM.

async ct_read(start: int, size: int) → bytearray

Read from counter area (CT).

async ct_write(start: int, size: int, data: bytearray) → int

Write to counter area (CT).

async db_fill(db_number: int, filler: int, size: int = 0) → int

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

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

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

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

Delete a block from PLC.

async disconnect() → int

Disconnect from S7 PLC.

Returns:

0 on success

async download(data: bytearray, block_num: int = -1) → int

Download block to PLC.

async eb_read(start: int, size: int) → bytearray

Read from process input area (PE).

async eb_write(start: int, size: int, data: bytearray) → int

Write to process input area (PE).

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

Upload a block from PLC with header and footer info.

async get_block_info(block_type: Block, db_number: int) → TS7BlockInfo

Get block information.

get_connected() → bool

Check if client is connected.

async get_cp_info() → S7CpInfo

Get communication processor info (SZL 0x0131).

async get_cpu_info() → S7CpuInfo

Get CPU component identification (SZL 0x001C).

async get_cpu_state() → str

Get CPU state (running/stopped).

async get_order_code() → S7OrderCode

Get module order code and firmware version (SZL 0x0011).

async get_plc_datetime() → datetime

Get PLC date/time.

async get_protection() → S7Protection

Get protection settings (SZL 0x0232).

async iso_exchange_buffer(data: bytearray) → bytearray

Exchange raw ISO PDU.

async list_blocks() → BlocksList

List blocks available in PLC.

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

List blocks of a specific type.

Supports multi-packet responses.

async mb_read(start: int, size: int) → bytearray

Read from marker/flag area (MK).

async mb_write(start: int, size: int, data: bytearray) → int

Write to marker/flag area (MK).

async plc_cold_start() → int

Cold start PLC CPU.

async plc_hot_start() → int

Hot start PLC CPU.

async plc_stop() → int

Stop PLC CPU.

async read_area(area: Area, db_number: int, start: int, size: int) → bytearray

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]]

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

Read SZL (System Status List).

Supports multi-packet responses.

async read_szl_list() → bytes

Read list of available SZL IDs.

async set_plc_datetime(dt: datetime) → int

Set PLC date/time.

async set_plc_system_datetime() → int

Set PLC time to system time.

async tm_read(start: int, size: int) → bytearray

Read from timer area (TM).

async tm_write(start: int, size: int, data: bytearray) → int

Write to timer area (TM).

async upload(block_num: int) → bytearray

Upload block from PLC (3-step: START_UPLOAD, UPLOAD, END_UPLOAD).

async write_area(area: Area, db_number: int, start: int, data: bytearray) → int

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

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