automat

automat copied to clipboard

With the coffee machine example in the readme, we go from 3 state attributes (beans, water, lid), to just one (beans), to make the following examples simpler. I'm a little confused about the complex case though. It seems like with n independent boolean state attributes, you need 2^n automat states, as well as appropriate upons . So for the coffee machine example:

@_machine.state(initial=True)
def no_beans_no_water_open_lid(self):
    pass
@_machine.state()
def no_beans_water_open_lid(self):
    pass
@_machine.state()
def no_beans_no_water_closed_lid(self):
    pass
@_machine.state()
def no_beans_water_closed_lid(self):
    pass
@_machine.state()
def beans_no_water_open_lid(self):
    pass
@_machine.state()
def beans_water_open_lid(self):
    pass
@_machine.state()
def beans_no_water_closed_lid(self):
    pass
@_machine.state()
def beans_water_closed_lid(self):
    pass

Or am I just thinking about this completely wrong?

Or am I just thinking about this completely wrong?

In terms of representing all the possible states: you've totally got the right idea. An exhaustive exploration of all possible states is (inherently) combinatorial.

The idea behind using Automat is that you probably don't need to model the entire state space; if someone goes outside the bounds you've defined in your simplified model, you just want them to get an error, not a potentially invalid method invocation. In other words, consider Automat to be a guard around your code, making sure it won't be run and accidentally corrupt state if it hits a sequence of states you didn't anticipate.

For state machines that legitimately have large (> 4) numbers of independent boolean states, modeling them as a single machine with named states is probably not appropriate, and it would make sense to split them into multiple communicating machines. For a more detailed simulation of this coffee machine, for example, we'd probably want to separately model the water tank and the bean receptacle. Exhaustively modeling this would probably require #41, so we should implement that :), because the state machines would need to communicate in potentially mutually recursive/reentrant ways (brewing would drain the water tank, for example, which might emit an output that would affect the heating element…)

So, with that description of the present and existing plans in mind, it might be worth discussing future directions a little bit.

It might be interesting to figure out some other way to deal with denser matrices of boolean flags in individual state machines as well. I don't have any particular ideas in this area, but we could provide invariants or names for flags rather than names for states. If anyone else has an idea about how this might be done I'd be interested to hear it.

Ok, so most state graphs probably won't be as connected as all that? But then would you want to raise errors for certain transitions? Or is that covered by #46?

I think the likely solution to #46 is to allow applications to specify their own exception type for missing transitions.

And yeah, I would imagine the state graph would not be that connected - if you actually have to deal with all the inputs in a completely fully-connected state graph, automat is probably more verbose than just using flags to get to the same result.

But, I think your example is interesting there. You've defined a bunch of extra states above, but what are the interesting extra edges that you imagine are missing in this application? Exploring that might get us to an interesting result.

Well, imagining we had the brew_button input and _heat_the_heating_element output from the readme, and supposing that brewing is only ok from the beans_water_closed_lid state, that gives us the following upon:

beans_water_closed_lid.upon(brew_button,
                            enter=no_beans_no_water_closed_lid,
                            outputs=[_heat_the_heating_element])

But what if we want to give more info about why we can't brew than NoTransition. So we might have something like:

@_machine.output()
def _please_close_lid(self):
    "Tell the user to close the lid."
    self.display_message("please close lid")

beans_water_open_lid.upon(brew_button,
                            enter=beans_water_open_lid,
                            outputs=[ _please_close_lid])

This is sort of like the connected/not_connected/send_message example, but you'd need a bunch of these transitions?

That seems like a pretty specific / conscious choice of behavior though; like, it's not just that you have a common behavior across all "open lid" states; some of the states have higher priority than others, and the output you've written here looks very specific.

Yes. I guess the general case I'm poking at here is where the user won't specifically know what inputs will advance them to a desired state. I mean, I suppose there could be some visual indicator that the brew_button input is active, or a running display of all the things that need to happen before it becomes active?

In the case of such a coffee machine, there's a plethora of user-interface choices once could make (a little readout that would give you messages in english, a hardware interlock on the brew button, blinking lights) but I'm having trouble seeing how this might translate to automat. Maybe the coffee example is just bad for the documentation?

Sorry, I think I got off track a little bit. Let me try and restate. First the two points I am confused about:

