python - 在 Transitions 中将具有多个状态的模型附加到多台机器

标签 python workflow transition state-machine

Transitions 是一个用于实现状态机的python库。
末尾的 Transitions 文档中所述alternative initialization patterns 部分,我们可以使用不同的 model_attribute 将多台机器附加到具有多个状态的模型上。值。
我正在以类似于文档中提到的方式实现它:

class Matter():
    pass

lump = Matter()

states = ['solid','liquid','gas']
transitions = [
    { 'trigger': 'melt', 'source': 'solid', 'dest': 'liquid'},
    { 'trigger': 'evaporate', 'source': 'liquid', 'dest': 'gas'}
]

matter_machine = Machine(lump, states=states, transitions=transitions, initial='solid', model_attribute='state')
shipment_machine = Machine(lump, states=states, transitions=transitions, initial='liquid', model_attribute='shipping_state')


lump.melt()
lump.evaporate()
print(lump.state)
>>> gas
print(lump.shipping_state)
>>> liquid
我应该如何访问具有 model_attribute='shipping_state' 的机器,即 shipment_machine ,因为所有转换仅在第一台机器上执行,即 matter_machine因为它首先被初始化。我无法在 shipment_machine 上执行任何转换.
另外我想告诉你,我打算做的是为matter_machine配备一个子机。例如,对于给定的模型,我有一台名为 Car 的机器。 ,我想有一台子机Driver对于它,具有各自的状态和转换。

最佳答案

目前,我在这里看到两个不同的问题。首先,如何使用“shipment_state”以及如何使用嵌套/并行状态。
问题 1:具有两个(独立)状态属性的模型。
在模型上触发事件的最常见方法是使用 transitions 的便利函数。添加到您的模型中。 README声明如下:

Each method triggers the corresponding transition. You don't have to explicitly define these methods anywhere; the name of each transition is bound to the model passed to the Machine initializer (in this case, lump). To be more precise, your model should not already contain methods with the same name as event triggers since transitions will only attach convenience methods to your model if the spot is not already taken.


我在这里强调了给你带来麻烦的部分。默认行为阻止使用 添加两台不同机器的触发器。同名 . transitions如果对象尚未包含名为“melt”的方法/属性,则只会添加便利函数(例如 lump.melt() )。
方案1.A:直接访问机器事件
除了添加的便利功能之外,还有更多方法可以触发事件。
例如,您可以直接触发事件。不过,您需要传递有问题的模型:

class Matter():
    pass

lump = Matter()

states = ['solid','liquid','gas']
transitions = [
    { 'trigger': 'melt', 'source': 'solid', 'dest': 'liquid'},
    { 'trigger': 'evaporate', 'source': 'liquid', 'dest': 'gas'}
]

matter_machine = Machine(lump, states=states, transitions=transitions, initial='solid', model_attribute='state')
shipment_machine = Machine(lump, states=states, transitions=transitions, initial='solid', model_attribute='shipping_state')

matter_machine.events['melt'].trigger(lump)  # will process event for 'state'
assert lump.state == 'liquid'
assert lump.shipping_state == 'solid'
shipment_machine.events['melt'].trigger(lump)  # will process event for 'shipment_state'
assert lump.state == lump.shipping_state
assert lump.shipping_state == 'liquid'
解决方案 1.B:覆盖机器分配策略。
不触及现有属性/字段的标准行为已经过调整,以防止混淆用户编写的自定义代码。一些用户仅按名称触发事件(例如 model.trigger('melt') )并将名为触发器( def melt(self, ...) )的方法添加到他们的转换回调中。但是这种行为可以通过覆盖 Machine._checked_assignment 来改变。 .在 FAQ Transitions 部分中的 notebook 不会向我的模型添加便捷方法,您会发现一个名为 CallingMachine 的示例性覆盖:
class CallingMachine(Machine):

    def _checked_assignment(self, model, name, func):
        if hasattr(model, name):
            predefined_func = getattr(model, name)
            def nested_func(*args, **kwargs):
                predefined_func(*args, **kwargs)
                func(*args, **kwargs)
            setattr(model, name, nested_func)
        else:
            setattr(model, name, func)
CallingMachine像默认机器一样工作,但会将属性包装到函数中并调用它以及事件触发器。这是一种非常简单的方法,因为它不检查先前的赋值属性是否实际上是可调用的。如果你仔细看,你也会发现我改变了predefined_func并传递参数,而不是像在常见问题解答中那样仅仅调用它。在您的用例中,我们可以假设已经存在的方法是另一个触发器函数。如果您想在此处实际混合可调用对象,则此假设可能会引起麻烦。最后,与 CallingMachine而不是 Machine将触发一个事件 在两台机器上 :
matter_machine = CallingMachine(lump, states=states, transitions=transitions, initial='solid', model_attribute='state')
shipment_machine = CallingMachine(lump, states=states, transitions=transitions, initial='solid', model_attribute='shipping_state')

