Multiple instances of the statemachine are not properly separated

[Rosi2143]

• Python State Machine version:
• Python version: Python 3.8.10
• Operating System: 1.0.2 (Requirement already satisfied: python-statemachine in c:\users\rasen\appdata\local\packages\pythonsoftwarefoundation.python.3.8_qbz5n2kfra8p0\localcache\local-packages\python38\site-packages (1.0.2))

Description

I created a statemachine and instantiated it multiple times. Scenario is the control of my 14 thermostats for my home automation.

What I Did

Here is the reduced "minimal" setup to show the error.
The failing test is

    test_state(sm_list[2], "Three")

the statemachine remains in state "Two"

"""Example of python module statemachine: https://pypi.org/project/python-statemachine/"""

import sys
import os
import logging
import inspect


from statemachine import State, StateMachine

log = logging.getLogger(__name__)
log.setLevel(logging.DEBUG)

# https://docs.python.org/3/howto/logging-cookbook.html
# create console handler with a higher log level
ch = logging.StreamHandler()
ch.setLevel(logging.DEBUG)
# create formatter and add it to the handlers
formatter = logging.Formatter(
    '%(asctime)s - %(name)s - %(levelname)s - %(message)s')
ch.setFormatter(formatter)
# add the handlers to the logger
log.addHandler(ch)


class TestStateMachine(StateMachine):

    # States
    st_1 = State("One", initial=True)
    st_2 = State("Two")
    st_3 = State("Three")

    one = False
    two = True
    three = True

    def __init__(self, name="unnamed"):
        # variables
        self.sm_name = name
        super(TestStateMachine, self).__init__()

    # Transitions
    tr_change = (st_1.to(st_1, cond="cond_one_is_active") |
                 st_1.to(st_2, cond="cond_two_is_active") |
                 st_1.to(st_3, cond="cond_three_is_active") |
                 st_2.to(st_1, cond="cond_one_is_active") |
                 st_2.to(st_2, cond="cond_two_is_active") |
                 st_2.to(st_3, cond="cond_three_is_active"))
    # Conditions

    def cond_one_is_active(self):
        return self.one

    def cond_two_is_active(self):
        return self.two

    def cond_three_is_active(self):
        return self.three


def test_state(state_machine, state):
    print(state_machine.current_state.name)
    assert state_machine.current_state.name == state


def test_single_sm():

    sm = TestStateMachine()
    test_state(sm, "One")
    sm.send("tr_change")
    test_state(sm, "Two")


def test_multiple_sm():

    sm_list = {}
    for index in [1, 2, 3]:
        sm_list[index] = TestStateMachine(str(index))
        print(sm_list[index].sm_name)
        test_state(sm_list[index], "One")

        sm_list[index].send("tr_change")
        test_state(sm_list[index], "Two")

    sm_list[2].two = False
    sm_list[2].send("tr_change")
    test_state(sm_list[2], "Three")

test_single_sm()
test_multiple_sm()

Output:

python \\openhabian3\openHAB-conf\automation\test\statemachine_test.py
One
Two
1
One
Two
2
One
Two
3
One
Two
One
Two
3
One
Two
Two
Traceback (most recent call last):
  File "\\openhabian3\openHAB-conf\automation\test\statemachine_test.py", line 91, in <module>
    test_multiple_sm()
  File "\\openhabian3\openHAB-conf\automation\test\statemachine_test.py", line 87, in test_multiple_sm
    test_state(sm_list[2], "Three")
  File "\\openhabian3\openHAB-conf\automation\test\statemachine_test.py", line 63, in test_state
    assert state_machine.current_state.name == state
AssertionError

Hint:

when I change

    sm_list[2].two = False

