Python脚本状态模式如何管理状态机

wen 实用脚本 2026-07-19 3

本文目录导读：

Python脚本状态模式如何管理状态机

基础状态模式实现
更完善的状态机实现
带数据和条件的复杂状态机
使用第三方库简化状态机
状态模式的最佳实践
主要特点和优势
应用场景

我来详细解释Python中如何使用状态模式管理状态机,包括几种常见实现方式：

基础状态模式实现

from abc import ABC, abstractmethod
class State(ABC):
    """抽象状态类"""
    @abstractmethod
    def handle(self, context):
        pass
class ConcreteStateA(State):
    def handle(self, context):
        print("当前状态: A")
        print("执行A状态的逻辑...")
        # 状态转换到B
        context.state = ConcreteStateB()
class ConcreteStateB(State):
    def handle(self, context):
        print("当前状态: B")
        print("执行B状态的逻辑...")
        # 状态转换到A
        context.state = ConcreteStateA()
class Context:
    """上下文环境"""
    def __init__(self):
        self.state = ConcreteStateA()
    def request(self):
        self.state.handle(self)
# 使用示例
context = Context()
context.request()  # 状态A -> B
context.request()  # 状态B -> A

更完善的状态机实现

from enum import Enum, auto
from typing import Dict, Callable, Any
class State(Enum):
    IDLE = auto()
    RUNNING = auto()
    PAUSED = auto()
    STOPPED = auto()
    ERROR = auto()
class Event(Enum):
    START = auto()
    PAUSE = auto()
    RESUME = auto()
    STOP = auto()
    ERROR = auto()
class StateMachine:
    def __init__(self):
        self.current_state = State.IDLE
        self.state_history = []
        # 定义状态转换表: {当前状态: {事件: 下一个状态}}
        self.transitions = {
            State.IDLE: {
                Event.START: State.RUNNING,
                Event.ERROR: State.ERROR
            },
            State.RUNNING: {
                Event.PAUSE: State.PAUSED,
                Event.STOP: State.STOPPED,
                Event.ERROR: State.ERROR
            },
            State.PAUSED: {
                Event.RESUME: State.RUNNING,
                Event.STOP: State.STOPPED,
                Event.ERROR: State.ERROR
            },
            State.STOPPED: {
                Event.START: State.RUNNING
            },
            State.ERROR: {
                Event.START: State.RUNNING  # 重置
            }
        }
        # 进入每个状态时的回调函数
        self.entry_actions: Dict[State, Callable] = {
            State.IDLE: self.on_idle,
            State.RUNNING: self.on_running,
            State.PAUSED: self.on_paused,
            State.STOPPED: self.on_stopped,
            State.ERROR: self.on_error
        }
    def on_idle(self):
        print("系统已就绪")
    def on_running(self):
        print("系统正在运行")
    def on_paused(self):
        print("系统已暂停")
    def on_stopped(self):
        print("系统已停止")
    def on_error(self):
        print("系统出错")
    def handle_event(self, event: Event) -> bool:
        """处理事件并返回是否成功转换"""
        current_transitions = self.transitions.get(self.current_state, {})
        if event not in current_transitions:
            print(f"从状态 {self.current_state.name} 无法处理事件 {event.name}")
            return False
        next_state = current_transitions[event]
        print(f"{self.current_state.name} -> {next_state.name}")
        # 记录历史
        self.state_history.append((self.current_state, next_state))
        # 更新状态并执行进入动作
        self.current_state = next_state
        # 调用进入状态的回调
        entry_action = self.entry_actions.get(next_state)
        if entry_action:
            entry_action()
        return True
    def get_state_history(self) -> list:
        """获取状态转换历史"""
        return self.state_history
# 使用示例
sm = StateMachine()
# 执行一系列状态转换
sm.handle_event(Event.START)   # IDLE -> RUNNING
sm.handle_event(Event.PAUSE)   # RUNNING -> PAUSED
sm.handle_event(Event.RESUME)  # PAUSED -> RUNNING
sm.handle_event(Event.STOP)    # RUNNING -> STOPPED
sm.handle_event(Event.START)   # STOPPED -> RUNNING
sm.handle_event(Event.ERROR)   # RUNNING -> ERROR

带数据和条件的复杂状态机

from dataclasses import dataclass
from typing import Optional, List, Dict, Callable
from datetime import datetime
@dataclass
class Order:
    """订单数据"""
    order_id: str
    items: List[str]
    total_amount: float
    customer_email: str
    created_at: datetime
class OrderState(Enum):
    CREATED = auto()
    CONFIRMED = auto()
    PAID = auto()
    PROCESSING = auto()
    SHIPPED = auto()
    DELIVERED = auto()
    CANCELLED = auto()
    REFUNDED = auto()
