LoginSignup
5
3

More than 5 years have passed since last update.

Pythonの状態遷移パッケージ(transitions)を理解する【コールバック編2】

Last updated at Posted at 2019-03-16

Pythonで状態遷移を実装したり動作確認をしたい方に、Pythonの状態遷移パッケージ「transitions」の使い方を説明していきたいと思います。
transitionsはPythonで状態遷移を実現するためのパッケージで、状態遷移そのものは組込みとか制御などでよく使われるものですが、それをPythonで実現したい場合にこのパッケージが有用かと思います。
その他、transitionsの概要やインストール方法、当記事で作成している状態遷移図といったグラフ表示機能の導入(GraphMachine)や設定については準備編の記事を参照頂けたらと思います。

今回の話

コールバック編1」ではコールバックの種類や優先順位について説明しましたが、今回は主に以下の内容となります。

  • コールバックへのデータ(引数)の渡し方
  • コールバック内でのステートマシン各種情報の取得方法
  • キュー実行(Queued transitions)

コールバック関数に外部からデータを渡してあれこれ処理させたい時や、コールバック内でトリガーイベントを起こす場合に当記事の内容が参考になるかと思います。
以下、順に説明していきたいと思います。

コールバックへのデータの渡し方

コールバック内にデータ渡す場合、引数で渡すことになりますが以下の設定が必要になります。

  • Machine定義にsend_event=Trueを追加
  • クラス内コールバックメソッドにevent引数を宣言(クラス内の状態遷移に絡む全コールバックメソッドに必要)
  • トリガーイベントを引き起こす際、メソッドに引数値を渡す

さて、早速ですがコールバックへデータを渡すステートマシンの定義例について示します。

引数をとるコールバックの定義例
from transitions import Machine

states = ['A', 'B']   #状態の定義
transitions = [
    {'trigger':'fromAtoB', 'source':'A', 'dest':'B', 'after':'action_after'},
    {'trigger':'fromBtoA', 'source':'B', 'dest':'A'},
]

class Model(object):
    def action_after(self, event):
        print('第一引数 : {}'.format(event.args[0]))
        print('第二引数 : {}'.format(event.args[1]))
        print('辞書形式引数 :{}'.format(event.kwargs.get('data')))

model = Model()
machine = Machine(model=model, states=states, transitions=transitions, initial=states[0], 
                  auto_transitions=False, ordered_transitions=False, send_event=True)

ステートマシンとしては状態AとBをもち、fromAtoBトリガーイベントで状態AからBに、fromBtoAトリガーイベントで状態BからAに遷移するものになります。
image.png
またfromAtoBトリガーイベントにafterコールバックとしてaction_afterが設定されています。
従ってfromAtoBトリガーイベントが起こった際にaction_afterコールバックが呼ばれますが、このfromAtoBトリガーイベントにデータを引数で渡すことになります。

ともあれ上記定義例に対し、実際にデータを与えトリガーイベントを起こしてみたいと思います。

引数をとるコールバックの動作例
>>> model.state                 #現在状態の確認
'A'
>>> model.fromAtoB(100, '200', data='test')
第一引数 : 100
第二引数 : 200
辞書形式引数 :test
True
>>> model.state                 #遷移後状態の確認
'B'

image.png
このようにトリガーイベントであるfromAtoBの引数に値を指定する事でコールバックであるaction_afterに各引数の値が渡されます。
渡し方は通常のメソッドと同じように引数キーワードを指定しない場合はevent.argsにタプル形式で渡されます。
引数キーワード(例では'data'というキーワードが付いている)を指定した場合はevent.kwargsに辞書形式で渡されます。

注意点として、Machineクラス定義時にsend_event=Trueにすることでこのようにデータを渡すことが可能になるので、付け忘れに注意ください。
また、event引数はトリガーイベントに紐付く全てのコールバックに同様に渡されることになるので、引数を利用しないコールバックであってもevent引数の宣言は必要になります。
なおトリガーイベントを起こした際にコールバックを呼ばない、もしくはデータを利用としないコールバックの場合(例えば当定義において状態BでfromBtoAトリガーイベントを起こす場合)、トリガーイベントメソッドに引数を入れる必要はありません。