1. Multiple independent condition variables producing a large number of states and transitions.
2. Being able to provide specific feedback for why a particular input is not acceptable in the current state.

The following illustrates point 1:

from automat import MethodicalMachine

class CoffeeBrewer(object):
    _machine = MethodicalMachine()
    
    def __init__(self):
        self._beans = "no beans"

    ### States ###

    @_machine.state(initial=True)
    def no_beans_no_water_open_lid(self):
        pass
    @_machine.state()
    def no_beans_water_open_lid(self):
        pass
    @_machine.state()
    def no_beans_no_water_closed_lid(self):
        pass
    @_machine.state()
    def no_beans_water_closed_lid(self):
        pass
    @_machine.state()
    def beans_no_water_open_lid(self):
        pass
    @_machine.state()
    def beans_water_open_lid(self):
        pass
    @_machine.state()
    def beans_no_water_closed_lid(self):
        pass
    @_machine.state()
    def beans_water_closed_lid(self):
        pass

    ### Inputs ###

    @_machine.input()
    def brew_button(self):
        "The user pressed the 'brew' button."
    @_machine.input()
    def put_in_beans(self, beans):
        "The user put in some beans."
    @_machine.input()
    def put_in_water(self):
        "The user put in the water."
    @_machine.input()
    def toggle_lid(self):
        "The user moved the lid."

    ### Outputs ###

    @_machine.output()
    def _heat_the_heating_element(self):
        "Heat up the heating element, which should cause coffee to happen."
    @_machine.output()
    def _save_beans(self, beans):
        "The beans are now in the machine; save them."
        self._beans = beans
    @_machine.output()
    def _pour_coffee(self):
        beans = self._beans
        self._beans = "no beans"
        return f"A cup of coffee made with {beans}."

    ### Valid Transitions ###

    no_beans_no_water_open_lid.upon(put_in_beans, enter=beans_no_water_open_lid,
                                    outputs=[_save_beans])
    no_beans_water_open_lid.upon(put_in_beans, enter=beans_water_open_lid,
                                 outputs=[_save_beans])
    no_beans_no_water_open_lid.upon(put_in_water, enter=no_beans_water_open_lid,
                                    outputs=[])
    beans_no_water_open_lid.upon(put_in_water, enter=beans_water_open_lid,
                                 outputs=[])
    no_beans_no_water_open_lid.upon(toggle_lid, enter=no_beans_no_water_closed_lid,
                                    outputs=[])
    beans_no_water_open_lid.upon(toggle_lid, enter=beans_no_water_closed_lid,
                                 outputs=[])
    no_beans_water_open_lid.upon(toggle_lid, enter=no_beans_water_closed_lid,
                                 outputs=[])
    beans_water_open_lid.upon(toggle_lid, enter=beans_water_closed_lid,
                              outputs=[])
    no_beans_no_water_closed_lid.upon(toggle_lid, enter=no_beans_no_water_open_lid,
                                      outputs=[])
    beans_no_water_closed_lid.upon(toggle_lid, enter=beans_no_water_open_lid,
                                   outputs=[])
    no_beans_water_closed_lid.upon(toggle_lid, enter=no_beans_water_open_lid,
                                   outputs=[])
    beans_water_closed_lid.upon(toggle_lid, enter=beans_water_open_lid,
                                outputs=[])
    beans_water_closed_lid.upon(brew_button, enter=no_beans_no_water_closed_lid,
                                outputs=[_heat_the_heating_element,
                                         _pour_coffee],
                                collector=lambda iterable: list(iterable)[-1])

One solution you mentioned is to use separate, communicating automats, though I'm not sure of the details of this, as say if we try to brew, there are beans but no water, and we empty the beans first, then error out on the water, we'd need some sort of transaction mechanism to restore the beans or something?

I don't really have any better idea of how it would work though.

