如何使用轉換庫管理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()