引数を利用しないトリガーイベントの動作例
>>> model.state                 #現在状態の確認
'B'
>>> model.fromBtoA()
True
>>> model.state                 #遷移後状態の確認
'A'

image.png

ちなみにトリガーイベントを起こす場合、イベント名(上記ではfromAtoB)で起こしましたがtriggerメソッドでも起こすことが可能です。
その場合におけるデータの渡し方は以下の通りです。

model.trigger('fromAtoB', 100, '200', data='test')

これは動作例のfromAtoBメソッドでトリガーイベントを与えた時と同じ結果になります。

データ渡しの応用例

引数にデータを渡すコードの応用として、例えばコールバックに渡されたデータを元にガード判定を行い、遷移するかしないかを制御することができます。
例として、コールバックに入力された温度情報(temp)が一定温度以上(LIMIT_TEMP)より高ければhot状態へ、それ以下であればnormal状態へ遷移するステートマシンは以下のように実現できます。

引数をとるコールバックの応用例(温度による状態変化)
from transitions import Machine

states = ['normal', 'hot']   #状態の定義
transitions = [
    {'trigger':'change_temp', 'source':'normal', 'dest': 'hot',    'prepare':'set_temp',  'conditions':'is_hot'},
    {'trigger':'change_temp', 'source':'hot',    'dest': 'normal', 'prepare':'set_temp',  'unless':'is_hot'},
]

class Model(object):
    def __init__(self):
        self.temp = 25
        self.LIMIT_TEMP = 45        # 閾値温度

    # 温度設定
    def set_temp(self, event):
        self.temp = event.kwargs.get('temp')

    # 高温判定
    def is_hot(self, event):
        return (self.temp > self.LIMIT_TEMP)

model = Model()
machine = Machine(model=model, states=states, transitions=transitions, initial=states[0], 
                  auto_transitions=False, ordered_transitions=False, send_event=True)   

こちらを図示すると以下の通りです。同じトリガーイベントを用いて遷移する形なりますが、それぞれガード判定がついています。
image.png
上記の状態遷移の定義について実際にデータを与えて動きを見てみましょう。
まずは閾値(45℃)以下のデータをコールバックへ渡した場合です。

応用例の動作(状態変化なし)
>>> model.state                 #現在状態の確認
'normal'
>>> model.change_temp(temp=40)  #temp=40℃としてトリガーイベント
False
>>> model.state                 #イベント後状態の確認
'normal'

上記の通り、トリガーイベントを与えても状態が初期状態(normal)から遷移していません。
内部の動きとしては、状態がnormalなのでトリガーイベント→prepareコールバック(set_temp)の実行→condititionsコールバック(is_hot)の実行という順で動作します。
conditionsは真のガード判定なので、is_hotでTrueが返らないと状態遷移を阻止します。したがって現在の状態(normal)が維持された形になります。

次に閾値温度を超えるデータを与えた例を見てみます。

応用例の動作(状態変化あり:normal→hot)
>>> model.state                 #現在状態の確認
'normal'
>>> model.change_temp(temp=50)  #temp=50℃としてトリガーイベント
True
>>> model.state                 #イベント後状態の確認
'hot'

image.png
上記はコールバックの呼び出し順としては同じになりますが、is_hotの戻り値がTrueになるので遷移が許可されhot状態に遷移しています。
さらにhot状態でtemp=0のデータをいれてトリガーイベントを起こした場合が以下になります。

応用例コードの動作(状態変化あり:hot→normal)
>>> model.state                 #現在状態の確認
'hot'
>>> model.change_temp(temp=0)   #temp=0℃としてトリガーイベント
True
>>> model.state                 #イベント後状態の確認
'normal'

image.png
こちらはunless(偽のガード判定)によりis_hot=Falseで遷移する形になり、hot状態からnormal状態に遷移しています。
このようにデータを渡すことで遷移を制御することが可能になります。