For point two, the above code emits NoTransition errors when, for example, we try to put in beans while the lid is closed. Even if we were to be able to specify a custom error class, that's not that much info besides "You can't do that". So in the current state, it seems you'd have to specify helpful outputs and transitions, like the following.

    ### Error Outputs ###

    @_machine.output()
    def _beans_lid_closed(self):
        print("Cannot add beans while lid is closed")
    @_machine.output()
    def _water_lid_closed(self):
        print("Cannot add water while lid is closed")
    @_machine.output()
    def _beans_already(self):
        print("There are already beans")
    @_machine.output()
    def _water_already(self):
        print("There is already water")
    @_machine.output()
    def _not_yet(self):
        print("Must have beans, water and lid must be closed.")

    ### Invalid Transitions ###

    no_beans_no_water_closed_lid.upon(put_in_beans, enter=no_beans_no_water_closed_lid,
                                      outputs=[_beans_lid_closed])
    no_beans_water_closed_lid.upon(put_in_beans, enter=no_beans_water_closed_lid,
                                   outputs=[_beans_lid_closed])
    no_beans_no_water_closed_lid.upon(put_in_water, enter=no_beans_no_water_closed_lid,
                                      outputs=[_water_lid_closed])
    beans_no_water_closed_lid.upon(put_in_water, enter=beans_no_water_closed_lid,
                                   outputs=[_water_lid_closed])
    beans_no_water_open_lid.upon(put_in_beans, enter=beans_no_water_open_lid,
                                 outputs=[_beans_already])
    beans_water_open_lid.upon(put_in_beans, enter=beans_water_open_lid,
                              outputs=[_beans_already])
    no_beans_water_open_lid.upon(put_in_water, enter=no_beans_water_open_lid,
                                 outputs=[_water_already])
    beans_water_open_lid.upon(put_in_water, enter=beans_water_open_lid,
                              outputs=[_water_already])
    no_beans_no_water_open_lid.upon(brew_button, enter=no_beans_no_water_open_lid,
                                    outputs=[_not_yet])
    beans_no_water_open_lid.upon(brew_button, enter=beans_no_water_open_lid,
                                    outputs=[_not_yet])
    no_beans_water_open_lid.upon(brew_button, enter=no_beans_water_open_lid,
                                    outputs=[_not_yet])
    beans_water_open_lid.upon(brew_button, enter=beans_water_open_lid,
                                    outputs=[_not_yet])
    no_beans_no_water_closed_lid.upon(brew_button, enter=no_beans_no_water_closed_lid,
                                    outputs=[_not_yet])
    beans_no_water_closed_lid.upon(brew_button, enter=beans_no_water_closed_lid,
                                    outputs=[_not_yet])
    no_beans_water_closed_lid.upon(brew_button, enter=no_beans_water_closed_lid,
                                    outputs=[_not_yet])

I was thinking that, were this a common enough pattern, there could be a method for it instead of upon, that re-entered the same state, took a string and used it to raise a No Transition. sort of like:

no_beans_no_water_closed_lid.cant(put_in_beans, reason="Cannot add beans while lid is closed")

I would also accept that all this is more complicated than automat is wanting to be.

I think that there may be some confusion here about what the inputs in the example are supposed to be.

The implication was that this was some software to automate a coffee machine, and put_in_beans means, as its docstring says, "The user put in some beans.", which is to say that there is some sensor which detects the beans, that has triggered, indicating the beans have already been put in.

The output from such a device when the user attempts to put beans into the machine when the lid is closed would be a bunch of coffee beans in a giant mess on your counter, not a different input to the state machine :-).

The reason that this is important to the example here is that what you're constructing with all these states is a simulation of a coffee machine, which means the inputs are coming from somewhere else, and inherently mean something else.

I would also accept that all this is more complicated than automat is wanting to be.

No, I think that there's a case in here we very much would want Automat to handle. I think it would potentially need some significant new features in order to do it, because what it looks like you want to do here is to express that there's a state matrix that is the cartesian product of (beans, water, lid) and you want to be able to call out individual states or sets of states within that matrix to have particular outputs in aggregate without having to name every single one of them.

I've previously advised many people using automat not to be afraid of writing for loops that add the same edge to half a dozen states at once, and in fact having each be an addressable object like this intentionally makes looping over sets of them easier: so making it easier to scale up your state machines to large numbers of outputs is definitely something I'd be open to adding.

However, I think that when adding features like this it's very easy to go off the rails, so I just want to make sure that the use-case is crystal clear.

Yes, I guess I was definitely thinking more in terms of a simulation of a coffee machine than in terms of control software for a physical coffee machine. When, for example, the result of trying to add beans with the lid closed should be something like "The beans bounce off the closed lid and fall all over the counter. You sheepishly scoop them up and put them back in the jar, hoping nobody else saw that."

