transitions
Release 0.9.2
多くの拡張機能を備えた Python による軽量のオブジェクト指向ステート マシン実装。 Python 2.7 以降および 3.0 以降と互換性があります。
pip install transitions
...または、GitHub からリポジトリのクローンを作成してから、次のようにします。
python setup.py install
彼らは、良い例は 100 ページの API ドキュメント、100 万のディレクティブ、または 1,000 の単語に相当すると言います。
まあ、「彼ら」はおそらく嘘をついています...しかし、とにかく例を次に示します。
from transitions import Machine
import random
class NarcolepticSuperhero ( object ):
# Define some states. Most of the time, narcoleptic superheroes are just like
# everyone else. Except for...
states = [ 'asleep' , 'hanging out' , 'hungry' , 'sweaty' , 'saving the world' ]
def __init__ ( self , name ):
# No anonymous superheroes on my watch! Every narcoleptic superhero gets
# a name. Any name at all. SleepyMan. SlumberGirl. You get the idea.
self . name = name
# What have we accomplished today?
self . kittens_rescued = 0
# Initialize the state machine
self . machine = Machine ( model = self , states = NarcolepticSuperhero . states , initial = 'asleep' )
# Add some transitions. We could also define these using a static list of
# dictionaries, as we did with states above, and then pass the list to
# the Machine initializer as the transitions= argument.
# At some point, every superhero must rise and shine.
self . machine . add_transition ( trigger = 'wake_up' , source = 'asleep' , dest = 'hanging out' )
# Superheroes need to keep in shape.
self . machine . add_transition ( 'work_out' , 'hanging out' , 'hungry' )
# Those calories won't replenish themselves!
self . machine . add_transition ( 'eat' , 'hungry' , 'hanging out' )
# Superheroes are always on call. ALWAYS. But they're not always
# dressed in work-appropriate clothing.
self . machine . add_transition ( 'distress_call' , '*' , 'saving the world' ,
before = 'change_into_super_secret_costume' )
# When they get off work, they're all sweaty and disgusting. But before
# they do anything else, they have to meticulously log their latest
# escapades. Because the legal department says so.
self . machine . add_transition ( 'complete_mission' , 'saving the world' , 'sweaty' ,
after = 'update_journal' )
# Sweat is a disorder that can be remedied with water.
# Unless you've had a particularly long day, in which case... bed time!
self . machine . add_transition ( 'clean_up' , 'sweaty' , 'asleep' , conditions = [ 'is_exhausted' ])
self . machine . add_transition ( 'clean_up' , 'sweaty' , 'hanging out' )
# Our NarcolepticSuperhero can fall asleep at pretty much any time.
self . machine . add_transition ( 'nap' , '*' , 'asleep' )
def update_journal ( self ):
""" Dear Diary, today I saved Mr. Whiskers. Again. """
self . kittens_rescued += 1
@ property
def is_exhausted ( self ):
""" Basically a coin toss. """
return random . random () < 0.5
def change_into_super_secret_costume ( self ):
print ( "Beauty, eh?" )これで、ステート マシンがNarcolepticSuperheroに焼き付けられました。彼/彼女/それをちょっと外に出してみましょう...
> >> batman = NarcolepticSuperhero ( "Batman" )
> >> batman . state
'asleep'
> >> batman . wake_up ()
> >> batman . state
'hanging out'
> >> batman . nap ()
> >> batman . state
'asleep'
> >> batman . clean_up ()
MachineError : "Can't trigger event clean_up from state asleep!"
> >> batman . wake_up ()
> >> batman . work_out ()
> >> batman . state
'hungry'
# Batman still hasn't done anything useful...
> >> batman . kittens_rescued
0
# We now take you live to the scene of a horrific kitten entreement...
> >> batman . distress_call ()
'Beauty, eh?'
> >> batman . state
'saving the world'
# Back to the crib.
> >> batman . complete_mission ()
> >> batman . state
'sweaty'
> >> batman . clean_up ()
> >> batman . state
'asleep' # Too tired to shower!
# Another productive day, Alfred.
> >> batman . kittens_rescued
1私たちは実際のバットマンの心を読むことはできませんが、 NarcolepticSuperheroの現在の状態を視覚化することは確かにできます。
その方法を知りたい場合は、Diagrams 拡張機能を参照してください。
ステート マシンは、有限数の状態とそれらの状態間の遷移で構成される動作モデルです。各状態および遷移内で、いくつかのアクションを実行できます。ステート マシンは、何らかの初期状態で開始する必要があります。 transitionsを使用する場合、ステート マシンは複数のオブジェクトで構成され、一部の (マシン) には他の (モデル) を操作するための定義が含まれる場合があります。以下では、いくつかの核となる概念とその使用方法を見ていきます。
州。状態は、ステート マシン内の特定の状態または段階を表します。これは、プロセス内の動作またはフェーズの明確なモードです。
遷移。これは、ステート マシンをある状態から別の状態に変化させるプロセスまたはイベントです。
モデル。実際のステートフル構造。これは、遷移中に更新されるエンティティです。また、遷移中に実行されるアクションを定義することもできます。たとえば、遷移の直前や、状態に入るときや状態から出るときなどです。
機械。これは、モデル、状態、遷移、アクションを管理および制御するエンティティです。ステート マシンのプロセス全体を調整するのはコンダクターです。
トリガー。これは遷移を開始するイベントであり、遷移を開始する信号を送信するメソッドです。
アクション。特定の状態に入るとき、出るとき、または遷移中に実行される特定の操作またはタスク。アクションは、コールバックを通じて実装されます。コールバックは、何らかのイベントが発生したときに実行される関数です。
ステート マシンを立ち上げて実行するのは非常に簡単です。オブジェクトのlump ( Matterクラスのインスタンス) があり、その状態を管理したいとします。
class Matter ( object ):
pass
lump = Matter ()次のように、モデルのlumpにバインドされた (最小限の) 動作ステート マシンを初期化できます。
from transitions import Machine
machine = Machine ( model = lump , states = [ 'solid' , 'liquid' , 'gas' , 'plasma' ], initial = 'solid' )
# Lump now has a new state attribute!
lump . state
> >> 'solid'別の方法は、モデルをMachineイニシャライザに明示的に渡さないことです。
machine = Machine ( states = [ 'solid' , 'liquid' , 'gas' , 'plasma' ], initial = 'solid' )
# The machine instance itself now acts as a model
machine . state
> >> 'solid'今回はlumpモデルを引数として渡していないことに注意してください。 Machineに渡される最初の引数はモデルとして機能します。したがって、そこに何かを渡すと、すべての便利な関数がオブジェクトに追加されます。モデルが指定されていない場合は、 machineインスタンス自体がモデルとして機能します。
冒頭で「最小限」と述べたのは、このステート マシンは技術的には動作しますが、実際には何も行わないためです。 'solid'状態で開始されますが、遷移がまだ定義されていないため、別の状態に移行することはありません。
もう一度試してみましょう。
# The states
states = [ 'solid' , 'liquid' , 'gas' , 'plasma' ]
# And some transitions between states. We're lazy, so we'll leave out
# the inverse phase transitions (freezing, condensation, etc.).
transitions = [
{ 'trigger' : 'melt' , 'source' : 'solid' , 'dest' : 'liquid' },
{ 'trigger' : 'evaporate' , 'source' : 'liquid' , 'dest' : 'gas' },
{ 'trigger' : 'sublimate' , 'source' : 'solid' , 'dest' : 'gas' },
{ 'trigger' : 'ionize' , 'source' : 'gas' , 'dest' : 'plasma' }
]
# Initialize
machine = Machine ( lump , states = states , transitions = transitions , initial = 'liquid' )
# Now lump maintains state...
lump . state
> >> 'liquid'
# And that state can change...
# Either calling the shiny new trigger methods
lump . evaporate ()
lump . state
> >> 'gas'
# Or by calling the trigger method directly
lump . trigger ( 'ionize' )
lump . state
> >> 'plasma' Matterインスタンスに追加された輝かしい新しいメソッド ( evaporate() 、 ionize()など) に注目してください。各メソッドは、対応する遷移をトリガーします。上に示したように、遷移の名前を指定してtrigger()メソッドを呼び出すことによって、遷移を動的にトリガーすることもできます。詳細については、「トランジションのトリガー」セクションを参照してください。
あらゆる優れた状態マシン (そして多くの悪い状態マシンも間違いなく) の中心は状態の集合です。上記では、文字列のリストをMachineイニシャライザーに渡すことによって、有効なモデルの状態を定義しました。ただし、内部的には、状態は実際にはStateオブジェクトとして表されます。
さまざまな方法で状態を初期化および変更できます。具体的には、次のことができます。
Machineイニシャライザに渡す、またはStateオブジェクトをそれぞれ直接初期化するか、次のスニペットは、同じ目標を達成するためのいくつかの方法を示しています。
# import Machine and State class
from transitions import Machine , State
# Create a list of 3 states to pass to the Machine
# initializer. We can mix types; in this case, we
# pass one State, one string, and one dict.
states = [
State ( name = 'solid' ),
'liquid' ,
{ 'name' : 'gas' }
]
machine = Machine ( lump , states )
# This alternative example illustrates more explicit
# addition of states and state callbacks, but the net
# result is identical to the above.
machine = Machine ( lump )
solid = State ( 'solid' )
liquid = State ( 'liquid' )
gas = State ( 'gas' )
machine . add_states ([ solid , liquid , gas ])状態はマシンに追加されると一度初期化され、マシンから削除されるまで保持されます。言い換えると、状態オブジェクトの属性を変更した場合、次回その状態に入ったときにこの変更はリセットされません。他の動作が必要な場合に備えて、状態機能を拡張する方法を確認してください。
ただし、状態があり、状態間を移動 (遷移) できるだけでは、それ自体ではあまり役に立ちません。何かをしたい場合、状態に入るときまたは状態から出るときに何らかのアクションを実行する場合はどうすればよいでしょうか?ここでコールバックが登場します。
Stateステート マシンがその状態に入るとき、またはその状態から出るたびに呼び出される、 enterおよびexitコールバックのリストに関連付けることもできます。初期化中にコールバックを指定するには、コールバックを状態プロパティ ディクショナリ内のStateオブジェクト コンストラクターに渡すか、後で追加します。
便宜上、新しいStateがMachineに追加されるたびに、メソッドon_enter_«state name»およびon_exit_«state name»が Machine (モデル上ではなく!) 上に動的に作成され、これにより新しい Enter および Exit を動的に追加できるようになります。必要に応じて後でコールバックします。
# Our old Matter class, now with a couple of new methods we
# can trigger when entering or exit states.
class Matter ( object ):
def say_hello ( self ): print ( "hello, new state!" )
def say_goodbye ( self ): print ( "goodbye, old state!" )
lump = Matter ()
# Same states as above, but now we give StateA an exit callback
states = [
State ( name = 'solid' , on_exit = [ 'say_goodbye' ]),
'liquid' ,
{ 'name' : 'gas' , 'on_exit' : [ 'say_goodbye' ]}
]
machine = Machine ( lump , states = states )
machine . add_transition ( 'sublimate' , 'solid' , 'gas' )
# Callbacks can also be added after initialization using
# the dynamically added on_enter_ and on_exit_ methods.
# Note that the initial call to add the callback is made
# on the Machine and not on the model.
machine . on_enter_gas ( 'say_hello' )
# Test out the callbacks...
machine . set_state ( 'solid' )
lump . sublimate ()
> >> 'goodbye, old state!'
> >> 'hello, new state!'マシンが最初に初期化されるとき、 on_enter_«state name»コールバックは起動しないことに注意してください。たとえば、 on_enter_A()コールバックが定義されており、 initial='A'でMachineを初期化した場合、 on_enter_A()次に状態Aに入るまで起動されません。 (初期化時にon_enter_A()が確実に起動するようにする必要がある場合は、ダミーの初期状態を作成し、 __init__メソッド内でto_A()明示的に呼び出すことができます。)
Stateの初期化時にコールバックを渡したり、コールバックを動的に追加したりするだけでなく、モデル クラス自体にコールバックを定義することもでき、これによりコードの明瞭さが向上する可能性があります。例えば：
class Matter ( object ):
def say_hello ( self ): print ( "hello, new state!" )
def say_goodbye ( self ): print ( "goodbye, old state!" )
def on_enter_A ( self ): print ( "We've just entered state A!" )
lump = Matter ()
machine = Machine ( lump , states = [ 'A' , 'B' , 'C' ])これで、 lump状態Aに遷移するたびに、 Matterクラスで定義されたon_enter_A()メソッドが起動されます。
final=Trueの状態に入ったときにトリガーされるon_finalコールバックを利用できます。
from transitions import Machine , State
states = [ State ( name = 'idling' ),
State ( name = 'rescuing_kitten' ),
State ( name = 'offender_gone' , final = True ),
State ( name = 'offender_caught' , final = True )]
transitions = [[ "called" , "idling" , "rescuing_kitten" ], # we will come when called
{ "trigger" : "intervene" ,
"source" : "rescuing_kitten" ,
"dest" : "offender_gone" , # we
"conditions" : "offender_is_faster" }, # unless they are faster
[ "intervene" , "rescuing_kitten" , "offender_caught" ]]
class FinalSuperhero ( object ):
def __init__ ( self , speed ):
self . machine = Machine ( self , states = states , transitions = transitions , initial = "idling" , on_final = "claim_success" )
self . speed = speed
def offender_is_faster ( self , offender_speed ):
return self . speed < offender_speed
def claim_success ( self , ** kwargs ):
print ( "The kitten is safe." )
hero = FinalSuperhero ( speed = 10 ) # we are not in shape today
hero . called ()
assert hero . is_rescuing_kitten ()
hero . intervene ( offender_speed = 15 )
# >>> 'The kitten is safe'
assert hero . machine . get_state ( hero . state ). final # it's over
assert hero . is_offender_gone () # maybe next time ... 次のいずれかの方法でいつでもモデルの現在の状態を確認できます。
.state属性を検査する、またはis_«state name»()呼び出し現在の状態の実際のStateオブジェクトを取得したい場合は、 Machineインスタンスのget_state()メソッドを通じて実行できます。
lump . state
> >> 'solid'
lump . is_gas ()
> >> False
lump . is_solid ()
> >> True
machine . get_state ( lump . state ). name
> >> 'solid'必要に応じて、 Machineの初期化中にmodel_attribute引数を渡すことで、独自の状態属性名を選択できます。ただし、これによりis_«state name»()の名前もis_«model_attribute»_«state name»()に変更されます。同様に、自動遷移の名前はto_«state name»() to_«model_attribute»_«state name»() «state name»() になります。これは、複数のマシンが個別の状態属性名を持つ同じモデルで動作できるようにするために行われます。
lump = Matter ()
machine = Machine ( lump , states = [ 'solid' , 'liquid' , 'gas' ], model_attribute = 'matter_state' , initial = 'solid' )
lump . matter_state
> >> 'solid'
# with a custom 'model_attribute', states can also be checked like this:
lump . is_matter_state_solid ()
> >> True
lump . to_matter_state_gas ()
> >> True これまで、ステート名を指定し、その名前を使用してステート マシンを操作する方法を見てきました。より厳密な型指定とより多くの IDE コード補完を好む場合 (または、「sesquipedalophobia」という単語が怖くてもう入力できない場合) には、列挙型を使用することが求められているかもしれません。
import enum # Python 2.7 users need to have 'enum34' installed
from transitions import Machine
class States ( enum . Enum ):
ERROR = 0
RED = 1
YELLOW = 2
GREEN = 3
transitions = [[ 'proceed' , States . RED , States . YELLOW ],
[ 'proceed' , States . YELLOW , States . GREEN ],
[ 'error' , '*' , States . ERROR ]]
m = Machine ( states = States , transitions = transitions , initial = States . RED )
assert m . is_RED ()
assert m . state is States . RED
state = m . get_state ( States . RED ) # get transitions.State object
print ( state . name ) # >>> RED
m . proceed ()
m . proceed ()
assert m . is_GREEN ()
m . error ()
assert m . state is States . ERROR必要に応じて列挙型と文字列を混合できます (例: [States.RED, 'ORANGE', States.YELLOW, States.GREEN] )。ただし、内部的にはtransitions引き続き名前 ( enum.Enum.name ) で状態を処理することに注意してください。したがって、状態'GREEN'とStates.GREENを同時に持つことはできません。
上記の例のいくつかはすでに通過時のトランジションの使用法を示していますが、ここではそれらをさらに詳しく見ていきます。
状態と同様に、各遷移は内部的に独自のオブジェクト ( Transitionクラスのインスタンス) として表されます。一連の遷移を初期化する最も簡単な方法は、辞書または辞書のリストをMachine初期化子に渡すことです。これについてはすでに上で見ました。
transitions = [
{ 'trigger' : 'melt' , 'source' : 'solid' , 'dest' : 'liquid' },
{ 'trigger' : 'evaporate' , 'source' : 'liquid' , 'dest' : 'gas' },
{ 'trigger' : 'sublimate' , 'source' : 'solid' , 'dest' : 'gas' },
{ 'trigger' : 'ionize' , 'source' : 'gas' , 'dest' : 'plasma' }
]
machine = Machine ( model = Matter (), states = states , transitions = transitions )辞書で遷移を定義すると明確になるという利点がありますが、煩雑になる可能性があります。簡潔さを求める場合は、リストを使用してトランジションを定義することを選択することもできます。各リスト内の要素がTransition初期化の位置引数 (つまり、 trigger 、 source 、 destinationなど) と同じ順序であることを確認してください。
次のリストのリストは、上記の辞書のリストと機能的に同等です。
transitions = [
[ 'melt' , 'solid' , 'liquid' ],
[ 'evaporate' , 'liquid' , 'gas' ],
[ 'sublimate' , 'solid' , 'gas' ],
[ 'ionize' , 'gas' , 'plasma' ]
]あるいは、初期化後にMachineにトランジションを追加することもできます。
machine = Machine ( model = lump , states = states , initial = 'solid' )
machine . add_transition ( 'melt' , source = 'solid' , dest = 'liquid' )トランジションを実行するには、何らかのイベントがそれをトリガーする必要があります。これを行うには 2 つの方法があります。
基本モデルで自動的にアタッチされたメソッドを使用する場合:
> >> lump . melt ()
> >> lump . state
'liquid'
> >> lump . evaporate ()
> >> lump . state
'gas'これらのメソッドをどこにも明示的に定義する必要がないことに注意してください。各遷移の名前は、 Machineイニシャライザ (この場合は、 lump ) に渡されるモデルにバインドされます。これは、スポットがまだ取得されていない場合にのみtransitionsモデルにコンビニエンス メソッドをアタッチするため、モデルにイベント トリガーと同じ名前のメソッドが既に含まれていてはいけないことも意味します。その動作を変更したい場合は、FAQ を参照してください。
triggerメソッドを使用すると、モデルにアタッチされます (以前に存在しなかった場合)。このメソッドを使用すると、動的トリガーが必要な場合に名前でトランジションを実行できます。
> >> lump . trigger ( 'melt' )
> >> lump . state
'liquid'
> >> lump . trigger ( 'evaporate' )
> >> lump . state
'gas' デフォルトでは、無効な遷移をトリガーすると例外が発生します。
> >> lump . to_gas ()
> >> # This won't work because only objects in a solid state can melt
>> > lump . melt ()
transitions . core . MachineError : "Can't trigger event melt from state gas!"コード内の問題を警告するのに役立つため、この動作は一般的に望ましいものです。ただし、場合によっては、無効なトリガーを黙って無視したい場合もあります。これを行うには、 ignore_invalid_triggers=Trueを設定します (州ごとに、またはすべての州に対してグローバルに)。
> >> # Globally suppress invalid trigger exceptions
>> > m = Machine ( lump , states , initial = 'solid' , ignore_invalid_triggers = True )
> >> # ...or suppress for only one group of states
>> > states = [ 'new_state1' , 'new_state2' ]
> >> m . add_states ( states , ignore_invalid_triggers = True )
> >> # ...or even just for a single state. Here, exceptions will only be suppressed when the current state is A.
>> > states = [ State ( 'A' , ignore_invalid_triggers = True ), 'B' , 'C' ]
> >> m = Machine ( lump , states )
> >> # ...this can be inverted as well if just one state should raise an exception
>> > # since the machine's global value is not applied to a previously initialized state.
>> > states = [ 'A' , 'B' , State ( 'C' )] # the default value for 'ignore_invalid_triggers' is False
> >> m = Machine ( lump , states , ignore_invalid_triggers = True )特定の状態からどの遷移が有効であるかを知る必要がある場合は、 get_triggers使用できます。
m . get_triggers ( 'solid' )
> >> [ 'melt' , 'sublimate' ]
m . get_triggers ( 'liquid' )
> >> [ 'evaporate' ]
m . get_triggers ( 'plasma' )
> >> []
# you can also query several states at once
m . get_triggers ( 'solid' , 'liquid' , 'gas' , 'plasma' )
> >> [ 'melt' , 'evaporate' , 'sublimate' , 'ionize' ]このドキュメントを最初から読んでいる場合は、 get_triggers実際には、上記で明示的に定義されたto_liquidなどのトリガーよりも多くのトリガーを返すことに気づくでしょう。これらはauto-transitionsと呼ばれ、次のセクションで紹介します。
明示的に追加される遷移に加えて、状態がMachineインスタンスに追加されるたびに、 to_«state»()メソッドが自動的に作成されます。このメソッドは、マシンが現在どの状態にあるかに関係なく、ターゲットの状態に移行します。
lump . to_liquid ()
lump . state
> >> 'liquid'
lump . to_solid ()
lump . state
> >> 'solid'必要に応じて、 Machine初期化子でauto_transitions=False設定することで、この動作を無効にすることができます。
特定のトリガーは複数の遷移にアタッチでき、その一部は同じ状態で開始または終了する可能性があります。例えば：
machine . add_transition ( 'transmogrify' , [ 'solid' , 'liquid' , 'gas' ], 'plasma' )
machine . add_transition ( 'transmogrify' , 'plasma' , 'solid' )
# This next transition will never execute
machine . add_transition ( 'transmogrify' , 'plasma' , 'gas' )この場合、 transmogrify()を呼び出すと、モデルの状態が現在'plasma'であれば'solid'に設定され、それ以外の場合は'plasma'に設定されます。 (最初に一致したトランジションのみが実行されることに注意してください。したがって、上記の最後の行で定義されたトランジションは何も行いません。)
'*'ワイルドカードを使用して、トリガーですべての状態から特定の宛先に遷移させることもできます。
machine . add_transition ( 'to_liquid' , '*' , 'liquid' )ワイルドカード遷移は、add_transition() 呼び出し時に存在する状態にのみ適用されることに注意してください。モデルが遷移の定義後に追加された状態にあるときにワイルドカード ベースの遷移を呼び出すと、無効な遷移メッセージが表示され、ターゲットの状態には遷移しません。
再帰トリガー (ソースと宛先と同じ状態を持つトリガー) は、宛先として=指定することで簡単に追加できます。これは、同じ再帰トリガーを複数の状態に追加する必要がある場合に便利です。例えば：
machine . add_transition ( 'touch' , [ 'liquid' , 'gas' , 'plasma' ], '=' , after = 'change_shape' )これにより、トリガーとしてtouch()を使用し、各トリガーの後に実行されるchange_shapeを使用して、3 つの状態すべてに再帰的な遷移が追加されます。
再帰的遷移とは対照的に、内部遷移は実際に状態を離れることはありません。これは、 beforeやafterなどの遷移関連のコールバックは処理されますが、状態関連のコールバックexitまたはenter処理されないことを意味します。遷移を内部に定義するには、宛先をNoneに設定します。
machine . add_transition ( 'internal' , [ 'liquid' , 'gas' ], None , after = 'change_shape' )共通の要望は、状態遷移が厳密な線形シーケンスに従うことです。たとえば、状態['A', 'B', 'C']が与えられた場合、 A → B 、 B → C 、およびC → Aの有効な遷移が必要な場合があります (ただし、他のペアは必要ありません)。
この動作を容易にするために、Transitions はMachineクラスにadd_ordered_transitions()メソッドを提供します。
states = [ 'A' , 'B' , 'C' ]
# See the "alternative initialization" section for an explanation of the 1st argument to init
machine = Machine ( states = states , initial = 'A' )
machine . add_ordered_transitions ()
machine . next_state ()
print ( machine . state )
> >> 'B'
# We can also define a different order of transitions
machine = Machine ( states = states , initial = 'A' )
machine . add_ordered_transitions ([ 'A' , 'C' , 'B' ])
machine . next_state ()
print ( machine . state )
> >> 'C'
# Conditions can be passed to 'add_ordered_transitions' as well
# If one condition is passed, it will be used for all transitions
machine = Machine ( states = states , initial = 'A' )
machine . add_ordered_transitions ( conditions = 'check' )
# If a list is passed, it must contain exactly as many elements as the
# machine contains states (A->B, ..., X->A)
machine = Machine ( states = states , initial = 'A' )
machine . add_ordered_transitions ( conditions = [ 'check_A2B' , ..., 'check_X2A' ])
# Conditions are always applied starting from the initial state
machine = Machine ( states = states , initial = 'B' )
machine . add_ordered_transitions ( conditions = [ 'check_B2C' , ..., 'check_A2B' ])
# With `loop=False`, the transition from the last state to the first state will be omitted (e.g. C->A)
# When you also pass conditions, you need to pass one condition less (len(states)-1)
machine = Machine ( states = states , initial = 'A' )
machine . add_ordered_transitions ( loop = False )
machine . next_state ()
machine . next_state ()
machine . next_state () # transitions.core.MachineError: "Can't trigger event next_state from state C!" トランジションのデフォルトの動作は、イベントを即座に処理することです。これは、 on_enterメソッド内のイベントは、 afterにバインドされたコールバックが呼び出される前に処理されることを意味します。
def go_to_C ():
global machine
machine . to_C ()
def after_advance ():
print ( "I am in state B now!" )
def entering_C ():
print ( "I am in state C now!" )
states = [ 'A' , 'B' , 'C' ]
machine = Machine ( states = states , initial = 'A' )
# we want a message when state transition to B has been completed
machine . add_transition ( 'advance' , 'A' , 'B' , after = after_advance )
# call transition from state B to state C
machine . on_enter_B ( go_to_C )
# we also want a message when entering state C
machine . on_enter_C ( entering_C )
machine . advance ()
> >> 'I am in state C now!'
> >> 'I am in state B now!' # what?この例の実行順序は次のとおりです。
prepare -> before -> on_enter_B -> on_enter_C -> after.
キュー処理が有効な場合、次の遷移がトリガーされる前に遷移が終了します。
machine = Machine ( states = states , queued = True , initial = 'A' )
...
machine . advance ()
> >> 'I am in state B now!'
> >> 'I am in state C now!' # That's better!この結果、
prepare -> before -> on_enter_B -> queue(to_C) -> after -> on_enter_C.
重要な注意事項:キュー内のイベントを処理する場合、トリガー呼び出しは常にTrueを返します。これは、キューに入れられた呼び出しを含む遷移が最終的に正常に完了するかどうかをキューイング時に判断する方法がないためです。これは、単一のイベントのみが処理される場合にも当てはまります。
machine . add_transition ( 'jump' , 'A' , 'C' , conditions = 'will_fail' )
...
# queued=False
machine . jump ()
> >> False
# queued=True
machine . jump ()
> >> Trueモデルがマシンから削除されると、 transitionsによって関連するイベントもすべてキューから削除されます。
class Model :
def on_enter_B ( self ):
self . to_C () # add event to queue ...
self . machine . remove_model ( self ) # aaaand it's gone 場合によっては、特定の条件が発生した場合にのみ特定の遷移を実行したい場合があります。これを行うには、 conditions引数にメソッドまたはメソッドのリストを渡します。
# Our Matter class, now with a bunch of methods that return booleans.
class Matter ( object ):
def is_flammable ( self ): return False
def is_really_hot ( self ): return True
machine . add_transition ( 'heat' , 'solid' , 'gas' , conditions = 'is_flammable' )
machine . add_transition ( 'heat' , 'solid' , 'liquid' , conditions = [ 'is_really_hot' ])上記の例では、モデルが'solid'状態にあるときにheat()呼び出すと、 is_flammable True返した場合、状態'gas'に遷移します。それ以外の場合、 is_really_hot Trueを返すと、状態'liquid'に遷移します。
便宜上、条件とまったく同じように動作しますが、逆の'unless'引数もあります。
machine . add_transition ( 'heat' , 'solid' , 'gas' , unless = [ 'is_flammable' , 'is_really_hot' ])この場合、 is_flammable()とis_really_hot()の両方がFalse返すという条件で、 heat()が発火するたびにモデルは固体から気体に遷移します。
条件チェック メソッドは、トリガー メソッドに渡されるオプションの引数やデータ オブジェクトを受動的に受け取ることに注意してください。たとえば、次の呼び出し:
lump . heat ( temp = 74 )
# equivalent to lump.trigger('heat', temp=74) ... temp=74オプションの kwarg をis_flammable()チェックに渡します (おそらくEventDataインスタンスにラップされます)。詳細については、以下の「データの受け渡し」セクションを参照してください。
移行を続行する前に移行が可能であることを確認したい場合は、モデルに追加されているmay_<trigger_name>関数を使用できます。モデルには、トリガーを名前で確認するためのmay_trigger関数も含まれています。
# check if the current temperature is hot enough to trigger a transition
if lump . may_heat ():
# if lump.may_trigger("heat"):
lump . heat ()これにより、すべてのprepareコールバックが実行され、潜在的な遷移に割り当てられた条件が評価されます。遷移チェックは、遷移の宛先が (まだ) 利用できない場合にも使用できます。
machine . add_transition ( 'elevate' , 'solid' , 'spiritual' )
assert not lump . may_elevate () # not ready yet :(
assert not lump . may_trigger ( "elevate" ) # same result for checks via trigger name 状態だけでなく遷移にもコールバックをアタッチできます。すべての遷移には、遷移の実行前と実行後に呼び出すメソッドのリストを含む'before'と'after'属性があります。
class Matter ( object ):
def make_hissing_noises ( self ): print ( "HISSSSSSSSSSSSSSSS" )
def disappear ( self ): print ( "where'd all the liquid go?" )
transitions = [
{ 'trigger' : 'melt' , 'source' : 'solid' , 'dest' : 'liquid' , 'before' : 'make_hissing_noises' },
{ 'trigger' : 'evaporate' , 'source' : 'liquid' , 'dest' : 'gas' , 'after' : 'disappear' }
]
lump = Matter ()
machine = Machine ( lump , states , transitions = transitions , initial = 'solid' )
lump . melt ()
> >> "HISSSSSSSSSSSSSSSS"
lump . evaporate ()
> >> "where'd all the liquid go?"また、遷移が開始されるとすぐに、 'conditions'がチェックされる前、または他のコールバックが実行される前に実行される'prepare'コールバックもあります。
class Matter ( object ):
heat = False
attempts = 0
def count_attempts ( self ): self . attempts += 1
def heat_up ( self ): self . heat = random . random () < 0.25
def stats ( self ): print ( 'It took you %i attempts to melt the lump!' % self . attempts )
@ property
def is_really_hot ( self ):
return self . heat
states = [ 'solid' , 'liquid' , 'gas' , 'plasma' ]
transitions = [
{ 'trigger' : 'melt' , 'source' : 'solid' , 'dest' : 'liquid' , 'prepare' : [ 'heat_up' , 'count_attempts' ], 'conditions' : 'is_really_hot' , 'after' : 'stats' },
]
lump = Matter ()
machine = Machine ( lump , states , transitions = transitions , initial = 'solid' )
lump . melt ()
lump . melt ()
lump . melt ()
lump . melt ()
> >> "It took you 4 attempts to melt the lump!"現在の状態が名前付き遷移の有効なソースでない限り、 prepare呼び出されないことに注意してください。
すべての遷移の前後に実行されるデフォルトのアクションは、それぞれbefore_state_changeとafter_state_changeを使用して初期化中にMachineに渡すことができます。
class Matter ( object ):
def make_hissing_noises ( self ): print ( "HISSSSSSSSSSSSSSSS" )
def disappear ( self ): print ( "where'd all the liquid go?" )
states = [ 'solid' , 'liquid' , 'gas' , 'plasma' ]
lump = Matter ()
m = Machine ( lump , states , before_state_change = 'make_hissing_noises' , after_state_change = 'disappear' )
lump . to_gas ()
> >> "HISSSSSSSSSSSSSSSS"
> >> "where'd all the liquid go?"コールバックには、独立して実行する必要がある 2 つのキーワードもあります。a) 可能な遷移の数、b) 遷移が成功したかどうか、c) 他のコールバックの実行中にエラーが発生した場合でも。 prepare_eventを使用してMachineに渡されるコールバックは、可能な遷移 (およびそれらの個別のprepareコールバック) の処理が行われる前に1 回実行されます。 finalize_eventのコールバックは、処理された遷移の成功に関係なく実行されます。エラーが発生した場合は、 event_dataにerrorとして添付され、 send_event=Trueで取得できることに注意してください。
from transitions import Machine
class Matter ( object ):
def raise_error ( self , event ): raise ValueError ( "Oh no" )
def prepare ( self , event ): print ( "I am ready!" )
def finalize ( self , event ): print ( "Result: " , type ( event . error ), event . error )
states = [ 'solid' , 'liquid' , 'gas' , 'plasma' ]
lump = Matter ()
m = Machine ( lump , states , prepare_event = 'prepare' , before_state_change = 'raise_error' ,
finalize_event = 'finalize' , send_event = True )
try :
lump . to_gas ()
except ValueError :
pass
print ( lump . state )
# >>> I am ready!
# >>> Result: <class 'ValueError'> Oh no
# >>> initial場合によっては、物事が意図したとおりに機能しない場合があり、物事を継続させるために例外を処理し、混乱をクリーンアップする必要があります。これを行うためにコールバックをon_exceptionに渡すことができます。
from transitions import Machine
class Matter ( object ):
def raise_error ( self , event ): raise ValueError ( "Oh no" )
def handle_error ( self , event ):
print ( "Fixing things ..." )
del event . error # it did not happen if we cannot see it ...
states = [ 'solid' , 'liquid' , 'gas' , 'plasma' ]
lump = Matter ()
m = Machine ( lump , states , before_state_change = 'raise_error' , on_exception = 'handle_error' , send_event = True )
try :
lump . to_gas ()
except ValueError :
pass
print ( lump . state )
# >>> Fixing things ...
# >>> initialおそらくすでにお気づきかと思いますが、呼び出し可能オブジェクトを状態、条件、遷移に渡す標準的な方法は名前によるものです。コールバックと条件を処理するとき、 transitionsその名前を使用して、関連する呼び出し可能オブジェクトをモデルから取得します。メソッドを取得できず、メソッドにドットが含まれている場合、 transitions名前をモジュール関数へのパスとして扱い、インポートを試みます。あるいは、プロパティまたは属性の名前を渡すこともできます。これらは関数にラップされますが、明らかな理由によりイベント データを受信できません。 (バインドされた) 関数などの呼び出し可能関数を直接渡すこともできます。前述したように、呼び出し可能な名前のリスト/タプルをコールバック パラメーターに渡すこともできます。コールバックは追加された順序で実行されます。
from transitions import Machine
from mod import imported_func
import random
class Model ( object ):
def a_callback ( self ):
imported_func ()
@ property
def a_property ( self ):
""" Basically a coin toss. """
return random . random () < 0.5
an_attribute = False
model = Model ()
machine = Machine ( model = model , states = [ 'A' ], initial = 'A' )
machine . add_transition ( 'by_name' , 'A' , 'A' , conditions = 'a_property' , after = 'a_callback' )
machine . add_transition ( 'by_reference' , 'A' , 'A' , unless = [ 'a_property' , 'an_attribute' ], after = model . a_callback )
machine . add_transition ( 'imported' , 'A' , 'A' , after = 'mod.imported_func' )
model . by_name ()
model . by_reference ()
model . imported ()呼び出し可能な解決はMachine.resolve_callableで行われます。より複雑な呼び出し可能な解決戦略が必要な場合は、このメソッドをオーバーライドできます。
例
class CustomMachine ( Machine ):
@ staticmethod
def resolve_callable ( func , event_data ):
# manipulate arguments here and return func, or super() if no manipulation is done.
super ( CustomMachine , CustomMachine ). resolve_callable ( func , event_data )要約すると、現在、イベントをトリガーするには 3 つの方法があります。 lump.melt()などのモデルの便利な関数を呼び出したり、 lump.trigger("melt")などの名前でトリガーを実行したり、 machine.dispatch("melt")を使用して複数のモデルにイベントをディスパッチしたりできます (「代替の初期化パターン)。遷移時のコールバックは次の順序で実行されます。
| 折り返し電話 | 現在の状態 | コメント |
|---|---|---|
'machine.prepare_event' | source | 個々のトランジションが処理される前に1 回実行されます |
'transition.prepare' | source | 移行が開始されるとすぐに実行されます |
'transition.conditions' | source | 条件が満たされないと移行が停止する可能性があります |
'transition.unless' | source | 条件が満たされないと移行が停止する可能性があります |
'machine.before_state_change' | source | モデルで宣言されたデフォルトのコールバック |
'transition.before' | source | |
'state.on_exit' | source | ソース状態で宣言されたコールバック |
<STATE CHANGE> | ||
'state.on_enter' | destination | 宛先状態で宣言されたコールバック |
'transition.after' | destination | |
'machine.on_final' | destination | 子に対するコールバックが最初に呼び出されます |
'machine.after_state_change' | destination | モデルで宣言されたデフォルトのコールバック。内部遷移後にも呼び出されます |
'machine.on_exception' | source/destination | コールバックは例外が発生したときに実行されます |
'machine.finalize_event' | source/destination | コールバックは、遷移が発生しないか、例外が発生した場合でも実行されます。 |
いずれかのコールバックで例外が発生した場合、コールバックの処理は続行されません。これは、遷移前 ( state.on_exit以前) にエラーが発生した場合、遷移が停止されることを意味します。遷移が実行された後 ( state.on_enter以降) にレイズが発生した場合、状態の変更は持続し、ロールバックは発生しません。 machine.finalize_eventで指定されたコールバックは、ファイナライズ コールバック自体によって例外が発生しない限り、常に実行されます。各コールバック シーケンスは、次のステージが実行される前に終了する必要があることに注意してください。コールバックをブロックすると実行順序が停止するため、 triggerまたはdispatch呼び出し自体がブロックされます。コールバックを並行して実行したい場合は、非同期処理用の拡張機能AsyncMachineまたはスレッド用のLockedMachineを検討するとよいでしょう。
場合によっては、マシンの初期化時に登録されたコールバック関数に、モデルの現在の状態を反映するデータを渡す必要があります。トランジションを使用すると、これを 2 つの異なる方法で行うことができます。
まず (デフォルト)、位置引数またはキーワード引数をトリガー メソッド ( add_transition()呼び出したときに作成される) に直接渡すことができます。
class Matter ( object ):
def __init__ ( self ): self . set_environment ()
def set_environment ( self , temp = 0 , pressure = 101.325 ):
self . temp = temp
self . pressure = pressure
def print_temperature ( self ): print ( "Current temperature is %d degrees celsius." % self . temp )
def print_pressure ( self ): print ( "Current pressure is %.2f kPa." % self . pressure )
lump = Matter ()
machine = Machine ( lump , [ 'solid' , 'liquid' ], initial = 'solid' )
machine . add_transition ( 'melt' , 'solid' , 'liquid' , before = 'set_environment' )
lump . melt ( 45 ) # positional arg;
# equivalent to lump.trigger('melt', 45)
lump . print_temperature ()
> >> 'Current temperature is 45 degrees celsius.'
machine . set_state ( 'solid' ) # reset state so we can melt again
lump . melt ( pressure = 300.23 ) # keyword args also work
lump . print_pressure ()
> >> 'Current pressure is 300.23 kPa.'トリガーには任意の数の引数を渡すことができます。
このアプローチには重要な制限が 1 つあります。状態遷移によってトリガーされるすべてのコールバック関数は、すべての引数を処理できなければなりません。それぞれのコールバックが多少異なるデータを想定している場合、これにより問題が発生する可能性があります。
これを回避するために、Transitions はデータを送信するための代替方法をサポートしています。 Machine初期化時にsend_event=Trueを設定すると、トリガーへのすべての引数がEventDataインスタンスにラップされ、すべてのコールバックに渡されます。 ( EventDataオブジェクトは、何らかの目的でこれらにアクセスする必要がある場合に備えて、イベントに関連付けられたソース状態、モデル、遷移、マシン、トリガーへの内部参照も保持します。)
class Matter ( object ):
def __init__ ( self ):
self . temp = 0
self . pressure = 101.325
# Note that the sole argument is now the EventData instance.
# This object stores positional arguments passed to the trigger method in the
# .args property, and stores keywords arguments in the .kwargs dictionary.
def set_environment ( self , event ):
self . temp = event . kwargs . get ( 'temp' , 0 )
self . pressure = event . kwargs . get ( 'pressure' , 101.325 )
def print_pressure ( self ): print ( "Current pressure is %.2f kPa." % self . pressure )
lump = Matter ()
machine = Machine ( lump , [ 'solid' , 'liquid' ], send_event = True , initial = 'solid' )
machine . add_transition ( 'melt' , 'solid' , 'liquid' , before = 'set_environment' )
lump . melt ( temp = 45 , pressure = 1853.68 ) # keyword args
lump . print_pressure ()
> >> 'Current pressure is 1853.68 kPa.'これまでのすべての例では、新しいMachineインスタンスを別のモデル ( lump 、クラスMatterのインスタンス) にアタッチしました。この分離により ( Matterクラスに大量の新しいメソッドをモンキー パッチする必要がないため) 整理整頓が保たれますが、ステート マシンでどのメソッドが呼び出されるかを追跡する必要があるため、面倒になることもあります。 、およびステートマシンがバインドされているモデル上でどれが呼び出されるのか (たとえば、 lump.on_enter_StateA()とmachine.add_transition() )。
幸いなことに、Transitions は柔軟性があり、他の 2 つの初期化パターンをサポートしています。
まず、別のモデルをまったく必要としないスタンドアロンのステート マシンを作成できます。初期化中にモデル引数を省略するだけです。
machine = Machine ( states = states , transitions = transitions , initial = 'solid' )
machine . melt ()
machine . state
> >> 'liquid'この方法でマシンを初期化すると、すべてのトリガー イベント ( evaporate() 、 sublimate()など) とすべてのコールバック関数をMachineインスタンスに直接アタッチできます。
このアプローチには、ステート マシンのすべての機能を 1 か所に統合できるという利点がありますが、ステート ロジックを別のコントローラーではなくモデル自体に含めるべきだと考える場合は、少し不自然に感じるかもしれません。
別の (おそらくより良い) アプローチは、モデルをMachineクラスから継承させることです。 Transitions は、継承をシームレスにサポートするように設計されています。 (必ずMachineクラスの__init__メソッドをオーバーライドしてください!):
class Matter ( Machine ):
def say_hello ( self ): print ( "hello, new state!" )
def say_goodbye ( self ): print ( "goodbye, old state!" )
def __init__ ( self ):
states = [ 'solid' , 'liquid' , 'gas' ]
Machine . __init__ ( self , states = states , initial = 'solid' )
self . add_transition ( 'melt' , 'solid' , 'liquid' )
lump = Matter ()
lump . state
> >> 'solid'
lump . melt ()
lump . state
> >> 'liquid'ここでは、すべてのステート マシンの機能を既存のモデルに統合できます。多くの場合、必要なすべての機能を個別のスタンドアロンMachineインスタンスに貼り付けるよりも自然に感じられます。
マシンはMachine(model=[model1, model2, ...])のようなリストとして渡すことができる複数のモデルを処理できます。マシン インスタンス自体だけでなくモデルを追加する場合は、 Machine(model=[Machine.self_literal, model1, ...])のように、初期化中にクラス変数プレースホルダー (文字列) Machine.self_literalを渡すことができます。また、スタンドアロン マシンを作成し、 model=Noneコンストラクターに渡すことでmachine.add_modelを介してモデルを動的に登録することもできます。さらに、 machine.dispatchを使用して、現在追加されているすべてのモデルでイベントをトリガーできます。マシンが長期間使用でき、モデルが一時的なものでガベージ コレクションする必要がある場合は、必ずmachine.remove_modelを呼び出してください。
class Matter ():
pass
lump1 = Matter ()
lump2 = Matter ()
# setting 'model' to None or passing an empty list will initialize the machine without a model
machine = Machine ( model = None , states = states , transitions = transitions , initial = 'solid' )
machine . add_model ( lump1 )
machine . add_model ( lump2 , initial = 'liquid' )
lump1 . state
> >> 'solid'
lump2 . state
> >> 'liquid'
# custom events as well as auto transitions can be dispatched to all models
machine . dispatch ( "to_plasma" )
lump1 . state
> >> 'plasma'
assert lump1 . state == lump2 . state
machine . remove_model ([ lump1 , lump2 ])
del lump1 # lump1 is garbage collected
del lump2 # lump2 is garbage collectedステート マシン コンストラクターで初期状態を指定しない場合、 transitions 'initial'と呼ばれるデフォルトの状態が作成および追加されます。デフォルトの初期状態を望まない場合は、 initial=Noneを渡すことができます。ただし、この場合、モデルを追加するたびに初期状態を渡す必要があります。
machine = Machine ( model = None , states = states , transitions = transitions , initial = None )
machine . add_model ( Matter ())
> >> "MachineError: No initial state configured for machine, must specify when adding model."
machine . add_model ( Matter (), initial = 'liquid' )複数の状態を持つモデルは、異なるmodel_attribute値を使用して複数のマシンを接続できます。 「状態のチェック」で説明したように、これによりカスタムis/to_<model_attribute>_<state_name>関数が追加されます。
lump = Matter ()
matter_machine = Machine ( lump , states = [ 'solid' , 'liquid' , 'gas' ], initial = 'solid' )
# add a second machine to the same model but assign a different state attribute
shipment_machine = Machine ( lump , states = [ 'delivered' , 'shipping' ], initial = 'delivered' , model_attribute = 'shipping_state' )
lump . state
> >> 'solid'
lump . is_solid () # check the default field
> >> True
lump . shipping_state
> >> 'delivered'
lump . is_shipping_state_delivered () # check the custom field.
> >> True
lump . to_shipping_state_shipping ()
> >> True
lump . is_shipping_state_delivered ()
> >> Falseトランジションには、非常に基本的なログ機能が含まれています。多くのイベント (状態変化、遷移トリガー、条件チェックなど) は、標準の Python loggingモジュールを使用して INFO レベルのイベントとして記録されます。これは、スクリプトで標準出力へのログ記録を簡単に設定できることを意味します。
# Set up logging; The basic log level will be DEBUG
import logging
logging . basicConfig ( level = logging . DEBUG )
# Set transitions' log level to INFO; DEBUG messages will be omitted
logging . getLogger ( 'transitions' ). setLevel ( logging . INFO )
# Business as usual
machine = Machine ( states = states , transitions = transitions , initial = 'solid' )
...機械はピクルス化可能で、 pickleを保管したりロードしたりできます。 Python 3.3 以前の場合はdillが必要です。
import dill as pickle # only required for Python 3.3 and earlier
m = Machine ( states = [ 'A' , 'B' , 'C' ], initial = 'A' )
m . to_B ()
m . state
> >> B
# store the machine
dump = pickle . dumps ( m )
# load the Machine instance again
m2 = pickle . loads ( dump )
m2 . state
> >> B
m2 . states . keys ()
> >> [ 'A' , 'B' , 'C' ]お気づきかと思いますが、 transitions Python の動的機能の一部を使用して、モデルを処理する便利な方法を提供します。ただし、静的型チェッカーは、モデルの属性やメソッドが実行前に知られていないことを好みません。歴史的に、 transitions偶発的なオーバーライドを防ぐために、モデルにすでに定義されている便利なメソッドも割り当てられませんでした。
でも心配しないでください。マシン コンストラクター パラメーターmodel_override使用して、モデルの装飾方法を変更できます。 model_override=Trueを設定すると、 transitionsすでに定義されたメソッドのみをオーバーライドします。これにより、実行時に新しいメソッドが表示されなくなり、使用するヘルパー メソッドを定義できるようになります。
from transitions import Machine
# Dynamic assignment
class Model :
pass
model = Model ()
default_machine = Machine ( model , states = [ "A" , "B" ], transitions = [[ "go" , "A" , "B" ]], initial = "A" )
print ( model . __dict__ . keys ()) # all convenience functions have been assigned
# >> dict_keys(['trigger', 'to_A', 'may_to_A', 'to_B', 'may_to_B', 'go', 'may_go', 'is_A', 'is_B', 'state'])
assert model . is_A () # Unresolved attribute reference 'is_A' for class 'Model'
# Predefined assigment: We are just interested in calling our 'go' event and will trigger the other events by name
class PredefinedModel :
# state (or another parameter if you set 'model_attribute') will be assigned anyway
# because we need to keep track of the model's state
state : str
def go ( self ) -> bool :
raise RuntimeError ( "Should be overridden!" )
def trigger ( self , trigger_name : str ) -> bool :
raise RuntimeError ( "Should be overridden!" )
model = PredefinedModel ()
override_machine = Machine ( model , states = [ "A" , "B" ], transitions = [[ "go" , "A" , "B" ]], initial = "A" , model_override = True )
print ( model . __dict__ . keys ())
# >> dict_keys(['trigger', 'go', 'state'])
model . trigger ( "to_B" )
assert model . state == "B"すべての便利な関数を使用し、いくつかのコールバックを組み合わせて使用したい場合、多数の状態と遷移が定義されていると、モデルの定義がかなり複雑になる可能性があります。 transitionsのメソッドgenerate_base_model 、マシン構成からベースモデルを生成して、それを支援します。
from transitions . experimental . utils import generate_base_model
simple_config = {
"states" : [ "A" , "B" ],
"transitions" : [
[ "go" , "A" , "B" ],
],
"initial" : "A" ,
"before_state_change" : "call_this" ,
"model_override" : True ,
}
class_definition = generate_base_model ( simple_config )
with open ( "base_model.py" , "w" ) as f :
f . write ( class_definition )
# ... in another file
from transitions import Machine
from base_model import BaseModel
class Model ( BaseModel ): # call_this will be an abstract method in BaseModel
def call_this ( self ) -> None :
# do something
model = Model ()
machine = Machine ( model , ** simple_config )オーバーライドされるモデル メソッドを定義すると、少し追加の作業が追加されます。特に状態と遷移がモデルの前後のリストで定義されている場合、イベント名のスペルが正しいことを確認するために前後に切り替えるのは面倒な場合があります。状態を列挙型として定義すると、定型文や文字列操作の不確実性を軽減できます。 add_transitionsとeventを使用して、モデル クラス内でトランジションを直接定義することもできます。属性に値を割り当てるために関数デコレーターadd_transitions使用するか、event を使用するかは、好みのコード スタイルに応じて決定します。どちらも同じように動作し、同じシグネチャを持ち、（ほぼ）同じ IDE タイプのヒントが得られるはずです。これはまだ進行中の作業であるため、カスタム Machine クラスを作成し、遷移に with_model_settings を使用して、そのように定義された遷移を確認する必要があります。
from enum import Enum
from transitions . experimental . utils import with_model_definitions , event , add_transitions , transition
from transitions import Machine
class State ( Enum ):
A = "A"
B = "B"
C = "C"
class Model :
state : State = State . A
@ add_transitions ( transition ( source = State . A , dest = State . B ), [ State . C , State . A ])
@ add_transitions ({ "source" : State . B , "dest" : State . A })
def foo ( self ): ...
bar = event (
{ "source" : State . B , "dest" : State . A , "conditions" : lambda : False },
transition ( source = State . B , dest = State . C )
)
@ with_model_definitions # don't forget to define your model with this decorator!
class MyMachine ( Machine ):
pass
model = Model ()
machine = MyMachine ( model , states = State , initial = model . state )
model . foo ()
model . bar ()
assert model . state == State . C
model . foo ()
assert model . state == State . Aトランジションのコアは軽量に保たれていますが、その機能を拡張するさまざまな MixIn があります。現在サポートされているものは次のとおりです。
必要な機能が有効になっているステート マシン インスタンスを取得するには、2 つのメカニズムがあります。最初のアプローチでは、機能が必要な場合に、4 つのパラメーターgraph 、 nested 、 lockedまたはasyncio Trueに設定してコンビニエンスfactoryを利用します。
from transitions . extensions import MachineFactory
# create a machine with mixins
diagram_cls = MachineFactory . get_predefined ( graph = True )
nested_locked_cls = MachineFactory . get_predefined ( nested = True , locked = True )
async_machine_cls = MachineFactory . get_predefined ( asyncio = True )
# create instances from these classes
# instances can be used like simple machines
machine1 = diagram_cls ( model , state , transitions )
machine2 = nested_locked_cls ( model , state , transitions )この場合、基礎となるクラスを知る必要がないため、このアプローチは実験的な使用をターゲットとしています。ただし、クラスはtransitions.extensionsから直接インポートすることもできます。命名スキームは次のとおりです。
| 図表 | 入れ子になった | ロックされています | アシンシオ | |
|---|---|---|---|---|
| 機械 | ✘ | ✘ | ✘ | ✘ |
| グラフマシン | ✓ | ✘ | ✘ | ✘ |
| 階層型マシン | ✘ | ✓ | ✘ | ✘ |
| ロックされたマシン | ✘ | ✘ | ✓ | ✘ |
| 階層グラフマシン | ✓ | ✓ | ✘ | ✘ |
| ロックされたグラフマシン | ✓ | ✘ | ✓ | ✘ |
| ロックされた階層マシン | ✘ | ✓ | ✓ | ✘ |
| ロックされた階層グラフマシン | ✓ | ✓ | ✓ | ✘ |
| 非同期マシン | ✘ | ✘ | ✘ | ✓ |
| 非同期グラフマシン | ✓ | ✘ | ✘ | ✓ |
| 階層型非同期マシン | ✘ | ✓ | ✘ | ✓ |
| 階層型非同期グラフマシン | ✓ | ✓ | ✘ | ✓ |
機能豊富なステート マシンを使用するには、次のように記述できます。
from transitions . extensions import LockedHierarchicalGraphMachine as LHGMachine
machine = LHGMachine ( model , states , transitions ) Transitions には、状態をネストできる拡張モジュールが含まれています。これにより、コンテキストを作成し、状態がステート マシン内の特定のサブタスクに関連するケースをモデル化することができます。ネストされた状態を作成するには、遷移からNestedStateインポートするか、初期化引数nameとchildren含むディクショナリを使用します。オプションで、 initial使用して、ネストされた状態に入ったときに遷移するサブ状態を定義できます。
from transitions . extensions import HierarchicalMachine
states = [ 'standing' , 'walking' , { 'name' : 'caffeinated' , 'children' :[ 'dithering' , 'running' ]}]
transitions = [
[ 'walk' , 'standing' , 'walking' ],
[ 'stop' , 'walking' , 'standing' ],
[ 'drink' , '*' , 'caffeinated' ],
[ 'walk' , [ 'caffeinated' , 'caffeinated_dithering' ], 'caffeinated_running' ],
[ 'relax' , 'caffeinated' , 'standing' ]
]
machine = HierarchicalMachine ( states = states , transitions = transitions , initial = 'standing' , ignore_invalid_triggers = True )
machine . walk () # Walking now
machine . stop () # let's stop for a moment
machine . drink () # coffee time
machine . state
> >> 'caffeinated'
machine . walk () # we have to go faster
machine . state
> >> 'caffeinated_running'
machine . stop () # can't stop moving!
machine . state
> >> 'caffeinated_running'
machine . relax () # leave nested state
machine . state # phew, what a ride
> >> 'standing'
# machine.on_enter_caffeinated_running('callback_method') initialを使用する構成は次のようになります。
# ...
states = [ 'standing' , 'walking' , { 'name' : 'caffeinated' , 'initial' : 'dithering' , 'children' : [ 'dithering' , 'running' ]}]
transitions = [
[ 'walk' , 'standing' , 'walking' ],
[ 'stop' , 'walking' , 'standing' ],
# this transition will end in 'caffeinated_dithering'...
[ 'drink' , '*' , 'caffeinated' ],
# ... that is why we do not need do specify 'caffeinated' here anymore
[ 'walk' , 'caffeinated_dithering' , 'caffeinated_running' ],
[ 'relax' , 'caffeinated' , 'standing' ]
]
# ... HierarchicalMachineコンストラクターのinitialキーワードは、ネストされた状態 (例: initial='caffeinated_running' ) および並列状態とみなされる状態のリスト (例: initial=['A', 'B'] ) または現在の状態を受け入れます。別のモデル ( initial=model.state ) は、実質的に前述のオプションの 1 つになります。文字列を渡すとき、 transitionターゲット状態のinitialサブステートをチェックし、これをエントリ状態として使用することに注意してください。これは、サブ状態が初期状態を言及しなくなるまで再帰的に行われます。並列状態またはリストとして渡された状態は「そのまま」使用され、それ以上の初期評価は行われません。
以前に作成した状態オブジェクトはNestedStateまたはその派生クラスである必要があることに注意してください。単純なMachineインスタンスで使用される標準のStateクラスには、ネストに必要な機能がありません。
from transitions . extensions . nesting import HierarchicalMachine , NestedState
from transitions import State
m = HierarchicalMachine ( states = [ 'A' ], initial = 'initial' )
m . add_state ( 'B' ) # fine
m . add_state ({ 'name' : 'C' }) # also fine
m . add_state ( NestedState ( 'D' )) # fine as well
m . add_state ( State ( 'E' )) # does not work!ネストされた状態を操作するときに考慮する必要がある点: 状態名はNestedState.separatorと連結されます。現在、区切り文字はアンダースコア ('_') に設定されているため、基本マシンと同様に動作します。これは、状態fooのサブ状態barがfoo_barによって認識されることを意味します。 barのサブステートbaz foo_bar_bazなどと呼ばれます。サブステートに入るとき、すべての親ステートに対してenter呼び出されます。終了するサブステートにも同じことが当てはまります。第三に、ネストされた状態は、親の遷移動作を上書きする可能性があります。遷移が現在の状態で認識されていない場合、遷移はその親に委任されます。
これは、標準構成では、HSM の状態名にアンダースコアを含めてはいけないことを意味します。 transitionsの場合、 machine.add_state('state_name') state_nameという名前の状態を追加するのか、それとも状態stateにサブステートnameを追加するのかを判断することは不可能です。ただし、これでは十分でない場合もあります。たとえば、州名が複数の単語で構成されており、 CamelCaseの代わりにアンダースコアを使用してそれらを区切る必要がある場合などです。これに対処するには、分離に使用する文字を簡単に変更できます。 Python 3 を使用している場合は、派手な Unicode 文字を使用することもできます。ただし、区切り文字をアンダースコア以外に設定すると、一部の動作 (auto_transition とコールバックの設定) が変わります。
from transitions . extensions import HierarchicalMachine
from transitions . extensions . nesting import NestedState
NestedState . separator = '↦'
states = [ 'A' , 'B' ,
{ 'name' : 'C' , 'children' :[ '1' , '2' ,
{ 'name' : '3' , 'children' : [ 'a' , 'b' , 'c' ]}
]}
]
transitions = [
[ 'reset' , 'C' , 'A' ],
[ 'reset' , 'C↦2' , 'C' ] # overwriting parent reset
]
# we rely on auto transitions
machine = HierarchicalMachine ( states = states , transitions = transitions , initial = 'A' )
machine . to_B () # exit state A, enter state B
machine . to_C () # exit B, enter C
machine . to_C . s3 . a () # enter C↦a; enter C↦3↦a;
machine . state
> >> 'C↦3↦a'
assert machine . is_C . s3 . a ()
machine . to ( 'C↦2' ) # not interactive; exit C↦3↦a, exit C↦3, enter C↦2
machine . reset () # exit C↦2; reset C has been overwritten by C↦3
machine . state
> >> 'C'
machine . reset () # exit C, enter A
machine . state
> >> 'A'
# s.on_enter('C↦3↦a', 'callback_method') to_C_3_a()の代わりに、自動遷移はto_C.s3.a()として呼び出されます。サブステートが数字で始まる場合、Python の属性命名スキームに準拠するために、transitions は自動遷移FunctionWrapperにプレフィックス 's' ('3' は 's3' になります) を追加します。対話型の補完が必要ない場合は、 to('C↦3↦a')を直接呼び出すことができます。さらに、 on_enter/exit_<<state name>>はon_enter/exit(state_name, callback)に置き換えられます。状態チェックも同様の方法で実行できます。 is_C_3_a()の代わりに、 FunctionWrapperバリアントis_C.s3.a()を使用できます。
現在の状態が特定の状態のサブ状態であるかどうかを確認するために、 is_stateキーワードallow_substatesをサポートしています。
machine . state
> >> 'C.2.a'
machine . is_C () # checks for specific states
> >> False
machine . is_C ( allow_substates = True )
> >> True
assert machine . is_C . s2 () is False
assert machine . is_C . s2 ( allow_substates = True ) # FunctionWrapper support allow_substate as well HSM でも列挙型を使用できますが、 Enum値によって比較されることに注意してください。状態ツリーに複数の値がある場合、それらの状態を区別できません。
states = [ States . RED , States . YELLOW , { 'name' : States . GREEN , 'children' : [ 'tick' , 'tock' ]}]
states = [ 'A' , { 'name' : 'B' , 'children' : states , 'initial' : States . GREEN }, States . GREEN ]
machine = HierarchicalMachine ( states = states )
machine . to_B ()
machine . is_GREEN () # returns True even though the actual state is B_GREEN HierarchicalMachine並列状態とネストされた状態のより適切な分離をサポートするために、最初から書き直されました。これには、コミュニティからのフィードバックに基づいたいくつかの調整が含まれます。処理の順序と構成を理解するには、次の例を見てください。
from transitions . extensions . nesting import HierarchicalMachine
import logging
states = [ 'A' , 'B' , { 'name' : 'C' , 'parallel' : [{ 'name' : '1' , 'children' : [ 'a' , 'b' , 'c' ], 'initial' : 'a' ,
'transitions' : [[ 'go' , 'a' , 'b' ]]},
{ 'name' : '2' , 'children' : [ 'x' , 'y' , 'z' ], 'initial' : 'z' }],
'transitions' : [[ 'go' , '2_z' , '2_x' ]]}]
transitions = [[ 'reset' , 'C_1_b' , 'B' ]]
logging . basicConfig ( level = logging . INFO )
machine = HierarchicalMachine ( states = states , transitions = transitions , initial = 'A' )
machine . to_C ()
# INFO:transitions.extensions.nesting:Exited state A
# INFO:transitions.extensions.nesting:Entered state C
# INFO:transitions.extensions.nesting:Entered state C_1
# INFO:transitions.extensions.nesting:Entered state C_2
# INFO:transitions.extensions.nesting:Entered state C_1_a
# INFO:transitions.extensions.nesting:Entered state C_2_z
machine . go ()
# INFO:transitions.extensions.nesting:Exited state C_1_a
# INFO:transitions.extensions.nesting:Entered state C_1_b
# INFO:transitions.extensions.nesting:Exited state C_2_z
# INFO:transitions.extensions.nesting:Entered state C_2_x
machine . reset ()
# INFO:transitions.extensions.nesting:Exited state C_1_b
# INFO:transitions.extensions.nesting:Exited state C_2_x
# INFO:transitions.extensions.nesting:Exited state C_1
# INFO:transitions.extensions.nesting:Exited state C_2
# INFO:transitions.extensions.nesting:Exited state C
# INFO:transitions.extensions.nesting:Entered state B children代わりにparallel使用すると、 transitions渡されたリストのすべての状態に同時に入ります。どのサブ状態に入るかは、常に直接のサブ状態を指す必要があるinitialによって定義されます。新しい機能は、状態定義でtransitionsキーワードを渡すことによってローカル遷移を定義することです。上記で定義された遷移['go', 'a', 'b']はC_1でのみ有効です。 ['go', '2_z', '2_x']のようにサブステートを参照できますが、ローカルに定義された遷移で親ステートを直接参照することはできません。親ステートが終了すると、その子ステートも終了します。 Machineで知られるトランジションの処理順序 (トランジションが追加された順序でトランジションが考慮される) に加えて、 HierarchicalMachine階層も考慮されます。物質で定義された遷移は最初に評価され（例えばC_1_a C_2_z前に残されます）、ワイルドカード*で定義された遷移は（今のところ） 0.8.0ネストされた状態で始まる根状（この例A 、 B 、 C ）に遷移を追加するだけです直接追加することができ、最中に親状態の作成を発行します。
m = HierarchicalMachine ( states = [ 'A' ], initial = 'A' )
m . add_state ( 'B_1_a' )
m . to_B_1 ()
assert m . is_B ( allow_substates = True ) 0.9.1の実験：州またはHSM自体でon_finalコールバックを使用できます。 a）状態自体がfinalにタグ付けされ、入力されたばかり、またはb）すべての物質が最終的であると見なされ、少なくとも1つの物質が最終状態に入った場合、コールバックがトリガーされます。 b）条件b）条件を保持している場合、すべての親が最終と見なされる場合。これは、すべての物質が最終状態に達したときに処理が並行して行われ、HSMまたは親の状態に通知する必要がある場合に役立つ可能性があります。
from transitions . extensions import HierarchicalMachine
from functools import partial
# We initialize this parallel HSM in state A:
# / X
# / / yI
# A -> B - Y - yII [final]
# Z - zI
# zII [final]
def final_event_raised ( name ):
print ( "{} is final!" . format ( name ))
states = [ 'A' , { 'name' : 'B' , 'parallel' : [{ 'name' : 'X' , 'final' : True , 'on_final' : partial ( final_event_raised , 'X' )},
{ 'name' : 'Y' , 'transitions' : [[ 'final_Y' , 'yI' , 'yII' ]],
'initial' : 'yI' ,
'on_final' : partial ( final_event_raised , 'Y' ),
'states' :
[ 'yI' , { 'name' : 'yII' , 'final' : True }]
},
{ 'name' : 'Z' , 'transitions' : [[ 'final_Z' , 'zI' , 'zII' ]],
'initial' : 'zI' ,
'on_final' : partial ( final_event_raised , 'Z' ),
'states' :
[ 'zI' , { 'name' : 'zII' , 'final' : True }]
},
],
"on_final" : partial ( final_event_raised , 'B' )}]
machine = HierarchicalMachine ( states = states , on_final = partial ( final_event_raised , 'Machine' ), initial = 'A' )
# X will emit a final event right away
machine . to_B ()
# >>> X is final!
print ( machine . state )
# >>> ['B_X', 'B_Y_yI', 'B_Z_zI']
# Y's substate is final now and will trigger 'on_final' on Y
machine . final_Y ()
# >>> Y is final!
print ( machine . state )
# >>> ['B_X', 'B_Y_yII', 'B_Z_zI']
# Z's substate becomes final which also makes all children of B final and thus machine itself
machine . final_Z ()
# >>> Z is final!
# >>> B is final!
# >>> Machine is final! セマンティックな順序に加えて、特定のタスクに州のマシンを指定し、それらを再利用する計画を立てたい場合、ネストされた状態は非常に便利です。 0.8.0の前に、 HierarchicalMachineマシンインスタンス自体を統合するのではなく、それらのコピーを作成することで状態と移行を統合します。ただし、 0.8.0 (Nested)Stateインスタンスは参照されているため、1つのマシンの状態とイベントのコレクションの変更が他のマシンインスタンスに影響を与えることを意味します。ただし、モデルとその状態は共有されません。イベントとトランジションも参照によってコピーされ、 remapキーワードを使用しない場合は両方のインスタンスで共有されることに注意してください。この変更は、参照によって合格されたStateインスタンスも使用するMachineに沿って行われるように行われました。
count_states = [ '1' , '2' , '3' , 'done' ]
count_trans = [
[ 'increase' , '1' , '2' ],
[ 'increase' , '2' , '3' ],
[ 'decrease' , '3' , '2' ],
[ 'decrease' , '2' , '1' ],
[ 'done' , '3' , 'done' ],
[ 'reset' , '*' , '1' ]
]
counter = HierarchicalMachine ( states = count_states , transitions = count_trans , initial = '1' )
counter . increase () # love my counter
states = [ 'waiting' , 'collecting' , { 'name' : 'counting' , 'children' : counter }]
transitions = [
[ 'collect' , '*' , 'collecting' ],
[ 'wait' , '*' , 'waiting' ],
[ 'count' , 'collecting' , 'counting' ]
]
collector = HierarchicalMachine ( states = states , transitions = transitions , initial = 'waiting' )
collector . collect () # collecting
collector . count () # let's see what we got; counting_1
collector . increase () # counting_2
collector . increase () # counting_3
collector . done () # collector.state == counting_done
collector . wait () # collector.state == waiting HierarchicalMachineマシンがchildrenキーワードで渡された場合、このマシンの初期状態は新しい親状態に割り当てられます。上記の例では、 countingを入力するとcounting_1も入力されることがわかります。これが望ましくない動作であり、マシンが親状態でむしろ停止する必要がある場合、ユーザーは{'name': 'counting', 'children': counter, 'initial': False}のようにFalseのようにinitialを渡すことができます。
このような埋め込み状態コレクションを「返す」ことを望んでいる場合があります。これは、それが完了した後、スーパーステートの1つに出て通過する必要があることを意味します。この動作を達成するには、状態移行を再マッピングできます。上記の例では、州が完了doneた場合は、カウンターを返したいと思います。これは次のように行われます。
states = [ 'waiting' , 'collecting' , { 'name' : 'counting' , 'children' : counter , 'remap' : { 'done' : 'waiting' }}]
... # same as above
collector . increase () # counting_3
collector . done ()
collector . state
> >> 'waiting' # be aware that 'counting_done' will be removed from the state machine上記のように、 remap使用すると、元の状態マシンでは有効ではないため、イベントとトランジションをコピーします。再利用された状態マシンに最終状態がない場合は、もちろん手動で遷移を追加できます。 「カウンター」に「完了」状態がなかった場合、 ['done', 'counter_3', 'waiting']を追加するだけで、同じ動作を達成できます。
状態と遷移を参照ではなく値によってコピーする場合（たとえば、0.8以前の動作を維持する場合）、 NestedStateを作成し、マシンのイベントや状態の深いコピーを割り当てることでそうすることができます。それ。
from transitions . extensions . nesting import NestedState
from copy import deepcopy
# ... configuring and creating counter
counting_state = NestedState ( name = "counting" , initial = '1' )
counting_state . states = deepcopy ( counter . states )
counting_state . events = deepcopy ( counter . events )
states = [ 'waiting' , 'collecting' , counting_state ]複雑な状態マシンの場合、インスタンス化されたマシンではなく構成を共有する方が実行可能になる場合があります。特に、インスタンス化された機械はHierarchicalMachineから派生する必要があるためです。このような構成は、JSONまたはYAMLを介して簡単に保存およびロードできます（FAQを参照）。 HierarchicalMachineキーワードのchildrenまたはstatesのいずれかで物質を定義することができます。両方が存在する場合、 childrenのみが考慮されます。
counter_conf = {
'name' : 'counting' ,
'states' : [ '1' , '2' , '3' , 'done' ],
'transitions' : [
[ 'increase' , '1' , '2' ],
[ 'increase' , '2' , '3' ],
[ 'decrease' , '3' , '2' ],
[ 'decrease' , '2' , '1' ],
[ 'done' , '3' , 'done' ],
[ 'reset' , '*' , '1' ]
],
'initial' : '1'
}
collector_conf = {
'name' : 'collector' ,
'states' : [ 'waiting' , 'collecting' , counter_conf ],
'transitions' : [
[ 'collect' , '*' , 'collecting' ],
[ 'wait' , '*' , 'waiting' ],
[ 'count' , 'collecting' , 'counting' ]
],
'initial' : 'waiting'
}
collector = HierarchicalMachine ( ** collector_conf )
collector . collect ()
collector . count ()
collector . increase ()
assert collector . is_counting_2 ()追加のキーワード：
title （オプション）：生成された画像のタイトルを設定します。show_conditions （デフォルトのfalse）：遷移エッジの条件を表示しますshow_auto_transitions （デフォルトのfalse）：グラフの自動遷移を表示しますshow_state_attributes （デフォルトのfalse）：show callbacks（enter、exit）、タグ、グラフのタイムアウト遷移は、状態間のすべての有効な遷移を表示する基本的な状態図を生成できます。基本的な図のサポートは、Mermaidのライブエディター、GitlabまたはGithubおよびその他のWebサービスのMarkdownファイルで使用できる人魚の状態マシン定義を生成します。たとえば、このコード：
from transitions . extensions . diagrams import HierarchicalGraphMachine
import pyperclip
states = [ 'A' , 'B' , { 'name' : 'C' ,
'final' : True ,
'parallel' : [{ 'name' : '1' , 'children' : [ 'a' , { "name" : "b" , "final" : True }],
'initial' : 'a' ,
'transitions' : [[ 'go' , 'a' , 'b' ]]},
{ 'name' : '2' , 'children' : [ 'a' , { "name" : "b" , "final" : True }],
'initial' : 'a' ,
'transitions' : [[ 'go' , 'a' , 'b' ]]}]}]
transitions = [[ 'reset' , 'C' , 'A' ], [ "init" , "A" , "B" ], [ "do" , "B" , "C" ]]
m = HierarchicalGraphMachine ( states = states , transitions = transitions , initial = "A" , show_conditions = True ,
title = "Mermaid" , graph_engine = "mermaid" , auto_transitions = False )
m . init ()
pyperclip . copy ( m . get_graph (). draw ( None )) # using pyperclip for convenience
print ( "Graph copied to clipboard!" )この図を作成します（ドキュメントソースを確認して、マークダウン表記を確認してください）：
---
人魚グラフ
---
Statediagram-V2
方向LR
classdef s_default fill：white、color：black
classdef s_inactive fill：白、色：黒
classdef s_parallel：black、fill：white
classdef s_active色：赤、塗りつぶし：darksalmon
classdef s_previous color：青、塗りつぶし：azure
as as as as
クラスA s_previous
b
クラスB S_ACTIVE
cとして「C」を状態にします
c-> [*]
クラスC S_DEFAULT
状態c {
c_1として「1」を述べます
状態C_1 {
[*] - > c_1_a
c_1_aとして「a」を述べます
c_1_bとして「b」を述べます
C_1_B-> [*]
}
--
C_2として「2」を状態
状態C_2 {
[*] - > c_2_a
c_2_aとして「a」を述べます
c_2_bとして「b」を述べます
C_2_B-> [*]
}
}
C-> A：リセット
a-> b：init
b-> c：do
c_1_a-> c_1_b：go
c_2_a-> c_2_b：go
[*] - > a
より洗練されたグラフ機能を使用するには、 graphvizおよび/またはpygraphvizインストールする必要があります。パッケージgraphvizでグラフを生成するには、GraphVizを手動でインストールするか、パッケージマネージャーを介してインストールする必要があります。
sudo apt-get install graphviz graphviz-dev # Ubuntu and Debian
brew install graphviz # MacOS
conda install graphviz python-graphviz # (Ana)conda
これで、実際のPythonパッケージをインストールできます
pip install graphviz pygraphviz # install graphviz and/or pygraphviz manually...
pip install transitions[diagrams] # ... or install transitions with 'diagrams' extras which currently depends on pygraphviz
現在、 GraphMachine利用可能な場合はpygraphviz使用し、 pygraphviz見つからないときにgraphvizに戻ります。 graphvizも利用できない場合、 mermaidが使用されます。これはgraph_engine="graphviz" （または"mermaid" ）をコンストラクターに渡すことでオーバーライドできます。このデフォルトは将来変更される可能性があり、 pygraphvizサポートが削除される可能性があることに注意してください。 Model.get_graph()を使用すると、現在のグラフまたは関心領域（ROI）を取得し、次のように描画できます。
# import transitions
from transitions . extensions import GraphMachine
m = Model ()
# without further arguments pygraphviz will be used
machine = GraphMachine ( model = m , ...)
# when you want to use graphviz explicitly
machine = GraphMachine ( model = m , graph_engine = "graphviz" , ...)
# in cases where auto transitions should be visible
machine = GraphMachine ( model = m , show_auto_transitions = True , ...)
# draw the whole graph ...
m . get_graph (). draw ( 'my_state_diagram.png' , prog = 'dot' )
# ... or just the region of interest
# (previous state, active state and all reachable states)
roi = m . get_graph ( show_roi = True ). draw ( 'my_state_diagram.png' , prog = 'dot' )これは次のようなものを生成します：
使用するバックエンドとは無関係に、draw関数は、最初の引数としてファイル記述子またはバイナリストリームも受け入れます。このパラメーターをNoneに設定すると、バイトストリームが返されます。
import io
with open ( 'a_graph.png' , 'bw' ) as f :
# you need to pass the format when you pass objects instead of filenames.
m . get_graph (). draw ( f , format = "png" , prog = 'dot' )
# you can pass a (binary) stream too
b = io . BytesIO ()
m . get_graph (). draw ( b , format = "png" , prog = 'dot' )
# or just handle the binary string yourself
result = m . get_graph (). draw ( None , format = "png" , prog = 'dot' )
assert result == b . getvalue ()コールバックとして渡された参照と部分的なものは、可能な限り良好に解決されます。
from transitions . extensions import GraphMachine
from functools import partial
class Model :
def clear_state ( self , deep = False , force = False ):
print ( "Clearing state ..." )
return True
model = Model ()
machine = GraphMachine ( model = model , states = [ 'A' , 'B' , 'C' ],
transitions = [
{ 'trigger' : 'clear' , 'source' : 'B' , 'dest' : 'A' , 'conditions' : model . clear_state },
{ 'trigger' : 'clear' , 'source' : 'C' , 'dest' : 'A' ,
'conditions' : partial ( model . clear_state , False , force = True )},
],
initial = 'A' , show_conditions = True )
model . get_graph (). draw ( 'my_state_diagram.png' , prog = 'dot' )これはこれに似たものを生成するはずです：
参照の形式がニーズに合わない場合は、静的メソッドGraphMachine.format_referencesオーバーライドできます。参照を完全にスキップしたい場合は、 GraphMachine.format_references返してみませNone 。また、グラフを使用および編集する方法についてのより詳細な例については、IPython/Jupyterノートブックの例をご覧ください。
イベントディスパッチがスレッドで行われる場合、機能アクセス（！sic）がリエントラントロックで保護されているLockedMachine LockedHierarchicalMachineのいずれかを使用できます。これは、モデルまたは状態マシンのメンバー変数をいじくり回すことで、マシンを破損することを避けません。
from transitions . extensions import LockedMachine
from threading import Thread
import time
states = [ 'A' , 'B' , 'C' ]
machine = LockedMachine ( states = states , initial = 'A' )
# let us assume that entering B will take some time
thread = Thread ( target = machine . to_B )
thread . start ()
time . sleep ( 0.01 ) # thread requires some time to start
machine . to_C () # synchronized access; won't execute before thread is done
# accessing attributes directly
thread = Thread ( target = machine . to_B )
thread . start ()
machine . new_attrib = 42 # not synchronized! will mess with execution order Pythonコンテキストマネージャーは、 machine_contextキーワード引数を介して渡すことができます。
from transitions . extensions import LockedMachine
from threading import RLock
states = [ 'A' , 'B' , 'C' ]
lock1 = RLock ()
lock2 = RLock ()
machine = LockedMachine ( states = states , initial = 'A' , machine_context = [ lock1 , lock2 ]) machine_model経由のコンテキストは、 Machineに登録されているすべてのモデル間で共有されます。モデルごとのコンテキストも追加できます。
lock3 = RLock ()
machine . add_model ( model , model_context = lock3 )州のマシンは、単一のトリガー呼び出しのコンテキストであっても、州のマシンが複数回呼び出すため、すべてのユーザーが提供するコンテキストマネージャーが再入国することが重要です。
Python 3.7以降を使用している場合は、 AsyncMachine使用して非同期コールバックを使用できます。必要に応じて、同期と非同期のコールバックを混ぜることができますが、これには望ましくない副作用があるかもしれません。イベントを待つ必要があり、イベントループもあなたが処理する必要があることに注意してください。
from transitions . extensions . asyncio import AsyncMachine
import asyncio
import time
class AsyncModel :
def prepare_model ( self ):
print ( "I am synchronous." )
self . start_time = time . time ()
async def before_change ( self ):
print ( "I am asynchronous and will block now for 100 milliseconds." )
await asyncio . sleep ( 0.1 )
print ( "I am done waiting." )
def sync_before_change ( self ):
print ( "I am synchronous and will block the event loop (what I probably shouldn't)" )
time . sleep ( 0.1 )
print ( "I am done waiting synchronously." )
def after_change ( self ):
print ( f"I am synchronous again. Execution took { int (( time . time () - self . start_time ) * 1000 ) } ms." )
transition = dict ( trigger = "start" , source = "Start" , dest = "Done" , prepare = "prepare_model" ,
before = [ "before_change" ] * 5 + [ "sync_before_change" ],
after = "after_change" ) # execute before function in asynchronously 5 times
model = AsyncModel ()
machine = AsyncMachine ( model , states = [ "Start" , "Done" ], transitions = [ transition ], initial = 'Start' )
asyncio . get_event_loop (). run_until_complete ( model . start ())
# >>> I am synchronous.
# I am asynchronous and will block now for 100 milliseconds.
# I am asynchronous and will block now for 100 milliseconds.
# I am asynchronous and will block now for 100 milliseconds.
# I am asynchronous and will block now for 100 milliseconds.
# I am asynchronous and will block now for 100 milliseconds.
# I am synchronous and will block the event loop (what I probably shouldn't)
# I am done waiting synchronously.
# I am done waiting.
# I am done waiting.
# I am done waiting.
# I am done waiting.
# I am done waiting.
# I am synchronous again. Execution took 101 ms.
assert model . is_Done ()だから、なぜあなたはあなたが尋ねるかもしれないPython 3.7以降を使用する必要があるのですか？ Asyncサポートは以前に導入されました。 AsyncMachine移行が終了する前に新しいイベントが到着したときに実行中のコールバックを処理するためにcontextvarsを使用します。
async def await_never_return ():
await asyncio . sleep ( 100 )
raise ValueError ( "That took too long!" )
async def fix ():
await m2 . fix ()
m1 = AsyncMachine ( states = [ 'A' , 'B' , 'C' ], initial = 'A' , name = "m1" )
m2 = AsyncMachine ( states = [ 'A' , 'B' , 'C' ], initial = 'A' , name = "m2" )
m2 . add_transition ( trigger = 'go' , source = 'A' , dest = 'B' , before = await_never_return )
m2 . add_transition ( trigger = 'fix' , source = 'A' , dest = 'C' )
m1 . add_transition ( trigger = 'go' , source = 'A' , dest = 'B' , after = 'go' )
m1 . add_transition ( trigger = 'go' , source = 'B' , dest = 'C' , after = fix )
asyncio . get_event_loop (). run_until_complete ( asyncio . gather ( m2 . go (), m1 . go ()))
assert m1 . state == m2 . stateこの例は実際に2つのことを示しています。1つ目は、 AからAからBへの移行で呼ばれる「Go」はキャンセルされず、2番目にm2.fix()を呼び出しますB AからCまで。この分離は、 contextvarsなしでは不可能です。 prepareとconditions継続的な移行として扱われないことに注意してください。これは、 conditionsが評価された後、別のイベントがすでに発生しているにもかかわらず、移行が実行されることを意味します。タスクは、コールバックbeforeまたは後期として実行される場合にのみキャンセルされます。
AsyncMachine queued='model'がコンストラクターに渡されるときに使用できるモデル特別なキューモードを備えています。モデル固有のキューを使用すると、イベントは同じモデルに属している場合にのみキューに掲載されます。さらに、拡張された例外は、その例外を提起したモデルのイベントキューをクリアするだけです。簡単にするために、以下のasyncio.gatherのすべてのイベントが同時にトリガーされるのではなく、わずかに遅れていると仮定しましょう。
asyncio . gather ( model1 . event1 (), model1 . event2 (), model2 . event1 ())
# execution order with AsyncMachine(queued=True)
# model1.event1 -> model1.event2 -> model2.event1
# execution order with AsyncMachine(queued='model')
# (model1.event1, model2.event1) -> model1.event2
asyncio . gather ( model1 . event1 (), model1 . error (), model1 . event3 (), model2 . event1 (), model2 . event2 (), model2 . event3 ())
# execution order with AsyncMachine(queued=True)
# model1.event1 -> model1.error
# execution order with AsyncMachine(queued='model')
# (model1.event1, model2.event1) -> (model1.error, model2.event2) -> model2.event3マシンの構築後にキューモードを変更しないでください。
スーパーヒーローがいくつかのカスタム動作を必要とする場合は、マシンの状態を装飾することで、いくつかの追加機能を投入できます。
from time import sleep
from transitions import Machine
from transitions . extensions . states import add_state_features , Tags , Timeout
@ add_state_features ( Tags , Timeout )
class CustomStateMachine ( Machine ):
pass
class SocialSuperhero ( object ):
def __init__ ( self ):
self . entourage = 0
def on_enter_waiting ( self ):
self . entourage += 1
states = [{ 'name' : 'preparing' , 'tags' : [ 'home' , 'busy' ]},
{ 'name' : 'waiting' , 'timeout' : 1 , 'on_timeout' : 'go' },
{ 'name' : 'away' }] # The city needs us!
transitions = [[ 'done' , 'preparing' , 'waiting' ],
[ 'join' , 'waiting' , 'waiting' ], # Entering Waiting again will increase our entourage
[ 'go' , 'waiting' , 'away' ]] # Okay, let' move
hero = SocialSuperhero ()
machine = CustomStateMachine ( model = hero , states = states , transitions = transitions , initial = 'preparing' )
assert hero . state == 'preparing' # Preparing for the night shift
assert machine . get_state ( hero . state ). is_busy # We are at home and busy
hero . done ()
assert hero . state == 'waiting' # Waiting for fellow superheroes to join us
assert hero . entourage == 1 # It's just us so far
sleep ( 0.7 ) # Waiting...
hero . join () # Weeh, we got company
sleep ( 0.5 ) # Waiting...
hero . join () # Even more company o/
sleep ( 2 ) # Waiting...
assert hero . state == 'away' # Impatient superhero already left the building
assert machine . get_state ( hero . state ). is_home is False # Yupp, not at home anymore
assert hero . entourage == 3 # At least he is not alone現在、移行には次の状態機能が装備されています。
タイムアウト- しばらく経った後にイベントをトリガーします
timeout （int、オプション） - 渡された場合、入力された状態はtimeout秒後にタイムアウトしますon_timeout （string/callable、optional） - タイムアウト時間に到達したときに呼び出されますtimeoutが設定されているときにAttributeErrorを上げますが、 on_timeoutはそうではありませんタグ- 状態にタグを追加します
tags （リスト、オプション） - タグを状態に割り当てますState.is_<tag_name>状態がtag_nameでタグ付けされたときにTrueを返します、else Falseエラー- 状態を離れることができないときにMachineError上げる
Tagsから継承する（ Errorを使用する場合はTags使用しない）accepted （bool、optional） - 受け入れられた状態をマークしますtags渡すことができます。auto_transitionsがFalseに設定されている場合にのみ、エラーが発生します。それ以外の場合、すべての状態をto_<state>メソッドで終了できます。揮発性- 状態が入るたびにオブジェクトを初期化する
volatile （クラス、オプション） - 状態が入力されるたびに、タイプクラスのオブジェクトがモデルに割り当てられます。属性名はhookで定義されます。省略すると、代わりに空のvolatileObjectが作成されますhook （string、default = 'scope'） - 時間オブジェクトのモデルの属性名。独自のState拡張機能を作成して、同じ方法で追加できます。 add_state_featuresミックスインが期待されることに注意してください。これは、拡張機能が常にオーバーライドされた方法__init__呼び出し、 enterてexitことを意味します。拡張機能は州から継承される可能性がありますが、それなしでも機能します。 @add_state_featuresを使用すると、装飾されたマシンを漬けることができないという欠点があります（より正確には、動的に生成されたCustomStateを漬けることはできません）。これが、代わりに専用のカスタムステートクラスを書く理由かもしれません。選択した状態マシンによっては、カスタムステートクラスが特定の状態機能を提供する必要がある場合があります。たとえば、 HierarchicalMachineでは、カスタム状態がNestedStateのインスタンスである必要があります（ Stateは十分ではありません）。状態を挿入するには、それらをMachineのクラス属性state_clsに割り当てるか、 Machine.create_stateオーバーライドすることができます。create_state状態が作成されるたびに特定の手順を実行する必要があります。
from transitions import Machine , State
class MyState ( State ):
pass
class CustomMachine ( Machine ):
# Use MyState as state class
state_cls = MyState
class VerboseMachine ( Machine ):
# `Machine._create_state` is a class method but we can
# override it to be an instance method
def _create_state ( self , * args , ** kwargs ):
print ( "Creating a new state with machine '{0}'" . format ( self . name ))
return MyState ( * args , ** kwargs ) AsyncMachineのスレッドを完全に避けたい場合は、 Timeout状態機能をasyncio拡張機能からAsyncTimeoutに置き換えることができます。
import asyncio
from transitions . extensions . states import add_state_features
from transitions . extensions . asyncio import AsyncTimeout , AsyncMachine
@ add_state_features ( AsyncTimeout )
class TimeoutMachine ( AsyncMachine ):
pass
states = [ 'A' , { 'name' : 'B' , 'timeout' : 0.2 , 'on_timeout' : 'to_C' }, 'C' ]
m = TimeoutMachine ( states = states , initial = 'A' , queued = True ) # see remark below
asyncio . run ( asyncio . wait ([ m . to_B (), asyncio . sleep ( 0.1 )]))
assert m . is_B () # timeout shouldn't be triggered
asyncio . run ( asyncio . wait ([ m . to_B (), asyncio . sleep ( 0.3 )]))
assert m . is_C () # now timeout should have been processed queued=True TimeoutMachineコンストラクターにtrueを渡すことを検討する必要があります。これにより、イベントが順次処理され、タイムアウトとイベントが近接して発生したときに現れる可能性のある非同期レース条件を回避できます。
インスピレーションやチェックアウトdjango-transitionsのために、FAQを見ることができます。 Christian Ledermannによって開発されており、Githubでもホストされています。ドキュメントには、いくつかの使用例が含まれています。
まず、おめでとうございます！ドキュメントの終わりに到達しました！インストールする前にtransitions試してみたい場合は、mybinder.orgのインタラクティブなjupyterノートブックでそれを行うことができます。このボタンをクリックするだけです。
バグレポートやその他の問題については、GitHubで問題を開いてください。
使用状況については、スタックオーバーフローに投稿して、 pytransitionsタグで質問にタグを付けてください。拡張された例を忘れずに見ることを忘れないでください！
その他の質問、勧誘、または大規模な無制限の金銭的贈り物については、Tal Yarkoni（初期著者）および/またはAlexander Neumann（現在のメンテナー）にメールしてください。