コールバック内でのステートマシン各種情報の取得方法

ここではコールバックの受け取るデータ(event引数)について詳細に説明します。

前述の通り、各コールバックはMachineクラスへのsend_event=True設定とevent引数を用いて、外部から(引数経由で)データを取得できることを示しました。
一方で、この各コールバックに取得されるevent引数は、transitions内部ではEventDataクラスで実装されており、外部からのデータだけでなくステートマシン内部の情報も取得することができます。
以下にevent引数で取得できるメンバの一覧を示します。

メンバ 型/クラス 内容
args tuple トリガーイベントに渡された引数(タプル形式)
kwargs dict トリガーイベントに渡された引数(辞書形式)
state State 現在の状態(コールバックが実施されたState)
event Event コールバックを引き起こしたイベント
transition Transitions トリガーイベントによる遷移情報
model Model コールバックが登録されているモデル
machine Machine 上記モデルが登録されているMachine
error Exception トリガーイベント実施時に発生した例外
result bool 遷移成否(True:遷移成功, False:それ以外)

event引数のうち、argsとkwargsは前述のデータ渡しの内容になります。
ひとまずevent引数で取得できるメンバ全てを表示してみます。

event引数で取得できるメンバの定義例
from transitions import Machine

states = ['A', 'B']   #状態の定義
transitions = [
    {'trigger':'fromAtoB', 'source':'A', 'dest':'B', 'after':'action_after'},
]

class Model(object):
    def action_after(self, event):
        print('args:',      event.args)
        print('kwargs:',    event.kwargs)
        print('state:',     event.state)
        print('event:',     event.event)
        print('machine:',   event.machine)
        print('model:',     event.model)
        print('transition:',event.transition)
        print('error:',     event.error)
        print('result:',    event.result)

model = Model()
machine = Machine(model=model, states=states, transitions=transitions, initial=states[0], 
                  auto_transitions=False, ordered_transitions=False, send_event=True)

上記定義に対し引数を与えてトリガーイベントを起こしてみましょう。

event引数で取得できるメンバの動作例
>>> model.trigger('fromAtoB', 100, '200', data='test')
args: (100, '200')
kwargs: {'data': 'test'}
state: <State('B')@1609728922064>
event: <Event('fromAtoB')@1609728921896>
machine: <transitions.core.Machine object at 0x00000176CB521550>
model: <__main__.Model object at 0x00000176CB38B4A8>
transition: <Transition('A', 'B')@1609728921784>
error: None
result: False
True

このような形でステートマシン内部の情報がevent引数で取得することができました。

event引数の補足1(event引数で取得できるクラスについて)

このとおりevent引数を用いてトリガーイベントを示すEventや遷移情報を示すTransitions、およびMachineなどはクラスインスタンスとして渡されており、このままでは使いにくいので各クラスのメンバを使って情報を取得することになります。

以下は上記サンプルに加え、str型で各内部情報を取得する場合のサンプル例になります。

event引数で取得したstateの定義例
from transitions import Machine

states = ['A', 'B']   #状態の定義
transitions = [
    {'trigger':'fromAtoB', 'source':'A', 'dest':'B', 'before':'action_after'},
]

class Model(object):
    def __init__(self, name=''):
        self.name = name

    def action_after(self, event):
        print("state(name): {}".format(event.state.name))
        print("event(name): {}".format(event.event.name))
        print("machine(name): {}".format(event.machine.name))
        print("model(name): {}".format(event.model.name))
        print("transition: {}->{}".format(event.transition.source, event.transition.dest))

model = Model('my_model')
machine = Machine(model=model, states=states, transitions=transitions, initial=states[0], 
                  auto_transitions=False, ordered_transitions=False, send_event=True,
                  name='my_machine')

args、kwargs、error、resultについては割愛しています。
これも同様にイベントを起こして結果を表示すると以下のようになります。


