How to use the transitions.extensions.nesting.HierarchicalMachine function in transitions

To help you get started, we’ve selected a few transitions examples, based on popular ways it is used in public projects.

Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.

github opencord / voltha / voltha / extensions / omci / state_machines / image_agent.py View on Github external
queued=True,
                                    name='{}-window_section_machine'.format(self.__class__.__name__))

        #### Software Activation Sub State Machine ####
        OMCI_SWIMG_ACTIVATE_STATES = ['init_act', 'activating', 'busy', 'rebooting', 'committing', 'done', 'failed']
        OMCI_SWIMG_ACTIVATE_TRANSITIONS = [
            {'trigger': 'activate', 'source': ['init_act', 'busy'], 'dest': 'activating'},
            {'trigger': 'onu_busy', 'source': 'activating', 'dest': 'busy'},
            {'trigger': 'reboot',   'source': 'activating', 'dest': 'rebooting'},
            {'trigger': 'do_commit', 'source': ['activating', 'rebooting'], 'dest': 'committing'},
            # {'trigger': 'commit_ok', 'source': 'committing', 'dest': 'done'},
            {'trigger': 'reset_actimg', 'source': ['activating', 'rebooting', 'committing', 'failed'], 'dest': 'init_act'},
            # {'trigger': 'actimg_fail', 'source': ['init_act', 'activating', 'rebooting', 'committing'], 'dest': 'failed'}
        ]
        
        self.activate_machine = HMachine(model=self, 
                                         states=OMCI_SWIMG_ACTIVATE_STATES,
                                         transitions=OMCI_SWIMG_ACTIVATE_TRANSITIONS,
                                         initial='init_act',
                                         queued=True,
                                         name='{}-activate_machine'.format(self.__class__.__name__))
                                   
        #### Main State Machine ####
        OMCI_SWIMG_DOWNLOAD_STATES = [ 'init_image', 'starting_image', 'ending_image', 'endimg_busy', 'done_image',
                                      {'name': 'dwin', 'children': self.win_machine}, 
                                      {'name': 'actimg', 'children': self.activate_machine}
                                     ]
        OMCI_SWIMG_DOWNLOAD_TRANSITIONS = [
            {'trigger': 'start_image',      'source': 'init_image',     'dest': 'starting_image' },
            {'trigger': 'download_window',  'source': 'starting_image', 'dest': 'dwin_init_window' },
            {'trigger': 'download_success', 'source': 'dwin',           'dest': 'ending_image'   },
            {'trigger': 'onu_busy',         'source': 'ending_image',   'dest': 'endimg_busy'    },
github opencord / voltha / voltha / extensions / omci / state_machines / image_agent.py View on Github external
{'name': 'dwin', 'children': self.win_machine}, 
                                      {'name': 'actimg', 'children': self.activate_machine}
                                     ]
        OMCI_SWIMG_DOWNLOAD_TRANSITIONS = [
            {'trigger': 'start_image',      'source': 'init_image',     'dest': 'starting_image' },
            {'trigger': 'download_window',  'source': 'starting_image', 'dest': 'dwin_init_window' },
            {'trigger': 'download_success', 'source': 'dwin',           'dest': 'ending_image'   },
            {'trigger': 'onu_busy',         'source': 'ending_image',   'dest': 'endimg_busy'    },
            {'trigger': 'retry_endimg',     'source': 'endimg_busy',    'dest': 'ending_image'   },
            {'trigger': 'end_img_success',  'source': 'ending_image',   'dest': 'actimg_init_act'  },
            {'trigger': 'activate_done',    'source': 'actimg',         'dest': 'done_image'     },
            {'trigger': 'download_fail',    'source': '*',              'dest': 'done_image'     },
            {'trigger': 'reset_image',      'source': '*',              'dest': 'init_image'     },
        ]
        
        self.img_machine = HMachine(model=self, 
                                   states=OMCI_SWIMG_DOWNLOAD_STATES,
                                   transitions=OMCI_SWIMG_DOWNLOAD_TRANSITIONS,
                                   initial='init_image',
                                   queued=True,
                                   name='{}-image_download_machine'.format(self.__class__.__name__))
github pytransitions / transitions / transitions / extensions / diagrams.py View on Github external
def _get_graph(self, model, title=None, force_new=False, show_roi=False):
        if title is None:
            title = self.title
        if not hasattr(model, 'graph') or force_new:
            model.graph = NestedGraph(self).get_graph(title) if isinstance(self, HierarchicalMachine) \
                else Graph(self).get_graph(title)
            self.set_node_state(model.graph, model.state, state='active')

        return model.graph if not show_roi else self._graph_roi(model)
github pytransitions / transitions / transitions / extensions / factory.py View on Github external
An event type to be used with (subclasses of) `LockedHierarchicalMachine`
        and `LockedHierarchicalGraphMachine`.
    """
    pass


class HierarchicalGraphMachine(GraphMachine, HierarchicalMachine):
    """
        A hierarchical state machine with graph support.
    """

    transition_cls = NestedGraphTransition
    graph_cls = NestedGraph


class LockedHierarchicalMachine(LockedMachine, HierarchicalMachine):
    """
        A threadsafe hierarchical machine.
    """

    event_cls = LockedNestedEvent


class LockedGraphMachine(GraphMachine, LockedMachine):
    """
        A threadsafe machine with graph support.
    """
    pass


class LockedHierarchicalGraphMachine(GraphMachine, LockedMachine, HierarchicalMachine):
    """