class OrderStateMachine:
    def __init__(self, order: Order):
        self.order = order
        self.current_state = OrderState.CREATED
        self.state_data: Dict[str, Any] = {
            'last_update': datetime.now()
        }
        # 定义状态转换规则
        self.transitions = {
            OrderState.CREATED: {
                'confirm': (OrderState.CONFIRMED, self.validate_order),
                'cancel': (OrderState.CANCELLED, self.cancel_order)
            },
            OrderState.CONFIRMED: {
                'pay': (OrderState.PAID, self.process_payment),
                'cancel': (OrderState.CANCELLED, self.cancel_order)
            },
            OrderState.PAID: {
                'process': (OrderState.PROCESSING, self.start_processing),
                'cancel': (OrderState.CANCELLED, self.request_refund)
            },
            OrderState.PROCESSING: {
                'ship': (OrderState.SHIPPED, self.ship_order)
            },
            OrderState.SHIPPED: {
                'deliver': (OrderState.DELIVERED, self.confirm_delivery)
            },
            OrderState.DELIVERED: {
                'refund': (OrderState.REFUNDED, self.process_refund)
            }
        }
    def validate_order(self) -> bool:
        """验证订单"""
        if not self.order.customer_email:
            raise ValueError("客户邮箱不能为空")
        if self.order.total_amount <= 0:
            raise ValueError("订单金额必须大于0")
        print(f"订单 {self.order.order_id} 验证通过")
        return True
    def process_payment(self) -> bool:
        """处理支付"""
        print(f"处理订单 {self.order.order_id} 的支付：{self.order.total_amount}")
        self.state_data['payment_time'] = datetime.now()
        return True
    def start_processing(self) -> bool:
        """开始处理订单"""
        print(f"开始处理订单 {self.order.order_id}")
        return True
    def ship_order(self) -> bool:
        """发货"""
        print(f"订单 {self.order.order_id} 已发货")
        self.state_data['shipping_time'] = datetime.now()
        return True
    def confirm_delivery(self) -> bool:
        """确认收货"""
        print(f"订单 {self.order.order_id} 已送达")
        self.state_data['delivery_time'] = datetime.now()
        return True
    def cancel_order(self) -> bool:
        """取消订单"""
        print(f"订单 {self.order.order_id} 已取消")
        return True
    def request_refund(self) -> bool:
        """请求退款"""
        print(f"请求处理订单 {self.order.order_id} 的退款")
        return True
    def process_refund(self) -> bool:
        """处理退款"""
        print(f"订单 {self.order.order_id} 已完成退款")
        self.state_data['refund_time'] = datetime.now()
        return True
    def handle_action(self, action: str) -> bool:
        """处理动作"""
        current_transitions = self.transitions.get(self.current_state, {})
        if action not in current_transitions:
            print(f"订单状态 {self.current_state.name} 不能执行 {action} 操作")
            return False
        next_state, condition_func = current_transitions[action]
        try:
            # 执行条件检查
            if condition_func():
                old_state = self.current_state
                self.current_state = next_state
                self.state_data['last_update'] = datetime.now()
                print(f"订单 {self.order.order_id}: {old_state.name} -> {next_state.name}")
                return True
        except Exception as e:
            print(f"状态转换失败: {e}")
            return False
        return False
    def get_current_state(self) -> OrderState:
        return self.current_state
    def get_state_info(self) -> Dict:
        """获取状态信息"""
        return {
            'order_id': self.order.order_id,
            'current_state': self.current_state.name,
            'last_update': self.state_data['last_update'],
            **{k: v for k, v in self.state_data.items() if k != 'last_update'}
        }
# 使用示例
order = Order(
    order_id="ORD-001",
    items=["笔记本电脑", "鼠标"],
    total_amount=5999.99,
    customer_email="customer@example.com",
    created_at=datetime.now()
)
order_sm = OrderStateMachine(order)
# 执行订单流转
order_sm.handle_action('confirm')   # CREATED -> CONFIRMED
order_sm.handle_action('pay')       # CONFIRMED -> PAID
order_sm.handle_action('process')   # PAID -> PROCESSING
order_sm.handle_action('ship')      # PROCESSING -> SHIPPED
order_sm.handle_action('deliver')   # SHIPPED -> DELIVERED
# 查看状态信息
print("\n订单状态信息:")
info = order_sm.get_state_info()
for k, v in info.items():
    print(f"  {k}: {v}")

使用第三方库简化状态机

# 安装: pip install transitions
from transitions import Machine
from transitions.extensions import GraphMachine
from enum import Enum
class Status(Enum):
    STOPPED = 'stopped'
    RUNNING = 'running'
    PAUSED = 'paused'
