如何使用转换库在 python fsm 中管理状态转换

Question

我正在尝试使用transitions 模块实现状态机。 Python 版本 2.7.13 和转换是版本 0.4.4。

在项目文档中，所有示例都通过在命令提示符下键入函数调用来推进状态。以转换文档中第一个示例的 sn-p 为例，batman 的状态是通过调用命名函数 wake_up 和 work_out 来实现的

>>> batman.wake_up()
>>> batman.work_out()
>>> batman.state
'hungry'

我想让状态机自动通过以模型数据为条件的状态前进。下面的玩具示例是我想要做的，但依赖于使用虚拟函数作为指针来设置next_state。

有没有不涉及创建next_state 函数并像指针一样使用它的方法？转换文档有有序转换和条件转换但我真正想要的是有条件有序转换。

是否可以在不使用函数指针的情况下重写下面的代码？

from transitions import Machine

class AModel(object):
    def __init__(self):
        self.sv = 0  # state variable of the model

    def on_enter_sA(self):
        print "Entered sA"
        self.next_state = self.to_sB

    def on_enter_sB(self):
        print "Entered sB"
        if self.sv < 3:
            self.next_state = self.to_sB
        else:
            self.next_state = self.to_sC

    def on_enter_sC(self):
        print "Entered sC"
        if self.sv == 6:
            self.next_state = self.to_sD

    def on_enter_sD(self):
        print "Entered sD"
        self.next_state = self.to_sA

    def next_state(self):
        pass

#setup model and state machine  
model = AModel()

#init transitions model
list_of_states = ['sA','sB','sC','sD']
transitions = [
    {'trigger':'to_sA','source':'sD','dest':'sA'},
    {'trigger':'to_sB','source':'sA','dest':'sB'},
    {'trigger':'to_sC','source':'sB','dest':'sC'},
    {'trigger':'to_sD','source':'sC','dest':'sD'}
]
machine = Machine(model=model, states=list_of_states, initial='sA', 
             transitions=transitions)

model.next_state = model.to_sB #init next state pointer

#begin main
for i in range(0,8):
    print 'iter is: ' + str(i) + " -model state is:" +  model.state
    model.sv = i #update model state variable, local state logic
                 #will determine what next_state points to
    model.next_state()

谢谢！

Answer 1

以前曾请求过此功能（请参阅this issue）。如您所见，有人正在研究它。他可能会在不久的将来打开一个拉取请求。我没有审查他的更改，但当发生这种情况时肯定会这样做。

现在，您可以让您的模型处理条件检查并将其与有序转换相结合，以摆脱频繁更新 next_state 函数指针的需要。由于您只是检查索引，因此可能如下所示：

from transitions import Machine


class AModel(object):
    def __init__(self):
        self.sv = 0  # state variable of the model
        self.conditions = {  # each state 
            'sA': 0,
            'sB': 3,
            'sC': 6,
            'sD': 0,
        }

    def poll(self):
        if self.sv >= self.conditions[self.state]:
            self.next_state()


# setup model and state machine
model = AModel()

# init transitions model
list_of_states = ['sA', 'sB', 'sC', 'sD']
machine = Machine(model=model, states=list_of_states, initial='sA', ordered_transitions=True)

# begin main
for i in range(0, 10):
    print('iter is: ' + str(i) + " -model state is:" +  model.state)
    model.sv = i
    model.poll()

我假设您每次增加索引时都会轮询模型。如果是这种情况 self.sv == 6 和 self.sv >= 6 做同样的事情（sC 到 sD）。 但是，如果有意选择运算符，您可以更改模型条件检查以使用运算符值元组：

from transitions import Machine
import operator


class AModel(object):
    def __init__(self):
        self.sv = 0  # state variable of the model
        self.conditions = {  # each state
            'sA': (operator.ne, None),
            'sB': (operator.ge, 3),
            'sC': (operator.eq, 6),
            'sD': (operator.ne, None),
        }

    def poll(self):
        op, value = self.conditions[self.state]
        if op(self.sv, value):
            self.next_state()


# setup model and state machine
model = AModel()

# init transitions model
list_of_states = ['sA', 'sB', 'sC', 'sD']
machine = Machine(model=model, states=list_of_states, initial='sA', ordered_transitions=True)