Aside from the number of states and transitions and the above error responses, the simulation way of things is complicated by the opaqueness of the state machine, where the actor can't see what state the machine is in to decide what to do next. It's almost like you would want to have the boolean variables anyway, but controlled by the state machine instead of being checked by a pseudo state machine.

Edit: Perhaps the simple answer is not to give an actor direct access to state machine inputs, but to have some wrapper object that knows how to handle no-transition errors?

Perhaps the simple answer is not to give an actor direct access to state machine inputs, but to have some wrapper object that knows how to handle no-transition errors?

Yeah, having some kind of wrapper that can manipulate the state / input / etc objects before they become descriptors (i.e.: at the class scope) would definitely be very welcome.

Oh, I was just thinking like, non-automat methods on an automat-decorated object, that would call the inputs and return normal things in the case of NoTransition. Not sure what you mean by "before they become descriptors"...

I mean in the scope of the class body. They "become descriptors" at the moment that somebody does Foo().bar and the __get__ method gets called.

If you have

class Something(object):
    machine = MethodicalMachine()

it is very important to automat's design that Something().machine not be a public attribute; that is why the __get__ (descriptor) is there. However, MethodicalMachine itself can totally have public methods, that can be called to set up states or perform transformations on the traversal graph; not every edge needs to be created manually.

we could provide invariants or names for flags rather than names for states

What about something that looks like this? See inline comments for notes on changes.

class CoffeeBrewer(object):
    _machine = MethodicalMachine()
    
    @_machine.input()
    def brew_button(self):
        "The user pressed the 'brew' button."

    @_machine.input()
    def put_in_beans(self, beans):
        "The user put in some beans."

    @_machine.output()
    def _heat_the_heating_element(self):
        "Heat up the heating element, which should cause coffee to happen."
        self._heating_element.turn_on()

    @_machine.output()
    def _save_beans(self, beans):
        "The beans are now in the machine; save them."
        self._beans = beans

    # `state`s become `state_flag`s
    # and the possible states for the flag are declared.
    @_machine.state_flag(states=[True, False])
    def has_beans(self):
        "Indicates whether or not the bean hopper has beans."

    @_machine.state_flag(states=[True, False])
    def has_water(self):
        "Indicates whether or not the resivore is full."

    @_machine.state_flag(states=[True, False])
    def lid_is_open(self):
        "Indicates whether or not the lid is open."

    # State transitions are created on the machine 
    # because there is not a state object any more.
    _machine.transition(
        # States are a dict of flag value pairs.
        # transition() can validate the flags and values
        # and raise an error if they don't match the previous declaration.
        starting_state={'has_beans': True, 'has_water': True, 
                        'lid_is_open': False},
        input=brew_button,
        # transition() would also check that the flags present 
        # in the starting_state are exactly the set of flags in ending_state
        ending_state={'has_beans': False, 'has_water': False, 
                      'lid_is_open': False},
        outputs=[_heat_the_heating_element, _describe_coffee],
        collector=lambda iterable: list(iterable)[-1],
    )
    
    _machine.transition(
        # It also makes sense to allow partial state invariants 
        # if the transition only involves a subset of flags.
        starting_state={'has_beans': False},  
        input=put_in_beans,
        ending_state={'has_beans': True},
        outputs=[_save_beans],
    )

This would make it possible to have n possible states per flag.

class CeilingFan(object):
    _machine = MethodicalMachine()

    @_machine.input()
    def pull_fann_speed_cord(self):
        "Toggle the fan speed"

    # Having 4 states for a flag is no problem.
    @_machine.state_flag(states=['off', 'low', 'medium', 'high'])
    def fan_speed(self):
        "The current speed of the fan"

    _machine.transition(
        starting_state={'fan_speed': 'off'},
        input=pull_fann_speed_cord,
        ending_state={'fan_speed': 'high'},
        outputs=[],
    )

    _machine.transition(
        starting_state={'fan_speed': 'high'},
        input=pull_fann_speed_cord,
        ending_state={'fan_speed': 'medium'},
        outputs=[],
    )

    # ...

Minor quibble: I'd want to call it machine.flag, but otherwise this definitely looks interesting. I can't say for sure I'd accept it without looking at more examples, but would you be willing to implement a proof of concept in a PR so some people can play with it and see if it addresses the use-case well?

Sure. I'll give it a go.