Python 状态机设计

Question

与 this Stack Overflow question (C state-machine design) 相关，StackOverflow 的人能否与我（和社区）分享您的 Python 状态机设计技术？

目前，我正在寻找基于以下内容的引擎：

class TrackInfoHandler(object):
    def __init__(self):
        self._state="begin"
        self._acc=""

    ## ================================== Event callbacks

    def startElement(self, name, attrs):
        self._dispatch(("startElement", name, attrs))

    def characters(self, ch):
        self._acc+=ch

    def endElement(self, name):
        self._dispatch(("endElement", self._acc))
        self._acc=""

    ## ===================================
    def _missingState(self, _event):
        raise HandlerException("missing state(%s)" % self._state)

    def _dispatch(self, event):
        methodName="st_"+self._state
        getattr(self, methodName, self._missingState)(event)

    ## =================================== State related callbacks

但我确信在利用 Python 的动态特性（例如动态调度）的同时，有很多方法可以实现。

我正在为接收“事件”和“调度”的“引擎”设计技术，而这些技术基于机器的“状态”。

Answer 1

我真的不明白这个问题。 State 设计模式非常清晰。请参阅Design Patterns book。

class SuperState( object ):
    def someStatefulMethod( self ):
        raise NotImplementedError()
    def transitionRule( self, input ):
        raise NotImplementedError()

class SomeState( SuperState ):
    def someStatefulMethod( self ):
        actually do something()
    def transitionRule( self, input ):
        return NextState()

这是很常见的样板，用于 Java、C++、Python（我敢肯定还有其他语言）。

如果您的状态转换规则碰巧是微不足道的，则可以进行一些优化以将转换规则本身推送到超类中。

请注意，我们需要有前向引用，因此我们通过名称引用类，并使用eval 将类名转换为实际类。另一种方法是使转换规则实例变量而不是类变量，然后在定义所有类后创建实例。

class State( object ):
    def transitionRule( self, input ):
        return eval(self.map[input])()

class S1( State ): 
    map = { "input": "S2", "other": "S3" }
    pass # Overrides to state-specific methods

class S2( State ):
    map = { "foo": "S1", "bar": "S2" }

class S3( State ):
    map = { "quux": "S1" }

在某些情况下，您的事件并不像测试对象是否相等那么简单，因此更通用的转换规则是使用适当的函数-对象对列表。

class State( object ):
    def transitionRule( self, input ):
        next_states = [ s for f,s in self.map if f(input)  ]
        assert len(next_states) >= 1, "faulty transition rule"
        return eval(next_states[0])()

class S1( State ):
    map = [ (lambda x: x == "input", "S2"), (lambda x: x == "other", "S3" ) ]

class S2( State ):
    map = [ (lambda x: "bar" <= x <= "foo", "S3"), (lambda x: True, "S1") ]

由于规则是按顺序评估的，因此允许使用“默认”规则。

Answer 2

这可能取决于您的状态机有多复杂。对于简单的状态机，dicts（事件键到 DFA 的状态键，或事件键到 NFA 的状态键列表/集合/元组）的字典可能是最容易编写和理解的东西。

对于更复杂的状态机，我听说过关于 SMC 的好消息，它可以将声明性状​​态机描述编译成各种语言的代码，包括 Python。

Answer 3

this design pattern 用于使用装饰器来实现状态机。从页面上的描述：

装饰器用于指定哪些方法是类的事件处理程序。

页面上也有示例代码（比较长，这里就不贴了）。

Answer 4

我认为 S. Lott 的回答是实现状态机的更好方法，但如果您仍想继续使用您的方法，使用 (state,event) 作为您的 dict 的键会更好。修改你的代码：

class HandlerFsm(object):

  _fsm = {
    ("state_a","event"): "next_state",
    #...
  }

Answer 5

我不会考虑使用有限状态机来处理 XML。我认为通常的方法是使用堆栈：

class TrackInfoHandler(object):
    def __init__(self):
        self._stack=[]

    ## ================================== Event callbacks

    def startElement(self, name, attrs):
        cls = self.elementClasses[name]
        self._stack.append(cls(**attrs))

    def characters(self, ch):
        self._stack[-1].addCharacters(ch)

    def endElement(self, name):
        e = self._stack.pop()
        e.close()
        if self._stack:
            self._stack[-1].addElement(e)

对于每种元素，您只需要一个支持addCharacters、addElement 和close 方法的类。

编辑：为了澄清，是的，我的意思是说有限状态机通常是错误的答案，作为一种通用编程技术，它们是垃圾，你应该远离。

有一些非常容易理解、清晰描述的问题，FSM 是一个很好的解决方案。例如，lex 就是好东西。

也就是说，FSM 通常不能很好地应对变化。假设有一天你想添加一些状态，也许是“我们见过元素 X 了吗？”旗帜。在上面的代码中，您将布尔属性添加到适当的元素类，然后就完成了。在有限状态机中，状态和转换的数量增加了一倍。

