Конечный автомат (Finite State Machine) — Формальные модели поведения автономных систем

// Формальное определение на языке Rust с проверкой типов
enum SafetyState {
    Normal,
    Warning { distance: f32 },
    EmergencyStop,
    Fault { code: u32 },
}

enum SafetyEvent {
    ObstacleDetected(f32),
    ButtonPressed,
    SystemCheckPassed,
    ErrorDetected(u32),
}

impl SafetyFSM {
    fn transition(state: SafetyState, event: SafetyEvent) -> SafetyState {
        match (state, event) {
            (SafetyState::Normal, SafetyEvent::ObstacleDetected(d)) if d < 0.1 => 
                SafetyState::EmergencyStop,
            (SafetyState::Normal, SafetyEvent::ObstacleDetected(d)) if d < 0.5 => 
                SafetyState::Warning { distance: d },
            (SafetyState::Warning { .. }, SafetyEvent::ObstacleDetected(d)) if d >= 0.5 => 
                SafetyState::Normal,
            // ... другие переходы
            _ => state, // Защита от неопределённых переходов
        }
    }
}

class HierarchicalFSM {
    struct State {
        string name;
        function<void()> entry_action;
        function<void()> exit_action;
        function<void()> do_action;
        map<Event, pair<State*, Condition>> transitions;
        
        // Иерархия
        State* parent = nullptr;
        vector<State*> children;
        State* initial_child = nullptr;
        
        // История для возврата
        State* history_state = nullptr;
    };
    
    State* current_state;
    stack<State*> state_stack;
    
    void process_event(Event event) {
        // Поиск обработчика в иерархии снизу вверх
        State* handler = find_handler(current_state, event);
        if (handler) {
            execute_transition(handler, event);
        }
    }
    
    State* find_handler(State* state, Event event) {
        while (state) {
            if (state->transitions.contains(event)) {
                auto& [next_state, condition] = state->transitions[event];
                if (condition()) return state;
            }
            state = state->parent; // Поднимаемся вверх по иерархии
        }
        return nullptr;
    }
};

class TimedAutomaton:
    def __init__(self):
        self.clocks = {'t_action': 0, 't_safety': 0}
        self.invariants = {
            'EXECUTING': 't_action <= 5.0',  # Максимум 5 секунд на действие
            'WAITING': 't_safety <= 2.0',    # Ожидание подтверждения 2 секунды
        }
        self.guards = {
            ('EXECUTING', 'TIMEOUT'): 't_action >= 5.0',
            ('WAITING', 'RESPONSE_MISSING'): 't_safety >= 2.0',
        }
    
    def transition(self, from_state, event):
        guard = self.guards.get((from_state, event))
        if guard and self.evaluate_guard(guard):
            # Сброс часов при переходе
            if event == 'ACTION_COMPLETE':
                self.clocks['t_action'] = 0
            return self.get_target_state(from_state, event)
        return from_state

class ProbabilisticFSM:
    def __init__(self):
        self.transition_matrix = {
            'EXPLORING': {
                'object_detected': [
                    ('INSPECTING', 0.7),   # С вероятностью 70% исследовать
                    ('IGNORING', 0.2),     # С вероятностью 20% проигнорировать
                    ('AVOIDING', 0.1),     # С вероятностью 10% избегать
                ]
            }
        }
        
    def transition(self, state, event):
        if state in self.transition_matrix and event in self.transition_matrix[state]:
            transitions = self.transition_matrix[state][event]
            # Выбор на основе вероятностей
            choices, probabilities = zip(*transitions)
            next_state = np.random.choice(choices, p=probabilities)
            return next_state
        return state

template<typename State, typename Action, typename Model>
class StateActionFSM {
    struct Transition {
        State from;
        Action trigger;
        function<bool(const Model&)> guard;
        function<void(Model&)> action;
        State to;
    };
    
    State current_state;
    Model world_model;
    vector<Transition> transitions;
    
    void process_action(Action action) {
        // Поиск подходящего перехода
        for (const auto& trans : transitions) {
            if (trans.from == current_state && 
                trans.trigger == action &&
                trans.guard(world_model)) {
                
                // Выполнение перехода
                trans.action(world_model);
                current_state = trans.to;
                log_transition(trans);
                return;
            }
        }
        // Обработка неопределённого перехода
        handle_undefined_transition(current_state, action);
    }
};

Основной FSM:          Навигационный FSM:       Манипулятор FSM:
[IDLE]                [PLANNING]              [STOWED]
   |                      |                       |
   v                      v                       v
[NAVIGATING] <------> [EXECUTING]           [GRASPING]
   |                      |                       |
   v                      v                       v
[WORKING]            [REPLANNING]            [PLACING]