github pytransitions / transitions / transitions / extensions / nesting.py View on Github external
def add_model(self, model, initial=None):
        """ Extends transitions.core.Machine.add_model by applying a custom 'to' function to
            the added model.
        """
        _super(HierarchicalMachine, self).add_model(model, initial=initial)
        models = listify(model)
        for mod in models:
            mod = self if mod == 'self' else mod
            # TODO: Remove 'mod != self' in 0.7.0
            if hasattr(mod, 'to') and mod != self:
                _LOGGER.warning("%sModel already has a 'to'-method. It will NOT "
                                "be overwritten by NestedMachine", self.name)
            else:
                to_func = partial(self.to_state, mod)
                setattr(mod, 'to', to_func)
github opencord / voltha / voltha / extensions / omci / state_machines / image_agent.py View on Github external
def __init_state_machine(self):

        #### Download Window Sub State Machine ####
        OMCI_DOWNLOAD_WINDOW_STATE = ['init_window', 'sending_sections', 'window_success', 'window_failed']
        OMCI_DOWNLOAD_WINDOW_TRANSITIONS = [
            {'trigger': 'send_sections',  'source': 'init_window',     'dest': 'sending_sections'},
            # {'trigger': 'send_section_last',  'source': 'start_section', 'dest': 'last_section'  },
            {'trigger': 'rx_ack_success', 'source': 'sending_sections', 'dest': 'window_success' },
            {'trigger': 'rx_ack_failed',  'source': 'sending_sections', 'dest': 'window_failed'  },
            # {'trigger': 'retry_window',   'source': 'window_failed', 'dest': 'start_section'  },
            {'trigger': 'reset_window',   'source': '*',               'dest': 'init_window'    }
        ]    
        self.win_machine = HMachine(model=self, 
                                    states=OMCI_DOWNLOAD_WINDOW_STATE,
                                    transitions=OMCI_DOWNLOAD_WINDOW_TRANSITIONS,
                                    initial='init_window',
                                    queued=True,
                                    name='{}-window_section_machine'.format(self.__class__.__name__))

        #### Software Activation Sub State Machine ####
        OMCI_SWIMG_ACTIVATE_STATES = ['init_act', 'activating', 'busy', 'rebooting', 'committing', 'done', 'failed']
        OMCI_SWIMG_ACTIVATE_TRANSITIONS = [
            {'trigger': 'activate', 'source': ['init_act', 'busy'], 'dest': 'activating'},
            {'trigger': 'onu_busy', 'source': 'activating', 'dest': 'busy'},
            {'trigger': 'reboot',   'source': 'activating', 'dest': 'rebooting'},
            {'trigger': 'do_commit', 'source': ['activating', 'rebooting'], 'dest': 'committing'},
            # {'trigger': 'commit_ok', 'source': 'committing', 'dest': 'done'},
            {'trigger': 'reset_actimg', 'source': ['activating', 'rebooting', 'committing', 'failed'], 'dest': 'init_act'},
            # {'trigger': 'actimg_fail', 'source': ['init_act', 'activating', 'rebooting', 'committing'], 'dest': 'failed'}
github pytransitions / transitions / transitions / extensions / factory.py View on Github external
class LockedHierarchicalGraphMachine(GraphMachine, LockedMachine, HierarchicalMachine):
    """
        A threadsafe hiearchical machine with graph support.
    """

    transition_cls = NestedGraphTransition
    event_cls = LockedNestedEvent
    graph_cls = NestedGraph


# 3d tuple (graph, nested, locked)
_CLASS_MAP = {
    (False, False, False): Machine,
    (False, False, True): LockedMachine,
    (False, True, False): HierarchicalMachine,
    (False, True, True): LockedHierarchicalMachine,
    (True, False, False): GraphMachine,
    (True, False, True): LockedGraphMachine,
    (True, True, False): HierarchicalGraphMachine,
    (True, True, True): LockedHierarchicalGraphMachine
}
github pytransitions / transitions / transitions / extensions / factory.py View on Github external
"""
        A transition type to be used with (subclasses of) `HierarchicalGraphMachine` and
        `LockedHierarchicalGraphMachine`.
    """
    pass


class LockedNestedEvent(LockedEvent, NestedEvent):
    """
        An event type to be used with (subclasses of) `LockedHierarchicalMachine`
        and `LockedHierarchicalGraphMachine`.
    """
    pass


class HierarchicalGraphMachine(GraphMachine, HierarchicalMachine):
    """
        A hierarchical state machine with graph support.
    """

    transition_cls = NestedGraphTransition
    graph_cls = NestedGraph


class LockedHierarchicalMachine(LockedMachine, HierarchicalMachine):
    """
        A threadsafe hierarchical machine.
    """

    event_cls = LockedNestedEvent