Server

The s7.Server is the recommended server for testing. It wraps both a legacy S7 server and an S7CommPlus server, so test environments can serve both protocol stacks simultaneously.

from s7 import Server

server = Server()
server.start(tcp_port=1102)

For quick testing with the legacy server, you can also use the mainloop helper:

from snap7.server import mainloop

mainloop(tcp_port=1102)

s7.Server

Unified S7 server supporting both legacy S7 and S7CommPlus clients.

Wraps both a legacy snap7.server.Server and an S7CommPlusServer so that test environments can serve both protocol stacks simultaneously.

Usage:

from s7 import Server

server = Server()
server.start(tcp_port=102, s7commplus_port=11020)

class s7.server.Server[source]

Unified S7 server for testing.

Runs a legacy S7 server and optionally an S7CommPlus server side by side.

__getattr__(name: str) → Any[source]: Delegate unknown methods to the legacy server.

get_db(db_number: int) → DataBlock | None[source]: Get a registered data block.

property legacy_server: Server: Direct access to the legacy S7 server.

register_db(db_number: int, variables: dict[str, tuple[str, int]], size: int = 0) → DataBlock[source]

Register a data block on the S7CommPlus server.

Parameters:

db_number – Data block number
variables – Dict of {name: (type_name, offset)}
size – Total DB size in bytes (auto-calculated if 0)

Returns:

The created DataBlock

register_raw_db(db_number: int, data: bytearray) → DataBlock[source]

Register a raw data block on the S7CommPlus server.

Parameters:

db_number – Data block number
data – Raw bytearray backing the data block

Returns:

The created DataBlock

property s7commplus_server: S7CommPlusServer: Direct access to the S7CommPlus server.

start(tcp_port: int = 102, s7commplus_port: int | None = None, *, use_tls: bool = False, tls_cert: str | None = None, tls_key: str | None = None) → None[source]

Start the server(s).

Parameters:

tcp_port – Port for the legacy S7 server.
s7commplus_port – Port for the S7CommPlus server. If None, only the legacy server is started.
use_tls – Whether to enable TLS on the S7CommPlus server.
tls_cert – Path to TLS certificate (PEM).
tls_key – Path to TLS private key (PEM).

stop() → None[source]: Stop all servers.

snap7.Server (legacy)

Legacy S7 server implementation.

Provides a complete S7 server emulator for the classic S7 protocol. For new projects, use s7.Server instead, which supports both legacy S7 and S7CommPlus clients.

class snap7.server.CPUState(*values)[source]: S7 CPU states.

class snap7.server.Server(log: bool = True, **kwargs: object)[source]

Legacy S7 server implementation.

Emulates a Siemens S7 PLC for testing and development purposes. For new projects, use s7.Server instead.

Examples

>>> from s7 import Server
>>> server = Server()
>>> server.start()
>>> # ... register areas and handle clients
>>> server.stop()

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

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

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

Initialize S7 server.

Parameters:

log – Enable event logging
**kwargs – Ignored. Kept for backwards compatibility.

clear_events() → int[source]

Clear event queue.

Returns:: 0 on success

create() → None[source]: Create the server (no-op for compatibility).

destroy() → None[source]: Destroy the server.

event_text(event: SrvEvent) → str[source]

Get event text description.

Parameters:: event – Server event
Returns:: Event description string

get_mask(mask_kind: int) → int[source]

Get event mask.

Parameters:: mask_kind – Mask type (0=Event, 1=Log)
Returns:: Event mask value

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

Get server parameter.

Parameters:: param – Parameter type
Returns:: Parameter value
Raises:: RuntimeError – If parameter is not valid for server

get_status() → Tuple[str, str, int][source]

Get server status.

Returns:: Tuple of (server_status, cpu_status, client_count)

lock_area(area: SrvArea, index: int) → int[source]

Lock a memory area.

Parameters:

area – Memory area type
index – Area index

Returns:

0 on success

Raises:

RuntimeError – If area is not registered

pick_event() → SrvEvent | bool[source]

Pick an event from the queue.

Returns:: Server event if available, False if no events

register_area(area: SrvArea, index: int, userdata: bytearray | Array) → int[source]

Register a memory area with the server.

Parameters:

area – Memory area type
index – Area index/number
userdata – Initial data for the area (bytearray or ctypes array)

Returns:

0 on success

set_cpu_status(status: int) → int[source]

Set CPU status.

Parameters:: status – CPU status code (0=Unknown, 4=Stop, 8=Run)
Returns:: 0 on success
Raises:: ValueError – If status is invalid

set_events_callback(callback: Callable[[SrvEvent], Any]) → int[source]

Set callback for server events.

Parameters:: callback – Event callback function
Returns:: 0 on success

set_mask(kind: int = 0, mask: int = 0) → int[source]

Set event mask.

Parameters:

kind – Mask type (0=Event, 1=Log)
mask – Mask value

Returns:

0 on success

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

Set server parameter.

Parameters:

param – Parameter type
value – Parameter value

Returns:

0 on success

set_read_events_callback(callback: Callable[[SrvEvent], Any]) → int[source]

Set callback for read events.

Parameters:: callback – Read event callback function
Returns:: 0 on success

set_rw_area_callback(callback: Callable[[Any], int]) → int[source]

Set callback for read/write area operations.

This is a stub for API compatibility with the C library’s Srv_SetRWAreaCallback. In the native implementation, read/write operations are handled directly.

Parameters:: callback – RW area callback function
Returns:: 0 on success

start(tcp_port: int = 102) → int[source]

Start the S7 server.

Parameters:: tcp_port – TCP port to listen on
Returns:: 0 on success

start_to(ip: str, tcp_port: int = 102) → int[source]

Start server on a specific interface.

Parameters:

ip – IP address to bind to
tcp_port – TCP port to listen on

Returns:

0 on success

stop() → int[source]

Stop the S7 server.

Returns:: 0 on success

unlock_area(area: SrvArea, index: int) → int[source]

Unlock a memory area.

Parameters:

area – Memory area type
index – Area index

Returns:

0 on success

unregister_area(area: SrvArea, index: int) → int[source]

Unregister a memory area.

Parameters:

area – Memory area type
index – Area index

Returns:

0 on success

class snap7.server.ServerISOConnection(client_socket: socket)[source]

ISO connection wrapper for server-side communication.

__init__(client_socket: socket)[source]: Initialize server ISO connection.

accept_connection() → bool[source]: Accept ISO connection from client.

receive_data() → bytes[source]: Receive data from client.

send_data(data: bytes) → None[source]: Send data to client.

class snap7.server.ServerState(*values)[source]: S7 server states.

snap7.server.mainloop(tcp_port: int = 1102, init_standard_values: bool = False) → None[source]

Initialize a pure Python S7 server with default values.

Parameters:

tcp_port – Port that the server will listen on
init_standard_values – If True, initialize some default values