python - pytransitions/transitions:有没有更好的方法来存储访问状态的历史?
问题描述
我最近在 Python 中发现了一个轻量级、面向对象的状态机实现,称为转换 ( https://github.com/pytransitions/transitions )。所以我尝试使用这些状态机,尤其是 HierarchicalGraphMachine。我想要的一个很好的功能是即使机器没有移动(保持相同的状态)也可以存储访问状态的历史记录。
而且从我从示例中看到的情况来看,我们实际上无法以简单的方式做到这一点,因为当机器状态未更改时before_state_change
,并after_state_change
没有调用 and。所以我们不能在这种情况下扩展我们的历史。为了解决这个问题,我最终创建了一个 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 in the example) 或valid trigger but not reachable from the current state
(call go() from state B) 引起的异常?
再次感谢你的帮助。
解决方案
正如您已经提到的,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' - 即使没有发生转换或引发异常,也会执行回调
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
有关回调执行顺序的更多信息,请参阅文档的此部分。
如何也记录无效事件?
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)
推荐阅读
- php - 同一页面上的 AJAX 调用 $_POST 变量为空
- php - 本地 mysql 上的错误,但 RDS mysql 上没有
- python - 目标变量线性回归的对数
- r - 在 R 中显示带有包装 likert 的百分比
- php - 在屏幕上打印响应头的消息
- django - Django 中的高级搜索
- php - 调用未定义的方法 DateTime :: getTimezone () 错误 laravel Carbon
- django - AttributeError:“DatabaseWrapper”对象没有属性“set_schema_to_public”(tenat_schemas)
- javascript - 从映射组件打开对话框做出反应
- php - 创建 PHP API(将数据从 mySQL 转换为 JSON);