Python有限状态机——transitions
文章目录
- 简介
- 安装
- 初试
- 绘图
- 回调和状态检查懒方法
- 状态及检查懒方法修改前缀
- 枚举
- 转换状态
- 获取触发器(转换状态的函数)
- 获取转换逻辑
- 批量添加状态转换
- 自反转换
- 内部转换
- 顺序转换
- 队列转换
- 满足条件才转换状态
- 转换状态前后回调
- 状态机转换状态前后回调
- 异常处理
- 回调协议
- 回调执行顺序
- 传递参数
- 初始化模式
- 日志
- 扩展
- Diagrams
- [Hierarchical State Machine (HSM)](https://github.com/pytransitions/transitions#hierarchical-state-machine-hsm)
- 常用参数
- `Machine`
- `Machine.add_transition()`
- 封装
- 参考文献
简介
transitions
是一款轻量的、面向对象的 Python 有限状态机库。
有限状态机是有限个状态以及在这些状态之间的转移和动作等行为的数学模型,例如交通信号灯系统。
本文代码
安装
pip install transitions
初试
以物体状态变化为例,如状态转换表
条件/当前状态 | 固态 solid | 液态 liquid | 气态 gas | 等离子态 plasma |
---|---|---|---|---|
熔化 melt | 液态 liquid | |||
蒸发 evaporate | 气态 gas | |||
升华 sublimate | 气态 gas | |||
电离 ionize | 等离子态 plasma |
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) # solid
lump.melt()
print(lump.state) # liquid
绘图
下载 Graphviz 并添加到环境变量 Path
安装
pip install graphviz
代码
from transitions.extensions import GraphMachineclass 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 Machine
from transitions.extensions import GraphMachinedef draw_machine(machine, filename='result.png'):"""绘制有限状态机"""transitions = []for trigger, event in machine.events.items():for source, _transitions in event.transitions.items():for i in _transitions:transitions.append({'trigger': trigger, 'source': source, 'dest': i.dest})machine = GraphMachine(model=machine.model, states=list(machine.states.keys()), transitions=transitions,initial=machine.initial)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(filename, prog='dot')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 = Machine(Matter(), states=states, transitions=transitions, initial='solid', auto_transitions=False)
draw_machine(machine)
回调和状态检查懒方法
- 状态切换时自动调用函数
- 懒方法判断状态
from transitions import State, Machineclass Matter:def say_hello(self):print('hello, new state!')def say_goodbye(self):print('goodbye, old state!')def on_enter_liquid(self):print('We have just entered state liquid!')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, old state!
# hello, new state!
# We have just entered state liquid!print(lump.state) # liquid
print(lump.is_solid()) # Fasle
print(lump.is_liquid()) # True
状态及检查懒方法修改前缀
- 状态由原本的 state 变成了 <model_attribute>
- 状态检查懒方法由原本的 is_<状态名> 变成了 is_<model_attribute>_<状态名>
state
变成matter_state
is_solid()
变成is_matter_state_solid()
to_solid()
变成to_matter_state_gas()
from transitions import Machineclass Matter:def __init__(self):self.machine = Machine(model=self, states=['solid', 'liquid', 'gas'], initial='solid',model_attribute='matter_state')lump = Matter()
print(lump.matter_state) # solid
print(lump.is_matter_state_solid()) # True
print(lump.to_matter_state_gas()) # True
枚举
from enum import Enum
from transitions import Machineclass States(Enum):ERROR = 0RED = 1YELLOW = 2GREEN = 3transitions = [['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)
print(state.name) # RED
m.proceed()
m.proceed()
assert m.is_GREEN()
m.error()
assert m.state is States.ERROR
转换状态
transitions 为转换列表,每个元素是字典或列表
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'}
]
transitions = [['melt', 'solid', 'liquid'],['evaporate', 'liquid', 'gas'],['sublimate', 'solid', 'gas'],['ionize', 'gas', 'plasma']
]lump = Matter()
machine = Machine(model=lump, states=states, transitions=transitions, initial='solid')
print(lump.state) # solid
常使用参数 auto_transitions=False
来禁止自动转换
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) # 初始化状态为固态solid
print(machine.get_triggers('solid')) # 获取固态的触发器(转换状态的函数)
lump.to_plasma() # 尝试转换为等离子态
print(lump.state) # 竟然成功了,实际上不能直接转换为等离子态
print()lump = Matter()
machine = Machine(model=lump, states=states, transitions=transitions, initial='solid', auto_transitions=False) # 禁止自动转换
print(lump.state) # 初始化状态为固态solid
print(machine.get_triggers('solid')) # 获取固态的触发器(转换状态的函数)
try:lump.to_plasma() # 尝试转换为等离子态失败print(lump.state)
except Exception as e:print(e)
获取触发器(转换状态的函数)
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, ignore_invalid_triggers=True,auto_transitions=False) # 禁止自动转换print(machine.get_triggers('solid'))
print(machine.get_triggers('liquid'))
print(machine.get_triggers('plasma'))
print(machine.get_triggers('solid', 'liquid', 'gas', 'plasma')) # 一次获取多个
# ['melt', 'sublimate']
# ['evaporate']
# []
# ['melt', 'evaporate', 'sublimate', 'ionize']
获取转换逻辑
from transitions import Machinedef get_transitions(machine, auto_transitions=False):"""获取转换逻辑"""transitions = []for trigger, event in machine.events.items():if auto_transitions == False and trigger.startswith('to_'):continuefor source, _transitions in event.transitions.items():for i in _transitions:transitions.append({'trigger': trigger, 'source': source, 'dest': i.dest})return transitionsclass 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)
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'}]
批量添加状态转换
调用 Machine.add_transition(trigger, source, dest, conditions=None, unless=None, before=None, after=None, prepare=None)
参数 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') # 批量添加
machine.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']
自反转换
自己转换为自己
from transitions import Machineclass Matter:passdef 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) # 自反转换,并在转换后调用某函数
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
# liquid
# 1
# liquid
内部转换
不同于自反转换,内部转换不会真正转换状态,意味着转换的回调如 before
或 after
会被调用,而状态相关的回调如 exit
或 enter
则不会被调用。
from transitions import Machineclass Matter:passdef 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('internal', ['liquid', 'gas'], None, after=change_shape) # 自反转换,并在转换后调用某函数
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()
print(lump.state)
# solid
# 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
前调用
import randomfrom transitions import Machineclass Matter:passdef go_to_C():global machinemachine.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])
machine.add_transition('advance', 'A', 'B', after=after_advance)print(machine.state) # A
machine.on_enter_B(go_to_C) # 转换为B时调用go_to_C
machine.on_enter_C(entering_C) # 转换为C时调用entering_C
machine.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 Falsedef 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 random
from transitions import Machineclass Matter:heat = False # 是否沸腾attempts = 0 # 尝试次数def heat_up(self):"""随机沸腾"""self.heat = random.random() < 0.25def count_attempts(self):self.attempts += 1def stats(self):print('It took you %i attempts to melt the lump!' % self.attempts)@propertydef 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:pass
print(lump.state)
# I am ready!
# Result: <class 'ValueError'> 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:pass
print(lump.state)
# Fixing things ...
# initial
回调协议
可能你意识到,给 states、conditions、transitions 赋予的回调通过名称,会从模型中检索相关可调用对象。如果检索不到,或包含点,则会将其视作通往模块函数的路径,并尝试导入。
或者,可以传递属性或属性名,但这样无法接收 event 数据
也可以传递函数之类的可调用对象
还可以传递可调用对象的列表/元组,将按顺序执行
my_tool.py
def imported_func():print(1)
from transitions import Machine
from my_tool import imported_funcimport randomclass Model(object):def a_callback(self):imported_func()@propertydef a_property(self):return random.random() < 0.5an_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):@staticmethoddef resolve_callable(func, event_data):super(CustomMachine, CustomMachine).resolve_callable(func, event_data)
回调执行顺序
回调 | 当前状态 | 备注 |
---|---|---|
‘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 = tempself.pressure = pressuredef 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 = 0self.pressure = 101.325def 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)
# plasma
assert lump1.state == lump2.statemachine.remove_model([lump1, lump2])
del lump1
del 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
为True
from 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)
Diagrams
Hierarchical State Machine (HSM)
常用参数
Machine
- model:需要管理状态的对象(s)
- states:有效状态的列表或枚举
- initial:初始状态
- transitions:可选的转换列表(每个元素是字典或列表)
- send_event:为
True
时,传递给触发器方法的任何参数都包装在EventData
对象中,允许对数据的间接和封装访问。为False
时,所有的位置参数和关键字参数将直接传递给所有回调方法 - auto_transitions:默认为
True
,自动给每个状态生成一个to_<state>()
触发器。 - 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()
- trigger:转换状态的方法名。
- source:原状态名,可为字符串或列表。
- dest:目标状态名,为
=
或同source
时,自反转换。为None
时则是内部转换,不会调用exit
和enter
的回调。 - conditions:满足条件才转换状态,为可调用对象及其列表。即所有返回为
True
才转换状态。 - unless:不满足条件才转换状态,类似
conditions
- before:转换状态前调用。
- after:转换状态后调用。
- prepare:触发器激活时调用。
- kwargs:关键字参数。
封装
1. 获取dest对应的trigger
只适用于一对一对应的情况
from transitions import Machinedef get_dest_trigger_map(machine):"""获取dest对应的trigger"""dest_trigger_map = {}for trigger, event in machine.events.items():for source, _transitions in event.transitions.items():for i in _transitions:dest_trigger_map[i.dest] = triggerreturn dest_trigger_mapclass 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)
dest_trigger_map = get_dest_trigger_map(machine)
print(dest_trigger_map)
2. 直接调用转换
from transitions import Machineclass Matter(Machine):def __init__(self):self.temp = 0states = ['solid', 'liquid', 'gas']Machine.__init__(self, states=states, send_event=True, initial='solid')self.add_transition('melt', 'solid', 'liquid', before='set_environment')def set_environment(self, event):self.temp = event.kwargs.get('temp', 0)def print_pressure(self):print(self.temp)lump = Matter()
lump.melt(temp=45)
lump.print_pressure() # 45
3. How to get the callback result?
参考文献
- transitions GitHub
- transitions-gui GitHub
- graphviz Documentation
- 将点文件(graphviz)转换为图像时,如何设置分辨率?
- get_triggers() in pytransitions not returning expected output
- How to pass data in the case of inheriting from the Machine class?
Python有限状态机——transitions相关推荐
- [python]有限状态机(FSM)简单实现
本文发表于恋花蝶的博客 http://blog.csdn.net/lanphaday,欢迎转载,但必须保留文章内容完整且包含本声明.违者必究. [python]有限状态机(FSM)简单实现 简述 有限 ...
- python 状态机设计(聊聊transitions)
文章目录 1. 状态机的基本概念 1.1 基本理论 1.2 状态机绘图 2. 在python中实现 2.1 状态详解 2.2 转换详解 2.3 实例 1. 状态机的基本概念 1.1 基本理论 有限状态 ...
- python网页填表教程_PythonSpot 中文系列教程 · 翻译完成
原文:PythonSpot Python Tutorials 协议:CC BY-NC-SA 4.0 欢迎任何人参与和完善:一个人可以走的很快,但是一群人却可以走的更远. 在线阅读 ApacheCN 学 ...
- 收集了一些python的文章
转载自:http://blog.csdn.net/xyw_blog/article/details/9128777 newthreading - safer concurrency for Pytho ...
- 收集到一些关于python的文章,存起来慢慢看。。。
2019独角兽企业重金招聘Python工程师标准>>> newthreading - safer concurrency for Python 安全并发(1回应) http://w ...
- PythonSpot 中文系列教程 · 翻译完成
原文:PythonSpot Python Tutorials 协议:CC BY-NC-SA 4.0 欢迎任何人参与和完善:一个人可以走的很快,但是一群人却可以走的更远. 在线阅读 ApacheCN 学 ...
- 喜迎四十万访问量,自荐十六篇好博文
喜迎四十万访问量,自荐十六篇好博文 作者:赖勇浩(http://blog.csdn.net/lanphaday) 2008年12月8日本博的访问量一举突破 40 万大关,猛回头,发现已经在 CSDN ...
- 《恋花蝶的博客》所有文章的官方阅读指引
<恋花蝶的博客>所有文章的官方阅读指引 恋花蝶的博客:http://blog.csdn.net/lanphaday 导语 至2007年7月4日,我的博客(http://blog.csdn. ...
- 【Buildroot】学习记录(2)配置注释
文章目录 一.前言 二.Buildroot目录结构 三.Buildroot配置选项 四.Target options(目标选项) 五.Build options(编译选项) 六.Toolchain(工 ...
最新文章
- 不使用梯度裁剪和使用梯度裁剪的对比(tensorflow)
- ACM寒假训练第二周总结
- DL框架之TensorFlow:深度学习框架TensorFlow Core(低级别TensorFlow API)的简介、安装、使用方法之详细攻略
- 【网络流24题】魔术球问题(最大流)
- linux将db2账户添加到组_超实用的shell脚本--Linux安全加固设置,值得收藏
- (转)JDK 1.5中的ENUM用法
- 【java】深入理解Java JVM虚拟机中init和clinit的区别
- spring的注入方式
- JavaScript之this,new,delete,call,apply
- 海康威视摄像头初始化设置(新相机的第一次配置相机恢复出厂设置)
- PS教程:通道抠图美女发丝
- 从董明珠雷军世纪之赌中看到什么样的格力和小米?
- 海比研究院专访伙伴云VP袁兆江:大厂入局,低代码赛道如何突围?
- Oracle - 同义词
- golang如何发送邮件(qq邮箱)
- RdhA蛋白序列发育树构建及iTol美化
- 英语词缀与英语派生词词典读书笔记,并总结输出思维导图
- 防火墙策略问题排查思路
- 硬件系列(三)--------wifi打印机之使用socket打印(无sdk)
- 蓝桥杯2021届C++B组省赛真题 杨辉三角形
热门文章
- GEE中影像镶嵌的mosaic函数的应用
- 众明星恭祝WOYOTV全新上线
- tensorflow的freeze graph及inference graph_transforms
- 互评Beta版本-SkyHunter
- redhead红帽系统配置yum源
- 阿米巴理念在根力多中的应用
- Effective Java 28 Use bounded wildcards to increase API flexibility
- 【名说】DB2 ERRORCODE=-4499, SQLSTATE=08001 linux环境完美解决方法
- nginx SSL证书
- java中文分词工具_中文分词常用方法简述