S7 Protocol

The s7protocol module implements the S7 PDU encoding and decoding layer.

S7 protocol implementation.

Handles S7 PDU encoding/decoding and protocol operations.

class snap7.s7protocol.S7Function(*values)[source]: S7 protocol function codes.

class snap7.s7protocol.S7PDUType(*values)[source]: S7 PDU type codes.

class snap7.s7protocol.S7Protocol[source]

S7 protocol implementation.

Handles encoding and decoding of S7 PDUs for communication with Siemens PLCs.

build_compress_request() → bytes[source]

Build PLC control request for memory compression.

Uses PI service “_MSZL” (compress memory).

Returns:: Complete S7 PDU for compress request

build_copy_ram_to_rom_request() → bytes[source]

Build PLC control request for copying RAM to ROM.

Uses PI service “_MSZL” with file system parameters.

Returns:: Complete S7 PDU for copy RAM to ROM request

build_cpu_state_request() → bytes[source]

Build CPU state request.

Returns:: Complete S7 PDU for CPU state query

build_delete_block_request(block_type: int, block_num: int) → bytes[source]

Build delete block request.

Uses PLC_CONTROL with PI service “_DELE” for block deletion.

Parameters:

block_type – Block type code
block_num – Block number

Returns:

Complete S7 PDU for delete block request

build_download_request(block_type: int, block_num: int, block_data: bytes) → bytes[source]

Build request download request.

Parameters:

block_type – Block type code
block_num – Block number
block_data – Block data to download

Returns:

Complete S7 PDU for request download

build_end_upload_request(upload_id: int) → bytes[source]

Build end upload request.

Parameters:: upload_id – Upload ID from start upload response
Returns:: Complete S7 PDU for end upload request

build_get_block_info_request(block_type: int, block_num: int) → bytes[source]

Build USER_DATA request for getting block information.

Parameters:

block_type – Block type code (0x38=OB, 0x41=DB, 0x42=SDB, 0x43=FC, 0x44=SFC, 0x45=FB, 0x46=SFB)
block_num – Block number

Returns:

Complete S7 PDU for get block info request

build_get_clock_request() → bytes[source]

Build USER_DATA request for reading PLC clock.

Returns:: Complete S7 PDU for get clock request

build_list_blocks_of_type_request(block_type: int) → bytes[source]

Build USER_DATA request for listing blocks of a specific type.

Parameters:: block_type – Block type code (e.g., 0x41 for DB)
Returns:: Complete S7 PDU for list blocks of type request

build_list_blocks_request() → bytes[source]

Build USER_DATA request for listing all blocks.

Returns:: Complete S7 PDU for list blocks request

build_multi_read_request(items: List[Tuple[int, int, int, int]]) → bytes[source]

Build S7 multi-variable read request PDU.

Encodes multiple address specifications into a single READ_AREA request so the PLC can return all data in one response.

Parameters:: items – List of (area, db_number, start_offset, byte_length) tuples.
Returns:: Complete S7 PDU.

build_plc_control_request(operation: str) → bytes[source]

Build PLC control request.

Parameters:: operation – Control operation (‘stop’, ‘hot_start’, ‘cold_start’)
Returns:: Complete S7 PDU for PLC control

build_read_request(area: S7Area, db_number: int, start: int, word_len: S7WordLen, count: int) → bytes[source]

Build S7 read request PDU.

Parameters:

area – Memory area to read from
db_number – DB number (for DB area)
start – Start address/offset
word_len – Data word length
count – Number of items to read

Returns:

Complete S7 PDU

build_read_szl_request(szl_id: int, szl_index: int) → bytes[source]

Build USER_DATA request for reading SZL (System Status List).

Parameters:

szl_id – SZL identifier
szl_index – SZL index

Returns:

Complete S7 PDU for read SZL request

build_set_clock_request(dt: datetime) → bytes[source]

Build USER_DATA request for setting PLC clock.

Parameters:: dt – Datetime to set
Returns:: Complete S7 PDU for set clock request

build_setup_communication_request(max_amq_caller: int = 1, max_amq_callee: int = 1, pdu_length: int = 480) → bytes[source]

Build S7 setup communication request.

This negotiates communication parameters with the PLC.

build_start_upload_request(block_type: int, block_num: int) → bytes[source]

Build start upload request.

Parameters:

block_type – Block type code (0x38=OB, 0x41=DB, 0x42=SDB, 0x43=FC, 0x44=SFC, 0x45=FB, 0x46=SFB)
block_num – Block number

Returns:

Complete S7 PDU for start upload request

build_upload_request(upload_id: int) → bytes[source]

Build upload request to get block data.

Parameters:: upload_id – Upload ID from start upload response
Returns:: Complete S7 PDU for upload request

build_userdata_followup_request(group: int, subfunction: int, sequence_number: int) → bytes[source]

Build USERDATA follow-up request for multi-packet responses.

Parameters:

group – USERDATA group (e.g., 4 for SZL, 3 for block info)
subfunction – Subfunction code
sequence_number – Sequence number from the previous response