class MachineWithGraph:
    def __init__(self):
        # 状态列表
        states = ['stopped', 'running', 'paused']
        # 转换规则
        transitions = [
            {'trigger': 'start', 'source': 'stopped', 'dest': 'running'},
            {'trigger': 'stop', 'source': ['running', 'paused'], 'dest': 'stopped'},
            {'trigger': 'pause', 'source': 'running', 'dest': 'paused'},
            {'trigger': 'resume', 'source': 'paused', 'dest': 'running'},
            {'trigger': 'error', 'source': '*', 'dest': 'stopped'}
        ]
        # 初始化状态机（带图形功能）
        self.machine = GraphMachine(
            model=self,
            states=states,
            transitions=transitions,
            initial='stopped',
            auto_transitions=False,
            show_conditions=False
        )
        # 设置状态进入回调
        self.machine.add_transition(
            'restart', 'stopped', 'running',
            after='on_restart'
        )
    def on_enter_running(self):
        print("进入运行状态")
    def on_enter_paused(self):
        print("进入暂停状态")
    def on_enter_stopped(self):
        print("进入停止状态")
    def on_restart(self):
        print("执行重启操作")
# 使用示例
model = MachineWithGraph()
# 执行状态转换
model.start()      # stopped -> running
model.pause()      # running -> paused
model.resume()     # paused -> running
model.stop()       # running -> stopped
model.error()      # stopped -> stopped (已经处于stopped)

状态模式的最佳实践

from abc import ABC, abstractmethod
from typing import Dict, Any, Callable
import logging
# 配置日志
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class State(ABC):
    """状态基类"""
    @abstractmethod
    def handle_event(self, context, event):
        pass
    @abstractmethod
    def get_name(self) -> str:
        pass
class StateManager:
    """状态管理器"""
    def __init__(self, initial_state: State):
        self._state = initial_state
        self._history = []
        self._observers = []
        self._data: Dict[str, Any] = {}
    @property
    def state(self) -> State:
        return self._state
    @state.setter
    def state(self, new_state: State):
        old_state = self._state
        self._state = new_state
        self._history.append({
            'from': old_state.get_name(),
            'to': new_state.get_name(),
            'timestamp': __import__('time').time()
        })
        self._notify_observers(old_state, new_state)
        logger.info(f"状态转换: {old_state.get_name()} -> {new_state.get_name()}")
    def add_observer(self, observer: Callable):
        """添加观察者"""
        self._observers.append(observer)
    def _notify_observers(self, old_state, new_state):
        """通知观察者状态变化"""
        for observer in self._observers:
            observer(old_state, new_state)
    def handle_event(self, event):
        """处理事件"""
        self._state.handle_event(self, event)
    def set_data(self, key: str, value: Any):
        """设置共享数据"""
        self._data[key] = value
    def get_data(self, key: str) -> Any:
        """获取共享数据"""
        return self._data.get(key)
    def get_history(self) -> list:
        """获取状态历史"""
        return self._history
# 使用示例
class IdleState(State):
    def handle_event(self, context, event):
        if event == "start":
            context.state = RunningState()
        elif event == "stop":
            print("已经停止")
        else:
            print(f"无法处理事件: {event}")
    def get_name(self) -> str:
        return "IDLE"
class RunningState(State):
    def handle_event(self, context, event):
        if event == "pause":
            context.state = PausedState()
        elif event == "stop":
            context.state = IdleState()
        else:
            print(f"无法处理事件: {event}")
    def get_name(self) -> str:
        return "RUNNING"
class PausedState(State):
    def handle_event(self, context, event):
        if event == "resume":
            context.state = RunningState()
        elif event == "stop":
            context.state = IdleState()
        else:
            print(f"无法处理事件: {event}")
    def get_name(self) -> str:
        return "PAUSED"
# 测试状态管理器
def status_change_callback(old_state, new_state):
    print(f"状态变化回调: {old_state.get_name()} -> {new_state.get_name()}")
# 使用
manager = StateManager(IdleState())
manager.add_observer(status_change_callback)
# 测试状态转换
manager.handle_event("start")   # IDLE -> RUNNING
manager.handle_event("pause")   # RUNNING -> PAUSED
manager.handle_event("resume")  # PAUSED -> RUNNING
manager.handle_event("stop")    # RUNNING -> IDLE
# 查看历史
print("\n状态历史:")
for h in manager.get_history():
    print(f"  {h['from']} -> {h['to']} at {h['timestamp']}")

主要特点和优势

封装性: 每个状态的行为都封装在自己的类中
开闭原则: 添加新状态不需要修改现有状态类
可扩展性: 易于添加新的事件和状态
可维护性: 状态转换逻辑集中管理
可测试性: 各个状态可以独立测试

应用场景

工作流管理系统
订单处理系统
游戏角色状态管理
用户界面导航
协议实现（如TCP状态机）
设备驱动状态管理

选择哪种实现方式取决于具体需求：简单场景用基本模式,复杂场景用完整的状态机实现或使用第三方库。