Как управлять переходами состояний в python fsm с помощью библиотеки переходов
Я пытаюсь реализовать конечный автомат, используя модуль переходов. 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()
Спасибо!
2 ответа
Эта функция была запрошена ранее (см. Эту проблему). Как вы также можете видеть, кто-то работает над этим. Он может открыть запрос на получение в ближайшем будущем. Я не проверял его изменения, но обязательно сделаю, когда это произойдет.
На данный момент вы можете позволить вашей модели обрабатывать проверки условий и комбинировать их с упорядоченными переходами, чтобы избавить от необходимости частого обновления 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
также регистрирует эти события.
Я не понял, как добавить это в модуль переходов, однако я сделал это для автоматического перехода на основе условий:
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()