GitHubじゃ!Pythonじゃ!

GitHubからPython関係の優良リポジトリを探したかったのじゃー、でも英語は出来ないから日本語で読むのじゃー、英語社会世知辛いのじゃー

pytransitions

transitions – 軽量でオブジェクト指向のFinite State MachineをPythonで実装

投稿日:

軽量でオブジェクト指向のFinite State MachineをPythonで実装

トランジション

軽量でオブジェクト指向の状態マシンのPython実装。 Python 2.7+および3.0+と互換性があります。

インストール

pip install transitions

…またはGitHubからレポをクローンしてから:

python setup.py install

目次

クイックスタート

彼らは 100ページのAPIドキュメンテーション、100万のディレクティブ、または1000ワードの価値があるとしています。

さて、「彼ら」はおそらく嘘をついています…しかし、とにかくここに例があります:

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

    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

非クイックスタート

基本的な初期化

ステートマシンを稼働させることは非常に簡単です。 オブジェクトのlumpMatterクラスのインスタンス)があり、その状態を管理したいとしましょう:

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 state!
lump.state
>>> 'solid'

この状態機械は技術的には機能しているが、実際に何もしないので、私は「最小限」と言います。 それは'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...
lump.evaporate()
lump.state
>>> 'gas'
lump.trigger('ionize')
lump.state
>>> 'plasma'

Matterインスタンスに接続された光沢のある新しいメソッド( evaporate()ionize()など)に注目してください。 各メソッドは、対応する遷移をトリガします。 これらのメソッドをどこにでも明示的に定義する必要はありません。 各遷移の名前は、 Machineイニシャライザに渡されたモデルにバインドされます(この場合、 lump )。 また、モデルにtriggerが追加されました。 このメソッドを使用すると、動的トリガが必要な場合に名前でトランジションを実行できます。

良い状態マシン(そして多くの悪いマシンの魂)は、状態の集合です。 上では、文字列のリストをMachine初期化子に渡すことで、有効なモデル状態を定義しました。 しかし、内部的には、状態は実際にはStateオブジェクトとして表されます。

状態を初期化したり変更したりするには、さまざまな方法があります。 具体的には、次のことができます。

  • 文字列を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])

ステートはマシンに追加されると一度だけ初期化され、ステートから削除されるまで保持されます。 言い換えれば、状態オブジェクトの属性を変更すると、次回その状態に入るときにこの変更がリセットされることはありません。 他の動作が必要な場合に備え状態機能拡張する方法を見てください。

コールバック

ステートマシンは、ステートマシンがそのステートに入る、またはステートを出るたびに呼び出される、 enterおよびexitコールバックのリストに関連付けることもできます。 初期化中にコールバックを指定することも、後でコールバックを追加することもできます。

便宜上、新しいStateMachineに追加されるたびに、 on_enter_«state name»およびon_exit_«state name»のメソッドはマシン上で動的に作成されます(モデルではありません)。後でコールバックを必要とする場合はコールバック。

# 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' }
    ]

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'Machineinitial='A'化すると、次にA状態に入るまでon_enter_A()は起動しません。 (初期化時にon_enter_A()が確実にon_enter_A()ようにする必要がある場合は、ダミーの初期状態を作成し、 to_A()メソッドの中で明示的にto_A()呼び出すことができます)。

状態の初期化時にコールバックを渡すか、動的に追加するだけでなく、モデルクラス自体でコールバックを定義することもできます。これにより、コードの明瞭性が向上する可能性があります。 例えば:

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'])

これで、状態Aに遷移するon_enter_A()に、 Matterクラスで定義されたon_enter_A()メソッドがon_enter_A()ます。

状態の確認

次のいずれかの方法で、モデルの現在の状態をいつでも確認できます。

  • .state属性を検査する、または
  • is_«state name»()呼び出す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'

トランジション

上の例のいくつかはすでに渡す際のトランジションの使用を示していますが、ここでそれらをさらに詳しく調べます。

状態と同様に、各トランジションは、自身のオブジェクトとして内部的に表現されます。 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初期化(つまり、 triggersourcedestinationなど)の位置引数と同じ順序になっていることを確認してください。

以下のリストのリストは、上記の辞書のリストと機能的に同等です:

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')

trigger引数は、ベースモデルにアタッチされる新しいトリガーメソッドの名前を定義します。 このメソッドが呼び出されると、遷移を実行しようとします:

>>> lump.melt()
>>> lump.state
'liquid'

デフォルトでは、無効なトリガーを呼び出すと例外が発生します。

>>> 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']

すべての状態の自動遷移

