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

__del__() → None[source]: Destructor.

__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, **kwargs: Any)[source]

Initialize S7 client.

Parameters:

lib_location – Ignored. Kept for backwards compatibility.
**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.

clear_session_password() → int[source]

Clear session password.

Returns:: 0 on success

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

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

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

error_text(error_code: int) → str[source]

Get error text for error code.

Parameters:: error_code – Error code to look up
Returns:: Human-readable error text

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 CP (Communication Processor) information.

Uses read_szl(0x0131) to get communication parameters.

Returns:: CP information structure

get_cpu_info() → S7CpuInfo[source]

Get CPU information.

Uses read_szl(0x001C) to get component identification data.

Returns:: CPU information structure

get_cpu_state() → str[source]

Get CPU state (running/stopped).

Returns:: CPU state string

get_exec_time() → int[source]

Get last operation execution time.

Returns:: Execution time in milliseconds

get_last_error() → int[source]

Get last error code.

Returns:: Last error code

get_order_code() → S7OrderCode[source]

Get order code.

Uses read_szl(0x0011) to get module identification.

Returns:: Order code structure

get_param(param: Parameter) → int[source]

Get client parameter.

Parameters:: param – Parameter number
Returns:: Parameter value

get_pdu_length() → int[source]

Get negotiated PDU length.

Returns:: PDU length in bytes

get_pg_block_info(data: bytearray) → TS7BlockInfo[source]

Get block info from raw block data.

Parameters:: data – Raw block data
Returns:: Block information structure

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.

Uses read_szl(0x0232) to get protection level.

Returns:: Protection structure

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_multi_vars(items: List[dict[str, Any]] | Array) → Tuple[int, Any][source]

Read multiple variables in a single request.

Parameters:: items – List of item specifications or S7DataItem array
Returns:: Tuple of (result, items with data)
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

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

set_connection_params(address: str, local_tsap: int, remote_tsap: int) → None[source]

Set connection parameters.

Parameters:

address – PLC IP address
local_tsap – Local TSAP
remote_tsap – Remote TSAP

set_connection_type(connection_type: int) → None[source]

Set connection type.

Parameters:: connection_type – Connection type (1=PG, 2=OP, 3=S7Basic)

set_param(param: Parameter, value: int) → int[source]

Set client parameter.

Parameters:

param – Parameter number
value – Parameter value

Returns:

0 on success

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

set_session_password(password: str) → int[source]

Set session password.

Parameters:: password – Session password
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