python - pytransitions/transitions : Is there any better way to store the history of visited state?

Question

我最近在 Python 中发现了一个轻量级、面向对象的状态机实现，称为转换 (https://github.com/pytransitions/transitions)。所以我正在尝试使用这些状态机，尤其是 HierarchicalGraphMachine。我想要的一个很好的功能是即使机器不移动(保持在相同状态)也可以存储访问状态的历史记录。
从我从示例中看到的情况来看，我们实际上无法以简单的方式做到这一点，因为 before_state_change和 after_state_change当机器的状态没有改变时没有被调用。所以在这种情况下我们不能扩展我们的历史。为了解决这个问题，我最终创建了一个 trigger_wrapper 函数:

def trigger_wrapper(self, trigger_name):
        
        previous_state = self.state

        result = None
        try:
            result = self.trigger(trigger_name)
        except AttributeError as attribute_err:
            print('Invalid trigger name: {}'.format(attribute_err))
        except MachineError as machine_err:
            print('Valid trigger name but not reachable: {}'.format(machine_err))
        except Exception as err:
            print('Cannot make transition with unknown error: {}'.format(err))
        
        if result is False:
            print('Trigger name reachable but condition(s) was not fulfilled')
            ....

        current_state = self.state

        # update history
        ..... 

        return result

然后，我们调用 trigger_wrapper 而不是 trigger:

before: machine.trigger('drink')
now:    machine.trigger_wrapper('drink').

除此之外，通过设置 ignore_invalid_triggers = False当初始化 Machine并使用此 trigger_wrapper函数，我们现在可以通过缓存异常知道机器无法移动的原因。
有没有更好的解决方案来保持跟踪访问状态？我认为另一种方法是覆盖触发器函数，但由于 NestedState 的原因，它看起来很复杂。 .
编辑 1 (遵循@aleneum 的建议)
感谢您的回复以及一个有趣的例子!!!
按照使用 finalize_event 的示例进行操作.一切顺利，但这个回调函数似乎不足以捕捉以下情况(我在代码中添加了 2 行额外的行):

... same setup as before
m.go()
m.internal()
m.reflexive()
m.condition()
m.go()    # MachineError: "Can't trigger event go from state B!"
m.goo()   # AttributeError: Do not know event named 'goo'.

>>> Expected: ['go', 'internal', 'reflexive', 'condition', 'go', 'goo'] 
>>> Expected: ['B', 'B', 'B', 'B', 'B', 'B']

换句话说，是否有另一个回调可以捕获由调用 invalid trigger 引起的异常？ (示例中的 goo)或由 valid trigger but not reachable from the current state 引起(从状态 B 调用 go())？
再次感谢你的帮助。

Answer 1

正如您已经提到的，before_state_change和 after_state_change仅在发生转换时调用。这并不一定意味着状态更改，因为内部和自反转换也会触发这些回调:

from transitions import Machine


def test():
    print("triggered")


m = Machine(states=['A', 'B'], transitions=[
    ['go', 'A', 'B'],
    dict(trigger='internal', source='B', dest=None),
    dict(trigger='reflexive', source='B', dest='='),
    dict(trigger='condition', source='B', dest='A', conditions=lambda: False)
], after_state_change=test, initial='A')


m.go()  # >>>  triggered
m.internal()  # >>> triggered
m.reflexive()  # >>> triggered
m.condition()  # no output

唯一不会触发 after_state_change 的事件这是 m.condition因为转换被(未满足的)条件停止了。
因此，当您的目标是跟踪实际进行的转换时， after_state_change是正确的地方。如果你想记录每个触发器/事件，你可以通过 finalize_event 来做到这一点。 :

'machine.finalize_event' - callbacks will be executed even if no transition took place or an exception has been raised

from transitions import Machine


event_log = []
state_log = []


def log_trigger(event_data):
    event_log.append(event_data.event.name)
    state_log.append(event_data.state)


m = Machine(states=['A', 'B'], transitions=[
    ['go', 'A', 'B'],
    dict(trigger='internal', source='B', dest=None),
    dict(trigger='reflexive', source='B', dest='='),
    dict(trigger='condition', source='B', dest='A', conditions=lambda event_data: False)
], finalize_event=log_trigger, initial='A', send_event=True)


m.go()
m.internal()
m.reflexive()
m.condition()

print(event_log)  # >>> ['go', 'internal', 'reflexive', 'condition']
print([state.name for state in state_log])  # >>> ['B', 'B', 'B', 'B']

回调传递给 finalize_event将始终被调用，即使转换引发异常。通过设置 send_event=True ，所有回调都会收到 EvenData包含事件、状态和转换信息以及出现问题时的错误对象的对象。这是我必须更改条件 lambda 表达式的方式。当 send_event=True , 所有回调都需要能够处理 EventData目的。
更多信息 finalize_event回调执行顺序可以在 this section中找到的文档。
如何记录无效事件？ finalize_event仅对有效事件调用，这意味着该事件必须存在并且还必须在当前源状态上有效。如果应该处理所有事件， Machine需要扩展:

from transitions import Machine

log = []


class LogMachine(Machine):

    def _get_trigger(self, model, trigger_name, *args, **kwargs):
        res = super(LogMachine, self)._get_trigger(model, trigger_name, *args, **kwargs)
        log.append((trigger_name, model.state))
        return res

# ...
m = LogMachine(states=..., ignore_invalid_triggers=True)
assert m.trigger("go")  # valid
assert not m.trigger("go")  # invalid
assert not m.trigger("gooo")  # invalid
print(log)  # >>> [('go', 'B'), ('go', 'B'), ('gooo', 'B')]

每个型号都饰有 trigger方法是 Machine._get_trigger 的一部分已分配 model范围。 Model.trigger可用于按名称触发事件，也可用于处理不存在的事件。您还需要通过 ignore_invalid_triggers=True不养 MachineError当事件无效时。
但是，如果应该记录所有事件，则从 Machine 拆分日志记录可能更可行/更易于维护。并处理处理事件的日志记录，例如:

m = Machine(..., ignore_invalid_triggers=True)
# ...
def on_event(event_name):
   logging.debug(f"Received event {event_name}")  # or log_event.append(event_name)
   m.trigger(event_name)
   logging.debug(f"Machine state is {m.state}")  # or log_state.append(m.state)