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_clear_session_password_request() → bytes[source]

Build USERDATA request to clear the session password.

Uses function group 5 (Security), subfunction 2.

Returns:

Complete S7 PDU.

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

static build_download_ended_response(sequence: int) → bytes[source]

Acknowledge the PLC’s DOWNLOAD_ENDED request.

static build_download_fragment_response(sequence: int, is_last: bool, data: bytes) → bytes[source]

Reply to a PLC DOWNLOAD_BLOCK request with one bounded fragment.

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 using the S7 PI service PDU format.

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_set_session_password_request(encoded_password: bytes) → bytes[source]

Build USERDATA request to set the session password.

Uses function group 5 (Security), subfunction 1.

Parameters:

encoded_password – 8-byte encoded password from encode_password().

Returns:

Complete S7 PDU.

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_end_upload_response(response: Dict[str, Any]) → None[source]

Require a canonical END_UPLOAD acknowledgement.

check_userdata_response(response: Dict[str, Any], expected_group: int | None = None, expected_subfunction: int | None = None) → None[source]

Check a USERDATA response for errors.

Verifies both the parameter-level error code and the data section return code.

Parameters:

response – Parsed S7 response from parse_response().

Raises:

S7ProtocolError – If the response indicates an error.

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

static encode_password(password: str) → bytes[source]

Encode an S7 session password into the 8-byte wire format.

Snap7 pads with spaces, XORs with 0x55, and chains each byte from the encoded byte two positions earlier.

Parameters:

password – Plaintext password (max 8 characters).

Returns:

8-byte encoded password block.

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

static parse_download_service_request(pdu: bytes, function: int, block_num: int) → int[source]

Validate a PLC request for a download slice or completion acknowledgement.

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_fragment(response: Dict[str, Any], *, first_fragment: bool = False, include_block_header: bool = False) → Tuple[bytes, bool][source]

Validate and decode one upload response fragment.

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