Returns:

Complete S7 PDU for follow-up request

build_write_request(area: S7Area, db_number: int, start: int, word_len: S7WordLen, data: bytes) → bytes[source]

Build S7 write request PDU.

Parameters:

area – Memory area to write to
db_number – DB number (for DB area)
start – Start address/offset
word_len – Data word length
data – Data to write

Returns:

Complete S7 PDU

check_control_response(response: Dict[str, Any]) → None[source]

Check PLC control response for errors.

Parameters:: response – Parsed S7 response
Raises:: S7ProtocolError – If control operation failed

check_write_response(response: Dict[str, Any]) → None[source]

Check write operation response for errors.

Parameters:: response – Parsed S7 response
Raises:: S7ProtocolError – If write operation failed

extract_cpu_state(response: Dict[str, Any]) → str[source]

Extract CPU state from response.

Parameters:: response – Parsed S7 response
Returns:: CPU state string in S7CpuStatus format (e.g., ‘S7CpuStatusRun’)

extract_multi_read_data(response: Dict[str, Any], block_count: int) → List[bytearray][source]

Extract per-block data from a multi-variable read response.

Parses the raw data section which contains N items, each with:

return_code (1 byte)
transport_size (1 byte)
bit_length (2 bytes, big-endian)
data (bit_length / 8 bytes)
fill byte (1 byte if byte_length is odd and not the last item)

Parameters:

response – Parsed S7 response from parse_response().
block_count – Expected number of data items.

Returns:

List of bytearrays, one per block.

Raises:

S7ProtocolError – If any item has a non-success return code.

extract_read_data(response: Dict[str, Any], word_len: S7WordLen, count: int) → List[Any][source]

Extract and decode data from read response.

Parameters:

response – Parsed S7 response
word_len – Expected data word length
count – Expected number of items

Returns:

List of decoded values

parse_get_block_info(response: Dict[str, Any]) → TS7BlockInfo[source]

Parse block info response directly into a TS7BlockInfo.

Consolidates the dict→struct conversion that used to live in both the sync and async clients.

parse_get_block_info_response(response: Dict[str, Any]) → Dict[str, Any][source]

Parse get block info response.

Parameters:: response – Parsed S7 response
Returns:: Dictionary with block info fields

parse_get_clock_response(response: Dict[str, Any]) → datetime[source]

Parse get clock response.

Parameters:: response – Parsed S7 response
Returns:: Datetime from PLC

parse_list_blocks(response: Dict[str, Any]) → BlocksList[source]

Parse list blocks response directly into a BlocksList.

Consolidates the dict→struct conversion that used to live in both the sync and async clients so the field mapping is declared once.

parse_list_blocks_of_type_response(response: Dict[str, Any]) → List[int][source]

Parse list blocks of type response and extract block numbers.

Parameters:: response – Parsed S7 response
Returns:: List of block numbers

parse_list_blocks_response(response: Dict[str, Any]) → Dict[str, int][source]

Parse list blocks response and extract block counts.

Parameters:: response – Parsed S7 response
Returns:: Dictionary mapping block type names to counts

parse_read_szl_response(response: Dict[str, Any], first_fragment: bool = True) → Dict[str, Any][source]

Parse read SZL response.

Parameters:

response – Parsed S7 response
first_fragment – If True (default), parse SZL header (ID+Index). If False, treat all data as raw payload (follow-up fragments).

Returns:

Dictionary with SZL ID, Index, and data

parse_response(pdu: bytes) → Dict[str, Any][source]

Parse S7 response PDU.

Parameters:: pdu – Complete S7 PDU
Returns:: Parsed response data

parse_start_upload_response(response: Dict[str, Any]) → Dict[str, Any][source]

Parse start upload response.

Returns:: Dictionary with upload_id and block_length

parse_upload_response(response: Dict[str, Any]) → bytes[source]

Parse upload response and extract block data.

Returns:: Block data bytes

validate_pdu_reference(response_sequence: int) → None[source]

Validate the PDU reference number from a response.

Compares the response sequence number against the expected (current) sequence.

Parameters:

response_sequence – Sequence number from the response PDU.

Raises:

S7StalePacketError – If response is older than expected (stale).
S7PacketLostError – If response is ahead of expected (packet loss).

class snap7.s7protocol.S7UserDataGroup(*values)[source]: S7 USER_DATA type groups (from s7_types.h).

class snap7.s7protocol.S7UserDataSubfunction(*values)[source]: S7 USER_DATA subfunctions.

snap7.s7protocol.build_block_info_from_dict(info: Dict[str, Any]) → TS7BlockInfo[source]: Populate a TS7BlockInfo from the dict returned by parse_get_block_info_response.

snap7.s7protocol.build_blocks_list_from_dict(counts: Dict[str, int]) → BlocksList[source]: Populate a BlocksList from the dict returned by parse_list_blocks_response.

snap7.s7protocol.get_return_code_description(return_code: int) → str[source]: Get human-readable description for S7 return code.