class OrthogonalRegionsFSM:
    def __init__(self):
        self.regions = {
            'mission': MissionFSM(),
            'navigation': NavigationFSM(),
            'manipulation': ManipulationFSM(),
        }
        self.coordination_rules = [
            # Правила координации между регионами
            lambda: (self.regions['navigation'].state == 'BLOCKED' and
                     self.regions['mission'].state == 'EXECUTING') ->
                self.regions['mission'].trigger('NAVIGATION_FAILED')
        ]
    
    def process_event(self, event):
        # Распространение события во все регионы
        for region_name, fsm in self.regions.items():
            fsm.process_event(event)
        
        # Применение правил координации
        for rule in self.coordination_rules:
            rule.apply_if_condition_met()

class LearningFSM:
    def __init__(self):
        self.transitions = {}  # Q × A → Q
        self.q_values = defaultdict(lambda: defaultdict(float))  # Q × A → R
        self.learning_rate = 0.1
        self.discount_factor = 0.9
    
    def choose_action(self, state, epsilon=0.1):
        # ε-жадная стратегия
        if random.random() < epsilon:
            return random.choice(self.available_actions(state))
        else:
            actions = self.available_actions(state)
            return max(actions, key=lambda a: self.q_values[state][a])
    
    def learn(self, state, action, reward, next_state):
        # Обновление Q-значений (упрощённый Q-learning)
        best_next_action = max(self.available_actions(next_state), 
                               key=lambda a: self.q_values[next_state][a])
        td_target = reward + self.discount_factor * self.q_values[next_state][best_next_action]
        td_error = td_target - self.q_values[state][action]
        self.q_values[state][action] += self.learning_rate * td_error
        
        # Адаптация переходов при достижении порога уверенности
        if self.confidence(state, action) > 0.95:
            self.transitions[(state, action)] = next_state

// Формулы для проверки свойств безопасности робота
vector<LTLFormula> safety_properties = {
    // Всегда: если EmergencyStop, то не двигаться
    LTL::Always(LTL::Implies(
        state == "EMERGENCY_STOP", 
        motor_speed == 0
    )),
    
    // В конце концов всегда можно вернуться в Normal
    LTL::Eventually(LTL::Always(
        LTL::Possible(state == "NORMAL")
    )),
    
    // Никогда не находиться одновременно в двух конфликтующих состояниях
    LTL::Always(LTL::Not(
        state == "MOVING" && state == "CHARGING"
    ))
};

// Проверка с помощью model checker'а
ModelChecker checker;
for (const auto& property : safety_properties) {
    bool holds = checker.verify(robot_fsm, property);
    if (!holds) {
        cout << "Нарушено свойство: " << property << endl;
        cout << "Контрпример: " << checker.get_counterexample() << endl;
    }
}

def analyze_reachability(fsm):
    visited = set()
    queue = [fsm.initial_state]
    deadlock_states = []
    
    while queue:
        state = queue.pop(0)
        if state in visited:
            continue
        visited.add(state)
        
        # Проверка на тупик
        if not fsm.get_enabled_events(state):
            deadlock_states.append(state)
        
        # Добавление достижимых состояний
        for event in fsm.get_possible_events(state):
            next_state = fsm.get_transition(state, event)
            if next_state not in visited:
                queue.append(next_state)
    
    return {
        'reachable_states': visited,
        'deadlock_states': deadlock_states,
        'coverage': len(visited) / len(fsm.all_states)
    }

def generate_test_paths(fsm):
    # Построение графа автомата
    graph = nx.DiGraph()
    for from_state, transitions in fsm.transitions.items():
        for event, to_state in transitions.items():
            graph.add_edge(from_state, to_state, event=event)
    
    # Нахождение эйлерова пути или цикла
    if nx.is_eulerian(graph):
        path = list(nx.eulerian_circuit(graph))
    else:
        # Добавление минимального числа рёбер для эйлерова графа
        graph = nx.eulerize(graph)
        path = list(nx.eulerian_circuit(graph))
    
    # Преобразование в последовательность тестов
    test_cases = []
    for (from_state, to_state, data) in path:
        test_cases.append({
            'initial_state': from_state,
            'event': data['event'],
            'expected_state': to_state
        })
    
    return test_cases

Требования безопасности (LTL):
1. □(charging → ¬moving)  // Во время зарядки не двигаться
2. □(emergency → ○(stopped))  // При аварии немедленно остановиться
3. ◇□(battery_low → charging)  // При низком заряде в конце концов зарядиться

Требования живости (Liveness):
1. ◇(order_received → ◇delivered)  // Заказ в конце концов будет доставлен
2. ◇(obstacle → ◇¬obstacle)  // Препятствие в конце концов будет преодолено

