transitions
Release 0.9.2
Облегченная объектно-ориентированная реализация конечного автомата на Python со множеством расширений. Совместим с Python 2.7+ и 3.0+.
pip install transitions
... или клонируйте репозиторий с GitHub, а затем:
python setup.py install
Говорят, хороший пример стоит 100 страниц документации API, миллиона директив или тысячи слов.
Ну, "они", наверное, врут... но всё же вот пример:
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 .
Если хотите узнать, как это сделать, ознакомьтесь с расширениями диаграмм.
Конечный автомат — это модель поведения, состоящая из конечного числа состояний и переходов между этими состояниями. Внутри каждого состояния и перехода может быть выполнено некоторое действие . Конечный автомат должен запускаться в некотором начальном состоянии . При использовании 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 в словаре свойств состояния или добавив их позже.
Для удобства всякий раз, когда к Machine добавляется новое State , методы 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' , '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() и вы инициализируете Machine с помощью initial='A' , on_enter_A() не будет запущен до тех пор, пока вы в следующий раз не войдете в состояние A . (Если вам нужно убедиться, что on_enter_A() срабатывает при инициализации, вы можете просто создать фиктивное начальное состояние, а затем явно вызвать to_A() внутри метода __init__ .)
Помимо передачи обратных вызовов при инициализации 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 срабатывает метод on_enter_A() определенный в классе Matter .
Вы можете использовать обратные вызовы on_final , которые будут запускаться при входе в состояние final=True .
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 для текущего состояния, вы можете сделать это с помощью метода get_state() экземпляра Machine .
lump . state
> >> 'solid'
lump . is_gas ()
> >> False
lump . is_solid ()
> >> True
machine . get_state ( lump . state ). name
> >> 'solid' Если вы хотите, вы можете выбрать собственное имя атрибута состояния, передав аргумент model_attribute при инициализации Machine . Это также изменит имя is_«state name»() на is_«model_attribute»_«state name»() . Аналогично, автоматические переходы будут называться to_«model_attribute»_«state name»() вместо to_«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 (или вы просто больше не можете вводить «сескипедалофобию», потому что это слово вас пугает), использование перечислений может быть тем, что вы ищете:
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' )Чтобы переход был выполнен, какое-то событие должно его инициировать . Есть два способа сделать это:
Использование автоматически присоединяемого метода в базовой модели:
> >> 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 и будут представлены в следующем разделе.
В дополнение к любым явно добавленным переходам, метод to_«state»() создается автоматически всякий раз, когда состояние добавляется к экземпляру Machine . Этот метод переходит в целевое состояние независимо от того, в каком состоянии сейчас находится машина:
lump . to_liquid ()
lump . state
> >> 'liquid'
lump . to_solid ()
lump . state
> >> 'solid' При желании вы можете отключить это поведение, установив auto_transitions=False в инициализаторе Machine .
К одному триггеру можно прикрепить несколько переходов, некоторые из которых потенциально могут начинаться или заканчиваться в одном и том же состоянии. Например:
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() установит состояние модели на 'solid' если она в данный момент является 'plasma' , и установит для нее значение 'plasma' в противном случае. (Обратите внимание, что будет выполнен только первый соответствующий переход; таким образом, переход, определенный в последней строке выше, ничего не сделает.)
Вы также можете сделать так, чтобы триггер вызывал переход из всех состояний в определенный пункт назначения, используя подстановочный знак '*' :
machine . add_transition ( 'to_liquid' , '*' , 'liquid' )Обратите внимание, что переходы с подстановочными знаками будут применяться только к состояниям, которые существуют на момент вызова add_transition(). Вызов перехода на основе подстановочных знаков, когда модель находится в состоянии, добавленном после определения перехода, вызовет недопустимое сообщение о переходе и не приведет к переходу в целевое состояние.
Рефлексивный триггер (триггер, имеющий то же состояние, что и источник и пункт назначения) можно легко добавить, указав = в качестве пункта назначения. Это удобно, если один и тот же рефлексивный триггер необходимо добавить к нескольким состояниям. Например:
machine . add_transition ( 'touch' , [ 'liquid' , 'gas' , 'plasma' ], '=' , after = 'change_shape' ) Это добавит рефлексивные переходы для всех трех состояний с touch() в качестве триггера и с change_shape выполняемой после каждого триггера.
В отличие от рефлексивных переходов, внутренние переходы никогда фактически не покидают состояние. Это означает, что обратные вызовы, связанные с переходом, такие как 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 предоставляет метод add_ordered_transitions() в классе Machine :
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!" Поведение по умолчанию в Transitions — мгновенная обработка событий. Это означает, что события внутри метода 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' ]) В приведенном выше примере вызов heat() , когда модель находится в состоянии 'solid' перейдет в состояние 'gas' если is_flammable возвращает True . В противном случае он перейдет в состояние 'liquid' , если is_really_hot возвращает True .
Для удобства существует также аргумент 'unless' , который ведет себя точно так же, как и условия, но в инвертированном виде:
machine . add_transition ( 'heat' , 'solid' , 'gas' , unless = [ 'is_flammable' , 'is_really_hot' ]) В этом случае модель будет переходить от твердого тела к газу при каждом срабатывании heat() , при условии, что и is_flammable() , и is_really_hot() возвращают False .
Обратите внимание, что методы проверки условий будут пассивно получать необязательные аргументы и/или объекты данных, передаваемые триггерным методам. Например, следующий вызов:
lump . heat ( temp = 74 )
# equivalent to lump.trigger('heat', temp=74) ... передал бы необязательный kwarg temp=74 в проверку 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?" Существует также обратный вызов 'prepare' , который выполняется, как только начинается переход, до проверки каких-либо 'conditions' или выполнения других обратных вызовов.
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 не будет вызываться, если текущее состояние не является допустимым источником для именованного перехода.
Действия по умолчанию, которые должны выполняться до или после каждого перехода, могут быть переданы Machine во время инициализации с помощью before_state_change и after_state_change соответственно:
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?" Есть также два ключевых слова для обратных вызовов, которые должны выполняться независимо: а) от того, сколько переходов возможно, б) если какой-либо переход успешен и в) даже если во время выполнения какого-либо другого обратного вызова возникает ошибка. Обратные вызовы, передаваемые в Machine с помощью prepare_event будут выполняться один раз , прежде чем произойдет обработка возможных переходов (и их отдельных обратных вызовов prepare ). Обратные вызовы 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 ) Таким образом, в настоящее время существует три способа запуска событий. Вы можете вызывать удобные функции модели, такие как lump.melt() , выполнять триггеры по имени, например lump.trigger("melt") или отправлять события на несколько моделей с помощью machine.dispatch("melt") (см. раздел о нескольких моделях в альтернативные шаблоны инициализации). Обратные вызовы при переходах затем выполняются в следующем порядке:
| Перезвонить | Текущее состояние | Комментарии |
|---|---|---|
'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.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 для многопоточности.
Иногда вам необходимо передать функциям обратного вызова, зарегистрированным при инициализации машины, некоторые данные, отражающие текущее состояние модели. Переходы позволяют сделать это двумя разными способами.
Во-первых (по умолчанию), вы можете передавать любые позиционные или ключевые аргументы непосредственно в методы триггера (создаваемые при вызове 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.'Вы можете передать триггеру любое количество аргументов.
У этого подхода есть одно важное ограничение: каждая функция обратного вызова, запускаемая переходом состояния, должна иметь возможность обрабатывать все аргументы. Это может вызвать проблемы, если каждый из обратных вызовов ожидает несколько разных данных.
Чтобы обойти эту проблему, Transitions поддерживает альтернативный метод отправки данных. Если вы установите send_event=True при инициализации Machine , все аргументы триггеров будут заключены в экземпляр 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() vs.machine.add_transition machine.add_transition() ).
К счастью, Transitions является гибким и поддерживает два других шаблона инициализации.
Во-первых, вы можете создать автономный конечный автомат, который вообще не требует другой модели. Просто опустите аргумент модели во время инициализации:
machine = Machine ( states = states , transitions = transitions , initial = 'solid' )
machine . melt ()
machine . state
> >> 'liquid' Если вы инициализируете машину таким образом, вы сможете прикрепить все триггерные события (например, evaporate() , sublimate() и т. д.) и все функции обратного вызова непосредственно к экземпляру Machine .
Преимущество этого подхода заключается в объединении всех функций конечного автомата в одном месте, но он может показаться немного неестественным, если вы считаете, что логика состояний должна содержаться внутри самой модели, а не в отдельном контроллере.
Альтернативный (потенциально лучший) подход — наследовать модель от класса Machine . Переходы предназначены для беспрепятственной поддержки наследования. (только не забудьте переопределить метод __init__ класса Machine !):
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.self_literal во время инициализации, например Machine(model=[Machine.self_literal, model1, ...]) . Вы также можете создать автономную машину и динамически регистрировать модели с помощью machine.add_model , передав конструктору model=None . Кроме того, вы можете использовать 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 Переходы включают в себя очень элементарные возможности ведения журнала. Ряд событий, а именно изменения состояния, триггеры перехода и условные проверки, регистрируются как события уровня INFO с использованием стандартного модуля logging Python. Это означает, что вы можете легко настроить ведение журнала на стандартный вывод в скрипте:
# 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" Если вы хотите использовать все удобные функции и добавить к этому несколько обратных вызовов, определение модели может оказаться довольно сложным, если у вас определено много состояний и переходов. generate_base_model в transitions может сгенерировать базовую модель из конфигурации машины, чтобы помочь вам в этом.
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 или событие для присвоения значений атрибутам, это зависит от вашего предпочтительного стиля кода. Оба они работают одинаково, имеют одинаковую сигнатуру и должны давать (почти) одинаковые подсказки типов IDE. Поскольку эта работа все еще находится в стадии разработки, вам нужно будет создать собственный класс Machine и использовать with_model_definitions для переходов, чтобы проверять переходы, определенные таким образом.
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Несмотря на то, что ядро переходов остается легким, существует множество MixIns, расширяющих его функциональность. В настоящее время поддерживаются:
Существует два механизма получения экземпляра конечного автомата с включенными желаемыми функциями. Первый подход использует удобную factory с graph из четырех параметров, nested , locked или asyncio установленным в значение True , если эта функция требуется:
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 . Схема именования следующая:
| Диаграммы | Вложенный | Заблокировано | Асинчио | |
|---|---|---|---|---|
| Машина | ✘ | ✘ | ✘ | ✘ |
| ГрафМашина | ✓ | ✘ | ✘ | ✘ |
| ИерархическаяМашина | ✘ | ✓ | ✘ | ✘ |
| Заблокированная машина | ✘ | ✘ | ✓ | ✘ |
| ИерархическийГрафМашина | ✓ | ✓ | ✘ | ✘ |
| ЗаблокированоГрафМашина | ✓ | ✘ | ✓ | ✘ |
| ЗаблокированоИерархическийМашина | ✘ | ✓ | ✓ | ✘ |
| ЗаблокированоИерархическийГрафМашина | ✓ | ✓ | ✓ | ✘ |
| Асинхронная машина | ✘ | ✘ | ✘ | ✓ |
| АсинхронныйГрафМашина | ✓ | ✘ | ✘ | ✓ |
| ИерархическийAsyncMachine | ✘ | ✓ | ✘ | ✓ |
| ИерархическийАсинхронныйГрафМашина | ✓ | ✓ | ✘ | ✓ |
Чтобы использовать многофункциональный конечный автомат, можно написать:
from transitions . extensions import LockedHierarchicalGraphMachine as LHGMachine
machine = LHGMachine ( model , states , 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' ]
]
# ... initial ключевое слово конструктора HierarchicalMachine принимает вложенные состояния (например, initial='caffeinated_running' ) и список состояний, которые считаются параллельным состоянием (например, initial=['A', 'B'] ) или текущим состоянием другая модель ( initial=model.state ), которая должна фактически быть одним из ранее упомянутых вариантов. Обратите внимание, что при передаче строки transition проверит целевое состояние на наличие initial подсостояний и использует его в качестве входного состояния. Это будет выполняться рекурсивно до тех пор, пока в подсостоянии не будет упоминаться начальное состояние. Параллельные состояния или состояния, переданные в виде списка, будут использоваться «как есть», и дальнейшая первоначальная оценка проводиться не будет.
Обратите внимание, что ранее созданный объект состояния должен быть NestedState или его производным классом. В стандартном классе State , используемом в простых экземплярах Machine , отсутствуют функции, необходимые для вложения.
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 . В настоящее время в качестве разделителя установлено подчеркивание («_»), поэтому его поведение аналогично базовой машине. Это означает, bar подсостояния из состояния foo будет известен foo_bar . baz подсостояния bar будет называться foo_bar_baz и так далее. При входе в подсостояние enter будет вызываться для всех родительских состояний. То же самое справедливо и для выхода из подсостояний. В-третьих, вложенные состояния могут перезаписать переходное поведение своих родителей. Если переход в текущее состояние неизвестен, он будет делегирован его родительскому элементу.
Это означает, что в стандартной конфигурации имена состояний в HSM НЕ ДОЛЖНЫ содержать подчеркивания. Для transitions невозможно определить, должен ли machine.add_state('state_name') добавлять состояние с именем state_name или добавлять name подсостояния к state state . Однако в некоторых случаях этого недостаточно. Например, если имена состояний состоят из более чем одного слова, и вы хотите/должны использовать подчеркивание для их разделения вместо CamelCase . Чтобы справиться с этим, вы можете довольно легко изменить символ, используемый для разделения. Вы даже можете использовать причудливые символы Юникода, если используете Python 3. Установка разделителя на что-то другое, кроме подчеркивания, меняет некоторые аспекты поведения (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() . Если ваше подсостояние начинается с цифры, переходы добавляют префикс «s» («3» становится «s3») к FunctionWrapper автоматического перехода, чтобы соответствовать схеме именования атрибутов Python. Если интерактивное завершение не требуется, 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 При использовании parallel вместо children transitions будут входить во все состояния переданного списка одновременно. Какое подсостояние войти, определяется параметром initial , который всегда должен указывать на прямое подсостояние. Новой особенностью является определение локальных переходов путем передачи ключевого слова transitions в определении состояния. Определенный выше переход ['go', 'a', 'b'] действителен только в C_1 . Хотя вы можете ссылаться на подсостояния, как это сделано в ['go', '2_z', '2_x'] вы не можете ссылаться на родительские состояния напрямую в локально определенных переходах. При выходе из родительского состояния его дочерние элементы также будут завершены. В дополнение к порядку обработки переходов, известному из Machine , где переходы рассматриваются в порядке их добавления, HierarchicalMachine также учитывает иерархию. Переходы, определенные в подсостояниях, будут оцениваться в первую очередь (например, C_1_a остается перед C_2_z ), а переходы, определенные с подстановочным знаком * , (на данный момент) будут добавлять переходы только к корневым состояниям (в этом примере A , B , C ). Начиная с вложенных состояний 0.8.0 может быть добавлен напрямую и выпустит создание родительских государств на лету:
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: вы можете использовать обратные вызовы on_final либо в состояниях, либо на самом HSM. Обратные вызовы будут запускаться, если а) само состояние помечено final и только что введено или б) все субстанции считаются окончательными, и, по крайней мере, одна субстанция только что вошла в окончательное состояние. В случае б) все родители также будут считаться окончательными, если условие б) верно для них. Это может быть полезно в тех случаях, когда обработка происходит параллельно, и ваш 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 экземпляры только что ссылаются , что означает, что изменения в сборе состояний и событий одной машины будут влиять на другой экземпляр машины. Модели и их состояние не будут делиться. Обратите внимание, что события и переходы также копируются с помощью ссылки и будут переданы обоими экземплярами, если вы не используете ключевое слово remap . Это изменение было сделано, чтобы быть в большей степени соответствовать Machine , которая также использует пропущенные экземпляры State посредством ссылки.
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 . Если это нежелательное поведение, и машина должна скорее остановиться в родительском состоянии, пользователь может передать initial как False Like {'name': 'counting', 'children': counter, 'initial': False} .
Иногда вы хотите, чтобы такая встроенная коллекция состояний была «возвращена», что означает, что после того, как это сделано, она должна выйти и перейти в одно из ваших супер -состояний. Чтобы достичь такого поведения, вы можете перенаправить переходы состояния. В приведенном выше примере мы хотели бы, чтобы счетчик вернулся, если будет 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 обратные вызовы (enter, exit), теги и тайм -ауты на графикеПереходы могут генерировать основные диаграммы состояния, отображающие все достоверные переходы между состояниями. Основная поддержка диаграммы генерирует определение машины State Mermaid, которое можно использовать с живым редактором Mermaid, в файлах Markdown в Gitlab или Github и других веб -службах. Например, этот код:
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!" )Создает эту диаграмму (проверьте источник документа, чтобы увидеть нотацию Markdown):
---
Русалка график
---
stateDiagram-V2
Направление LR
ClassDef S_DEFAULT FILL: Белый, Цвет: Черный
ClassDef s_inactive Fill: белый, цвет: черный
ClassDef S_PARALLEL Цвет: черный, заполните: белый
ClassDef S_ACTICE Color: Red, Fill: Darksalmon
ClassDef S_previous Color: Blue, Fill: azure
утверждать "как" как
Класс A S_PREVIOL
Государство "B" как b
Класс B S_ACTICE
Государство "c" как c
C -> [*]
Класс C S_DEFAULT
состояние c {
Государство "1" как c_1
состояние c_1 {
[*] -> C_1_A
Укажите "a" как c_1_a
Установите "B" как C_1_B
C_1_B -> [*]
}
--
Государство "2" как c_2
состояние c_2 {
[*] -> C_2_A
Укажите "a" как c_2_a
Государство "b" как c_2_b
C_2_B -> [*]
}
}
C -> A: сбросить
A -> b: init
B -> c: сделать
C_1_A -> C_1_B: Go
C_2_A -> C_2_B: Go
[*] -> а
Чтобы использовать более сложные функции графика, вам нужно будет установить graphviz и/или pygraphviz . Чтобы сгенерировать графики с помощью Package 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 , когда будет доступен, и вернется в graphviz , когда pygraphviz не может быть найдена. Если 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 None . Кроме того, посмотрите на наш пример ноутбуков Ipython/Jupyter для более подробного примера о том, как использовать и редактировать графики.
В тех случаях, когда отправка событий проводится в потоках, можно использовать либо LockedMachine , либо LockedHierarchicalMachine где доступ (! Sic), где можно закрепить функциональный доступ (! Sic). Это не спасает вас от повреждения вашей машины, возившись с переменными членами вашей модели или штатной машины.
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 или более поздней версии, вы можете спросить. Асинхронная поддержка была введена ранее. 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 Этот пример фактически иллюстрирует две вещи: во -первых, «go», вызванный переходом M1 от A к B не отменен, а второе, вызов m2.fix() остановит попытку перехода M2 от A к 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, необязательно) - если пройдет, введенное состояние будет время ожидания после секунды timeouton_timeout (string/callable, необязательно) - будет вызвано, когда будет достигнуто время ожиданияAttributeError , когда будет установлен timeout , но on_timeout неТеги - добавляет теги в штаты
tags (список, необязательно) - назначает теги в состояниеState.is_<tag_name> вернуть True , когда состояние было помечено с помощью tag_name , иначе False Ошибка - поднимает MachineError
Tags (если вы используете Error не используйте Tags )accepted (Bool, необязательно) - отмечает состояние, как принятоtags ключевых слов, содержащие «принятые»auto_transitions были установлены на False . В противном случае каждое состояние может быть выброшено с помощью методов to_<state> .Нестабильный - инициализирует объект каждый раз, когда вводится состояние
volatile (класс, необязательно) - Каждый раз, когда введено состояние, объект класса типа будет назначен модели. Имя атрибута определено hook . В случае пропущения, вместо этого будет создан пустой волатилеобъект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 на AsyncTimeout из расширения asyncio :
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 . Это убедится, что события обрабатываются последовательно и избегают асинхронных гоночных условий, которые могут появиться, когда тайм -аут и события происходят в непосредственной близости.
Вы можете взглянуть на FAQ для некоторого вдохновения или проверки django-transitions . Он был разработан Кристиан Ледерманн, а также принимается на GitHub. Документация содержит некоторые примеры использования.
Во -первых, поздравляю! Вы достигли конца документации! Если вы хотите попробовать transitions перед его установкой, вы можете сделать это в интерактивной ноутбуке Jupyter на mybinder.org. Просто нажмите эту кнопку.
Для отчетов об ошибках и других вопросов, пожалуйста, откройте проблему на GitHub.
Для вопросов использования публикуйте переполнение стека, убедившись пометить ваш вопрос с тегом pytransitions . Не забудьте взглянуть на расширенные примеры!
Для любых других вопросов, ходатайств или больших неограниченных денежных подарков по электронной почте Tal Yarkoni (первоначальный автор) и/или Александр Нейман (текущий сопровождающий).
