Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
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' },
{'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__))
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)
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):
"""
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)
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'}
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
}
"""
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