Common functionality¶

Contains helper functions.

class secsgem.common.CallbackHandler[source]¶

Bases: object

Handler for callbacks for HSMS/SECS/GEM events.

This handler manages callbacks for events that can happen on a handler for a connection.

class secsgem.common.EventProducer[source]¶

Bases: object

Manages the consumers for the events and handles firing events.

fire(event: str, data: Dict[str, Any])[source]¶

Fire a event.

calls all the available handlers for a specific event

Parameters:	event (string) – name of the event data (dict) – data connected to this event

targets¶: Targets used as consumer for this producer.

class secsgem.common.Fysom(cfg)[source]¶

Bases: object

Fysom state machine.

isstate(state)[source]¶: Get state.

can(event)[source]¶: Check if transition possible.

cannot(event)[source]¶: Check if transition is not possible.

secsgem.common.format_hex(text: bytes) → str[source]¶

Return byte arrays (string) formated as hex numbers.

Example:

>>> import secsgem.common
>>>
>>> data = b"asdfg"
>>> secsgem.common.format_hex(data)
'61:73:64:66:67'

Parameters:	text (string) – byte array
Returns:	Formated text
Return type:	string

secsgem.common.function_name(function) → str[source]¶

Get name of function or method.

Returns:	function/method name
Return type:	string

secsgem.common.indent_block(block: str, spaces: int = 2) → str[source]¶

Indent a multiline string by a number of spaces.

Parameters:	block (string) – input text spaces (integer) – number of spaces to prepend to each line
Returns:	indented text
Return type:	string

secsgem.common.is_windows() → bool[source]¶

Return True if running on windows.

Returns:	Is windows system
Return type:	bool

secsgem.common.is_errorcode_ewouldblock(errorcode: int) → bool[source]¶

Check if the errorcode is a would-block error.

Parameters:	errorcode – Code of the error
Returns:	True if blocking error code

class secsgem.common.Protocol[source]¶

Bases: abc.ABC

Abstract base class for a protocol.

events¶: Property for event handling.

timeouts¶: Property for timeout.

secs_decode¶: Get secs decode.

serialize_data() → Dict[str, Any][source]¶: Get protocol serialized data for debugging.

enable()[source]¶: Enable the connection.

disable()[source]¶: Disable the connection.

send_stream_function(function: SecsStreamFunction) → bool[source]¶

Send the packet and wait for the response.

Parameters:	function (`secsgem.secs.functionbase.SecsStreamFunction`) – packet to be sent

send_and_waitfor_response(function: SecsStreamFunction) → typing.Optional[Packet][source]¶

Send the packet and wait for the response.

Parameters:	function (`secsgem.secs.functionbase.SecsStreamFunction`) – packet to be sent
Returns:	Packet that was received
Return type:	`secsgem.common.Packet`

send_response(function: SecsStreamFunction, system: int) → bool[source]¶

Send response function for system.

Parameters:	function (`secsgem.secs.functionbase.SecsStreamFunction`) – function to be sent system (int) – system to reply to

class secsgem.common.Packet[source]¶

Bases: abc.ABC

Abstract base class for a packet.

header¶: Get the header.

class secsgem.common.Header(system: int, session_id: int, stream: int, function: int)[source]¶

Bases: abc.ABC

Abstract base class for a packet header.

session_id¶: Get session id.

stream¶: Get stream.

function¶: Get function.

system¶: Get system.

class secsgem.common.Timeouts[source]¶

Bases: object

Timeouts.

T3 = 45.0¶

T5 = 10.0¶

T6 = 5.0¶