transitionsはPythonで状態遷移を実現するためのパッケージですが、今回は状態遷移時に実施されるアクションを実現するために「コールバック」について紹介したいと思います。
この記事では状態遷移を実現するソフトウェア的な機構をステートマシンと呼んでいます。
#この記事の対象者と今回の内容
Pythonで状態遷移を実装したり動作確認をしたい方に、Pythonの状態遷移パッケージ「transitions」の使い方を説明していきたいと思います。状態遷移そのものは組込みとか制御などでよく使われるものですが、それをPythonで実現したい場合にこのパッケージが有用かと思います。
状態遷移を実現するだけであれば状態と遷移を定義しステートマシンを作れますが、実際はトリガーイベントが発生した際や遷移時に何か処理をすることも多いです。こういったトリガーイベント起因による処理をアクションなどといい、ソフトウェア的にはコールバック(callback)として実装されることが多いかと思います。transitionsパッケージでもこのコールバックを実装できますので詳細について説明していきたいと思います。
この状態遷移におけるアクションは必須ではありませんが、状態遷移を使って制御機構を実装する場合に非常に重要ですのでtransitionsで状態遷移を実現する場合は当記事の内容を参考に頂けたらと思います。
その他、transitionsの概要やインストール方法、当記事で作成している状態遷移図といったグラフ表示機能の導入(GraphMachine)や設定については準備編の記事を参照頂けたらと思います。
※今回の記事でも公式チュートリアルにならい遷移をtransitionsと呼んでますが、パッケージ名のtransitionsとややこしいので、当パッケージそのものを示す場合はtransitionsパッケージと明示することにします。
#transitionsにおけるコールバック
transitionsパッケージにおけるコールバックについて、簡単なサンプルコードを例に説明します。
まずはコールバックをもつステートマシンの定義になります。
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):
# afterのコールバック（自メソッド名を表示するだけ）
def action_after(self):
print('do "{}"'.format(sys._getframe().f_code.co_name))
model = Model()
machine = Machine(model=model, states=states, transitions=transitions, initial=states[0],
auto_transitions=False, ordered_transitions=False)
上記コードはAとBの2状態で、AからBの遷移にafterコールバックを定義したものとなります。
上記で定義したステートマシンに対し、トリガーイベントを起こし、コールバックを呼んでみます。
>>> model.state # 初期状態の確認
'A'
>>> model.fromAtoB() # トリガーイベントを起こす
do "action_after"
True
>>> model.state # 遷移後状態の確認
'B'
遷移を図示すると以下の通りになります。
今回はこの青矢印の遷移が起こった時にコールバックが呼ばれ実行されたことになります。
トリガーイベント（fromAtoB）を起こし、その際にtransitionリスト内の定義である'after'に指定された'action_after'メソッドが呼ばれている事を確認できると思います。
このようにtransitionsパッケージにおけるコールバックは、(一部を除いて)基本的にトリガーイベントを起点として発生する事になります。
またコールバックに指定するものは、Machineの引数modelに指定するオブジェクト内のメソッド（例ではmodelオブジェクトのクラスメソッド）になります。
#コールバックの種類
冒頭の例では非常に簡単なコールバックですが、transitionsパッケージで定義できるコールバックは大きく分けて3つになります。
- 遷移に紐付くコールバック（trantisions辞書で設定）
- 各状態に紐付くコールバック（Stateクラス/state辞書で設定）
- 全状態に紐付くコールバック（Machineクラス/machineオブジェクトで設定）
上記以外にtimeout等特殊なコールバックもあります。今回は順次それぞれについて詳細説明します。
遷移に紐付くコールバック
こちらは遷移編1と内容が被る部分がありますが再度内容を記しておきます。
遷移に紐付くコールバックは、その名の通り遷移それぞれに対し個別にコールバックを設定できます。従って「ある遷移にのみコールバックを設定したい場合」にこちらの定義方法を利用することになります。
遷移準備/遷移前/遷移後コールバックについて
まずは遷移に関する基本的なコールバックになります。
冒頭の例もこの遷移に紐付くコールバックであり、以下の通りtransitions辞書などで設定できます。
transitions = [
{'trigger':'fromAtoB', 'source':'A', 'dest':'B', 'after':'action_after'},
]
上記はfromAtoBトリガーイベントが起こった際の状態Aから状態Bへの遷移定義ですが、afterキーがコールバック定義の一つで、afterキーに指定しているaction_afterがコールバックされるクラス内メソッドになります。このようにコールバックとして呼び出されるクラス内メソッド名を文字列で記載します。
なお遷移に紐付くコールバックはトリガーイベントが起こった際、遷移元（source）と遷移後（dest）どちらで行うかでいくつか種類が別れています。
| 項目 | 定義名 | 実施される状態 | 説明 |
|---|---|---|---|
| 遷移準備コールバック | prepare | 遷移元 (souce) | triggerが発生した際に実行されるコールバック |
| 遷移前コールバック | before | 遷移元 (souce) | 状態が遷移する前に呼び出されるコールバック |
| 遷移後コールバック | after | 遷移先 (dest ) | 状態が遷移した後に呼び出されるコールバック |
冒頭の例ではafterのみしか設定していませんが、複数種類のコールバックを一度に定義することも可能です。
もちろんこれらprepareやbeforeコールバックを実行するクラス内メソッドをModelクラス内に定義しなければ実行エラーになりますので注意が必要です。
# prepare, before, afterを一つの遷移に設定した場合
transitions = [
{'trigger':'fromAtoB', 'source':'A', 'dest':'B',
'prepare':'action_prepare', 'before':'action_before', 'after':'action_after'}
]
上記を図示すると以下の通りになります。
この図は、状態AにおいてfromAtoBトリガーイベントを起こした直後になります。
状態Aにおいてトリガーイベントを起こしprepare, beforeが実施され、その後状態Bに移ってafterのコールバックが実施されることになります。
ガード判定について
UMLなどでもあるガードも実装することができ、transitionsパッケージにおいてガードは遷移に紐付くコールバックになります。（当記事ではガードを便宜上ガード判定と呼んでいます）
ガード判定は遷移を伴うトリガーイベントが起こったとしても、特定の条件を満たしていないと遷移をさせないというものになります。
この判定をコールバックとして実現しているのがtrantisionsパッケージのガード判定コールバックになります。
trantisionsパッケージにおけるガード判定の定義は以下二種があげられます。
| 項目 | 定義名 | 実施される状態 | 説明 |
|---|---|---|---|
| ガード判定(True) | conditions | 遷移元 (souce) | 指定されたコールバックがTrue時に遷移を許可 |
| ガード判定(False) | unless | 遷移元 (souce) | 指定されたコールバックがFalse時に遷移を許可 |
遷移編の焼き直しになりますが、ガード判定の簡単な例について示します。
import sys
from transitions import Machine
states = ['A', 'B'] # 状態の定義
transitions = [
{'trigger':'fromAtoB', 'source':'A', 'dest':'B', 'conditions':'action_conditions'}
]
class Model(object):
# ガード判定用コールバック(conditionsやunlessはbool型の戻り値が必要)
def action_conditions(self):
print('do "{}" on state ({})'.format(sys._getframe().f_code.co_name, self.state))
return False # conditionsではTrueを返すと遷移を許可する(unlessではFalse時に遷移を許可)
model = Model()
machine = Machine(model=model, states=states, transitions=transitions, initial=states[0],
auto_transitions=False, ordered_transitions=False)
上記コード定義後、以下のコードにより動作させてみると、conditionsに設定したクラス内メソッドがコールバックされ実施されている事が分かります。
内部的な一連の動きとしては、トリガーイベント発生 → condisionsのコールバック実施 → コールバックよりFalseが返る → 遷移阻止という流れになります。
>>> model.state
'A'
>>> model.fromAtoB() # トリガーイベントを起こしてみる
do "action_conditions" on state (A)
False
>>> model.state # 状態はAのままで遷移できていない
'A'
上記conditionsの定義において、action_conditionsの戻り値をTrueにすると、状態AにおいてfromAtoBトリガーイベントが起きた際、通常どおり状態Bに遷移する事になります。
unlessについてはconditionsと論理が逆になるだけ（Falseを返した際に遷移を許可）で、基本的な動作は変わりません。
また、conditionsとunlessを同時設定することも可能で、その場合はconditionsの判定が行われた後unlessが実施され、conditionsで遷移不許可となった場合はunlessは実施されない事に注意ください。
なお、prepareとbeforeの役割や違い、実行順については遷移編1でも述べてますが、以下コールバックの優先順位にて全コールバックの優先順位と絡めて述べたいと思います。
各状態に紐付くコールバック
こちらは状態編と内容が被る部分がありますが、もう少し詳しく説明したいと思います。
enterとexitコールバックについて
当コールバック設定は状態定義（stateリスト）で行うものになります。まずはサンプルコードを示します。
from transitions import Machine, State
#状態の定義
states = [State(name='A', on_exit=['action_on_exit']), # Stateクラスで定義可能
{'name':'B', 'on_enter':'action_on_enter'}] # 辞書でも定義可能
transitions = {'trigger':'fromAtoB', 'source':'A', 'dest':'B'}
class Model(object):
# on_enterのコールバック
def action_on_enter(self):
print('do "{}" on state ({})'.format(sys._getframe().f_code.co_name, self.state))
# on_exitのコールバック
def action_on_exit(self):
print('do "{}" on state ({})'.format(sys._getframe().f_code.co_name, self.state))
model = Model()
machine = Machine(model=model, states=states, transitions=transitions, initial='A',
auto_transitions=False, ordered_transitions=False)
状態のみを定義する場合は、冒頭のサンプルコードの通りstate=['A', 'B']と文字列のリストのみで良いですが、各状態に「入った際」や「出た際」にコールバックを起こす際は、以下のように設定します。
states = [State(name='A', on_exit=['action_on_exit']), # Stateクラス版（Machine.Stateのimportが必要）
{'name':'B', 'on_enter':'action_on_enter'}] # 辞書版
上記サンプルコードに対し実際に動作させてみます。
>>> model.state
'A'
>>> model.fromAtoB()
do "action_on_exit" on state (A)
do "action_on_enter" on state (B)
True
>>> model.state
'B'
上記の結果を図示すると以下の通りになります。
fromAtoBトリガーイベントが発生し遷移が行われ状態Aから抜けます。この際状態Aに定義されたon_exitのコールバックaction_on_exitが実施されてます。
その後、状態Bに入り、状態Bに定義されたon_enterのコールバックaction_on_enterが実施されているという流れになります。
このように、on_enterは定義した状態に入った時、on_exitは定義した状態から出た時に実施されることになります。
こちらを図示すると以下のようなイメージとなります。
時間切れコールバックについて
こちらはちょっと特殊なコールバックになりますが、状態に紐付くコールバックになります。
ある状態に入って、その状態にtimeoutで指定した時間滞在すると実施されるコールバックになります。定義例を以下に示します。
from time import sleep
from transitions import Machine
from transitions.extensions.states import add_state_features, Timeout
@add_state_features(Timeout)
class MachineWithTimeout(Machine):
pass
states = ['A',
Timeout(name='B', timeout=10, on_timeout='action_timeout')]
#上記は{'name': 'B', 'timeout': 10, 'on_timeout': 'action_timeout'}でも良い
class Model:
# on_timeoutのコールバック
def action_timeout(self):
print('do "{}" on state ({})'.format(sys._getframe().f_code.co_name, self.state))
model= Model()
machine = MachineWithTimeout(model=model, states=states, initial='A',
auto_transitions=False, ordered_transitions=True)
注意が必要なのはtimeoutはtransitionsパッケージの拡張機能になるので「transitions.extensions.states」でTimeoutクラスとadd_state_fuaturesのimportおよび定義が必要になります。
上記は状態Bに遷移後、状態Bの状態に10秒間滞在するとon_timeoutコールバックが呼ばれるものになります。実際に動作させてみます。
>>> model.state
'A'
>>> model.next_state()
True
>>> model.state
'B'
>>> sleep(30)
do "action_timeout" on state (B)
このように、sleep中にtimeoutコールバックが発生している事がわかります（next_state実施後、状態Bに移って10秒後に発生）
なお、timeout時間が設定されているのにもかかわらず、on_timeoutコールバックが指定されていない場合はtimeout例外が発生します。
各状態に関するコールバックのまとめ
各状態に関するコールバックをまとめると以下のとおりになります。
あくまでもコールバックが行われるのは「各状態に関するコールバック」が定義された"状態"になります。
| 項目 | 定義名 | 説明 |
|---|---|---|
| enterコールバック | on_enter | 定義された状態に入った際に実施されるコールバック |
| exitコールバック | on_exit | 定義された状態から出る際に実施されるコールバック |
| 時間切れコールバック | on_timeout | 定義された状態にtimeout時間経過した際に実施されるコールバック |
全状態に紐付くコールバック
最後に全状態に紐付くコールバックになります。
動作としては各状態に紐付くコールバックに近いものになりますが、ここで設定したコールバックは全状態が対象になります。
以下サンプルを見てみましょう。
from transitions import Machine
states = ['A', 'B'] #状態の定義
class Model(object):
# prepare_eventのコールバック
def action_prepare_event(self):
print('do "{}" on state ({})'.format(sys._getframe().f_code.co_name, self.state))
# before_state_changeのコールバック
def action_before_state_change(self):
print('do "{}" on state ({})'.format(sys._getframe().f_code.co_name, self.state))
# after_state_changeのコールバック
def action_after_state_change(self):
print('do "{}" on state ({})'.format(sys._getframe().f_code.co_name, self.state))
# finalize_eventのコールバック
def action_finalize_event(self):
print('do "{}" on state ({})'.format(sys._getframe().f_code.co_name, self.state))
model = Model()
machine = Machine(model=model, states=states, initial=states[0],
auto_transitions=False, ordered_transitions=True,
prepare_event='action_prepare_event',
before_state_change='action_before_state_change',
after_state_change='action_after_state_change',
finalize_event='action_finalize_event')
上記定義はステートマシンとしては単純な状態Aと状態Bによる2状態の順序遷移になります。
全状態に紐付くコールバックであるprepare_event、before_state_change、after_state_change、finalize_eventの4つ全てを定義しています。
こちらを実際に動かしてみます。
以下のサンプルでは状態A→状態Bへ遷移し、その後状態B→状態Aへと遷移する動作例になります。
>>> model.state # 初期状態の確認
'A'
>>> model.next_state() # トリガーイベント：状態A→状態Bへ
do "action_prepare_event" on state (A)
do "action_before_state_change" on state (A)
do "action_after_state_change" on state (B)
do "action_finalize_event" on state (B)
True
>>> model.state # 遷移後状態の確認
'B'
>>> model.next_state() # トリガーイベント：状態B→状態Aへ
do "action_prepare_event" on state (B)
do "action_before_state_change" on state (B)
do "action_after_state_change" on state (A)
do "action_finalize_event" on state (A)
True
>>> model.state # 遷移後状態の確認
'A'
動作例を見ると分かる通り、stateやtransitionsのコールバック設定がされていないのにも関わらず、状態遷移が発生する度に登録されたコールバックが呼ばれている事が確認できると思います。
このように、全状態に関するコールバックは設定したコールバックが各状態に入った際や出た際などで起こりますので、各状態における共通の初期化などに使えると思います。
なお、prepare_eventとfinalize_eventは少々特殊で、こちらは遷移してもしなくても実施されるコールバックになります。
したがって、ガード判定などで遷移が阻害されたとしても、prepare_eventはトリガーイベントが発生したらすぐに、finalize_eventは最後に実施されるコールバックになります。
ただし、他のコールバック内で例外などが発生してしまった場合は実施されませんので、その点はご注意ください。
「全状態に紐付くコールバック」をまとめると以下の通りになります。
| 項目 | 定義名 | 実施される状態 | 説明 |
|---|---|---|---|
| 準備コールバック | prepare_event | 遷移元 (souce) | triggerが発生した際に実行されるコールバック |
| 状態変化前コールバック | before_state_change | 遷移元 (souce) | 状態が遷移する前に呼び出されるコールバック |
| 状態変化後コールバック | after_state_change | 遷移先 (dest ) | 状態が遷移した後に呼び出されるコールバック |
| 最終コールバック | finalize_event | ※ | 最後に呼び出されるコールバック |
| ※遷移出来た際は遷移先（dest）にて、ガード判定で遷移が中断された場合は遷移元（source）にて実施される。 |
#コールバックの優先順位
さて、コールバックの種類として、「遷移に紐付くコールバック」「各状態に紐付くコールバック」「全状態に紐付くコールバック」の3種類を紹介しましたが、もちろんこれらコールバックは全て合わせて定義可能です。
一方で各種コールバックを複数合わせた際に気になるのが実施順だと思いますが、実施順には明確な定義があります。
以下、実施順序について記します。
なんらかのトリガーイベントが発生した時、下表の上から順に実施されることを示しています。
また下表において「遷移阻止時も実施」列はunlessやconditionsにより遷移が中断された際にコールバックが実施されるかどうかを示しています。
| 定義名 | 定義先 | 状態 | 遷移阻止時も実施 |
|---|---|---|---|
| prepare_event | machine | source | YES |
| prepare | transitions | source | YES |
| conditions | transitions | source | YES ※3 |
| unless | transitions | source | YES ※3, ※4 |
| before_state_change | machine | source | NO |
| before | transitions | source | NO |
| on_exit | state | source ※1 | NO |
| --- 状態変化 --- | |||
| on_enter | state | dest ※2 | NO |
| after | transitions | dest | NO |
| after_state_change | machine | dest | NO |
| finalize_event | machine | source/dest | YES ※5 |
| ※1：遷移元（source)のstateで定義されたon_exitコールバックが呼ばれる | |||
| ※2：遷移先（dest）のstateで定義されたon_enterコールバックが呼ばれる | |||
| ※3：conditionsとunlessが同時設定されていた場合、両者の遷移条件が成立していないと遷移しません。 | |||
| ※4：conditionsとunlessが同時設定されていた場合、conditionsで遷移阻止された場合はunlessも実施されません。 | |||
| ※5：遷移中断時は遷移元で実施され、on_exitの後に実施される。 |
全コールバックを定義した際のサンプルコード
これらを確認するためのサンプルコードを以下に示します。
少々長いですが、以下はtimeoutを除く全てのコールバックを定義してあります。
以下サンプルは状態Aのみで、トリガーイベントeventにより自己遷移（状態A → 状態A）するステートマシンになります
from transitions import State, Machine
states = [State(name='A', on_exit='action_on_exit', on_enter='action_on_enter')]
transitions = {'trigger':'event', 'source':'A', 'dest':'=',
'prepare' : 'action_prepare',
'conditions' : 'action_conditions',
'unless' : 'action_unless',
'before' : 'action_before',
'after' : 'action_after'}
class Model(object):
# prepare_event(全遷移に紐付く)のコールバック
def action_prepare_event(self):
print('do "{}"'.format(sys._getframe().f_code.co_name))
# prepare(遷移に紐付く)のコールバック
def action_prepare(self):
print('do "{}"'.format(sys._getframe().f_code.co_name))
# conditions(遷移に紐付く)のコールバック
def action_conditions(self):
print('do "{}"'.format(sys._getframe().f_code.co_name))
return True #遷移を許可
# unless(遷移に紐付く)のコールバック
def action_unless(self):
print('do "{}"'.format(sys._getframe().f_code.co_name))
return False #遷移を許可
# before_state_change(全遷移に紐付く)のコールバック
def action_before_state_change(self):
print('do "{}"'.format(sys._getframe().f_code.co_name))
# before(遷移に紐付く)のコールバック
def action_before(self):
print('do "{}"'.format(sys._getframe().f_code.co_name))
# on_exit(状態に紐付く)のコールバック
def action_on_exit(self):
print('do "{}"'.format(sys._getframe().f_code.co_name))
### --- change state ---
# on_enter(状態に紐付く)のコールバック
def action_on_enter(self):
print('do "{}"'.format(sys._getframe().f_code.co_name))
# after(遷移に紐付く)のコールバック
def action_after(self):
print('do "{}"'.format(sys._getframe().f_code.co_name))
# after_state_change(全遷移に紐付く)のコールバック
def action_after_state_change(self):
print('do "{}"'.format(sys._getframe().f_code.co_name))
# finalize_event(全遷移に紐付く)のコールバック
def action_finalize_event(self):
print('do "{}"'.format(sys._getframe().f_code.co_name))
#ファイル出力する場合はMatter('test')等ファイル名指定する, Notebook上で表示する場合は引数に指定なし
model = Model()
machine = Machine(model=model, states=states, transitions=transitions, initial=states[0],
auto_transitions=False, ordered_transitions=False)
# 全状態に紐付くコールバック定義の追加
machine.prepare_event = 'action_prepare_event'
machine.before_state_change = 'action_before_state_change'
machine.after_state_change = 'action_after_state_change'
machine.finalize_event = 'action_finalize_event'
こちらを動作させると以下の通りになります。
>>> model.event() #トリガーイベント：状態A→状態A
do "action_prepare_event"
do "action_prepare"
do "action_conditions"
do "action_unless"
do "action_before_state_change"
do "action_before"
do "action_on_exit"
do "action_on_enter"
do "action_after"
do "action_after_state_change"
do "action_finalize_event"
True
なお、もしもconditionsやunlessにより遷移が中断された場合は遷移や状態が変わった際のコールバックは実施されませんが、準備関連のコールバックや最終コールバックなどは実施されます。
以下にガード判定unlessでTrueを返し、遷移を阻止した場合の動作例を示します。
>>> model.event() #トリガーイベント：状態A→状態A
do "action_prepare_event"
do "action_prepare"
do "action_conditions"
do "action_unless"
do "action_finalize_event"
False
このようにunlessにより遷移が阻止されても準備系(prepareやprepare_event）や最終コールバック(finalize_event)が実施されている事に注意下さい。
各コールバックの使い分け
これまでで各種コールバックの実施順やタイミングなどは理解できたと思います。
色々とコールバックの設定があって大変ですが、
- 遷移それぞれにコールバックを設定したい場合は「遷移に紐付くコールバック」
- ある状態に入ったり出ていった際に実施したい場合は「状態に紐付くコールバック」
- 全状態に共通に処理を行わせたい場合は「全状態に紐付くコールバック」
を定義してやれば良いことになります。
基本的に制御関連では「ある状態において、あるイベントが起きた時、あるアクションが起きて別の状態に遷移する」という風な目線で設計することが多いと思いますので、大多数は「遷移に紐付くコールバック」を設計していけば良いと思いますが、設計していく中で、ある状態に入ったり出た際の初期化は「状態に紐付くコールバック」を、状態が変わるごとに初期化をしたい場合は「全状態に紐付くコールバック」をする場合が多いとは思います。
また「遷移に紐付くコールバック」を設計していて共通化できるものは、各状態や全状態に紐付くコールバックにした方がコールバックの見通しは良くなるので、全てのコールバックを「遷移に紐付くコールバック」として実装するのではなく、通常のソフトウェア設計と同じくコールバックが実施される範囲や共通化といった事を意識して定義することをオススメします。
prepare (prepare_event)やfinalize_eventの使いどころ
遷移阻止によらず実施されるコールバックとして、prepare（遷移に紐付く）、prepare_event（全状態に紐付く）、finalize_event（全状態に紐付く）がありますが、これらは使いどころが難しいかもしれません。
prepare（遷移に紐付くコールバック）は、特にガード判定の前処理として使う事をオススメします。
後のコールバック編2で紹介しますが、例えば「prepareコールバックでデータを受け取ったりして前処理を行い、前処理の結果、遷移しても良いと判定したら遷移許可を出す（conditionsコールバックの戻り値がTrueになるように処理する）」といった使い方も出来ると思います。特に遷移前のコールバックであるbeforeやbefore_state_changeは「遷移が確定しないと実施されない」ので、これらガード判定の前処理としては使えず注意が必要です。
全遷移に対しトリガーイベントが発生した際に必ず実施されるfinalize_eventは、個人的には遷移結果の表示やGraphMachineを用いた画像出力などをこちらのコールバックとしてやらせることが多いです。
まとめ
今回はtransitionsパッケージで設定出来るコールバックの種類や定義方法、および実施順序などについて紹介しました。
冒頭でも述べたとおり、コールバックはステートマシンに必須の機能ではありませんがコールバックなしには状態遷移で制御を行う事は難しくなるので、是非活用してもらえればと思います。その際、当記事が少しでもお役に立てたら幸いです。
今回も少々長くなってしまったので、一旦コールバック編をここで区切ります。次はコールバック編2としてコールバックメソッドへのデータの渡し方や、コールバックのキューといった内容について紹介する予定です。