to change the LAST StateMachine in my list it works :-(

    sm_list[3].two = False

I can reproduce the same error with index sm_list[1] the same way,

Assumed reason:

When I debug and set the breakpoints into the functions e.g. cond_one_is_active the instance of TestStateMachine is always "3".

[fgmacedo]

Hi @Rosi2143 , how are you?

Nice use case, let me know when you have something working, I'll be glad to know. Maybe you can post somewhere so I can learn about it. I have an ESP32 and like to learn more about home automation.

I've tried your example, and actually, I've found two minor "gotchas" on the tests, so I think everything is fine with the library regarding this "isolation" concern.

Machine extended state should be initialized at __init__

The first was to put the "extended state" inside the "instance" initialization, that in Python is done on the __init__ function, so when you set one = True at the class level, you're sharing this value across all instances, when under the __init__, the value is "per-instance".

    def __init__(self, name="unnamed"):
        # variables
        self.sm_name = name
        self.one = False
        self.two = True
        self.three = True

Python lists indexes start from zero

This is actually the main issue, when trying to change the value, there was a mistake on the indexes, so

- for index in [1, 2, 3]:
+ for index in [0, 1, 2]:

This is the test case fixed

import sys
import os
import logging
import inspect


from statemachine import State, StateMachine

log = logging.getLogger(__name__)
log.setLevel(logging.DEBUG)

# https://docs.python.org/3/howto/logging-cookbook.html
# create console handler with a higher log level
ch = logging.StreamHandler()
ch.setLevel(logging.DEBUG)
# create formatter and add it to the handlers
formatter = logging.Formatter(
    '%(asctime)s - %(name)s - %(levelname)s - %(message)s')
ch.setFormatter(formatter)
# add the handlers to the logger
log.addHandler(ch)


class TestStateMachine(StateMachine):

    # States
    st_1 = State("One", initial=True)
    st_2 = State("Two")
    st_3 = State("Three")

    def __init__(self, name="unnamed"):
        # variables
        self.sm_name = name
        self.one = False
        self.two = True
        self.three = True
        super(TestStateMachine, self).__init__()

    # Transitions
    tr_change = (st_1.to(st_1, cond="cond_one_is_active") |
                 st_1.to(st_2, cond="cond_two_is_active") |
                 st_1.to(st_3, cond="cond_three_is_active") |
                 st_2.to(st_1, cond="cond_one_is_active") |
                 st_2.to(st_2, cond="cond_two_is_active") |
                 st_2.to(st_3, cond="cond_three_is_active"))
    # Conditions

    def cond_one_is_active(self):
        return self.one

    def cond_two_is_active(self):
        return self.two

    def cond_three_is_active(self):
        return self.three


def test_state(state_machine, state):
    print(state_machine.current_state.name)
    assert state_machine.current_state.name == state


def test_single_sm():

    sm = TestStateMachine()
    test_state(sm, "One")
    sm.send("tr_change")
    test_state(sm, "Two")


def test_multiple_sm():

    sm_list = {}
    for index in [0, 1, 2]:
        sm_list[index] = TestStateMachine(str(index))
        print(sm_list[index].sm_name)
        test_state(sm_list[index], "One")

        sm_list[index].send("tr_change")
        test_state(sm_list[index], "Two")

    sm_list[2].two = False
    assert [(sm.current_state.id, sm.two) for sm in sm_list.values()] == [
        ('st_2', True), ('st_2', True), ('st_2', False)
    ]
    sm_list[2].send("tr_change")
    test_state(sm_list[2], "Three")

test_single_sm()
test_multiple_sm()

Please let me know if you have any other issue.

[fgmacedo]

Ow... extra hint if it helps...

Given that I've implemented a "dynamic dispatch", the library also works directly with attributes and properties and can read from them directly as cond parameters. So your code can be simplified to this (no need to write custom conditions if you already have booleans flags):

from statemachine import State
from statemachine import StateMachine


class TestStateMachine(StateMachine):

    # States
    st_1 = State("One", initial=True)
    st_2 = State("Two")
    st_3 = State("Three")

    # Transitions
    tr_change = (
        st_1.to(st_1, cond="one")
        | st_1.to(st_2, cond="two")
        | st_1.to(st_3, cond="three")
        | st_2.to(st_1, cond="one")
        | st_2.to(st_2, cond="two")
        | st_2.to(st_3, cond="three")
    )

    def __init__(self, name="unnamed"):
        # variables
        self.sm_name = name
        self.one = False
        self.two = True
        self.three = True
        super().__init__()

[fgmacedo]

Closing for now. If you have any other issue please let me know.

[KellyP5]

I am not sure this was solved, if I create multiple instances of a state machine -- the newest statemachines class variables will serve as the single point of truth for all state machines.

from statemachine import State
from statemachine import StateMachine
class TestStateMachine(StateMachine):
    # States
    st_1 = State("One", initial=True)
    st_2 = State("Two")
    st_3 = State("Three")
    # Transitions
    tr_change = (
        st_1.to(st_2, cond="two")
        | st_2.to(st_3, cond="three") 
        | st_3.to(st_1, cond="one")
    )
    def __init__(self, name="unnamed"):
        # variables
        self.sm_name = name
        self.one = False
        self.two = True
        self.three = False
        super().__init__()
        
s1 = TestStateMachine()
s2 = TestStateMachine()
s1.current_state
State('One', id='st_1', value='st_1', initial=True, final=False)
s2.current_state
State('One', id='st_1', value='st_1', initial=True, final=False)
s1.tr_change()
s1.current_state
State('Two', id='st_2', value='st_2', initial=False, final=False)
s2.current_state
State('One', id='st_1', value='st_1', initial=True, final=False)
s2.two = False
s2
TestStateMachine(model=Model(state=st_1), state_field='state', current_state='st_1')
s2.tr_change()

statemachine.exceptions.TransitionNotAllowed: Can't tr_change when in One.

s1.three = True
s1.three
True
s2.three
False
s1.tr_change() # this shouldn't throw an error because s1.three is now true

statemachine.exceptions.TransitionNotAllowed: Can't tr_change when in Two.

s2.three = True # When I change the variable on s2, it then allows me to transition the state of s1
s1.tr_change()
s1.current_state 
State('Two', id='st_2', value='st_2', initial=False, final=False)

[Rosi2143]

Hi @fgmacedo,

sorry for not replying earlier but I was on vacation.

I've seen that you already merged a patch. Thanks, I will try it out today.

Also I really like the idea of using boolean attributes/properties directly.

I will raise a MR again with some extention of the documentation for this.

Great to see that you are so active and respond very quickly.

[fgmacedo]

Nice news @Rosi2143 , I hope that you enjoyed your deserved free time :)

I've just released 1.0.3 now. Your feedback will be handy.

Best regards!

[Rosi2143]

I just re-run my test with your latest version (6dbc55c) and it works like a charm.

Thanks.

[fgmacedo]

Another "feature" that's not well documented on the callbacks is that you can either pass a string or method, or even a list of strings or methods. This is valid for all callbacks:

• State.enter
• State.exit
• Transition.validators
• Transition.cond
• Transition.unless
• Transition.on
• Transition.before
• Transition.after

def should_have_name(machine):
    "Any method will work"
    return machine.sm_name != "unnamed"


class TestStateMachine(StateMachine):
    # States
    st_1 = State("One", initial=True)
    st_2 = State("Two")
    st_3 = State("Three")
    # Transitions
    tr_change = (
        st_1.to(st_2, cond=["two", should_have_name])
        | st_2.to(st_3, cond=["three", "prepared"]) 
        | st_3.to(st_1, cond="one")
    )
    def __init__(self, name="unnamed"):
        # variables
        self.sm_name = name
        self.one = False
        self.two = True
        self.three = False
        self.prepared = False
        super().__init__()