client.remote_skill.RemoteSkill

class documentation

class RemoteSkill(RemoteCallable, UaDataChangeSubscriber):

Constructor: RemoteSkill(name, ua_node, server, parent, ...)

The remote interface for a server-side BaseSkill.

Method	`__init__`	Create a new instance. Must be asynchronously initialized with `init()` before use.
Async Method	`halt`	Halt the skill. With `wait`, block until it is halted.
Async Method	`read_current_state`	Read the current state of the remote skill.
Async Method	`reset`	Reset the skill. With `wait`, block until it is ready.
Async Method	`start`	Start the skill. With `wait`, block until it is running (or already completed).
Async Method	`suspend`	Suspend the skill. With `wait`, block until it is suspended.
Async Method	`ua_on_data_change`	Handle current state data change notifications from OPC UA.
Async Method	`ua_read_type`	Read the OPC UA type, resolved from node ID browse name.
Async Method	`wait_for_state`	Advanced. Block until the skill reaches the given state.
Instance Variable	`state_changed`	Callback when the current state or its status changed.
Property	`current_state`	Cached current state of the remote skill.
Property	`current_state_sample`	Cached current state with its status and timestamps.
Property	`is_finite`	Whether the remote skill is finite or not (= continuous).
Property	`suspendable`	Indicates whether the skill is suspendable.
Async Method	`_browse_state_machine`	Undocumented
Async Method	`_handle_state_change`	Undocumented
Method	`_init_handle_node`	Undocumented
Async Method	`_wait_for_states`	Undocumented
Instance Variable	`_current_state`	Undocumented
Instance Variable	`_halt_count`	Undocumented
Instance Variable	`_state_condition`	Undocumented
Instance Variable	`_ua_nodes`	Undocumented
Instance Variable	`_ua_type`	Undocumented

def __init__(self, *, name: str, ua_node: Node, server: RemoteServer, parent: BaseRemoteComponent, browse_rules: Iterable[BrowseRule] | None): ¶

Create a new instance. Must be asynchronously initialized with init() before use.

async def halt(self, *, wait: bool = False): ¶

Halt the skill. With wait, block until it is halted.

async def read_current_state(self) -> SkillState: ¶

Read the current state of the remote skill.

async def reset(self, *, wait: bool = False): ¶

Reset the skill. With wait, block until it is ready.

Raises
`SkillHaltedError`	The skill halted while waiting.

async def start(self, *, wait: bool = False): ¶

Start the skill. With wait, block until it is running (or already completed).

Raises
`SkillHaltedError`	The skill halted while waiting.

async def suspend(self, *, wait: bool = False): ¶

Suspend the skill. With wait, block until it is suspended.

Raises
`SkillHaltedError`	The skill halted while waiting.

@override

async def ua_on_data_change(self, node: Node, val: Any, data: DataChangeNotif): ¶

Handle current state data change notifications from OPC UA.

@override

async def ua_read_type(self) -> str: ¶

Read the OPC UA type, resolved from node ID browse name.

async def wait_for_state(self, target_state: SkillState): ¶

Advanced. Block until the skill reaches the given state.

Returns immediately if the skill is already in that state. Use asyncio.timeout() to bound the wait.

Only the latest state is observed, so short-lived states can be missed and the wait then never returns. Prefer stable states (READY, SUSPENDED, COMPLETED, HALTED). Avoid transient states (e.g. STARTING), and be cautious with RUNNING for short finite skills.

Raises
`SkillHaltedError`	The skill halted while waiting for any state other than `HALTED`.

state_changed = ¶

Callback when the current state or its status changed.

@property

current_state: SkillState = ¶

Cached current state of the remote skill.

Is HALTED (the safest assumption) until the server reported a state, and while it reports no value, e.g. with a bad status (see current_state_sample).

@property

current_state_sample: Sample[SkillState] = ¶

Cached current state with its status and timestamps.

Has the status BadWaitingForInitialData until the server reported a state.

@property

is_finite: bool = ¶

Whether the remote skill is finite or not (= continuous).

@property

suspendable: bool = ¶

Indicates whether the skill is suspendable.

async def _browse_state_machine(self, ua_node: Node): ¶

Undocumented

async def _handle_state_change(self, data_value: ua.DataValue): ¶

Undocumented

@override

def _init_handle_node(self, ua_node: Node, browse_name: ua.QualifiedName) -> list[Awaitable]: ¶

Undocumented

async def _wait_for_states(self, target_states: Collection[SkillState], *, halt_count: int): ¶

Undocumented

_current_state = ¶

Undocumented

_halt_count: int = ¶

Undocumented

_state_condition = ¶

Undocumented

_ua_nodes: _RemoteSkillUaNodes = ¶

Undocumented

_ua_type = ¶

Undocumented