Python 状态机实战教程
Python 写状态机的详细教程
Python 状态机实战教程
Python 状态机实战教程: 从零到工业级设计
1. 什么是状态机?
1.1. 核心定义
状态机 (Finite State Machine, FSM) 是一种数学模型, 用于描述一个对象在其生命周期内所经历的状态以及触发状态切换的事件.
- 状态 (State): 系统在某一时刻的形态 (如: 关机、开机、故障).
- 事件/触发器 (Event/Trigger): 引起状态改变的原因 (如: 按下电源键).
- 转换 (Transition): 从一个状态切换到另一个状态的过程.
- 动作 (Action): 在转换过程中执行的代码逻辑 (如: 开机前自检).
FSM = (States, Events, Transitions, Actions)
状态机 = 一个在任意时刻只处于一个“状态”的系统, 状态只能通过“合法的转换”改变.
1.2. 为什么状态机重要（而不是 if-else）
1. if-else 的本质问题
1
2
3
4
5
6
if state == "idle":
...
elif state == "calibrating":
...
elif state == "exposing":
...
问题不在“丑”, 而在 失控:
- 状态约束是隐式的
- 非法状态跳转不会被阻止
- 行为分散在代码各处
- 很难回答一个问题: “系统当前允许做什么?”
2. 状态机解决的不是“写法”，而是“约束”
状态机本质上是在做三件事:
- 显式建模系统阶段
- 收紧状态变化的可能性
- 把副作用绑定到状态边界
含义不是“调用一个函数”, 而是: “凡是进入 calibrating 状态，必须先完成这个动作”
1.3. 状态机在工程中的三种层次
1. 层次一: 流程 FSM (Workflow FSM)
最常见: ``idle → calibrating → exposing → processing → idle`
特点:
- 顺序明确
- 状态有限
- 行为可测
2. 层次二: 控制 FSM (Control FSM)
特点：
- 异常路径多
- 强制跳转 (panic / reset)
- 状态间并非线性
这类 FSM 的价值: 保证“无论系统在哪里, 故障都能被收敛”
3. 层次三: 协议 / 业务 FSM (Protocol FSM)
如：
- 网络管理
- 设备控制协议
- 登录 / 认证流程
特点：
- FSM 本身就是 “协议文档”
- 状态是 API 约束的一部分
1.4. 状态的工程化分类
1.4.1. 状态 ≠ 行为
状态是“阶段”，不是“动作”
❌ 错误示例:
1
2
3
connecting
sending
waiting
✅ 正确示例:
1
2
3
disconnected
connected
authenticated
1.4.2. 状态的生命周期钩子
| 钩子 | 语义 |
|---|---|
| on_enter | 状态初始化 |
| on_exit | 状态清理 |
| before | 转换前准备 |
| after | 转换后副作用 |
FSM 的核心价值之一: 防止作者写 add_transition 时忘记调用检查函数
1.5. 一个成熟 FSM 的判断标准
- 状态是有限且命名清晰的
- 非法操作会失败
- 故障路径被统一收敛
- 行为绑定在状态边界而不是 scattered
2. Python 状态机库: Transitions
在 Python 领域, transitions 是最流行、最轻量且功能最全的状态机库. 它采用对象注入的方式, 让你的类平滑地获得状态控制能力.
2.1. 安装方法:
1
pip install transitions -i https://mirrors.aliyun.com/pypi/simple/
2.2. 引用方法:
1
from transitions import Machine
3. 核心 API: Machine 函数详解
3.1 基础参数
在初始化 Machine 时, 以下参数定义了状态机的骨架:
| 参数 | 说明 | 作用流程 |
|---|---|---|
model |
状态机依附的对象 (通常设为 self) |
初始化时 |
states |
状态列表 (字符串或字典) | 定义阶段 |
initial |
初始状态名 | 启动瞬间 |
transitions |
转换规则列表 | 定义阶段 |
3.2 进阶参数 (高级用法)
当你需要工业级的严谨性时, 会用到这些:
| 参数 | 说明 | 作用流程 |
|---|---|---|
before_state_change |
全局前置回调 | 只要发生任何转换, 最先执行 |
after_state_change |
全局后置回调 | 转换成功后, 最后执行 |
auto_transitions |
自动生成 to_state() 方法 |
允许从任意状态强制跳转 |
send_event |
回调函数接收 event 对象 |
让回调函数能获取触发详情 (如源状态) |
**举例: **
Python
1
2
3
4
5
6
7
self.machine = Machine(
model=self,
states=['A', 'B'],
initial='A',
send_event=True, # 开启后, 回调函数写成 def func(self, event)
auto_transitions=False # 严谨模式: 禁止通过 to_B() 这种方式非法跳转; true: 可通过 to_B() 方法跳转
)
4. 如何设计状态机?
4.1 状态 vs 中间态: 如何区分?
- 状态 (State): 应该是“持久的”. 系统可以稳定停留在这一刻, 等待外部指令. 如:
idle(等用户按快门) 、error(停下来等维修). - 中间态 (Action/Hook): 应该是“瞬间的”. 它只是转换过程中的一个动作. 例如:
open_shutter(开快门) 不是一个状态, 而是一个before动作.
4.2 小技巧与注意点
- 防御式设计: 使用
conditions拦截非法请求 (如: SD卡满时不准拍照). - 职责单一: 状态名应该是形容词或被动语态 (
exposing), 而触发器应该是动词 (capture). - 不要在
on_enter里写耗时循环: 这会阻塞整个状态机的跳转. 如果是耗时任务, 应考虑异步. - 善用通配符
\*: 对于“硬件故障”这种任何时候都可能发生的事, 用source='*'统一指向error.
4.3 推荐学习资源
- Transitions 官方文档 (GitHub) —— 最权威的参考.
- Game Programming Patterns: State —— 深入理解状态模式.
- Refactoring Guru: State Pattern —— 图解设计模式.
5. 实战项目: 全自动天文摄影系统
5.1 状态机设计思路
设计一个复杂的逻辑系统时, 开发者通常遵循 “自顶向下” 的原则.
5.1.1. 业务逻辑拆解 (The Workflow)
设计之初, 将相机抽象为五个核心阶段：
- 静默期 (
idle)：省电、待机，这是所有逻辑的起点和终点。 - 准备期 (
calibrating)：相机最繁忙的自检阶段。它必须确认温度达标、对焦准确。它是拍摄前的“门禁”。 - 执行期 (
exposing)：最核心的产出阶段。此时硬件受保护，不接受非紧急指令。 - 收尾期 (
processing)：数据搬运阶段。此时传感器可以休息，但 CPU 和存储在工作。 - 异常期 (
error)：系统的“安全气囊”。一旦发生任何非预期事件，立即跳入此状态。
2. 状态机的核心设计技巧
- 状态的“原子性”： 将
exposing和processing分开。这样设计的好处: 如果未来我们要升级系统支持“连拍”，我们可以让相机在processing（保存上一张图）的同时，逻辑上已经回到了calibrating准备下一张。 - 双重保险 (Conditions vs Unless)： 在
capture触发器中，我们不仅使用了conditions（满足天黑才拍），还使用了unless（如果不天黑就报错）。这种互斥设计保证了逻辑没有灰色地带。 - 全局异常捕获： 使用通配符
*监听hardware_failure。这意味着无论代码运行到哪一行, 只要底层驱动报故障, 状态机都能第一时间切断当前流程, 保护昂贵的感光元件。
3. 状态转移图 (ASCII / TXT 格式)
Plaintext
================================================================================
全自动天文摄影系统状态转移图 (FSM)
================================================================================
[ 重置 (reset) ] <-------------------------------------------+
| |
v |
+-------------------+ start_session +------------------------+
| IDLE | ---------------------------> | CALIBRATING |
| (等候指令) | <--------------------------- | (自检/制冷/对焦) |
+-------------------+ (自动跳转/to_idle) +------------------------+
^ |
| | capture()
| | [条件检查]
| | 1. 天够黑吗?
| | 2. 空间够吗?
| hardware_failure(*) |
+------------------------------------------+ |
| | |
| v v
+-------------------+ save_complete +------------------------+
| PROCESSING | <--------------------- | EXPOSING |
| (保存/降噪) | | (快门开启/跟踪中) |
+-------------------+ +------------------------+
| |
| | finish_exposure()
+------------------------------------------+
|
v
+------------------------+
| ERROR |
| (故障/环境异常) |
+------------------------+
--------------------------------------------------------------------------------
注:
1. [*] 代表通配符，表示从任何状态遇到该触发器都会跳入 ERROR。
2. [Conditions] 在 CALIBRATING -> EXPOSING 的箭头上做拦截。
3. [Hooks] 在 EXPOSING 状态的前后分别挂载了 "开启快门" 和 "开启跟踪" 的动作。
================================================================================
4. 关键点详细说明
整理设计规范：
- 关于
to_idle()的强制性： 在 TXT 图中可以看到，to_idle是从任何地方指回IDLE的虚线。在 Python 代码中，这是由auto_transitions=True支撑的。在大神的代码里，这通常作为finally块或异常处理的最后一道防线 - 关于
is_shutter_open的布尔值： 这是为了处理“逻辑状态”与“物理现实”的延迟。 状态机切换状态是毫秒级的，但机械快门关闭可能需要几十毫秒。通过self.is_shutter_open标志位，我们可以确保在finish_exposure转换完成前，程序能够准确知道物理硬件是否安全。 - 关于
on_enter与on_exit的妙用： 在例子中，calibrating状态使用了字典定义。on_enter: ['check_system_temp']：这就像是 Z30 拨到拍摄档时自动清理传感器。它保证了只要系统处于这个状态，前提条件就一定被检查过。- 这种写法避免了你在每一个
add_transition里重复写检查逻辑，是代码复用的最高体现。
5. 典型学习链接
- Transitions GitHub Repository:
- 必读理由：最详尽的 README，包含了“图表绘制”、“异步状态机”、“分层状态机”等所有进阶玩法。
- Refactoring Guru - State Pattern (中文):
- 必读理由：如果你想从零用 C++ 写一个状态机而不依赖库，这里有最清晰的类图和代码结构说明。
- XState Docs (Concepts):
- 必读理由：虽然是 JS 的库，但它的文档对“什么是状态机”、“什么是正交状态”解释得最专业，其可视化工具是行业标杆。
- Python-Transitions Snippets:
- 技巧来源：在 Issues 标签下搜索 “Best practices”，可以看到大神们如何处理复杂的并发状态转换。
5.2 完整代码实现
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
from transitions import Machine
import random
class AstroCameraSystem:
# 1. 定义状态 (使用字典格式，增加进入/退出状态的回调)
# 状态的定义是整个系统的骨架。之所以采用这种“字符串与字典混合”的写法，是为了在简洁性与功能性之间取得平衡
states = [
'idle', # 空闲状态
{ # 字典格式; 自动化管理状态的进入与退出行为
'name': 'calibrating', # 校准状态
# 类似面向对象的 init 方法
'on_enter': ['check_system_temp'], # 进入状态时检查温度 (将检查逻辑绑定状态本身, 防止作者写 add_transition 时忘记调用检查函数 & 复用性)
# 类似面向对象的 del 方法
'on_exit': 'notify_calibration_done' # 退出状态时通知 (保证系统进入下个状态前, 上个状态留下的 "烂摊子" 已经处理完毕)
},
'exposing', # 曝光状态
'processing', # 处理中状态
'error' # 错误状态
]
def __init__(self):
self.sensor_temp = 25 # 传感器初始温度
self.disk_space = 100 # 剩余磁盘空间 %
self.is_shutter_open = False # 快门状态
# 2. 初始化 Machine (涵盖大部分全局参数)
self.machine = Machine(
model=self,
states=AstroCameraSystem.states,
initial='idle',
before_state_change='log_transition_start', # 全局回调: 任何状态转换前/后都会执行
after_state_change='log_transition_end',
auto_transitions=True, # 自动生成 to_state() 方法, 允许强制跳转
send_event=True # 如果转换非法, 直接抛出异常而不是返回 False
)
# 3. 添加复杂的转换规则 (add_transition)
# 简单转换: 空闲 -> 校准
self.machine.add_transition(
trigger='start_session', # 触发器名称, 自动给 camera 对象添加 start_session() 方法
source='idle', # 触发器作用的源状态
dest='calibrating' # 触发器转换到的目标状态
)
# 带条件判断的转换：校准 -> 曝光
# 只有在 is_dark() 为 True 且 disk_has_space 为 True 时才能拍摄
self.machine.add_transition(
trigger='capture', # 触发器名称
source='calibrating', # 源状态: 校准
dest='exposing', # 目标状态: 曝光
conditions=['is_dark', 'disk_has_space'], # 条件列表, is_dark() & disk_has_space() 都返回 True 才能转换
before='open_shutter', # 前置动作: 转换前打开快门
after='track_stars' # 后置动作: 转换后开始跟踪星点
)
# 转换失败的处理：如果不满足 capture 条件，可以定义另一个同名 trigger 跳转到 error
self.machine.add_transition(
trigger='capture', # 同名触发器
source='calibrating', # 源状态: 校准
dest='error', # 目标状态: 报错
unless=['is_dark'], # 不满足 is_dark() 条件
after='log_weather_error' # 后置动作: 记录错误日志
)
# 曝光完成 -> 处理中
self.machine.add_transition(
trigger='finish_exposure', # 触发器名称
source='exposing', # 源状态: 曝光
dest='processing', # 目标状态: 处理中
before='close_shutter' # 前置动作: 关闭快门
)
# 任务循环：处理完回到校准状态，准备下一张
self.machine.add_transition(
trigger='save_complete', # 触发器名称
source='processing', # 源状态: 处理中
dest='calibrating', # 目标状态: 校准
after='reduce_disk_space' # 后置动作: 减少磁盘空间
)
# 通配符与强制跳转: 任何状态下遇到硬件故障 -> 报错
self.machine.add_transition(
trigger='hardware_failure', # 触发器名称
source='*', # 源状态: 通配符 *, 代表任何状态
dest='error' # 目标状态: 报错
)
# 重置：从错误状态回到空闲
self.machine.add_transition(
trigger='reset', # 触发器名称
source='error', # 源状态: 报错
dest='idle' # 目标状态: 空闲
)
# --- 回调函数与条件逻辑 ---
def log_transition_start(self, event):
print(f"\n[系统日志] 尝试从 {event.state.name} 转换到新状态...")
def log_transition_end(self, event):
print(f"[系统日志] 转换成功！当前状态: {self.state}")
def check_system_temp(self, event):
print(f"[硬件] 正在检查传感器温度... 当前: {self.sensor_temp}°C")
def notify_calibration_done(self, event):
print("[校准] 传感器校准完毕。")
def open_shutter(self, event):
self.is_shutter_open = True
print("[快门] 开启。开始接收光子。")
def close_shutter(self, event):
self.is_shutter_open = False
print("[快门] 关闭。")
def track_stars(self, event):
print("[赤道仪] 开始自动跟踪星点...")
def reduce_disk_space(self, event):
self.disk_space -= 10
print(f"[存储] 文件已保存。剩余空间: {self.disk_space}%")
def log_weather_error(self, event):
print("[错误] 无法拍摄：检测到环境光过强。")
# --- 条件检查 (Conditions) ---
def is_dark(self, event):
# 模拟环境光检查
dark = random.choice([True, False])
print(f"[传感器] 环境光检查: {'天黑适合拍摄' if dark else '太亮了'}")
return dark
def disk_has_space(self, event):
return self.disk_space > 0
# --- 运行演示 ---
if __name__ == "__main__":
camera = AstroCameraSystem()
try:
camera.start_session() # 进入校准状态
camera.capture() # 尝试拍摄 (受随机光照条件影响)
if camera.is_exposing():
camera.finish_exposure()
camera.save_complete()
# 模拟硬件故障
camera.hardware_failure()
# 使用自动生成的强制跳转方法
camera.to_idle()
print(f"\n最终状态: {camera.state}")
except Exception as e:
print(f"操作失败: {e}")
6. 重点难点深度解析
-
Q1: 为什么要区分
idle和calibrating? 在专业设备中,idle是低功耗待机, 而calibrating涉及硬件操作 (如传感器制冷). 将校准独立出来, 可以让系统在连续拍摄时, 每张照片之间只回到calibrating进行快速检查, 而不是退回idle重启整个系统. -
Q2:
self.is_shutter_open的标志位意义? 这叫逻辑与物理状态同步. 状态机负责“想”, 标志位负责“做”. 在强制跳转to_idle()时, 如果标志位为True, 程序应强制触发关门动作, 避免物理损坏. -
Q3: 为什么说
to_idle()是自动生成的? 这是 Transitions 库的特性. 它会根据states列表里的名字, 自动给model注入to_<状态名>()方法. 它是一种越权操作, 跳过所有conditions和source限制, 常用于紧急复位. -
Q4: 同名 Trigger (
capture) 怎么工作? 状态机会按添加顺序检查转换规则. 如果第一个capture的conditions不满足, 它不会报错, 而是自动尝试第二个capture规则. 这允许我们非常优雅地处理“分流逻辑” (如: 成功则拍照, 失败则报错).
本文由作者按照
CC BY 4.0
进行授权