lockable_state_machine.py

Go to the documentation of this file.

#!/usr/bin/env python
from flexbe_core.core.ros_state_machine import RosStateMachine


class LockableStateMachine(RosStateMachine):
    """
    A state machine that can be locked.
    When locked, no transition can be done regardless of the resulting outcome.
    However, if any outcome would be triggered, the outcome will be stored
    and the state won't be executed anymore until it is unlocked and the stored outcome is set.
    """
    path_for_switch = None

    def __init__(self, *args, **kwargs):
        super(LockableStateMachine, self).__init__(*args, **kwargs)
        self._locked = False

    def get_deep_state(self):
        """
        Looks for the current state (traversing all state machines down to the real state).

        @return: The current state (not state machine)
        """
        container = self
        while isinstance(container._current_state, LockableStateMachine):
            container = container._current_state
        return container._current_state

    def _is_internal_transition(self, transition_target):
        return transition_target in self._labels

    def transition_allowed(self, state, outcome):
        if outcome is None or outcome == 'None':
            return True
        transition_target = self._transitions[state].get(outcome)
        return (self._is_internal_transition(transition_target) or
                (not self._locked and (self.parent is None or
                                       self.parent.transition_allowed(self.name, transition_target))))

    # for switching

    def execute(self, userdata):
        if (LockableStateMachine.path_for_switch is not None
                and LockableStateMachine.path_for_switch.startswith(self.path)):
            path_segments = LockableStateMachine.path_for_switch.replace(self.path, "", 1).split("/")
            wanted_state = path_segments[1]
            self._current_state = self._labels[wanted_state]
            if len(path_segments) <= 2:
                LockableStateMachine.path_for_switch = None
        return super(LockableStateMachine, self).execute(userdata)

    def replace_userdata(self, userdata):
        self._userdata = userdata

    def replace_state(self, state):
        old_state = self._labels[state.name]
        state._parent = old_state._parent
        self._states[self._states.index(old_state)] = state
        self._labels[state.name] = state

    def remove_state(self, state):
        del self._labels[state.name]
        self._states.remove(state)

    # for locking

    def lock(self, path):
        if path == self.path:
            self._locked = True
            return True
        elif self._parent is not None:
            return self._parent.lock(path)
        else:
            return False

    def unlock(self, path):
        if path == self.path:
            self._locked = False
            return True
        elif self._parent is not None:
            return self._parent.unlock(path)
        else:
            return False

    def is_locked(self):
        return self._locked

    def is_locked_inside(self):
        if self._locked:
            return True
        for state in self._states:
            result = False
            if isinstance(state, LockableStateMachine):
                result = state.is_locked_inside()
            else:
                result = state.is_locked()
            if result is True:
                return True
        return False

    def get_locked_state(self):
        if self._locked:
            return self
        for state in self._states:
            if state.is_locked():
                return state
            elif isinstance(state, LockableStateMachine):
                return state.get_locked_state()
        return None