Persistence, timeouts, and branching conditionals?

[NSchrading]

• Python State Machine version: 2.1.2
• Python version: 3.10
• Operating System: Windows 10

Description

Is there a way to define persistence or timeouts for states within the state machine? For instance, if I have a state machine that should be evaluating data on a particular cycle rate, and if the data is "good" for a number of cycles / time, it should transition to a data_good state and if it does not reach that data_good state within a max number of cycles / time it goes to a data_bad state, what is the best way to structure this? Here's a toy example with pseudocode of what I'm trying to accomplish:

class MyStateMachine(StateMachine)
    check_data = State()
    data_good = State()
    data_bad = State()

    cycle = check_data.to(data_good, cond="evaluate_data")

    # must be true for 3 cycles, else it returns False. If it is False for 10 cycles then it raises an exception
    @persistence_cycles(3, 10)
    def evaluate_data(self, model):
        return model.data.get("value") > 10.0
    
    
# main logic
# separate thread is updating model.data on a cycle rate
while True:
    my_statemachine.cycle()
    time.sleep(cycle_rate)

I can make a decorator, persistence_cycles to accomplish the transition to data_good, but I want it to transition also to data_bad if the max cycles is reached. I tried making the decorator raise an exception and then use validators like cycle = check_data.to(data_good, cond="evaluate_data") | check_data.to(data_bad, validators="evaluate_data") but I believe because evaluate_data returns True/False that validator immediately transitions to data_bad.

Separately, I'm wondering if there is a better way to structure this kind of thing. I like being able to have a single cycle event with various cond to move through states, as it vastly simplifies the main logic. This is sort of related to the discussion here: #420. For more complex branching with timeouts and abort conditions, I'm wondering if it is necessary to move the logic out into separate threads that trigger events to move between states. What I want is to be able to easily define actions that the state machine takes on each cycle depending on the state that it is in.

[fgmacedo]

Hi @NSchrading, how are you? I'm just sending a quick reply to help out. We can delve deeper if this doesn't resolve your issue.

Have you checked out the Guess the Number Machine? example?

It features an internal state machine state (counter) and uses conditionals in a sequence of potential transitions to determine the appropriate next state, which appears similar to your issue.

Regarding running in a loop, our system doesn't have an internal loop, so to achieve continuous operation, you would need to write one yourself, as shown in the snippet you provided.

Hope this helps. Let me know!

[NSchrading]

Thanks for the additional information @fgmacedo. I hadn't seen the guess the number machine example, but I understand how to use internal state to track things like counters. Applying that approach I was able to achieve the results I wanted with a combination of an internal transition to count cycles and move to the bad state and a different transition guard to move to the good state. For example:

class MyStateMachine(StateMachine)
    check_data = State()
    data_good = State()
    data_bad = State()

    MAX_CYCLE_COUNT = 10
    cycle_count = 0

    cycle = check_data.to(data_good, cond="evaluate_data") | check_data.to.itself(internal=True, on="while_checking")
    do_data_bad = check_data.to(data_bad)

    # must be true for 3 cycles, else it returns False.
    @persistence_cycles(3)
    def evaluate_data(self, model):
        return model.data.get("value") > 10.0

    def while_checking(self):
        if self.cycle_count > self.MAX_CYCLE_COUNT:
            self.do_data_bad()
        self.cycle_count += 1
    
# main logic
# separate thread is updating model.data on a cycle rate
while True:
    my_statemachine.cycle()
    time.sleep(cycle_rate)

A problem I found with this approach is that you need to be careful about the order of the transitions. If you put the internal transition before the transition with the evaluate_data guard, then the state machine logic always prefers the internal transition, never evaluating the evaluate_data guard.

A couple additional questions:

Would you consider this a bug?
Do you have a recommendation to achieve the same thing without the possibility of making a mistake with ordering?
Is it ok to place state transitions inside of state events (e.g. the self.do_data_bad() call inside of while_checking)?

[fgmacedo]

@NSchrading trying to understand what is the problem. Can you share the code of persistence_cycles?

[fgmacedo]