class DeliveryRobotFSM:
    def __init__(self):
        # Верхний уровень: миссия
        self.mission = HierarchicalFSM(
            initial_state='WAITING_FOR_ORDER',
            states={
                'WAITING_FOR_ORDER': {
                    'order_received': ('PROCESSING_ORDER', self.validate_order),
                },
                'PROCESSING_ORDER': {
                    'order_valid': ('EXECUTING_DELIVERY', self.plan_route),
                    'order_invalid': ('WAITING_FOR_ORDER', self.notify_error),
                },
                'EXECUTING_DELIVERY': {
                    'at_destination': ('COMPLETING_DELIVERY', self.prepare_delivery),
                    'delivery_failed': ('HANDLING_FAILURE', self.analyze_failure),
                },
                'COMPLETING_DELIVERY': {
                    'delivery_complete': ('WAITING_FOR_ORDER', self.confirm_delivery),
                },
                'HANDLING_FAILURE': {
                    'failure_resolved': ('EXECUTING_DELIVERY', self.replan),
                    'failure_permanent': ('WAITING_FOR_ORDER', self.abort_delivery),
                },
            }
        )
        
        # Средний уровень: навигация
        self.navigation = ParallelFSM([
            LocalNavigationFSM(),   # Избегание препятствий
            GlobalNavigationFSM(),  # Следование маршруту
            LocalizationFSM(),      # Определение позиции
        ])
        
        # Нижний уровень: безопасность
        self.safety = SafetyFSM(
            sensors=['lidar', 'bumper', 'current_sensor'],
            emergency_actions=['cut_power', 'activate_brakes']
        )

MODULE main
VAR
    state: {WAITING, EXECUTING, CHARGING, EMERGENCY};
    battery: 0..100;
    obstacle: boolean;

ASSIGN
    init(state) := WAITING;
    next(state) := case
        state = WAITING & battery > 20 & !obstacle: EXECUTING;
        state = EXECUTING & battery < 15: CHARGING;
        state = EXECUTING & obstacle: EMERGENCY;
        state = CHARGING & battery > 90: WAITING;
        state = EMERGENCY & !obstacle: WAITING;
        1: state;  -- по умолчанию остаёмся в текущем состоянии
    esac;

-- Проверка свойств безопасности
SPEC AG (state = EMERGENCY -> AX state != EXECUTING)  -- После аварии не двигаться
SPEC AG (battery < 5 -> AF state = CHARGING)         -- Низкий заряд ведёт к зарядке
SPEC !EF (state = EXECUTING & state = CHARGING)      -- Невозможно одновременно выполнять и заряжаться

# Автоматическая генерация кода из спецификации
code_generator = FSMCodeGenerator(
    target_language='cpp',
    template='embedded_rtos',
    optimizations=['minimize_states', 'optimize_transitions']
)

generated_code = code_generator.generate(self.delivery_robot_fsm)

# Генерация тестов с покрытием всех переходов
test_generator = TestGenerator(coverage_criteria='transition_coverage')
test_suite = test_generator.generate(self.delivery_robot_fsm)

# Статическая проверка свойств
verification_result = model_checker.verify(
    model=self.delivery_robot_fsm,
    properties=safety_properties + liveness_properties
)

class LifecycleFSMNode : public rclcpp_lifecycle::LifecycleNode {
    enum State {UNCONFIGURED, INACTIVE, ACTIVE, FINALIZED};
    
    CallbackReturn on_configure(const State &) override {
        // Переход в INACTIVE
        return CallbackReturn::SUCCESS;
    }
    
    CallbackReturn on_activate(const State &) override {
        // Переход в ACTIVE
        publisher_->on_activate();
        return CallbackReturn::SUCCESS;
    }
    
    CallbackReturn on_deactivate(const State &) override {
        // Возврат в INACTIVE
        publisher_->on_deactivate();
        return CallbackReturn::SUCCESS;
    }
};

class HybridFSMRL:
    def __init__(self):
        self.fsm = DeterministicFSM()  # Для гарантированной безопасности
        self.rl_agent = PPOAgent()     # Для адаптивного поведения
        
    def decide_action(self, state):
        # FSM определяет допустимые действия
        allowed_actions = self.fsm.get_allowed_actions(state)
        
        # RL агент выбирает оптимальное действие среди допустимых
        if allowed_actions:
            q_values = self.rl_agent.predict(state)
            # Фильтрация по допустимым действиям
            allowed_q_values = {a: q for a, q in q_values.items() 
                                if a in allowed_actions}
            if allowed_q_values:
                return max(allowed_q_values, key=allowed_q_values.get)
        
        # Резервное действие от FSM
        return self.fsm.get_default_action(state)

class DistributedFSM:
    def __init__(self, robot_id, neighbors):
        self.local_fsm = LocalFSM()
        self.neighbors = neighbors
        self.consensus_algorithm = PaxosFSM()
        
    def global_transition(self, global_event):
        # Достижение консенсуса о глобальном переходе
        proposal = (self.robot_id, global_event, time.time())
        agreed_event = self.consensus_algorithm.propose(proposal)
        
        if agreed_event:
            # Все роботы выполняют синхронизированный переход
            self.local_fsm.process_event(agreed_event)
            self.broadcast_transition(agreed_event)