50

这个 Stack Overflow 问题(C 状态机设计)相关,Stack Overflow 的人能否与我(和社区)分享您的 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 的动态特性(例如动态调度)的同时,有很多方法可以解决这个问题。

我追求的是“引擎”的设计技术,它接收“事件”和“调度”,而不是基于机器“状态”的那些。

4

12 回答 12

40

我真的不明白这个问题。状态设计模式非常清晰。请参阅设计模式一书

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") ]

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

于 2010-01-20T14:27:44.677 回答
12

在 2009 年 4 月的 Python 杂志上,我写了一篇关于在 Python 中嵌入 State 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 中已被删除。)

于 2010-01-20T17:19:18.523 回答
9

有这种使用装饰器来实现状态机的设计模式。从页面上的描述:

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

页面上也有示例代码(它很长,所以我不会在这里粘贴)。

于 2010-01-20T14:51:51.560 回答
5

我对 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
于 2014-03-13T18:48:51.913 回答
3

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

class HandlerFsm(object):

  _fsm = {
    ("state_a","event"): "next_state",
    #...
  }
于 2010-01-20T14:53:30.043 回答
2

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

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

于 2010-01-20T14:45:07.240 回答
2

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

   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}
于 2011-05-28T01:18:25.007 回答
2

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

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

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

目前,SCXML 是一个工作草案;但它很快获得 W3C 推荐的可能性很高(这是第 9 稿)。

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

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

于 2011-09-19T12:07:36.307 回答
1

我不会考虑使用有限状态机来处理 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)

对于每种元素,您只需要一个支持addCharactersaddElementclose方法的类。

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

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

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

一开始需要有限状态的问题通常会演变为需要更多状态,比如可能是一个数字,此时您的 FSM 方案是 toast,或者更糟糕的是,您将其演变为某种通用状态机,然后您'真的有麻烦了。你走得越远,你的规则就越像代码一样——但是你发明的一种慢速解释语言的代码,没有人知道,没有调试器,也没有工具。

于 2010-01-20T15:24:44.513 回答
1

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

于 2016-08-25T15:58:01.660 回答
0

其他相关项目:

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

于 2013-06-05T09:16:52.573 回答
0

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

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.

我在这里发布了“征求意见” 。

于 2019-06-07T20:43:12.893 回答