server.base_skill.

class documentation

class _BaseSkill(BaseCallable, AbstractSkill):

Known subclasses: server.base_skill.BaseCheck, server.base_skill.BaseSkill

Constructor: _BaseSkill(name, suspendable, minimum_access_level, parent, ...)

Undocumented

Method	`__init__`	Cooperative Constructor. Create a new skill instance.
Async Method	`call_other_continuous_skill`	Call another continuous skill from this skill.
Async Method	`call_other_finite_skill`	Call another finite skill from this skill.
Async Method	`wait_for_state`	Undocumented
Property	`current_state`	Return the current state of the skill.
Property	`ua_state_machine`	Undocumented
Async Method	`_condition_is_suspendable`	Undocumented
Async Method	`_get_sub_node`	Undocumented
Async Method	`_halt_running_called_skills`	Undocumented
Async Method	`_init`	Undocumented
Async Method	`_write_current_state`	Undocumented
Instance Variable	`_called_skills`	Undocumented
Instance Variable	`_condition`	Undocumented
Instance Variable	`_current_state`	Undocumented
Instance Variable	`_suspendable`	Undocumented
Instance Variable	`_ua_current_state`	OPC UA node if the state variable of the state machine.
Instance Variable	`_ua_state_machine`	OPC UA base node of state machine.

def __init__(self, *, name: str | None = None, suspendable: bool = False, minimum_access_level: int | None = None, parent: UaObject | None = None, **kwargs): ¶

overridden in server.base_skill.BaseSkill

Cooperative Constructor. Create a new skill instance.

Parameters
name:`str \| None`	name of the skill.
suspendable:`bool`	whether the skill can be suspended/resumed.
minimum_access_level:`int \| None`	Minimum required access level to start the skill (see User for more info).
parent:`UaObject \| None`	Undocumented
**kwargs	Undocumented

async def call_other_continuous_skill(self, skill: BaseSkill, *, wait_for_running: bool = True, timeout: float | None = 10.0): ¶

Call another continuous skill from this skill.

Parameters
skill:`BaseSkill`	the continuous skill to call.
wait_for_running:`bool`	should this method block until the timeout is reached or the skill is running?
timeout:`float \| None`	timeout in seconds, can be disabled by setting it to None.

Raises
`TypeError`	if given `skill` is not continuous.
`asyncio.TimeoutError`	if the `timeout` is reached.

@asynccontextmanager

async def call_other_finite_skill(self, skill: BaseSkill, *, wait_for_completion: bool = True, timeout: float | None = 60.0, reset_after_completion: bool = True) -> AsyncGenerator: ¶

Call another finite skill from this skill.

Parameters
skill:`BaseSkill`	the finite skill to call.
wait_for_completion:`bool`	should this method block until the timeout is reached or the skill is completed?
timeout:`float \| None`	timeout in seconds, can be disabled by setting it to None.
reset_after_completion:`bool`	should the called skill be reset after completion?

Returns
`AsyncGenerator`	Undocumented

Raises
`TypeError`	if given `skill` is not finite.
`asyncio.TimeoutError`	if the `timeout` is reached.

@override

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

Undocumented

@property
@override

current_state: SkillState = ¶

Return the current state of the skill.

@property
@override

ua_state_machine: Node = ¶

Undocumented

@override

async def _condition_is_suspendable(self, _user: UserAuthorization) -> bool: ¶

Undocumented

@override

async def _get_sub_node(self) -> Node: ¶

Undocumented

async def _halt_running_called_skills(self): ¶

Undocumented

@override

async def _init(self): ¶

Undocumented

@override

async def _write_current_state(self, state: SkillState): ¶

Undocumented

_called_skills: list[BaseSkill] = ¶

Undocumented

_condition = ¶

Undocumented

_current_state = ¶

Undocumented

_suspendable = ¶

Undocumented

_ua_current_state: Node = ¶

OPC UA node if the state variable of the state machine.

_ua_state_machine: Node = ¶

OPC UA base node of state machine.

server.base_skill._BaseSkill

`server.base_skill._BaseSkill`