python transitions库 表示状态机

  • Post author:
  • Post category:python




一、综述

在生活中,常常会遇到状态的转换。比如水,在低温0度以下会结冰。用一个状态转换图就是

在这里插入图片描述

此等例子比比皆是,这就是最基本的状态转换图。此图应用广泛,比如工业制造领域跟踪设备的状态。本节内容就是用python去模拟各种状态的转换过程。



二、准备



1. python解释器



2. 安装transition库

pip install transitions



3. 源码网址

https://pypi.org/project/transitions/



三、状态机的使用



1.

以下是一个综合代码,不过不用着急,我会慢慢讲解的。主要体现一下整体性

from transitions import Machine
from transitions.extensions.states import add_state_features, Timeout, Tags
import random

TASK_EXE_TIME = 0.1  # 任务执行时间 单位:秒
PROBABILITY = 0.8 # 扫描成功的概率值


@add_state_features(Tags, Timeout)
class CustomStateMachine(Machine):
    pass

class PLC(object):
    """plc状态机"""
    # 'initial'  初始态,一片空白
    # 'load'     物料到位
    # 'scanfinish' 扫描完成
    # 'exe_task'   执行任务
    # 'taskfinish'  任务完成
    # 'unload'  plc出站完成状态
    states = ['initial', 'load', 'scanfinish', {'name': 'exe_task', 'timeout': TASK_EXE_TIME, 'on_timeout': 'go'},
              'taskfinish', 'unload']
    transitions = [
        ['supplies_ready', 'initial', 'load'],
        {'trigger': 'scan', 'source': 'load', 'dest': 'scanfinish',
         'conditions': 'is_scan_success'},
        ['manual_scan', 'load', 'scanfinish'],
        ['task_release_finish', 'scanfinish', 'exe_task'],
        {'trigger': 'mes_finish', 'source': 'taskfinish', 'dest': 'unload', 'before': 'read_ps'},
        ['reset_signal','unload', 'initial']
    ]

    def __init__(self, plc_name):
        self.plc_name = plc_name

        # Initialize the state machine
        self.machine = CustomStateMachine(model=self, states=PLC.states, initial='initial', transitions=PLC.transitions)

    @property
    def is_scan_success(self):
        """
        以一定的概率返回扫描成功还是失败,成功的概率,自定义
        :return: True ,False
        """
        return random.random() <= PROBABILITY

    def go(self):
        """超时,从exe_task,自动转换到taskfinish"""
        self.to_taskfinish()

    def read_ps(self):
        print("读取P+S码")



2. 快速掌握

from transitions import Machine # 导入状态机
import random
class Water():
	"""定义一个水类内部可空白"""
	pass
water = Water() 

states = ['gas', 'liquid', 'solid'] # 定义状态 气态、液态、固态

# 状态机的模型设为water对象, 状态设为气态、液态、固态,初始状态为液态
machine = Machine(model=water, states=states, initial='liquid')

# 设定转换过程
# trigger为触发条件(本质是一个函数的引用,该函数自动生成) , source是原状态 ,dest为目标状态。应把各种状态转换过程完善。这里仅举几例。
machine.add_transition(trigger='heat', source='liquid', dest='gas')
machine.add_transition(trigger='cool', source='liquid', dest='solid')
machine.add_transition(trigger='sublimation', source='solid', dest='gas')

# add_transition函数还指定before, after参数,分别表示状态装换之前和之后执行的函数
# 还可以指定 conditions=,表示当条件函数返回为True时,才执行状态转换。

# water.state查询水的状态
# water.heat() 将液态转换为气态



3. 详细内容



1. States

状态可根据自己的需求定义

from transitions import State
states = [
    State(name='solid'),
    'liquid',
    { 'name': 'gas'}
    ]



2. 回调函数

states = [
    State(name='solid', on_exit=['say_goodbye']),
    'liquid',
    { 'name': 'gas' }
    ]

machine.on_enter_gas('say_hello')

两种方式为状态添加回调函数。

1 通过on_exit字段,表示离开该状态时,执行该函数

2 machine.on_enter_gas(‘say_hello’) 通过函数方式添加



4. Enumerations

transitions = [['heat','liquid', 'gas'],
               ['cool', 'liquid','solid'],
               ['sublimation', 'solid','gas']

通过枚举出所有的状态,若需要精确定义,可将内层列表部分改成字典来指定。



5. Logging

Transitions includes very rudimentary logging capabilities. A number of events – namely, state changes, transition triggers, and conditional checks – are logged as INFO-level events using the standard Python logging module. This means you can easily configure logging to standard output in a script:

转换包括非常基本的日志功能。使用标准Python日志模块将许多事件(即状态更改、转换触发器和条件检查)记录为info级事件。这意味着你可以很容易地配置日志到标准输出在一个脚本:

# Set up logging; The basic log level will be DEBUG
import logging
logging.basicConfig(level=logging.DEBUG)
# Set transitions' log level to INFO; DEBUG messages will be omitted
logging.getLogger('transitions').setLevel(logging.INFO)

# Business as usual
machine = Machine(states=states, transitions=transitions, initial='solid')



6. Adding features to states 添加新特性

from time import sleep
from transitions import Machine
from transitions.extensions.states import add_state_features, Tags, Timeout

# 定义一个新的机器,继承Machine,并添加Tags, Timeout的装饰器
@add_state_features(Tags, Timeout)
class CustomStateMachine(Machine):
    pass

# 有了新机器,就可以在states中添加timeout和on_timeout字段。
# timeout 指定超时时间,单位是秒
# on_timeout 指定时间超时后应该执行什么函数
states = ['initial', 'load', 'scanfinish', {'name': 'exe_task', 'timeout': 2, 'on_timeout': 'go'}, 'taskfinish', 'unload']

# 类似的 还可以指定标签值。标签值就是:若在当前状态,则machine.get_state('preparing').is_busy 会返回True,否则,False
states = [{'name': 'preparing', 'tags': ['home', 'busy']},
          {'name': 'waiting', 'timeout': 1, 'on_timeout': 'go'},
machine =CustomStateMachine(model=hero,states=states,transitions=transitions, initial='preparing')



7. 更多内容请查阅源网址,见2.3。

有问题可私信。



版权声明:本文为xin_IT_able原创文章,遵循 CC 4.0 BY-SA 版权协议,转载请附上原文出处链接和本声明。