Skip to content

Instantly share code, notes, and snippets.

@poros
Created March 25, 2017 17:40
Show Gist options
  • Save poros/45e067c0f2e1a3ace85aab1e05424762 to your computer and use it in GitHub Desktop.
Save poros/45e067c0f2e1a3ace85aab1e05424762 to your computer and use it in GitHub Desktop.
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)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment