<s> import collections <EOL> import functools <EOL> from futurist import waiters <EOL> from taskflow import deciders as de <EOL> from taskflow . engines . action_engine . actions import retry as ra <EOL> from taskflow . engines . action_engine . actions import task as ta <EOL> from taskflow . engines . action_engine import analyzer as an <EOL> from taskflow . engines . action_engine import builder as bu <EOL> from taskflow . engines . action_engine import compiler as com <EOL> from taskflow . engines . action_engine import completer as co <EOL> from taskflow . engines . action_engine import scheduler as sched <EOL> from taskflow . engines . action_engine import scopes as sc <EOL> from taskflow . engines . action_engine import traversal as tr <EOL> from taskflow import exceptions as exc <EOL> from taskflow import states as st <EOL> from taskflow . utils import misc <EOL> from taskflow . flow import ( LINK_DECIDER , LINK_DECIDER_DEPTH ) <EOL> _EdgeDecider = collections . namedtuple ( '<STR_LIT>' , <EOL> '<STR_LIT>' ) <EOL> class Runtime ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , compilation , storage , atom_notifier , <EOL> task_executor , retry_executor , <EOL> options = None ) : <EOL> self . _atom_notifier = atom_notifier <EOL> self . _task_executor = task_executor <EOL> self . _retry_executor = retry_executor <EOL> self . _storage = storage <EOL> self . _compilation = compilation <EOL> self . _atom_cache = { } <EOL> self . _options = misc . safe_copy_dict ( options ) <EOL> def _walk_edge_deciders ( self , graph , atom ) : <EOL> """<STR_LIT>""" <EOL> predecessors_iter = graph . predecessors_iter <EOL> nodes = collections . deque ( ( u_node , atom ) <EOL> for u_node in predecessors_iter ( atom ) ) <EOL> visited = set ( ) <EOL> while nodes : <EOL> u_node , v_node = nodes . popleft ( ) <EOL> u_node_kind = graph . node [ u_node ] [ '<STR_LIT>' ] <EOL> u_v_data = graph . adj [ u_node ] [ v_node ] <EOL> try : <EOL> decider = u_v_data [ LINK_DECIDER ] <EOL> decider_depth = u_v_data . get ( LINK_DECIDER_DEPTH ) <EOL> if decider_depth is None : <EOL> decider_depth = de . Depth . ALL <EOL> yield _EdgeDecider ( u_node , u_node_kind , <EOL> decider , decider_depth ) <EOL> except KeyError : <EOL> pass <EOL> if u_node_kind == com . FLOW and u_node not in visited : <EOL> visited . add ( u_node ) <EOL> nodes . extend ( ( u_u_node , u_node ) <EOL> for u_u_node in predecessors_iter ( u_node ) ) <EOL> def compile ( self ) : <EOL> """<STR_LIT>""" <EOL> change_state_handlers = { <EOL> com . TASK : functools . partial ( self . task_action . change_state , <EOL> progress = <NUM_LIT:0.0> ) , <EOL> com . RETRY : self . retry_action . change_state , <EOL> } <EOL> schedulers = { <EOL> com . RETRY : self . retry_scheduler , <EOL> com . TASK : self . task_scheduler , <EOL> } <EOL> check_transition_handlers = { <EOL> com . TASK : st . check_task_transition , <EOL> com . RETRY : st . check_retry_transition , <EOL> } <EOL> actions = { <EOL> com . TASK : self . task_action , <EOL> com . RETRY : self . retry_action , <EOL> } <EOL> graph = self . _compilation . execution_graph <EOL> for node , node_data in graph . nodes_iter ( data = True ) : <EOL> node_kind = node_data [ '<STR_LIT>' ] <EOL> if node_kind in com . FLOWS : <EOL> continue <EOL> elif node_kind in com . ATOMS : <EOL> check_transition_handler = check_transition_handlers [ node_kind ] <EOL> change_state_handler = change_state_handlers [ node_kind ] <EOL> scheduler = schedulers [ node_kind ] <EOL> action = actions [ node_kind ] <EOL> else : <EOL> raise exc . CompilationFailure ( "<STR_LIT>" <EOL> "<STR_LIT>" % node_kind ) <EOL> metadata = { } <EOL> deciders_it = self . _walk_edge_deciders ( graph , node ) <EOL> walker = sc . ScopeWalker ( self . compilation , node , names_only = True ) <EOL> metadata [ '<STR_LIT>' ] = walker <EOL> metadata [ '<STR_LIT>' ] = check_transition_handler <EOL> metadata [ '<STR_LIT>' ] = change_state_handler <EOL> metadata [ '<STR_LIT>' ] = scheduler <EOL> metadata [ '<STR_LIT>' ] = tuple ( deciders_it ) <EOL> metadata [ '<STR_LIT:action>' ] = action <EOL> self . _atom_cache [ node . name ] = metadata <EOL> @ property <EOL> def compilation ( self ) : <EOL> return self . _compilation <EOL> @ property <EOL> def storage ( self ) : <EOL> return self . _storage <EOL> @ property <EOL> def options ( self ) : <EOL> return self . _options <EOL> @ misc . cachedproperty <EOL> def analyzer ( self ) : <EOL> return an . Analyzer ( self ) <EOL> @ misc . cachedproperty <EOL> def builder ( self ) : <EOL> return bu . MachineBuilder ( self , waiters . wait_for_any ) <EOL> @ misc . cachedproperty <EOL> def completer ( self ) : <EOL> return co . Completer ( self ) <EOL> @ misc . cachedproperty <EOL> def scheduler ( self ) : <EOL> return sched . Scheduler ( self ) <EOL> @ misc . cachedproperty <EOL> def task_scheduler ( self ) : <EOL> return sched . TaskScheduler ( self ) <EOL> @ misc . cachedproperty <EOL> def retry_scheduler ( self ) : <EOL> return sched . RetryScheduler ( self ) <EOL> @ misc . cachedproperty <EOL> def retry_action ( self ) : <EOL> return ra . RetryAction ( self . _storage , <EOL> self . _atom_notifier , <EOL> self . _retry_executor ) <EOL> @ misc . cachedproperty <EOL> def task_action ( self ) : <EOL> return ta . TaskAction ( self . _storage , <EOL> self . _atom_notifier , <EOL> self . _task_executor ) <EOL> def _fetch_atom_metadata_entry ( self , atom_name , metadata_key ) : <EOL> return self . _atom_cache [ atom_name ] [ metadata_key ] <EOL> def check_atom_transition ( self , atom , current_state , target_state ) : <EOL> """<STR_LIT>""" <EOL> check_transition_handler = self . _fetch_atom_metadata_entry ( <EOL> atom . name , '<STR_LIT>' ) <EOL> return check_transition_handler ( current_state , target_state ) <EOL> def fetch_edge_deciders ( self , atom ) : <EOL> """<STR_LIT>""" <EOL> return self . _fetch_atom_metadata_entry ( atom . name , '<STR_LIT>' ) <EOL> def fetch_scheduler ( self , atom ) : <EOL> """<STR_LIT>""" <EOL> return self . _fetch_atom_metadata_entry ( atom . name , '<STR_LIT>' ) <EOL> def fetch_action ( self , atom ) : <EOL> """<STR_LIT>""" <EOL> metadata = self . _atom_cache [ atom . name ] <EOL> return metadata [ '<STR_LIT:action>' ] <EOL> def fetch_scopes_for ( self , atom_name ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> return self . _fetch_atom_metadata_entry ( atom_name , '<STR_LIT>' ) <EOL> except KeyError : <EOL> return None <EOL> def reset_atoms ( self , atoms , state = st . PENDING , intention = st . EXECUTE ) : <EOL> """<STR_LIT>""" <EOL> tweaked = [ ] <EOL> for atom in atoms : <EOL> if state or intention : <EOL> tweaked . append ( ( atom , state , intention ) ) <EOL> if state : <EOL> change_state_handler = self . _fetch_atom_metadata_entry ( <EOL> atom . name , '<STR_LIT>' ) <EOL> change_state_handler ( atom , state ) <EOL> if intention : <EOL> self . storage . set_atom_intention ( atom . name , intention ) <EOL> return tweaked <EOL> def reset_all ( self , state = st . PENDING , intention = st . EXECUTE ) : <EOL> """<STR_LIT>""" <EOL> return self . reset_atoms ( self . analyzer . iterate_nodes ( com . ATOMS ) , <EOL> state = state , intention = intention ) <EOL> def reset_subgraph ( self , atom , state = st . PENDING , intention = st . EXECUTE ) : <EOL> """<STR_LIT>""" <EOL> execution_graph = self . _compilation . execution_graph <EOL> atoms_it = tr . depth_first_iterate ( execution_graph , atom , <EOL> tr . Direction . FORWARD ) <EOL> return self . reset_atoms ( atoms_it , state = state , intention = intention ) <EOL> def retry_subflow ( self , retry ) : <EOL> """<STR_LIT>""" <EOL> tweaked = self . reset_atoms ( [ retry ] , state = None , intention = st . EXECUTE ) <EOL> tweaked . extend ( self . reset_subgraph ( retry ) ) <EOL> return tweaked </s>
