Блог

Я пытаюсь использовать иерархическую машину с тремя уровнями вложенности main -> nested -> deeper . Я бы ожидал, что конечные автоматы будут выполняться один за другим, а затем состояния будут переназначены обратно на первую машину. Итак, я ожидал бы, что конечное состояние есть done , но оно есть nested_deeper_working , поэтому, очевидно, я что-то упускаю.

Обходной путь здесь заключается в использовании queued=False , тогда он работает так, как ожидалось. Но недостатком является то, что стек вызовов действительно длинный, а трассировка в случае какой-либо ошибки чертовски длинная.

Извините за длинный пример, я не смог сделать его короче. В реальной жизни я использую MainMachine как общий производственный контроль, он запускает меньшие машины для стирания, прошивки, калибровки или тестирования устройства. Они представлены NestedMachine . Внутри этих машин используются самые маленькие машины, например. для жесткого сброса — одна тестовая последовательность или около того. Это DeeperMachine в данном случае.

 pytransitions 0.8.10
python 3.7.3
 

GenericMachine класс — это просто абстрактный класс. Здесь я определяю состояния по умолчанию initial done , а также базовую конфигурацию.

 from transitions.extensions import HierarchicalMachine

class GenericMachine(HierarchicalMachine):
    def __init__(self, states, transitions, model=None):
        generic_states = [
            {"name": "initial", "on_enter": self.entry_initial},
            {"name": "done", "on_enter": self.entry_done},
        ]
        states  = generic_states
        super().__init__(
            states=states,
            transitions=transitions,
            model=model,
            send_event=True,
            queued=True,
        )

    def entry_initial(self, event_data):
        raise NotImplementedError

    def entry_done(self, event_data):
        raise NotImplementedError
 

MainMachine это самая высокая машина в иерархии, и она запускает NestedMachine . Ожидается, что после завершения всех вложенных машин done состояние будет выполнено.

 class MainMachine(GenericMachine):
    def __init__(self):
        nested = NestedMachine()
        remap = {"done": "done"}
        states = [
            {"name": "nested", "children": nested, "remap": remap},
        ]
        transitions = [
            ["go", "initial", "nested"],
        ]
        super().__init__(states, transitions, model=self)

    def entry_done(self, event_data):
        print("job finished")
 

NestedMachine действует как второй уровень вложенности. Он запускает DeeperMachine и переназначает done состояние.

 class NestedMachine(GenericMachine):
    def __init__(self):
        deeper = DeeperMachine()
        remap = {"done": "done"}
        states = [
            {"name": "deeper", "children": deeper, "remap": remap},
        ]
        transitions = [
            ["go", "initial", "deeper"],
        ]
        super().__init__(states, transitions)

    def entry_initial(self, event_data):
        event_data.model.go()
 

Третий уровень вложенности реализован DeeperMachine . После завершения работы go событие переходит в done состояние и переходит обратно через NestedMachine to MainMachine

 class DeeperMachine(GenericMachine):
    def __init__(self):
        states = [
            {"name": "working", "on_enter": self.entry_working},
        ]
        transitions = [
            ["go", "initial", "working"],
            ["go", "working", "done"],
        ]
        super().__init__(states, transitions, model=self)

    def entry_initial(self, event_data):
        event_data.model.go()

    def entry_working(self, event_data):
        event_data.model.go()
 

Тест создает экземпляр MainMachine и запускает первое событие. Ожидается, что будут вызваны вложенные машины, и после выполнения задания оно будет переназначено через done состояния обратно в MainMachine .

 import logging as log

def main():
    log.basicConfig(level=log.DEBUG)
    log.getLogger("transitions").setLevel(log.INFO)

    machine = MainMachine()
    machine.go()

    assert machine.state == "done"

if __name__ == "__main__":
    main()
 

Подтверждено как ошибка

https://github.com/pytransitions/transitions/issues/554

Решенный в dev-0.9, примерный и рабочий код хорошо работает для меня.