Scalable design of a system with multiple state machines, where valid state transitions depend on the state of other machines

Question

This is a pretty huge question, however, I would appreciate it if you could just review the design and not my implementations of it. The Implementation and test sections could be ignored, they are only there to aid the description of the design, in case I didn't describe it very well. This makes this question have a smaller scope.

Problem

I have a system with multiple pieces of equipment that can be in many states (E.g turned on, open, in zone 1, etc). The total number of possible states of the entire system is very large, as there are many systems. I need to design some software to restrict the number of possible states into a subset that has been deemed desirable.

For the sake of this question, I will reduce the complexity of this system so that it only contains two pieces of equipment that each only have two states, "On" and "Off".

The total number of this states this system can be is therefore 4:

#| item 1 | item 2 |
#|   On   |   On   |
#|   On   |   Off  |
#|   Off  |   On   |
#|   Off  |   Off  |

For this example, let's say that the states that are deemed desirable are the ones where at most only 1 item is on at a time. This reduces the number of states down to 3 and the state machine is.

#                     ,----------------------------,
#                     v                            |
#    ,----------[BothOffState]--------,            |
#    | turnOn1()                      | turnOn2()  |
#    v                                v            |
# [item1_OnState ]             [item2_OnState]     |
#        | turnOff1()                | turnOff2()  |
#        `---------------------------'-------------'
#

Approach 1

Create the state machine for the whole system, as shown above. The state machine would contain a state object that represents a valid state that my system can have. The state object would have functions to transition into another valid state that is possible to reach from this current state. The state objects would only have functions to transition to states that it has a valid transition to, and every state that I create would represent a valid state.

Implementation:

class IState(metaclass=ABCMeta):
     def __init__(self, fsm):
        print("system : " + fsm.currentState.__class__.__name__ + " -> " + self.__class__.__name__)
        self._fsm = fsm

class BothOffState(IState):
    def __init__(self, fsm):
        super().__init__(fsm)

    def turnOn1(self):
        self._fsm.currentState = item1_OnState(self._fsm)

    def turnOn2(self):
        self._fsm.currentState = item2_OnState(self._fsm)

class item1_OnState(IState):
    def __init__(self, fsm):
        super().__init__(fsm)

    def turnOff1(self):
        self._fsm.currentState = BothOffState(self._fsm)

class item2_OnState(IState):
    def __init__(self, fsm):
        super().__init__(fsm)

    def turnOff2(self):
        self._fsm.currentState = BothOffState(self._fsm)

class FSM:
    currentState = None
    def __init__(self):
        self.currentState = BothOffState(self)

Test:

if __name__ == "__main__":

    system = FSM()
    print("<turning on 1>")
    system.currentState.turnOn1()
    #system.currentState.turnOn2() AttributeError because this state transition doesn't exist
    print("<turning off 1>")
    system.currentState.turnOff1()
    print("<turning on 2>")
    system.currentState.turnOn2()

    #Output:
    #
    # system : NoneType -> BothOffState
    # <turning on 1>
    # system : BothOffState -> item1_OnState
    # <turning off 1>
    # system : item1_OnState -> BothOffState
    # <turning on 2>
    # system : BothOffState -> item2_OnState

Problem with this approach

This seems fine but it is not very scalable. If there are 20 items, and each has an average of 5 states, this would mean creating 3.2 million state objects to represent all of the possible states of the whole system. Even if half of them are considered undesirable and so are not created, this is still too many to realistically implement.

Approach 2: Scalable design of a system with multiple state machines, where valid state transitions depend on the state of other machines:

Instead of using 1 mega state-machine for the whole system, instead, create smaller state machines for each item that can interact with each other. Instead of states directly transitioning into each other, they will go into an intermediate state where they will evaluate if it is a valid state transition within the context of the wider system. Failure will result in it returning to the state it entered from, and success would move to the desired state

The state machines would now look like:

#      item1 state machine                   item2 state machine
#
#        [OffState] <--------,                  [OffState] <--------,
#            | turnOn()      |                      | turnOn()      |
#            v         eval()|                      v         eval()| 
#  [EvaluateCanTurnOnState]->|            [EvaluateCanTurnOnState]->|
#            | eval()        |                      | eval()        |
#            v               |                      v               |
#        [OnState]           |                  [OnState]           |
#            | turnOff()     |                      | turnOff()     |
#            '---------------'                      '---------------'
# State machines are linked, as the input to one of the state transitions `eval()` is the other state machine

In this example, the 2 systems have identical states, however, the idea still works with heterogeneous systems.

When the FSM's are created they will be given a reference to any other state machine that they have a dependency on. The intermediate Eval states will use this reference to decide if the next state should be the desired state or if it should go back to the previous state.

Implementation:

class IState(metaclass=ABCMeta):
     def __init__(self, fsm):
        print(fsm.name + " : " + fsm.currentState.__class__.__name__ + " -> " + self.__class__.__name__)
        self._fsm = fsm

class OffState(IState):
    def __init__(self, fsm):
        super().__init__(fsm)

    def turnOn(self):
        self._fsm.currentState = EvaluateCanTurnOnState(self._fsm)
        self._fsm.currentState.eval(self._fsm.otherStateMachine)

