3. Online状态机模型设计：定义Online/Offline/Busy/Idle等状态，设计状态转换图与状态机类

在联合调试工程案例中，Online状态机是连接设备在线状态与日志分析引擎的核心枢纽。本节将定义四种基础状态（Online / Offline / Busy / Idle），设计完整的状态转换图，并实现一个可复用的状态机类。

3.1 状态定义与语义

每个状态代表设备或服务在某一时刻的可通信性与负载能力。下表列出四种状态的语义、触发条件及典型日志特征：

3.2 状态转换图

下图展示了四种状态之间的合法转换路径，箭头上的标签表示触发事件：

3.3 状态机类设计

基于上述转换图，我们使用 Python 实现一个泛化状态机类，支持事件驱动、状态回调与日志记录。核心设计原则：

• 状态注册：通过字典维护状态与合法转换事件
• 事件驱动：外部调用 trigger(event) 触发转换
• 回调机制：进入/离开状态时自动调用注册的回调函数
• 日志埋点：每次转换自动输出结构化日志

from enum import Enum
from typing import Callable, Dict, List, Optional

class State(Enum):
    OFFLINE = "Offline"
    ONLINE  = "Online"
    IDLE    = "Idle"
    BUSY    = "Busy"

class Event(Enum):
    CONNECT    = "connect"       # 连接建立
    DISCONNECT = "disconnect"    # 连接断开
    HEARTBEAT  = "heartbeat"     # 心跳恢复
    TIMEOUT    = "timeout"       # 心跳超时
    LOAD_HIGH  = "load_high"     # 负载升高
    LOAD_LOW   = "load_low"      # 负载降低
    TASK_START = "task_start"    # 任务开始
    TASK_DONE  = "task_done"     # 任务完成

class OnlineStateMachine:
    """Online 状态机类，支持事件驱动与回调"""

    def __init__(self, initial_state: State = State.OFFLINE):
        self._state = initial_state
        self._transitions: Dict[State, Dict[Event, State]] = {
            State.OFFLINE: {
                Event.CONNECT:   State.ONLINE,
                Event.HEARTBEAT: State.ONLINE,
            },
            State.ONLINE: {
                Event.DISCONNECT: State.OFFLINE,
                Event.TIMEOUT:    State.OFFLINE,
                Event.LOAD_HIGH:  State.BUSY,
                Event.LOAD_LOW:   State.IDLE,
            },
            State.IDLE: {
                Event.DISCONNECT: State.OFFLINE,
                Event.TIMEOUT:    State.OFFLINE,
                Event.TASK_START: State.BUSY,
                Event.LOAD_HIGH:  State.BUSY,
            },
            State.BUSY: {
                Event.DISCONNECT: State.OFFLINE,
                Event.TIMEOUT:    State.OFFLINE,
                Event.TASK_DONE:  State.IDLE,
                Event.LOAD_LOW:   State.IDLE,
            },
        }
        self._callbacks: Dict[str, List[Callable]] = {
            "on_enter": [],
            "on_exit":  [],
            "on_error": [],
        }

    @property
    def current_state(self) -> State:
        return self._state

    def register_callback(self, hook: str, func: Callable) -> None:
        """注册回调函数，hook 可选 'on_enter' / 'on_exit' / 'on_error'"""
        if hook in self._callbacks:
            self._callbacks[hook].append(func)

    def trigger(self, event: Event) -> bool:
        """触发事件，返回是否成功转换"""
        if self._state not in self._transitions:
            self._invoke_callbacks("on_error", f"未知状态: {self._state}")
            return False

        if event not in self._transitions[self._state]:
            self._invoke_callbacks("on_error",
                f"非法转换: {self._state.value} -> {event.value}")
            return False

        new_state = self._transitions[self._state][event]
        old_state = self._state

        # 离开旧状态
        self._invoke_callbacks("on_exit", old_state, event)

        # 更新状态
        self._state = new_state

        # 进入新状态
        self._invoke_callbacks("on_enter", new_state, event)

        # 输出结构化日志
        print(f"[StateMachine] {old_state.value} --({event.value})--> {new_state.value}")
        return True

    def _invoke_callbacks(self, hook: str, *args) -> None:
        for cb in self._callbacks.get(hook, []):
            cb(*args)

# 使用示例
if __name__ == "__main__":
    sm = OnlineStateMachine(State.OFFLINE)

    def log_enter(state, event):
        print(f"  >> 进入状态: {state.value} (事件: {event.value})")

    def log_exit(state, event):
        print(f"  << 离开状态: {state.value} (事件: {event.value})")

    sm.register_callback("on_enter", log_enter)
    sm.register_callback("on_exit",  log_exit)

    # 模拟一次完整生命周期
    sm.trigger(Event.CONNECT)      # Offline -> Online
    sm.trigger(Event.LOAD_LOW)     # Online -> Idle
    sm.trigger(Event.TASK_START)   # Idle   -> Busy
    sm.trigger(Event.TASK_DONE)    # Busy   -> Idle
    sm.trigger(Event.TIMEOUT)      # Idle   -> Offline

3.4 状态转换合法性矩阵

为便于工程审查与测试，将全部合法转换整理为矩阵形式。行代表当前状态，列代表触发事件，单元格内容为目标状态或“❌ 非法”：

3.5 状态机与日志分析引擎的对接接口

状态机类需要暴露以下接口供日志分析模块调用：

• get_state() → 返回当前状态枚举值
• get_state_history() → 返回最近 N 次状态转换记录（含时间戳）
• on_state_change(callback) → 注册状态变更回调，用于实时推送日志
• validate_transition(event) → 预检某事件是否合法，不实际转换