明示的に追加されたトランジションに加えて、ステートがMachineインスタンスに追加されるたびにto_«state»()メソッドが自動的に作成されます。 このメソッドは、マシンが現在どの状態にあるかにかかわらず、ターゲット状態に遷移します。

lump.to_liquid()
lump.state
>>> 'liquid'
lump.to_solid()
lump.state
>>> 'solid'

必要にauto_transitions=Falseて、 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つの状態すべてに対して反射的な遷移を追加します。

順序付けられた遷移

一般的な要望は、状態遷移が厳密な線形シーケンスに従うことである。 たとえば、与えられた状態['A', 'B', 'C']では、 ABBCCA (しかし他のペアはありません)に対して有効な遷移を望むかもしれません。

この動作を容易にするために、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'

キューに入れられた遷移

Transitionsのデフォルトの動作は、イベントを即座に処理することです。 つまり、 on_enterメソッド内のイベントは、 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)

# 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)
...
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 返しTrue 。キューに入れられたコールを含む遷移が最終的に正常に完了するかどうかを判断する方法がないためです。 これは、単一のイベントだけが処理されている場合にも当てはまります。

machine.add_transition('jump', 'A', 'C', conditions='will_fail')
...
# queued=False
machine.jump()
>>> False
# queued=True
machine.jump()
>>> True

条件付き遷移

場合によっては、特定の条件が発生した場合にのみ、特定の遷移を実行することが必要な場合もあります。 これを行うには、 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'])

上記の例では、 is_flammableTrue返す場合、モデルが'solid'状態'solid'ときにheat()呼び出すとstate 'gas'遷移します。 それ以外の場合は、 is_really_hotTrue返すと状態'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 is_flammable()オプション)kwargをis_flammable()チェック(おそらくEventDataインスタンスにラップされてEventDataます)にEventDataます。 詳細については、以下の「 データ受け渡し 」を参照してください。

コールバック

トランジションとステートにコールバックをアタッチすることができます。 すべての遷移には、遷移前と遷移後に呼び出すメソッドのリストを含む'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 is_really_hot(self): return self.heat
    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)

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_changebefore_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され、 send_event=True取得できることにevent_dataerror

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

実行命令

要約すると、トランジションのコールバックは次の順序で実行されます。

折り返し電話 現在の状態 コメント
'machine.prepare_event' source 個々のトランジションが処理される前に一度実行される
'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.after_state_change' destination モデルで宣言されたデフォルトのコールバック
'machine.finalize_event' source/destination 遷移が起こらなかったり、例外が発生してもコールバックは実行されます

データを渡す

時には、マシンの初期化時に登録されたコールバック関数を、モデルの現在の状態を反映するデータに渡す必要がある場合もあります。 トランジションでは、これを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.'

代替の初期化パターン

今までのすべての例では、別のモデル( lump 、クラスMatterインスタンス)に新しいMachineインスタンスをアタッチしました。 While this separation keeps things tidy (because you don’t have to monkey patch a whole bunch of new methods into the Matter class), it can also get annoying, since it requires you to keep track of which methods are called on the state machine, and which ones are called on the model that the state machine is bound to (eg, lump.on_enter_StateA() vs. machine.add_transition() ).

Fortunately, Transitions is flexible, and supports two other initialization patterns.

First, you can create a standalone state machine that doesn’t require another model at all. Simply omit the model argument during initialization:

machine = Machine(states=states, transitions=transitions, initial='solid')
machine.melt()
machine.state
>>> 'liquid'

If you initialize the machine this way, you can then attach all triggering events (like evaporate() , sublimate() , etc.) and all callback functions directly to the Machine instance.

This approach has the benefit of consolidating all of the state machine functionality in one place, but can feel a little bit unnatural if you think state logic should be contained within the model itself rather than in a separate controller.

An alternative (potentially better) approach is to have the model inherit from the Machine class. Transitions is designed to support inheritance seamlessly. (just be sure to override class Machine ‘s __init__ method!):

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'

Here you get to consolidate all state machine functionality into your existing model, which often feels more natural way than sticking all of the functionality we want in a separate standalone Machine instance.

A machine can handle multiple models which can be passed as a list like Machine(model=[model1, model2, ...]) . In cases where you want to add models as well as the machine instance itself, you can pass the string placeholder 'self' during initialization like Machine(model=['self', model1, ...]) . You can also create a standalone machine, and register models dynamically via machine.add_model . Remember to call machine.remove_model if machine is long-lasting and your models are temporary and should be garbage collected:

class Matter():
    pass

lump1 = Matter()
lump2 = Matter()

