Python状态机transitions库详解
概述
状态机(State Machine)是计算机科学里的基础概念,它描述了物体可能的所有状态,在不同的事件或条件下,物体状态会发生相应的改变。在编程中,状态机可以应用到许多场景中,比如游戏状态切换、流程控制等。Python状态机transitions库是一个非常优秀的、易于使用的状态机库,本文将详细介绍该库的使用。
安装
transitions库可以通过pip安装:
pip install transitions
建立状态机
在transitions中,我们通过定义不同状态和状态间的转移来建立状态机。比如我们有一个简单的场景,要实现一个电视机的开关:
from transitions import Machine
class TV(object):
# 定义状态
states = ['off', 'on']
def __init__(self):
self.machine = Machine(model=self, states=TV.states, initial='off')
# 定义状态转移
self.machine.add_transition('power_on', 'off', 'on')
self.machine.add_transition('power_off', 'on', 'off')
首先我们定义了一个TV类,它有两种状态:off和on。我们通过Machine类来创建状态机对象,将其绑定到我们的类中。initial参数指定了初始状态,这里我们将其设为off。接着我们定义了两个方法,power_on和power_off,分别用来实现开机和关机。
接下来,我们可以创建一个TV对象并测试:
t = TV()
print(t.state) # off
t.power_on()
print(t.state) # on
t.power_off()
print(t.state) # off
状态转移回调函数
我们也可以通过状态转移回调函数来实现复杂的状态转移,例如我们想让电视在关机时弹出确认对话框,在开机时显示欢迎词,我们可以这么做:
from transitions import Machine
class TV(object):
# 定义状态
states = ['off', 'on']
def __init__(self):
self.machine = Machine(model=self, states=TV.states, initial='off')
# 定义状态转移
self.machine.add_transition('power_on', 'off', 'on', before='say_hello')
self.machine.add_transition('power_off', 'on', 'off', before='check_shutdown')
# 开机回调函数
def say_hello(self):
print('欢迎使用电视机!')
# 关机回调函数
def check_shutdown(self):
input_str = input('您确定要关闭电视机吗?(y/n)')
if input_str.lower() == 'y':
print('正在关闭电视机...')
else:
print('取消关闭。')
t = TV()
t.power_on()
t.power_off()
通过before参数,我们可以在状态转移前调用指定的回调函数。这里我们在开机时调用了say_hello函数,它将输出欢迎词;在关机时调用了check_shutdown函数,它将显示一个确认对话框,让用户决定是否关闭电视机。
示例一:闹钟
我们来看一个更实际的例子,假设我们要实现一个闹钟程序,它有以下四种状态:未设置、已设置、已响铃未关闭、已响铃已关闭。根据这些状态,我们需要实现以下一些操作:
- set_alarm():设置闹钟时间;
- snooze():将闹钟延迟10分钟;
- alarm_on():开启闹钟;
- alarm_off():关闭闹钟;
- alarm_stop():关闭闹钟铃声。
首先我们需要定义闹钟:
from datetime import datetime, timedelta
from transitions import Machine
class Alarm(object):
# 定义状态
states = ['not_set', 'set', 'ringing', 'halted']
def __init__(self):
self.machine = Machine(model=self, states=Alarm.states, initial='not_set')
# 定义状态转移
self.machine.add_transition('set_alarm', 'not_set', 'set')
self.machine.add_transition('alarm_on', 'set', 'ringing')
# snooze操作可以在已响铃未关闭、已响铃已关闭状态下使用
self.machine.add_transition('snooze', ['ringing', 'halted'], 'ringing', before='delay_alarm')
self.machine.add_transition('alarm_off', '*', 'halted')
self.machine.add_transition('alarm_stop', 'ringing', 'halted', before='stop_ring')
# 处理snooze
def delay_alarm(self):
self.alarm_time = datetime.now() + timedelta(minutes=10)
print('已将闹钟延迟至{}时{}分\n'.format(self.alarm_time.hour, self.alarm_time.minute))
# 处理alarm_stop
def stop_ring(self):
print('闹钟已关闭铃声')
def __str__(self):
if self.state == 'not_set':
return '闹钟未设置'
elif self.state == 'set':
return '闹钟已设置,时间为{}时{}分'.format(self.alarm_time.hour, self.alarm_time.minute)
elif self.state == 'ringing':
return '闹钟响铃中...'
elif self.state == 'halted':
return '闹钟已关闭'
我们定义了一个Alarm类,有四种状态:not_set、set、ringing、halted。在构造函数中,我们定义了状态转移,并且使用了before参数指定了两个回调函数。snnoze操作可以在两种状态下使用,因此,我们简单地将其两个状态中的前缀都设为ring。
对于闹钟来说,它最核心的功能就是算时间,因此我们需要一个时间计算器,它会每秒钟检查一遍当前时间是否达到闹铃时间,达到了就调用alarm_on()方法。我们可以这么写:
import time
alarm = Alarm()
# 设置闹钟时间
alarm.set_alarm(alarm_time=datetime.now() + timedelta(seconds=15))
while alarm.state != 'halted':
# 发出响铃状态检查
if alarm.state == 'ringing':
if datetime.now() > alarm.alarm_time:
alarm.alarm_off()
else:
print(str(alarm))
# 设置闹钟时间
elif alarm.state == 'not_set':
input_str = input('请输入闹钟时间(格式如07:00):')
alarm_time_str = datetime.now().strftime('%Y-%m-%d') + ' ' + input_str
alarm_time = datetime.strptime(alarm_time_str, '%Y-%m-%d %H:%M')
print('已设置闹钟时间{}时{}分'.format(alarm_time.hour, alarm_time.minute))
alarm.set_alarm(alarm_time=alarm_time)
# 其他状态
else:
print(str(alarm))
time.sleep(1)
print('退出闹钟程序')
这个循环代码包含了闹钟的交互逻辑,每秒钟检查一次闹钟状态,如果闹钟处于ringing状态且闹钟时间已过了,就关闭闹钟;如果闹钟未设置,就让用户输入闹钟时间并设置;否则就输出当前状态。
示例二:多个状态机组合
有时候,我们会碰到多个业务状态机需要组合使用的问题,比如在一个工厂中,一批原材料到来后必须进行原材料检测、原材料质检、领取并清空检测台等操作,才能安排生产。这个过程可以抽象成三个状态机,它们是并列的关系。
首先我们需要定义三个状态机类:MaterialCheck、MaterialQC和MaterialClaim。这里我们只给出MaterialCheck的定义,其他状态机定义与之类似:
from transitions import Machine
class MaterialCheck(object):
# 定义状态
states = ['ready', 'testing', 'pass', 'fail']
def __init__(self):
self.machine = Machine(model=self, states=MaterialCheck.states, initial='ready')
self.machine.add_transition('start_test', 'ready', 'testing')
self.machine.add_transition('test_pass', 'testing', 'pass')
self.machine.add_transition('test_fail', 'testing', 'fail')
它的状态包括ready、testing、pass和fail。在构造函数中,我们定义了三个转移事件:start_test、test_pass、test_fail,分别表示开始检测、检测通过和检测不通过三种状态。
接下来,我们可以定义主状态机组合这三个子状态机:
class MaterialAllocation(object):
# 定义状态
states = ['allocate', 'check', 'qc', 'production', 'complete']
transitions = [
{'trigger': 'start_check', 'source': 'allocate', 'dest': 'check'},
{'trigger': 'start_qc', 'source': 'check', 'dest': 'qc', 'conditions': 'qc_ready'},
{'trigger': 'start_production', 'source': 'qc', 'dest': 'production', 'conditions': 'production_ready'},
{'trigger': 'complete_production', 'source': 'production', 'dest': 'complete'}
]
def __init__(self, check, qc, claim):
self.machine = Machine(model=self, states=MaterialAllocation.states, transitions=MaterialAllocation.transitions, initial='allocate')
self.check = check
self.qc = qc
self.claim = claim
# 检查是否有原材料需要检测
def check_ready(self):
return self.claim.count_materials > 0
# 检查是否需要质检
def qc_ready(self):
return self.check.state == 'pass'
# 检查是否已经准备好生产
def production_ready(self):
return self.qc.state == 'pass'
check = MaterialCheck()
qc = MaterialQC()
claim = MaterialClaim()
allocation = MaterialAllocation(check, qc, claim)
# 开始原材料的领取、检测、质检、生产流程
while allocation.state != 'complete':
# 领取原材料
if allocation.state == 'allocate':
claim.count_materials -= 1
allocation.start_check()
# 进行原材料检测
elif allocation.state == 'check':
if allocation.check.state == 'ready':
allocation.check.start_test()
elif allocation.check.state == 'pass':
allocation.start_qc()
else:
print('原材料检测不通过。')
allocation = MaterialAllocation(check, MaterialQC(), claim)
# 进行质量检查
elif allocation.state == 'qc':
if allocation.qc.state == 'ready':
allocation.qc.start_qc()
elif allocation.qc.state == 'pass':
allocation.start_production()
else:
print('质检不通过。')
allocation = MaterialAllocation(check, MaterialQC(), claim)
# 进行生产
elif allocation.state == 'production':
allocation.complete_production()
# 生产完成
else:
print('生产完成!')
这里我们定义了一个MaterialAllocation类作为主状态机,并将MaterialCheck、MaterialQC和MaterialClaim状态机作为参数传入。在主状态机构造函数中,我们定义了状态转移,并在后面的循环中使用特定的流程处理每个状态机的运行。在整个过程中,我们异步执行了三个子状态机,它们通过组合的方式实现了一个完整的业务流程。
至此,我们已经学习了如何使用transitions库来构建一个状态机,包括基本状态、状态转移和转移回调函数、业务场景的案例,相信通过这篇文章的学习和实践,大家已经对状态机有了更深入的认识。
本站文章如无特殊说明,均为本站原创,如若转载,请注明出处:python状态机transitions库详解 - Python技术站