#!/bin/env python
# -*- utf-8 -*-
"""Boiler Plate code for implementing a state machine according to
state design pattern
"""
class State(object):
"""Base class for a state in a statemachine
No statemachine class is needed. State transitions have to be
defined in self.transitions.
If access to self is needed in transitions they may be defined
using the set_transitions method.
The object is callable and by calling it input is handled and
the next state is returned.
Example usage:
def handle(self, msg):
self._state = self._state(msg)
"""
transitions = [(None, None, None)]
def __init__(self, state_context):
self._context = state_context
self.set_transitions()
def set_transitions(self):
"""subclass may override this if access to self is
needed in transitions
"""
pass
def __call__(self, inp):
new_state = self
state_action_list = [(state, action) for cls, state,
action in self.transitions if
isinstance(inp, cls)]
if 1 == len(state_action_list):
state, action = state_action_list[0]
if action is not None:
try:
action(inp)
new_state = globals except TypeError:
action(self, inp)[state]
new_state = state(self._context)
elif 1 < len(state_action_list):
raise AmbiguousTransitionError(inp, self)
print ("[DEBUG] %s -- %s --> %s" %
(self.__class__.__name__,
inp.__class__.__name__,
new_state.__class__.__name__ ))
return new_state
class AmbiguousTransitionError(Exception):
"""Raised if more than one state transition was found for a
given input
"""
pass
################################################## Example
class MsgA(object):
"""example"""
pass
class MsgB(object):
"""example"""
pass
class MsgC(object):
"""example"""
pass
class StateA(State):
"""example"""
transitions = [(MsgA, "StateB", None)]pass
class StateB(State):
"""example"""
def set_transitions(self):
self.transitions = [(MsgB, "StateC", self._print_msg)]
@staticmethod
def _print_msgprint_msg(inp):
"""example"""
print "Inp: ", inp
class StateC(State):
"""example"""
def set_transitionsprint_context(self, unused_inp):
print "Context: ", self._context
StateA.transitions = [(MsgBMsgA, "StateB"StateB, None),]
StateB.transitions = [(MsgCMsgB, "StateA"StateC, selfStateB._print_contextprint_msg)]
StateC.transitions def= _print_context[(selfMsgB, unused_inpStateB, None):,
"""example"""
print "Context:(MsgC, "StateA, selfStateC._contextprint_context)]
class Context(object):
def __init__(self):
self._state = StateA(self)
def handle(self, msg):
self._state = self._state(msg)
if __name__ == "__main__":
CONTEXT = Context()
CONTEXT.handle(MsgA())
CONTEXT.handle(MsgB())
CONTEXT.handle(MsgC())
CONTEXT.handle(MsgB())
#!/bin/env python
# -*- utf-8 -*-
class State(object):
"""Base class for a state in a statemachine
No statemachine class is needed. State transitions have to be
defined in self.transitions.
If access to self is needed in transitions they may be defined
using the set_transitions method.
The object is callable and by calling it input is handled and
the next state is returned.
Example usage:
def handle(self, msg):
self._state = self._state(msg)
"""
transitions = [(None, None, None)]
def __init__(self, state_context):
self._context = state_context
self.set_transitions()
def set_transitions(self):
"""subclass may override this if access to self is
needed in transitions
"""
pass
def __call__(self, inp):
new_state = self
state_action_list = [(state, action) for cls, state,
action in self.transitions if
isinstance(inp, cls)]
if 1 == len(state_action_list):
state, action = state_action_list[0]
if action is not None:
action(inp)
new_state = globals()[state](self._context)
elif 1 < len(state_action_list):
raise AmbiguousTransitionError(inp, self)
print ("[DEBUG] %s -- %s --> %s" %
(self.__class__.__name__,
inp.__class__.__name__,
new_state.__class__.__name__ ))
return new_state
class AmbiguousTransitionError(Exception):
"""Raised if more than one state transition was found for a
given input
"""
pass
################################################## Example
class MsgA(object):
"""example"""
pass
class MsgB(object):
"""example"""
pass
class MsgC(object):
"""example"""
pass
class StateA(State):
"""example"""
transitions = [(MsgA, "StateB", None)]
class StateB(State):
"""example"""
def set_transitions(self):
self.transitions = [(MsgB, "StateC", self._print_msg)]
@staticmethod
def _print_msg(inp):
"""example"""
print "Inp: ", inp
class StateC(State):
"""example"""
def set_transitions(self):
self.transitions = [(MsgB, "StateB", None),
(MsgC, "StateA", self._print_context)]
def _print_context(self, unused_inp):
"""example"""
print "Context: ", self._context
class Context(object):
def __init__(self):
self._state = StateA(self)
def handle(self, msg):
self._state = self._state(msg)
if __name__ == "__main__":
CONTEXT = Context()
CONTEXT.handle(MsgA())
CONTEXT.handle(MsgB())
CONTEXT.handle(MsgC())
CONTEXT.handle(MsgB())
#!/bin/env python
# -*- utf-8 -*-
"""Boiler Plate code for implementing a state machine according to
state design pattern
"""
class State(object):
"""Base class for a state in a statemachine
No statemachine class is needed. State transitions have to be
defined in self.transitions.
The object is callable and by calling it input is handled and
the next state is returned.
Example usage:
def handle(self, msg):
self._state = self._state(msg)
"""
transitions = [(None, None, None)]
def __init__(self, state_context):
self._context = state_context
def __call__(self, inp):
new_state = self
state_action_list = [(state, action) for cls, state,
action in self.transitions if
isinstance(inp, cls)]
if 1 == len(state_action_list):
state, action = state_action_list[0]
if action is not None:
try:
action(inp)
except TypeError:
action(self, inp)
new_state = state(self._context)
elif 1 < len(state_action_list):
raise AmbiguousTransitionError(inp, self)
print ("[DEBUG] %s -- %s --> %s" %
(self.__class__.__name__,
inp.__class__.__name__,
new_state.__class__.__name__ ))
return new_state
class AmbiguousTransitionError(Exception):
"""Raised if more than one state transition was found for a
given input
"""
pass
################################################## Example
class MsgA(object): pass
class MsgB(object): pass
class MsgC(object): pass
class StateA(State): pass
class StateB(State):
@staticmethod
def print_msg(inp):
print "Inp: ", inp
class StateC(State):
def print_context(self, unused_inp):
print "Context: ", self._context
StateA.transitions = [(MsgA, StateB, None)]
StateB.transitions = [(MsgB, StateC, StateB.print_msg)]
StateC.transitions = [(MsgB, StateB, None),
(MsgC, StateA, StateC.print_context)]
class Context(object):
def __init__(self):
self._state = StateA(self)
def handle(self, msg):
self._state = self._state(msg)
if __name__ == "__main__":
CONTEXT = Context()
CONTEXT.handle(MsgA())
CONTEXT.handle(MsgB())
CONTEXT.handle(MsgC())
CONTEXT.handle(MsgB())
State Design Pattern in Python
I'm trying to find the best - read: readable, maintainable, robust, threadsafe, usable - solution for a State Machine in python. For this I've been looking at the State Design Pattern. However I want a pythonic and lean solution (saying not implementing functions in subclasses that are not needed).
Long story short, here's what I came up with. What do you think?
#!/bin/env python
# -*- utf-8 -*-
class State(object):
"""Base class for a state in a statemachine
No statemachine class is needed. State transitions have to be
defined in self.transitions.
If access to self is needed in transitions they may be defined
using the set_transitions method.
The object is callable and by calling it input is handled and
the next state is returned.
Example usage:
def handle(self, msg):
self._state = self._state(msg)
"""
transitions = [(None, None, None)]
def __init__(self, state_context):
self._context = state_context
self.set_transitions()
def set_transitions(self):
"""subclass may override this if access to self is
needed in transitions
"""
pass
def __call__(self, inp):
new_state = self
state_action_list = [(state, action) for cls, state,
action in self.transitions if
isinstance(inp, cls)]
if 1 == len(state_action_list):
state, action = state_action_list[0]
if action is not None:
action(inp)
new_state = globals()[state](self._context)
elif 1 < len(state_action_list):
raise AmbiguousTransitionError(inp, self)
print ("[DEBUG] %s -- %s --> %s" %
(self.__class__.__name__,
inp.__class__.__name__,
new_state.__class__.__name__ ))
return new_state
class AmbiguousTransitionError(Exception):
"""Raised if more than one state transition was found for a
given input
"""
pass
################################################## Example
class MsgA(object):
"""example"""
pass
class MsgB(object):
"""example"""
pass
class MsgC(object):
"""example"""
pass
class StateA(State):
"""example"""
transitions = [(MsgA, "StateB", None)]
class StateB(State):
"""example"""
def set_transitions(self):
self.transitions = [(MsgB, "StateC", self._print_msg)]
@staticmethod
def _print_msg(inp):
"""example"""
print "Inp: ", inp
class StateC(State):
"""example"""
def set_transitions(self):
self.transitions = [(MsgB, "StateB", None),
(MsgC, "StateA", self._print_context)]
def _print_context(self, unused_inp):
"""example"""
print "Context: ", self._context
class Context(object):
def __init__(self):
self._state = StateA(self)
def handle(self, msg):
self._state = self._state(msg)
if __name__ == "__main__":
CONTEXT = Context()
CONTEXT.handle(MsgA())
CONTEXT.handle(MsgB())
CONTEXT.handle(MsgC())
CONTEXT.handle(MsgB())
The output looks like this:
python State.py
[DEBUG] StateA -- MsgA --> StateB
Inp: <__main__.MsgB object at 0x1eb6250>
[DEBUG] StateB -- MsgB --> StateC
Context: <__main__.Context object at 0x1eb6150>
[DEBUG] StateC -- MsgC --> StateA
[DEBUG] StateA -- MsgB --> StateA
lang-py