server.base_skill.BaseSkill

class documentation

class BaseSkill(BaseCallable, AbstractSkill):

Constructor: BaseSkill(name, suspendable, minimum_access_level, precondition_check, ...)

Abstract base skill without state machine logic.

For skills with logic in Python, there is FiniteSkillLogicMixin and ContinuousSkillLogicMixin.

Derived classes must implement at least the reset(), start() and halt() methods for a non-suspendable skill. The _condition_skill_ready() method can be overridden in order to add skill specific conditions before the start() method is allowed to be executed by the state machine.

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`	Write the given `state` without any transition logic checks.
Instance Variable	`_called_skills`	Undocumented
Instance Variable	`_condition`	Undocumented
Instance Variable	`_current_state`	Undocumented
Instance Variable	`_feasibility_check`	Undocumented
Instance Variable	`_precondition_check`	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, precondition_check: BaseSkill | None = None, feasibility_check: BaseSkill | None = None, **kwargs): ¶

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).
precondition_check:`BaseSkill \| None`	Optional reference to corresponding separate precondition check for this skill.
feasibility_check:`BaseSkill \| None`	Optional reference to corresponding separate feasibility check for this skill.
**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

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

Write the given state without any transition logic checks.

Does nothing if given state is already current_state.

_called_skills: list[BaseSkill] = ¶

Undocumented

_condition = ¶

Undocumented

_current_state = ¶

Undocumented

_feasibility_check = ¶

Undocumented

_precondition_check = ¶

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.