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库来构建一个状态机，包括基本状态、状态转移和转移回调函数、业务场景的案例，相信通过这篇文章的学习和实践，大家已经对状态机有了更深入的认识。