Partner

The Partner class implements S7 peer-to-peer communication for bidirectional data exchange using BSend/BRecv. Both partners have equal rights and can send data asynchronously.

from s7 import Partner

partner = Partner(active=True)
partner.port = 102
partner.r_id = 0x00000001
partner.start_to("0.0.0.0", "192.168.1.10", 0x1300, 0x1301)
partner.set_send_data(b"Hello")
partner.b_send()
partner.stop()

s7.Partner

Unified S7 partner for peer-to-peer communication.

Wraps snap7.partner.Partner so that the s7 package is a drop-in replacement for snap7, including partner functionality.

Usage:

from s7 import Partner

partner = Partner(active=True)
partner.port = 102
partner.r_id = 0x00000001
partner.start_to("0.0.0.0", "192.168.1.10", 0x1300, 0x1301)
partner.set_send_data(b"Hello")
partner.b_send()
partner.stop()

class s7.partner.Partner(active: bool = False, **kwargs: object)[source]

Pure Python S7 partner implementation.

Implements peer-to-peer S7 communication where both partners can send and receive data asynchronously. Supports both active (initiates connection) and passive (waits for connection) modes.

Examples

>>> import snap7
>>> partner = snap7.Partner(active=True)
>>> partner.start_to("0.0.0.0", "192.168.1.10", 0x0100, 0x0102)
>>> partner.set_send_data(b"Hello")
>>> partner.b_send()
>>> partner.stop()

__enter__() → Partner[source]: Context manager entry.

__exit__(exc_type: Type[BaseException] | None, exc_val: BaseException | None, exc_tb: TracebackType | None) → None[source]: Context manager exit.

__init__(active: bool = False, **kwargs: object) → None[source]

Initialize S7 partner.

Parameters:

active – If True, this partner initiates the connection. If False, this partner waits for incoming connections.
**kwargs – Ignored. Kept for backwards compatibility.

as_b_recv() → int[source]

Start asynchronous receive (non-blocking).

Begins listening for incoming partner data in the background. Use check_as_b_recv_completion() or wait_as_b_recv_completion() to check for results.

Returns:: 0 on success (receive initiated), -1 on error

as_b_send() → int[source]

Send data asynchronously (non-blocking).

Note: Call set_send_data() first to set the data to send.

Returns:: 0 on success (send initiated)

b_recv() → int[source]

Receive data synchronously (blocking).

Returns:: 0 on success

b_send() → int[source]

Send data synchronously (blocking).

Note: Call set_send_data() first to set the data to send.

Returns:: 0 on success

check_as_b_recv_completion() → int[source]

Check if async receive completed.

Returns:: 0 if data available, 1 if in progress, -1 on error

check_as_b_send_completion() → Tuple[str, c_int][source]

Check if async send completed.

Returns:: Tuple of (status_string, operation_result)

create(active: bool = False) → None[source]

Creates a Partner.

Note: For pure Python implementation, the partner is created in __init__. This method exists for API compatibility.

Parameters:: active – If True, this partner initiates connections

destroy() → int[source]

Destroy the Partner.

Returns:: 0 on success

get_last_error() → c_int[source]

Get last error code.

Returns:: Last error code

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

Get partner parameter.

Parameters:: parameter – Parameter to read
Returns:: Parameter value

get_recv_data() → bytes | None[source]

Get data received by b_recv() or async receive.

Returns:: Received data or None

get_recv_r_id() → int[source]

Get the R-ID from the last received PDU.

Returns:: R-ID value (0 if no data has been received yet)

get_stats() → Tuple[c_uint, c_uint, c_uint, c_uint][source]

Get partner statistics.

Returns:: Tuple of (bytes_sent, bytes_recv, send_errors, recv_errors)

get_status() → c_int[source]

Get partner status.

Returns:: Status code (0=stopped, 1=running, 2=connected)

get_times() → Tuple[c_int, c_int][source]

Get last operation times.

Returns:: Tuple of (last_send_time_ms, last_recv_time_ms)

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

Set partner parameter.

Parameters:

parameter – Parameter to set
value – Value to set

Returns:

0 on success

set_recv_callback(callback: Callable[[bytes], None] | None = None) → int[source]

Register a callback for incoming data.

The callback is invoked with the received bytes whenever data arrives via b_recv() or async receive.

Parameters:: callback – Function called with received data, or None to clear.
Returns:: 0 on success

set_send_callback(callback: Callable[[int], None] | None = None) → int[source]

Register a callback for completed async sends.

Parameters:: callback – Function called with the result code, or None to clear.
Returns:: 0 on success

set_send_data(data: bytes) → None[source]

Set data to be sent by b_send() or as_b_send().

Parameters:: data – Data to send

start() → int[source]

Start the partner with default parameters.

Returns:: 0 on success

start_to(local_ip: str, remote_ip: str, local_tsap: int, remote_tsap: int) → int[source]

Start the partner with specific connection parameters.

Parameters:

local_ip – Local IP address to bind to
remote_ip – Remote partner IP address (for active mode)
local_tsap – Local TSAP
remote_tsap – Remote TSAP

Returns:

0 on success

stop() → int[source]

Stop the partner and disconnect.

Returns:: 0 on success

wait_as_b_recv_completion(timeout: int = 0) → int[source]

Wait for async receive to complete.

Parameters:: timeout – Timeout in milliseconds (0 for infinite)
Returns:: 0 on success, -1 on timeout/error
Raises:: RuntimeError – If no async receive was ever started

