from typing import List from transitions import Machine class FSMBase: states, transitions = [], [] def __init__(self, initial: str = None, accepted_states: List[str] = None): self.message_handlers = [] self.accepted_states = accepted_states or [self.states[-1]] self.machine = Machine(model=self, states=self.states, transitions=self.transitions, initial=initial or self.states[0], send_event=True, ignore_invalid_triggers=True, after_state_change='forward_message') def forward_message(self, event_data): message = event_data.kwargs.get('message') for msghandler in self.message_handlers: msghandler(message) def subscribe_message_handler(self, msghandler): self.message_handlers.append(msghandler) def unsubscribe_message_handler(self, msghandler): self.message_handlers.remove(msghandler) def is_solved(self): return self.state in self.accepted_states