A problem I found with this approach is that you need to be careful about the order of the transitions. If you put the internal transition before the transition with the evaluate_data guard, then the state machine logic always prefers the internal transition, never evaluating the evaluate_data guard
Would you consider this a bug?

No. Consider looking at this feature as applying a variation of the "chain of responsibility" design pattern.

This behavior is intentional and by design, at the cost of being aware that the order matters. The feature allows for multiple transitions on the same event, with each transition checked in the order they are declared. The first transition that meets the conditions is executed. If none of the transitions meet the conditions, the state machine either raises an exception (if allow_event_without_transition=False) or does nothing.

Do you have a recommendation to achieve the same thing without the possibility of making a mistake with ordering?

I don't see any alternative for now, without leaking the "conditionals" for outside the state machine, which I think will result in a worse design.

I believe we can also enhance our documentation on this topic to provide clearer guidance.

[fgmacedo]

Is it ok to place state transitions inside of state events (e.g. the self.do_data_bad() call inside of while_checking)?

Yes, it's ok. Just be aware of the "RTC" model:

In a run-to-completion (RTC) processing model (default), the state machine executes each event to completion before processing the next event. This means that the state machine completes all the actions associated with an event before moving on to the next event. This guarantees that the system is always in a consistent state.

If the machine is in rtc mode, the event is put on a queue.

This means that even if you are calling the event trigger while inside another transition, the event trigger call will put the trigger on a queue and only execute the trigger after the current transition.

[fgmacedo]

Example using chained transitions with conditionals

Another way to model your state machine is to fully embrace the conditionals. IMO this is the best approach.

>>> from time import sleep
>>> from statemachine import StateMachine, State

>>> class Model:
...     def __init__(self, data: dict):
...         self.data = data

>>> class DataCheckerMachine(StateMachine):
...     check_data = State(initial=True)
...     data_good = State(final=True)
...     data_bad = State(final=True)
...
...     MAX_CYCLE_COUNT = 10
...     cycle_count = 0
...
...     cycle = (
...         check_data.to(data_good, cond="data_looks_good")
...         | check_data.to(data_bad, cond="max_cycle_reached")
...         | check_data.to.itself(internal=True)
...     )
...
...     def data_looks_good(self):
...         return self.model.data.get("value") > 10.0
...
...     def max_cycle_reached(self):
...         return self.cycle_count > self.MAX_CYCLE_COUNT
...
...     def after_cycle(self, event: str, source: State, target: State):
...         print(f'Running {event} {self.cycle_count} from {source!s} to {target!s}.')
...         self.cycle_count += 1
...

Run until we reach the max cycle without success:

>>> data = {"value": 1}
>>> sm1 = DataCheckerMachine(Model(data))
>>> cycle_rate = 0.1
>>> while not sm1.current_state.final:
...     sm1.cycle()
...     sleep(cycle_rate)
Running cycle 0 from Check data to Check data.
Running cycle 1 from Check data to Check data.
Running cycle 2 from Check data to Check data.
Running cycle 3 from Check data to Check data.
Running cycle 4 from Check data to Check data.
Running cycle 5 from Check data to Check data.
Running cycle 6 from Check data to Check data.
Running cycle 7 from Check data to Check data.
Running cycle 8 from Check data to Check data.
Running cycle 9 from Check data to Check data.
Running cycle 10 from Check data to Check data.
Running cycle 11 from Check data to Data bad.

Run simulating that the data turns good on the 5th iteration:

>>> data = {"value": 1}
>>> sm2 = DataCheckerMachine(Model(data))
>>> cycle_rate = 0.1
>>> while not sm2.current_state.final:
...     sm2.cycle()
...     if sm2.cycle_count == 5:
...         print("Now data looks good!")
...         data["value"] = 20  
...     sleep(cycle_rate)
Running cycle 0 from Check data to Check data.
Running cycle 1 from Check data to Check data.
Running cycle 2 from Check data to Check data.
Running cycle 3 from Check data to Check data.
Running cycle 4 from Check data to Check data.
Now data looks good!
Running cycle 5 from Check data to Data good.

[fgmacedo]

Hi @NSchrading , since this has been idle for a few weeks and we already have examples, I think it's safe to close the issue. Feel free to reopen or start another one.

Best!