class EvaluateCanTurnOnState(IState):
    def __init__(self, fsm):
        super().__init__(fsm)

    def eval(self, otherFsm):
        if otherFsm.currentState.__class__.__name__ == "OffState":
            self._fsm.currentState = OnState(self._fsm)
        else:
            self._fsm.currentState = OffState(self._fsm)

class OnState(IState):
    def __init__(self, fsm):
        super().__init__(fsm)

    def turnOff(self):
        self._fsm.currentState = OffState(self._fsm)

class FSM:
    currentState = None
    otherStateMachine = None

    def __init__(self, name):
        self.name = name
        self.currentState = OffState(self)

    def setOther(self, otherStateMachine):
        self.otherStateMachine = otherStateMachine

Test:

if __name__ == "__main__":

    fsm1 = FSM("item1")
    fsm2 = FSM("item2")
    fsm1.setOther(fsm2)
    fsm2.setOther(fsm1)

    fsm1.currentState.turnOn()
    fsm2.currentState.turnOn()
    fsm1.currentState.turnOff()
    fsm2.currentState.turnOn()

    #Output:
    #
    # item1 : NoneType -> OffState
    # item2 : NoneType -> OffState
    # item1 : OffState -> EvaluateCanTurnOnState
    # item1 : EvaluateCanTurnOnState -> OnState
    # item2 : OffState -> EvaluateCanTurnOnState
    # item2 : EvaluateCanTurnOnState -> OffState
    # item1 : OnState -> OffState
    # item2 : OffState -> EvaluateCanTurnOnState
    # item2 : EvaluateCanTurnOnState -> OnState

Discussion

The second approach seems more scalable, as the states of the whole system to not have to be explicitly defined. The dependencies between each state machine are captured during construction of the object, and if the number of dependent machines grows, this could be tidied up with a builder object.

However, I have never seen this design before (because I don't really know where to look). I do not know if the complexity of this will actually become unmaintainable or prone to bugs.

Surely this is a common problem and has already been solved? What is the standard design to use in a situation like this? If there isn't a standard design pattern, do you think the design I have suggested is good design?

Answer 1

This doesn't have to be complicated. Definitely avoid Approach 1 - having a dedicated class for state combinations is not a good idea. Follow vaguely Approach 2, but

• Do not have a class OffState, nor a class for any specific state
• Do not have a dedicated class for EvaluateCanTurnOnState
• Track states with enumeration members
• Have an equipment superclass, where each subclass implements a state transition predicate

Example:

from enum import Enum
from typing import Type, List


class Equipment:
    States: Type[Enum]

    def __init__(self):
        self.state: Equipment.States = None

    def change(self, new_state: 'Equipment.States'):
        if not self.can_change(new_state):
            raise ValueError(
                f'{type(self).__name__} cannot change '
                f'from {self.state} to {new_state}'
            )
        self.state = new_state

    def can_change(self, new_state: 'Equipment.States') -> bool:
        raise NotImplementedError()


class ExclusiveEq(Equipment):
    class States(Enum):
        OFF = 0
        ON = 1

    def __init__(self, name: str):
        super().__init__()
        self.name = name

    def __str__(self):
        return self.name

    def can_change(self, new_state: 'ExclusiveEq.States') -> bool:
        if new_state != self.States.ON:
            return True
        return all(
            not isinstance(r, ExclusiveEq)
            or r is self
            or r.state != self.States.ON
            for r in registry
        )


registry: List[Equipment] = [
    ExclusiveEq('blender'),
    ExclusiveEq('coffeemaker'),
    ExclusiveEq('ion cannon'),
]

registry[0].change(ExclusiveEq.States.ON)
registry[0].change(ExclusiveEq.States.OFF)
registry[1].change(ExclusiveEq.States.ON)
registry[1].change(ExclusiveEq.States.OFF)
registry[2].change(ExclusiveEq.States.ON)

try:
    registry[0].change(ExclusiveEq.States.ON)
    raise AssertionError('This should have failed')
except ValueError:
    pass

Answer 2

Approach 3

As a third approach, you might consider a hierarchical state machine. The system as a whole could have a state machine with states such as starting_up, running, shutting_down. Then different kinds of components or groups of components could have state machines whose transition table depend on the system state (or there are different tables for different states). Likewise, the system can change state based on the states of the sub state machines (e.g., when they are all running, then the system can transition to the running state). Further, each component has it's own state machine that depends on the state of it's parent state machine. The state machines at any given level are more or less independent of each other. They change state based on the inputs, but unrecognized inputs get passed up to their parent state machine.

Consider a basic HVAC system. The system may have states: OFF, COOL, HEAT. A thermostat can send a signal that the temperature is above or below the temperature set point. The A/C component has a state machine that response to the thermostat signal if the system state machine is the the COOL state. It can also respond to internal signals such as compressor temperature, or refrigerant suction pressure, etc. Similarly, the furnace can respond if the system is in the HEAT state, and can also respond to internal signals such as low pilot light temperature or high flue temperature.