# begin main
for i in range(0, 10):
    print('iter is: ' + str(i) + " -model state is:" + model.state)
    model.sv = i
    model.poll()

在这两种情况下，输出都是：

iter is: 0 -model state is:sA
iter is: 1 -model state is:sB
iter is: 2 -model state is:sB
iter is: 3 -model state is:sB
iter is: 4 -model state is:sC
iter is: 5 -model state is:sC
iter is: 6 -model state is:sC
iter is: 7 -model state is:sD
iter is: 8 -model state is:sA
iter is: 9 -model state is:sB

但是，我再次假设了一些可能是错误的：我假设如果满足条件就改变状态就足够了。这就是条件的运作方式。但也许您实际上打算在每次轮询模型时退出并进入状态。在这种情况下，您可以使用auto_transitions 并使用getattr 动态检索它们：

from transitions import Machine


class AModel(object):
    def __init__(self):
        self.sv = 0  # state variable of the model
        self.conditions = {  # each state
            'sA': 0,
            'sB': 3,
            'sC': 6,
            'sD': 0,
        }

    def poll(self):
        if self.sv >= self.conditions[self.state]:
            self.next_state()  # go to next state
        else:
            getattr(self, 'to_%s' % self.state)()  # enter current state again

    def on_enter(self):
        print('entered state %s' % self.state)

    def on_exit(self):
        print('exited state %s' % self.state)


# setup model and state machine
model = AModel()

# init transitions model
list_of_states = ['sA', 'sB', 'sC', 'sD']
machine = Machine(model=model, states=list_of_states, initial='sA',
                  ordered_transitions=True, before_state_change='on_exit',
                  after_state_change='on_enter')

# begin main
for i in range(0, 10):
    print('iter is: ' + str(i) + " -model state is:" +  model.state)
    model.sv = i
    model.poll()

为了简单起见，我添加了在每次进入或退出状态时打印消息的函数。如果您使用记录器，则不需要这样做，因为transitions 也会记录这些事件。

Answer 2

我还没有想出如何将它添加到转换模块中，但是我将它调整为根据条件自动转换：

class StateMachine:
    def __init__(self, name, states, transitions, initialState):
        self.name=name
        self.states=set()
        self.transitions=set()
        self.state=None

        for s in states:
            _s=State(s)
            self.states.add(_s)
            if self.state==None:
                self.state=_s
            elif s==initialState:
                self.state=_s

        for t in transitions:
            # self.addTransition(t)

    def addTransition(self, transition):
        name=transition[0]
        for s in self.states:
            #fromState
            if s.name==transition[1]:
                fromState=s
            #toState
            elif s.name==transition[2]:
                toState=s
        #condition
        condition=getattr(self, transition[3])
        #action     
        if len(transition)==5:
            action = getattr(self, transition[4])()
        else:
            action=self.passAction
        t=Transition(name, fromState, toState, condition, action)
        fromState.transitions.add(t)

    def passAction(self):
        print('pass!!!')

    def run(self):
        self.state=self.state.testTransitions()

    def __repr__(self):
        return self.name+':'+str(self.states)

class State:
    def __init__(self, name='state'):
        self.name=name
        self.transitions=set()

    def __repr__(self):
        return self.name+':'+str(self.transitions)

    def testTransitions(self):
        state=self
        for t in self.transitions:
            if t.condition() is True:
                t.action()
                state=t.toState
        return state

class Transition:
    def __init__(self, name, fromState, toState, condition, action): 
        self.name=name
        self.fromState=fromState
        self.toState=toState
        self.condition=condition
        self.action=action

    def __repr__(self):
        return self.name

class TestSM(StateMachine):
    def __init__(self, name, states, transitions, initialState):
        StateMachine.__init__(self, name, states, transitions, initialState)

    def testCondition(self):
        print('testCondition!!!')
        return True



states=['a', 'b', 'c']
transitions=[
    ['aTOb', 'a', 'b', 'testCondition'],
    ['bTOc', 'b', 'c', 'testCondition'],
    ['cTOa', 'c', 'a', 'testCondition'],
    ]
sm=TestSM('testSM', states, transitions, 'a')

for s in sm.states:
    print(s.name)

print('fin')        

为了运行状态机，只需运行'run'函数，例如：

sm.run()