python 初探狀態機transiti
- 2020 年 1 月 8 日
- 筆記
偶然接觸一個python的關於狀態機的庫，簡單demo如下：# coding=utf-8 ############################################################## # 目標："solid", "liquid", "gas", "plasma" 四種互相轉換... # 環境： pip install transitions # 這個庫簡直碉堡了... ############################################################## from transitions import Machine class Matter(object): pass model = Matter() # 定義狀態 states = ["solid", "liquid", "gas", "plasma"] # 定義狀態轉移 transitions = [ {'trigger':'melt', 'source': 'solid', 'dest': 'liquid'}, {'trigger':'evaporate', 'source': 'solid', 'dest': 'gas'}, {'trigger':'sublimate', 'source': 'liquid', 'dest': 'gas'}, {'trigger':'ionize', 'source': 'gas', 'dest': 'plasma'}, ] # 初始化 machine = Machine(model=model, states=states, transitions=transitions, initial='solid') print model.state model.melt() print model.state對源程式碼一探究竟。發現看不懂的地方甚多，那就分解一步步去解析吧。 Machine —————> States = {'s1':s1, 's2':s2, 's3': s3} —————> Events = {e1, e2} e1—————>Transition = [t1,t2,t3...] —————> models = [m1, m2] m1 —————> to_s1(), to_s2()大概就是Machine這個類裡面維護比較重要的三個元素，States={}，Events={}, models= []，接下來先看怎麼填充這3個元素。 0、
pip install transitions==0.6.1
python == 2.7.10
1、 按照demo來說，我們傳入的States和transitions：
States = ["solid", "liquid", "gas", "plasma"] transitions = [ {'trigger':'melt', 'source': 'solid', 'dest': 'liquid'}, {'trigger':'evaporate', 'source': 'solid', 'dest': 'gas'}, {'trigger':'sublimate', 'source': 'liquid', 'dest': 'gas'}, {'trigger':'ionize', 'source': 'gas', 'dest': 'plasma'}, ]
2、 States Machine的構造函數中，傳入了一個狀態list，觸發self.add_states(states)：
if states is not None: self.add_states(states)
然後在add_states()中，用字元串state分別創建一個State實例，然後把這個實例加入到字典self.states中，key就是字元串state
for state in states: if isinstance(state, string_types): state = self._create_state(state, on_enter=on_enter, on_exit=on_exit, ignore_invalid_triggers=ignore, **kwargs) self.states[state.name] = state
所以到這裡我們的Machine實例字典參數如下：
self.states = { 'solid': <State('solid')@4516320272>, 'liquid': <State('liquid')@4516320336>, 'gas': <State('gas')@4516320400>, 'plasma': <State('plasma')@4516320464> }
同樣在add_states()中，最後來一句：
if self.auto_transitions: for s in self.states.keys(): self.add_transition('to_%s' % s, self.wildcard_all, s)
這就很有趣了，每一個狀態，都觸發一次add_transition()，比如我們四個狀態中的solid，觸發時名字為to_solid，self.wildcard_all就是*號，表示source源狀態，solid表示desc目標狀態。那麼看下add_transiton()給我們帶來什麼呢？
if trigger not in self.events: self.events[trigger] = self._create_event(trigger, self)
因為to_solid一定不在self.events={}中，所以if條件為True，先是用trigger=to_solid創建一個Event實例，然後加到self.events這個字典中。 所以到這裡我們的Machine實例字典參數如下：
self.events = { 'to_liquid': <Event('to_liquid')@4516320592>, 'to_gas': <Event('to_gas')@4516320912>, 'to_plasma': <Event('to_plasma')@4516321232>, 'to_solid': <Event('to_solid')@4516320528> }
繼續往下走，add_transition()接下來執行：
if isinstance(source, string_types): source = list(self.states.keys()) if source == self.wildcard_all else [source] else: source = [s.name if self._has_state(s) else s for s in listify(source)]
這是算出source，如果我們進來的參數是self.wildcard_all，那麼source=["solid", "liquid", "gas", "plasma"]，否則source=["solid"]，add_transition()接下來執行：
for s in source: d = s if dest == self.wildcard_same else dest if self._has_state(d): d = d.name t = self._create_transition(s, d, conditions, unless, before, after, prepare, **kwargs) self.events[trigger].add_transition(t)
這是對於每一個source，我們都算出一對Source2Dest如:("solid", "solid")，用這一對去生成一個Transition實例t，然後self.events["to_solid"]這個event，添加這個t到event.transitions中。 所以到了這裡我們的Machine實例參數如下:
self.events["to_solid"].transitions = { 'solid': [<Transition('solid', 'solid')@4516320656>], 'plasma': [<Transition('plasma', 'solid')@4516320848>], 'gas': [<Transition('gas', 'solid')@4516320784>], 'liquid': [<Transition('liquid', 'solid')@4516320720>] } self.events["to_liquid"].transitions = { 'solid': [<Transition('solid', 'liquid')@4516320656>], 'plasma': [<Transition('plasma', 'liquid')@4516320848>], 'gas': [<Transition('gas', 'liquid')@4516320784>], 'liquid': [<Transition('liquid', 'liquid')@4516320720>] } self.events["to_gas"].transitions = { 'solid': [<Transition('solid', 'gas')@4516320656>], 'plasma': [<Transition('plasma', 'gas')@4516320848>], 'gas': [<Transition('gas', 'gas')@4516320784>], 'liquid': [<Transition('liquid', 'gas')@4516320720>] } self.events["to_plasma"].transitions = { 'solid': [<Transition('solid', 'plasma')@4516320656>], 'plasma': [<Transition('plasma', 'plasma')@4516320848>], 'gas': [<Transition('gas', 'plasma')@4516320784>], 'liquid': [<Transition('liquid', 'plasma')@4516320720>] }
3、 Events Machine的構造函數中，傳入了transitions的list，觸發了add_transitions()：
if transitions is not None: self.add_transitions(transitions)
查看下add_transitions()發生了什麼：
for t in listify(transitions): if isinstance(t, list): self.add_transition(*t)
好的，是分別觸發add_transition()函數，那麼根據1中分析，我們自定義傳入了4個trigger，那麼此時此刻的self.events如下：
self.events = { 'to_liquid': <Event('to_liquid')@4516320592>, 'to_gas': <Event('to_gas')@4516320912>, 'to_plasma': <Event('to_plasma')@4516321232>, 'to_solid': <Event('to_solid')@4516320528>， 'sublimate': <Event('sublimate')@4516446608>, 'evaporate': <Event('evaporate')@4516446480>, 'melt': <Event('melt')@4516446416>, 'ionize': <Event('ionize')@4516446736> }
然後除了to_liquid, to_gas, to_plasma, to_solid，其他的參數如下：
self.events["melt"].transitions = { 'solid': [<Transition('solid', 'liquid')@4438964880>] } self.events["ionize"].transitions = { 'gas': [<Transition('gas', 'plasma')@4438965264>] } self.events["sublimate"].transitions = { 'liquid': [<Transition('liquid', 'gas')@4438965136>] } self.events["evaporate"].transitions = { 'solid': [<Transition('solid', 'gas')@4438965008>] }
4、 model Machine的構造函數中，傳入了model這list，觸發了add_model()：
if model: self.add_model(model)
在add_model()中，先是遍歷model（沒錯，它被變成一個list，如果單個元素，就是[model]）。這裡我們就只看一個model。
if hasattr(model, 'trigger'): logger.warning("%sModel already contains an attribute 'trigger'. Skip method binding ", self.name) else: model.trigger = partial(get_trigger, model)
這段程式碼是先給model加一個函數trigger()，以後就能夠用model.trigger()調用啦。 實際上調用的是偏函數get_trigger(model, trigger_name).
def get_trigger(model, trigger_name, *args, **kwargs): func = getattr(model, trigger_name, None) if func: return func(*args, **kwargs) raise AttributeError("Model has no trigger named '%s'" % trigger_name)
有意思，所以以後我們就能用model.trigger("melt")，來調用某一個觸發函數。
接著繼續在add_model()中：
for trigger, _ in self.events.items(): self._add_trigger_to_model(trigger, model)
這個是遍歷我們Machine實例中的self.events的鍵值對，拿到每個觸發名字：to_liquid, to_solid, to_gas, to_plasma, melt, ionize, sublimate, evaporate，遍歷調用self._add_trigger_to_model(trigger, model)：
def _add_trigger_to_model(self, trigger, model): trig_func = partial(self.events[trigger].trigger, model) setattr(model, trigger, trig_func)
其實這個還是給model綁定偏函數，以後我們就能這麼調用model.melt()。 所以到了這裡我們Machine實例關於model參數如下：
model.trigger("xxx") model.to_liquid() model.to_solid() model.to_gas() model.to_plasma() model.melt() model.mionize() model.sublimate() model.evaporate()
接著繼續在add_model()中：
for _, state in self.states.items(): self._add_model_to_state(state, model)
和上面差不多，遍歷調用_add_model_to_state(state, model)
def _add_model_to_state(self, state, model): setattr(model, 'is_%s' % state.name, partial(self.is_state, state.name, model))
所以到了這裡我們Machine實例關於model參數如下：
model.is_solid model.is_liquid model.is_gas model.is_plasms model.state # 這個是model.setState()把初始化狀態如solid設置過來生成的屬性
好了，我們知道了model現在多了很多函數和屬性。 好的，現在我們具體來看看調用model.melt()到底發生了什麼呢？它其實是調用self.events['melt'].trigger(model)， 好的，看看self.events['melt'].trigger(model)裡面是啥？
def trigger(self, model, *args, **kwargs): f = partial(self._trigger, model, *args, **kwargs) return self.machine._process(f)
好的，原來是調用self.events['melt']._trigger(model)，那麼這個實際處理函數_trigger()裡面又是啥呢？
state = self.machine.get_state(model.state) if state.name not in self.transitions: msg = "%sCan't trigger event %s from state %s!" % (self.machine.name, self.name, state.name) if state.ignore_invalid_triggers: logger.warning(msg) return False else: raise MachineError(msg)
第一句就先找出當前model的狀態是什麼：<State('solid')@4372239248>. 然後判斷state.name 在不在 當前event的transitions中：{'solid': [<Transition('solid', 'liquid')@4372347728>]} 好吧，確實存在，那麼就不用if語句塊啦。_trigger()繼續往下：
event_data = EventData(state, self, self.machine, model, args=args, kwargs=kwargs)
這個生成一個event_data實例，顧名思義，它就是一放數據的，裡面放了當前machine，當前state，當前event,當前的model，transitions，執行transition()的結果result，當然不僅僅是放數據而已，它還會update()，更新machine狀態。好吧，_trigger()繼續往下：
for t in self.transitions[state.name]: event_data.transition = t if t.execute(event_data): event_data.result = True break
如上我們的self.transitions['solid']只有一個元素t：{'solid': [<Transition('solid', 'liquid')@4372347728>]}，把這個元素t賦值給event_data後，開始執行t,等會執行結果還是要放到event_data中呢。 總結下，到這裡我們執行model.melt()，其實就是執行self.events['melt'].transitions['solid'].execute(). 好的，那我們繼續看看這個t.execute(event_data)怎麼個執行法。 因為我們沒帶什麼prepare, conditions等高級參數，所以我們把這個execute()簡化成：
def execute(self, event_data): self._change_state(event_data) return True
好吧，繼續看看transition._change_state()，顧名思義，它就是專門處理狀態的更新：
def _change_state(self, event_data): event_data.machine.get_state(self.source).exit(event_data) event_data.machine.set_state(self.dest, event_data.model) event_data.update(event_data.model) event_data.machine.get_state(self.dest).enter(event_data)
沒什麼好說的，就是單純的處理前狀態"solid"的exit()，把當前狀態置成desc的狀態="liquid", 返回True。 總結下，Machine就是有兩個self.events={}, self.states=[]輔助圍著self.model轉，維護好這個model的狀態，即可！ 這個transitions庫大量運用偏函數呀和動態生成屬性函數什麼的，不過這也正常，畢竟從demo上看來，demo越簡單，底下的操作越複雜。 以上