wait_as_b_send_completion(timeout: int = 0) → int[source]

Wait for async send to complete.

Parameters:: timeout – Timeout in milliseconds (0 for infinite)
Returns:: 0 on success, non-zero on error/timeout
Raises:: RuntimeError – If no async operation is in progress

class s7.partner.PartnerStatus[source]: Partner status constants.

snap7.Partner (legacy)

Pure Python S7 partner implementation.

S7 peer-to-peer communication for bidirectional data exchange. Unlike client-server where client requests and server responds, partners have equal rights and can send data asynchronously.

class snap7.partner.Partner(active: bool = False, **kwargs: object)[source]

Pure Python S7 partner implementation.

Implements peer-to-peer S7 communication where both partners can send and receive data asynchronously. Supports both active (initiates connection) and passive (waits for connection) modes.

Examples

>>> import snap7
>>> partner = snap7.Partner(active=True)
>>> partner.start_to("0.0.0.0", "192.168.1.10", 0x0100, 0x0102)
>>> partner.set_send_data(b"Hello")
>>> partner.b_send()
>>> partner.stop()

__enter__() → Partner[source]: Context manager entry.

__exit__(exc_type: Type[BaseException] | None, exc_val: BaseException | None, exc_tb: TracebackType | None) → None[source]: Context manager exit.

__init__(active: bool = False, **kwargs: object) → None[source]

Initialize S7 partner.

Parameters:

active – If True, this partner initiates the connection. If False, this partner waits for incoming connections.
**kwargs – Ignored. Kept for backwards compatibility.

as_b_recv() → int[source]

Start asynchronous receive (non-blocking).

Begins listening for incoming partner data in the background. Use check_as_b_recv_completion() or wait_as_b_recv_completion() to check for results.

Returns:: 0 on success (receive initiated), -1 on error

as_b_send() → int[source]

Send data asynchronously (non-blocking).

Note: Call set_send_data() first to set the data to send.

Returns:: 0 on success (send initiated)

b_recv() → int[source]

Receive data synchronously (blocking).

Returns:: 0 on success

b_send() → int[source]

Send data synchronously (blocking).

Note: Call set_send_data() first to set the data to send.

Returns:: 0 on success

check_as_b_recv_completion() → int[source]

Check if async receive completed.

Returns:: 0 if data available, 1 if in progress, -1 on error

check_as_b_send_completion() → Tuple[str, c_int][source]

Check if async send completed.

Returns:: Tuple of (status_string, operation_result)

create(active: bool = False) → None[source]

Creates a Partner.

Note: For pure Python implementation, the partner is created in __init__. This method exists for API compatibility.

Parameters:: active – If True, this partner initiates connections

destroy() → int[source]

Destroy the Partner.

Returns:: 0 on success

get_last_error() → c_int[source]

Get last error code.

Returns:: Last error code

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

Get partner parameter.

Parameters:: parameter – Parameter to read
Returns:: Parameter value

get_recv_data() → bytes | None[source]

Get data received by b_recv() or async receive.

Returns:: Received data or None

get_recv_r_id() → int[source]

Get the R-ID from the last received PDU.

Returns:: R-ID value (0 if no data has been received yet)

get_stats() → Tuple[c_uint, c_uint, c_uint, c_uint][source]

Get partner statistics.

Returns:: Tuple of (bytes_sent, bytes_recv, send_errors, recv_errors)

get_status() → c_int[source]

Get partner status.

Returns:: Status code (0=stopped, 1=running, 2=connected)

get_times() → Tuple[c_int, c_int][source]

Get last operation times.

Returns:: Tuple of (last_send_time_ms, last_recv_time_ms)

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

Set partner parameter.

Parameters:

parameter – Parameter to set
value – Value to set

Returns:

0 on success

set_recv_callback(callback: Callable[[bytes], None] | None = None) → int[source]

Register a callback for incoming data.

The callback is invoked with the received bytes whenever data arrives via b_recv() or async receive.

Parameters:: callback – Function called with received data, or None to clear.
Returns:: 0 on success

set_send_callback(callback: Callable[[int], None] | None = None) → int[source]

Register a callback for completed async sends.

Parameters:: callback – Function called with the result code, or None to clear.
Returns:: 0 on success

set_send_data(data: bytes) → None[source]

Set data to be sent by b_send() or as_b_send().

Parameters:: data – Data to send

start() → int[source]

Start the partner with default parameters.

Returns:: 0 on success

start_to(local_ip: str, remote_ip: str, local_tsap: int, remote_tsap: int) → int[source]

Start the partner with specific connection parameters.

Parameters:

local_ip – Local IP address to bind to
remote_ip – Remote partner IP address (for active mode)
local_tsap – Local TSAP
remote_tsap – Remote TSAP

Returns:

0 on success

stop() → int[source]

Stop the partner and disconnect.

Returns:: 0 on success

wait_as_b_recv_completion(timeout: int = 0) → int[source]

Wait for async receive to complete.

Parameters:: timeout – Timeout in milliseconds (0 for infinite)
Returns:: 0 on success, -1 on timeout/error
Raises:: RuntimeError – If no async receive was ever started

wait_as_b_send_completion(timeout: int = 0) → int[source]

Wait for async send to complete.

Parameters:: timeout – Timeout in milliseconds (0 for infinite)
Returns:: 0 on success, non-zero on error/timeout
Raises:: RuntimeError – If no async operation is in progress

class snap7.partner.PartnerStatus[source]: Partner status constants.