如何构建兼具HierarchicalMachine与AsyncMachine特性的Pytransitions状态机？

问题描述 #

想要构建同时支持分层结构和异步操作的状态机，尝试多重继承HierarchicalMachine和AsyncMachine后出现报错：

RuntimeError: AsyncMachine should not call Machine._process. Use Machine._process_async instead.

原代码示例：

from transitions.extensions.markup import MarkupMachine
from transitions.extensions.factory import HierarchicalMachine
from transitions.extensions.asyncio import AsyncMachine
QUEUED = False

class Unhealthy(HierarchicalMachine, AsyncMachine):
    def __init__(self):
        states = [{"name":'aborted', "on_enter":[]},
                    {"name":'clearancetimeouterr', "on_enter":[]},
                    {"name":"awaitingclearanceerr", 'on_enter':[]},
                    {"name":"cleared", 'on_enter':[]}
                    ]
        transitions = [{"trigger":"abort", "source":"aborted", "dest":"awaitingclearanceerr"},
                        {"trigger":"awaitingclearanceerr", "source":"clearancetimeout", "dest":"awaitingclearanceerr"},
                        {"trigger":"cleared", "source":"awaitingclearanceerr", "dest":"cleared"}]
        super().__init__(states=states, transitions=transitions, initial="awaitingclearanceerr", queued=QUEUED)
        

class Healthy(HierarchicalMachine, AsyncMachine):
    def __init__(self):
        unhealthy = Unhealthy()
        states = [{"name":'idle', 'on_enter':[]},
                    {"name":"busy", 'on_enter':[]},
                    {"name":"done", 'on_enter':[]}]

        transitions = [{'trigger':'start', 'source':'idle', 'dest':'busy'},
                        {"trigger":"done", "source":"busy", "dest":"done"},
                        {"trigger":"idle", "source":"awaiting_clearance", "dest":"idle"}]
        super().__init__(states=states, transitions=transitions, initial="idle", queued=QUEUED)



class StateMachine(HierarchicalMachine, MarkupMachine, AsyncMachine):
    def __init__(self):
        unhealthy= Unhealthy()
        healthy = Healthy()
        states = [{'name':"idle"}, {"name":'healthy', 'children':healthy}, {"name":"unhealthy", "children":unhealthy}]
        super().__init__(states=states, initial="idle", queued=QUEUED)
        self.add_transition("start_machine", "idle", "healthy")
        self.add_transition('abort', 'healthy', 'unhealthy')

问题原因 #

HierarchicalMachine的内部逻辑依赖同步的_process方法处理状态转换，但AsyncMachine强制要求使用异步的_process_async方法，多重继承后两者的执行逻辑冲突，导致报错。

解决方案 #

不需要同时继承HierarchicalMachine和AsyncMachine——AsyncMachine本身支持分层状态的定义，只需在状态配置中通过children属性关联子状态机即可实现分层结构。

修改后的完整代码 #

from transitions.extensions.markup import MarkupMachine
from transitions.extensions.asyncio import AsyncMachine
QUEUED = False

class Unhealthy(AsyncMachine):
    def __init__(self):
        states = [{"name":'aborted', "on_enter":[]},
                    {"name":'clearancetimeouterr', "on_enter":[]},
                    {"name":"awaitingclearanceerr", 'on_enter':[]},
                    {"name":"cleared", 'on_enter':[]}
                    ]
        # 修正原代码中source拼写错误（clearancetimeout -> clearancetimeouterr）
        transitions = [{"trigger":"abort", "source":"aborted", "dest":"awaitingclearanceerr"},
                        {"trigger":"awaitingclearanceerr", "source":"clearancetimeouterr", "dest":"awaitingclearanceerr"},
                        {"trigger":"cleared", "source":"awaitingclearanceerr", "dest":"cleared"}]
        super().__init__(states=states, transitions=transitions, initial="awaitingclearanceerr", queued=QUEUED)
        

class Healthy(AsyncMachine):
    def __init__(self):
        states = [{"name":'idle', 'on_enter':[]},
                    {"name":"busy", 'on_enter':[]},
                    {"name":"done", 'on_enter':[]}]

        # 修正原代码中不存在的source状态（awaiting_clearance -> done）
        transitions = [{'trigger':'start', 'source':'idle', 'dest':'busy'},
                        {"trigger":"done", "source":"busy", "dest":"done"},
                        {"trigger":"idle", "source":"done", "dest":"idle"}]
        super().__init__(states=states, transitions=transitions, initial="idle", queued=QUEUED)



class StateMachine(AsyncMachine, MarkupMachine):
    def __init__(self):
        unhealthy= Unhealthy()
        healthy = Healthy()
        states = [{'name':"idle"}, 
                  {"name":'healthy', 'children':healthy}, 
                  {"name":"unhealthy", "children":unhealthy}]
        super().__init__(states=states, initial="idle", queued=QUEUED)
        self.add_transition("start_machine", "idle", "healthy")
        self.add_transition('abort', 'healthy', 'unhealthy')

使用说明 #

因为使用了AsyncMachine，所有触发状态转换的方法必须用await调用，示例：

import asyncio

async def main():
    sm = StateMachine()
    await sm.start_machine()
    print(sm.state)  # 输出: healthy
    await sm.abort()
    print(sm.state)  # 输出: unhealthy

asyncio.run(main())

内容的提问来源于stack exchange，提问作者Tanisha Mittal