文章目录
1、为什么要用状态机？
用python做一个比较复杂的小项目，需要根据不同的输入，控制摄像头采集执行不同的任务。虽然用流程方式实现了，但阅读起来费劲，还容易出错。所以就用了状态机。
2、状态机是什么？
有限状态机（Finite-state machine, FSM），又称有限状态自动机，简称状态机，是表示有限个状态以及在这些状态之间的转移和动作等行为的数学模型。
FSM是一种算法思想，简单而言，有限状态机由一组状态、一个初始状态、输入和根据输入及现有状态转换为下一个状态的转换函数组成。
有限状态机是有限个状态以及在这些状态之间的转移和动作等行为的数学模型，例如交通信号灯系统。
3、状态图是什么？
一般可以用状态图来对一个状态机进行精确地描述。
例如这个可乐机的状态图，这里可乐卖50美分 。
4、transitions是什么？
transitions是专门设计好的应用于python的一个有限状态机设计库。
当然状态机的设计可以有很多其他的方法，选用该方法其一是因为更加简单，其二是专业人士写的状态机代码相比于自己写的肯定更加完备一点。
官网
https://github.com/pytransitions/transitions
安装
$ pip3 install transitions使用状态机必须定义的两个要素
state：状态节点transition：状态转移。用于从一个状态节点移动到另一个状态节点
二、实战应用
以物体状态变化为例
1、规划state、transition
state：状态节点的说明
所谓状态机，最先接触的肯定是状态，因此在设计状态机之前需要先明确有哪几种状态。
大自然中物质有4种状态：固态、液态、气态、等离子态（等离子态是热的带电气体）。
所以这里我们规划了物质的4种状态，如下表
| 条件/当前状态 | 固态 solid | 液态 liquid | 气态 gas | 等离子态 plasma |
|---|---|---|---|---|
| 熔化 melt | 液态 liquid | |||
| 蒸发 evaporate | 气态 gas | |||
| 升华 sublimate | 气态 gas | |||
| 电离 ionize | 等离子态 plasma |
transition：状态转移
[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-vV8fPnDL-1673197686411)(…/pic/image-20230108094511221.png)]
固态→熔化→液态液态→蒸发→气态固态→升华→气态气态→电离→等离子态2、编写代码
创建一个继承
object的类Matter的实体对象lump，然后调用transitions.Machine()将状态机绑定到这个实体对象上。最终得到了两个东西，一个是状态机对象
machine，一个是具体的实体对象lump。之后设定状态机是用
machine，运行状态机是用具体的实体对象lump。
创建一个基础类
先定义一个类，把它当成基础模型，Matter就是物质类，lump不好理解的话就理解成’冰’。
## 1、创建基础类class Matter: passlump = Matter()state：状态的定义
支持两种定义方式
方式一：状态可以是列表的形式，列表里的形式可以是类、字符串或字典：
from transitions import Machine, State## 1、创建基础类class Matter: passlump = Matter()## 2、定义状态states = [ State(name='solid'), # 类,方式二：也可以应用到transitions模块提供的State类进行初始化，再用add_states()进行添加：
from transitions import Machine, State## 1、创建基础类class Matter: passlump = Matter()## 2、绑定machine = Machine(lump)## 3、定义状态solid = State('solid')liquid = State('liquid')gas = State('gas')plasma = State('plasma')## 4、将状态加入到状态机上machine.add_states([solid, liquid, gas]) # callbacks相关
有三种方式：
- 使用states的
on_enter、on_exit属性进行设置。
on_enter：进入该状态后做的事，此时状态已经转换了。on_exit：即将离开某状态时做的事，此时状态未转换。- 给基础类增加动态方法：
on_enter_{stateName}、on_exit_{stateName}- transition的
before属性、after属性
before: 在 属性转换方法 调用即将开始前进行回调，此时属性未转变。after: 在 属性转换方法 调用即将结束前进行回调，此时属性已改变。先后顺序（使用转换条件进行状态转换，即执行
lump.melt()，实现由solid -> liquid）：
- transition的回调
before属性- state=solid的回调
on_exit属性- 基础类动态方法的回调
on_exit_solid()方法- 【此时状态已改变】
- state=liquid的回调
on_enter属性- 基础类动态方法的回调
on_enter_liquid()方法- transition的回调
after属性
第一种：states的on_enter、on_exit
可以理解为“回调”或者“钩子”。
on_enter：进入该状态后做的事（会执行Matter类中相应名字的函数），注意此时状态已经转换了。on_exit：即将离开某状态时做的事（会执行Matter类中相应名字的函数），注意此时状态未转换。
from transitions import Machine, State## 1、创建基础类class Matter(object): def say_hello(self): print("hello, new state!") def say_goodbye(self): print("goodbye, old state!")lump = Matter()## 2、定义状态，同时给某些状态设置回调states = [ State(name='solid', on_exit=['say_goodbye']), 'liquid', {'name': 'gas', 'on_exit': ['say_goodbye']}, {'name': 'plasma', 'on_enter': ['say_hello']}]## 3、绑定并将状态加入到状态机上machine = Machine(lump, states=states)## 4、追加设置某些状态的回调machine.on_enter_liquid('say_hello') # 使用'on_enter_{stateName},动态方法'的方式设置进入liquid状态时执行的回调第二种：给基础类增加动态方法：on_enter_{stateName}、on_exit_{stateName}
# 代码看不懂可以先往下看，回头再来看即可from transitions import State, Machineclass Matter: def say_hello(self): print('hello, ' + self.state + ' state!') def say_goodbye(self): print('goodbye, ' + self.state + ' state!') def on_exit_solid(self): print('Current ' + self.state + ' state!') def on_enter_liquid(self): print('Current ' + self.state + ' state!') def __init__(self): states = [ State(name='solid', on_enter='say_hello', on_exit='say_goodbye'), State(name='liquid', on_enter='say_hello', on_exit='say_goodbye'), State(name='gas', on_enter='say_hello', on_exit='say_goodbye'), State(name='plasma', on_enter='say_hello', on_exit='say_goodbye'), ] self.machine = Machine(model=self, states=states, initial=states[0]) self.machine.add_transition(trigger='melt', source='solid', dest='liquid') self.machine.add_transition(trigger='evaporate', source='liquid', dest='gas') self.machine.add_transition(trigger='sublimate', source='solid', dest='gas') self.machine.add_transition(trigger='ionize', source='gas', dest='plasma')lump = Matter()print(lump.state) # solidlump.melt()# goodbye, solid state!# Current solid state!# hello, liquid state!# Current liquid state!print(lump.state) # liquidprint(lump.is_solid()) # Fasleprint(lump.is_liquid()) # True第三种：transition的回调before属性、after属性
from transitions import State, Machineclass Matter: def say_hello(self): print('hello, ' + self.state + ' state!') def say_goodbye(self): print('goodbye, ' + self.state + ' state!') def on_exit_solid(self): print('exit Current ' + self.state + ' state!') def on_enter_liquid(self): print('enter Current ' + self.state + ' state!') def hello(self): print('hello after, ' + self.state + ' state!') def goodbye(self): print('goodbye before, ' + self.state + ' state!') def __init__(self): states = [ State(name='solid', on_enter='say_hello', on_exit='say_goodbye'), State(name='liquid', on_enter='say_hello', on_exit='say_goodbye'), State(name='gas', on_enter='say_hello', on_exit='say_goodbye'), State(name='plasma', on_enter='say_hello', on_exit='say_goodbye'), ] self.machine = Machine(model=self, states=states, initial=states[0]) # before: 在 属性转换方法 调用开始之前进行回调，此时属性未转变 # after: 在 属性转换方法 调用即将结束前进行回调，此时属性已改变 self.machine.add_transition(trigger='melt', source='solid', dest='liquid', before='goodbye', after='hello') self.machine.add_transition(trigger='evaporate', source='liquid', dest='gas') self.machine.add_transition(trigger='sublimate', source='solid', dest='gas') self.machine.add_transition(trigger='ionize', source='gas', dest='plasma')lump = Matter()print(lump.state) # solidlump.melt()# goodbye before, solid state! # transition的回调`before`属性# goodbye, solid state! # state=solid的回调`on_exit`属性# exit Current solid state! # 基础类动态方法的回调`on_exit_{stateName}`方法# hello, liquid state! # state=liquid的回调`on_enter`属性# enter Current liquid state! # 基础类动态方法的回调`on_enter_{stateName}`方法# hello after, liquid state! # transition的回调`after`属性print(lump.state) # liquidprint(lump.is_solid()) # Fasleprint(lump.is_liquid()) # Truetransition：状态转换的定义
支持两种定义方式
方式一：列表的形式，状态切换的格式如下所示：
from transitions import Machine, State## 1、创建基础类class Matter(object): def say_hello(self): print("hello, new state!") def say_goodbye(self): print("goodbye, old state!")lump = Matter()## 2、定义状态，同时给某些状态设置回调states = [ State(name='solid', on_exit=['say_goodbye']), 'liquid', {'name': 'gas', 'on_exit': ['say_goodbye']}, State(name='plasma', on_enter=['say_hello'])]## 3、定义转换规则transitions = [ { 'trigger': 'melt', 'source': 'solid', 'dest': 'liquid' }, { 'trigger': 'evaporate', 'source': 'liquid', 'dest': 'gas' }, { 'trigger': 'sublimate', 'source': 'solid', 'dest': 'gas' }, { 'trigger': 'ionize', 'source': 'gas', 'dest': 'plasma' }, { 'trigger': 'any', 'source': '*', 'dest': 'solid' } # *为任意状态]## 4、绑定并将状态及转换规则加入到状态机上machine = Machine(model=lump, states=states, transitions=transitions)定义转换规则可以简化为:
transitions = [ ['melt', 'solid', 'liquid'], ['evaporate', 'liquid', 'gas'], ['sublimate', 'solid', 'gas'], ['ionize', 'gas', 'plasma'], ['any', '*', 'solid']]方式二：通过add_transition函数添加转换：
from transitions import Machine, State## 1、创建基础类class Matter(object): def say_hello(self): print("hello, new state!") def say_goodbye(self): print("goodbye, old state!")lump = Matter()## 2、定义状态，同时给某些状态设置回调states = [ State(name='solid', on_exit=['say_goodbye']), 'liquid', {'name': 'gas', 'on_exit': ['say_goodbye']}, State(name='plasma', on_enter=['say_hello'])]# 3、绑定并将状态加入到状态机上，同时指定初始状态为'solid'# 如果未指定初始状态则默认为'initial'machine = Machine(model=lump, states=states, initial='solid')# 4、添加转换条件machine.add_transition('melt', source='solid', dest='liquid')# machine.add_transition('melt', source='solid', dest='plasma') # 只有第一个匹配melt的transition有效，因为它们trigger相同，source也相同，所以即使写了此行也不会生效。# machine.add_transition('melt', source='gas', dest='plasma') # 相同trigger时source不同的择机匹配,即melt()的source支持solid或gas，如果source都不是则抛出异常machine.add_transition('evaporate', source='liquid', dest='gas')machine.add_transition('sublimate', source='solid', dest='gas')machine.add_transition('ionize', source='gas', dest='plasma')使用
## 把 完成"状态转换的定义"后的任意代码放到此处即可# 打印当前状态print(lump.state)# is_{stateName}判断当前状态是否为solidprint(lump.is_solid()) # 返回布尔值# (非常不推荐使用,会无视设置的转换条件)强制转换到liquid状态lump.to_liquid()# (推荐)使用转换条件进行状态转换,如不符合设置的转换条件则会抛出异常lump.melt()# (也可)使用trigger方法调用melt转化,如不符合设置的转换条件则会抛出异常lump.trigger("melt")# 获取solid状态触发器machine.get_triggers('solid')3、代码总结
单纯绘图
主要用来前期理清 state、transition思路
from transitions.extensions import GraphMachine""" GraphMachine依赖graphviz，需要先安装`pip3 install graphviz` 并确保Graphviz可执行文件在您的系统上PATH https://github.com/xflr6/graphviz http://graphviz.org/"""class Matter: passstates = ['固态 solid', '液态 liquid', '气态 gas', '等离子态 plasma']transitions = [ {'trigger': '熔化 melt', 'source': '固态 solid', 'dest': '液态 liquid'}, {'trigger': '蒸发 evaporate', 'source': '液态 liquid', 'dest': '气态 gas'}, {'trigger': '升华 sublimate', 'source': '固态 solid', 'dest': '气态 gas'}, {'trigger': '电离 ionize', 'source': '气态 gas', 'dest': '等离子态 plasma'}]machine = GraphMachine(Matter(), states=states, transitions=transitions, initial='固态 solid')# machine = GraphMachine(lump, states=states, transitions=transitions, initial='solid', show_auto_transitions=True) # 完整流程graph = machine.get_graph()graph.edge_attr['fontname'] = 'Microsoft Yahei'graph.node_attr['fontname'] = 'Microsoft Yahei'graph.graph_attr['fontname'] = 'Microsoft Yahei'graph.graph_attr['dpi'] = '300' # 设置分辨率graph.graph_attr.pop('label') # 删除标题graph.draw('result.png', prog='dot')封装写法
from transitions import Machineclass Matter: states = ['solid', 'liquid', 'gas', 'plasma'] # 状态有固态、液态、气态、等离子态 transitions = [ {'trigger': 'melt', 'source': 'solid', 'dest': 'liquid'}, {'trigger': 'evaporate', 'source': 'liquid', 'dest': 'gas'}, {'trigger': 'sublimate', 'source': 'solid', 'dest': 'gas'}, {'trigger': 'ionize', 'source': 'gas', 'dest': 'plasma'} ] def __init__(self): """不定型物体 固态→熔化→液态 液态→蒸发→气态 固态→升华→气态 气态→电离→等离子态 """ self.machine = Machine(model=self, states=Matter.states, transitions=Matter.transitions, initial='solid')lump = Matter()print(lump.state) # solidlump.melt()print(lump.state) # liquid枚举写法
from enum import Enumfrom transitions import Machineclass States(Enum): ERROR = 0 RED = 1 YELLOW = 2 GREEN = 3transitions = [ # 注意这里有两个名为proceed的trigger，代表着匹配到哪个source就用哪个，如果都没匹配到则抛出异常 ['proceed', States.RED, States.YELLOW], ['proceed', States.YELLOW, States.GREEN], ['error', '*', States.ERROR]]m = Machine(states=States, transitions=transitions, initial=States.RED)assert m.is_RED(), "当前状态不是红色"assert m.state is States.RED, "当前状态不是红色"state = m.get_state(States.RED) # auto_transitions参数设置状态机
当 True（默认值）时，每个状态都会在基本模型中自动具有关联的 to_{state}（） 便利触发器。
from transitions import Machineclass Matter: passstates = ['solid', 'liquid', 'gas', 'plasma']transitions = [ {'trigger': 'melt', 'source': 'solid', 'dest': 'liquid'}, {'trigger': 'evaporate', 'source': 'liquid', 'dest': 'gas'}, {'trigger': 'sublimate', 'source': 'solid', 'dest': 'gas'}, {'trigger': 'ionize', 'source': 'gas', 'dest': 'plasma'}]lump = Matter()machine = Machine(model=lump, states=states, transitions=transitions, initial='solid')print(lump.state) # solidprint(machine.get_triggers('solid')) # 获取固态的触发器（转换状态的函数）lump.to_plasma() # 强制转换为等离子态 (不推荐使用强制转换，这里仅测试)print(lump.state) # plasmaprint()lump = Matter()machine = Machine(model=lump, states=states, transitions=transitions, initial='solid', auto_transitions=False) # 禁止自动转换print(lump.state) # 初始化状态为固态solidprint(machine.get_triggers('solid')) # 获取固态的触发器（转换状态的函数）, ['melt', 'sublimate']try: lump.to_plasma() # 尝试转换为等离子态,失败except Exception as e: print(e) # 'Matter' object has no attribute 'to_plasma'lump.melt()print(lump.state) # liquid获取转换逻辑，即获取transition
from transitions import Machineclass Matter: passdef get_transitions(machine, auto_transitions=False): """获取转换逻辑""" transitions = [] for trigger, event in machine.events.items(): if auto_transitions is False and trigger.startswith('to_'): continue for source, _transitions in event.transitions.items(): for i in _transitions: transitions.append({'trigger': trigger, 'source': source, 'dest': i.dest}) return transitionslump = Matter()states = ['solid', 'liquid', 'gas', 'plasma']transitions = [ {'trigger': 'melt', 'source': 'solid', 'dest': 'liquid'}, {'trigger': 'evaporate', 'source': 'liquid', 'dest': 'gas'}, {'trigger': 'sublimate', 'source': 'solid', 'dest': 'gas'}, {'trigger': 'ionize', 'source': 'gas', 'dest': 'plasma'}]machine = Machine(model=lump, states=states, transitions=transitions)print(get_transitions(machine))"""[ {'trigger': 'melt', 'source': 'solid', 'dest': 'liquid'}, {'trigger': 'evaporate', 'source': 'liquid', 'dest': 'gas'}, {'trigger': 'sublimate', 'source': 'solid', 'dest': 'gas'}, {'trigger': 'ionize', 'source': 'gas', 'dest': 'plasma'}]"""批量定义转换规则
add_transition()的参数
source可为列表或字符串或通配符*
from transitions import Machineclass Matter: passlump = Matter()states = ['solid', 'liquid', 'gas', 'plasma']transitions = [ {'trigger': 'melt', 'source': 'solid', 'dest': 'liquid'}, {'trigger': 'evaporate', 'source': 'liquid', 'dest': 'gas'}, {'trigger': 'sublimate', 'source': 'solid', 'dest': 'gas'}, {'trigger': 'ionize', 'source': 'gas', 'dest': 'plasma'}]machine = Machine(model=lump, states=states, transitions=transitions, auto_transitions=False) # 禁止自动转换machine.add_transition('transmogrify', ['solid', 'liquid', 'gas'], 'plasma') # 批量添加不同sourcemachine.add_transition('to_liquid', '*', 'liquid') # 通配符添加print(machine.get_triggers('solid'))print(machine.get_triggers('liquid'))print(machine.get_triggers('gas'))print(machine.get_triggers('plasma'))# ['melt', 'sublimate', 'transmogrify', 'to_liquid']# ['evaporate', 'transmogrify', 'to_liquid']# ['ionize', 'transmogrify', 'to_liquid']# ['to_liquid']自反转换 machine.add_transition(dest = “=”)
自己转换为自己，注意状态转换是实际发生的，所以会调用state相关的回调
on_exit或on_enter。
from transitions import Machineclass Matter: def say_hello(self): print("hello, new state!") def say_goodbye(self): print("goodbye, old state!")def change_shape(): print(1)lump = Matter()states = ['solid', 'liquid', 'gas', 'plasma']transitions = [ {'trigger': 'melt', 'source': 'solid', 'dest': 'liquid'}, {'trigger': 'evaporate', 'source': 'liquid', 'dest': 'gas'}, {'trigger': 'sublimate', 'source': 'solid', 'dest': 'gas'}, {'trigger': 'ionize', 'source': 'gas', 'dest': 'plasma'}]machine = Machine(model=lump, states=states, transitions=transitions, initial='solid', auto_transitions=False) # 禁止自动转换machine.add_transition('touch', ['liquid', 'gas', 'plasma'], '=', after=change_shape) # 自反转换，并在转换后调用某函数machine.on_enter_liquid('say_hello')print(machine.get_triggers('solid'))print(machine.get_triggers('liquid'))print(machine.get_triggers('gas'))print(machine.get_triggers('plasma'))print()# ['melt', 'sublimate']# ['evaporate', 'touch']# ['ionize', 'touch']# ['touch']print(lump.state)lump.melt()print(lump.state)lump.touch()print(lump.state)# solid# hello, new state!# liquid# hello, new state!# 1# liquid内部转换 machine.add_transition(dest=None)
不同于自反转换，内部转换不会真正转换状态，意味着transition的回调before或after会被调用，而state相关的回调on_exit或on_enter则不会被调用。
from transitions import Machineclass Matter: def say_hello(self): print("hello, new state!") def say_goodbye(self): print("goodbye, old state!")def change_shape(): print(1)lump = Matter()states = [ {'name': 'solid'}, {'name': 'liquid', "on_exit": ['say_goodbye']}, 'gas', 'plasma']transitions = [ {'trigger': 'melt', 'source': 'solid', 'dest': 'liquid'}, {'trigger': 'evaporate', 'source': 'liquid', 'dest': 'gas'}, {'trigger': 'sublimate', 'source': 'solid', 'dest': 'gas'}, {'trigger': 'ionize', 'source': 'gas', 'dest': 'plasma'}]machine = Machine(model=lump, states=states, transitions=transitions, initial='solid', auto_transitions=False) # 禁止自动转换machine.add_transition('internal', ['liquid', 'gas'], None, after=change_shape) # 自反转换，并在转换后调用某函数machine.on_enter_liquid('say_hello')print(machine.get_triggers('solid'))print(machine.get_triggers('liquid'))print(machine.get_triggers('gas'))print(machine.get_triggers('plasma'))print()# ['melt', 'sublimate']# ['evaporate', 'internal']# ['ionize', 'internal']# []print(lump.state)lump.melt()print(lump.state)lump.internal() # 内部转换,不会真正转换状态,意味着transition的回调如before或after会被调用,而state相关的回调如on_exit或on_enter则不会被调用print(lump.state)# solid# hello, new state!# liquid# 1# liquid顺序转换
常见的需求是转换状态按顺序转换，如给定状态 ['A', 'B', 'C']，转换状态的情况只有 A → B、 B → C、 C → A。
调用 add_ordered_transitions()
import randomfrom transitions import Machineclass Matter: passdef check(): success = random.randint(0, 1) # 生成一个指定范围内的整数，包含端点 if success: print('转换状态') else: print('不转换状态') return successdef f1(): return check()def f2(): return check()def f3(): return check()states = ['A', 'B', 'C']machine = Machine(states=states, initial=states[0])machine.add_ordered_transitions()# machine.add_ordered_transitions(['A', 'C', 'B']) # 指定顺序# machine.add_ordered_transitions(conditions=check) # 为进行转换必须通过的条件，True则转换状态# machine.add_ordered_transitions(conditions=[f1, f2, f3]) # 同上，列表里的个数与states同，一一对应# machine.add_ordered_transitions(loop=False) # 禁止循环for _ in range(4): # 转换四次 print(machine.state) machine.next_state() # A # B # C # A队列转换
队列特点是先进先出。
转换状态会马上处理事件，意味着 on_enter 会在 after 前调用。
from transitions import Machineclass Matter: passdef go_to_C(): global machine machine.to_C()def after_advance(): print('I am in state B now!')def entering_C(): print('I am in state C now!')states = ['A', 'B', 'C']machine = Machine(states=states, initial=states[0])# after: 在属性转换方法调用即将结束前进行回调，此时属性已改变。machine.add_transition('advance', 'A', 'B', after=after_advance)print(machine.state) # Amachine.on_enter_B(go_to_C) # 转换为B时调用go_to_Cmachine.on_enter_C(entering_C) # 转换为C时调用entering_Cmachine.advance()print(machine.state) # C调用链为：prepare -> before -> on_enter_B -> on_enter_C -> after
满足条件才转换状态
调用 Machine.add_transition()
- conditions：满足条件才转换状态，为可调用对象及其列表。即所有返回为
True才转换状态。 - unless：不满足条件才转换状态，类似
conditions
from transitions import Machineclass Matter: def is_flammable(self): """是否易燃物""" return False def is_really_hot(self): """高温""" return Truelump = Matter()states = ['solid', 'liquid', 'gas', 'plasma']transitions = [ {'trigger': 'melt', 'source': 'solid', 'dest': 'liquid'}, {'trigger': 'evaporate', 'source': 'liquid', 'dest': 'gas'}, {'trigger': 'sublimate', 'source': 'solid', 'dest': 'gas'}, {'trigger': 'ionize', 'source': 'gas', 'dest': 'plasma'}]machine = Machine(model=lump, states=states, transitions=transitions, initial='solid', auto_transitions=False) # 禁止自动转换machine.add_transition('heat', 'solid', 'gas', conditions='is_flammable') # 加热时如果是易燃物会变气态# machine.add_transition('heat', 'solid', 'gas', unless=['is_flammable', 'is_really_hot']) # 加热时不易燃且不高温变气态machine.add_transition('heat', 'solid', 'liquid', conditions=['is_really_hot']) # 加热时如果高温会变液态print(lump.state)lump.heat()print(lump.state)# solid# liquid转换状态前后回调
调用 Machine.add_transition()
- before：转换状态前调用。
- after：转换状态后调用。
- prepare：触发器激活时调用。
from transitions import Machineclass Matter: def make_hissing_noises(self): print("HISSSSSSSSSSSSSSSS") def disappear(self): print("where'd all the liquid go?")lump = Matter()states = ['solid', 'liquid', 'gas', 'plasma']transitions = [ {'trigger': 'melt', 'source': 'solid', 'dest': 'liquid', 'before': 'make_hissing_noises'}, {'trigger': 'evaporate', 'source': 'liquid', 'dest': 'gas', 'after': 'disappear'}]machine = Machine(model=lump, states=states, transitions=transitions, initial='solid')print(lump.state)lump.melt()print(lump.state)lump.evaporate()print(lump.state)# solid# HISSSSSSSSSSSSSSSS# liquid# where'd all the liquid go?# gas加热物体并在沸腾的时候输出加热了几次
import randomfrom transitions import Machineclass Matter: heat = False # 是否沸腾 attempts = 0 # 尝试次数 def heat_up(self): """随机沸腾""" self.heat = random.random() < 0.25 def count_attempts(self): self.attempts += 1 def stats(self): print('It took you %i attempts to melt the lump!' % self.attempts) @property def is_really_hot(self): return self.heatlump = Matter()states = ['solid', 'liquid', 'gas', 'plasma']transitions = [ {'trigger': 'melt', 'source': 'solid', 'dest': 'liquid', 'prepare': ['heat_up', 'count_attempts'], 'conditions': 'is_really_hot', 'after': 'stats'},]machine = Machine(model=lump, states=states, transitions=transitions, initial='solid')print(lump.state)lump.melt()lump.melt()lump.melt()lump.melt()print(lump.state)# solid# It took you 4 attempts to melt the lump!# liquid除非当前状态是命名转换的有效源，否则不会调用 prepare
状态机转换状态前后回调
初始化状态机对象 Machine
- before_state_change：转换状态前调用可调用对象。
- after_state_change：转换状态后调用可调用对象。
from transitions import Machineclass Matter: def before(self): print('before') def after(self): print('after') def prepare(self): print('prepare') def before_state_change(self): print('before_state_change') def after_state_change(self): print('after_state_change')lump = Matter()states = ['solid', 'liquid', 'gas', 'plasma']transitions = [ {'trigger': 'melt', 'source': 'solid', 'dest': 'liquid', 'before': 'before', 'after': 'after', 'prepare': 'prepare'},]machine = Machine(model=lump, states=states, transitions=transitions, initial='solid', before_state_change='before_state_change', after_state_change='after_state_change')lump.melt()# prepare# before_state_change# before# after# after_state_change异常处理
from transitions import Machineclass Matter: def raise_error(self, event): raise ValueError('Oh no') def prepare(self, event): print('I am ready!') def finalize(self, event): print('Result: ', type(event.error), event.error)lump = Matter()states = ['solid', 'liquid', 'gas', 'plasma']machine = Machine(model=lump, states=states, prepare_event='prepare', before_state_change='raise_error', finalize_event='finalize', send_event=True)try: lump.to_gas()except ValueError: passprint(lump.state)# I am ready!# Result: Oh no # initial可以传递给 on_exception
from transitions import Machineclass Matter(object): def raise_error(self, event): raise ValueError('Oh no') def handle_error(self, event): print('Fixing things ...') del event.error # 看不到就不会发生states = ['solid', 'liquid', 'gas', 'plasma']lump = Matter()m = Machine(lump, states, before_state_change='raise_error', on_exception='handle_error', send_event=True)try: lump.to_gas()except ValueError: passprint(lump.state)# Fixing things ...# initial回调协议
可能你意识到，给 states、conditions、transitions 赋予的回调通过名称，会从模型中检索相关可调用对象。如果检索不到，或包含点，则会将其视作通往模块函数的路径，并尝试导入。
或者，可以传递属性或属性名，但这样无法接收 event 数据
也可以传递函数之类的可调用对象
还可以传递可调用对象的列表/元组，将按顺序执行
# my_tool.pydef imported_func(): print(1)# main.pyfrom transitions import Machinefrom my_tool import imported_funcimport randomclass Model(object): def a_callback(self): imported_func() @property def a_property(self): return random.random() < 0.5 an_attribute = Falsemodel = Model()machine = Machine(model=model, states=['A'], initial='A')machine.add_transition('by_name', 'A', 'A', conditions='a_property', after='a_callback')machine.add_transition('by_reference', 'A', 'A', unless=['a_property', 'an_attribute'], after=model.a_callback)machine.add_transition('imported', 'A', 'A', after='my_tool.imported_func')model.by_name()model.by_reference()model.imported()# 1# 1回调协议在 Machine.resolve_callable 中实现，如果需要更复杂的解析策略，可以重写此方法
from transitions import Machineclass CustomMachine(Machine): @staticmethod def resolve_callable(func, event_data): super(CustomMachine, CustomMachine).resolve_callable(func, event_data)回调执行顺序
有三种方法触发事件：
- lump.melt()
- lump.trigger(‘melt’)
- machine.dispatch(‘melt’)
并按以下顺序执行转换状态的回调
| 回调 | 当前状态 | 备注 |
|---|---|---|
| ‘machine.prepare_event’ | source | 在处理个别转换前执行一次 |
| ‘transition.prepare’ | source | 在转换开始时执行 |
| ‘transition.conditions’ | source | 条件可能会失效并停止转换 |
| ‘transition.unless’ | source | 条件可能会失效并停止转换 |
| ‘machine.before_state_change’ | source | 在模型上声明的默认回调函数 |
| ‘transition.before’ | source | |
| ‘state.on_exit’ | source | 在源状态上声明的回调 |
| STATE CHANGE | ||
| ‘state.on_enter’ | destination | 在目标状态上声明的回调函数 |
| ‘transition.after’ | destination | |
| ‘machine.after_state_change’ | destination | 在模型上声明的默认回调函数 |
| ‘machine.on_exception’ | source/destination | 在引发异常时执行回调函数 |
| ‘machine.finalize_event’ | source/destination | 即使没有发生转换或引发异常，也将执行回调 |
传递参数
给状态机的回调函数传递参数：
- 位置参数
- 关键字参数
from transitions import Machineclass Matter: def __init__(self): self.set_environment() def set_environment(self, temp=0, pressure=101.325): self.temp = temp self.pressure = pressure def print_temperature(self): print(f'Current temperature is {self.temp} degrees celsius.') def print_pressure(self): print(f'Current pressure is {self.pressure:.2f} kPa.')lump = Matter()machine = Machine(model=lump, states=['solid', 'liquid'], initial='solid')machine.add_transition(trigger='melt', source='solid', dest='liquid', before='set_environment')lump.melt(45) # 相当于调用lump.trigger('melt', 45)lump.print_temperature()# Current temperature is 45 degrees celsius.machine.set_state('solid')lump.melt(pressure=300.23)lump.print_pressure()# Current pressure is 300.23 kPa.如果初始化状态机时加入参数 send_event=True ，所有调用触发器的参数将被封装进 EventData 实例 并传递给所有回调函数，但只能用关键字参数。
EventData 还维护了相关的内部引用，如原状态、转换、状态机、触发器等。
from transitions import Machineclass Matter: def __init__(self): self.temp = 0 self.pressure = 101.325 def set_environment(self, event): self.temp = event.kwargs.get('temp', 0) self.pressure = event.kwargs.get('pressure', 101.325) def print_temperature(self): print(f'Current temperature is {self.temp} degrees celsius.') def print_pressure(self): print(f'Current pressure is {self.pressure:.2f} kPa.')lump = Matter()machine = Machine(model=lump, states=['solid', 'liquid'], initial='solid', send_event=True)machine.add_transition(trigger='melt', source='solid', dest='liquid', before='set_environment')lump.melt(temp=45)lump.print_temperature()# Current temperature is 45 degrees celsius.machine.set_state('solid')lump.melt(pressure=300.23)lump.print_pressure()# Current pressure is 300.23 kPa.初始化模式
类继承 Machine，将状态机功能整合到现有类中，比将功能放在独立 machine 实例中更自然
from transitions import Machineclass Matter(Machine): def say_hello(self): print('hello, new state!') def say_goodbye(self): print('goodbye, old state!') def __init__(self): states = ['solid', 'liquid', 'gas'] Machine.__init__(self, states=states, initial='solid') self.add_transition(trigger='melt', source='solid', dest='liquid')lump = Matter()print(lump.state)lump.melt()print(lump.state)# solid# liquid状态机能处理多个模型，例如Machine(model=[model1, model2, ...])
也可以创建单例，machine.add_model(model=None)
调用 machine.dispatch 用于给所有添加了状态机的模型触发事件
调用 machine.remove_model 对暂时使用状态机的模型进行垃圾回收
from transitions import Machineclass Matter: passlump1 = Matter()lump2 = Matter()states = ['solid', 'liquid', 'gas', 'plasma']transitions = [ {'trigger': 'melt', 'source': 'solid', 'dest': 'liquid'}, {'trigger': 'evaporate', 'source': 'liquid', 'dest': 'gas'}, {'trigger': 'sublimate', 'source': 'solid', 'dest': 'gas'}, {'trigger': 'ionize', 'source': 'gas', 'dest': 'plasma'}]machine = Machine(model=None, states=states, transitions=transitions, initial='solid')machine.add_model(lump1)machine.add_model(lump2, initial='liquid')print(lump1.state)print(lump2.state)# solid# liquidmachine.dispatch('to_plasma')print(lump1.state)# plasmaassert lump1.state == lump2.statemachine.remove_model([lump1, lump2])del lump1del lump2状态机的初始化状态为 'initial'，如果不希望有初始化状态，可以传递参数 initial=None，在这种情况下，每次添加模型时都需要指定一个初始状态
from transitions import Machineclass Matter: passstates = ['solid', 'liquid', 'gas', 'plasma']transitions = [ {'trigger': 'melt', 'source': 'solid', 'dest': 'liquid'}, {'trigger': 'evaporate', 'source': 'liquid', 'dest': 'gas'}, {'trigger': 'sublimate', 'source': 'solid', 'dest': 'gas'}, {'trigger': 'ionize', 'source': 'gas', 'dest': 'plasma'}]machine = Machine(model=None, states=states, transitions=transitions, initial=None)try: machine.add_model(Matter())except Exception as e: print(e) # MachineError: No initial state configured for machine, must specify when adding model.machine.add_model(Matter(), initial='liquid')日志
import loggingfrom transitions import Machineclass Matter: passlogging.basicConfig(level=logging.DEBUG)logging.getLogger('transitions').setLevel(logging.INFO)lump = Matter()states = ['solid', 'liquid', 'gas', 'plasma']transitions = [ {'trigger': 'melt', 'source': 'solid', 'dest': 'liquid'}, {'trigger': 'evaporate', 'source': 'liquid', 'dest': 'gas'}, {'trigger': 'sublimate', 'source': 'solid', 'dest': 'gas'}, {'trigger': 'ionize', 'source': 'gas', 'dest': 'plasma'}]machine = Machine(model=lump, states=states, transitions=transitions, initial='solid')print(lump.state)lump.melt()print(lump.state)# solid# liquid# INFO:transitions.core:Finished processing state solid exit callbacks.# INFO:transitions.core:Finished processing state liquid enter callbacks.扩展
转换的核心是轻量级的，同时有各种各样的 MixIns 来扩展它的功能：
- Diagrams：可视化状态机当前状态
- Hierarchical State Machines：用于嵌套和重用
- Threadsafe Locks：并行执行
- Async callbacks：异步执行
- Custom States：扩展状态相关的行为
有两种方式获取所需特性的状态机实例：
-
工厂模式
MachineFactory：设置参数graph、nested、locked、asyncio为Truefrom transitions.extensions import MachineFactorydiagram_cls = MachineFactory.get_predefined(graph=True)nested_locked_cls = MachineFactory.get_predefined(nested=True, locked=True)async_machine_cls = MachineFactory.get_predefined(asyncio=True)class Matter: passmodel = Matter()states = ['solid', 'liquid', 'gas', 'plasma']transitions = [ {'trigger': 'melt', 'source': 'solid', 'dest': 'liquid'}, {'trigger': 'evaporate', 'source': 'liquid', 'dest': 'gas'}, {'trigger': 'sublimate', 'source': 'solid', 'dest': 'gas'}, {'trigger': 'ionize', 'source': 'gas', 'dest': 'plasma'}]machine1 = diagram_cls(model=model, states=states, transitions=transitions)machine2 = nested_locked_cls(model=model, states=states, transitions=transitions) -
直接从
transitions.extensions导入对应的类Diagrams Nested Locked Asyncio Machine ✘ ✘ ✘ ✘ GraphMachine ✓ ✘ ✘ ✘ HierarchicalMachine ✘ ✓ ✘ ✘ LockedMachine ✘ ✘ ✓ ✘ HierarchicalGraphMachine ✓ ✓ ✘ ✘ LockedGraphMachine ✓ ✘ ✓ ✘ LockedHierarchicalMachine ✘ ✓ ✓ ✘ LockedHierarchicalGraphMachine ✓ ✓ ✓ ✘ AsyncMachine ✘ ✘ ✘ ✓ AsyncGraphMachine ✓ ✘ ✘ ✓ HierarchicalAsyncMachine ✘ ✓ ✘ ✓ HierarchicalAsyncGraphMachine ✓ ✓ ✘ ✓ from transitions.extensions import LockedHierarchicalGraphMachine as LHGMachineclass Matter: passmodel = Matter()states = ['solid', 'liquid', 'gas', 'plasma']transitions = [ {'trigger': 'melt', 'source': 'solid', 'dest': 'liquid'}, {'trigger': 'evaporate', 'source': 'liquid', 'dest': 'gas'}, {'trigger': 'sublimate', 'source': 'solid', 'dest': 'gas'}, {'trigger': 'ionize', 'source': 'gas', 'dest': 'plasma'}]machine = LHGMachine(model=model, states=states, transitions=transitions)
常用参数
Machine - 初始化状态机对象
https://github.com/pytransitions/transitions/blob/master/transitions/core.py#L470
- model：需要管理状态的对象(s)
- states：有效状态的列表或枚举
- initial：初始状态
- transitions：可选的转换列表（每个元素是字典或列表）
- send_event：为
True时，传递给触发器方法的任何参数都包装在EventData对象中，允许对数据的间接和封装访问。为False时，所有的位置参数和关键字参数将直接传递给所有回调方法 - auto_transitions：默认为
True，自动给每个状态生成一个to_触发器。() - ordered_transitions：为
True时，则在初始化结束时调用add_ordered_transitions()。 - ignore_invalid_triggers：为
True时，对当前状态无效的触发器调用将被忽略 - before_state_change：转换状态前调用可调用对象。
- after_state_change：转换状态后调用可调用对象。
- name：设置名称，将其用作记录器输出前缀
- queued：为
True时，顺序处理转换。在状态回调函数中执行的触发器将入队并稍后执行。由于队列的性质，所有转换都返回True，因为条件检查不能在排队时进行。 - prepare_event：状态转换前调用可调用对象
- finalize_event：状态转换后调用可调用对象。
- on_exception：引发异常时调用可调用对象。
Machine.add_transition() - 给状态机添加转换方法
https://github.com/pytransitions/transitions/blob/master/transitions/core.py#L935
- trigger：转换状态的方法名。
- source：原状态名，可为字符串或列表。
- dest：目标状态名，为
=或同source时，自反转换。为None时则是内部转换，不会调用exit和enter的回调。 - conditions：满足条件才转换状态，为可调用对象及其列表。即所有返回为
True才转换状态。 - unless：不满足条件才转换状态，类似
conditions - before：转换状态前调用。
- after：转换状态后调用。
- prepare：触发器激活时调用。
- kwargs：关键字参数。
