openpectus.lang.exec.events

Event subsystem that defines life-time events during method execution.

Attributes

`logger`
`ScopeType`

Classes

`RunStateChange`	Enum where members are also (and must be) strings
`EventListener`	Defines the listener interface and base class for engine life-time events.
`PerformanceTimer`	Monitors execution time of event handlers and warns if they're slow.
`EventEmitter`	Emitter of EventListener events.
`BlockInfo`
`ScopeInfo`

Module Contents

openpectus.lang.exec.events.logger

class openpectus.lang.exec.events.RunStateChange

Bases: enum.StrEnum

Enum where members are also (and must be) strings

PAUSE = 'Pause'

UNPAUSE = 'Unpause'

openpectus.lang.exec.events.ScopeType

class openpectus.lang.exec.events.EventListener

Defines the listener interface and base class for engine life-time events.

run_id: str | None = None

__str__()

Return type:: str

on_engine_configured(): Invoked once on engine startup, after configuration and after the connection to hardware has been established.

_on_before_start(run_id)

Is run by emitter to avoid having all subclasses call super().on_start().

Parameters:: run_id (str)

on_start(run_id)

Is invoked by the Start command when method is started.

Parameters:: run_id (str)

on_block_start(block_info)

Invoked just after a new block is started, before on_tick, in the same engine tick.

Parameters:: block_info (BlockInfo)

on_block_end(block_info, new_block_info)

Invoked just after a block is completed, before on_tick, in the same engine tick. If the new block is the root/program scope, the name is the empty string.

Parameters:

block_info (BlockInfo)
new_block_info (BlockInfo | None)

on_scope_start(scope_info)

Parameters:: scope_info (ScopeInfo)

on_scope_activate(scope_info)

Parameters:: scope_info (ScopeInfo)

on_scope_end(scope_info)

Parameters:: scope_info (ScopeInfo)

on_notify_command(argument)

Parameters:: argument (str)

on_tick(tick_time, increment_time)

Is invoked on each tick.

Intended for NA (calculated/derived) tags to calculate the value for the tick and apply it to the value property.

Parameters:

tick_time (float)
increment_time (float)

on_runstate_change(state_change)

Is invoked when method interpretation is complete.

Parameters:: state_change (RunStateChange)

on_method_end(): Is invoked when interpretation of the last method line is complete. Note: This does not necessarily mean that interpretation ends: - Any alarms still have their condition evaluated and may execute - The method may be modified by the user and the new lines will start executing

on_stop(): Is invoked by the Stop command when method is stopped.

on_engine_shutdown(): Invoked once when engine shuts down

on_method_error(exception)

Invoked by engine when an error state is encountered which is severe enough to cause pause

Parameters:: exception (Exception)

on_connection_status_change(status)

Invoked by system tag Connection Status on value change

Parameters:: status (Literal['Disconnected', 'Connected'])

class openpectus.lang.exec.events.PerformanceTimer(listener)

Bases: EventListener

Monitors execution time of event handlers and warns if they’re slow.

Parameters:: listener (EventListener)

_t0: float = 0.0

_warned = False

_listener

_is_tag

_event_name: str = ''

__enter__()

__exit__(exc_type, exc_value, traceback)

on_engine_configured(): Invoked once on engine startup, after configuration and after the connection to hardware has been established.

_on_before_start(run_id)

Is run by emitter to avoid having all subclasses call super().on_start().

Parameters:: run_id (str)

on_start(run_id)

Is invoked by the Start command when method is started.

Parameters:: run_id (str)

on_block_start(block_info)

Invoked just after a new block is started, before on_tick, in the same engine tick.

Parameters:: block_info (BlockInfo)

on_block_end(block_info, new_block_info)

Invoked just after a block is completed, before on_tick, in the same engine tick. If the new block is the root/program scope, the name is the empty string.

Parameters:

block_info (BlockInfo)
new_block_info (BlockInfo | None)

on_scope_start(scope_info)

Parameters:: scope_info (ScopeInfo)

on_scope_activate(scope_info)

Parameters:: scope_info (ScopeInfo)

on_scope_end(scope_info)

Parameters:: scope_info (ScopeInfo)

on_notify_command(argument)

Parameters:: argument (str)

on_tick(tick_time, increment_time)

Is invoked on each tick.

Intended for NA (calculated/derived) tags to calculate the value for the tick and apply it to the value property.

Parameters:

tick_time (float)
increment_time (float)

on_runstate_change(state_change)

Is invoked when method interpretation is complete.

Parameters:: state_change (RunStateChange)

on_method_end(): Is invoked when interpretation of the last method line is complete. Note: This does not necessarily mean that interpretation ends: - Any alarms still have their condition evaluated and may execute - The method may be modified by the user and the new lines will start executing

on_stop(): Is invoked by the Stop command when method is stopped.

on_engine_shutdown(): Invoked once when engine shuts down

on_method_error(exception)

Invoked by engine when an error state is encountered which is severe enough to cause pause

Parameters:: exception (Exception)

on_connection_status_change(status)

Invoked by system tag Connection Status on value change

Parameters:: status (Literal['Disconnected', 'Connected'])

__str__()

Return type:: str

class openpectus.lang.exec.events.EventEmitter(listeners)

Emitter of EventListener events.

Enables emitting execution events to a collection of listeners.

Parameters:: listeners (Iterable[EventListener])

_use_perf_timers = True

_listeners: list[EventListener] = []

__str__()

Return type:: str

add_listener(listener)

Parameters:: listener (EventListener)

emit_on_engine_configured()

emit_on_start(run_id)

Parameters:: run_id (str)

emit_on_tick(tick_time, increment_time)

Parameters:

tick_time (float)
increment_time (float)

emit_on_block_start(block_name, tick)

Parameters:

block_name (str)
tick (int)

emit_on_block_end(block_name, new_block_name, tick)

Parameters:

block_name (str)
new_block_name (str)
tick (int)

emit_on_scope_start(node_id, scope_type, argument)

Parameters:

node_id (str)
scope_type (ScopeType)
argument (str)

emit_on_scope_activate(node_id, scope_type, argument)

Parameters:

node_id (str)
scope_type (ScopeType)
argument (str)

emit_on_scope_end(node_id, scope_type, argument)

Parameters:

node_id (str)
scope_type (ScopeType)
argument (str)

emit_on_notify_command(argument)

Parameters:: argument (str)

emit_on_runstate_change(state_change)

Parameters:: state_change (RunStateChange)

emit_on_method_end()

emit_on_stop()

emit_on_engine_shutdown()

emit_on_method_error(exception)

Parameters:: exception (Exception)

emit_on_connection_status_change(status)

Parameters:: status (Literal['Disconnected', 'Connected'])

class openpectus.lang.exec.events.BlockInfo

name: str

tick: int

property key: str

Gets the block name as a unique name across a single excution.

Return type:: str

class openpectus.lang.exec.events.ScopeInfo

node_id: str

argument: str

scope_type: ScopeType