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

Question

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

在项目文档中，所有示例都通过在命令提示符处键入函数调用来完成状态。以从在转变文档的第一示例的片断，的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')   

为了运行状态机，只需运行“运行'功能，例如：

sm.run()