Attaching a model with multiple states to multiple machine in Transitions

Viewed 698

Transitions is a python library for implementing state machines.

As mentioned in the documentation of Transitions at the end of alternative initialization patterns section, we can attach multiple machines to a model with multiple states using different model_attribute values.

I am implementing it similar to the way mentioned in the documentation:

class Matter():
    pass

lump = Matter()

states = ['solid','liquid','gas']
transitions = [
    { 'trigger': 'melt', 'source': 'solid', 'dest': 'liquid'},
    { 'trigger': 'evaporate', 'source': 'liquid', 'dest': 'gas'}
]

matter_machine = Machine(lump, states=states, transitions=transitions, initial='solid', model_attribute='state')
shipment_machine = Machine(lump, states=states, transitions=transitions, initial='liquid', model_attribute='shipping_state')


lump.melt()
lump.evaporate()
print(lump.state)
>>> gas
print(lump.shipping_state)
>>> liquid

How shall I access the Machine which has the model_attribute='shipping_state', that is shipment_machine, as all the transitions get carried out only on the first machine that is the matter_machine as it is initialized first. I am not able to perform any transition on the shipment_machine.

Also I would like to inform you that what I intend to do is to have a sub-machine for the matter_machine. That is for example say for a given model I have a machine named Car, I wish to have a sub-machine Driver for it, with their respective states and transitions.

1 Answers

Currently, I see two separate questions here. First, how to work with 'shipment_state' and how to work with nested/parallel states.

Problem 1: A model with two (independent) state attributes.

The most common approach to trigger events on a model is to use the convenience funtions that transitions add to your model. The README states the following:

Each method triggers the corresponding transition. You don't have to explicitly define these methods anywhere; the name of each transition is bound to the model passed to the Machine initializer (in this case, lump). To be more precise, your model should not already contain methods with the same name as event triggers since transitions will only attach convenience methods to your model if the spot is not already taken.

I highlighted the part which causes you trouble here. The default behaviour prevents adding triggers of two different machines with the same name. transitions will only add convenience functions (e.g lump.melt()) if the object does not already contain a method/attribute called 'melt'.

Solution 1.A: Access machine events directly

There are more ways to trigger events than just the added covenience functions. For instance, you could trigger events directly. You need to pass the model in question though:


class Matter():
    pass

lump = Matter()

states = ['solid','liquid','gas']
transitions = [
    { 'trigger': 'melt', 'source': 'solid', 'dest': 'liquid'},
    { 'trigger': 'evaporate', 'source': 'liquid', 'dest': 'gas'}
]

matter_machine = Machine(lump, states=states, transitions=transitions, initial='solid', model_attribute='state')
shipment_machine = Machine(lump, states=states, transitions=transitions, initial='solid', model_attribute='shipping_state')

matter_machine.events['melt'].trigger(lump)  # will process event for 'state'
assert lump.state == 'liquid'
assert lump.shipping_state == 'solid'
shipment_machine.events['melt'].trigger(lump)  # will process event for 'shipment_state'
assert lump.state == lump.shipping_state
assert lump.shipping_state == 'liquid'

Solution 1.B: Override the machines assignment strategy.

The standard behaviour of not touching existing attributes/fields has been adapted to prevent messing with custom code written by the user. Some users only trigger events by name (e.g. model.trigger('melt')) and add methods named like triggers (def melt(self, ...)) to their transition callbacks. But this behaviour can be changed by overriding Machine._checked_assignment. In the FAQ notebook in the section Transitions does not add convencience methods to my model you find an examplary override named CallingMachine:

class CallingMachine(Machine):

    def _checked_assignment(self, model, name, func):
        if hasattr(model, name):
            predefined_func = getattr(model, name)
            def nested_func(*args, **kwargs):
                predefined_func(*args, **kwargs)
                func(*args, **kwargs)
            setattr(model, name, nested_func)
        else:
            setattr(model, name, func)

CallingMachine works like the default machine but will wrap an attribute into a function and call it as well as the event trigger. This is a very simplistic approach though since it doesn't check whether the previous assignment attribute is actually callable. If you look closely, you will also see that I changed predefined_func and pass parameters instead of just calling it like in the FAQ. In your use case we can assume that an already present method is another trigger function. If you want to actually mix callables here, this assumption might cause trouble. Finally, with CallingMachine instead of Machine an event will be triggered on both machines:

