Graphical representation/ How to execute code while in a State

[hoderlump]

Hi everone.

Thanks for this cool project! I really like the versability and the fact that it is very robust.
There is one thing that bothers me a little. Maybe my understanding is not correct here but I try to lllistrate it.

I have the following simplyfied state machine:

class mySuperStatemachine(StateMachine):
   
    # initialisation states:
    my_init_State = State("my_init_State", initial=True)
    my_worker_State = State("my_worker_State", final=True)
	# failure states:
    sm_configuration_invalid_State = State("sm_configuration_invalid_State", final=True)
	
	# transitions:
    run = (
            my_init_State.to(my_worker_State, cond="valid_sm_configuration")
            | my_init_State.to(sm_configuration_invalid_State, unless="valid_sm_configuration")
			
    def __init__(self):
        # conditions and guards:
        self.valid_sm_configuration = False
		

    def on_enter_my_init_State(self):
		# do stuff that takes a lot of time...
        self.valid_sm_configuration = True

This is the main loop:

sm = mySuperStatemachine()
sm.graph = DotGraphMachine(sm)
sm.add_observer(LogObserver("mySuperStatemachine", sm))


while not sm.current_state.final:
    sm.run()

This is the observer; It prints the current state and creates a svg of the state machine and marks the current state:

class LogObserver(object):
    def __init__(self, sm):
        self.graph = sm.graph
        self.sm = sm

    def on_run(self, event, source, target):
        print(f"StateMachine: {source.id}--({event})-->{target.id}")
        dot = self.sm.graph()
        dot.write_svg("graphviz.svg")

It runs quite well!
But the graphical representation is always late. As all the graphical representation is being created before the state is entered (during the run transition), the previous state is always marked instead of the target state.

I tried to change the things by executing all my worker code in on_exit_ functions but this does not work well for me: I set validators inside the code (self.valid_sm_configuration = False) and that is too late in the on_exit_ function.

Is there any way, i can execute code during the state, so after entering and before exiting?
Maybe there is another way: I could to create the graph in the observer when the state is entered. But then i have 2 on_enter_ functions per state and i do not know which of those is executed first...

Thx a lot!

[fgmacedo]

Hi @hoderlump , how are you?

Nice example! The hook on_<event_name> is executed just before the transition occurs. So seems to me that you want to use the after_<event_name> hook. See https://python-statemachine.readthedocs.io/en/latest/actions.html#transition-actions.

This should do the trick... I've updated a few things on your example to take advantage of the latest features (like state names are now optional).

class LogObserver(object):
    def __init__(self):
        self.counter = 0

    def after_transition(self, machine, event, source, target):
        self.counter += 1
        print(f"StateMachine: {source.id}--({event})-->{target.id}")
        machine._graph().write_svg(f"{machine.name}_{self.counter:0>5}.svg")


class mySuperStatemachine(StateMachine):

    # initialization states:
    my_init_State = State(initial=True)
    my_worker_State = State(final=True)

    # failure states:
    sm_configuration_invalid_State = State(final=True)

    # transitions:
    run = (
        my_init_State.to(my_worker_State, cond="valid_sm_configuration")
        | my_init_State.to(sm_configuration_invalid_State, unless="valid_sm_configuration")
    )

    def __init__(self):
        # conditions and guards:
        self.valid_sm_configuration = False

    def run_until_done(self):
        while not self.current_state.final:
            self.run()

    def on_enter_my_init_State(self):
		# do stuff that takes a lot of time...
        self.valid_sm_configuration = True

The main loop:

sm = mySuperStatemachine()
sm.add_observer(LogObserver())

sm.run_until_done()

[hoderlump]

Hi Fernando,

thanks for you quick help! :)
I implemented all of your changes and it looks promising and even simpler. I like the look and feel of 2.0.
The problem is: The order of execution is still not like I'd like it. I try to illustrate it by a litte modification of the example code:

class LogObserver(object):
    def __init__(self):
        self.counter = 0

    def after_transition(self, machine, event, source, target):
        self.counter += 1
        print(f"StateMachine: {source.id}--({event})-->{target.id}")
        machine._graph().write_svg(f"{machine.name}_{self.counter:0>5}.svg")


class mySuperStatemachine(StateMachine):

    # initialization states:
    not_being_pregnant = State(initial=True)
    being_pregnant = State()
    giving_birth = State(final=True) # Not the final state in reality. For illustration purposes only. Everyone life long!! :)

    # transitions:
    run = (
        not_being_pregnant.to(being_pregnant)
        | being_pregnant.to(giving_birth, cond="nine_months_pass")
    )

    def __init__(self):
        # conditions and guards:
        self.nine_months_pass = False

    def run_until_done(self):
        while not self.current_state.final:
            self.run()

    def on_enter_being_pregnant(self):
	time.sleep(60*60*24*30*9)
        self.self.nine_months_pass = True

It's obvious that the "being_pregnant" state is active for 9 months in the example above.
The graph and the print from the Observer now always show the old state "not_being_pregnant" as active state for nine months.
I assume this is because the on_enter_being_pregnant function is executed before the after_transition function.

I could solve it now by changing the after_transition function in the Observer to on_enter_state. Now the active states are like I'd like them to be. But thus I have 2 on_enter_ functions for every state and it's not defined in which order they are executed. So, I don't know if this is a good solution to go.

[fgmacedo]

Hi @hoderlump , I'm afraid that I don't get the expected behavior. Can you elaborate on the example as an automated test case that fails?

def test_should_x():
    sm = mySuperStatemachine()
    assert ...

[hoderlump]

Hi @fgmacedo . Sorry for the not being explicit. I could solve this and I think it was a misunderstanding what caused the problem.
If anyone is interested or has the same problem:

I try with pseudocode:

def test_should_x():
    sm = mySuperStatemachine()
    sm.run()
    # Now the state machine left the init state and executes the function `on_enter_being_pregnant(self)`.
    # As there is a 9 month wait cycle in that function, the function` after_transition(self, machine, event, source, target)` will be executed much later....
    # Thus mySuperStatemachine_00001.svg where the being_pregnant state is marked active is created 9 months later
    # The mySuperStatemachine_00000.svg still shows not_being_pregnant as active state.
    # The assert will have to be started in a parallel thread as it also will only be executed after 9 months. After 9 months it will be passed though. It will fail before:
    assert active_state(newest_svg) == sm.active_state

As written above, I could solve this by changing the after_transition(self, machine, event, source, target) function in the Observer to on_enter_state(self, machine, event, source, target). I recognised that on_enter_state(self, machine, event, source, target) is executed before the on_enter_being_pregnant(self) function, as defined here and this is actual valid code:
on_enter_state()
is exceuted before
on_enter_<state.id>()

Sorry for the confusion and wasting your time here, I should have rtfm better. I think, this can be closed.
I really appreciate your work!!
Greetings

[fgmacedo]

Hi @hoderlump , don't worry. Glad It worked and you have figured out a way to accomplish your task!

In fact, you gave me an idea to improve the docs, the ordering list could actually be a table, informing if the current state was changed when the action is called.

Also, the possibility of injecting the machine parameter is missing from the list at dynamic dispatch. I don't think that this section has a good name 😅 .

Best!