machine = Machine(states=states, transitions=transitions, initial='solid', add_self=False)

machine.add_model(lump1)
machine.add_model(lump2, initial='liquid')

lump1.state
>>> 'solid'
lump2.state
>>> 'liquid'

machine.remove_model([lump1, lump2])
del lump1  # lump1 is garbage collected
del lump2  # lump2 is garbage collected

If you don’t provide an initial state in the state machine constructor, you must provide one every time you add a model:

machine = Machine(states=states, transitions=transitions, add_self=False)

machine.add_model(Matter())
>>> "MachineError: No initial state configured for machine, must specify when adding model."
machine.add_model(Matter(), initial='liquid')

ロギング

Transitions includes very rudimentary logging capabilities. A number of events – namely, state changes, transition triggers, and conditional checks – are logged as INFO-level events using the standard Python logging module. This means you can easily configure logging to standard output in a script:

# 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')
...

(Re-)Storing machine instances

Machines are picklable and can be stored and loaded with pickle . For Python 3.3 and earlier dill is required.

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']

拡張機能

Even though the core of transitions is kept lightweight, there are a variety of MixIns to extend its functionality. Currently supported are:

  • Diagrams to visualize the current state of a machine
  • Hierarchical State Machines for nesting and reuse
  • Threadsafe Locks for parallel execution
  • Custom States for extended state-related behaviour

There are two mechanisms to retrieve a state machine instance with the desired features enabled. The first approach makes use of the convenience factory with the three parameters graph , nested and locked set to True if the certain feature is required:

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)

# 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)

This approach targets experimental use since in this case the underlying classes do not have to be known. However, classes can also be directly imported from transitions.extensions . The naming scheme is as follows:

ダイアグラム ネストされた ロックされた
機械
GraphMachine
HierarchicalMachine
LockedMachine
HierarchicalGraphMachine
LockedGraphMachine
LockedHierarchicalMachine
LockedHierarchicalGraphMachine

To use a full featured state machine, one could write:

from transitions.extensions import LockedHierarchicalGraphMachine as Machine

#enable ALL the features!
machine = Machine(model, states, transitions)

ダイアグラム

追加キーワード:

  • title (optional): Sets the title of the generated image.
  • show_conditions (default False): Shows conditions at transition edges
  • show_auto_transitions (default False): Shows auto transitions in graph

Transitions can generate basic state diagrams displaying all valid transitions between states. To use the graphing functionality, you’ll need to have pygraphviz installed:

pip install pygraphviz  # install pygraphviz manually...
pip install transitions[diagrams]  # ... or install transitions with 'diagrams' extras

With GraphMachine enabled, a PyGraphviz AGraph object is generated during machine initialization and is constantly updated when the machine state changes:

from transitions.extensions import GraphMachine as Machine
m = Model()
machine = Machine(model=m, ...)
# in cases where auto transitions should be visible
# Machine(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)
m.get_graph(show_roi=True).draw('my_state_diagram.png', prog='dot')

This produces something like this:

Also, have a look at our example IPython/Jupyter notebooks for a more detailed example.

Hierarchical State Machine (HSM)

Transitions includes an extension module which allows to nest states. This allows to create contexts and to model cases where states are related to certain subtasks in the state machine. To create a nested state, either import NestedState from transitions or use a dictionary with the initialization arguments name and children . Optionally, initial can be used to define a sub state to transit to, when the nested state is entered.

from transitions.extensions import HierarchicalMachine as Machine

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 = Machine(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')

A configuration making use of initial could look like this:

# ...
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']
]
# ...

Some things that have to be considered when working with nested states: State names are concatenated with NestedState.separator . Currently the separator is set to underscore (‘_’) and therefore behaves similar to the basic machine. This means a substate bar from state foo will be known by foo_bar . A substate baz of bar will be referred to as foo_bar_baz and so on. When entering a substate, enter will be called for all parent states. The same is true for exiting substates. Third, nested states can overwrite transition behaviour of their parents. If a transition is not known to the current state it will be delegated to its parent.

