server.mixins.skill_logic_mixins._SkillLogicMixin

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
boolAlways True, else an exception is raised.
Raises
UaStatusCodeErrorIf 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
boolAlways True, else an exception is raised.
Raises
UaStatusCodeErrorIf 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
boolAlways True, else an exception is raised.
Raises
UaStatusCodeErrorIf 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. Behavior of continuous skills can be configured. By default, they go to SkillState.HALTING state, but if desired can stay in SkillState.RUNNING.

Note: This continues to be executed during SkillState.SUSPENDING and SkillState.SUSPENDED states. If your skill supports suspending, you have to check in what state the skill is!

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 =

Skill Finite State Machine.

_paused_event: asyncio.Event =

Undocumented

_resume_event: asyncio.Event =

Undocumented