Server

class c104.Server

This class represents a local server and provides access to meta information and containing stations

__init__(self: c104.Server, ip: str = '0.0.0.0', port: int = 2404, tick_rate_ms: int = 1000, max_connections: int = 0, transport_security: c104.TransportSecurity | None = None) → None

create a new 104er server

Parameters:

  • ip (str) – listening server ip address

  • port (int) – listening server port

  • tick_rate_ms (int) – server thread update interval

  • max_connections (int) – maximum number of clients allowed to connect

  • transport_security (TransportSecurity) – TLS configuration object

Example

>>> my_server = c104.Server(ip="0.0.0.0", port=2404, tick_rate_ms=1000, max_connections=0)
add_station(self: c104.Server, common_address: int) → c104.Station | None

add a new station to this server and return the new station object

Parameters:

common_address (int) – station common address (value between 0 and 65535)

Returns:

station object, if station was added, else None

Return type:

Station

Example

>>> station_1 = my_server.add_station(common_address=15)
get_station(self: c104.Server, common_address: int) → c104.Station | None

get a station object via common address

Parameters:

common_address (int) – station common address (value between 0 and 65535)

Returns:

station object, if found, else None

Return type:

Station

Example

>>> station_2 = my_server.get_connection(common_address=14)
on_clock_sync(self: c104.Server, callable: Callable[[c104.Server, str, datetime.datetime], c104.ResponseState]) → None

set python callback that will be executed on incoming clock sync command

Callable signature

Parameters:

  • server (Server) – server instance

  • ip (str) – client connection request ip

  • date_time (datetime.datetime) – clients current clock time

Returns:

success or failure of clock sync command

Return type:

ResponseState

Raises:

ValueError – If callable signature does not match exactly

Example

>>> import datetime
>>>
>>> def sv_on_clock_sync(server: c104.Server, ip: str, date_time: datetime.datetime) -> c104.ResponseState:
>>>     print("->@| Time {0} from {1} | SERVER {2}:{3}".format(date_time, ip, server.ip, server.port))
>>>     return c104.ResponseState.SUCCESS
>>>
>>> my_server.on_clock_sync(callable=sv_on_clock_sync)
on_connect(self: c104.Server, callable: Callable[[c104.Server, ip], bool]) → None

set python callback that will be executed on incoming connection requests

Callable signature

Parameters:

  • server (Server) – server instance

  • ip (str) – client connection request ip

Returns:

accept or reject the connection request

Return type:

bool

Raises:

ValueError – If callable signature does not match exactly

Example

>>> def sv_on_connect(server: c104.Server, ip: str) -> bool:
>>>     print("<->| {0} | SERVER {1}:{2}".format(ip, server.ip, server.port))
>>>     return ip == "127.0.0.1"
>>>
>>> my_server.on_connect(callable=sv_on_connect)
on_receive_raw(self: c104.Server, callable: Callable[[c104.Server, bytes], None]) → None

set python callback that will be executed on incoming message

Callable signature

Parameters:

  • server (Server) – server instance

  • data (bytes) – raw message bytes

Return type:

None

Raises:

ValueError – If callable signature does not match exactly

Example

>>> def sv_on_receive_raw(server: c104.Server, data: bytes) -> None:
>>>     print("-->| {1} [{0}] | SERVER {2}:{3}".format(data.hex(), c104.explain_bytes(apdu=data), server.ip, server.port))
>>>
>>> my_server.on_receive_raw(callable=sv_on_receive_raw)
on_send_raw(self: c104.Server, callable: Callable[[c104.Server, bytes], None]) → None

set python callback that will be executed on outgoing message

Callable signature

Parameters:

  • server (Server) – server instance

  • data (bytes) – raw message bytes

Return type:

None

Raises:

ValueError – If callable signature does not match exactly

Example

>>> def sv_on_send_raw(server: c104.Server, data: bytes) -> None:
>>>     print("<--| {1} [{0}] | SERVER {2}:{3}".format(data.hex(), c104.explain_bytes(apdu=data), server.ip, server.port))
>>>
>>> my_server.on_send_raw(callable=sv_on_send_raw)
on_unexpected_message(self: c104.Server, callable: Callable[[c104.Server, c104.IncomingMessage, c104.Umc], None]) → None

set python callback that will be executed on unexpected incoming messages

Callable signature

Parameters:
Return type:

None

Raises:

ValueError – If callable signature does not match exactly

Example

>>> def sv_on_unexpected_message(server: c104.Server, message: c104.IncomingMessage, cause: c104.Umc) -> None:
>>>     print("->?| {1} from CLIENT OA {0} | SERVER {2}:{3}".format(message.originator_address, cause, server.ip, server.port))
>>>
>>> my_server.on_unexpected_message(callable=sv_on_unexpected_message)
start(self: c104.Server) → None

open local server socket for incoming connections

Raises:

RuntimeError – If server thread failed to start

Example

>>> my_server.start()
stop(self: c104.Server) → None

stop local server socket

Example

>>> my_server.stop()
property active_connection_count

get number of active (open and not muted) connections to clients (read-only)

Type:

int

property has_active_connections

test if Server has active (open and not muted) connections to clients (read-only)

Type:

bool

property has_open_connections

test if Server has open connections to clients (read-only)

Type:

bool

property has_stations

test if local server has at least one station (read-only)

Type:

bool

property ip

ip address the server will accept connections on, “0.0.0.0” = any (read-only)

Type:

str

property is_running

test if server is running (read-only)

Type:

bool

property max_connections

maximum number of open connections, 0 = no limit

Type:

int

property open_connection_count

get number of open connections to clients (read-only)

Type:

int

property port

port number the server will accept connections on (read-only)

Type:

int

property stations

list of all local Station objects (read-only)

Type:

List[Station]