一开始需要有限状态的问题通常会演变为需要更多状态，比如可能是一个数字，此时要么你的 FSM 方案是烤面包，要么更糟糕的是，你将它演变成某种广义状态机，到那时你真的有麻烦了。你走得越远，你的规则就越像代码一样——但是你发明的一种慢速解释语言的代码，没有人知道，没有调试器，也没有工具。

Answer 6

在 2009 年 4 月的 Python 杂志上，我写了一篇关于在 Python 中嵌入状态 DSL 的文章，使用 pyparsing 和 imputil。此代码将允许您编写模块 trafficLight.pystate：

# trafficLight.pystate

# define state machine
statemachine TrafficLight:
    Red -> Green
    Green -> Yellow
    Yellow -> Red

# define some class level constants
Red.carsCanGo = False
Yellow.carsCanGo = True
Green.carsCanGo = True

Red.delay = wait(20)
Yellow.delay = wait(3)
Green.delay = wait(15)

DSL 编译器将创建所有必要的 TrafficLight、Red、Yellow 和 Green 类，以及适当的状态转换方法。代码可以使用类似这样的方式调用这些类：

import statemachine
import trafficLight

tl = trafficLight.Red()
for i in range(6):
    print tl, "GO" if tl.carsCanGo else "STOP"
    tl.delay()
    tl = tl.next_state()

（不幸的是，imputil 在 Python 3 中已被删除。）

Answer 7

以下代码是一个非常简单的解决方案。唯一有趣的部分是：

   def next_state(self,cls):
      self.__class__ = cls

每个状态的所有逻辑都包含在一个单独的类中。通过替换正在运行的实例的“__class__”来更改“状态”。

#!/usr/bin/env python

class State(object):
   call = 0 # shared state variable
   def next_state(self,cls):
      print '-> %s' % (cls.__name__,),
      self.__class__ = cls

   def show_state(self,i):
      print '%2d:%2d:%s' % (self.call,i,self.__class__.__name__),

class State1(State):
   __call = 0  # state variable
   def __call__(self,ok):
      self.show_state(self.__call)
      self.call += 1
      self.__call += 1
      # transition
      if ok: self.next_state(State2)
      print '' # force new line

class State2(State):
   __call = 0
   def __call__(self,ok):
      self.show_state(self.__call)
      self.call += 1
      self.__call += 1
      # transition
      if ok: self.next_state(State3)
      else: self.next_state(State1)
      print '' # force new line

class State3(State):
   __call = 0
   def __call__(self,ok):
      self.show_state(self.__call)
      self.call += 1
      self.__call += 1
      # transition
      if not ok: self.next_state(State2)
      print '' # force new line

