Pythonで状態遷移を実装したり動作確認をしたい方に、Pythonの状態遷移パッケージ「transitions」の使い方を説明していきたいと思います。
transitionsはPythonで状態遷移を実現するためのパッケージで、状態遷移そのものは組込みとか制御などでよく使われるものですが、それをPythonで実現したい場合にこのパッケージが有用かと思います。
その他、transitionsの概要やインストール方法、当記事で作成している状態遷移図といったグラフ表示機能の導入(GraphMachine)や設定については準備編の記事を参照頂けたらと思います。
今回の話
「状態編2」ではtransitionsの「状態」に関するカスタムステートであるタグ機能と終端状態例外について紹介しました。
今回は残りのカスタムステートについて説明します。主に以下の内容になりますが、基本的に全て状態に設定される内容になります。
- 各状態でのクラスオブジェクト生成 (Volatile機能)
- 状態のタイムアウト (Timeout機能)
- add_state_featuresとカスタムステートについて
状態編2ではタグや状態遷移例外など、特定状態の識別や終端状態での例外的な話でしたが、今回は各状態でのオブジェクト生成や特殊なコールバックの話になります。
また独自のカスタムステートを作る方法についても若干触れています。まずは上から順に説明していきます。
各状態でのクラスオブジェクト生成 (Volatile機能)
Cをやってきた方にとってコンパイル時に最適化を抑制するためにvolatile宣言することがままありますが、ちょっとその印象とは違います。
Volatileを辞書で引くと「揮発性」という訳がヒットしますが、transitionsにおけるVolatileクラスはその意味合いが強い印象です。
transitionsパッケージにおけるVolatile機能は「状態に入る度にクラス型オブジェクトがmodelに割り当てられる」というもので、そのクラス型オブジェクトはその状態にいる間のみで有効になります。
ともあれ例をみてみましょう。
from transitions import Machine
from transitions.extensions.states import add_state_features, Volatile
# Volatile機能を使う宣言
@add_state_features(Volatile)
class CustomMachine(Machine):
pass
# 状態毎に作成されるクラスの定義
class VolatileClass(object):
def __init__(self):
self.data = 1
def increase(self):
self.data += 1
# ステートマシンが割り当てられるクラス定義
class Model(object):
pass
# Volatile機能付き状態の定義
states = [{'name':'A', 'volatile':VolatileClass},
{'name':'B', 'volatile':VolatileClass}]
model = Model()
machine = CustomMachine(model=model, states=states, initial=states[0]["name"],
auto_transitions=False, ordered_transitions=True)
こちらをGraphMahineで再定義して図示すると以下の通りになります。
ordered_transiitons=Trueにより順序遷移が付与されていることに注意ください。
なんの変哲も無い2状態のステートマシンですが、状態に入る度にVolatileClassが状態に対し生成されます。
以下、動作例になります。初期状態(状態A)ではまだVolatileClassオブジェクトが割り当てられていないので、まず次の状態に遷移します。
>>> model.next_state()
True
状態Bに遷移したので状態BにおいてVolatileClassオブジェクトが生成されました。それでは内容を確認していきましょう。
まずVolitileClassですが以下のようにself.dataというメンバ変数を持ち、それを1インクリメントするincreaseメソッドを持ちます。
class VolatileClass(object):
def __init__(self):
self.data = 1
def increase(self):
self.data += 1
現在このクラス型オブジェクトが状態Bに作られており、以下のように操作することができます。
>>> model.state # 現在の状態を確認
'B'
>>> model.scope.data # VolatileClassのメンバ変数dataにアクセス
1
>>> model.scope.increase() # VolatileClassのメソッドを実行
>>> model.scope.data
2
さらにメンバ変数を追加するといった事も可能です。
以下、元々持っていなかったtextメンバ変数を追加してみましょう。
>>> model.scope.text # メンバ変数textは元々持っていないので例外が発生
Traceback (most recent call last):
File "<input>", line 1, in <module>
AttributeError: 'VolatileClass' object has no attribute 'text'
>>> model.scope.text = 'text-B!' # メンバ変数textを追加
>>> model.scope.text
'text-B!'
このように状態Bに割り当てられたVolatileClassオブジェクトをscopeという属性名を用いてアクセスすることができます。
さて、次は状態Aに戻って同様にVolatileClassを確認してみましょう。
>>> model.next_state()
True
これで状態AにもVolatileClassが生成されました。
model.state # 現在の状態を確認
'A'
model.scope.data # VolatileClassのメンバ変数dataにアクセス
1
model.scope.text # 状態Bで追加したメンバ変数textを確認
Traceback (most recent call last):
File "<input>", line 1, in <module>
AttributeError: 'VolatileClass' object has no attribute 'text'
この例から分かる通り、「ある状態で生成されたVolatileClassオブジェクトは引き継がれない」という事になります。
それでは状態Bに再度戻った際、先ほど状態BでのVolatileClassオブジェクトはどうなっているでしょうか。状態Bに戻って確認してみます。
>>> model.next_state()
True
先ほどは状態BのVolatileObjectにdata=2、text='text-B!'という値/文字列が割り当てられていました。
model.state # 現在の状態を確認
'B'
model.scope.data # VolatileClassのメンバ変数dataにアクセス
1
model.scope.text # 前回状態Bにいた時に追加したメンバ変数textを確認
Traceback (most recent call last):
File "<input>", line 1, in <module>
AttributeError: 'VolatileClass' object has no attribute 'text'
これで大体わかったと思いますが、以前の状態Bで追加した内容も綺麗さっぱり初期化されおり、以前の操作にかかわらず状態に入った際に新しくVolatileClassオブジェクトが作られ初期化されるということになります。ですので、状態に入る毎にVolatileClassのコンストラクタ (def __init__(self))が実施されます。
こういう特性を持っているためVolatile (揮発性)という名が付いているのだと思います。
Volatile機能のまとめと注意点
Volatileで追加指定可能な状態定義をまとめると以下の通りになります。
- volatile：状態に入った際に生成されるクラス指定
- hook：クラスオブジェクトへのアクセス属性名（default : 'scope'）
なお、VolatileクラスはStateクラスの派生クラスなので、Stateクラスと同様のメソッドやメンバ変数を扱えるだけでなく、他のState派生クラスと同様に以下のように宣言することも可能です。
states = [Volatile(name='A', volatile=VolatileClass, hook='volatile'),
Volatile(name='B', volatile=VolatileClass, hook='temp')]
Volatile機能のまとめと注意点は以下となります。
- 状態毎に作成されるクラスVolatileClass(クラス名は変更可)を定義
- 状態定義時にvolatileキーにVolatileClassを指定することで、状態に入る度にVolatileClassオブジェクトが生成される（初期状態には生成されていない）
- 生成されるタイミングはon_enterコールバックが実施される前
- VolatileClassオブジェクトには属性名scopeでアクセスでき、属性名は状態定義時にhookキーで変更可能。
- 一度作られたVolatileClassオブジェクトは状態が変わると破棄され、新たな状態や再度同じ状態に入っても新規に作られる。
- 自己遷移ではVolatileClassオブジェクトは新たに生成されるが、内部遷移ではVolatileClassオブジェクトは保持される。
Volatile機能の捕捉
以下、上記の注意点についていくつか補足します。
Volatile機能の属性名scopeについて
VolatileClassオブジェクトへのアクセスについてはscopeという属性名を用いてアクセスしましたが、こちらは状態定義時にhookキーを用いて変更できます。
指定しない場合は冒頭の例のようにデフォルトで'scope'という名が割り当てられます。
states = [{'name':'A', 'volatile':VolatileClass, 'hook':'volatile'}, # volatileという名でVolatileClassにアクセス
{'name':'B', 'volatile':VolatileClass, 'hook':'temp'}] # tempという名でVolatileClassにアクセス
したがって、このように属性名を変更した場合、状態Aと状態BではVolatileClassオブジェクトへのアクセスには別々の名前を用いてアクセスする必要があるので注意が必要です。
>>> model.state
'A'
>>> model.volatile.data
1
>>> model.state
'B'
>>> model.temp.data
1
自己遷移と内部遷移でのVolatile機能の挙動
状態編では特に遷移について言及していませんが、遷移には自己遷移と内部遷移があります。（詳細は遷移編1へ）
両者とも自己の状態に返るというものですが、コールバックの発生を含め微妙に動作が異なります。
公式ドキュメントに記載はなかったのですが、実験してみたところやはりこのVolatile機能についても自己遷移と内部遷移の違いがあるので記しておきます。
まず以下のような自己遷移(to_Aトリガーイベント)と内部遷移(meトリガーイベント)を持つ状態Aがあり、これにVolatile機能を付与しておきます。
コードとしては以下のようになります。
from transitions import Machine
from transitions.extensions.states import add_state_features, Volatile
@add_state_features(Volatile)
class CustomMachine(Machine):
pass
# 状態毎に作成されるクラスの定義
class VolatileClass(object):
def __init__(self):
self.data = 1
def increase(self):
self.data += 1
# ステートマシンが割り当てられるクラス定義
class Model(object):
pass
# Volatile機能付き状態の定義
states = [{'name':'A', 'volatile':VolatileClass}]
# 遷移の定義
transitions = [{'trigger':'to_A', 'source':'A', 'dest':'='}, # 自己遷移
{'trigger':'me', 'source':'A', 'dest':None}] # 内部遷移
model = Model()
machine = CustomMachine(model=model, transitions=transitions,
states=states, initial=states[0]["name"],
auto_transitions=False, ordered_transitions=False)
初期状態はVolatileClassオブジェクトは割り当てられていないのでまずは自身に遷移して生成しておきます。
model.to_A()
True
これで状態AにVolatileClassオブジェクトが割り当てられました。
ここでVolatileClassオブジェクトのメンバ変数dataに操作をして値を変更しておきます。
>>> model.scope.increase()
>>> model.scope.increase()
>>> model.scope.data
3
現在メンバ変数data=3です。ここで自己遷移と内部遷移の違いを確認してみます。
>>> model.me() # 内部遷移を実行
True
>>> model.scope.data
3
>>> model.to_A() # 自己遷移を実行
True
>>> model.scope.data
1
このように自己遷移ではVolatileClassオブジェクトが新たに生成されていますが、内部遷移ではVolatileClassオブジェクトは新たに生成されず保持されています。
言い換えれば内部遷移ではVolatileClassオブジェクトを引き継いだままトリガーイベントを起こし、遷移に付随するコールバックを起こすことが可能になります。（状態に付随するコールバックは内部遷移では起こらない点に注意）
状態のタイムアウト (Timeout機能)
こちらはタイトルのままで、ある状態に入った時、一定時間経過したらコールバックを実施するというものになります。
以下実際のサンプルを見てみましょう。
from time import sleep
from transitions import Machine
from transitions.extensions.states import add_state_features, Timeout
@add_state_features(Timeout)
class CustomMachine(Machine):
pass
# Timeout機能付き状態の定義
states = [{'name': 'A', 'timeout': 3, 'on_timeout': 'action_timeout'},
{'name': 'B', 'timeout': 3, 'on_timeout': 'action_timeout'}]
class Model:
def action_timeout(self):
print('timeout! on state ({})'.format(self.state))
model= Model()
machine = CustomMachine(model=model, states=states, initial=states[0]['name'],
auto_transitions=False, ordered_transitions=True)
こちらは毎度おなじみ2状態で順序遷移を持つステートマシンです。今回はこちらの各状態にtimeout機能がついています。
それではこのステートマシンを動作させてみます。ひとまず今回は3秒でタイムアウトする設定としています。
>>> model.state
'A'
>>> model.next_state()
True
>>> model.state
'B'
>>> sleep(5)
timeout! on state (B)
このように状態Aから状態Bに遷移してからtimeout時間経過するとコールバックaction_timeoutが動作していることを確認できたと思います。
なおこちらもVolatileと同様に遷移が発生していない初期状態ではタイムアウトは発生しないという事になります。
Timeout機能のまとめと注意点
Timeoutで追加指定可能な状態定義をまとめると以下の通りになります。
- timeout：タイムアウトするまでの時間（秒）
- on_timeout：タイムアウト時のコールバックの指定
なお、TimeoutクラスはStateクラスの派生クラスなので、Stateクラスと同様のメソッドやメンバ変数を扱えるだけでなく、他のState派生クラスと同様に以下のように宣言することも可能です。
states = [Timeout(name='A', timeout=2, on_timeout='action_timeout'),
Timeout(name='B', timeout=2, on_timeout='action_timeout')]
Timeout機能のまとめと注意点は以下となります。
- タイムアウトする時間をtimeoutで指定。
- 状態に入りtimeout時間経過するとon_timeoutコールバックが実施される。
- タイムアウトがカウントダウンを開始するのは状態に入ってon_enterコールバックが実施される直前。
- タイムアウトは状態から抜けると停止し、遷移後の状態にtimeoutが設定されて入れば新たに入った状態のTimeoutのカウントダウンが開始される。
- on_timeoutコールバックで発生した例外はon_timeoutコールバック外で捕捉できない。
- 自己遷移ではTimeoutのカウントダウンはリセットされ再カウントされるが、内部遷移ではTimeoutのカウントダウンはリセットされない。
- timeout時間が設定されているのにもかかわらずon_timeoutコールバックを指定していない場合、AttributeError例外が発生する。
Timeout機能の捕捉
以下、上記の注意点についていくつか補足します。
on_timeoutコールバック内での例外について
タイムアウト自体は別スレッドで実行するので、タイムアウトで発生した例外の捕捉が行われないなど注意が必要です。
これは例えばtry～except内でトリガーイベントを起こしても、on_timeoutコールバック内で何らかの例外が発生した場合、その例外をtry～exceptでキャッチできず例外発生としてコードが停止することを意味しています。
確認コードは長いので以下にしまっておきます。
確認に使用したコードと実行例はここをクリックして下さい。
from time import sleep
from transitions import Machine
from transitions.extensions.states import add_state_features, Timeout
@add_state_features(Timeout)
class CustomMachine(Machine):
pass
states = [{'name': 'A', 'timeout': 3, 'on_timeout': 'action_timeout'},
{'name': 'B', 'timeout': 3, 'on_timeout': 'action_timeout'}]
class Model:
def action_timeout(self):
raise ValueError('timeout内での例外') #ここで例外を発生させる
print('timeout! on state ({})'.format(self.state))
model= Model()
machine = CustomMachine(model=model, states=states, initial=states[0]['name'],
auto_transitions=False, ordered_transitions=True)
以下のようにtry~except内でトリガーイベントを起こしてみます
>>> try:
... model.next_state()
... except:
... print('これは実施されない')
Exception in thread Thread-26:
Traceback (most recent call last):
～略～
ValueError: timeout内での例外
上記のようにexceptで例外をキャッチできず例外として発生し、コードが停止する
自己遷移と内部遷移でのTimeout機能の挙動
こちらもVolatile機能と同様に自己遷移と内部遷移とで挙動が若干異なります。
まず以下のような自己遷移(to_Aトリガーイベント)と内部遷移(meトリガーイベント)を持つ状態Aがあり、これにTimeout機能を付与しておきます。
コードとしては以下のようになります。
from time import sleep
from transitions import Machine
from transitions.extensions.states import add_state_features, Timeout
@add_state_features(Timeout)
class CustomMachine(Machine):
pass
# ステートマシンが割り当てられるクラス定義
class Model(object):
def action_timeout(self):
print('timeout! on state ({})'.format(self.state))
# Timeout機能付き状態の定義
states = [{'name':'A', 'timeout': 1, 'on_timeout': 'action_timeout'}]
# 遷移の定義
transitions = [{'trigger':'to_A', 'source':'A', 'dest':'='}, # 自己遷移
{'trigger':'me', 'source':'A', 'dest':None}] # 内部遷移
model = Model()
machine = CustomMachine(model=model, transitions=transitions,
states=states, initial=states[0]["name"],
auto_transitions=False, ordered_transitions=False)
さてこれを自己遷移と内部遷移それぞれで動かしてみたいと思います。
>>> model.to_A() # 自己遷移を実行(タイムアウト発生あり)
... sleep(3)
timeout! on state (A)
>>> model.me() # 内部遷移を実行(タイムアウト発生なし)
... sleep(3)
このように内部遷移ではon_timeoutコールバックが発生していませんが、これはto_Aトリガーイベントにより既にon_timeoutコールバックが発生してしまっているためであり、内部遷移ではカウントがクリアされず再度on_timeoutが発生しないという状況になっています。もちろんこの状態でto_Aトリガーイベントにより自己遷移すれば再度カウントはクリアされ時間経過すればon_timeoutコールバックが発生します。
言い換えれば自己遷移ではカウントはクリアされタイムアウト動作が初期化されるが、内部遷移してもカウントとタイムアウト動作は初期化されず継続するということになります。
タイムアウトの繰り返し実行
基本的にこのTimeout機能によるタイムアウトは状態を変えない限りワンショット(1回きり)の実行になります。
一方で自己遷移を利用することで繰り返しタイムアウトを発生させ、on_timeoutコールバックを定周期で実施させることが可能です。
以下の例をみてみましょう。
from time import sleep
from transitions import Machine
from transitions.extensions.states import add_state_features, Timeout
@add_state_features(Timeout)
class CustomMachine(Machine):
pass
# ステートマシンが割り当てられるクラス定義
class Model(object):
def action_timeout(self):
print('timeout! on state ({})'.format(self.state))
self.to_A() # タイムアウトしたら再度自己遷移する
# Timeout機能付き状態の定義
states = [{'name':'A', 'timeout': 1, 'on_timeout': 'action_timeout'}]
# 遷移の定義
transitions = [{'trigger':'to_A', 'source':'A', 'dest':'='}] # 自己遷移
model = Model()
machine = CustomMachine(model=model, transitions=transitions,
states=states, initial=states[0]["name"],
auto_transitions=False, ordered_transitions=False)
これはタイムアウトした際に再度自己遷移してタイムアウトのカウントをリセットさせ、on_timeoutコールバックを繰り返し実行できるようにした例です。
上記を定義した後、以下のコードで実行してみましょう。
>>> model.to_A()
... sleep(10)
このように繰り返しtimeoutが実施されている事が確認できました。
※なおtimeoutは前述のとおり別スレッドで動作するので、本サンプルの停止にはCtrl+Cなどでプログラミング動作そのものを停止してください。
add_state_featuresとカスタムステートについて
状態の拡張機能であるカスタムステートを定義する際にはadd_state_featuresでそれらカスタムステートを実現するためのクラスを呼び出す必要があります。
このadd_state_featuresについても補足があるので今回紹介しておきます。
複数のカスタムステートの利用
今回カスタムステートを紹介する際、1機能毎にadd_state_featuresに対しState派生クラスを導入しました、実際は複数のカスタムステートを同時に割り当てることができます。
from time import sleep
from transitions import Machine
from transitions.extensions.states import add_state_features, Timeout, Error
# 終端状態例外とタイムアウトの同時付与
@add_state_features(Error, Timeout)
class CustomMachine(Machine):
pass
class Model(object):
def action_timeout(self):
print('timeout!')
try:
self.to_error() # タイムアウトしたら次の状態へ
except Exception as err: # 終端状態例外はここで発生する
print("終端状態例外:", err)
states = [{'name':'A', 'timeout': 2, 'on_timeout': 'action_timeout'},
{'name':'error'}] # 終端状態例外付き
transitions = [{'trigger':'start', 'source':'A', 'dest':'='}, # 自己遷移
{'trigger':'to_error', 'source':'A', 'dest':'error'}]
model = Model()
machine = CustomMachine(model=model, transitions=transitions,
states=states, initial=states[0]["name"],
auto_transitions=False, ordered_transitions=False)
これは以下のようなステートマシンになります。
タイムアウトが発生した際、終端状態例外付きの終端状態に遷移します。
今回はタイムアウト時間が2sec設定で5sec待つコードで確認して挙動を確認してみます。
>>> model.start()
... sleep(5)
timeout!
終端状態例外: "Error state 'error' reached!"
このように複数の状態を付けることができます。
独自のカスタムステート
実は今回紹介した元々実装されているカスタムステート以外に自身でカスタムステートを作成することができます。
詳細は言及しませんが、基本的にはカスタムステート実現するクラスを実装し、そのクラスをadd_state_featuresを使って取り込むだけです。
from transitions import State, Machine
# Stateクラスから継承する
class OriginalState(State):
def __init__(self, *args, **kwargs):
# ここに状態定義時の初期化を書く
# 独自キーワードはword = kwargs.pop('word', False)のように書いて辞書形式引数から受け取る
super(OriginalState, self).__init__(*args, **kwargs)
def enter(self, event_data):
# ここに状態に入った際に行われる処理を書く
super(OriginalState, self).enter(event_data)
def exit(self, event_data):
# ここに状態から出た際に行われる処理を書く
super(OriginalState, self).exit(event_data)
Stateクラスから継承されるのでenterやexit書かなくても動作しますが、必要に応じてオーバーライドしてください。
加えて上記カスタムステートをMachineに取り込む時は以下のように書きます。
@add_state_features(OriginalState)
class CustomMachine(Machine):
pass
これで独自のカスタムステートを反映したMachineを作成することができるはずです。
実装例としては、本家のコードが参考になるのでそちらも見てみるとよいかもしれません。
まとめ
今回は各状態でのクラスオブジェクト生成 (Volatile機能)とタイムアウト(Timeout機能)について示しました。
Volatile機能の使いどころは難しいですが、状態が変わる毎に初期化されるクラスオブジェクトを利用する場合には活躍すると思います。
Timeout機能については特定状態でタイムアウトさせる場合に便利ですが、タイムアウト時にon_timeoutコールバック内で発生する例外には注意が必要です。
最後に独自のadd_state_featuresとカスタムステートを実現する方法を紹介しました。ここまでできればtransitionsをかなり使いこなしていると思います。
ひとまず今回で状態編は完了になります。大体transitionsパッケージの状態に関する機能は紹介できたと思います。少しでも参考になればと思います。