DP.17:ステートマシンの仕組みを適用する。 - Stateパターン -【Python】
【１】Stateパターン
Stateパターンは
オブジェクトにステート（動作モード）を持たせておき、関数コールなどの何かしらのトリガーでステートを切り替える（トランシジョンする）。
といった感じの書き方。
■用語について
・ステート(状態)：
システムの現在の（アクティブな）状態のこと。
・トランシジョン：
ある状態から別の状態に切り替わること。トリガーとなるイベントや条件によって開始される
【２】ステートマシン概要
「ステートマシン（※）」は
デジタル回路やプログラムの設計等で「状態（動作モード）」と「実行できる動作」を用いて、その「状態遷移をモデル化する」便利なツールみたいなもの。
※「有限オートマトン」とか「有限状態機械：FSM（finite state machine）」等とも呼ばれる。
■ステートマシンのイメージ図
■状態遷移例１：AM・FMラジオ
・AMラジオを聞く場合、FMラジオを聞く場合、スイッチを押すなどして電波の受信モードを切り替える。
（→「電波の受信モードが変わる」という状態の変化。トリガーはスイッチ切り替え。）
・聞いているラジオ局を切り替える場合、チューニングをして目的の電波を受信してラジオ局を切り替える。
（→「受信しているエリアが変わる」という状態の変化。トリガーは電波のチューニング。）
■状態遷移例２：自動販売機
自動販売機は「選択した商品」に対し、「投入したお金の量」に応じて様々な動作をする。
・商品売り切れ → 商品を選択できないようにする（入力を拒否する）
・金額不足 → 商品を選択できないようにする（入力を拒否する）
・ぴったりの金額 → 商品を出したらおつりを出さない
・金額余剰 → 商品を出しておつりを出す（連続で商品選択させない場合）
・・・・等々
こんな感じで「ステートマシンは様々なものをモデル化できるところが強み」でもある。
■Non-computational（日常生活で使われる場合）の例
・自動販売機
・エレベータ
・信号機
・パーキングメータ
・・・等々
■computational（主にコンピュータの世界で使われる場合）の例
・ゲームプログラミング
・ハードウェアデザイン
・プロトコルデザイン
・プログラミング言語の構文解析
・・・等々
その他、具体的なソフトウェアの例としては
・Djangoで状態遷移を実現するフレームワーク：django-fsm
・独自言語で状態遷移を記述し、そこから指定の言語コードを生成するコンパイラ：SMC
長々とステートマシン概要と例を挙げてきたが、
要するに「Stateパターンはステートマシンの仕組みをソフトウェアエンジニアリングの世界に適用したもの」ってこと。
【３】ソフトウェアエンジニアリングにおけるStateパターンの利用例
Stateパターンは多くの問題に適用できる。（ステートマシンを使って解決できる問題は、Stateパターンを使うことができる）
例えば、
・OS（組み込み含む）のプロセス・モデル
・コンパイラ（構文解析などで、抽象的な構文木を構築に「状態」を利用する）
・イベント駆動型システム
・コンピュータゲーム（CPUが防御状態から攻撃状態に変わる、等）
【４】実装の仕方：transitionsのインストール
Stateパターンの実装の際は、Stateクラス（ベースクラス）を用意して、それを継承して各クラスを作っていく形が定番だが、一から作るのは面倒くさいので「transitions」ライブラリを利用する。
■インストール
pip install transitions「requirements.txt」を用意して「pip install -r requirements.txt」でもOK。
【５】例題：コンピュータのプロセスモデル
例題としては「transitionsのクイックスタート」の内容で十分ではあるが、別例と「コンピュータのプロセスモデル」をtransitionsで実装してみる。
具体的には次のような状態遷移図となるようなプログラムを作成する。
■コンピュータのプロセスモデルの状態遷移図
■StateMachineの定義
「state machine」の定義は「jsonファイル」にまとめておき、それを読み込ませる形にする。（※もちろん「クイックスタートの例」のようにプログラム内にうめこんでもよい）
■stateMachine.json
{
"states": [
"created",
"waiting",
"running",
"terminated",
"blocked",
"swapped_out_waiting",
"swapped_out_blocked"
],
"transitions": [
{"trigger": "wait", "source": ["created","running","blocked","swapped_out_waiting"], "dest": "waiting", "after":"run_info"},
{"trigger": "run", "source": "waiting", "dest": "running"},
{"trigger": "terminate", "source": "running", "dest": "terminated", "before":"terminate_info"},
{"trigger": "block", "source": ["running","swapped_out_blocked"], "dest": "blocked", "after":"block_info"},
{"trigger": "swap_wait", "source": "waiting", "dest": "swapped_out_waiting", "after":"swap_wait_info"},
{"trigger": "swap_block", "source": "blocked", "dest": "swapped_out_blocked", "after":"swap_block_info"}
],
"initial": "created"
}▲ jsonファイル内に「states」や「transitions」、「initial」を記載しておく
■main.py(抜粋)
from transitions import Machine
import json
class MyComputerProcess:
##### State Machineの初期設定とtransitionの定義
def __init__(self, name):
self.name = name
self.config = None # jsonファイル内のデータロード用
with open("stateMachine.json",mode='r', encoding='utf-8') as f:
self.config = json.load(f)
# ステートマシーン初期設定(残りの設定はロードしたjsonファイルロード＋アンパックで設定)
self.machine = Machine(model=self, **self.config)
####### 以下transition時にコールされる関数など
# after-wait
def run_info(self):
print(f"{self.name} is running ! (trigger:wait() ,after)")
# before-terminate
def terminate_info(self):
print(f"{self.name} terminated ! (trigger:terminate() ,before)")
# after-block
def block_info(self):
print(f"{self.name} is blocked ! (trigger:block() ,after)")
# after-swap_wait
def swap_wait_info(self):
print(f"{self.name} is swapped out and waiting ! (trigger:swap_wait() ,after)")
# after-swap_block
def swap_block_info(self):
print(f"{self.name} is swapped out and blocked ! (trigger:swap_block() ,after) ")▲「jsonファイル」は１回目でも少し記載したように「open()」と「json.load()」を組み合わせて中身を読み込む。
読み込んだjsonファイルの記述内容は「transitions.Machineオブジェクト」へ「**演算子」を使ってアンパックして設定している。
※transitions.Machineオブジェクトが受けつける引数
transitions.Machine(model='self',
states=None,
initial='initial',
transitions=None,
send_event=False,
auto_transitions=True,
ordered_transitions=False,
ignore_invalid_triggers=None,
before_state_change=None,
after_state_change=None,
name=None,
queued=False,
prepare_event=None,
finalize_event=None,
model_attribute='state',
on_exception=None,
**kwargs) ■使用例
my_proccess1 = MyComputerProcess('process1') # 最初はcreated
# transitionさせる
my_proccess1.wait() # waitイベント発生：created → waiting
my_proccess1.run() # runイベント発生：waiting → running
... ...【６】全体コード
from transitions import Machine
import json
class MyComputerProcess:
##### State Machineの初期設定とtransitionの定義
def __init__(self, name):
self.name = name
self.config = None # jsonファイル内のデータロード用
with open("stateMachine.json",mode='r', encoding='utf-8') as f:
self.config = json.load(f)
# ステートマシーン初期設定(残りの設定はロードしたjsonファイルロード＋アンパックで指定)
self.machine = Machine(model=self, **self.config)
####### 以下transition時にコールされる関数など
# after-wait
def run_info(self):
print(f"{self.name} is running ! (trigger:wait() ,after)")
# before-terminate
def terminate_info(self):
print(f"{self.name} terminated ! (trigger:terminate() ,before)")
# after-block
def block_info(self):
print(f"{self.name} is blocked ! (trigger:block() ,after)")
# after-swap_wait
def swap_wait_info(self):
print(f"{self.name} is swapped out and waiting ! (trigger:swap_wait() ,after)")
# after-swap_block
def swap_block_info(self):
print(f"{self.name} is swapped out and blocked ! (trigger:swap_block() ,after) ")
def main():
my_proccess1 = MyComputerProcess('process1')
print(my_proccess1.state) # 初期状態確認（created)
print('----------')
my_proccess1.wait() # waitイベント発生：created → waiting
print(my_proccess1.state)
print('----------')
my_proccess1.run() # runイベント発生：waiting → running
print(my_proccess1.state)
print('----------')
my_proccess1.block() # runイベント発生：running → blocked
print(my_proccess1.state)
if __name__ == '__main__':
main()# 実行結果
created
----------
process1 is running ! (trigger:wait() ,after)
waiting
----------
running
----------
process1 is blocked ! (trigger:block() ,after)
blocked
【７】図として書き出す（graphvizとpygraphviz）
定義したステートマシンのステートチャートをプログラムから書き出すには「graphviz」と「pygraphviz」が必要。なお、graphviz→pygraphvizの順にインストールする必要があるので注意する。
（１）graphvizのインストール
以下からOSに合わせてインストールする
※インストール確認
コマンドプロンプト以下を入力
dot -v→ graphvizのバージョンが表示されればOK
（２）pygraphvizのインストール
同様にOSに合わせてインストールする。
基本的にpipでいいのだが、pipコマンドが先にインストールしたgraphvizを見つけられない場合、ドキュメントの通りオプションをつけてインストールすればよい。
なお、「requirements.txt」にオプションつけてpipを動かしてもよい。例えば今回の「requirements.txt」ファイルは次の通り。
（※graphvizのインストール先は「D:\ProgramFiles\Graphviz」としてデフォルトからちょっと変更した）
■requirements.txt
transitions
pygraphviz --global-option=build_ext --global-option="-ID:\ProgramFiles\Graphviz\include" --global-option="-LD:\ProgramFiles\Graphviz\lib"▲ようはpipコマンドがインストールした「Graphviz」の「includeフォルダ」と「libフォルダ」の場所を把握できればOK。
■使い方
from transitions import Machine
from transitions.extensions import GraphMachine
... ...
# ステートマシーンオブジェクト作成
my_proccess1 = MyComputerProcess('process1')
... ...
# GraphMachineオブジェクト
machine = GraphMachine(my_proccess1,states=my_proccess1.config["states"],
transitions=my_proccess1.config["transitions"],
initial=my_proccess1.config["initial"],
show_conditions=True)
# graphviz + pygraphvizで描画（dotコマンドをたたいてtest.png画像を書き出す）
my_proccess1.get_graph().draw('test.png',prog='dot')【８】おまけ：簡単なCD再生プレイヤーのステートマシンとその状態遷移図の書き出し
記事の最初の方で示した「ステートマシンのイメージ図」も「trasitions」と「graphviz+pygraphviz」で出力した。そのコードも掲載しておく。
■cdPlayerStateMachine.json
{
"states": [
"idle",
"playing",
"stopped",
"paused"
],
"transitions": [
{"trigger": "play", "source": ["idle","stopped","paused"], "dest": "playing", "before":"play_media"},
{"trigger": "stop", "source": ["playing","stopped"], "dest": "stopped","before":"stop_media"},
{"trigger": "pause", "source": "playing", "dest": "paused", "before":"pause_media"}
],
"initial": "idle"
}■cdplayer.py
from transitions import Machine
import json
from transitions.extensions import GraphMachine
class MyCDPlayer:
##### State Machineの初期設定とtransitionの定義
def __init__(self, media_name):
self.media_name = media_name
self.config = None # jsonファイル内のデータロード用
with open("cdPlayerStateMachine.json",mode='r', encoding='utf-8') as f:
self.config = json.load(f)
# ステートマシーン初期設定(残りの設定はロードしたjsonファイルロード＋アンパックで指定)
self.machine = Machine(model=self, **self.config)
####### 以下transition時にコールされる関数など
# before trigger play
def play_media(self):
print(f"--- now [ {self.state} ] state --")
if self.state == self.config['states'][0]:
print("first play...seeking data")
elif self.state == self.config['states'][2]:
print("restarting media (from stopped)")
elif self.state == self.config['states'][3]:
print("resuming media (from paused)")
# before trigger stop
def stop_media(self):
print(f"--- now [ {self.state} ] state --")
if self.state == self.config['states'][1]:
print("stopping (from playing)")
elif self.state == self.config['states'][2]:
print("already stopped !")
# before trigger pause
def pause_media(self):
print(f"--- now [ {self.state} ] state --")
print("pausing media")
def main():
my_cdplayer = MyCDPlayer('myMusicSample.cue')
my_cdplayer.play() # idle → playingへ遷移
print("")
my_cdplayer.stop() # playing → stoppedへ遷移
print("")
my_cdplayer.play() # stopped → playingへ遷移
print("")
my_cdplayer.pause() # playing → pausedへ遷移
print("")
my_cdplayer.play() # paused → playing
# graphviz + pygraphvizで描画
machine = GraphMachine(my_cdplayer,states=my_cdplayer.config["states"],
transitions=my_cdplayer.config["transitions"],
#show_auto_transitions=True, # 考えられる全ルートを表示
initial=my_cdplayer.config["initial"],
show_conditions=True)
my_cdplayer.get_graph().draw('cdplayer.png',prog='dot')
if __name__ == '__main__':
main()# 実行結果
--- now [ idle ] state --
first play...seeking data
--- now [ playing ] state --
stopping (from playing)
--- now [ stopped ] state --
restarting media (from stopped)
--- now [ playing ] state --
pausing media
--- now [ paused ] state --
resuming media (from paused)
以下のような「cdplayer.png」も生成される。
もっと応援したいなと思っていただけた場合、よろしければサポートをおねがいします。いただいたサポートは活動費に使わせていただきます。