transitions
Release 0.9.2
Una implementación de máquina de estados ligera y orientada a objetos en Python con muchas extensiones. Compatible con Python 2.7+ y 3.0+.
pip install transitions
... o clonar el repositorio de GitHub y luego:
python setup.py install
Dicen que un buen ejemplo vale más que 100 páginas de documentación API, un millón de directivas o mil palabras.
Bueno, "ellos" probablemente mientan... pero aquí hay un ejemplo de todos modos:
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?" ) Ahí, ahora ha integrado una máquina de estados en NarcolepticSuperhero . Saquémoslo a dar una vuelta...
> >> 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 Si bien no podemos leer la mente del Batman real, seguramente podemos visualizar el estado actual de nuestro NarcolepticSuperhero .
Echa un vistazo a las extensiones de Diagramas si quieres saber cómo hacerlo.
Una máquina de estados es un modelo de comportamiento compuesto por un número finito de estados y transiciones entre esos estados. Dentro de cada estado y transición se puede realizar alguna acción . Una máquina de estados necesita comenzar en algún estado inicial . Cuando se utilizan transitions , una máquina de estados puede constar de múltiples objetos donde algunos ( máquinas ) contienen definiciones para la manipulación de otros ( modelos ). A continuación, veremos algunos conceptos básicos y cómo trabajar con ellos.
Estado . Un estado representa una condición o etapa particular en la máquina de estados. Es un modo distinto de comportamiento o fase de un proceso.
Transición . Este es el proceso o evento que hace que la máquina de estados cambie de un estado a otro.
Modelo . La estructura con estado real. Es la entidad que se actualiza durante las transiciones. También puede definir acciones que se ejecutarán durante las transiciones. Por ejemplo, justo antes de una transición o cuando se entra o se sale de un estado.
Máquina . Esta es la entidad que gestiona y controla el modelo, estados, transiciones y acciones. Es el conductor el que orquesta todo el proceso de la máquina estatal.
Desencadenar . Este es el evento que inicia una transición, el método que envía la señal para iniciar una transición.
Acción . Operación o tarea específica que se realiza cuando se entra, se sale de un determinado estado o durante una transición. La acción se implementa mediante devoluciones de llamada , que son funciones que se ejecutan cuando ocurre algún evento.
Poner en funcionamiento una máquina de estados es bastante sencillo. Digamos que tienes el objeto lump (una instancia de la clase Matter ) y quieres administrar sus estados:
class Matter ( object ):
pass
lump = Matter () Puede inicializar una máquina en estado de funcionamiento ( mínima ) vinculada al lump del modelo de esta manera:
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' Una alternativa es no pasar explícitamente un modelo al inicializador Machine :
machine = Machine ( states = [ 'solid' , 'liquid' , 'gas' , 'plasma' ], initial = 'solid' )
# The machine instance itself now acts as a model
machine . state
> >> 'solid' Tenga en cuenta que esta vez no pasé el modelo lump como argumento. El primer argumento pasado a Machine actúa como modelo. Entonces, cuando pase algo allí, todas las funciones de conveniencia se agregarán al objeto. Si no se proporciona ningún modelo, la propia instancia machine actúa como modelo.
Cuando al principio dije "mínimo", fue porque si bien esta máquina de estados está técnicamente operativa, en realidad no hace nada. Comienza en el estado 'solid' , pero nunca pasará a otro estado, porque no se han definido transiciones... ¡todavía!
Intentemos de nuevo.
# 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 los nuevos y brillantes métodos adjuntos a la instancia Matter ( evaporate() , ionize() , etc.). Cada método desencadena la transición correspondiente. Las transiciones también se pueden activar dinámicamente llamando al método trigger() proporcionado con el nombre de la transición, como se muestra arriba. Más sobre esto en la sección Activar una transición.
El alma de cualquier máquina de estados buena (y de muchas malas, sin duda) es un conjunto de estados. Arriba, definimos los estados válidos del modelo pasando una lista de cadenas al inicializador Machine . Pero internamente, los Estados en realidad están representados como objetos State .
Puede inicializar y modificar estados de varias maneras. Específicamente, puedes:
Machine dando los nombres de los estados, oState , oLos siguientes fragmentos ilustran varias formas de lograr el mismo 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 ])Los estados se inicializan una vez cuando se agregan a la máquina y persistirán hasta que se eliminen de ella. En otras palabras: si modifica los atributos de un objeto de estado, este cambio NO se restablecerá la próxima vez que ingrese a ese estado. Eche un vistazo a cómo ampliar las funciones de estado en caso de que necesite algún otro comportamiento.
Pero simplemente tener estados y poder moverse entre ellos (transiciones) no es muy útil por sí solo. ¿Qué pasa si quieres hacer algo, realizar alguna acción cuando entras o sales de un estado? Aquí es donde entran las devoluciones de llamada .
Un State también se puede asociar con una lista de devoluciones de llamadas enter y exit , que se llaman cada vez que la máquina de estado entra o sale de ese estado. Puede especificar devoluciones de llamada durante la inicialización pasándolas a un constructor de objetos State , en un diccionario de propiedades de estado, o agregándolas más tarde.
Para mayor comodidad, cada vez que se agrega un nuevo State a una Machine , los métodos on_enter_«state name» y on_exit_«state name» se crean dinámicamente en la Máquina (¡no en el modelo!), lo que le permite agregar dinámicamente nuevas entradas y salidas. devoluciones de llamada más tarde si las necesita.
# 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!' Tenga en cuenta que la devolución de llamada on_enter_«state name» no se activará cuando se inicialice una máquina por primera vez. Por ejemplo, si tiene definida una devolución de llamada on_enter_A() e inicializa la Machine con initial='A' , on_enter_A() no se activará hasta la próxima vez que ingrese al estado A . (Si necesita asegurarse de que on_enter_A() se active durante la inicialización, simplemente puede crear un estado inicial ficticio y luego llamar explícitamente to_A() dentro del método __init__ ).
Además de pasar devoluciones de llamada al inicializar un State o agregarlas dinámicamente, también es posible definir devoluciones de llamada en la propia clase de modelo, lo que puede aumentar la claridad del código. Por ejemplo:
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' ]) Ahora, cada vez que lump pase al estado A , se activará el método on_enter_A() definido en la clase Matter .
Puede utilizar devoluciones de llamada on_final que se activarán cuando se ingrese un estado con 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 ... Siempre puedes comprobar el estado actual del modelo mediante:
.state , ois_«state name»() Y si desea recuperar el objeto State real para el estado actual, puede hacerlo a través del método get_state() de la instancia Machine .
lump . state
> >> 'solid'
lump . is_gas ()
> >> False
lump . is_solid ()
> >> True
machine . get_state ( lump . state ). name
> >> 'solid' Si lo desea, puede elegir su propio nombre de atributo de estado pasando el argumento model_attribute mientras inicializa Machine . Sin embargo, esto también cambiará el nombre de is_«state name»() a is_«model_attribute»_«state name»() . De manera similar, las transiciones automáticas se denominarán to_«model_attribute»_«state name»() en lugar de to_«state name»() . Esto se hace para permitir que varias máquinas trabajen en el mismo modelo con nombres de atributos de estado individuales.
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 Hasta ahora hemos visto cómo podemos dar nombres de estados y utilizarlos para trabajar con nuestra máquina de estados. Si prefiere una escritura más estricta y una mayor finalización del código IDE (o simplemente ya no puede escribir 'sesquipedalofobia' porque la palabra le asusta), usar Enumeraciones podría ser lo que está buscando:
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 Puede mezclar enumeraciones y cadenas si lo desea (por ejemplo [States.RED, 'ORANGE', States.YELLOW, States.GREEN] ), pero tenga en cuenta que internamente, transitions seguirán manejando estados por nombre ( enum.Enum.name ). Por lo tanto, no es posible tener los estados 'GREEN' y States.GREEN al mismo tiempo.
Algunos de los ejemplos anteriores ya ilustran el uso de transiciones de pasada, pero aquí las exploraremos con más detalle.
Al igual que con los estados, cada transición se representa internamente como su propio objeto: una instancia de la clase Transition . La forma más rápida de inicializar un conjunto de transiciones es pasar un diccionario o una lista de diccionarios al inicializador de la Machine . Ya vimos esto arriba:
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 transiciones en los diccionarios tiene la ventaja de ser claro, pero puede resultar engorroso. Si busca brevedad, puede optar por definir transiciones usando listas. Sólo asegúrese de que los elementos de cada lista estén en el mismo orden que los argumentos posicionales en la inicialización Transition (es decir, trigger , source , destination , etc.).
La siguiente lista de listas es funcionalmente equivalente a la lista de diccionarios anterior:
transitions = [
[ 'melt' , 'solid' , 'liquid' ],
[ 'evaporate' , 'liquid' , 'gas' ],
[ 'sublimate' , 'solid' , 'gas' ],
[ 'ionize' , 'gas' , 'plasma' ]
] Alternativamente, puede agregar transiciones a una Machine después de la inicialización:
machine = Machine ( model = lump , states = states , initial = 'solid' )
machine . add_transition ( 'melt' , source = 'solid' , dest = 'liquid' )Para que se ejecute una transición, es necesario que algún evento la desencadene . Hay dos maneras de hacer esto:
Usando el método adjunto automáticamente en el modelo base:
> >> lump . melt ()
> >> lump . state
'liquid'
> >> lump . evaporate ()
> >> lump . state
'gas' Tenga en cuenta que no es necesario definir explícitamente estos métodos en ningún lugar; el nombre de cada transición está vinculado al modelo pasado al inicializador Machine (en este caso, lump ). Esto también significa que su modelo no debería contener métodos con el mismo nombre que los activadores de eventos, ya que transitions solo adjuntarán métodos convenientes a su modelo si el lugar aún no está ocupado. Si desea modificar ese comportamiento, consulte las preguntas frecuentes.
Usando el método trigger , ahora adjunto a su modelo (si no ha estado allí antes). Este método le permite ejecutar transiciones por nombre en caso de que se requiera activación dinámica:
> >> lump . trigger ( 'melt' )
> >> lump . state
'liquid'
> >> lump . trigger ( 'evaporate' )
> >> lump . state
'gas' De forma predeterminada, activar una transición no válida generará una excepción:
> >> 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!" Este comportamiento es generalmente deseable, ya que ayuda a alertarle sobre problemas en su código. Pero en algunos casos, es posible que desee ignorar silenciosamente los desencadenantes no válidos. Puede hacer esto configurando ignore_invalid_triggers=True (ya sea estado por estado o globalmente para todos los 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 ) Si necesita saber qué transiciones son válidas a partir de un determinado estado, puede utilizar 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' ] Si ha seguido esta documentación desde el principio, notará que get_triggers en realidad devuelve más activadores que los definidos explícitamente que se muestran arriba, como to_liquid , etc. Éstas se denominan auto-transitions y se presentarán en la siguiente sección.
Además de cualquier transición agregada explícitamente, se crea automáticamente un método to_«state»() cada vez que se agrega un estado a una instancia Machine . Este método pasa al estado objetivo sin importar en qué estado se encuentre actualmente la máquina:
lump . to_liquid ()
lump . state
> >> 'liquid'
lump . to_solid ()
lump . state
> >> 'solid' Si lo desea, puede desactivar este comportamiento configurando auto_transitions=False en el inicializador Machine .
Un desencadenador determinado se puede asociar a múltiples transiciones, algunas de las cuales potencialmente pueden comenzar o terminar en el mismo estado. Por ejemplo:
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' ) En este caso, llamar transmogrify() establecerá el estado del modelo en 'solid' si actualmente es 'plasma' y, en caso contrario, lo establecerá en 'plasma' . (Tenga en cuenta que solo se ejecutará la primera transición coincidente; por lo tanto, la transición definida en la última línea anterior no hará nada).
También puedes hacer que un disparador provoque una transición de todos los estados a un destino particular usando el comodín '*' :
machine . add_transition ( 'to_liquid' , '*' , 'liquid' )Tenga en cuenta que las transiciones con comodines solo se aplicarán a los estados que existen en el momento de la llamada add_transition(). Llamar a una transición basada en comodines cuando el modelo está en un estado agregado después de que se definió la transición generará un mensaje de transición no válido y no realizará la transición al estado objetivo.
Se puede agregar fácilmente un activador reflexivo (activador que tiene el mismo estado que origen y destino) especificando = como destino. Esto es útil si se debe agregar el mismo disparador reflexivo a varios estados. Por ejemplo:
machine . add_transition ( 'touch' , [ 'liquid' , 'gas' , 'plasma' ], '=' , after = 'change_shape' ) Esto agregará transiciones reflexivas para los tres estados con touch() como disparador y con change_shape ejecutado después de cada disparador.
A diferencia de las transiciones reflexivas, las transiciones internas nunca abandonarán realmente el Estado. Esto significa que las devoluciones de llamadas relacionadas con la transición, como before o after se procesarán, mientras que las devoluciones de llamadas relacionadas con el estado exit o enter no. Para definir una transición como interna, establezca el destino en None .
machine . add_transition ( 'internal' , [ 'liquid' , 'gas' ], None , after = 'change_shape' ) Un deseo común es que las transiciones de estados sigan una secuencia lineal estricta. Por ejemplo, dados los estados ['A', 'B', 'C'] , es posible que desee transiciones válidas para A → B , B → C y C → A (pero ningún otro par).
Para facilitar este comportamiento, Transitions proporciona un método add_ordered_transitions() en la clase 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!" El comportamiento predeterminado en Transiciones es procesar eventos instantáneamente. Esto significa que los eventos dentro de un método on_enter se procesarán antes de que se llamen las devoluciones de llamada vinculadas a 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?El orden de ejecución de este ejemplo es
prepare -> before -> on_enter_B -> on_enter_C -> after.
Si el procesamiento en cola está habilitado, se finalizará una transición antes de que se active la siguiente transición:
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!Esto resulta en
prepare -> before -> on_enter_B -> queue(to_C) -> after -> on_enter_C.
Nota importante: al procesar eventos en una cola, la llamada de activación siempre devolverá True , ya que no hay manera de determinar en el momento de la cola si una transición que involucra llamadas en cola finalmente se completará exitosamente. Esto es cierto incluso cuando solo se procesa un evento.
machine . add_transition ( 'jump' , 'A' , 'C' , conditions = 'will_fail' )
...
# queued=False
machine . jump ()
> >> False
# queued=True
machine . jump ()
> >> True Cuando se elimina un modelo de la máquina, transitions también eliminarán todos los eventos relacionados de la cola.
class Model :
def on_enter_B ( self ):
self . to_C () # add event to queue ...
self . machine . remove_model ( self ) # aaaand it's gone A veces solo desea que se ejecute una transición particular si ocurre una condición específica. Puedes hacer esto pasando un método, o una lista de métodos, en el 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' ]) En el ejemplo anterior, llamar heat() cuando el modelo está en estado 'solid' pasará al estado 'gas' si is_flammable devuelve True . De lo contrario, pasará al estado 'liquid' si is_really_hot devuelve True .
Por conveniencia, también hay un argumento 'unless' que se comporta exactamente como las condiciones, pero al revés:
machine . add_transition ( 'heat' , 'solid' , 'gas' , unless = [ 'is_flammable' , 'is_really_hot' ]) En este caso, el modelo pasaría de sólido a gas cada vez que se active heat() , siempre que tanto is_flammable() como is_really_hot() devuelvan False .
Tenga en cuenta que los métodos de verificación de condiciones recibirán pasivamente argumentos opcionales y/u objetos de datos pasados a los métodos de activación. Por ejemplo, la siguiente llamada:
lump . heat ( temp = 74 )
# equivalent to lump.trigger('heat', temp=74) ... pasaría el kwarg opcional temp=74 a la verificación is_flammable() (posiblemente envuelto en una instancia EventData ). Para obtener más información sobre esto, consulte la sección Pasar datos a continuación.
Si desea asegurarse de que una transición sea posible antes de continuar con ella, puede usar las funciones may_<trigger_name> que se han agregado a su modelo. Su modelo también contiene la función may_trigger para verificar un disparador por nombre:
# check if the current temperature is hot enough to trigger a transition
if lump . may_heat ():
# if lump.may_trigger("heat"):
lump . heat () Esto ejecutará todas las devoluciones de llamada prepare y evaluará las condiciones asignadas a las posibles transiciones. Las comprobaciones de transición también se pueden utilizar cuando el destino de una transición no está disponible (todavía):
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 Puede adjuntar devoluciones de llamada a transiciones y estados. Cada transición tiene atributos 'before' y 'after' que contienen una lista de métodos para llamar antes y después de que se ejecute la transición:
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?" También hay una devolución de llamada 'prepare' que se ejecuta tan pronto como comienza una transición, antes de que se verifiquen las 'conditions' o se ejecuten otras devoluciones de llamada.
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!" Tenga en cuenta que no se llamará prepare a menos que el estado actual sea una fuente válida para la transición nombrada.
Las acciones predeterminadas que deben ejecutarse antes o después de cada transición se pueden pasar a Machine durante la inicialización con before_state_change y 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?" También hay dos palabras clave para las devoluciones de llamada que deben ejecutarse independientemente a) de cuántas transiciones son posibles, b) si alguna transición tiene éxito y c) incluso si se genera un error durante la ejecución de alguna otra devolución de llamada. Las devoluciones de llamada pasadas a Machine con prepare_event se ejecutarán una vez antes de que se procesen las posibles transiciones (y sus devoluciones de llamada prepare individuales). Las devoluciones de llamada de finalize_event se ejecutarán independientemente del éxito de las transiciones procesadas. Tenga en cuenta que si se produce un error, se adjuntará a event_data como error y se podrá recuperar con 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 A veces las cosas simplemente no funcionan como se esperaba y necesitamos manejar las excepciones y limpiar el desorden para que todo siga funcionando. Podemos pasar devoluciones de llamada a on_exception para hacer esto:
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 probablemente ya se habrá dado cuenta, la forma estándar de pasar invocables a estados, condiciones y transiciones es por nombre. Al procesar devoluciones de llamada y condiciones, transitions usarán su nombre para recuperar el elemento invocable relacionado del modelo. Si el método no se puede recuperar y contiene puntos, transitions tratarán el nombre como una ruta a una función del módulo e intentarán importarlo. Alternativamente, puede pasar nombres de propiedades o atributos. Se incluirán en funciones pero no podrán recibir datos de eventos por razones obvias. También puede pasar elementos invocables como funciones (enlazadas) directamente. Como se mencionó anteriormente, también puede pasar listas/tuplas de nombres invocables a los parámetros de devolución de llamada. Las devoluciones de llamada se ejecutarán en el orden en que se agregaron.
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 () La resolución invocable se realiza en Machine.resolve_callable . Este método se puede anular en caso de que se requieran estrategias de resolución invocables más complejas.
Ejemplo
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 ) En resumen, actualmente existen tres formas de desencadenar eventos. Puede llamar a las funciones de conveniencia de un modelo como lump.melt() , ejecutar activadores por nombre como lump.trigger("melt") o enviar eventos en múltiples modelos con machine.dispatch("melt") (consulte la sección sobre múltiples modelos en patrones de inicialización alternativos). Luego, las devoluciones de llamada sobre las transiciones se ejecutan en el siguiente orden:
| Llamar de vuelta | Estado actual | Comentarios |
|---|---|---|
'machine.prepare_event' | source | ejecutado una vez antes de que se procesen las transiciones individuales |
'transition.prepare' | source | ejecutado tan pronto como comienza la transición |
'transition.conditions' | source | las condiciones pueden fallar y detener la transición |
'transition.unless' | source | las condiciones pueden fallar y detener la transición |
'machine.before_state_change' | source | devoluciones de llamada predeterminadas declaradas en el modelo |
'transition.before' | source | |
'state.on_exit' | source | devoluciones de llamada declaradas en el estado de origen |
<STATE CHANGE> | ||
'state.on_enter' | destination | devoluciones de llamada declaradas en el estado de destino |
'transition.after' | destination | |
'machine.on_final' | destination | las devoluciones de llamada a niños se llamarán primero |
'machine.after_state_change' | destination | devoluciones de llamada predeterminadas declaradas en el modelo; También será llamado después de transiciones internas. |
'machine.on_exception' | source/destination | Las devoluciones de llamada se ejecutarán cuando se haya generado una excepción. |
'machine.finalize_event' | source/destination | Las devoluciones de llamada se ejecutarán incluso si no se produjo ninguna transición o se generó una excepción. |
Si alguna devolución de llamada genera una excepción, el procesamiento de las devoluciones de llamada no continúa. Esto significa que cuando ocurre un error antes de la transición (en state.on_exit o antes), se detiene. En caso de que haya un aumento después de que se haya realizado la transición (en state.on_enter o posterior), el cambio de estado persiste y no se produce ninguna reversión. Las devoluciones de llamada especificadas en machine.finalize_event siempre se ejecutarán a menos que la excepción sea generada por una devolución de llamada de finalización. Tenga en cuenta que cada secuencia de devolución de llamada debe finalizar antes de que se ejecute la siguiente etapa. El bloqueo de las devoluciones de llamada detendrá la orden de ejecución y, por lo tanto, bloqueará el trigger o la llamada dispatch . Si desea que las devoluciones de llamada se ejecuten en paralelo, puede echar un vistazo a las extensiones AsyncMachine para procesamiento asincrónico o LockedMachine para subprocesamiento.
A veces es necesario pasar a las funciones de devolución de llamada registradas en la inicialización de la máquina algunos datos que reflejan el estado actual del modelo. Transitions le permite hacer esto de dos maneras diferentes.
Primero (el valor predeterminado), puede pasar cualquier argumento posicional o de palabra clave directamente a los métodos de activación (creados cuando llama 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.'Puede pasar cualquier número de argumentos que desee al disparador.
Hay una limitación importante en este enfoque: cada función de devolución de llamada activada por la transición de estado debe poder manejar todos los argumentos. Esto puede causar problemas si cada una de las devoluciones de llamada espera datos algo diferentes.
Para solucionar esto, Transitions admite un método alternativo para enviar datos. Si configura send_event=True en la inicialización Machine , todos los argumentos de los activadores se incluirán en una instancia EventData y se pasarán a cada devolución de llamada. (El objeto EventData también mantiene referencias internas al estado de origen, modelo, transición, máquina y disparador asociados con el evento, en caso de que necesite acceder a ellos por cualquier motivo).
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.' En todos los ejemplos hasta ahora, hemos adjuntado una nueva instancia Machine a un modelo separado ( lump , una instancia de la clase Matter ). Si bien esta separación mantiene las cosas ordenadas (porque no es necesario parchear un montón de métodos nuevos en la clase Matter ), también puede resultar molesto, ya que requiere que se realice un seguimiento de qué métodos se llaman en la máquina de estados. y cuáles se llaman en el modelo al que está vinculada la máquina de estados (por ejemplo, lump.on_enter_StateA() frente a machine.add_transition() ).
Afortunadamente, Transitions es flexible y admite otros dos patrones de inicialización.
Primero, puede crear una máquina de estados independiente que no requiera ningún otro modelo. Simplemente omita el argumento del modelo durante la inicialización:
machine = Machine ( states = states , transitions = transitions , initial = 'solid' )
machine . melt ()
machine . state
> >> 'liquid' Si inicializa la máquina de esta manera, puede adjuntar todos los eventos desencadenantes (como evaporate() , sublimate() , etc.) y todas las funciones de devolución de llamada directamente a la instancia Machine .
Este enfoque tiene la ventaja de consolidar toda la funcionalidad de la máquina de estados en un solo lugar, pero puede parecer un poco antinatural si cree que la lógica de estado debería estar contenida dentro del modelo mismo en lugar de en un controlador separado.
Un enfoque alternativo (potencialmente mejor) es hacer que el modelo herede de la clase Machine . Transitions está diseñado para admitir la herencia sin problemas. (¡solo asegúrese de anular el método __init__ de la clase 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' Aquí puede consolidar todas las funciones de la máquina de estado en su modelo existente, lo que a menudo parece más natural que colocar todas las funciones que queremos en una instancia Machine independiente y separada.
Una máquina puede manejar múltiples modelos que se pueden pasar como una lista como Machine(model=[model1, model2, ...]) . En los casos en los que desee agregar modelos además de la instancia de la máquina en sí, puede pasar el marcador de posición de la variable de clase (cadena) Machine.self_literal durante la inicialización como Machine(model=[Machine.self_literal, model1, ...]) . También puede crear una máquina independiente y registrar modelos dinámicamente a través de machine.add_model pasando model=None al constructor. Además, puede utilizar machine.dispatch para activar eventos en todos los modelos agregados actualmente. Recuerde llamar a machine.remove_model si la máquina es duradera y sus modelos son temporales y deben recolectarse como basura:
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 Si no proporciona un estado inicial en el constructor de la máquina de estados, transitions crearán y agregarán un estado predeterminado llamado 'initial' . Si no desea un estado inicial predeterminado, puede pasar initial=None . Sin embargo, en este caso es necesario pasar un estado inicial cada vez que agrega un 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' ) Los modelos con múltiples estados podrían conectar múltiples máquinas usando diferentes valores model_attribute . Como se mencionó en Verificación del estado, esto agregará funciones personalizadas 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 Las transiciones incluyen capacidades de registro muy rudimentarias. Una serie de eventos (a saber, cambios de estado, activadores de transición y comprobaciones condicionales) se registran como eventos de nivel INFO utilizando el módulo logging estándar de Python. Esto significa que puede configurar fácilmente el registro en la salida estándar en un 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' )
... Las máquinas son decapables y se pueden almacenar y cargar con pickle . Para Python 3.3 y versiones anteriores se requiere 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 probablemente habrás notado, transitions utilizan algunas de las características dinámicas de Python para brindarte formas prácticas de manejar modelos. Sin embargo, a los verificadores de tipos estáticos no les gusta que los atributos y métodos del modelo no se conozcan antes del tiempo de ejecución. Históricamente, transitions tampoco asignaban métodos de conveniencia ya definidos en los modelos para evitar anulaciones accidentales.
¡Pero no te preocupes! Puede utilizar el parámetro del constructor de la máquina model_override para cambiar cómo se decoran los modelos. Si configura model_override=True , transitions solo anularán los métodos ya definidos. Esto evita que aparezcan nuevos métodos en tiempo de ejecución y también le permite definir qué métodos auxiliares desea utilizar.
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" Si desea utilizar todas las funciones de conveniencia y agregar algunas devoluciones de llamada a la mezcla, definir un modelo puede volverse bastante complicado cuando tiene muchos estados y transiciones definidas. El método generate_base_model en transitions puede generar un modelo base a partir de la configuración de una máquina para ayudarle con eso.
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 los métodos del modelo que se anularán añade un poco de trabajo adicional. Puede resultar engorroso alternar para asegurarse de que los nombres de los eventos estén escritos correctamente, especialmente si los estados y las transiciones se definen en listas antes o después de su modelo. Puede reducir el texto estándar y la incertidumbre de trabajar con cadenas definiendo estados como enumeraciones. También puedes definir transiciones directamente en tu clase de modelo con la ayuda de add_transitions y event . Depende de usted si usa el decorador de funciones add_transitions o event para asignar valores a los atributos, según su estilo de código preferido. Ambos funcionan de la misma manera, tienen la misma firma y deberían dar como resultado (casi) las mismas sugerencias de tipo IDE. Como todavía es un trabajo en progreso, necesitarás crear una clase Machine personalizada y usar with_model_definitions para las transiciones para verificar las transiciones definidas de esa manera.
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 . AAunque el núcleo de las transiciones se mantiene liviano, existe una variedad de MixIns para ampliar su funcionalidad. Actualmente soportados son:
Hay dos mecanismos para recuperar una instancia de máquina de estado con las funciones deseadas habilitadas. El primer enfoque hace uso de la factory de conveniencia con los cuatro parámetros graph , nested , locked o asyncio configurado en True si la función es necesaria:
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 ) Este enfoque apunta al uso experimental ya que en este caso no es necesario conocer las clases subyacentes. Sin embargo, las clases también se pueden importar directamente desde transitions.extensions . El esquema de nomenclatura es el siguiente:
| Diagramas | anidado | bloqueado | asincio | |
|---|---|---|---|---|
| Máquina | ✘ | ✘ | ✘ | ✘ |
| Máquina gráfica | ✓ | ✘ | ✘ | ✘ |
| Máquina Jerárquica | ✘ | ✓ | ✘ | ✘ |
| Máquina bloqueada | ✘ | ✘ | ✓ | ✘ |
| Máquina de gráficos jerárquicos | ✓ | ✓ | ✘ | ✘ |
| Máquina De Gráficos Bloqueada | ✓ | ✘ | ✓ | ✘ |
| Máquina jerárquica bloqueada | ✘ | ✓ | ✓ | ✘ |
| Máquina de gráficos jerárquicos bloqueados | ✓ | ✓ | ✓ | ✘ |
| Máquina asíncrona | ✘ | ✘ | ✘ | ✓ |
| Máquina de gráficos asíncronos | ✓ | ✘ | ✘ | ✓ |
| JerárquicoAsyncMachine | ✘ | ✓ | ✘ | ✓ |
| JerárquicoAsyncGraphMachine | ✓ | ✓ | ✘ | ✓ |
Para utilizar una máquina de estados rica en funciones, se podría escribir:
from transitions . extensions import LockedHierarchicalGraphMachine as LHGMachine
machine = LHGMachine ( model , states , transitions ) Transitions incluye un módulo de extensión que permite anidar estados. Esto nos permite crear contextos y modelar casos donde los estados están relacionados con ciertas subtareas en la máquina de estados. Para crear un estado anidado, importe NestedState desde transiciones o use un diccionario con los argumentos de inicialización name y children . Opcionalmente, se puede utilizar initial para definir un subestado al que transitar, cuando se ingresa al estado anidado.
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') Una configuración que utilice initial podría verse así:
# ...
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' ]
]
# ... La palabra clave initial del constructor HierarchicalMachine acepta estados anidados (por ejemplo, initial='caffeinated_running' ) y una lista de estados que se considera un estado paralelo (por ejemplo, initial=['A', 'B'] ) o el estado actual de otro modelo ( initial=model.state ) que debería ser efectivamente una de las opciones mencionadas anteriormente. Tenga en cuenta que al pasar una cadena, transition comprobará el estado objetivo en busca de subestados initial y lo utilizará como estado de entrada. Esto se hará de forma recursiva hasta que un subestado no mencione un estado inicial. Los estados paralelos o un estado aprobado como lista se utilizarán "tal cual" y no se realizarán más evaluaciones iniciales.
Tenga en cuenta que su objeto de estado creado previamente debe ser un NestedState o una clase derivada del mismo. La clase State estándar utilizada en instancias Machine simples carece de las características necesarias para el anidamiento.
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! Algunas cosas que se deben considerar cuando se trabaja con estados anidados: Los nombres de los estados se concatenan con NestedState.separator . Actualmente, el separador está configurado para subrayar ('_') y, por lo tanto, se comporta de manera similar a la máquina básica. Esto significa que foo_bar conocerá una bar de subestado del estado foo . Un subestado baz de bar se denominará foo_bar_baz y así sucesivamente. Al ingresar a un subestado, se llamará enter para todos los estados principales. Lo mismo ocurre con los subestados que salen. En tercer lugar, los estados anidados pueden sobrescribir el comportamiento de transición de sus padres. Si no se conoce una transición en el estado actual, se delegará a su padre.
Esto significa que en la configuración estándar, los nombres de estado en los HSM NO DEBEN contener guiones bajos. Para transitions es imposible saber si machine.add_state('state_name') debe agregar un estado llamado state_name o agregar un name de subestado al estado state . Sin embargo, en algunos casos esto no es suficiente. Por ejemplo, si los nombres de los estados constan de más de una palabra y desea/necesita utilizar un guión bajo para separarlos en lugar de CamelCase . Para solucionar esto, puedes cambiar el carácter utilizado para la separación con bastante facilidad. Incluso puedes usar caracteres Unicode elegantes si usas Python 3. Sin embargo, establecer el separador en algo más que guión bajo cambia parte del comportamiento (auto_transition y configuración de devoluciones de llamada):
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') En lugar de to_C_3_a() la transición automática se llama como to_C.s3.a() . Si su subestado comienza con un dígito, las transiciones agregan un prefijo 's' ('3' se convierte en 's3') al FunctionWrapper de transición automática para cumplir con el esquema de nomenclatura de atributos de Python. Si no se requiere la finalización interactiva, se puede llamar directamente to('C↦3↦a') . Además, on_enter/exit_<<state name>> se reemplaza con on_enter/exit(state_name, callback) . Los controles estatales pueden realizarse de manera similar. En lugar de is_C_3_a() , se puede utilizar la variante FunctionWrapper is_C.s3.a() .
Para comprobar si el estado actual es un subestado de un estado específico, is_state admite la palabra clave 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 También puedes usar enumeraciones en HSM, pero ten en cuenta que Enum se comparan por valor. Si tiene un valor más de una vez en un árbol de estados, esos estados no se pueden distinguir.
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 se ha reescrito desde cero para admitir estados paralelos y un mejor aislamiento de estados anidados. Esto implica algunos ajustes basados en los comentarios de la comunidad. Para tener una idea del procesamiento del pedido y la configuración, observe el siguiente ejemplo:
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 Cuando se usa parallel en lugar de children , transitions ingresarán a todos los estados de la lista pasada al mismo tiempo. El subestado al que se debe ingresar está definido por initial , que siempre debe apuntar a un subestado directo. Una característica novedosa es definir transiciones locales pasando la palabra clave transitions en una definición de estado. La transición definida anteriormente ['go', 'a', 'b'] solo es válida en C_1 . Si bien puede hacer referencia a subestados como se hace en ['go', '2_z', '2_x'] no puede hacer referencia a estados principales directamente en transiciones definidas localmente. Cuando se sale de un estado padre, también se saldrá de sus hijos. Además del orden de procesamiento de las transiciones conocido de Machine , donde las transiciones se consideran en el orden en que se agregaron, HierarchicalMachine también considera la jerarquía. Las transiciones definidas en subestados se evaluarán primero (por ejemplo, C_1_a se deja antes de C_2_z ) y las transiciones definidas con comodín * (por ahora) solo agregarán transiciones a estados raíz (en este ejemplo A , B , C ). A partir de 0.8.0 estados anidados se puede agregar directamente y emitirá la creación de los estados matriz sobre la marcha:
m = HierarchicalMachine ( states = [ 'A' ], initial = 'A' )
m . add_state ( 'B_1_a' )
m . to_B_1 ()
assert m . is_B ( allow_substates = True ) Experimental en 0.9.1: Puede hacer uso de devoluciones de llamada on_final en los estados o en el propio HSM. Las devoluciones de llamada se activarán si a) el estado en sí está etiquetado con final y se acaba de ingresar ob) todos los subestados se consideran definitivos y al menos un subestado acaba de ingresar a un estado final. En el caso de B) todos los padres también se considerarán finales si la condición b) es cierto para ellos. Esto podría ser útil en los casos en que el procesamiento ocurre en paralelo y su HSM o cualquier estado principal debe ser notificado cuando todos los subtitados hayan alcanzado un 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! Además del orden semántico, los estados anidados son muy útiles si desea especificar máquinas estatales para tareas específicas y planea reutilizarlas. Antes de 0.8.0 , una HierarchicalMachine no integraría la instancia de la máquina en sí, sino los estados y las transiciones creando copias de ellos. Sin embargo, dado que las instancias (Nested)State de 0.8.0 (anidadas) se hacen referencias , lo que significa que los cambios en la colección de estados y eventos de una máquina influirán en la otra instancia de la máquina. Sin embargo, los modelos y su estado no se compartirán. Tenga en cuenta que los eventos y las transiciones también se copian por referencia y serán compartidos por ambas instancias si no usa la palabra clave remap . Este cambio se realizó para estar más en línea con Machine que también utiliza instancias State aprobadas por referencia.
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 Si se pasa una HierarchicalMachine con la palabra clave children , el estado inicial de esta máquina se asignará al nuevo estado principal. En el ejemplo anterior, vemos que ingresar counting también ingresará counting_1 . Si este es un comportamiento no deseado y la máquina debería detenerse en el estado principal, el usuario puede pasar initial como False como {'name': 'counting', 'children': counter, 'initial': False} .
A veces desea que una colección estatal tan integrada 'regrese', lo que significa que después de que se realice, debe salir y transitar a uno de sus súper estados. Para lograr este comportamiento, puede reasignar las transiciones de estado. En el ejemplo anterior nos gustaría que el contador regrese si se alcanzara el estado done . Esto se hace de la siguiente manera:
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 se mencionó anteriormente, el uso de remap copiará eventos y transiciones ya que no podrían ser válidos en la máquina de estado original. Si una máquina de estado reutilizada no tiene un estado final, por supuesto, puede agregar las transiciones manualmente. Si 'Counter' no tuviera estado 'hecho', podríamos simplemente agregar ['done', 'counter_3', 'waiting'] para lograr el mismo comportamiento.
En los casos en que desea que los estados y las transiciones se copien por valor en lugar de referencia (por ejemplo, si desea mantener el comportamiento anterior a 0.8), puede hacerlo creando un NestedState y asignando copias profundas de los eventos y los estados de la máquina a él.
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 complejas, compartir configuraciones en lugar de máquinas instanciadas podría ser más factible. Especialmente porque las máquinas instanciadas deben derivarse de HierarchicalMachine . Dichas configuraciones se pueden almacenar y cargar fácilmente a través de JSON o YAML (ver las preguntas frecuentes). HierarchicalMachine permite definir sustitutos con la palabra clave children o states . Si ambos están presentes, solo children serán considerados.
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 ()Palabras clave adicionales:
title (opcional): establece el título de la imagen generada.show_conditions (falso predeterminado): muestra condiciones en los bordes de transiciónshow_auto_transitions (falso predeterminado): muestra transiciones automáticas en gráficoshow_state_attributes (falso predeterminado): show llamadas (ingresar, salir), etiquetas y tiempos de espera en gráficoLas transiciones pueden generar diagramas de estado básicos que muestran todas las transiciones válidas entre los estados. El soporte del diagrama básico genera una definición de máquina de estado de Mermaid que se puede utilizar con el editor en vivo de Mermaid, en archivos de Markdown en GitLab o GitHub y otros servicios web. Por ejemplo, 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!" )Produce este diagrama (verifique la fuente del documento para ver la notación de Markdown):
---
Gráfico de sirena
---
statediagram-v2
Dirección LR
classdef s_default llena: blanco, color: negro
Classdef S_inactive relleno: blanco, color: negro
Classdef S_parallel Color: Negro, Llena: Blanco
Classdef S_Active Color: Rojo, Llena: Darksalmon
Classdef S_Previous Color: Azul, Llena: Azure
declarar "a" como un
Clase A S_Previous
Estado "B" como B
Clase B S_Active
Estado "C" como C
C -> [*]
Clase C S_DEFAULT
estado c {
Estado "1" como C_1
Estado C_1 {
[*] -> C_1_A
Indique "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
Indique "A" como C_2_A
Estado "B" como C_2_B
C_2_b -> [*]
}
}
C -> A: reiniciar
A -> B: init
B -> C: hacer
C_1_A -> C_1_B: GO
C_2_A -> C_2_B: ir
[*] -> A
Para usar una funcionalidad gráfica más sofisticada, deberá tener instalados graphviz y/o pygraphviz . Para generar gráficos con el paquete graphviz , debe instalar GraphViz manualmente o mediante un administrador de paquetes.
sudo apt-get install graphviz graphviz-dev # Ubuntu and Debian
brew install graphviz # MacOS
conda install graphviz python-graphviz # (Ana)conda
Ahora puede instalar los paquetes de Python reales
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
Actualmente, GraphMachine usará pygraphviz cuando esté disponible y volverá a graphviz cuando no se puede encontrar pygraphviz . Si graphviz tampoco está disponible, mermaid se utilizará. Esto se puede anular pasando graph_engine="graphviz" (o "mermaid" ) al constructor. Tenga en cuenta que este valor predeterminado podría cambiar en el futuro y se puede eliminar el soporte pygraphviz . Con Model.get_graph() puede obtener el gráfico actual o la región de interés (ROI) y dibujarlo así:
# 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' )Esto produce algo como esto:
Independientemente del backend que usa, la función de dibujo también acepta un descriptor de archivo o un flujo binario como primer argumento. Si establece este parámetro en None , se devolverá el flujo de bytes:
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 ()Las referencias y parciales aprobados como devoluciones de llamada se resolverán lo más bien posible:
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' )Esto debería producir algo similar a esto:
Si el formato de referencias no se adapta a sus necesidades, puede anular el método estático GraphMachine.format_references . Si desea omitir la referencia por completo, simplemente deje que GraphMachine.format_references no devuelva None . Además, eche un vistazo a nuestro ejemplo de cuadernos Ipython/Jupyter para un ejemplo más detallado sobre cómo usar y editar gráficos.
En los casos en que el envío de eventos se realiza en hilos, se puede usar LockedMachine omachine LockedHierarchicalMachine donde el acceso a la función (! Sic) está asegurado con cerraduras reentrantes. Esto no le evita corromper a su máquina jugando con variables de miembros de su modelo o 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 Cualquier administrador de contexto de Python puede aprobarse a través del argumento de palabras clave 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 ]) Cualquier contexto a través de machine_model se compartirá entre todos los modelos registrados con la Machine . También se pueden agregar contextos por modelo:
lock3 = RLock ()
machine . add_model ( model , model_context = lock3 )Es importante que todos los gerentes de contexto proporcionados por el usuario sean reentrantes ya que la máquina de estado los llamará varias veces, incluso en el contexto de una sola invocación de activación.
Si está utilizando Python 3.7 o posterior, puede usar AsyncMachine para trabajar con devoluciones de llamada asíncronas. Puede mezclar devoluciones de llamada sincrónicas y asincrónicas si lo desea, pero esto puede tener efectos secundarios no deseados. Tenga en cuenta que los eventos deben esperarse y que usted también debe manejar el bucle del evento.
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 () Entonces, ¿por qué necesitas usar Python 3.7 o más tarde? El soporte de Async se ha introducido anteriormente. AsyncMachine utiliza contextvars para manejar la ejecución de devoluciones de llamada cuando llegan nuevos eventos antes de que se haya finalizado una transición:
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 ejemplo en realidad ilustra dos cosas: primero, que 'go' llamado en la transición de M1 de A a Be B no se cancela y el segundo, llamar m2.fix() detendrá el intento de transición de M2 de A a B ejecutando 'Fix' de A a C . Esta separación no sería posible sin contextvars . Tenga en cuenta que prepare y conditions no se tratan como transiciones continuas. Esto significa que después de que se han evaluado conditions , se ejecuta una transición a pesar de que ya ocurrió otro evento. Las tareas solo se cancelarán cuando se ejecuten como una devolución de llamada before o posterior.
AsyncMachine presenta un modo de cola especial del modelo que se puede usar cuando queued='model' se pasa al constructor. Con una cola específica del modelo, los eventos solo se colocarán en cola cuando pertenecen al mismo modelo. Además, una excepción elevada solo eliminará la cola de eventos del modelo que planteó esa excepción. En aras de la simplicidad, supongamos que cada evento en asyncio.gather a continuación no se desencadena al mismo tiempo, sino un poco retrasado:
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.event3Tenga en cuenta que los modos de cola no deben cambiarse después de la construcción de la máquina.
Si sus superhéroes necesitan un comportamiento personalizado, puede incluir una funcionalidad adicional mediante la decoración de los estados de la 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 aloneActualmente, las transiciones vienen equipadas con las siguientes características del estado:
Tiempo de espera : desencadena un evento después de pasar un tiempo
timeout (int, opcional): si se pasa, un estado ingresado tiempo después de timeout segundoson_timeout (string/callable, opcional) - se llamará cuando se haya alcanzado el tiempo de tiempo de esperaAttributeError cuando se establece timeout , pero on_timeout no esEtiquetas : agrega etiquetas a los estados
tags (lista, opcional): asigna etiquetas a un estadoState.is_<tag_name> devolverá True cuando el estado haya sido etiquetado con tag_name , de lo contrario False Error : plantea un MachineError cuando no se puede dejar un estado
Tags (si usa Error no use Tags )accepted (bool, opcional): marca un estado según lo aceptadotags de palabras clave, que contienen 'aceptado'auto_transitions se ha establecido en False . De lo contrario, cada estado se puede salir con métodos to_<state> .Volátil : inicializa un objeto cada vez que se ingresa un estado
volatile (clase, opcional): cada vez que se ingrese el estado, se asignará un objeto de clase de tipo al modelo. El nombre del atributo se define por hook . Si se omite, se creará un VolatileObject vacío en su lugarhook (String, default = 'Scope'): el nombre del atributo del modelo para el objeto temporal. Puede escribir sus propias extensiones State y agregarlas de la misma manera. Simplemente tenga en cuenta que add_state_features espera mezclas . Esto significa que su extensión siempre debe llamar a los métodos anulados __init__ , enter y exit . Su extensión puede heredar del estado , pero también funcionará sin ella. Usar @add_state_features tiene un inconveniente que es que las máquinas decoradas no se pueden encurtir (más precisamente, el CustomState generado dinámicamente no se puede encurtir). Esta podría ser una razón para escribir una clase estatal personalizada dedicada. Dependiendo de la máquina de estado elegida, su clase estatal personalizada puede necesitar proporcionar ciertas características de estado. Por ejemplo, HierarchicalMachine requiere que su estado personalizado sea una instancia de NestedState ( State no es suficiente). Para inyectar sus estados, puede asignarlos al atributo de clase de su Machine state_cls o anular Machine.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 ) Si desea evitar por completo los hilos en su AsyncMachine , puede reemplazar la función de estado Timeout con AsyncTimeout desde la extensión 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 Debe considerar pasar queued=True al constructor TimeoutMachine . Esto asegurará que los eventos se procesen secuencialmente y eviten las condiciones de carreras asincrónicas que pueden aparecer cuando el tiempo de espera y el evento ocurren en proximidad.
Puede echar un vistazo a las preguntas frecuentes para obtener alguna inspiración o contratar django-transitions . Ha sido desarrollado por Christian Ledermann y también está alojado en Github. La documentación contiene algunos ejemplos de uso.
Primero, ¡felicitaciones! ¡Llegaste al final de la documentación! Si desea probar transitions antes de instalarlo, puede hacerlo en un cuaderno interactivo de Jupyter en mybinder.org. Simplemente haga clic en este botón.
Para informes de errores y otros problemas, abra un problema en GitHub.
Para preguntas de uso, publique en Stack Overflow, asegurándose de etiquetar su pregunta con la etiqueta pytransitions . ¡No olvides echar un vistazo a los ejemplos extendidos!
Para cualquier otra pregunta, solicitudes o grandes obsequios monetarios sin restricciones, envíe un correo electrónico a Tal Yarkoni (autor inicial) y/o Alexander Neumann (mantenedor actual).
