poros
3/25/2017 - 5:40 PM

Basic state machine implementation in Python using namedtuples as states and functions as transitions

Basic state machine implementation in Python using namedtuples as states and functions as transitions

from collections import namedtuple
import re
import sys


SIGNALFX_API = 'https://app.signalfx.com/#/'
RES_TYPE_RE= re.compile(r"^(?P<type>\S+?)\..*:$")


# STATES
NoResource = namedtuple('NoResource', 'output')
EmptyResource = namedtuple('EmptyResource', '')
NewResource = namedtuple('NewResource', ('type',))
NamedResource = namedtuple('NamedResource', ('type', 'name'))
IdResource = namedtuple('IdResouce', ('type', 'id'))
CompleteResource = namedtuple('CompleteResource', ('type', 'name', 'id'))
End = namedtuple('End', '')


# ACTIONS
def show_no_state(state):
    print(state.output)


def show(state):
    print(state.type)
    print(state.name)
    print(SIGNALFX_API + state.type + '/' + state.id + '/edit\n')


# TRANSITIONS
def itself(state, input):
    return state


def end(state, input):
    return End()


def empty(state, input):
    return EmptyResource()


def no_state(state, input):
    if 'No state' in input:
        return NoResource(input)


def new_resource(state, input):
    match = RES_TYPE_RE.match(input)
    if match:
        return NewResource(match.group('type'))


def name_resource(state, input):
    input = input.strip()
    if input.startswith('name ='):
        name = input.split('name = ', 1)[1].strip()
        if isinstance(state, NewResource):
            return NamedResource(state.type, name)
        else:
            return CompleteResource(state.type, name, state.id)


def id_resource(state, input):
    input = input.strip()
    if input.startswith('id ='):
        res_id = input.split('id =', 1)[1].strip()
        if isinstance(state, NewResource):
            return IdResource(state.type, res_id)
        else:
            return CompleteResource(state.type, state.name, res_id)


# STATE MACHINE
TRANSITIONS = {
    EmptyResource: (no_state, new_resource, itself),
    NewResource: (name_resource, id_resource, itself),
    NamedResource: (id_resource, itself),
    IdResource: (name_resource, itself),
    CompleteResource: (new_resource, empty),
    NoResource: (end,),
    End: (itself,),
}


ACTIONS = {
    NoResource: show_no_state,
    CompleteResource: show,
}


def parse(state, line):
    for tr in TRANSITIONS[type(state)]:
        new_state = tr(state, line)
        if new_state is not None:
            if type(new_state) in ACTIONS:
                ACTIONS[type(new_state)](new_state)
            return new_state
    raise ValueError("Unexpected state line: {0}\nState: {1}".format(line, state))


def parse_resource_state(resource_state):
    reduce(parse, resource_state, EmptyResource())


if __name__ == "__main__":
    parse_resource_state(sys.stdin)