server.mixins.skill_logic_mixins.

class documentation

class _SkillLogicMixin(AbstractSkill, AbstractUaLogger):

Known subclasses: server.mixins.skill_logic_mixins.ContinuousSkillLogicMixin, server.mixins.skill_logic_mixins.FiniteSkillLogicMixin

Constructor: _SkillLogicMixin(**kwargs)

View In Hierarchy

Base class for all python-native skills (with state machine logic).

Method	`__init__`	Cooperative constructor.
Async Method	`halt`	Undocumented
Async Method	`lifecycle_skill_state_machine`	Initialize the skill state machine.
Async Method	`reset`	Undocumented
Async Method	`start`	Undocumented
Async Method	`suspend`	Undocumented
Method	`_add_transitions`	Undocumented
Async Method	`_after_halting`	Is called by the finite state machine after the `SkillState.HALTING` state is entered.
Async Method	`_after_resetting`	Is called by the finite state machine after the `SkillState.RESETTING` state is entered.
Async Method	`_after_running`	Undocumented
Async Method	`_after_starting`	Is called by the finite state machine after the `SkillState.STARTING` state is entered.
Async Method	`_after_suspending`	Is called by the finite state machine after the `SkillState.SUSPENDING` state is entered.
Async Method	`_condition_reset_allowed`	Implement additional custom conditions checked before resetting the skill.
Async Method	`_condition_start_allowed`	Implement additional custom conditions checked before starting the skill.
Async Method	`_condition_suspend_allowed`	Implement additional custom conditions checked before suspending the skill.
Async Method	`_handle_halting`	Handle user-specific logic during the `SkillState.HALTING` state.
Async Method	`_handle_resetting`	Handle user-specific logic during `SkillState.RESETTING` state.
Async Method	`_handle_running`	Handle user-specific logic during the `SkillState.RUNNING` state.
Async Method	`_handle_starting`	Handle user-specific logic during the `SkillState.STARTING` state.
Async Method	`_handle_suspending`	Handle user-specific logic during the `SkillState.SUSPENDING` state.
Async Method	`_suspend_point`	Blocks if current_state is `SkillState.SUSPENDING` and allow advance to `SkillState.SUSPENDED`.
Async Method	`_ua_halt`	Undocumented
Async Method	`_ua_reset`	Undocumented
Async Method	`_ua_start`	Undocumented
Async Method	`_ua_suspend`	Undocumented
Async Method	`_wrap_handle_method`	Undocumented
Instance Variable	`__tasks`	Keeps track of tasks related to skill execution and their name. These tasks will get canceled and emptied before handle_halting() is called.
Instance Variable	`_machine`	Skill Finite State Machine.
Instance Variable	`_paused_event`	Undocumented
Instance Variable	`_resume_event`	Undocumented

def __init__(self, **kwargs): ¶

Cooperative constructor.

@override

async def halt(self): ¶

Undocumented

@lifecycle(after=BaseCallable.lifecycle_callable)

async def lifecycle_skill_state_machine(self) -> AsyncGenerator[None]: ¶

Initialize the skill state machine.

@override

async def reset(self): ¶

Undocumented

@override

async def start(self): ¶

Undocumented

@override

async def suspend(self): ¶

Undocumented

def _add_transitions(self): ¶

overridden in server.mixins.skill_logic_mixins.FiniteSkillLogicMixin

Undocumented

async def _after_halting(self, *args, **kwargs): ¶

Is called by the finite state machine after the SkillState.HALTING state is entered.

async def _after_resetting(self, *args, **kwargs): ¶

Is called by the finite state machine after the SkillState.RESETTING state is entered.

@abstractmethod

async def _after_running(self, *args, **kwargs): ¶

overridden in server.mixins.skill_logic_mixins.ContinuousSkillLogicMixin, server.mixins.skill_logic_mixins.FiniteSkillLogicMixin

Undocumented

async def _after_starting(self, *args, **kwargs): ¶

Is called by the finite state machine after the SkillState.STARTING state is entered.

async def _after_suspending(self, *args, **kwargs): ¶

Is called by the finite state machine after the SkillState.SUSPENDING state is entered.

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

Implement additional custom conditions checked before resetting the skill.

Returns
`bool`	Always `True`, else an exception is raised.

Raises
`UaStatusCodeError`	If the skill is not allowed to reset.

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

Implement additional custom conditions checked before starting the skill.

Returns
`bool`	Always `True`, else an exception is raised.

Raises
`UaStatusCodeError`	If the skill is not allowed to start.

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

Implement additional custom conditions checked before suspending the skill.

Returns
`bool`	Always `True`, else an exception is raised.

Raises
`UaStatusCodeError`	If the skill is not allowed to suspend.

async def _handle_halting(self): ¶

Handle user-specific logic during the SkillState.HALTING state.

After this function is done, the skill will automatically advance to the SkillState.HALTED state.

async def _handle_resetting(self): ¶

Handle user-specific logic during SkillState.RESETTING state.

After this function is done, the skill will automatically advance to the SkillState.READY state.

@abstractmethod

async def _handle_running(self): ¶

Handle user-specific logic during the SkillState.RUNNING state.

After this function is done, the skill will automatically advance to the SkillState.COMPLETED state if it is finite. Continuous skills stay in SkillState.RUNNING.

Note: This continues to be executed during SkillState.SUSPENDING state. Use _suspend_point method to block the task executing this function and advance to SkillState.SUSPENDED state.

async def _handle_starting(self): ¶

Handle user-specific logic during the SkillState.STARTING state.

After this function is done, the skill will automatically advance to the SkillState.RUNNING state.

async def _handle_suspending(self): ¶

Handle user-specific logic during the SkillState.SUSPENDING state.

After this function is returns, the skill will automatically advance to the SkillState.SUSPENDED state.

async def _suspend_point(self): ¶

Blocks if current_state is SkillState.SUSPENDING and allow advance to SkillState.SUSPENDED.

Call from _handle_running at any point where it's safe to pause.

@uamethod
@override

async def _ua_halt(self, _parent: NodeId, session: SessionProtocol) -> StatusCode: ¶

Undocumented

@uamethod
@override

async def _ua_reset(self, _parent: NodeId, session: SessionProtocol) -> StatusCode: ¶

Undocumented

@uamethod
@override

async def _ua_start(self, _parent: NodeId, session: SessionProtocol) -> StatusCode: ¶

Undocumented

@uamethod
@override

async def _ua_suspend(self, _parent: NodeId, session: SessionProtocol) -> StatusCode: ¶

Undocumented

async def _wrap_handle_method(self, *, handle_method: Callable[[], Awaitable], success_trigger: TRIGGER | None, fail_trigger: TRIGGER = 'halt', cancel_others: bool = False): ¶

Undocumented

__tasks: dict[str, Task] = ¶

Keeps track of tasks related to skill execution and their name. These tasks will get canceled and emptied before handle_halting() is called.

_machine: UaFiniteStateMachine[SkillState] = ¶

Skill Finite State Machine.

_paused_event: asyncio.Event = ¶

Undocumented

_resume_event: asyncio.Event = ¶

Undocumented

server.mixins.skill_logic_mixins._SkillLogicMixin

`server.mixins.skill_logic_mixins._SkillLogicMixin`