client.remote_skill.RemoteSkill

class documentation

class RemoteSkill(RemoteCallable, UaDataChangeSubscriber):

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

View In Hierarchy

The remote interface for a server-side BaseSkill.

Method __init__ Create a new instance. Must be asynchronously initialized with init() before use.
Method dto Return the data transfer object.
Async Method halt Call the halt method on the remote skill. User is responsible for handling any errors.
Async Method read_current_state Read the current state of the remote skill.
Async Method reset Call the reset method on the remote skill. User is responsible for handling any errors.
Async Method start Call the start method on the remote skill. User is responsible for handling any errors.
Async Method suspend Call the suspend method on the remote skill. User is responsible for handling any errors.
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 Block until the skill either reached the given state or was halted. Given timeout is in seconds.
Instance Variable state_changed Undocumented
Property current_state Cached current state of the remote skill.
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
Instance Variable _current_state 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.

def dto(self) -> RemoteSkillDTO:

Return the data transfer object.

async def halt(self, wait_for: SkillState | None = None):

Call the halt method on the remote skill. User is responsible for handling any errors.

async def read_current_state(self) -> SkillState:

Read the current state of the remote skill.

async def reset(self, *, wait_for: SkillState | None = None):

Call the reset method on the remote skill. User is responsible for handling any errors.

async def start(self, wait_for: SkillState | None = None):

Call the start method on the remote skill. User is responsible for handling any errors.

async def suspend(self, wait_for: SkillState | None = None):

Call the suspend method on the remote skill. User is responsible for handling any errors.

@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):

Block until the skill either reached the given state or was halted. Given timeout is in seconds.

state_changed =

Undocumented

@property
current_state: SkillState =

Cached current state of the remote skill.

@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, val: str | LocalizedText):

Undocumented

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

Undocumented

_current_state =

Undocumented

_ua_nodes: _RemoteSkillUaNodes =

Undocumented

_ua_type =

Undocumented