Python状态机（transitions模块）

使用states的`on_enter`、`on_exit`属性进行设置。`on_enter`：进入该状态后做的事，此时状态已经转换了。`on_exit`：即将离开某状态时做的事，此时状态未转换。给基础类增加动态方法：`on_enter_{stateName}`、`on_exit_{stateName}`transition的`before`属性、`after`属性`before`: 在属性转换方

Jenrey

11141人浏览 · 2023-01-09 01:11:17

Jenrey · 2023-01-09 01:11:17 发布

文章目录

二、实战应用
常用参数
- Machine - 初始化状态机对象
- Machine.add_transition() - 给状态机添加转换方法

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
熔化 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:
    pass

lump = Matter()

state：状态的定义

支持两种定义方式

方式一：状态可以是列表的形式，列表里的形式可以是类、字符串或字典：

from transitions import Machine, State

## 1、创建基础类
class Matter:
    pass

lump = Matter()


## 2、定义状态
states = [
    State(name='solid'),  # 类,<State('solid')@4472798992>
    'liquid',  # 字符串
    {'name': 'gas'},  # 字典
    "plasma"
]


## 3、将状态加入到状态机,并将状态机绑定到lump实例对象上
machine = Machine(lump, states)  # <transitions.core.Machine object at 0x10bb1a8b0>

方式二：也可以应用到transitions模块提供的State类进行初始化，再用add_states()进行添加：

from transitions import Machine, State

## 1、创建基础类
class Matter:
    pass

lump = 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]) # <transitions.core.Machine object at 0x10ac6df10>

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, Machine

class 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)  # solid

lump.melt()
# goodbye, solid state!
# Current solid state!
# hello, liquid state!
# Current liquid state!

print(lump.state)  # liquid
print(lump.is_solid())  # Fasle
print(lump.is_liquid())  # True

第三种：transition的回调`before`属性、`after`属性

from transitions import State, Machine

class 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)  # solid
lump.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)  # liquid
print(lump.is_solid())  # Fasle
print(lump.is_liquid())  # True

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、定义转换规则
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}判断当前状态是否为solid
print(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:
    pass


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 = 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


class 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

枚举写法

from enum import Enum
from transitions import Machine


class States(Enum):
    ERROR = 0
    RED = 1
    YELLOW = 2
    GREEN = 3


transitions = [
    # 注意这里有两个名为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)  # <State('RED')@4436112768>
print(state.name)  # RED
m.proceed()  # RED -> YELLOW
m.proceed()  # YELLOW -> GREEN
assert m.is_GREEN(), "当前状态不是绿色"
m.error()
assert m.state is States.ERROR

auto_transitions参数设置状态机

当 True（默认值）时，每个状态都会在基本模型中自动具有关联的 to_{state}（）便利触发器。

from transitions import Machine

class Matter:
    pass

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'}
]

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)  # plasma
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'))  # 获取固态的触发器（转换状态的函数）, ['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 Machine

class Matter:
    pass

def 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 transitions

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)
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 Machine

class Matter:
    pass

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, auto_transitions=False)  # 禁止自动转换
machine.add_transition('transmogrify', ['solid', 'liquid', 'gas'], 'plasma')  # 批量添加不同source
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']

自反转换 machine.add_transition(dest = “=”)

自己转换为自己，注意状态转换是实际发生的，所以会调用state相关的回调on_exit或on_enter。

from transitions import Machine

class 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 Machine

class 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 random
from transitions import Machine

class Matter:
    pass

def check():
    success = random.randint(0, 1)  # 生成一个指定范围内的整数，包含端点
    if success:
        print('转换状态')
    else:
        print('不转换状态')
    return success

def 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 Machine

class Matter:
    pass

def 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)  # 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 Machine

class Matter:
    def is_flammable(self):
        """是否易燃物"""
        return False

    def is_really_hot(self):
        """高温"""
        return True

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('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 Machine

class 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 Machine

class 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.heat

lump = 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 Machine

class 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 Machine

class 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 Machine

class 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)

# main.py
from transitions import Machine
from my_tool import imported_func

import random

class Model(object):

    def a_callback(self):
        imported_func()

    @property
    def a_property(self):
        return random.random() < 0.5

    an_attribute = False

model = 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 Machine

class 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 Machine

class 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 Machine

class 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 Machine

class 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 Machine

class Matter:
    pass

lump1 = 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
# liquid

machine.dispatch('to_plasma')
print(lump1.state)
# plasma
assert lump1.state == lump2.state

machine.remove_model([lump1, lump2])
del lump1
del lump2

状态机的初始化状态为 'initial'，如果不希望有初始化状态，可以传递参数 initial=None，在这种情况下，每次添加模型时都需要指定一个初始状态

from transitions import Machine

class Matter:
    pass

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=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 logging

from transitions import Machine

class Matter:
    pass

logging.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 MachineFactory

diagram_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:
    pass

model = 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 LHGMachine

class Matter:
    pass

model = 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_<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() - 给状态机添加转换方法

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：关键字参数。

开放原子开发者工作坊

开放原子开发者工作坊旨在鼓励更多人参与开源活动，与志同道合的开发者们相互交流开发经验、分享开发心得、获取前沿技术趋势。工作坊有多种形式的开发者活动，如meetup、训练营等，主打技术交流，干货满满，真诚地邀请各位开发者共同参与！

更多推荐

操作系统大会&openEuler Summit 2024参会指南，请查收！

开放原子开发者工作坊

推动工业软件核心技术攻关，开源工业软件算法集成大赛正式启动！

开放原子开发者工作坊

第二届openEuler生态大会（中国·湖南）成功举办

10月30日，第二届openEuler生态大会（中国·湖南）成功举办。

开放原子开发者工作坊

所有评论(0)

查看更多评论

Jenrey

@JENREY

已为社区贡献3条内容

Python状态机（transitions模块）

Jenrey

文章目录

1、为什么要用状态机？

2、状态机是什么？

3、状态图是什么？

4、transitions是什么？

官网

安装

使用状态机必须定义的两个要素

二、实战应用

1、规划state、transition

state：状态节点的说明

transition：状态转移

2、编写代码

创建一个基础类

state：状态的定义

callbacks相关

第一种：states的on_enter、on_exit

第二种：给基础类增加动态方法：on_enter_{stateName}、on_exit_{stateName}

第三种：transition的回调before属性、after属性

transition：状态转换的定义

使用

3、代码总结

单纯绘图

封装写法

枚举写法

auto_transitions参数设置状态机

获取转换逻辑，即获取transition

批量定义转换规则

自反转换 machine.add_transition(dest = “=”)

内部转换 machine.add_transition(dest=None)

顺序转换

队列转换

满足条件才转换状态

转换状态前后回调

状态机转换状态前后回调

异常处理

回调协议

回调执行顺序

传递参数

初始化模式

日志

扩展

常用参数

Machine - 初始化状态机对象

Machine.add_transition() - 给状态机添加转换方法

所有评论(0)

Jenrey

第一种：states的`on_enter`、`on_exit`

第二种：给基础类增加动态方法：`on_enter_{stateName}`、`on_exit_{stateName}`

第三种：transition的回调`before`属性、`after`属性