Client

Pure Python S7 client implementation.

Drop-in replacement for the ctypes-based client with native Python implementation.

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

Pure Python S7 client implementation.

Drop-in replacement for the ctypes-based client that provides native Python communication with Siemens S7 PLCs without requiring the Snap7 C library.

Examples

>>> import snap7
>>> client = snap7.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.