```Python
>>> model.trigger('fromAtoB', 100, '200', data='test')
state(name): A
event(name): fromAtoB
machine(name): my_machine: 
model(name): my_model
transition: A->B
True

注意点として、Modelについては元々ユーザクラスでありnameメンバは存在しないので、Modelクラス定義時にコンストラクタ内でself.nameに値を設定しています。
Machineに関してはname引数というものがあるので、これを指定すればMachine名を取得できるようになります。
メンバtransitionは遷移元(source)と遷移先(dest)がメンバとして保持されているので、これをそれぞれ取得している形になります。
Transitionクラスの詳細は遷移編1のTransitionの構成メンバをご確認下さい。
なお状態(state)についてはevent引数を用いなくてもself.stateで現在の状態をstr型として取得できますが、stateに紐付くコールバック(on_enter等)情報等を取得する場合はevent.stateから得る必要があります。

event引数の補足2(event.resultについて)

event引数では遷移の結果であるresultを取得できるという旨が公式のコードにも書いてありますが、実際の動作を以下のコードで確認してみます。

event.resultの結果比較の定義例
from transitions import Machine

states = ['A', 'B']
transitions = [
    {'trigger':'fromAtoB', 'source':'A', 'dest':'B', 'after':'after', 'before':'before'},
]

class Model(object):
    def before(self, event): print('result(before):', event.result)
    def after(self, event): print('result(after):', event.result)
    def finalize(self, event): print('result(finalize):', event.result)

model = Model()
machine = Machine(model=model, states=states, transitions=transitions, initial=states[0], 
                  auto_transitions=False, ordered_transitions=False, send_event=True,
                  finalize_event='finalize')

これを動作させてみます。状態AからBへの遷移はガード判定などがないので、遷移は成功します。
従って期待値としては全てのコールバックのevent.resultはTrueになることです。

event.resultの結果比較の動作
>>> model.fromAtoB()
result(before): False
result(after): False
result(finalize): True
True

結果をみると分かる通り、実際に遷移が成功しているのに関わらず遷移成否を示すresultはfinalize_eventで定義したコールバック上でのみTrueとなっています。
これはバグの可能性もありますが、現在のところこのような動作になっているのでevent.resultにより遷移成否を用いて処理を変えたい場合はfinalize_eventのコールバックで実施した方が良いでしょう。

event引数の補足3(event.errorについて)

この通りselfでは取得できないステートマシンの内部情報をevent引数で取得できることがわかりますが、応用として別のコールバックで生じた例外をerrorとして取得することもできます。
以下、公式のTutorialを少々修正したものになります。

from transitions import Machine

states = ['A', 'B']
transitions = [
    {'trigger':'fromAtoB', 'source':'A', 'dest':'B', 'after':'action_after'},
]

class Model(object):
    def action_after(self, event):
        raise ValueError('Error Test')

    def action_finalize(self, event): 
        print('error:', event.error)
        print('result:', event.result)

model = Model()
machine = Machine(model=model, states=states, transitions=transitions, initial=states[0], 
                  auto_transitions=False, ordered_transitions=False, send_event=True,
                  finalize_event='action_finalize')

これを動作させてみます。
トリガーイベントを起こした際に例外で止まってしまうので念のためtry/except内で実施しています(例なのでexceptでpassしているところには目をつぶって下さいね・・・)

>>> try:
...     model.fromAtoB()
... except ValueError:
...     pass
error: Error Test
result: False

このとおりaction_afterコールバックで発生した例外をキャッチしてaction_finalize内のevent.errorが取得している事が確認できます。この際、event.errorは発生した例外のクラスインスタンスが渡され、この場合はValueErrorとなります。

キュー実行(Queued transitions)

最後はキュー実行について説明します。
これは「コールバック内でトリガーイベントを発生させて状態遷移させる」場合に重要な考えになりますので、そのような実装をする場合は抑えておいた方が良い内容です。

まずはコールバック内でトリガーイベントを発生させ状態遷移させる場合の定義例を見て下さい。

Queuedなし時場合のコールバック(デフォルト動作)定義例
from transitions import Machine, State

states = [State(name='A'),
          State(name='B', on_enter='enter_B'),      #状態Bに入る時にenter_Bコールバック
          State(name='C', on_enter='enter_C')]      #状態Cに入る時にenter_Cコールバック

transitions = [
    {'trigger':'fromAtoB', 'source':'A', 'dest':'B', 'after':'after_A'},
    {'trigger':'fromBtoC', 'source':'B', 'dest':'C'},
]

class Model(object):
    def after_A(self):
        print('after-A on state({})'.format(self.state))

    def enter_B(self):
        self.fromBtoC() #状態Bから状態Cへ

    def enter_C(self): 
        print('enter-C on state({})'.format(self.state))

model = Model()
machine = Machine(model=model, states=states, transitions=transitions, initial=states[0], 
                  auto_transitions=False, ordered_transitions=False)

上記定義は遷移に紐付くコールバック(after)と状態に紐付くコールバック(on_enter)が設定されています。
状態Aではafterが設定されていることから、状態Aから遷移後にafter_Aコールバックが実施されます。一方で状態Bと状態Cにはon_enterコールバックが設定されているので、それぞれの状態に入った際に実施されることになります。さらに状態Bのon_enterにはfromBtoCトリガーイベントが実施されているので、状態AでfromAtoBトリガーイベントを起こすと自動的に状態Cまで遷移する例になります。
つまりはこれらコールバックの実施順に問題が生じるのが今回の話になります。

ともあれ分かりにくいのでこれを図示すると以下の通りになります。
image.png
なお、コールバック編1でも示したようにコールバックには優先順位があります。

遷移後の実施順でいうとafterコールバック(図中の青字)よりもon_enterコールバック(図中の赤字)の方が遷移が先なので、状態AでfromAtoBトリガーイベントを実施した際はまずenter_Bが実施され、その後after_Aコールバックが実施されて状態Cに遷移するというのが想定しうる動きです。
想定しうるコールバックの順番を並べると

  1. enter_Bが状態Bで実施 (表示なし)
  2. after_Aが状態Bで実施 (enter-A on state(B)と表示される)
  3. enter_Cが状態Cで実施 (enter-C on state(C)と表示される)

という形になるかと思います。
では実際にデフォルト設定(キュー実行設定なし)で動かしてみます。

キュー実行設定なし時の実行
>>> model.fromAtoB()
enter-C on state(C)
after-A on state(C)
True

ちょっと話が違うじゃないか!と言いたくなりますが、これこそが「キュー実行設定なし」によるコールバックの実施順序問題になります。
上記例のコールバック実施順を図示すると以下のようなイメージになります。
image.png
これはtransitionsパッケージの思想として、デフォルトでは発生したイベントを先に処理するという思想だからのようです。
つまりenter_Bコールバックで発生したfromBtoCイベントがafter_Aコールバックよりも先に処理された事で状態Cに移り、加えてenter_Cコールバックが先に処理されるようです。

このイベントとコールバックの優先順位については変更することができ、これを想定通り動作(状態Bでenter_Bおよびafter_Aが実施された後、状態Cに移りenter_Cが実施)させる為にはMachineクラスでキュー実行設定を有効(queued=True)にします。

# Machine以外は上記と同じなので割愛
machine = Machine(model=model, states=states, transitions=transitions, initial=states[0], 
                  auto_transitions=False, ordered_transitions=False, queued=True)

この設定で、同様に状態AでfromAtoBトリガーイベントをおこしてみます。

キュー実行設定あり時の実行
>>> model.fromAtoB()
after-A on state(B)
enter-C on state(C)
True

これで想定通り、各状態前後のコールバックが実施されてから状態が移るようになりました。
コールバック内でのトリガーイベントで問題が発生するという話をしましたが、正しくはキュー実行設定なしのデフォルト動作では遷移に伴うコールバックより遷移が優先されるということであり、状態に紐付くコールバックは遷移順に実行されるという点に注意下さい。つまりデフォルトでは状態に紐付くコールバック(on_enterやafter_state_change)は遷移順にコールバックされ、それらが終わった後に遷移に伴うコールバック(afterなど)が実施されるということのようです。ややこしい。

ですから基本的には(遷移と絡まないtimeoutコールバック以外の)コールバック内で遷移を伴うトリガーイベントを呼ばない、もしくはキュー実行設定を有効(queued=True)にしてコールバック内でトリガーイベントを起こすというのが安全策だと思われます。
以下キュー実行の設定あり/なし時のコールバックの実施順についてまとめておきます。

  • キュー実行設定なし(queued=False)時のコールバック実施順(デフォルト動作)
    遷移に伴うコールバック(after)より先に遷移が行われ、状態に紐付くコールバックが先に実施される。

    順番 状態 コールバック 備考
    1 B enter_B (State.on_enter) fromBtoCイベント発生
    2 C enter_C (State.on_enter) enter_Bでこれが差し込まれる
    3 C after_A (transitions.after) afterの方が優先順位低いのでココで実施
  • キュー実行設定あり(queued=True)時のコールバック実施順
    状態がAからBに遷移し、AからBの遷移に伴う一連のコールバック(enter, after)が終わってから次の状態Cに遷移している。

    順番 状態 コールバック 備考
    1 B enter_B (State.on_enter) fromBtoCイベント発生
    2 B after_A (transitions.after) afterの方が優先順位低いが状態Bで実施
    3 C enter_C (State.on_enter) 状態Bで実施されるコールバックが終わってから実施

キュー実行の応用

このようにキュー実行は混乱するので、実際のところコールバックで状態を遷移させるような複雑な実装は避けた方がよいといえると思います。
一方でキュー実行を使うと面白いことができるので一部紹介しておきます。

あまり通常の状態遷移設計では行いませんが、一旦トリガーをかけたら常に決まった順序で遷移を延々と実施したい場合があったとします。
例えばA → B → C → A → B → ・・・のような形で状態A, B, Cをループ動作させてみる場合、遷移に伴うコールバック(after)と状態に紐付くコールバック(on_enter)とを混在させる場合は以下のようにします。

from transitions import Machine, State
from time import sleep

states = [State(name='A', on_enter='fromAtoB'),     # 状態Aに入ったらB状態に行くトリガーイベントを起こす
          State(name='B', on_enter='fromBtoC'),     # 状態Bに入ったらC状態に行くトリガーイベントを起こす
          State(name='C', on_enter='fromCtoA')]     # 状態Cに入ったらA状態に行くトリガーイベントを起こす

transitions = [
    {'trigger':'fromAtoB', 'source':'A', 'dest':'B', 'after':'after_A'},
    {'trigger':'fromBtoC', 'source':'B', 'dest':'C', 'after':'after_B'},
    {'trigger':'fromCtoA', 'source':'C', 'dest':'A', 'after':'after_C'},
]

class Model(object):
    def after_A(self):
        print('after-A on state({})'.format(self.state))

    def after_B(self):
        print('after-B on state({})'.format(self.state))

    def after_C(self):
        print('after-C on state({})'.format(self.state))

    def sleep_each_state(self):
        sleep(1)            #次の状態までの待ち時間。各状態から入った際に実施される。

model = Model()
machine = Machine(model=model, states=states, transitions=transitions, initial=states[0], 
                  auto_transitions=False, ordered_transitions=False, queued=True,
                  after_state_change='sleep_each_state')

これを実施すると以下のようになります。
※このプログラムはキュー実行の問題があるためキュー実行設定あり(queue=True)にしないと動作しません。

さらにGraphMachineを使ってJupyterNotebook上で状態遷移を図示すると以下のような感じに状態が延々と繰り返される物になります。

GraphMachineでの実装例はここをクリックして下さい。

GrpahMachineのインストールはOSごとにやり方があるので、詳細は準備編をご確認ください。
以下はWindows上にGraphVizをインストールしてJupyterNotebookに貼り付けて実施したものです。
画像ファイルとして出力する場合はModelクラスでインスタンスを作る際にファイル名を引数に入れて下さい。

上記gif画像の作成に使ったGraphMachine(GraphVis使用版)
import graphviz
from IPython.display import Image, display, clear_output
from transitions import State
from transitions.extensions import GraphMachine
from time import sleep

states = [State(name='A', on_enter='fromAtoB'),
          State(name='B', on_enter='fromBtoC'),
          State(name='C', on_enter='fromCtoA')]

transitions = [
    {'trigger':'fromAtoB', 'source':'A', 'dest':'B'},
    {'trigger':'fromBtoC', 'source':'B', 'dest':'C'},
    {'trigger':'fromCtoA', 'source':'C', 'dest':'A'},
]

class Model(object):
    def __init__(self, filename=None):
        self.output = filename

    def sleep_each_state(self, *args, **kwargs):
        sleep(1)

    def action_output_graph(self, *args, **kwargs):
        dg = model.get_graph(*args, **kwargs).generate()
        dg.node_attr.update(shape='circle', height="1.2") 
        if isinstance(self.output, str):
            graphviz.Source(dg, filename=self.output, format='png').render(cleanup=True) 
        else:
            clear_output(wait=True)
            display(Image(dg.pipe(format='png'))) 

#ファイル出力する場合はModel('test')等ファイル名指定する, Notebook上で表示する場合は引数に指定なし
model = Model()
machine = GraphMachine(model=model, states=states, transitions=transitions, initial=states[0], 
                       auto_transitions=False, ordered_transitions=False,
                       title='', show_auto_transitions=False, show_conditions=True, ignore_invalid_triggers=True,
                       queued=False, after_state_change=['sleep_each_state', 'action_output_graph'])

model.action_output_graph() #初回のみ手動実施

上記で定義した後以下を実施

model.fromAtoB()

なお、あくまでJupyterNotebook上での実施になります。またPyCharmでJupyterNotebook動かすとclear_outputが上手く動かないので、ブラウザベースのJupyterNotebookで動かしてください。
あと上記はPyGraphvizではなくWindows+GraphVizを使った例になります。Linux/macOSでの表示は準備編のココをご参考に。

キュー実行応用の補足

記事を書いている時に気付いたのですが、上記の状態を遷移するものはordered_transitionを使えば更に圧縮できて、且つキュー実行設定が不要になります。

キュー実行設定を使わないループ
from transitions import Machine
from time import sleep

states = ['A', 'B', 'C']

class Model(object):
    def after_state(self, event):
        print('after-{} on state({})'.format(event.transition.source, self.state))

    def sleep_each_state(self, event):
        sleep(1)

model = Model()
machine = Machine(model=model, states=states, initial=states[0], 
                  auto_transitions=False, ordered_transitions=True, 
                  queued=False, send_event=True,
                  after_state_change=['sleep_each_state', 'after_state', 'next_state'])

動作にはnext_stateトリガーイベントを使います。

>>> model.next_state()
after-A on state(B)
after-B on state(C)
after-C on state(A)
after-A on state(B)
after-B on state(C)
after-C on state(A)
以下繰り返し

これはあくまでも状態に紐付くコールバックのみで実装されており、状態に紐付くコールバックは順序通り実施されるので、想定通りの動作をしていることになります。
ちょっと蛇足的な話になってしまいましたが、このようにqueueを必ず使わないと実現できないという話ばかりではないので、複雑な実装になってしまった場合は代替手段があるか再考してみても良いかもしれません。

まとめ

今回も長くなってしまいましたがtransitionsパッケージのコールバックに関する勘所はある程度紹介できたと思います。
event引数を使ってコールバックにデータを渡すことで、ガード判定がより実現しやすいものになったのを確認できたと思います。
キュー実行設定は非常に混乱する所が多いので、コールバックでトリガーイベントを発生させるような実装をする際はよく検討いただき、その際にこの項目などを参考いただければと思います。
当記事が少しでもお役に立てればと思います。

5
3
0

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
5
3