Article about Finite State Machine - Step 3 - The State Machine library

_FSM_LIBRARY_IMPL.md

Article about Finite State Machine - Step 3 - The State Machine library

• https://helpdev.com.br/2020/11/02/controlando-transacoes-de-status-com-finite-state-machine-parte-1/
• https://helpdev.com.br/2020/11/09/controlando-transacoes-de-status-com-finite-state-machine-parte-2/

actions.py

import random


def hidden_to_visible_action(pokemon):
    print(f'Trying to find {pokemon.name}...')
    value = random.randint(0, pokemon.level)
    if value > 3:
        print('Uh oh! Pokemon 404!')
        return False

    print(f'Wild {pokemon.name} appeared!')
    return True


def injured_to_attack_action(pokemon):
    # TODO - Some action to extract the pokemon life
    return True

domain.py

from enum import Enum, auto


class States(Enum):
    HIDDEN = auto()
    VISIBLE = auto()
    INJURED = auto()
    CAPTURED = auto()
    DEAD = auto()


class Events(Enum):
    FIND = auto()
    OBSERVE = auto()
    ATTACK = auto()
    CATCH = auto()

    @staticmethod
    def get_by_name(name):
        return Events[name.upper()]


class Pokemon:
    def __init__(self, poke_id, name, level):
        self.__poke_id = poke_id
        self.__name = name
        if level > 5:
            raise ValueError('Pokemon level cant be more than 5')
        self.__level = level

    @property
    def poke_id(self):
        return self.__poke_id

    @property
    def name(self):
        return self.__name

    @property
    def level(self):
        return self.__level

fsm_factory.py

from fsm.fsm_transaction import StateMachineTransaction
from fsm.sate_machine import StateMachine, _StateMachineImpl


class StateMachineFactory:

    def __init__(self, states, final_states):
        self.__states = states
        self.__final_states = final_states
        self.__transactions = {}

    def add_transaction(self, transaction: StateMachineTransaction):
        if transaction.state not in self.__states:
            raise ValueError(f'Transaction state {transaction.state.name} not in {self.__states}')
        elif self.__transactions.get(transaction.state):
            self.__transactions[transaction.state][transaction.event] = transaction
        else:
            self.__transactions[transaction.state] = {transaction.event: transaction}
        return self

    def build_machine(self, machine_id, current_state) -> StateMachine:
        return _StateMachineImpl(machine_id, current_state, self.__final_states, self.__transactions)

fsm_transaction.py

class StateMachineTransaction:
    def __init__(self, state, event, state_target, action_method=None):
        """
        Define a transaction of one state.

        :param state: The State enum
        :param event: The Event enum
        :param state_target: The State enum target
        :param action_method: The method to be executed when this transaction is called.
         The method needed return a Boolean and receive some object was argument
        """
        self.__state = state
        self.__event = event
        self.__target = state_target
        self._action_method = action_method

    @property
    def state(self):
        return self.__state

    @property
    def target(self):
        return self.__target

    @property
    def event(self):
        return self.__event

    @property
    def action_method(self):
        return self._action_method

main.py

from fsm.fsm_factory import StateMachineFactory
from fsm.fsm_transaction import StateMachineTransaction
from pokecatcher.actions import hidden_to_visible_action, injured_to_attack_action
from pokecatcher.domain import States, Events
from pokecatcher.pokecatcher_fsm import PokeCatcherFsm

if __name__ == '__main__':
    fsm_factory = StateMachineFactory(
        states=States, final_states=(States.CAPTURED, States.DEAD)
    ).add_transaction(StateMachineTransaction(
        state=States.HIDDEN, event=Events.FIND, target=States.VISIBLE, action_method=hidden_to_visible_action)
    ).add_transaction(StateMachineTransaction(
        state=States.VISIBLE, event=Events.OBSERVE, target=States.VISIBLE)
    ).add_transaction(StateMachineTransaction(
        state=States.VISIBLE, event=Events.ATTACK, target=States.INJURED)
    ).add_transaction(StateMachineTransaction(
        state=States.INJURED, event=Events.ATTACK, target=States.DEAD, action_method=injured_to_attack_action)
    ).add_transaction(StateMachineTransaction(
        state=States.INJURED, event=Events.CATCH, target=States.CAPTURED))

    PokeCatcherFsm(fsm_factory) \
        .start_main_loop()

pokecatcher_fsm.py

from pokecatcher.domain import States, Events, Pokemon

wild_animals = {
    '1': Pokemon(1, "Bulbasaur", 3),
    '2': Pokemon(2, "Charizard", 5),
    '3': Pokemon(3, "Magikarp", 1)
}


class PokeCatcherFsm:

    def __init__(self, machine_factory):
        self.__machines = {}
        self.__fsm_factory = machine_factory

    def start_main_loop(self):
        print("Input the command: poke_id,action / Like: 1,find:")
        while True:
            data = input().split(",")
            pokemon = wild_animals.get(data[0])
            event = Events.get_by_name(data[1])
            self.execute(pokemon, event)

    def execute(self, pokemon, event):
        machine = self.__get_machine(pokemon)
        accepted_event = machine.send_event(event, pokemon)
        if accepted_event is False:
            print(f'The event {event.name} is not accepted for status {machine.current_state.name}')

    def __get_machine(self, pokemon):
        if not self.__machines.get(pokemon.poke_id):
            self.__machines[pokemon.poke_id] = self.__fsm_factory.build_machine(pokemon.poke_id, States.HIDDEN)
        return self.__machines[pokemon.poke_id]

state_machine.py

from abc import ABC, abstractmethod


class StateMachine(ABC):
    """
    Use the fsm.state_machine_factory.StateMachineFactory to build a new Machine
    """

    @abstractmethod
    def send_event(self, event, message=None):
        pass

    @property
    @abstractmethod
    def machine_id(self):
        pass

    @property
    @abstractmethod
    def current_state(self):
        pass


class _StateMachineImpl(StateMachine):

    def __init__(self, machine_id, initial_state, final_states, transactions):
        self.__machine_id = machine_id
        self.__current_state = initial_state
        self.__final_states = final_states
        self.__transactions = transactions
        self.__ended = self.__is_ended()

    def send_event(self, event, message=None):
        if self.__ended:
            print(f'The machine {self.__machine_id} was ended')
            return True

        transaction = self.__get_transaction_from_event(event)
        if not transaction:
            return False

        if self.__finalize_transaction_change(message, transaction):
            self.__current_state = transaction.target
            self.__ended = self.__is_ended()

        return True

    @staticmethod
    def __finalize_transaction_change(message, transaction):
        return transaction.action_method(message) if transaction.action_method else True

    def __get_transaction_from_event(self, event):
        transaction_group = self.__transactions.get(self.__current_state)
        if not transaction_group or not transaction_group.get(event):
            return None
        return transaction_group[event]

    def __is_ended(self):
        return self.__current_state in self.__final_states

    @property
    def machine_id(self):
        return self.__machine_id

    @property
    def current_state(self):
        return self.__current_state