if __name__ == '__main__':
   sm = State1()
   for v in [1,1,1,0,0,0,1,1,0,1,1,0,0,1,0,0,1,0,0]:
      sm(v)
   print '---------'
   print vars(sm

结果：

 0: 0:State1 -> State2 
 1: 0:State2 -> State3 
 2: 0:State3 
 3: 1:State3 -> State2 
 4: 1:State2 -> State1 
 5: 1:State1 
 6: 2:State1 -> State2 
 7: 2:State2 -> State3 
 8: 2:State3 -> State2 
 9: 3:State2 -> State3 
10: 3:State3 
11: 4:State3 -> State2 
12: 4:State2 -> State1 
13: 3:State1 -> State2 
14: 5:State2 -> State1 
15: 4:State1 
16: 5:State1 -> State2 
17: 6:State2 -> State1 
18: 6:State1 
---------
{'_State1__call': 7, 'call': 19, '_State3__call': 5, '_State2__call': 7}

Answer 8

我认为工具PySCXML 也需要仔细研究。

本项目使用 W3C 定义：State Chart XML (SCXML): State Machine Notation for Control Abstraction

SCXML 提供基于 CCXML 和 Harel 状态表的通用状态机执行环境

目前，SCXML 是一个工作草案；但它很快就会获得 W3C 推荐的可能性很高（这是第 9 稿）。

要强调的另一个有趣的点是，有一个 Apache Commons 项目旨在创建和维护一个 Java SCXML 引擎，该引擎能够执行使用 SCXML 文档定义的状态机，同时抽象出环境接口......

对于某些其他工具，支持这项技术将在未来 SCXML 离开其草案状态时出现......

Answer 9

其他相关项目：

• http://fsme.sourceforge.net/

• https://code.google.com/p/visio2python/

您可以绘制状态机，然后在您的代码中使用它。

Answer 10

我也对 state_machines 的当前选项不满意，所以我编写了 state_machine 库。

您可以通过pip install state_machine 安装它并像这样使用它：

@acts_as_state_machine
class Person():
    name = 'Billy'

    sleeping = State(initial=True)
    running = State()
    cleaning = State()

    run = Event(from_states=sleeping, to_state=running)
    cleanup = Event(from_states=running, to_state=cleaning)
    sleep = Event(from_states=(running, cleaning), to_state=sleeping)

    @before('sleep')
    def do_one_thing(self):
        print "{} is sleepy".format(self.name)

    @before('sleep')
    def do_another_thing(self):
        print "{} is REALLY sleepy".format(self.name)

    @after('sleep')
    def snore(self):
        print "Zzzzzzzzzzzz"

    @after('sleep')
    def big_snore(self):
        print "Zzzzzzzzzzzzzzzzzzzzzz"

person = Person()
print person.current_state == person.sleeping       # True
print person.is_sleeping                            # True
print person.is_running                             # False
person.run()
print person.is_running                             # True
person.sleep()

# Billy is sleepy
# Billy is REALLY sleepy
# Zzzzzzzzzzzz
# Zzzzzzzzzzzzzzzzzzzzzz

print person.is_sleeping                            # True

Answer 11

我绝对不建议自己实现这样一个众所周知的模式。如果您需要自定义功能，只需使用像 transitions 这样的开源实现并围绕它包装另一个类。在this post 中，我解释了为什么我更喜欢这种特定的实现及其特性。

Answer 12

这是我提出的“有状态对象”的解决方案，但对于您的预期目的来说效率相当低，因为状态更改相对昂贵。但是，它可能适用于不经常更改状态或仅经历有限数量的状态更改的对象。好处是一旦状态改变，就没有多余的间接了。

class T:
    """
    Descendant of `object` that rectifies `__new__` overriding.

    This class is intended to be listed as the last base class (just
    before the implicit `object`).  It is a part of a workaround for

      * https://bugs.python.org/issue36827
    """

    @staticmethod
    def __new__(cls, *_args, **_kwargs):
        return object.__new__(cls)

class Stateful:
    """
    Abstract base class (or mixin) for "stateful" classes.

    Subclasses must implement `InitState` mixin.
    """

    @staticmethod
    def __new__(cls, *args, **kwargs):
        # XXX: see https://stackoverflow.com/a/9639512
        class CurrentStateProxy(cls.InitState):
            @staticmethod
            def _set_state(state_cls=cls.InitState):
                __class__.__bases__ = (state_cls,)

        class Eigenclass(CurrentStateProxy, cls):
            __new__ = None  # just in case

        return super(__class__, cls).__new__(Eigenclass, *args, **kwargs)

# XXX: see https://bugs.python.org/issue36827 for the reason for `T`.
class StatefulThing(Stateful, T):
    class StateA:
        """First state mixin."""

        def say_hello(self):
            self._say("Hello!")
            self.hello_count += 1
            self._set_state(self.StateB)
            return True

        def say_goodbye(self):
            self._say("Another goodbye?")
            return False

    class StateB:
        """Second state mixin."""

        def say_hello(self):
            self._say("Another hello?")
            return False

        def say_goodbye(self):
            self._say("Goodbye!")
            self.goodbye_count += 1
            self._set_state(self.StateA)
            return True

    # This one is required by `Stateful`.
    class InitState(StateA):
        """Third state mixin -- the initial state."""

        def say_goodbye(self):
            self._say("Why?")
            return False

    def __init__(self, name):
        self.name = name
        self.hello_count = self.goodbye_count = 0

    def _say(self, message):
        print("{}: {}".format(self.name, message))

    def say_hello_followed_by_goodbye(self):
        self.say_hello() and self.say_goodbye()

# ----------
# ## Demo ##
# ----------
if __name__ == "__main__":
    t1 = StatefulThing("t1")
    t2 = StatefulThing("t2")
    print("> t1, say hello.")
    t1.say_hello()
    print("> t2, say goodbye.")
    t2.say_goodbye()
    print("> t2, say hello.")
    t2.say_hello()
    print("> t1, say hello.")
    t1.say_hello()
    print("> t1, say hello followed by goodbye.")
    t1.say_hello_followed_by_goodbye()
    print("> t2, say goodbye.")
    t2.say_goodbye()
    print("> t2, say hello followed by goodbye.")
    t2.say_hello_followed_by_goodbye()
    print("> t1, say goodbye.")
    t1.say_goodbye()
    print("> t2, say hello.")
    t2.say_hello()
    print("---")
    print( "t1 said {} hellos and {} goodbyes."
           .format(t1.hello_count, t1.goodbye_count) )
    print( "t2 said {} hellos and {} goodbyes."
           .format(t2.hello_count, t2.goodbye_count) )

    # Expected output:
    #
    #     > t1, say hello.
    #     t1: Hello!
    #     > t2, say goodbye.
    #     t2: Why?
    #     > t2, say hello.
    #     t2: Hello!
    #     > t1, say hello.
    #     t1: Another hello?
    #     > t1, say hello followed by goodbye.
    #     t1: Another hello?
    #     > t2, say goodbye.
    #     t2: Goodbye!
    #     > t2, say hello followed by goodbye.
    #     t2: Hello!
    #     t2: Goodbye!
    #     > t1, say goodbye.
    #     t1: Goodbye!
    #     > t2, say hello.
    #     t2: Hello!
    #     ---
    #     t1 said 1 hellos and 1 goodbyes.
    #     t2 said 3 hellos and 2 goodbyes.

我已发布“征求意见”here。