In some cases underscore as a separator is not sufficient. For instance if state names consists of more than one word and a concatenated naming such as state_A_name_state_C would be confusing. Setting the separator to something else than underscore changes some of the behaviour (auto_transition and setting callbacks). You can even use unicode characters if you use python 3:

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 = Machine(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'
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')

Instead of to_C_3_a() auto transition is called as to_C.s3.a() . If your substate starts with a digit, transitions adds a prefix ‘s’ (‘3’ becomes ‘s3’) to the auto transition FunctionWrapper to comply with the attribute naming scheme of python. If interactive completion is not required, to('C↦3↦a') can be called directly. Additionally, on_enter/exit_<<state name>> is replaced with on_enter/exit(state_name, callback) .

To check whether the current state is a substate of a specific state is_state supports the keyword allow_substates :

machine.state
>>> 'C.2.a'
machine.is_C() # checks for specific states
>>> False
machine.is_C(allow_substates=True)
>>> True

Reuse of previously created HSMs

Besides semantic order, nested states are very handy if you want to specify state machines for specific tasks and plan to reuse them. Be aware that this will embed the passed machine’s states. This means if your states had been altered before , this change will be persistent.

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 = Machine(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 = Machine(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

If a HierarchicalStateMachine is passed with the children keyword, the initial state of this machine will be assigned to the new parent state. In the above example we see that entering counting will also enter counting_1 . If this is undesired behaviour and the machine should rather halt in the parent state, the user can pass initial as False like {'name': 'counting', 'children': counter, 'initial': False} .

Sometimes you want such an embedded state collection to ‘return’ which means after it is done it should exit and transit to one of your states. To achieve this behaviour you can remap state transitions. In the example above we would like the counter to return if the state done was reached. This is done as follows:

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

If a reused state machine does not have a final state, you can of course add the transitions manually. If ‘counter’ had no ‘done’ state, we could just add ['done', 'counter_3', 'waiting'] to achieve the same behaviour.

Note that the HierarchicalMachine will not integrate the machine instance itself but the states and transitions by creating copies of them. This way you are able to continue using your previously created instance without interfering with the embedded version.

Threadsafe(-ish) State Machine

In cases where event dispatching is done in threads, one can use either LockedMachine or LockedHierarchicalMachine where function access (!sic) is secured with reentrant locks. This does not save you from corrupting your machine by tinkering with member variables of your model or state machine.

from transitions.extensions import LockedMachine as Machine
from threading import Thread
import time

states = ['A', 'B', 'C']
machine = Machine(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

Any python context manager can be passed in via the machine_context keyword argument:

from transitions.extensions import LockedMachine as Machine
from threading import RLock

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

lock1 = RLock()
lock2 = RLock()

machine = Machine(states=states, initial='A', machine_context=[lock1, lock2])

Any contexts via machine_model will be shared between all models registered with the Machine . Per-model contexts can be added as well:

lock3 = RLock()

machine.add_model(model, model_context=lock3)

It’s important that all user-provided context managers are re-entrant since the state machine will call them multiple times, even in the context of a single trigger invocation.

Adding features to states

If your superheroes need some custom behaviour, you can throw in some extra functionality by decorating machine states:

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

Currently, transitions comes equipped with the following state features:

  • Timeout — triggers an event after some time has passed

    • keyword: timeout (int, optional) — if passed, an entered state will timeout after timeout seconds
    • keyword: on_timeout (string/callable, optional) — will be called when timeout time has been reached
    • will raise an AttributeError when timeout is set but on_timeout is not
  • Tags — adds tags to states

    • keyword: tags (list, optional) — assigns tags to a state
    • State.is_<tag_name> will return True when the state has been tagged with tag_name , else False
  • Error — raises a MachineError when a state cannot be left

    • inherits from Tags (if you use Error do not use Tags )
    • keyword: accepted (bool, optional) — marks a state as accepted
    • alternatively the keyword tags can be passed, containing ‘accepted’
  • Volatile — initialises an object every time a state is entered

    • keyword: volatile (class, optional) — every time the state is entered an object of type class will be assigned to the model. The attribute name is defined by hook . If omitted, an empty VolatileObject will be created instead
    • keyword: hook (string, default=’scope’) — The model’s attribute name fore the temporal object.

You can write your own State extensions and add them the same way. Just note that add_state_features expects Mixins . This means your extension should always call the overridden methods __init__ , enter and exit . Your extension may inherit from State but will also work without it. In case you prefer to write your own custom states from scratch be aware that some state extensions require certain state features. HierarchicalStateMachine requires your custom state to be an instance of NestedState ( State is not sufficient). To inject your states you can either assign them to your Machine ‘s class attribute state_cls or override Machine.create_state in case you need some specific procedures done whenever a state is created:

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)

I have a [bug report/issue/question]…

For bug reports and other issues, please open an issue on GitHub.

For usage questions, post on Stack Overflow, making sure to tag your question with the transitions and python tags. Do not forget to have a look at the extended examples !

For any other questions, solicitations, or large unrestricted monetary gifts, email Tal Yarkoni .







-pytransitions

執筆者: