transitions
Release 0.9.2
Uma implementação de máquina de estado leve e orientada a objetos em Python com muitas extensões. Compatível com Python 2.7+ e 3.0+.
pip install transitions
... ou clone o repositório do GitHub e então:
python setup.py install
Dizem que um bom exemplo vale 100 páginas de documentação de API, um milhão de diretivas ou mil palavras.
Bem, "eles" provavelmente mentem... mas aqui está um exemplo:
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?" ) Pronto, agora você incorporou uma máquina de estado em NarcolepticSuperhero . Vamos levá-lo para dar uma volta...
> >> 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 Embora não possamos ler a mente do Batman real, certamente podemos visualizar o estado atual do nosso NarcolepticSuperhero .
Dê uma olhada nas extensões de diagramas se quiser saber como.
Uma máquina de estados é um modelo de comportamento composto por um número finito de estados e transições entre esses estados. Dentro de cada estado e transição alguma ação pode ser executada. Uma máquina de estados precisa começar em algum estado inicial . Ao utilizar transitions , uma máquina de estados pode consistir em múltiplos objetos onde alguns ( máquinas ) contêm definições para a manipulação de outros ( modelos ). A seguir, veremos alguns conceitos básicos e como trabalhar com eles.
Estado . Um estado representa uma condição ou estágio específico na máquina de estados. É um modo distinto de comportamento ou fase de um processo.
Transição . Este é o processo ou evento que faz com que a máquina de estados mude de um estado para outro.
Modelo . A estrutura com estado real. É a entidade que é atualizada durante as transições. Também pode definir ações que serão executadas durante as transições. Por exemplo, logo antes de uma transição ou quando um estado é inserido ou encerrado.
Máquina . Esta é a entidade que gerencia e controla o modelo, estados, transições e ações. É o maestro que orquestra todo o processo da máquina estatal.
Acionar . Este é o evento que inicia uma transição, o método que envia o sinal para iniciar uma transição.
Ação . Operação ou tarefa específica executada quando um determinado estado é inserido, encerrado ou durante uma transição. A ação é implementada através de callbacks , que são funções que são executadas quando algum evento acontece.
Colocar uma máquina de estado em funcionamento é muito simples. Digamos que você tenha o objeto lump (uma instância da classe Matter ) e queira gerenciar seus estados:
class Matter ( object ):
pass
lump = Matter () Você pode inicializar uma máquina de estado funcional ( mínima ) vinculada ao lump do modelo assim:
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' Uma alternativa é não passar explicitamente um modelo para o inicializador Machine :
machine = Machine ( states = [ 'solid' , 'liquid' , 'gas' , 'plasma' ], initial = 'solid' )
# The machine instance itself now acts as a model
machine . state
> >> 'solid' Observe que desta vez não passei o modelo lump como argumento. O primeiro argumento passado para Machine atua como modelo. Então quando eu passar algo lá, todas as funções de conveniência serão adicionadas ao objeto. Se nenhum modelo for fornecido, a própria instância machine atuará como modelo.
Quando no início eu disse “mínimo”, foi porque embora esta máquina de estado esteja tecnicamente operacional, na verdade não faz nada. Começa no estado 'solid' , mas nunca passará para outro estado, porque nenhuma transição está definida... ainda!
Vamos tentar novamente.
# 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' Observe os novos métodos anexados à instância Matter ( evaporate() , ionize() , etc.). Cada método desencadeia a transição correspondente. As transições também podem ser acionadas dinamicamente chamando o método trigger() fornecido com o nome da transição, conforme mostrado acima. Mais sobre isso na seção Acionando uma transição.
A alma de qualquer boa máquina estatal (e de muitas más, sem dúvida) é um conjunto de estados. Acima, definimos os estados válidos do modelo passando uma lista de strings para o inicializador Machine . Mas internamente, os estados são, na verdade, representados como objetos State .
Você pode inicializar e modificar estados de diversas maneiras. Especificamente, você pode:
Machine fornecendo o(s) nome(s) do(s) estado(s), ouState ouOs trechos a seguir ilustram várias maneiras de atingir o mesmo objetivo:
# 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 ])Os estados são inicializados uma vez quando adicionados à máquina e persistirão até serem removidos dela. Em outras palavras: se você alterar os atributos de um objeto de estado, essa alteração NÃO será redefinida na próxima vez que você entrar nesse estado. Dê uma olhada em como estender os recursos de estado caso você precise de algum outro comportamento.
Mas apenas ter estados e poder movimentar-se entre eles (transições) não é muito útil por si só. E se você quiser fazer algo, realizar alguma ação ao entrar ou sair de um estado? É aqui que entram os retornos de chamada .
Um State também pode ser associado a uma lista de retornos de chamada enter e exit , que são chamados sempre que a máquina de estado entra ou sai desse estado. Você pode especificar retornos de chamada durante a inicialização, passando-os para um construtor de objeto State , em um dicionário de propriedades de estado ou adicionando-os posteriormente.
Por conveniência, sempre que um novo State é adicionado a uma Machine , os métodos on_enter_«state name» e on_exit_«state name» são criados dinamicamente na Machine (não no modelo!), o que permite adicionar dinamicamente novas entradas e saídas retornos de chamada mais tarde, se você precisar deles.
# 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!' Observe que o retorno de chamada on_enter_«state name» não será acionado quando uma máquina for inicializada pela primeira vez. Por exemplo, se você tiver um retorno de chamada on_enter_A() definido e inicializar a Machine com initial='A' , on_enter_A() não será acionado até a próxima vez que você entrar no estado A . (Se você precisar ter certeza de que on_enter_A() dispara na inicialização, você pode simplesmente criar um estado inicial fictício e então chamar explicitamente to_A() dentro do método __init__ .)
Além de passar retornos de chamada ao inicializar um State ou adicioná-los dinamicamente, também é possível definir retornos de chamada na própria classe do modelo, o que pode aumentar a clareza do código. Por exemplo:
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' ]) Agora, sempre que lump transitar para o estado A , o método on_enter_A() definido na classe Matter será acionado.
Você pode usar retornos de chamada on_final que serão acionados quando um estado com final=True for inserido.
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 ... Você sempre pode verificar o estado atual do modelo:
.state ouis_«state name»() E se quiser recuperar o objeto State real para o estado atual, você pode fazer isso por meio do método get_state() da instância Machine .
lump . state
> >> 'solid'
lump . is_gas ()
> >> False
lump . is_solid ()
> >> True
machine . get_state ( lump . state ). name
> >> 'solid' Se desejar, você pode escolher seu próprio nome de atributo de estado passando o argumento model_attribute ao inicializar Machine . Isso também mudará o nome de is_«state name»() para is_«model_attribute»_«state name»() . Da mesma forma, as transições automáticas serão nomeadas to_«model_attribute»_«state name»() em vez de to_«state name»() . Isso é feito para permitir que várias máquinas trabalhem no mesmo modelo com nomes de atributos de estado individuais.
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 Até agora vimos como podemos dar nomes de estados e usar esses nomes para trabalhar com nossa máquina de estados. Se você prefere uma digitação mais rigorosa e mais preenchimento de código IDE (ou simplesmente não consegue mais digitar 'sesquipedalofobia' porque a palavra te assusta), usar Enumerações pode ser o que você está procurando:
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 Você pode misturar enums e strings se quiser (por exemplo [States.RED, 'ORANGE', States.YELLOW, States.GREEN] ), mas observe que internamente, transitions ainda tratarão os estados por nome ( enum.Enum.name ). Assim, não é possível ter os estados 'GREEN' e States.GREEN ao mesmo tempo.
Alguns dos exemplos acima já ilustram o uso de transições na passagem, mas aqui iremos explorá-los com mais detalhes.
Tal como acontece com os estados, cada transição é representada internamente como seu próprio objeto – uma instância da classe Transition . A maneira mais rápida de inicializar um conjunto de transições é passar um dicionário, ou lista de dicionários, para o inicializador Machine . Já vimos isso acima:
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 ) Definir transições em dicionários traz o benefício da clareza, mas pode ser complicado. Se você deseja brevidade, pode optar por definir transições usando listas. Apenas certifique-se de que os elementos em cada lista estejam na mesma ordem que os argumentos posicionais na inicialização Transition (ou seja, trigger , source , destination , etc.).
A seguinte lista de listas é funcionalmente equivalente à lista de dicionários acima:
transitions = [
[ 'melt' , 'solid' , 'liquid' ],
[ 'evaporate' , 'liquid' , 'gas' ],
[ 'sublimate' , 'solid' , 'gas' ],
[ 'ionize' , 'gas' , 'plasma' ]
] Alternativamente, você pode adicionar transições a uma Machine após a inicialização:
machine = Machine ( model = lump , states = states , initial = 'solid' )
machine . add_transition ( 'melt' , source = 'solid' , dest = 'liquid' )Para que uma transição seja executada, algum evento precisa acioná -la. Existem duas maneiras de fazer isso:
Usando o método anexado automaticamente no modelo base:
> >> lump . melt ()
> >> lump . state
'liquid'
> >> lump . evaporate ()
> >> lump . state
'gas' Observe como você não precisa definir explicitamente esses métodos em nenhum lugar; o nome de cada transição está vinculado ao modelo passado para o inicializador Machine (neste caso, lump ). Isso também significa que seu modelo ainda não deve conter métodos com o mesmo nome dos gatilhos de eventos, pois transitions só anexarão métodos de conveniência ao seu modelo se o local ainda não estiver ocupado. Se você quiser modificar esse comportamento, dê uma olhada no FAQ.
Usando o método trigger , agora anexado ao seu modelo (se ainda não existia). Este método permite executar transições por nome caso seja necessário o acionamento dinâmico:
> >> lump . trigger ( 'melt' )
> >> lump . state
'liquid'
> >> lump . trigger ( 'evaporate' )
> >> lump . state
'gas' Por padrão, acionar uma transição inválida gerará uma exceção:
> >> 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!" Esse comportamento geralmente é desejável, pois ajuda a alertá-lo sobre problemas no seu código. Mas, em alguns casos, você pode querer ignorar silenciosamente os gatilhos inválidos. Você pode fazer isso definindo ignore_invalid_triggers=True (estado por estado ou globalmente para todos os estados):
> >> # 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 ) Se precisar saber quais transições são válidas a partir de um determinado estado, você pode usar 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' ] Se você seguiu esta documentação desde o início, notará que get_triggers na verdade retorna mais gatilhos do que os definidos explicitamente mostrados acima, como to_liquid e assim por diante. Elas são chamadas auto-transitions e serão apresentadas na próxima seção.
Além de quaisquer transições adicionadas explicitamente, um método to_«state»() é criado automaticamente sempre que um estado é adicionado a uma instância Machine . Este método faz a transição para o estado de destino, independentemente do estado em que a máquina está atualmente:
lump . to_liquid ()
lump . state
> >> 'liquid'
lump . to_solid ()
lump . state
> >> 'solid' Se desejar, você pode desativar esse comportamento definindo auto_transitions=False no inicializador Machine .
Um determinado gatilho pode ser anexado a múltiplas transições, algumas das quais podem potencialmente começar ou terminar no mesmo estado. Por exemplo:
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' ) Nesse caso, chamar transmogrify() definirá o estado do modelo como 'solid' se estiver atualmente 'plasma' e como 'plasma' caso contrário. (Observe que apenas a primeira transição correspondente será executada; portanto, a transição definida na última linha acima não fará nada.)
Você também pode fazer com que um gatilho cause uma transição de todos os estados para um destino específico usando o curinga '*' :
machine . add_transition ( 'to_liquid' , '*' , 'liquid' )Observe que as transições curinga serão aplicadas apenas aos estados que existem no momento da chamada add_transition(). Chamar uma transição baseada em curinga quando o modelo estiver em um estado adicionado após a definição da transição gerará uma mensagem de transição inválida e não fará a transição para o estado de destino.
Um gatilho reflexivo (gatilho que possui o mesmo estado de origem e destino) pode ser facilmente adicionado especificando = como destino. Isso é útil se o mesmo gatilho reflexivo precisar ser adicionado a vários estados. Por exemplo:
machine . add_transition ( 'touch' , [ 'liquid' , 'gas' , 'plasma' ], '=' , after = 'change_shape' ) Isso adicionará transições reflexivas para todos os três estados com touch() como gatilho e com change_shape executado após cada gatilho.
Em contraste com as transições reflexivas, as transições internas nunca sairão realmente do estado. Isso significa que os retornos de chamada relacionados à transição, como before ou after serão processados, enquanto os retornos de chamada relacionados ao estado, exit ou enter , não. Para definir uma transição como interna, defina o destino como None .
machine . add_transition ( 'internal' , [ 'liquid' , 'gas' ], None , after = 'change_shape' ) Um desejo comum é que as transições de estado sigam uma sequência linear estrita. Por exemplo, dados os estados ['A', 'B', 'C'] , você pode querer transições válidas para A → B , B → C e C → A (mas nenhum outro par).
Para facilitar esse comportamento, Transitions fornece um método add_ordered_transitions() na classe 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!" O comportamento padrão em Transitions é processar eventos instantaneamente. Isso significa que os eventos dentro de um método on_enter serão processados antes que os retornos de chamada vinculados ao after sejam chamados.
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?A ordem de execução deste exemplo é
prepare -> before -> on_enter_B -> on_enter_C -> after.
Se o processamento em fila estiver ativado, uma transição será concluída antes que a próxima transição seja acionada:
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!Isto resulta em
prepare -> before -> on_enter_B -> queue(to_C) -> after -> on_enter_C.
Observação importante: ao processar eventos em uma fila, a chamada do acionador sempre retornará True , pois não há como determinar, no momento da fila, se uma transição envolvendo chamadas na fila será concluída com êxito. Isto é verdade mesmo quando apenas um único evento é processado.
machine . add_transition ( 'jump' , 'A' , 'C' , conditions = 'will_fail' )
...
# queued=False
machine . jump ()
> >> False
# queued=True
machine . jump ()
> >> True Quando um modelo é removido da máquina, transitions também removerão todos os eventos relacionados da fila.
class Model :
def on_enter_B ( self ):
self . to_C () # add event to queue ...
self . machine . remove_model ( self ) # aaaand it's gone Às vezes, você só deseja que uma transição específica seja executada se ocorrer uma condição específica. Você pode fazer isso passando um método, ou lista de métodos, no argumento 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' ]) No exemplo acima, chamar heat() quando o modelo estiver no estado 'solid' fará a transição para o estado 'gas' se is_flammable retornar True . Caso contrário, ele fará a transição para o estado 'liquid' se is_really_hot retornar True .
Por conveniência, há também um argumento 'unless' que se comporta exatamente como as condições, mas invertido:
machine . add_transition ( 'heat' , 'solid' , 'gas' , unless = [ 'is_flammable' , 'is_really_hot' ]) Nesse caso, o modelo faria a transição de sólido para gasoso sempre que heat() disparasse, desde que is_flammable() e is_really_hot() retornassem False .
Observe que os métodos de verificação de condição receberão passivamente argumentos opcionais e/ou objetos de dados passados para métodos de disparo. Por exemplo, a seguinte chamada:
lump . heat ( temp = 74 )
# equivalent to lump.trigger('heat', temp=74) ... passaria o kwarg opcional temp=74 para a verificação is_flammable() (possivelmente agrupada em uma instância EventData ). Para obter mais informações, consulte a seção Transmissão de dados abaixo.
Se quiser ter certeza de que uma transição é possível antes de prosseguir com ela, você pode usar as funções may_<trigger_name> que foram adicionadas ao seu modelo. Seu modelo também contém a função may_trigger para verificar um gatilho pelo nome:
# check if the current temperature is hot enough to trigger a transition
if lump . may_heat ():
# if lump.may_trigger("heat"):
lump . heat () Isso executará todos os callbacks prepare e avaliará as condições atribuídas às transições potenciais. As verificações de transição também podem ser usadas quando o destino de uma transição (ainda) não está disponível:
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 Você pode anexar retornos de chamada a transições e também a estados. Cada transição tem atributos 'before' e 'after' que contêm uma lista de métodos a serem chamados antes e depois da execução da transição:
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?" Há também um retorno de chamada 'prepare' que é executado assim que uma transição começa, antes de quaisquer 'conditions' serem verificadas ou outros retornos de chamada serem executados.
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!" Observe que prepare não será chamado a menos que o estado atual seja uma fonte válida para a transição nomeada.
As ações padrão destinadas a serem executadas antes ou depois de cada transição podem ser passadas para Machine durante a inicialização com before_state_change e after_state_change respectivamente:
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?" Existem também duas palavras-chave para retornos de chamada que devem ser executados independentemente: a) de quantas transições são possíveis, b) se alguma transição for bem-sucedida e c) mesmo se um erro for gerado durante a execução de algum outro retorno de chamada. Os retornos de chamada passados para Machine com prepare_event serão executados uma vez antes do processamento de possíveis transições (e seus retornos de chamada prepare individuais). Os retornos de chamada de finalize_event serão executados independentemente do sucesso das transições processadas. Observe que se ocorrer um erro, ele será anexado a event_data como error e poderá ser recuperado com 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 Às vezes as coisas simplesmente não funcionam como planejado e precisamos lidar com exceções e limpar a bagunça para manter as coisas funcionando. Podemos passar callbacks para on_exception para fazer isso:
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 Como você provavelmente já percebeu, a forma padrão de passar chamadas para estados, condições e transições é pelo nome. Ao processar retornos de chamada e condições, transitions usarão seus nomes para recuperar o callable relacionado do modelo. Se o método não puder ser recuperado e contiver pontos, transitions tratarão o nome como um caminho para uma função de módulo e tentarão importá-lo. Alternativamente, você pode passar nomes de propriedades ou atributos. Eles serão agrupados em funções, mas não poderão receber dados de eventos por motivos óbvios. Você também pode passar callables, como funções (vinculadas), diretamente. Conforme mencionado anteriormente, você também pode passar listas/tuplas de nomes de chamadas para os parâmetros de retorno de chamada. Os retornos de chamada serão executados na ordem em que foram adicionados.
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 () A resolução que pode ser chamada é feita em Machine.resolve_callable . Este método pode ser substituído caso sejam necessárias estratégias de resolução exigíveis mais complexas.
Exemplo
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 ) Em resumo, existem atualmente três maneiras de acionar eventos. Você pode chamar as funções de conveniência de um modelo, como lump.melt() , executar gatilhos por nome, como lump.trigger("melt") ou despachar eventos em vários modelos com machine.dispatch("melt") (consulte a seção sobre vários modelos em padrões de inicialização alternativos). Os retornos de chamada nas transições são executados na seguinte ordem:
| Ligar de volta | Estado atual | Comentários |
|---|---|---|
'machine.prepare_event' | source | executado uma vez antes de transições individuais serem processadas |
'transition.prepare' | source | executado assim que a transição começa |
'transition.conditions' | source | condições podem falhar e interromper a transição |
'transition.unless' | source | condições podem falhar e interromper a transição |
'machine.before_state_change' | source | retornos de chamada padrão declarados no modelo |
'transition.before' | source | |
'state.on_exit' | source | retornos de chamada declarados no estado de origem |
<STATE CHANGE> | ||
'state.on_enter' | destination | retornos de chamada declarados no estado de destino |
'transition.after' | destination | |
'machine.on_final' | destination | retornos de chamada em filhos serão chamados primeiro |
'machine.after_state_change' | destination | retornos de chamada padrão declarados no modelo; também será chamado após transições internas |
'machine.on_exception' | source/destination | retornos de chamada serão executados quando uma exceção for levantada |
'machine.finalize_event' | source/destination | retornos de chamada serão executados mesmo que nenhuma transição tenha ocorrido ou uma exceção tenha sido gerada |
Se algum retorno de chamada gerar uma exceção, o processamento dos retornos de chamada não continuará. Isso significa que quando ocorre um erro antes da transição (em state.on_exit ou anterior), ela é interrompida. Caso haja um aumento após a transição ter sido realizada (em state.on_enter ou posterior), a mudança de estado persiste e nenhuma reversão ocorre. Os retornos de chamada especificados em machine.finalize_event sempre serão executados, a menos que a exceção seja gerada pelo próprio retorno de chamada de finalização. Observe que cada sequência de retorno de chamada deve ser concluída antes da execução do próximo estágio. O bloqueio de retornos de chamada interromperá a ordem de execução e, portanto, bloqueará o trigger ou a própria chamada dispatch . Se você deseja que os retornos de chamada sejam executados em paralelo, você pode dar uma olhada nas extensões AsyncMachine para processamento assíncrono ou LockedMachine para threading.
Às vezes você precisa passar para as funções de retorno de chamada registradas na inicialização da máquina alguns dados que refletem o estado atual do modelo. Transitions permite que você faça isso de duas maneiras diferentes.
Primeiro (o padrão), você pode passar qualquer argumento posicional ou de palavra-chave diretamente para os métodos de gatilho (criados quando você chama 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.'Você pode passar quantos argumentos quiser para o gatilho.
Há uma limitação importante nesta abordagem: toda função de retorno de chamada acionada pela transição de estado deve ser capaz de lidar com todos os argumentos. Isso pode causar problemas se cada retorno de chamada esperar dados um pouco diferentes.
Para contornar isso, o Transitions oferece suporte a um método alternativo de envio de dados. Se você definir send_event=True na inicialização Machine , todos os argumentos para os gatilhos serão agrupados em uma instância EventData e passados para cada retorno de chamada. (O objeto EventData também mantém referências internas ao estado de origem, modelo, transição, máquina e gatilho associado ao evento, caso você precise acessá-los para qualquer coisa.)
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.' Em todos os exemplos até agora, anexamos uma nova instância Machine a um modelo separado ( lump , uma instância da classe Matter ). Embora essa separação mantenha as coisas organizadas (porque você não precisa corrigir vários métodos novos na classe Matter ), ela também pode ser irritante, pois exige que você acompanhe quais métodos são chamados na máquina de estado e quais são chamados no modelo ao qual a máquina de estado está vinculada (por exemplo, lump.on_enter_StateA() vs. machine.add_transition() ).
Felizmente, o Transitions é flexível e oferece suporte a dois outros padrões de inicialização.
Primeiro, você pode criar uma máquina de estado independente que não exija outro modelo. Simplesmente omita o argumento do modelo durante a inicialização:
machine = Machine ( states = states , transitions = transitions , initial = 'solid' )
machine . melt ()
machine . state
> >> 'liquid' Se você inicializar a máquina dessa maneira, poderá anexar todos os eventos de acionamento (como evaporate() , sublimate() , etc.) e todas as funções de retorno de chamada diretamente à instância Machine .
Essa abordagem tem a vantagem de consolidar todas as funcionalidades da máquina de estado em um só lugar, mas pode parecer um pouco antinatural se você achar que a lógica de estado deve estar contida no próprio modelo, e não em um controlador separado.
Uma abordagem alternativa (potencialmente melhor) é fazer com que o modelo seja herdado da classe Machine . O Transitions foi projetado para suportar herança perfeitamente. (apenas certifique-se de substituir o método __init__ da classe 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' Aqui você consolida todas as funcionalidades da máquina de estado em seu modelo existente, o que muitas vezes parece mais natural do que colocar todas as funcionalidades que desejamos em uma instância Machine independente separada.
Uma máquina pode lidar com vários modelos que podem ser passados como uma lista como Machine(model=[model1, model2, ...]) . Nos casos em que você deseja adicionar modelos , bem como a própria instância da máquina, você pode passar o espaço reservado para variável de classe (string) Machine.self_literal durante a inicialização como Machine(model=[Machine.self_literal, model1, ...]) . Você também pode criar uma máquina autônoma e registrar modelos dinamicamente por meio de machine.add_model passando model=None para o construtor. Além disso, você pode usar machine.dispatch para acionar eventos em todos os modelos adicionados atualmente. Lembre-se de chamar machine.remove_model se a máquina for duradoura e seus modelos forem temporários e devem ser coletados como lixo:
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 Se você não fornecer um estado inicial no construtor da máquina de estado, transitions criarão e adicionarão um estado padrão chamado 'initial' . Se você não quiser um estado inicial padrão, poderá passar initial=None . Porém, neste caso você precisa passar um estado inicial toda vez que adicionar um modelo.
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' ) Modelos com vários estados podem anexar várias máquinas usando valores model_attribute diferentes. Conforme mencionado em Verificando estado, isso adicionará funções is/to_<model_attribute>_<state_name> personalizadas:
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 Transitions inclui recursos de registro muito rudimentares. Vários eventos – ou seja, mudanças de estado, gatilhos de transição e verificações condicionais – são registrados como eventos de nível INFO usando o módulo logging padrão do Python. Isso significa que você pode configurar facilmente o log para saída padrão em um 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' )
... As máquinas podem ser decapadas e podem ser armazenadas e carregadas com pickle . Para Python 3.3 e anteriores, é necessário 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' ] Como você provavelmente notou, transitions usam alguns recursos dinâmicos do Python para fornecer maneiras práticas de lidar com modelos. No entanto, os verificadores de tipo estático não gostam que atributos e métodos do modelo não sejam conhecidos antes do tempo de execução. Historicamente, transitions também não atribuíam métodos de conveniência já definidos nos modelos para evitar substituições acidentais.
Mas não se preocupe! Você pode usar o parâmetro do construtor de máquina model_override para alterar a forma como os modelos são decorados. Se você definir model_override=True , transitions substituirão apenas os métodos já definidos. Isso evita que novos métodos apareçam em tempo de execução e também permite definir quais métodos auxiliares você deseja usar.
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" Se você quiser usar todas as funções de conveniência e adicionar alguns retornos de chamada, definir um modelo pode ficar bem complicado quando você tem muitos estados e transições definidos. O método generate_base_model nas transitions pode gerar um modelo base a partir de uma configuração de máquina para ajudá-lo com isso.
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 ) Definir métodos de modelo que serão substituídos adiciona um pouco de trabalho extra. Pode ser complicado alternar para garantir que os nomes dos eventos estejam escritos corretamente, especialmente se os estados e as transições forem definidos em listas antes ou depois do seu modelo. Você pode reduzir o clichê e a incerteza de trabalhar com strings definindo estados como enums. Você também pode definir transições diretamente na sua classe de modelo com a ajuda de add_transitions e event . Depende de você usar o decorador de função add_transitions ou event para atribuir valores aos atributos, dependendo do seu estilo de código preferido. Ambos funcionam da mesma maneira, têm a mesma assinatura e devem resultar em (quase) as mesmas dicas de tipo IDE. Como este ainda é um trabalho em andamento, você precisará criar uma classe Machine personalizada e usar with_model_definitions para transições para verificar as transições definidas dessa forma.
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 . AMesmo que o núcleo das transições seja leve, há uma variedade de MixIns para estender sua funcionalidade. Atualmente suportados são:
Existem dois mecanismos para recuperar uma instância de máquina de estado com os recursos desejados habilitados. A primeira abordagem faz uso da factory de conveniência com os quatro parâmetros graph , nested , locked ou asyncio definidos como True se o recurso for necessário:
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 ) Esta abordagem visa o uso experimental, pois neste caso as classes subjacentes não precisam ser conhecidas. No entanto, as classes também podem ser importadas diretamente de transitions.extensions . O esquema de nomenclatura é o seguinte:
| Diagramas | Aninhado | Bloqueado | Assíncio | |
|---|---|---|---|---|
| Máquina | ✘ | ✘ | ✘ | ✘ |
| Máquina gráfica | ✓ | ✘ | ✘ | ✘ |
| Máquina Hierárquica | ✘ | ✓ | ✘ | ✘ |
| Máquina bloqueada | ✘ | ✘ | ✓ | ✘ |
| Máquina gráfica hierárquica | ✓ | ✓ | ✘ | ✘ |
| Máquina LockedGraph | ✓ | ✘ | ✓ | ✘ |
| Máquina Hierárquica Bloqueada | ✘ | ✓ | ✓ | ✘ |
| LockedHierarchicalGraphMachine | ✓ | ✓ | ✓ | ✘ |
| Máquina assíncrona | ✘ | ✘ | ✘ | ✓ |
| Máquina AsyncGraph | ✓ | ✘ | ✘ | ✓ |
| HierárquicaAsyncMachine | ✘ | ✓ | ✘ | ✓ |
| Máquina HierárquicaAsyncGraph | ✓ | ✓ | ✘ | ✓ |
Para usar uma máquina de estado rica em recursos, pode-se escrever:
from transitions . extensions import LockedHierarchicalGraphMachine as LHGMachine
machine = LHGMachine ( model , states , transitions ) Transitions inclui um módulo de extensão que permite aninhar estados. Isto nos permite criar contextos e modelar casos onde os estados estão relacionados a determinadas subtarefas na máquina de estados. Para criar um estado aninhado, importe NestedState das transições ou use um dicionário com os argumentos de inicialização name e children . Opcionalmente, initial pode ser usado para definir um subestado para o qual transitar, quando o estado aninhado for inserido.
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') Uma configuração usando initial poderia ser assim:
# ...
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' ]
]
# ... A palavra-chave initial do construtor HierarchicalMachine aceita estados aninhados (por exemplo, initial='caffeinated_running' ) e uma lista de estados que é considerado um estado paralelo (por exemplo, initial=['A', 'B'] ) ou o estado atual de outro modelo ( initial=model.state ) que deve ser efetivamente uma das opções mencionadas anteriormente. Observe que ao passar uma string, transition verificará o estado de destino para os subestados initial e usará isso como um estado de entrada. Isto será feito recursivamente até que um subestado não mencione um estado inicial. Os estados paralelos ou um estado aprovado como uma lista serão usados “como estão” e nenhuma avaliação inicial adicional será realizada.
Observe que o objeto de estado criado anteriormente deve ser um NestedState ou uma classe derivada dele. A classe State padrão usada em instâncias simples Machine não possui os recursos necessários para aninhamento.
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! Algumas coisas que devem ser consideradas ao trabalhar com estados aninhados: Os nomes dos estados são concatenados com NestedState.separator . Atualmente o separador está definido como sublinhado ('_') e, portanto, se comporta de forma semelhante à máquina básica. Isso significa que uma bar de subestado do estado foo será conhecida por foo_bar . Um subestado baz de bar será referido como foo_bar_baz e assim por diante. Ao inserir um subestado, enter será chamado para todos os estados pais. O mesmo se aplica à saída de subestados. Terceiro, os estados aninhados podem substituir o comportamento de transição de seus pais. Se uma transição não for conhecida para o estado atual, ela será delegada ao seu pai.
Isso significa que na configuração padrão, os nomes de estado nos HSMs NÃO DEVEM conter sublinhados. Para transitions é impossível dizer se machine.add_state('state_name') deve adicionar um estado chamado state_name ou adicionar um name de subestado ao estado state . Em alguns casos, isto não é suficiente. Por exemplo, se os nomes dos estados consistirem em mais de uma palavra e você quiser/precisar usar sublinhado para separá-los em vez de CamelCase . Para lidar com isso, você pode alterar facilmente o caractere usado para separação. Você pode até usar caracteres Unicode sofisticados se usar Python 3. Definir o separador para algo diferente de sublinhado altera parte do comportamento (auto_transition e configuração de retornos de chamada):
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') Em vez de to_C_3_a() a transição automática é chamada como to_C.s3.a() . Se o seu subestado começar com um dígito, as transições adicionam um prefixo 's' ('3' torna-se 's3') ao FunctionWrapper de transição automática para cumprir o esquema de nomenclatura de atributos do Python. Se a conclusão interativa não for necessária, to('C↦3↦a') pode ser chamado diretamente. Além disso, on_enter/exit_<<state name>> é substituído por on_enter/exit(state_name, callback) . As verificações do Estado podem ser realizadas de maneira semelhante. Em vez de is_C_3_a() , a variante FunctionWrapper is_C.s3.a() pode ser usada.
Para verificar se o estado atual é um subestado de um estado específico, is_state suporta a palavra-chave 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 Você também pode usar enumerações em HSMs, mas lembre-se de que Enum são comparados por valor. Se você tiver um valor mais de uma vez em uma árvore de estados, esses estados não poderão ser distinguidos.
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 foi reescrito do zero para oferecer suporte a estados paralelos e melhor isolamento de estados aninhados. Isso envolve alguns ajustes com base no feedback da comunidade. Para ter uma ideia da ordem de processamento e configuração, dê uma olhada no exemplo a seguir:
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 Ao usar parallel em vez de children , transitions entrarão em todos os estados da lista passada ao mesmo tempo. Qual subestado entrar é definido por initial que deve sempre apontar para um subestado direto. Um novo recurso é definir transições locais passando a palavra-chave transitions em uma definição de estado. A transição definida acima ['go', 'a', 'b'] só é válida em C_1 . Embora você possa fazer referência a subestados como feito em ['go', '2_z', '2_x'] você não pode fazer referência a estados pais diretamente em transições definidas localmente. Quando um estado pai é encerrado, seus filhos também serão encerrados. Além da ordem de processamento das transições conhecida em Machine , onde as transições são consideradas na ordem em que foram adicionadas, HierarchicalMachine também considera a hierarquia. As transições definidas em subestados serão avaliadas primeiro (por exemplo, C_1_a é deixado antes de C_2_z ) e as transições definidas com curinga * Will (por enquanto) adicionam apenas transições aos estados raiz (neste exemplo A , B , C ) começando com 0,8.0 estados aninhados pode ser adicionado diretamente e emitirá a criação de estados-pais on-the-fly:
m = HierarchicalMachine ( states = [ 'A' ], initial = 'A' )
m . add_state ( 'B_1_a' )
m . to_B_1 ()
assert m . is_B ( allow_substates = True ) Experimental em 0.9.1: Você pode usar retornos de chamada on_final nos estados ou no próprio HSM. Os retornos de chamada serão acionados se a) o próprio estado for marcado com final e acaba de ser inserido ou b) todos os subestados forem considerados finais e pelo menos um substituto acabou de entrar em um estado final. No caso de B) todos os pais serão considerados finais se a condição b) se fortalecer para eles. Isso pode ser útil nos casos em que o processamento ocorre em paralelo e seu HSM ou qualquer estado pai deve ser notificado quando todos os subestados atingirem um estado final:
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! Além da ordem semântica, os estados aninhados são muito úteis se você deseja especificar máquinas de estado para tarefas específicas e planejar reutilizá -las. Antes de 0,8.0 , uma HierarchicalMachine não integraria a própria instância da máquina, mas os estados e as transições criando cópias deles. No entanto, como as instâncias estatais de 0,8.0 (Nested)State são apenas referenciadas , o que significa que as alterações na coleção de estados e eventos de uma máquina influenciarão a outra instância da máquina. Modelos e seu estado não serão compartilhados. Observe que eventos e transições também são copiados por referência e serão compartilhados por ambas as instâncias se você não usar a palavra -chave remap . Essa alteração foi feita para estar mais alinhada com Machine , que também usa instâncias State passadas por referência.
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 Se uma HierarchicalMachine for aprovada com a palavra -chave children , o estado inicial desta máquina será atribuído ao novo estado pai. No exemplo acima, vemos que a entrada counting também entrará em counting_1 . Se esse comportamento indesejado e a máquina deve interromper no estado pai, o usuário pode passar initial como False como {'name': 'counting', 'children': counter, 'initial': False} .
Às vezes, você deseja uma coleção de estado tão incorporada para 'devolver', o que significa que, depois de ser feito, deve sair e transitar para um de seus super estados. Para alcançar esse comportamento, você pode remapar as transições do estado. No exemplo acima, gostaríamos que o balcão retornasse se o done fosse alcançado. Isso é feito da seguinte maneira:
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 Como mencionado acima, o uso remap copiará eventos e transições, pois eles não poderiam ser válidos na máquina estadual original. Se uma máquina de estado reutilizada não tiver um estado final, é claro que você pode adicionar as transições manualmente. Se 'contador' não tivesse 'estado', poderíamos simplesmente adicionar ['done', 'counter_3', 'waiting'] para alcançar o mesmo comportamento.
Nos casos em que você deseja que estados e transições sejam copiados por valor, em vez de referência (por exemplo, se você deseja manter o comportamento pré-0.8), pode fazê-lo criando um NestedState e atribuindo cópias profundas dos eventos e estados da máquina a isto.
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 ] Para máquinas de estado complexas, o compartilhamento de configurações em vez de máquinas instanciadas pode ser mais viável. Especialmente porque as máquinas instanciadas devem ser derivadas da HierarchicalMachine . Tais configurações podem ser armazenadas e carregadas facilmente via JSON ou YAML (consulte as perguntas frequentes). HierarchicalMachine permite definir subestados com as children -chave ou states . Se ambos estiverem presentes, apenas children serão consideradas.
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 ()Palavras -chave adicionais:
title (Opcional): define o título da imagem gerada.show_conditions (padrão false): mostra condições nas bordas de transiçãoshow_auto_transitions (padrão false): mostra transições automáticas no gráficoshow_state_attributes (Padrão false): Mostrar retornos de chamada (ENTER, EXIT), Tags e Timeouts no gráficoAs transições podem gerar diagramas básicos de estado exibindo todas as transições válidas entre os estados. O suporte ao diagrama básico gera uma definição de máquina de estado de sereia que pode ser usada com o editor ao vivo da sereia, em arquivos de marcação no GitLab ou Github e em outros serviços da Web. Por exemplo, este código:
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!" )Produz este diagrama (verifique a fonte do documento para ver a notação de marcação):
---
Gráfico de sereia
---
StatediaGram-V2
direção lr
ClassDef S_Default Preenchimento: Branco, Cor: Preto
ClassDef S_Inactive Preenchimento: Branco, Cor: Preto
ClassDef S_parallelel Cor: preto, preenchimento: branco
ClassDef S_Active Cor: Red, preenchimento: Darksalmon
Classdef S_previous Color: Blue, preenchimento: Azure
Estado "A" como um
Classe A S_previous
Estado "B" como B
Classe B S_Active
Estado "C" como C
C -> [*]
Classe C S_Default
Estado C {
Estado "1" como C_1
Estado C_1 {
[*] -> c_1_a
Estado "A" como C_1_A
Estado "B" como C_1_B
C_1_B -> [*]
}
--
estado "2" como c_2
Estado C_2 {
[*] -> c_2_a
Estado "A" como C_2_A
Estado "B" como C_2_B
C_2_B -> [*]
}
}
C -> A: Redefinir
A -> B: init
B -> C: Faça
C_1_a -> c_1_b: vá
C_2_a -> c_2_b: vá
[*] -> a
Para usar a funcionalidade gráfica mais sofisticada, você precisará instalar graphviz e/ou pygraphviz . Para gerar gráficos com o pacote graphviz , você precisa instalar o GraphViz manualmente ou através de um gerenciador de pacotes.
sudo apt-get install graphviz graphviz-dev # Ubuntu and Debian
brew install graphviz # MacOS
conda install graphviz python-graphviz # (Ana)conda
Agora você pode instalar os pacotes python reais
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
Atualmente, GraphMachine usará pygraphviz quando disponível e voltará ao graphviz quando pygraphviz não puder ser encontrado. Se graphviz também não estiver disponível, mermaid será usada. Isso pode ser substituído pela passagem para graph_engine="graphviz" (ou "mermaid" ) para o construtor. Observe que esse padrão pode mudar no futuro e o suporte pygraphviz pode ser descartado. Com Model.get_graph() você pode obter o gráfico atual ou a região de interesse (ROI) e desenhá -lo assim:
# 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' )Isso produz algo assim:
Independente do back -end que você usa, a função Draw também aceita um descritor de arquivo ou um fluxo binário como o primeiro argumento. Se você definir este parâmetro como None , o fluxo de bytes será devolvido:
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 ()Referências e parciais passados como retornos de chamada serão resolvidos o mais bom possível:
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' )Isso deve produzir algo semelhante a isso:
Se o formato das referências não atender às suas necessidades, você poderá substituir o método estático GraphMachine.format_references . Se você deseja pular a referência completamente, basta deixar GraphMachine.format_references retornar None . Além disso, dê uma olhada no nosso exemplo de notebooks ipython/jupyter para obter um exemplo mais detalhado sobre como usar e editar gráficos.
Nos casos em que a despacho de eventos é feita em threads, pode -se usar Machine LockedMachine ou LockedHierarchicalMachine , onde o acesso à função (! SIC) é protegido com bloqueios reentrantes. Isso não o salva de corromper sua máquina, mexendo com variáveis de membros do seu modelo ou máquina de estado.
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 Qualquer gerente de contexto Python pode ser passado através do argumento da palavra -chave 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 ]) Quaisquer contextos via machine_model serão compartilhados entre todos os modelos registrados na Machine . Os contextos por modelo também podem ser adicionados:
lock3 = RLock ()
machine . add_model ( model , model_context = lock3 )É importante que todos os gerentes de contexto fornecidos pelo usuário sejam reentrantes, pois a máquina de estado os chama várias vezes, mesmo no contexto de uma única invocação de gatilho.
Se você estiver usando o Python 3.7 ou posterior, poderá usar AsyncMachine para trabalhar com retornos de chamada assíncronos. Você pode misturar retornos de chamada síncronos e assíncronos, se quiser, mas isso pode ter efeitos colaterais indesejados. Observe que os eventos precisam ser aguardados e o loop do evento também deve ser tratado por você.
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 () Então, por que você precisa usar o Python 3.7 ou mais tarde, pode perguntar. O suporte assíncrono foi introduzido anteriormente. AsyncMachine faz uso de contextvars para lidar com retornos de chamada em execução quando novos eventos chegarem antes que uma transição seja concluída:
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 Este exemplo realmente ilustra duas coisas: primeiro, que 'Go' chamou na transição de M1 de A para B para B não é cancelado e o segundo, chamando m2.fix() interromperá a tentativa de transição de M2 de A para B executando 'Fix' de A a C Essa separação não seria possível sem contextvars . Observe que prepare e conditions não são tratadas como transições contínuas. Isso significa que, após conditions terem sido avaliadas, uma transição é executada, embora outro evento já tenha acontecido. As tarefas só serão canceladas quando executadas como um retorno de chamada before ou posterior.
AsyncMachine apresenta um modo de fila-especial modelo que pode ser usado quando queued='model' é passada para o construtor. Com uma fila específica do modelo, os eventos só serão filmados quando pertencem ao mesmo modelo. Além disso, uma exceção elevada apenas limpará a fila de eventos do modelo que levantou essa exceção. Por uma questão de simplicidade, vamos supor que todos os eventos em asyncio.gather abaixo não sejam acionados ao mesmo tempo, mas um pouco atrasado:
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.event3Observe que os modos de fila não devem ser alterados após a construção da máquina.
Se seus super -heróis precisarem de algum comportamento personalizado, você poderá lançar alguma funcionalidade extra decorando os estados da máquina:
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 aloneAtualmente, as transições são equipadas com os seguintes recursos do estado:
Tempo limite - desencadeia um evento depois de algum tempo passou
timeout (int, opcional) - Se aprovado, um estado inserido terá tempo limite após timeout segundoson_timeout (String/Callable, Opcional) - será chamado quando o tempo de tempo limite for alcançadoAttributeError quando timeout estiver definido, mas on_timeout não éTags - adiciona tags aos estados
tags (lista, opcional) - atribui tags a um estadoState.is_<tag_name> retornará True quando o estado for marcado com tag_name , else False Erro - levanta um MachineError quando um estado não pode ser deixado
Tags (se você usar Error não use Tags )accepted (bool, opcional) - marca um estado como aceitotags de palavra -chave podem ser passadas, contendo 'aceito'auto_transitions tiver sido definido como False . Caso contrário, todo estado pode ser exitado com os métodos to_<state> .Volátil - inicializa um objeto toda vez que um estado é inserido
volatile (classe, opcional) - Toda vez que o estado é inserido, um objeto de classe de tipo será atribuído ao modelo. O nome do atributo é definido pelo hook . Se omitido, um volatileObject vazio será criado em vez dissohook (string, default = 'scope') - o nome do atributo do modelo para o objeto temporal. Você pode escrever suas próprias extensões State e adicioná -las da mesma maneira. Basta observar que add_state_features espera mixins . Isso significa que sua extensão deve sempre chamar os métodos substituídos __init__ , enter e exit . Sua extensão pode herdar do estado , mas também funcionará sem ela. O uso de @add_state_features possui uma desvantagem que é que as máquinas decoradas não podem ser em conserva (com mais precisão, o CustomState gerado dinamicamente não pode ser em conserva). Isso pode ser um motivo para escrever uma classe de estado personalizada dedicada. Dependendo da máquina estadual escolhida, sua classe de estado personalizada pode precisar fornecer determinados recursos do estado. Por exemplo, HierarchicalMachine exige que seu estado personalizado seja uma instância do Estado de NestedState ( State não é suficiente). Para injetar seus estados, você pode atribuí -los ao atributo de classe da sua Machine state_cls ou Substitua Machine.create_state , caso precise de alguns procedimentos específicos feitos sempre que um estado for criado:
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 ) Se você deseja evitar threads em sua AsyncMachine completamente, poderá substituir o recurso de estado Timeout pelo AsyncTimeout da extensão 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 Você deve considerar passar queued=True ao construtor TimeoutMachine . Isso garantirá que os eventos sejam processados sequencialmente e evitem condições de corrida assíncronas que possam aparecer quando o tempo limite e o evento ocorrem nas proximidades.
Você pode dar uma olhada no FAQ para obter alguma inspiração ou checkout django-transitions . Foi desenvolvido por Christian Ledermann e também está hospedado no Github. A documentação contém alguns exemplos de uso.
Primeiro, parabéns! Você chegou ao final da documentação! Se você quiser experimentar transitions antes de instalá -lo, poderá fazer isso em um notebook interativo Jupyter em mybinder.org. Basta clicar neste botão.
Para relatórios de bugs e outros problemas, abra um problema no Github.
Para perguntas de uso, poste no Stack Overflow, certifique -se de marcar sua pergunta com a tag pytransitions . Não se esqueça de dar uma olhada nos exemplos estendidos!
Para quaisquer outras perguntas, solicitações ou grandes presentes monetários irrestritos, envie um e -mail para Tal Yarkoni (autor inicial) e/ou Alexander Neumann (mantenedor atual).
