Server

python-snap7 provides two server implementations:

s7commplus.Server: S7CommPlus server emulator
s7.server.Server: Legacy S7 server for testing

from s7commplus import Server

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

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

from s7.server import mainloop

mainloop(tcp_port=1102)

s7commplus.Server

S7CommPlus server emulator for testing.

Emulates an S7-1200/1500 PLC for integration testing without real hardware. Handles the S7CommPlus protocol including: - COTP connection setup (reuses ISOTCPConnection transport) - CreateObject session handshake - Explore (browse registered data blocks and variables) - GetMultiVariables / SetMultiVariables (read/write by address) - Internal PLC memory model with thread-safe access - V2 protocol emulation with TLS and IntegrityId tracking

Supports both V1 (no TLS) and V2 (TLS + IntegrityId) emulation.

Usage:

server = S7CommPlusServer()
server.register_db(1, {"temperature": ("Real", 0), "pressure": ("Real", 4)})
server.start(port=11020)

# V2 server with TLS:
server = S7CommPlusServer(protocol_version=ProtocolVersion.V2)
server.start(port=11020, use_tls=True, tls_cert="cert.pem", tls_key="key.pem")

class s7commplus.server.CPUState(*values)[source]: Emulated CPU operational state.

class s7commplus.server.DBVariable(name: str, soft_datatype: int, byte_offset: int)[source]: A variable in a data block.

class s7commplus.server.DataBlock(number: int, size: int = 1024)[source]

An emulated PLC data block with named variables.

add_variable(name: str, type_name: str, byte_offset: int) → None[source]

Register a named variable in this data block.

Parameters:

name – Variable name (e.g. “temperature”)
type_name – PLC type name (e.g. “Real”, “Int”, “Bool”)
byte_offset – Byte offset within the data block

read(offset: int, size: int) → bytes[source]: Read bytes from the data block.

read_variable(name: str) → tuple[int, bytes][source]

Read a named variable.

Returns:: Tuple of (wire_datatype, raw_bytes)

write(offset: int, data: bytes) → None[source]: Write bytes to the data block.

write_variable(name: str, data: bytes) → None[source]: Write a named variable.

class s7commplus.server.S7CommPlusServer(protocol_version: int = ProtocolVersion.V1, public_key_fingerprint: str | None = None, session_challenge: bytes | None = None)[source]

S7CommPlus PLC emulator for testing.

Emulates an S7-1200/1500 PLC with: - Internal data block storage with named variables - S7CommPlus protocol handling (V1 and V2) - V2 TLS support with IntegrityId tracking - Multi-client support (threaded) - CPU state management

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

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

Register a data block with named variables.

Parameters:

db_number – Data block number (e.g. 1 for DB1)
variables – Dict mapping variable name to (type_name, byte_offset) e.g. {“temperature”: (“Real”, 0), “count”: (“Int”, 4)}
size – Data block size in bytes

Returns:

The created DataBlock

Example:

server.register_db(1, {
    "temperature": ("Real", 0),
    "pressure": ("Real", 4),
    "running": ("Bool", 8),
    "count": ("DInt", 10),
})

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

Register a data block with raw data (no named variables).

Parameters:

db_number – Data block number
data – Initial data block content

Returns:

The created DataBlock

start(host: str = '0.0.0.0', port: int = 11020, use_tls: bool = False, tls_cert: str | None = None, tls_key: str | None = None, tls_ca: str | None = None) → None[source]

Start the server.

Parameters:

host – Bind address
port – TCP port to listen on
use_tls – Whether to wrap client sockets with TLS after InitSSL
tls_cert – Path to server TLS certificate (PEM)
tls_key – Path to server private key (PEM)
tls_ca – Path to CA certificate for client verification (PEM)

stop() → None[source]: Stop the server.

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