<s> from __future__ import division <EOL> import base64 <EOL> import copy <EOL> import gc <EOL> import cPickle as pickle <EOL> from vistrails . core . common import InstanceObject , VistrailsInternalError <EOL> from vistrails . core . data_structures . bijectivedict import Bidict <EOL> from vistrails . core import debug <EOL> import vistrails . core . interpreter . base <EOL> from vistrails . core . interpreter . base import AbortExecution <EOL> from vistrails . core . log . controller import DummyLogController <EOL> from vistrails . core . modules . basic_modules import identifier as basic_pkg , Generator <EOL> from vistrails . core . modules . module_registry import get_module_registry <EOL> from vistrails . core . modules . vistrails_module import ModuleBreakpoint , ModuleConnector , ModuleError , ModuleErrors , ModuleHadError , ModuleSuspended , ModuleWasSuspended <EOL> from vistrails . core . utils import DummyView <EOL> import vistrails . core . system <EOL> import vistrails . core . vistrail . pipeline <EOL> class ViewUpdatingLogController ( object ) : <EOL> class Loop ( object ) : <EOL> def __init__ ( self , logger , view ) : <EOL> self . log = logger <EOL> self . view = view <EOL> def end_loop_execution ( self ) : <EOL> self . log . finish_loop_execution ( ) <EOL> def begin_iteration ( self , looped_obj , iteration ) : <EOL> self . log . start_iteration ( looped_obj , iteration ) <EOL> def end_iteration ( self , looped_obj ) : <EOL> self . log . finish_iteration ( looped_obj ) <EOL> def __init__ ( self , logger , view , remap_id , ids , <EOL> module_executed_hook = [ ] ) : <EOL> self . log = logger <EOL> self . view = view <EOL> self . remap_id = remap_id <EOL> self . ids = set ( ids ) <EOL> self . nb_modules = len ( self . ids ) <EOL> self . module_executed_hook = module_executed_hook <EOL> self . errors = { } <EOL> self . executed = { } <EOL> self . suspended = { } <EOL> self . cached = { } <EOL> def signalSuccess ( self , obj ) : <EOL> self . executed [ obj . id ] = True <EOL> for callable_ in self . module_executed_hook : <EOL> callable_ ( obj . id ) <EOL> def signalError ( self , obj , error ) : <EOL> self . errors [ obj . id ] = error <EOL> def begin_update ( self , obj ) : <EOL> i = self . remap_id ( obj . id ) <EOL> self . view . set_module_active ( i ) <EOL> def begin_compute ( self , obj ) : <EOL> i = self . remap_id ( obj . id ) <EOL> self . view . set_module_computing ( i ) <EOL> reg = get_module_registry ( ) <EOL> module_name = reg . get_descriptor ( obj . __class__ ) . name <EOL> self . log . start_execution ( obj , i , module_name ) <EOL> def update_progress ( self , obj , progress = <NUM_LIT:0.0> ) : <EOL> i = self . remap_id ( obj . id ) <EOL> self . view . set_module_progress ( i , progress ) <EOL> def begin_loop_execution ( self , obj , total_iterations = None ) : <EOL> return ViewUpdatingLogController . Loop ( <EOL> self . log . start_loop_execution ( obj , total_iterations ) , <EOL> self . view ) <EOL> def _handle_suspended ( self , obj , error ) : <EOL> """<STR_LIT>""" <EOL> jm = obj . job_monitor ( ) <EOL> reg = get_module_registry ( ) <EOL> name = reg . get_descriptor ( obj . __class__ ) . name <EOL> iteration = self . log . get_iteration_from_module ( obj ) <EOL> if iteration is not None : <EOL> name += '<STR_LIT>' % iteration <EOL> error . name = name <EOL> jm . addParent ( error ) <EOL> def end_update ( self , obj , error = None , errorTrace = None , <EOL> was_suspended = False ) : <EOL> try : <EOL> i = self . remap_id ( obj . id ) <EOL> except KeyError : <EOL> return <EOL> if was_suspended : <EOL> self . _handle_suspended ( obj , error ) <EOL> self . suspended [ obj . id ] = error <EOL> self . view . set_module_suspended ( i , error ) <EOL> if error . children : <EOL> for child in error . children : <EOL> self . end_update ( child . module , child , was_suspended = True ) <EOL> elif error is None : <EOL> self . view . set_module_success ( i ) <EOL> else : <EOL> self . view . set_module_error ( i , error . msg , error . errorTrace ) <EOL> if i in self . ids : <EOL> self . ids . remove ( i ) <EOL> self . view . set_execution_progress ( <EOL> <NUM_LIT:1.0> - ( ( len ( self . ids ) + len ( Generator . generators ) ) * <NUM_LIT:1.0> / <EOL> ( self . nb_modules + len ( Generator . generators ) ) ) ) <EOL> msg = '<STR_LIT>' if error is None else error . msg <EOL> self . log . finish_execution ( obj , msg , errorTrace , <EOL> was_suspended ) <EOL> def update_cached ( self , obj ) : <EOL> self . cached [ obj . id ] = True <EOL> i = self . remap_id ( obj . id ) <EOL> reg = get_module_registry ( ) <EOL> module_name = reg . get_descriptor ( obj . __class__ ) . name <EOL> self . log . start_execution ( obj , i , module_name , <EOL> cached = <NUM_LIT:1> ) <EOL> self . view . set_module_not_executed ( i ) <EOL> self . log . finish_execution ( obj , '<STR_LIT>' ) <EOL> def set_computing ( self , obj ) : <EOL> i = self . remap_id ( obj . id ) <EOL> self . view . set_module_computing ( i ) <EOL> def annotate ( self , obj , d ) : <EOL> self . log . insert_module_annotations ( obj , d ) <EOL> def add_machine ( self , machine ) : <EOL> return self . log . add_machine ( machine ) <EOL> def add_exec ( self , exec_ ) : <EOL> return self . log . add_exec ( exec_ ) <EOL> Variant_desc = None <EOL> InputPort_desc = None <EOL> class CachedInterpreter ( vistrails . core . interpreter . base . BaseInterpreter ) : <EOL> def __init__ ( self ) : <EOL> vistrails . core . interpreter . base . BaseInterpreter . __init__ ( self ) <EOL> self . debugger = None <EOL> self . create ( ) <EOL> def create ( self ) : <EOL> from vistrails . core . modules . module_utils import FilePool <EOL> self . _file_pool = FilePool ( ) <EOL> self . _persistent_pipeline = vistrails . core . vistrail . pipeline . Pipeline ( ) <EOL> self . _objects = { } <EOL> self . filePool = self . _file_pool <EOL> self . _streams = [ ] <EOL> def clear ( self ) : <EOL> self . _file_pool . cleanup ( ) <EOL> self . _persistent_pipeline . clear ( ) <EOL> for obj in self . _objects . itervalues ( ) : <EOL> obj . clear ( ) <EOL> self . _objects = { } <EOL> def __del__ ( self ) : <EOL> self . clear ( ) <EOL> def clean_modules ( self , modules_to_clean ) : <EOL> """<STR_LIT>""" <EOL> if not modules_to_clean : <EOL> return <EOL> g = self . _persistent_pipeline . graph <EOL> modules_to_clean = ( set ( modules_to_clean ) & <EOL> set ( self . _persistent_pipeline . modules . iterkeys ( ) ) ) <EOL> dependencies = g . vertices_topological_sort ( modules_to_clean ) <EOL> for v in dependencies : <EOL> self . _persistent_pipeline . delete_module ( v ) <EOL> del self . _objects [ v ] <EOL> def clean_non_cacheable_modules ( self ) : <EOL> """<STR_LIT>""" <EOL> non_cacheable_modules = [ i for <EOL> ( i , mod ) in self . _objects . iteritems ( ) <EOL> if not mod . is_cacheable ( ) ] <EOL> self . clean_modules ( non_cacheable_modules ) <EOL> def _clear_package ( self , identifier ) : <EOL> """<STR_LIT>""" <EOL> modules = [ mod . id <EOL> for mod in self . _persistent_pipeline . module_list <EOL> if mod . module_descriptor . identifier == identifier ] <EOL> self . clean_modules ( modules ) <EOL> def make_connection ( self , conn , src , dst ) : <EOL> """<STR_LIT>""" <EOL> iport = conn . destination . name <EOL> oport = conn . source . name <EOL> src . enable_output_port ( oport ) <EOL> src . load_type_check_descs ( ) <EOL> if isinstance ( src , src . InputPort_desc . module ) : <EOL> typecheck = [ False ] <EOL> else : <EOL> typecheck = src . get_type_checks ( conn . source . spec ) <EOL> dst . set_input_port ( iport , <EOL> ModuleConnector ( src , oport , conn . source . spec , <EOL> typecheck ) ) <EOL> def setup_pipeline ( self , pipeline , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> def fetch ( name , default ) : <EOL> return kwargs . pop ( name , default ) <EOL> controller = fetch ( '<STR_LIT>' , None ) <EOL> locator = fetch ( '<STR_LIT>' , None ) <EOL> current_version = fetch ( '<STR_LIT>' , None ) <EOL> view = fetch ( '<STR_LIT>' , DummyView ( ) ) <EOL> vistrail_variables = fetch ( '<STR_LIT>' , None ) <EOL> aliases = fetch ( '<STR_LIT>' , None ) <EOL> params = fetch ( '<STR_LIT>' , None ) <EOL> extra_info = fetch ( '<STR_LIT>' , None ) <EOL> logger = fetch ( '<STR_LIT>' , DummyLogController ) <EOL> sinks = fetch ( '<STR_LIT>' , None ) <EOL> reason = fetch ( '<STR_LIT>' , None ) <EOL> actions = fetch ( '<STR_LIT>' , None ) <EOL> done_summon_hooks = fetch ( '<STR_LIT>' , [ ] ) <EOL> module_executed_hook = fetch ( '<STR_LIT>' , [ ] ) <EOL> stop_on_error = fetch ( '<STR_LIT>' , True ) <EOL> parent_exec = fetch ( '<STR_LIT>' , None ) <EOL> job_monitor = fetch ( '<STR_LIT>' , None ) <EOL> reg = get_module_registry ( ) <EOL> if len ( kwargs ) > <NUM_LIT:0> : <EOL> raise VistrailsInternalError ( '<STR_LIT>' <EOL> '<STR_LIT>' % kwargs ) <EOL> def create_null ( ) : <EOL> """<STR_LIT>""" <EOL> getter = reg . get_descriptor_by_name <EOL> descriptor = getter ( basic_pkg , '<STR_LIT>' ) <EOL> return descriptor . module ( ) <EOL> def create_constant ( param , module ) : <EOL> """<STR_LIT>""" <EOL> getter = reg . get_descriptor_by_name <EOL> desc = getter ( param . identifier , param . type , param . namespace ) <EOL> constant = desc . module ( ) <EOL> constant . id = module . id <EOL> if param . strValue != '<STR_LIT>' : <EOL> constant . setValue ( param . strValue ) <EOL> else : <EOL> constant . setValue ( <EOL> constant . translate_to_string ( constant . default_value ) ) <EOL> return constant <EOL> to_delete = [ ] <EOL> errors = { } <EOL> if controller is not None : <EOL> controller . validate ( pipeline ) <EOL> else : <EOL> pipeline . validate ( ) <EOL> self . resolve_aliases ( pipeline , aliases ) <EOL> if vistrail_variables : <EOL> self . resolve_variables ( vistrail_variables , pipeline ) <EOL> self . update_params ( pipeline , params ) <EOL> ( tmp_to_persistent_module_map , <EOL> conn_map , <EOL> module_added_set , <EOL> conn_added_set ) = self . add_to_persistent_pipeline ( pipeline ) <EOL> for i in module_added_set : <EOL> persistent_id = tmp_to_persistent_module_map [ i ] <EOL> module = self . _persistent_pipeline . modules [ persistent_id ] <EOL> obj = self . _objects [ persistent_id ] = module . summon ( ) <EOL> obj . interpreter = self <EOL> obj . id = persistent_id <EOL> obj . signature = module . _signature <EOL> if module . has_annotation_with_key ( '<STR_LIT>' ) : <EOL> annotate_output = module . get_annotation_by_key ( '<STR_LIT>' ) <EOL> if annotate_output : <EOL> obj . annotate_output = True <EOL> for f in module . functions : <EOL> connector = None <EOL> if len ( f . params ) == <NUM_LIT:0> : <EOL> connector = ModuleConnector ( create_null ( ) , '<STR_LIT:value>' , <EOL> f . get_spec ( '<STR_LIT>' ) ) <EOL> elif len ( f . params ) == <NUM_LIT:1> : <EOL> p = f . params [ <NUM_LIT:0> ] <EOL> try : <EOL> constant = create_constant ( p , module ) <EOL> connector = ModuleConnector ( constant , '<STR_LIT:value>' , <EOL> f . get_spec ( '<STR_LIT>' ) ) <EOL> except Exception , e : <EOL> debug . unexpected_exception ( e ) <EOL> err = ModuleError ( <EOL> module , <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" % ( <EOL> p . strValue , <EOL> debug . format_exception ( e ) ) ) <EOL> errors [ i ] = err <EOL> to_delete . append ( obj . id ) <EOL> else : <EOL> tupleModule = vistrails . core . interpreter . base . InternalTuple ( ) <EOL> tupleModule . length = len ( f . params ) <EOL> for ( j , p ) in enumerate ( f . params ) : <EOL> try : <EOL> constant = create_constant ( p , module ) <EOL> constant . update ( ) <EOL> connector = ModuleConnector ( constant , '<STR_LIT:value>' , <EOL> f . get_spec ( '<STR_LIT>' ) ) <EOL> tupleModule . set_input_port ( j , connector ) <EOL> except Exception , e : <EOL> debug . unexpected_exception ( e ) <EOL> err = ModuleError ( <EOL> module , <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" % ( <EOL> p . strValue , <EOL> debug . format_exception ( e ) ) ) <EOL> errors [ i ] = err <EOL> to_delete . append ( obj . id ) <EOL> connector = ModuleConnector ( tupleModule , '<STR_LIT:value>' , <EOL> f . get_spec ( '<STR_LIT>' ) ) <EOL> if connector : <EOL> obj . set_input_port ( f . name , connector , is_method = True ) <EOL> for i in conn_added_set : <EOL> persistent_id = conn_map [ i ] <EOL> conn = self . _persistent_pipeline . connections [ persistent_id ] <EOL> src = self . _objects [ conn . sourceId ] <EOL> dst = self . _objects [ conn . destinationId ] <EOL> self . make_connection ( conn , src , dst ) <EOL> if self . done_summon_hook : <EOL> self . done_summon_hook ( self . _persistent_pipeline , self . _objects ) <EOL> for callable_ in done_summon_hooks : <EOL> callable_ ( self . _persistent_pipeline , self . _objects ) <EOL> tmp_id_to_module_map = { } <EOL> for i , j in tmp_to_persistent_module_map . iteritems ( ) : <EOL> tmp_id_to_module_map [ i ] = self . _objects [ j ] <EOL> return ( tmp_id_to_module_map , tmp_to_persistent_module_map . inverse , <EOL> module_added_set , conn_added_set , to_delete , errors ) <EOL> def execute_pipeline ( self , pipeline , tmp_id_to_module_map , <EOL> persistent_to_tmp_id_map , ** kwargs ) : <EOL> def fetch ( name , default ) : <EOL> return kwargs . pop ( name , default ) <EOL> controller = fetch ( '<STR_LIT>' , None ) <EOL> locator = fetch ( '<STR_LIT>' , None ) <EOL> current_version = fetch ( '<STR_LIT>' , None ) <EOL> view = fetch ( '<STR_LIT>' , DummyView ( ) ) <EOL> vistrail_variables = fetch ( '<STR_LIT>' , None ) <EOL> aliases = fetch ( '<STR_LIT>' , None ) <EOL> params = fetch ( '<STR_LIT>' , None ) <EOL> extra_info = fetch ( '<STR_LIT>' , None ) <EOL> logger = fetch ( '<STR_LIT>' , DummyLogController ) <EOL> sinks = fetch ( '<STR_LIT>' , None ) <EOL> reason = fetch ( '<STR_LIT>' , None ) <EOL> actions = fetch ( '<STR_LIT>' , None ) <EOL> module_executed_hook = fetch ( '<STR_LIT>' , [ ] ) <EOL> done_summon_hooks = fetch ( '<STR_LIT>' , [ ] ) <EOL> clean_pipeline = fetch ( '<STR_LIT>' , False ) <EOL> stop_on_error = fetch ( '<STR_LIT>' , True ) <EOL> parent_exec = fetch ( '<STR_LIT>' , None ) <EOL> job_monitor = fetch ( '<STR_LIT>' , None ) <EOL> if len ( kwargs ) > <NUM_LIT:0> : <EOL> raise VistrailsInternalError ( '<STR_LIT>' <EOL> '<STR_LIT>' % kwargs ) <EOL> def get_remapped_id ( id ) : <EOL> return persistent_to_tmp_id_map [ id ] <EOL> logging_obj = ViewUpdatingLogController ( <EOL> logger = logger , <EOL> view = view , <EOL> remap_id = get_remapped_id , <EOL> ids = pipeline . modules . keys ( ) , <EOL> module_executed_hook = module_executed_hook ) <EOL> parameter_changes = [ ] <EOL> def change_parameter ( obj , name , value ) : <EOL> parameter_changes . append ( ( get_remapped_id ( obj . id ) , <EOL> name , value ) ) <EOL> def make_change_parameter ( obj ) : <EOL> return lambda * args : change_parameter ( obj , * args ) <EOL> for i , obj in tmp_id_to_module_map . iteritems ( ) : <EOL> obj . in_pipeline = True <EOL> obj . logging = logging_obj <EOL> obj . change_parameter = make_change_parameter ( obj ) <EOL> obj . moduleInfo [ '<STR_LIT>' ] = locator <EOL> obj . moduleInfo [ '<STR_LIT:version>' ] = current_version <EOL> obj . moduleInfo [ '<STR_LIT>' ] = i <EOL> obj . moduleInfo [ '<STR_LIT>' ] = pipeline <EOL> obj . moduleInfo [ '<STR_LIT>' ] = controller <EOL> if controller : <EOL> obj . moduleInfo [ '<STR_LIT>' ] = controller . jobMonitor <EOL> else : <EOL> obj . moduleInfo [ '<STR_LIT>' ] = job_monitor <EOL> if extra_info is not None : <EOL> obj . moduleInfo [ '<STR_LIT>' ] = extra_info <EOL> if reason is not None : <EOL> obj . moduleInfo [ '<STR_LIT>' ] = reason <EOL> if actions is not None : <EOL> obj . moduleInfo [ '<STR_LIT>' ] = actions <EOL> if sinks is not None : <EOL> persistent_sinks = [ tmp_id_to_module_map [ sink ] <EOL> for sink in sinks <EOL> if sink in tmp_id_to_module_map ] <EOL> else : <EOL> persistent_sinks = [ tmp_id_to_module_map [ sink ] <EOL> for sink in pipeline . graph . sinks ( ) ] <EOL> self . _streams . append ( Generator . generators ) <EOL> Generator . generators = [ ] <EOL> for obj in persistent_sinks : <EOL> abort = False <EOL> try : <EOL> obj . update ( ) <EOL> continue <EOL> except ModuleWasSuspended : <EOL> continue <EOL> except ModuleHadError : <EOL> pass <EOL> except AbortExecution : <EOL> break <EOL> except ModuleSuspended , ms : <EOL> ms . module . logging . end_update ( ms . module , ms , <EOL> was_suspended = True ) <EOL> continue <EOL> except ModuleErrors , mes : <EOL> for me in mes . module_errors : <EOL> me . module . logging . end_update ( me . module , me ) <EOL> logging_obj . signalError ( me . module , me ) <EOL> abort = abort or me . abort <EOL> except ModuleError , me : <EOL> me . module . logging . end_update ( me . module , me , me . errorTrace ) <EOL> logging_obj . signalError ( me . module , me ) <EOL> abort = me . abort <EOL> except ModuleBreakpoint , mb : <EOL> mb . module . logging . end_update ( mb . module ) <EOL> logging_obj . signalError ( mb . module , mb ) <EOL> abort = True <EOL> if stop_on_error or abort : <EOL> break <EOL> if not logging_obj . errors and not logging_obj . suspended and Generator . generators : <EOL> result = True <EOL> abort = False <EOL> while result is not None : <EOL> try : <EOL> for m in Generator . generators : <EOL> result = m . generator . next ( ) <EOL> continue <EOL> except AbortExecution : <EOL> break <EOL> except ModuleErrors , mes : <EOL> for me in mes . module_errors : <EOL> me . module . logging . end_update ( me . module , me ) <EOL> logging_obj . signalError ( me . module , me ) <EOL> abort = abort or me . abort <EOL> except ModuleError , me : <EOL> me . module . logging . end_update ( me . module , me , me . errorTrace ) <EOL> logging_obj . signalError ( me . module , me ) <EOL> abort = me . abort <EOL> except ModuleBreakpoint , mb : <EOL> mb . module . logging . end_update ( mb . module ) <EOL> logging_obj . signalError ( mb . module , mb ) <EOL> abort = True <EOL> except Exception , e : <EOL> debug . unexpected_exception ( e ) <EOL> debug . critical ( "<STR_LIT>" % e , <EOL> debug . format_exc ( ) ) <EOL> abort = True <EOL> if stop_on_error or abort : <EOL> break <EOL> Generator . generators = self . _streams . pop ( ) <EOL> if self . done_update_hook : <EOL> self . done_update_hook ( self . _persistent_pipeline , self . _objects ) <EOL> objs = { } <EOL> errs = { } <EOL> execs = { } <EOL> suspends = { } <EOL> caches = { } <EOL> to_delete = [ ] <EOL> for ( tmp_id , obj ) in tmp_id_to_module_map . iteritems ( ) : <EOL> if clean_pipeline : <EOL> to_delete . append ( obj . id ) <EOL> objs [ tmp_id ] = obj <EOL> if obj . id in logging_obj . errors : <EOL> errs [ tmp_id ] = logging_obj . errors [ obj . id ] <EOL> if not clean_pipeline : <EOL> to_delete . append ( obj . id ) <EOL> executed = False <EOL> if obj . id in logging_obj . executed : <EOL> execs [ tmp_id ] = logging_obj . executed [ obj . id ] <EOL> executed = True <EOL> if obj . id in logging_obj . suspended : <EOL> suspends [ tmp_id ] = logging_obj . suspended [ obj . id ] <EOL> if not clean_pipeline : <EOL> to_delete . append ( obj . id ) <EOL> executed = True <EOL> if obj . id in logging_obj . cached : <EOL> caches [ tmp_id ] = logging_obj . cached [ obj . id ] <EOL> executed = True <EOL> if not executed : <EOL> execs [ tmp_id ] = False <EOL> return ( to_delete , objs , errs , execs , suspends , caches , parameter_changes ) <EOL> def finalize_pipeline ( self , pipeline , to_delete , objs , errs , execs , <EOL> suspended , cached , ** kwargs ) : <EOL> def fetch ( name , default ) : <EOL> return kwargs . pop ( name , default ) <EOL> reset_computed = fetch ( '<STR_LIT>' , True ) <EOL> view = fetch ( '<STR_LIT>' , None ) <EOL> self . clean_modules ( to_delete ) <EOL> if view is not None : <EOL> for i in objs : <EOL> if i in errs : <EOL> view . set_module_error ( i , errs [ i ] . msg , errs [ i ] . errorTrace ) <EOL> elif i in suspended and suspended [ i ] : <EOL> view . set_module_suspended ( i , suspended [ i ] ) <EOL> elif i in execs and execs [ i ] : <EOL> view . set_module_success ( i ) <EOL> elif i in cached and cached [ i ] : <EOL> view . set_module_not_executed ( i ) <EOL> else : <EOL> view . set_module_persistent ( i ) <EOL> if reset_computed : <EOL> for module in self . _objects . itervalues ( ) : <EOL> module . computed = False <EOL> def execute ( self , pipeline , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> new_kwargs = { } <EOL> def fetch ( name , default ) : <EOL> new_kwargs [ name ] = r = kwargs . pop ( name , default ) <EOL> return r <EOL> controller = fetch ( '<STR_LIT>' , None ) <EOL> locator = fetch ( '<STR_LIT>' , None ) <EOL> current_version = fetch ( '<STR_LIT>' , None ) <EOL> view = fetch ( '<STR_LIT>' , DummyView ( ) ) <EOL> vistrail_variables = fetch ( '<STR_LIT>' , None ) <EOL> aliases = fetch ( '<STR_LIT>' , None ) <EOL> params = fetch ( '<STR_LIT>' , None ) <EOL> extra_info = fetch ( '<STR_LIT>' , None ) <EOL> logger = fetch ( '<STR_LIT>' , DummyLogController ) <EOL> sinks = fetch ( '<STR_LIT>' , None ) <EOL> reason = fetch ( '<STR_LIT>' , None ) <EOL> actions = fetch ( '<STR_LIT>' , None ) <EOL> done_summon_hooks = fetch ( '<STR_LIT>' , [ ] ) <EOL> module_executed_hook = fetch ( '<STR_LIT>' , [ ] ) <EOL> stop_on_error = fetch ( '<STR_LIT>' , True ) <EOL> parent_exec = fetch ( '<STR_LIT>' , None ) <EOL> job_monitor = fetch ( '<STR_LIT>' , None ) <EOL> if len ( kwargs ) > <NUM_LIT:0> : <EOL> raise VistrailsInternalError ( '<STR_LIT>' <EOL> '<STR_LIT>' % kwargs ) <EOL> self . clean_non_cacheable_modules ( ) <EOL> if controller is not None : <EOL> vistrail = controller . vistrail <EOL> else : <EOL> vistrail = None <EOL> logger = logger . start_workflow_execution ( <EOL> parent_exec , <EOL> vistrail , pipeline , current_version ) <EOL> new_kwargs [ '<STR_LIT>' ] = logger <EOL> self . annotate_workflow_execution ( logger , reason , aliases , params ) <EOL> res = self . setup_pipeline ( pipeline , ** new_kwargs ) <EOL> modules_added = res [ <NUM_LIT:2> ] <EOL> conns_added = res [ <NUM_LIT:3> ] <EOL> to_delete = res [ <NUM_LIT:4> ] <EOL> errors = res [ <NUM_LIT:5> ] <EOL> if len ( errors ) == <NUM_LIT:0> : <EOL> res = self . execute_pipeline ( pipeline , * ( res [ : <NUM_LIT:2> ] ) , ** new_kwargs ) <EOL> else : <EOL> res = ( to_delete , res [ <NUM_LIT:0> ] , errors , { } , { } , { } , [ ] ) <EOL> for ( i , error ) in errors . iteritems ( ) : <EOL> view . set_module_error ( i , error . msg , error . errorTrace ) <EOL> self . finalize_pipeline ( pipeline , * ( res [ : - <NUM_LIT:1> ] ) , ** new_kwargs ) <EOL> result = InstanceObject ( objects = res [ <NUM_LIT:1> ] , <EOL> errors = res [ <NUM_LIT:2> ] , <EOL> executed = res [ <NUM_LIT:3> ] , <EOL> suspended = res [ <NUM_LIT:4> ] , <EOL> parameter_changes = res [ <NUM_LIT:6> ] , <EOL> modules_added = modules_added , <EOL> conns_added = conns_added ) <EOL> logger . finish_workflow_execution ( result . errors , suspended = result . suspended ) <EOL> return result <EOL> def annotate_workflow_execution ( self , logger , reason , aliases , params ) : <EOL> """<STR_LIT>""" <EOL> d = { } <EOL> d [ "<STR_LIT>" ] = reason <EOL> if aliases is not None and isinstance ( aliases , dict ) : <EOL> d [ "<STR_LIT>" ] = pickle . dumps ( aliases ) <EOL> if params is not None and isinstance ( params , list ) : <EOL> d [ "<STR_LIT>" ] = pickle . dumps ( params ) <EOL> logger . insert_workflow_exec_annotations ( d ) <EOL> def add_to_persistent_pipeline ( self , pipeline ) : <EOL> """<STR_LIT>""" <EOL> module_id_map = Bidict ( ) <EOL> connection_id_map = Bidict ( ) <EOL> modules_added = set ( ) <EOL> connections_added = set ( ) <EOL> pipeline . refresh_signatures ( ) <EOL> verts = pipeline . graph . vertices_topological_sort ( ) <EOL> for new_module_id in verts : <EOL> new_sig = pipeline . subpipeline_signature ( new_module_id ) <EOL> if not self . _persistent_pipeline . has_subpipeline_signature ( new_sig ) : <EOL> persistent_module = copy . copy ( pipeline . modules [ new_module_id ] ) <EOL> persistent_id = self . _persistent_pipeline . fresh_module_id ( ) <EOL> persistent_module . id = persistent_id <EOL> self . _persistent_pipeline . add_module ( persistent_module ) <EOL> self . _persistent_pipeline . modules [ persistent_id ] . _signature = base64 . b16encode ( new_sig ) . lower ( ) <EOL> module_id_map [ new_module_id ] = persistent_id <EOL> modules_added . add ( new_module_id ) <EOL> else : <EOL> i = self . _persistent_pipeline . subpipeline_id_from_signature ( new_sig ) <EOL> module_id_map [ new_module_id ] = i <EOL> for connection in pipeline . connections . itervalues ( ) : <EOL> new_sig = pipeline . connection_signature ( connection . id ) <EOL> if not self . _persistent_pipeline . has_connection_signature ( new_sig ) : <EOL> persistent_connection = copy . copy ( connection ) <EOL> persistent_id = self . _persistent_pipeline . fresh_connection_id ( ) <EOL> persistent_connection . id = persistent_id <EOL> persistent_connection . sourceId = module_id_map [ <EOL> connection . sourceId ] <EOL> persistent_connection . destinationId = module_id_map [ <EOL> connection . destinationId ] <EOL> self . _persistent_pipeline . add_connection ( persistent_connection ) <EOL> connection_id_map [ connection . id ] = persistent_id <EOL> connections_added . add ( connection . id ) <EOL> else : <EOL> i = self . _persistent_pipeline . connection_id_from_signature ( new_sig ) <EOL> connection_id_map [ connection . id ] = i <EOL> self . _persistent_pipeline . compute_signatures ( ) <EOL> return ( module_id_map , connection_id_map , <EOL> modules_added , connections_added ) <EOL> def find_persistent_entities ( self , pipeline ) : <EOL> """<STR_LIT>""" <EOL> persistent_p = self . _persistent_pipeline <EOL> object_map = { } <EOL> module_id_map = { } <EOL> connection_id_map = { } <EOL> pipeline . refresh_signatures ( ) <EOL> verts = pipeline . graph . vertices_topological_sort ( ) <EOL> for module_id in verts : <EOL> sig = pipeline . subpipeline_signature ( module_id ) <EOL> if persistent_p . has_subpipeline_signature ( sig ) : <EOL> i = persistent_p . subpipeline_id_from_signature ( sig ) <EOL> module_id_map [ module_id ] = i <EOL> object_map [ module_id ] = self . _objects [ i ] <EOL> else : <EOL> module_id_map [ module_id ] = None <EOL> object_map [ module_id ] = None <EOL> for connection in pipeline . connections . itervalues ( ) : <EOL> sig = pipeline . connection_signature ( connection . id ) <EOL> if persistent_p . has_connection_signature ( sig ) : <EOL> connection_id_map [ connection . id ] = persistent_p . connection_id_from_signature ( sig ) <EOL> else : <EOL> connection_id_map [ connection . id ] = None <EOL> return ( object_map , module_id_map , connection_id_map ) <EOL> __instance = None <EOL> @ staticmethod <EOL> def get ( ) : <EOL> if not CachedInterpreter . __instance : <EOL> CachedInterpreter . __instance = CachedInterpreter ( ) <EOL> return CachedInterpreter . __instance <EOL> @ staticmethod <EOL> def cleanup ( ) : <EOL> if CachedInterpreter . __instance : <EOL> CachedInterpreter . __instance . clear ( ) <EOL> objs = gc . collect ( ) <EOL> @ staticmethod <EOL> def flush ( ) : <EOL> if CachedInterpreter . __instance : <EOL> CachedInterpreter . __instance . clear ( ) <EOL> CachedInterpreter . __instance . create ( ) <EOL> objs = gc . collect ( ) <EOL> @ staticmethod <EOL> def clear_package ( identifier ) : <EOL> if CachedInterpreter . __instance : <EOL> CachedInterpreter . __instance . _clear_package ( identifier ) <EOL> import unittest <EOL> class TestCachedInterpreter ( unittest . TestCase ) : <EOL> def test_cache ( self ) : <EOL> from vistrails . core . modules . basic_modules import StandardOutput <EOL> old_compute = StandardOutput . compute <EOL> StandardOutput . compute = lambda s : None <EOL> try : <EOL> from vistrails . core . db . locator import XMLFileLocator <EOL> from vistrails . core . vistrail . controller import VistrailController <EOL> from vistrails . core . db . io import load_vistrail <EOL> """<STR_LIT>""" <EOL> locator = XMLFileLocator ( vistrails . core . system . vistrails_root_directory ( ) + <EOL> '<STR_LIT>' ) <EOL> ( v , abstractions , thumbnails , mashups ) = load_vistrail ( locator ) <EOL> controller = VistrailController ( v , locator , abstractions , <EOL> thumbnails , mashups ) <EOL> p1 = v . getPipeline ( '<STR_LIT>' ) <EOL> n = v . get_version_number ( '<STR_LIT>' ) <EOL> controller . change_selected_version ( n ) <EOL> controller . flush_delayed_actions ( ) <EOL> p1 = controller . current_pipeline <EOL> view = DummyView ( ) <EOL> interpreter = CachedInterpreter . get ( ) <EOL> result = interpreter . execute ( p1 , <EOL> locator = v , <EOL> current_version = n , <EOL> view = view , <EOL> ) <EOL> p2 = v . getPipeline ( '<STR_LIT>' ) <EOL> controller . change_selected_version ( n ) <EOL> controller . flush_delayed_actions ( ) <EOL> p2 = controller . current_pipeline <EOL> result = interpreter . execute ( p2 , <EOL> locator = v , <EOL> current_version = n , <EOL> view = view , <EOL> ) <EOL> self . assertEqual ( len ( result . modules_added ) , <NUM_LIT:1> ) <EOL> finally : <EOL> StandardOutput . compute = old_compute <EOL> if __name__ == '<STR_LIT:__main__>' : <EOL> unittest . main ( ) </s>