lump.melt()
assert lump.state == lump.shipping_state
assert lump.shipping_state == 'liquid'
问题2:将不同的机器/型号组合成一个嵌套结构。
解决方案 2.A:使用并发和嵌套
除了在类/模型级别将状态拼接在一起,您还可以使用并发分层状态机组合状态或嵌套配置。并发最近添加到 transitions在版本0.8.0 .对于您的汽车/司机示例,这可能如下所示:
from transitions.extensions.nesting import HierarchicalMachine as HSM

# our car
car_config = dict(name="car", children=['stopped', 'accelerating', 'decelerating', 'driving'], 
                  initial='stopped',
                  transitions=[
                      ['accelerate', '*', 'accelerating'],  # gotta go fast
                      ['braking', ['driving', 'accelerating'], 'decelerating'],  # slower!
                      ['braking', 'stopped', 'stopped'],  # can't be slower than stopped
                      ['release', ['accelerating', 'decelerating'], 'driving'],  # releasing the pedal will result in a steady speed ...
                      ['release', 'stopped', 'stopped']  # ... which could be 0 km/h
                  ])

driver_config = dict(name="driver", children=['rested', 'tired', 'sleepy', 'jamming'],
                     initial='rested',
                     transitions=[
                         ['driving', 'rested', 'tired'],  # driving can be exhausting
                         ['driving', 'tired', 'sleepy'],  # when you tired you can get even more sleepy
                         ['tunein', '*', 'jamming'],  # music allways helps ...
                         ['driving', 'jamming', 'rested'],  # to replenish some energy
                         ['resting', '*', 'rested']  # sometimes a break is better though
                     ])

states = ['off', {'name': 'running', 'parallel': [car_config, driver_config]}]

m = HSM(states=states, initial='off', auto_transitions=False,
        transitions=[['start', 'off', 'running']])
print(m.state)  # >>> off
m.start()  
print(m.state)  # >>> ['running_car_stopped', 'running_driver_rested']
m.accelerate()
print(m.state)  # >>> ['running_car_accelerating', 'running_driver_rested']
m.driving()
m.driving()
print(m.state)  # >>> ['running_car_accelerating', 'running_driver_sleepy']
作为图表,这大致如下所示:
enter image description here
解决方案 2.B:将机器作为多个模型的“规则手册”
分层机器和并发性可能会变得困惑(因为一切都比概念性的交通灯更复杂;我认为真正的机器也是复杂的野兽)。
将一些想法放在转换/状态命名和 ignore_invalid_triggers=True 中可能足以处理汽车/司机的例子。
我们将状态和转换合并为大集合,但要确保它们不重叠。
这样我们就可以用一台机器管理多个模型/组件作为一种规则手册:
from transitions import Machine

class Car:

    def driving(self):
        print("vrooom")
        # e.g. calculate passed distance

class Driver:
    pass

car_states = ['stopped', 'accelerating', 'decelerating', 'driving']
driver_states = ['rested', 'tired', 'sleepy', 'jamming']

transitions = [
    # car transitions
    ['accelerate', car_states, 'accelerating'],
    ['braking', ['driving', 'accelerating'], 'decelerating'],
    ['braking', 'stopped', 'stopped'],
    ['release', ['accelerating', 'decelerating'], 'driving'],
    ['release', 'stopped', 'stopped'],
    # driver
    ['driving', 'rested', 'tired'],
    ['driving', 'tired', 'sleepy'],
    ['tunein', driver_states, 'jamming'],
    ['driving', 'jamming', 'rested'],
    ['resting', driver_states, 'rested']
]

car = Car()
driver = Driver()
# initialize a 'blank' combined rule book
# ignore_invalid_triggers=True allows us to call m.dispatch even when a trigger isn't defined for all models'
# current state
m = Machine(model=None, states=car_states + driver_states, transitions=transitions, auto_transitions=False,
            initial=None, ignore_invalid_triggers=True)
# we add models after construction since they have different initial states
m.add_model(car, initial='stopped')
m.add_model(driver, initial='rested')
car.accelerate()
# dispatch triggers an event on ALL models
m.dispatch('driving')  # >>> vrooom
assert car.is_accelerating()
assert driver.is_tired()

关于python - 在 Transitions 中将具有多个状态的模型附加到多台机器,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/63123581/

相关文章:

python - 当字段名在变量中时,如何访问类的字段?

python - 我可以在 C 中使用 Python/Ruby ORM 吗?

python - Django Admin 中的欧洲日期输入

像 RStudio 一样的 Python 工作流?

.net - Windows Workflow Foundation 持久性架构解释?

ios - View Controller 的过渡 - 从左到右

image - Vulkan - 如何知道当前图像布局是什么?

python - 如何使用 Python 在网格中创建 10 个随机 x、y 坐标

css - Firefox 和 Chrome 3D 转换问题

github - 如何在不使用release分支的情况下使用git flow?