Connection

class c104.Connection

This class represents connections from a client to a remote server and provides access to meta information and containing stations

add_station(self: c104.Connection, common_address: int) → c104.Station | None

add a new station to this connection 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 = my_connection.add_station(common_address=15)

clock_sync(self: c104.Connection, common_address: int, wait_for_response: bool = True) → bool

send a clock synchronization command to the remote terminal unit (server) the clients OS time is used

Parameters:

common_address (int) – station common address (value between 0 and 65535)
wait_for_response (bool) – block call until command success or failure reponse received?

Returns:

True, if connection is Open, False otherwise

Return type:

bool

Example

>>> if not my_connection.clock_sync(common_address=47):
>>>     raise ValueError("Cannot send clock sync command")

connect(self: c104.Connection) → None

initiate connection to remote terminal unit (server) in a background thread (non-blocking)

Example

>>> my_connection.connect()

counter_interrogation(self: c104.Connection, common_address: int, cause: c104.Cot = c104.Cot.ACTIVATION, qualifier: c104.Qoi = c104.Qoi.Station, wait_for_response: bool = True) → bool

send a counter interrogation command to the remote terminal unit (server)

Parameters:

common_address (int) – station common address (value between 0 and 65535)
cause (Cot (Cause of Transmission)) – cause of transmission
qualifier (Qoi (Qualifier of interrogation)) – qualifier of interrogation
wait_for_response (bool) – block call until command success or failure reponse received?

Returns:

True, if connection is Open, False otherwise

Return type:

bool

Raises:

ValueError – If qualifier is invalid

Example

>>> if not my_connection.counter_interrogation(common_address=47, cause=c104.Cot.ACTIVATION, qualifier=c104.Qoi.Station):
>>>     raise ValueError("Cannot send counter interrogation command")

disconnect(self: c104.Connection) → None

close connection to remote terminal unit (server)

Example

>>> my_connection.disconnect()

get_station(self: c104.Connection, 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_14 = my_connection.get_station(common_address=14)

interrogation(self: c104.Connection, common_address: int, cause: c104.Cot = c104.Cot.ACTIVATION, qualifier: c104.Qoi = c104.Qoi.Station, wait_for_response: bool = True) → bool

send an interrogation command to the remote terminal unit (server)

Parameters:

common_address (int) – station common address (value between 0 and 65535)
cause (Cot (Cause of Transmission)) – cause of transmission
qualifier (Qoi (Qualifier of interrogation)) – qualifier of interrogation
wait_for_response (bool) – block call until command success or failure reponse received?

Returns:

True, if connection is Open, False otherwise

Return type:

bool

Raises:

ValueError – If qualifier is invalid

Example

>>> if not my_connection.interrogation(common_address=47, cause=c104.Cot.ACTIVATION, qualifier=c104.Qoi.Station):
>>>     raise ValueError("Cannot send interrogation command")

mute(self: c104.Connection) → bool

tell the remote terminal unit (server) that this connection is muted, prohibit monitoring messages

Returns:: True, if connection is Open, False otherwise
Return type:: bool

Example

>>> if not my_connection.mute():
>>>     raise ValueError("Cannot mute connection")

on_receive_raw(self: c104.Connection, callable: Callable[[c104.Connection, bytes], None]) → None

set python callback that will be executed on incoming message

Callable signature

Parameters:

connection (Connection) – connection instance
data (bytes) – raw message bytes

Return type:

None

Raises:

ValueError – If callable signature does not match exactly

Example

>>> def con_on_receive_raw(connection: c104.Connection, data: bytes) -> None:
>>>     print("-->| {1} [{0}] | CON {2}:{3}".format(data.hex(), c104.explain_bytes(apdu=data), connection.ip, connection.port))
>>>
>>> my_connection.on_receive_raw(callable=con_on_receive_raw)

on_send_raw(self: c104.Connection, callable: Callable[[c104.Connection, bytes], None]) → None

set python callback that will be executed on outgoing message

Callable signature

Parameters:

connection (Connection) – connection instance
data (bytes) – raw message bytes

Return type:

None

Raises:

ValueError – If callable signature does not match exactly

Example

>>> def con_on_send_raw(connection: c104.Connection, data: bytes) -> None:
>>>     print("<--| {1} [{0}] | CON {2}:{3}".format(data.hex(), c104.explain_bytes(apdu=data), connection.ip, connection.port))
>>>
>>> my_connection.on_send_raw(callable=con_on_send_raw)

on_state_change(self: c104.Connection, callable: Callable[[c104.Connection, c104.ConnectionState], None]) → None

set python callback that will be executed on connection state changes

Callable signature

Parameters:

connection (Connection) – connection instance
state (ConnectionState) – latest connection state

Return type:

None

Raises:

ValueError – If callable signature does not match exactly

Example

>>> def con_on_state_change(connection: c104.Connection, state: c104.ConnectionState) -> None:
>>>     print("CON {0}:{1} STATE changed to {2}".format(connection.ip, connection.port, state))
>>>
>>> my_connection.on_state_change(callable=con_on_state_change)

test(self: c104.Connection, common_address: int, with_time: bool = True, wait_for_response: bool = True) → bool

send a test command to the remote terminal unit (server) the clients OS time is used

Parameters:

common_address (int) – station common address (value between 0 and 65535)
with_time (bool) – send with or without timestamp
wait_for_response (bool) – block call until command success or failure response received?

Returns:

True, if connection is Open, False otherwise

Return type:

bool

Example

>>> if not my_connection.test(common_address=47):
>>>     raise ValueError("Cannot send test command")

unmute()

mute(self: c104.Connection) -> bool

tell the remote terminal unit (server) that this connection is not muted, allow monitoring messages

Returns:: True, if connection is Open, False otherwise
Return type:: bool

Example

>>> if not my_connection.unmute():
>>>     raise ValueError("Cannot unmute connection")

property has_stations

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

Type:: bool

property ip

remote terminal units (server) ip (read-only)

Type:: str

property is_connected

test if connection is opened (read-only)

Type:: bool

property is_muted

test if connection is muted (read-only)

Type:: bool

property originator_address

primary originator address of this connection (0-255)

Type:: int

property port

remote terminal units (server) port (read-only)

Type:: int

property stations: List[Station] list of all Station objects (read-only)