matter_machine = CallingMachine(lump, states=states, transitions=transitions, initial='solid', model_attribute='state')
shipment_machine = CallingMachine(lump, states=states, transitions=transitions, initial='solid', model_attribute='shipping_state')

lump.melt()
assert lump.state == lump.shipping_state
assert lump.shipping_state == 'liquid'

Problem 2: Combining different machines/models into a nested structure.

Solution 2.A: Using concurrency and nesting

Besides stitching states together on a class/model level you can also combine states or nest configurations by using concurrent hierarchical state machines. Concurrency has been added recently to transitions in version 0.8.0. For your car/driver example this could look like this:

from transitions.extensions.nesting import HierarchicalMachine as HSM

# our car
car_config = dict(name="car", children=['stopped', 'accelerating', 'decelerating', 'driving'], 
                  initial='stopped',
                  transitions=[
                      ['accelerate', '*', 'accelerating'],  # gotta go fast
                      ['braking', ['driving', 'accelerating'], 'decelerating'],  # slower!
                      ['braking', 'stopped', 'stopped'],  # can't be slower than stopped
                      ['release', ['accelerating', 'decelerating'], 'driving'],  # releasing the pedal will result in a steady speed ...
                      ['release', 'stopped', 'stopped']  # ... which could be 0 km/h
                  ])

driver_config = dict(name="driver", children=['rested', 'tired', 'sleepy', 'jamming'],
                     initial='rested',
                     transitions=[
                         ['driving', 'rested', 'tired'],  # driving can be exhausting
                         ['driving', 'tired', 'sleepy'],  # when you tired you can get even more sleepy
                         ['tunein', '*', 'jamming'],  # music allways helps ...
                         ['driving', 'jamming', 'rested'],  # to replenish some energy
                         ['resting', '*', 'rested']  # sometimes a break is better though
                     ])

states = ['off', {'name': 'running', 'parallel': [car_config, driver_config]}]

m = HSM(states=states, initial='off', auto_transitions=False,
        transitions=[['start', 'off', 'running']])
print(m.state)  # >>> off
m.start()  
print(m.state)  # >>> ['running_car_stopped', 'running_driver_rested']
m.accelerate()
print(m.state)  # >>> ['running_car_accelerating', 'running_driver_rested']
m.driving()
m.driving()
print(m.state)  # >>> ['running_car_accelerating', 'running_driver_sleepy']

As a graph, this looks roughly like this:

enter image description here

Solution 2.B: Machine as a 'rule book' for multiple models

Hierarchical machines and concurrency can get messy (as everything more complex than a conceptional traffic light; I assume the real ones are complex beasts, too). Putting some thoughts into transition/state naming and ignore_invalid_triggers=Truemay be enough to deal with a car/driver example. We merge states and transitions into big sets but make sure that they don't overlap. This way we can manage multiple models/components with one machine as sort of rule book:

from transitions import Machine

class Car:

    def driving(self):
        print("vrooom")
        # e.g. calculate passed distance

class Driver:
    pass

car_states = ['stopped', 'accelerating', 'decelerating', 'driving']
driver_states = ['rested', 'tired', 'sleepy', 'jamming']

transitions = [
    # car transitions
    ['accelerate', car_states, 'accelerating'],
    ['braking', ['driving', 'accelerating'], 'decelerating'],
    ['braking', 'stopped', 'stopped'],
    ['release', ['accelerating', 'decelerating'], 'driving'],
    ['release', 'stopped', 'stopped'],
    # driver
    ['driving', 'rested', 'tired'],
    ['driving', 'tired', 'sleepy'],
    ['tunein', driver_states, 'jamming'],
    ['driving', 'jamming', 'rested'],
    ['resting', driver_states, 'rested']
]

car = Car()
driver = Driver()
# initialize a 'blank' combined rule book
# ignore_invalid_triggers=True allows us to call m.dispatch even when a trigger isn't defined for all models'
# current state
m = Machine(model=None, states=car_states + driver_states, transitions=transitions, auto_transitions=False,
            initial=None, ignore_invalid_triggers=True)
# we add models after construction since they have different initial states
m.add_model(car, initial='stopped')
m.add_model(driver, initial='rested')
car.accelerate()
# dispatch triggers an event on ALL models
m.dispatch('driving')  # >>> vrooom
assert car.is_accelerating()
assert driver.is_tired()
Related