<s> """<STR_LIT>""" <EOL> from __future__ import division <EOL> import vistrails . core . cache . hasher <EOL> from vistrails . core . debug import format_exception <EOL> from vistrails . core . modules . module_registry import get_module_registry <EOL> from vistrails . core . modules . vistrails_module import Module , new_module , Converter , NotCacheable , ModuleError <EOL> from vistrails . core . modules . config import ConstantWidgetConfig , QueryWidgetConfig , ParamExpWidgetConfig , ModuleSettings , IPort , OPort , CIPort <EOL> import vistrails . core . system <EOL> from vistrails . core . utils import InstanceObject <EOL> from vistrails . core import debug <EOL> from abc import ABCMeta <EOL> from ast import literal_eval <EOL> from itertools import izip <EOL> import mimetypes <EOL> import os <EOL> import pickle <EOL> import re <EOL> import shutil <EOL> import zipfile <EOL> import urllib <EOL> try : <EOL> import hashlib <EOL> sha_hash = hashlib . sha1 <EOL> except ImportError : <EOL> import sha <EOL> sha_hash = sha . new <EOL> version = '<STR_LIT>' <EOL> name = '<STR_LIT>' <EOL> identifier = '<STR_LIT>' <EOL> old_identifiers = [ '<STR_LIT>' ] <EOL> constant_config_path = "<STR_LIT>" <EOL> query_config_path = "<STR_LIT>" <EOL> paramexp_config_path = "<STR_LIT>" <EOL> def get_port_name ( port ) : <EOL> if hasattr ( port , '<STR_LIT:name>' ) : <EOL> return port . name <EOL> else : <EOL> return port [ <NUM_LIT:0> ] <EOL> class meta_add_value_ports ( type ) : <EOL> def __new__ ( cls , name , bases , dct ) : <EOL> """<STR_LIT>""" <EOL> mod = type . __new__ ( cls , name , bases , dct ) <EOL> if '<STR_LIT>' in mod . __dict__ : <EOL> input_ports = mod . _input_ports <EOL> if not any ( get_port_name ( port_info ) == '<STR_LIT:value>' <EOL> for port_info in input_ports ) : <EOL> mod . _input_ports = [ ( '<STR_LIT:value>' , mod ) ] <EOL> mod . _input_ports . extend ( input_ports ) <EOL> else : <EOL> mod . _input_ports = [ ( '<STR_LIT:value>' , mod ) ] <EOL> if '<STR_LIT>' in mod . __dict__ : <EOL> output_ports = mod . _output_ports <EOL> if not any ( get_port_name ( port_info ) == '<STR_LIT:value>' <EOL> for port_info in output_ports ) : <EOL> mod . _output_ports = [ ( '<STR_LIT:value>' , mod ) ] <EOL> mod . _output_ports . extend ( output_ports ) <EOL> else : <EOL> mod . _output_ports = [ ( '<STR_LIT:value>' , mod ) ] <EOL> return mod <EOL> class Constant ( Module ) : <EOL> """<STR_LIT>""" <EOL> _settings = ModuleSettings ( abstract = True ) <EOL> _output_ports = [ OPort ( "<STR_LIT>" , "<STR_LIT>" ) ] <EOL> __metaclass__ = meta_add_value_ports <EOL> @ staticmethod <EOL> def validate ( x ) : <EOL> raise NotImplementedError <EOL> @ staticmethod <EOL> def translate_to_python ( x ) : <EOL> raise NotImplementedError <EOL> def compute ( self ) : <EOL> """<STR_LIT>""" <EOL> v = self . get_input ( "<STR_LIT:value>" ) <EOL> b = self . validate ( v ) <EOL> if not b : <EOL> raise ModuleError ( self , "<STR_LIT>" ) <EOL> self . set_output ( "<STR_LIT:value>" , v ) <EOL> self . set_output ( "<STR_LIT>" , self . translate_to_string ( v ) ) <EOL> def setValue ( self , v ) : <EOL> self . set_output ( "<STR_LIT:value>" , self . translate_to_python ( v ) ) <EOL> self . upToDate = True <EOL> @ staticmethod <EOL> def translate_to_string ( v ) : <EOL> return str ( v ) <EOL> @ staticmethod <EOL> def get_widget_class ( ) : <EOL> return None <EOL> @ staticmethod <EOL> def query_compute ( value_a , value_b , query_method ) : <EOL> if query_method == '<STR_LIT>' or query_method is None : <EOL> return ( value_a == value_b ) <EOL> elif query_method == '<STR_LIT>' : <EOL> return ( value_a != value_b ) <EOL> return False <EOL> def new_constant ( name , py_conversion = None , default_value = None , validation = None , <EOL> widget_type = None , <EOL> str_conversion = None , base_class = Constant , <EOL> compute = None , query_compute = None ) : <EOL> """<STR_LIT>""" <EOL> d = { } <EOL> if py_conversion is not None : <EOL> d [ "<STR_LIT>" ] = py_conversion <EOL> elif base_class == Constant : <EOL> raise ValueError ( "<STR_LIT>" ) <EOL> if validation is not None : <EOL> d [ "<STR_LIT>" ] = validation <EOL> elif base_class == Constant : <EOL> raise ValueError ( "<STR_LIT>" ) <EOL> if default_value is not None : <EOL> d [ "<STR_LIT>" ] = default_value <EOL> if str_conversion is not None : <EOL> d [ '<STR_LIT>' ] = str_conversion <EOL> if compute is not None : <EOL> d [ '<STR_LIT>' ] = compute <EOL> if query_compute is not None : <EOL> d [ '<STR_LIT>' ] = query_compute <EOL> if widget_type is not None : <EOL> @ staticmethod <EOL> def get_widget_class ( ) : <EOL> return widget_type <EOL> d [ '<STR_LIT>' ] = get_widget_class <EOL> m = new_module ( base_class , name , d ) <EOL> m . _input_ports = [ ( '<STR_LIT:value>' , m ) ] <EOL> m . _output_ports = [ ( '<STR_LIT:value>' , m ) ] <EOL> return m <EOL> class Boolean ( Constant ) : <EOL> _settings = ModuleSettings ( <EOL> constant_widget = '<STR_LIT>' % constant_config_path ) <EOL> default_value = False <EOL> @ staticmethod <EOL> def translate_to_python ( x ) : <EOL> s = x . upper ( ) <EOL> if s == '<STR_LIT>' : <EOL> return True <EOL> if s == '<STR_LIT>' : <EOL> return False <EOL> raise ValueError ( '<STR_LIT>' <EOL> '<STR_LIT>' % x ) <EOL> @ staticmethod <EOL> def validate ( x ) : <EOL> return isinstance ( x , bool ) <EOL> class Float ( Constant ) : <EOL> _settings = ModuleSettings ( constant_widgets = [ <EOL> QueryWidgetConfig ( '<STR_LIT>' % query_config_path ) , <EOL> ParamExpWidgetConfig ( '<STR_LIT>' % paramexp_config_path ) ] ) <EOL> default_value = <NUM_LIT:0.0> <EOL> @ staticmethod <EOL> def translate_to_python ( x ) : <EOL> return float ( x ) <EOL> @ staticmethod <EOL> def validate ( x ) : <EOL> return isinstance ( x , ( int , long , float ) ) <EOL> @ staticmethod <EOL> def query_compute ( value_a , value_b , query_method ) : <EOL> value_a = float ( value_a ) <EOL> value_b = float ( value_b ) <EOL> if query_method == '<STR_LIT>' or query_method is None : <EOL> return ( value_a == value_b ) <EOL> elif query_method == '<STR_LIT:<>' : <EOL> return ( value_a < value_b ) <EOL> elif query_method == '<STR_LIT:>>' : <EOL> return ( value_a > value_b ) <EOL> elif query_method == '<STR_LIT>' : <EOL> return ( value_a <= value_b ) <EOL> elif query_method == '<STR_LIT>' : <EOL> return ( value_a >= value_b ) <EOL> class Integer ( Float ) : <EOL> _settings = ModuleSettings ( constant_widgets = [ <EOL> QueryWidgetConfig ( '<STR_LIT>' % query_config_path ) , <EOL> ParamExpWidgetConfig ( '<STR_LIT>' % paramexp_config_path ) ] ) <EOL> default_value = <NUM_LIT:0> <EOL> @ staticmethod <EOL> def translate_to_python ( x ) : <EOL> if x . startswith ( '<STR_LIT>' ) : <EOL> return int ( x , <NUM_LIT:16> ) <EOL> else : <EOL> return int ( x ) <EOL> @ staticmethod <EOL> def validate ( x ) : <EOL> return isinstance ( x , ( int , long ) ) <EOL> class String ( Constant ) : <EOL> _settings = ModuleSettings ( <EOL> configure_widget = "<STR_LIT>" , <EOL> constant_widgets = [ <EOL> ConstantWidgetConfig ( '<STR_LIT>' % constant_config_path , <EOL> widget_type = '<STR_LIT>' ) , <EOL> QueryWidgetConfig ( '<STR_LIT>' % query_config_path ) ] ) <EOL> _output_ports = [ OPort ( "<STR_LIT>" , "<STR_LIT>" , optional = True ) ] <EOL> default_value = "<STR_LIT>" <EOL> @ staticmethod <EOL> def translate_to_python ( x ) : <EOL> assert isinstance ( x , ( str , unicode ) ) <EOL> return str ( x ) <EOL> @ staticmethod <EOL> def validate ( x ) : <EOL> return isinstance ( x , str ) <EOL> @ staticmethod <EOL> def query_compute ( value_a , value_b , query_method ) : <EOL> if query_method == '<STR_LIT>' or query_method is None : <EOL> return ( value_b in value_a ) <EOL> elif query_method == '<STR_LIT>' : <EOL> return ( value_a == value_b ) <EOL> elif query_method == '<STR_LIT>' : <EOL> try : <EOL> m = re . match ( value_b , value_a ) <EOL> if m is not None : <EOL> return ( m . end ( ) == len ( value_a ) ) <EOL> except re . error : <EOL> pass <EOL> return False <EOL> try : <EOL> from IPython import display <EOL> except ImportError : <EOL> display = None <EOL> class PathObject ( object ) : <EOL> def __init__ ( self , name ) : <EOL> self . name = name <EOL> self . _ipython_repr = None <EOL> def __repr__ ( self ) : <EOL> return "<STR_LIT>" % self . name <EOL> __str__ = __repr__ <EOL> def __getattr__ ( self , name ) : <EOL> if name . startswith ( '<STR_LIT>' ) and name . endswith ( '<STR_LIT:_>' ) : <EOL> if self . _ipython_repr is None : <EOL> filetype , encoding = mimetypes . guess_type ( self . name ) <EOL> if filetype and filetype . startswith ( '<STR_LIT>' ) : <EOL> self . _ipython_repr = display . Image ( filename = self . name ) <EOL> else : <EOL> self . _ipython_repr = False <EOL> if self . _ipython_repr is not False : <EOL> return getattr ( self . _ipython_repr , name ) <EOL> raise AttributeError <EOL> class Path ( Constant ) : <EOL> _settings = ModuleSettings ( constant_widget = ( "<STR_LIT>" % <EOL> constant_config_path ) ) <EOL> _input_ports = [ IPort ( "<STR_LIT:value>" , "<STR_LIT>" ) , <EOL> IPort ( "<STR_LIT:name>" , "<STR_LIT>" , optional = True ) ] <EOL> _output_ports = [ OPort ( "<STR_LIT:value>" , "<STR_LIT>" ) ] <EOL> @ staticmethod <EOL> def translate_to_python ( x ) : <EOL> return PathObject ( x ) <EOL> @ staticmethod <EOL> def translate_to_string ( x ) : <EOL> return str ( x . name ) <EOL> @ staticmethod <EOL> def validate ( v ) : <EOL> return isinstance ( v , PathObject ) <EOL> def get_name ( self ) : <EOL> n = None <EOL> if self . has_input ( "<STR_LIT:value>" ) : <EOL> n = self . get_input ( "<STR_LIT:value>" ) . name <EOL> if n is None : <EOL> self . check_input ( "<STR_LIT:name>" ) <EOL> n = self . get_input ( "<STR_LIT:name>" ) <EOL> return n <EOL> def set_results ( self , n ) : <EOL> self . set_output ( "<STR_LIT:value>" , PathObject ( n ) ) <EOL> self . set_output ( "<STR_LIT>" , n ) <EOL> def compute ( self ) : <EOL> n = self . get_name ( ) <EOL> self . set_results ( n ) <EOL> Path . default_value = PathObject ( '<STR_LIT>' ) <EOL> def path_parameter_hasher ( p ) : <EOL> def get_mtime ( path ) : <EOL> t = int ( os . path . getmtime ( path ) ) <EOL> if os . path . isdir ( path ) : <EOL> for subpath in os . listdir ( path ) : <EOL> subpath = os . path . join ( path , subpath ) <EOL> if os . path . isdir ( subpath ) : <EOL> t = max ( t , get_mtime ( subpath ) ) <EOL> return t <EOL> h = vistrails . core . cache . hasher . Hasher . parameter_signature ( p ) <EOL> try : <EOL> t = get_mtime ( p . strValue ) <EOL> except OSError : <EOL> return h <EOL> hasher = sha_hash ( ) <EOL> hasher . update ( h ) <EOL> hasher . update ( str ( t ) ) <EOL> return hasher . digest ( ) <EOL> class File ( Path ) : <EOL> """<STR_LIT>""" <EOL> _settings = ModuleSettings ( constant_signature = path_parameter_hasher , <EOL> constant_widget = ( "<STR_LIT>" % <EOL> constant_config_path ) ) <EOL> _input_ports = [ IPort ( "<STR_LIT:value>" , "<STR_LIT>" ) , <EOL> IPort ( "<STR_LIT>" , "<STR_LIT>" , optional = True ) ] <EOL> _output_ports = [ OPort ( "<STR_LIT:value>" , "<STR_LIT>" ) , <EOL> OPort ( "<STR_LIT>" , "<STR_LIT>" , optional = True ) ] <EOL> def compute ( self ) : <EOL> n = self . get_name ( ) <EOL> if ( self . has_input ( "<STR_LIT>" ) and self . get_input ( "<STR_LIT>" ) ) : <EOL> vistrails . core . system . touch ( n ) <EOL> if not os . path . isfile ( n ) : <EOL> raise ModuleError ( self , '<STR_LIT>' % n ) <EOL> self . set_results ( n ) <EOL> self . set_output ( "<STR_LIT>" , n ) <EOL> class Directory ( Path ) : <EOL> _settings = ModuleSettings ( constant_signature = path_parameter_hasher , <EOL> constant_widget = ( "<STR_LIT>" % <EOL> constant_config_path ) ) <EOL> _input_ports = [ IPort ( "<STR_LIT:value>" , "<STR_LIT>" ) , <EOL> IPort ( "<STR_LIT>" , "<STR_LIT>" , optional = True ) ] <EOL> _output_ports = [ OPort ( "<STR_LIT:value>" , "<STR_LIT>" ) , <EOL> OPort ( "<STR_LIT>" , "<STR_LIT>" ) ] <EOL> def compute ( self ) : <EOL> n = self . get_name ( ) <EOL> if ( self . has_input ( "<STR_LIT>" ) and <EOL> self . get_input ( "<STR_LIT>" ) ) : <EOL> try : <EOL> vistrails . core . system . mkdir ( n ) <EOL> except Exception , e : <EOL> raise ModuleError ( self , '<STR_LIT>' % format_exception ( e ) ) <EOL> if not os . path . isdir ( n ) : <EOL> raise ModuleError ( self , '<STR_LIT>' % n ) <EOL> self . set_results ( n ) <EOL> dir_list = os . listdir ( n ) <EOL> output_list = [ ] <EOL> for item in dir_list : <EOL> full_path = os . path . join ( n , item ) <EOL> output_list . append ( PathObject ( full_path ) ) <EOL> self . set_output ( '<STR_LIT>' , output_list ) <EOL> class OutputPath ( Path ) : <EOL> _settings = ModuleSettings ( constant_widget = ( "<STR_LIT>" % <EOL> constant_config_path ) ) <EOL> _input_ports = [ IPort ( "<STR_LIT:value>" , "<STR_LIT>" ) ] <EOL> _output_ports = [ OPort ( "<STR_LIT:value>" , "<STR_LIT>" ) ] <EOL> def get_name ( self ) : <EOL> n = None <EOL> if self . has_input ( "<STR_LIT:value>" ) : <EOL> n = self . get_input ( "<STR_LIT:value>" ) . name <EOL> if n is None : <EOL> self . check_input ( "<STR_LIT:name>" ) <EOL> n = self . get_input ( "<STR_LIT:name>" ) <EOL> return n <EOL> def set_results ( self , n ) : <EOL> self . set_output ( "<STR_LIT:value>" , PathObject ( n ) ) <EOL> self . set_output ( "<STR_LIT>" , n ) <EOL> def compute ( self ) : <EOL> n = self . get_name ( ) <EOL> self . set_results ( n ) <EOL> class FileSink ( NotCacheable , Module ) : <EOL> """<STR_LIT>""" <EOL> _input_ports = [ IPort ( "<STR_LIT:file>" , File ) , <EOL> IPort ( "<STR_LIT>" , OutputPath ) , <EOL> IPort ( "<STR_LIT>" , Boolean , optional = True , <EOL> default = True ) , <EOL> IPort ( "<STR_LIT>" , Boolean , optional = True ) ] <EOL> def compute ( self ) : <EOL> input_file = self . get_input ( "<STR_LIT:file>" ) <EOL> output_path = self . get_input ( "<STR_LIT>" ) <EOL> full_path = output_path . name <EOL> if os . path . isfile ( full_path ) : <EOL> if self . get_input ( '<STR_LIT>' ) : <EOL> try : <EOL> os . remove ( full_path ) <EOL> except OSError , e : <EOL> msg = ( '<STR_LIT>' <EOL> '<STR_LIT>' % full_path ) <EOL> raise ModuleError ( self , msg ) <EOL> else : <EOL> raise ModuleError ( self , <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> try : <EOL> vistrails . core . system . link_or_copy ( input_file . name , full_path ) <EOL> except OSError , e : <EOL> msg = "<STR_LIT>" % ( full_path , e ) <EOL> raise ModuleError ( self , msg ) <EOL> if ( self . has_input ( "<STR_LIT>" ) and <EOL> self . get_input ( "<STR_LIT>" ) or <EOL> not self . has_input ( "<STR_LIT>" ) ) : <EOL> if self . moduleInfo . has_key ( '<STR_LIT>' ) : <EOL> if self . moduleInfo [ '<STR_LIT>' ] . has_key ( '<STR_LIT>' ) : <EOL> folder = self . moduleInfo [ '<STR_LIT>' ] [ '<STR_LIT>' ] <EOL> base_fname = os . path . basename ( full_path ) <EOL> ( base_fname , file_extension ) = os . path . splitext ( base_fname ) <EOL> base_fname = os . path . join ( folder , base_fname ) <EOL> filename = base_fname + file_extension <EOL> counter = <NUM_LIT:2> <EOL> while os . path . exists ( filename ) : <EOL> filename = base_fname + "<STR_LIT>" % ( counter , <EOL> file_extension ) <EOL> counter += <NUM_LIT:1> <EOL> try : <EOL> vistrails . core . system . link_or_copy ( input_file . name , filename ) <EOL> except OSError , e : <EOL> msg = "<STR_LIT>" % ( <EOL> full_path , filename ) <EOL> debug . warning ( "<STR_LIT:%s>" % msg , e ) <EOL> class DirectorySink ( NotCacheable , Module ) : <EOL> """<STR_LIT>""" <EOL> _input_ports = [ IPort ( "<STR_LIT>" , Directory ) , <EOL> IPort ( "<STR_LIT>" , OutputPath ) , <EOL> IPort ( "<STR_LIT>" , Boolean , optional = True , default = "<STR_LIT:True>" ) ] <EOL> def compute ( self ) : <EOL> input_dir = self . get_input ( "<STR_LIT>" ) <EOL> output_path = self . get_input ( "<STR_LIT>" ) <EOL> full_path = output_path . name <EOL> if os . path . exists ( full_path ) : <EOL> if self . get_input ( "<STR_LIT>" ) : <EOL> try : <EOL> if os . path . isfile ( full_path ) : <EOL> os . remove ( full_path ) <EOL> else : <EOL> shutil . rmtree ( full_path ) <EOL> except OSError , e : <EOL> msg = ( '<STR_LIT>' <EOL> '<STR_LIT>' % full_path ) <EOL> raise ModuleError ( <EOL> self , <EOL> '<STR_LIT>' % ( msg , format_exception ( e ) ) ) <EOL> else : <EOL> msg = ( '<STR_LIT>' <EOL> '<STR_LIT>' % full_path ) <EOL> raise ModuleError ( self , msg ) <EOL> try : <EOL> shutil . copytree ( input_dir . name , full_path ) <EOL> except OSError , e : <EOL> msg = '<STR_LIT>' % ( input_dir . name , full_path ) <EOL> raise ModuleError ( self , '<STR_LIT>' % ( msg , format_exception ( e ) ) ) <EOL> class WriteFile ( Converter ) : <EOL> """<STR_LIT>""" <EOL> _input_ports = [ IPort ( '<STR_LIT>' , String ) , <EOL> IPort ( '<STR_LIT>' , String , optional = True , default = "<STR_LIT>" ) , <EOL> IPort ( '<STR_LIT>' , String , optional = True ) ] <EOL> _output_ports = [ OPort ( '<STR_LIT>' , File ) ] <EOL> def compute ( self ) : <EOL> contents = self . get_input ( '<STR_LIT>' ) <EOL> suffix = self . force_get_input ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> result = self . interpreter . filePool . create_file ( suffix = suffix ) <EOL> if self . has_input ( '<STR_LIT>' ) : <EOL> contents = contents . decode ( '<STR_LIT:utf-8>' ) <EOL> contents = contents . encode ( self . get_input ( '<STR_LIT>' ) ) <EOL> with open ( result . name , '<STR_LIT:wb>' ) as fp : <EOL> fp . write ( contents ) <EOL> self . set_output ( '<STR_LIT>' , result ) <EOL> class ReadFile ( Converter ) : <EOL> """<STR_LIT>""" <EOL> _input_ports = [ IPort ( '<STR_LIT>' , File ) , <EOL> IPort ( '<STR_LIT>' , String , optional = True ) ] <EOL> _output_ports = [ OPort ( '<STR_LIT>' , String ) ] <EOL> def compute ( self ) : <EOL> filename = self . get_input ( '<STR_LIT>' ) . name <EOL> with open ( filename , '<STR_LIT:rb>' ) as fp : <EOL> contents = fp . read ( ) <EOL> if self . has_input ( '<STR_LIT>' ) : <EOL> contents = contents . decode ( self . get_input ( '<STR_LIT>' ) ) <EOL> contents = contents . encode ( '<STR_LIT:utf-8>' ) <EOL> self . set_output ( '<STR_LIT>' , contents ) <EOL> class Color ( Constant ) : <EOL> _settings = ModuleSettings ( constant_widgets = [ <EOL> '<STR_LIT>' % constant_config_path , <EOL> ConstantWidgetConfig ( '<STR_LIT>' % constant_config_path , <EOL> widget_type = '<STR_LIT>' ) , <EOL> QueryWidgetConfig ( '<STR_LIT>' % query_config_path ) , <EOL> ParamExpWidgetConfig ( '<STR_LIT>' % paramexp_config_path , <EOL> widget_type = '<STR_LIT>' ) , <EOL> ParamExpWidgetConfig ( '<STR_LIT>' % paramexp_config_path , <EOL> widget_type = '<STR_LIT>' ) ] ) <EOL> _input_ports = [ IPort ( "<STR_LIT:value>" , "<STR_LIT>" ) ] <EOL> _output_ports = [ OPort ( "<STR_LIT:value>" , "<STR_LIT>" ) ] <EOL> default_value = InstanceObject ( tuple = ( <NUM_LIT:1> , <NUM_LIT:1> , <NUM_LIT:1> ) ) <EOL> @ staticmethod <EOL> def translate_to_python ( x ) : <EOL> return InstanceObject ( <EOL> tuple = tuple ( [ float ( a ) for a in x . split ( '<STR_LIT:U+002C>' ) ] ) ) <EOL> @ staticmethod <EOL> def translate_to_string ( v ) : <EOL> return '<STR_LIT:U+002C>' . join ( '<STR_LIT>' % c for c in v . tuple ) <EOL> @ staticmethod <EOL> def validate ( x ) : <EOL> return isinstance ( x , InstanceObject ) and hasattr ( x , '<STR_LIT>' ) <EOL> @ staticmethod <EOL> def to_string ( r , g , b ) : <EOL> return "<STR_LIT>" % ( r , g , b ) <EOL> @ staticmethod <EOL> def query_compute ( value_a , value_b , query_method ) : <EOL> def rgb_to_xyz ( r , g , b ) : <EOL> if r > <NUM_LIT> : <EOL> r = ( ( r + <NUM_LIT> ) / <NUM_LIT> ) ** <NUM_LIT> <EOL> else : <EOL> r = r / <NUM_LIT> <EOL> if g > <NUM_LIT> : <EOL> g = ( ( g + <NUM_LIT> ) / <NUM_LIT> ) ** <NUM_LIT> <EOL> else : <EOL> g = g / <NUM_LIT> <EOL> if b > <NUM_LIT> : <EOL> b = ( ( b + <NUM_LIT> ) / <NUM_LIT> ) ** <NUM_LIT> <EOL> else : <EOL> b = b / <NUM_LIT> <EOL> r *= <NUM_LIT:100> <EOL> g *= <NUM_LIT:100> <EOL> b *= <NUM_LIT:100> <EOL> x = r * <NUM_LIT> + g * <NUM_LIT> + b * <NUM_LIT> <EOL> y = r * <NUM_LIT> + g * <NUM_LIT> + b * <NUM_LIT> <EOL> z = r * <NUM_LIT> + g * <NUM_LIT> + b * <NUM_LIT> <EOL> return ( x , y , z ) <EOL> def xyz_to_cielab ( x , y , z ) : <EOL> ref_x , ref_y , ref_z = ( <NUM_LIT> , <NUM_LIT> , <NUM_LIT> ) <EOL> x /= ref_x <EOL> y /= ref_y <EOL> z /= ref_z <EOL> if x > <NUM_LIT> : <EOL> x = x ** ( <NUM_LIT:1> / <NUM_LIT> ) <EOL> else : <EOL> x = ( <NUM_LIT> * x ) + ( <NUM_LIT:16> / <NUM_LIT> ) <EOL> if y > <NUM_LIT> : <EOL> y = y ** ( <NUM_LIT:1> / <NUM_LIT> ) <EOL> else : <EOL> y = ( <NUM_LIT> * y ) + ( <NUM_LIT:16> / <NUM_LIT> ) <EOL> if z > <NUM_LIT> : <EOL> z = z ** ( <NUM_LIT:1> / <NUM_LIT> ) <EOL> else : <EOL> z = ( <NUM_LIT> * z ) + ( <NUM_LIT:16> / <NUM_LIT> ) <EOL> L = ( <NUM_LIT> * y ) - <NUM_LIT:16> <EOL> a = <NUM_LIT> * ( x - y ) <EOL> b = <NUM_LIT:200> * ( y - z ) <EOL> return ( L , a , b ) <EOL> def rgb_to_cielab ( r , g , b ) : <EOL> return xyz_to_cielab ( * rgb_to_xyz ( r , g , b ) ) <EOL> value_a_rgb = ( float ( a ) for a in value_a . split ( '<STR_LIT:U+002C>' ) ) <EOL> value_b_rgb = ( float ( b ) for b in value_b . split ( '<STR_LIT:U+002C>' ) ) <EOL> value_a_lab = rgb_to_cielab ( * value_a_rgb ) <EOL> value_b_lab = rgb_to_cielab ( * value_b_rgb ) <EOL> diff = sum ( ( v_1 - v_2 ) ** <NUM_LIT:2> <EOL> for v_1 , v_2 in izip ( value_a_lab , value_b_lab ) ) ** ( <NUM_LIT:0.5> ) <EOL> if query_method is None : <EOL> query_method = '<STR_LIT>' <EOL> return diff < float ( query_method ) <EOL> class StandardOutput ( NotCacheable , Module ) : <EOL> """<STR_LIT>""" <EOL> _input_ports = [ IPort ( "<STR_LIT:value>" , '<STR_LIT>' ) ] <EOL> def compute ( self ) : <EOL> v = self . get_input ( "<STR_LIT:value>" ) <EOL> if isinstance ( v , PathObject ) : <EOL> try : <EOL> fp = open ( v . name , '<STR_LIT:rb>' ) <EOL> except IOError : <EOL> print v <EOL> else : <EOL> try : <EOL> CHUNKSIZE = <NUM_LIT> <EOL> chunk = fp . read ( CHUNKSIZE ) <EOL> if chunk : <EOL> sys . stdout . write ( chunk ) <EOL> while len ( chunk ) == CHUNKSIZE : <EOL> chunk = fp . read ( CHUNKSIZE ) <EOL> if chunk : <EOL> sys . stdout . write ( chunk ) <EOL> sys . stdout . write ( '<STR_LIT:\n>' ) <EOL> finally : <EOL> fp . close ( ) <EOL> else : <EOL> print v <EOL> class Tuple ( Module ) : <EOL> """<STR_LIT>""" <EOL> _settings = ModuleSettings ( configure_widget = <EOL> "<STR_LIT>" ) <EOL> def __init__ ( self ) : <EOL> Module . __init__ ( self ) <EOL> self . input_ports_order = [ ] <EOL> self . values = tuple ( ) <EOL> def transfer_attrs ( self , module ) : <EOL> Module . transfer_attrs ( self , module ) <EOL> self . input_ports_order = [ p . name for p in module . input_port_specs ] <EOL> def compute ( self ) : <EOL> values = tuple ( [ self . get_input ( p ) <EOL> for p in self . input_ports_order ] ) <EOL> self . values = values <EOL> self . set_output ( "<STR_LIT:value>" , values ) <EOL> class Untuple ( Module ) : <EOL> """<STR_LIT>""" <EOL> _settings = ModuleSettings ( configure_widget = <EOL> "<STR_LIT>" ) <EOL> def __init__ ( self ) : <EOL> Module . __init__ ( self ) <EOL> self . output_ports_order = [ ] <EOL> def transfer_attrs ( self , module ) : <EOL> Module . transfer_attrs ( self , module ) <EOL> self . output_ports_order = [ p . name for p in module . output_port_specs ] <EOL> self . output_ports_order . reverse ( ) <EOL> def compute ( self ) : <EOL> if self . has_input ( "<STR_LIT>" ) : <EOL> tuple = self . get_input ( "<STR_LIT>" ) <EOL> values = tuple . values <EOL> else : <EOL> values = self . get_input ( "<STR_LIT:value>" ) <EOL> for p , value in izip ( self . output_ports_order , values ) : <EOL> self . set_output ( p , value ) <EOL> class ConcatenateString ( Module ) : <EOL> """<STR_LIT>""" <EOL> fieldCount = <NUM_LIT:4> <EOL> _input_ports = [ IPort ( "<STR_LIT>" % i , "<STR_LIT>" ) <EOL> for i in xrange ( <NUM_LIT:1> , <NUM_LIT:1> + fieldCount ) ] <EOL> _output_ports = [ OPort ( "<STR_LIT:value>" , "<STR_LIT>" ) ] <EOL> def compute ( self ) : <EOL> result = "<STR_LIT>" . join ( self . force_get_input ( '<STR_LIT>' % i , '<STR_LIT>' ) <EOL> for i in xrange ( <NUM_LIT:1> , <NUM_LIT:1> + self . fieldCount ) ) <EOL> self . set_output ( '<STR_LIT:value>' , result ) <EOL> class Not ( Module ) : <EOL> """<STR_LIT>""" <EOL> _input_ports = [ IPort ( '<STR_LIT:input>' , '<STR_LIT>' ) ] <EOL> _output_ports = [ OPort ( '<STR_LIT:value>' , '<STR_LIT>' ) ] <EOL> def compute ( self ) : <EOL> value = self . get_input ( '<STR_LIT:input>' ) <EOL> self . set_output ( '<STR_LIT:value>' , not value ) <EOL> class ListType ( object ) : <EOL> __metaclass__ = ABCMeta <EOL> ListType . register ( list ) <EOL> try : <EOL> import numpy <EOL> except ImportError : <EOL> numpy = None <EOL> else : <EOL> ListType . register ( numpy . ndarray ) <EOL> class List ( Constant ) : <EOL> _settings = ModuleSettings ( configure_widget = <EOL> "<STR_LIT>" ) <EOL> _input_ports = [ IPort ( "<STR_LIT:value>" , "<STR_LIT>" ) , <EOL> IPort ( "<STR_LIT>" , "<STR_LIT>" , depth = <NUM_LIT:1> ) , <EOL> IPort ( "<STR_LIT>" , "<STR_LIT>" ) ] <EOL> _output_ports = [ OPort ( "<STR_LIT:value>" , "<STR_LIT>" ) ] <EOL> default_value = [ ] <EOL> def __init__ ( self ) : <EOL> Constant . __init__ ( self ) <EOL> self . input_ports_order = [ ] <EOL> def transfer_attrs ( self , module ) : <EOL> Module . transfer_attrs ( self , module ) <EOL> self . input_ports_order = [ p . name for p in module . input_port_specs ] <EOL> @ staticmethod <EOL> def validate ( x ) : <EOL> return isinstance ( x , ListType ) <EOL> @ staticmethod <EOL> def translate_to_python ( v ) : <EOL> return literal_eval ( v ) <EOL> @ staticmethod <EOL> def translate_to_string ( v , dims = None ) : <EOL> if dims is None : <EOL> if numpy is not None and isinstance ( v , numpy . ndarray ) : <EOL> dims = v . ndim <EOL> else : <EOL> dims = <NUM_LIT:1> <EOL> if dims == <NUM_LIT:1> : <EOL> return '<STR_LIT>' % '<STR_LIT:U+002CU+0020>' . join ( repr ( c ) <EOL> for c in v ) <EOL> else : <EOL> return '<STR_LIT>' % '<STR_LIT:U+002CU+0020>' . join ( List . translate_to_string ( c , dims - <NUM_LIT:1> ) <EOL> for c in v ) <EOL> def compute ( self ) : <EOL> head , middle , items , tail = [ ] , [ ] , [ ] , [ ] <EOL> got_value = False <EOL> if self . has_input ( '<STR_LIT:value>' ) : <EOL> Constant . compute ( self ) <EOL> middle = self . outputPorts [ '<STR_LIT:value>' ] <EOL> got_value = True <EOL> if self . has_input ( '<STR_LIT>' ) : <EOL> head = self . get_input ( '<STR_LIT>' ) <EOL> got_value = True <EOL> if self . input_ports_order : <EOL> items = [ self . get_input ( p ) <EOL> for p in self . input_ports_order ] <EOL> got_value = True <EOL> if self . has_input ( '<STR_LIT>' ) : <EOL> tail = self . get_input ( '<STR_LIT>' ) <EOL> got_value = True <EOL> if not got_value : <EOL> self . get_input ( '<STR_LIT:value>' ) <EOL> self . set_output ( '<STR_LIT:value>' , head + middle + items + tail ) <EOL> class Dictionary ( Constant ) : <EOL> default_value = { } <EOL> _input_ports = [ CIPort ( "<STR_LIT>" , "<STR_LIT>" ) , <EOL> IPort ( "<STR_LIT>" , "<STR_LIT>" ) ] <EOL> @ staticmethod <EOL> def translate_to_python ( v ) : <EOL> return literal_eval ( v ) <EOL> @ staticmethod <EOL> def validate ( x ) : <EOL> return isinstance ( x , dict ) <EOL> def compute ( self ) : <EOL> d = { } <EOL> if self . has_input ( '<STR_LIT:value>' ) : <EOL> Constant . compute ( self ) <EOL> d . update ( self . outputPorts [ '<STR_LIT:value>' ] ) <EOL> if self . has_input ( '<STR_LIT>' ) : <EOL> pairs_list = self . get_input_list ( '<STR_LIT>' ) <EOL> d . update ( pairs_list ) <EOL> if self . has_input ( '<STR_LIT>' ) : <EOL> d . update ( self . get_input ( '<STR_LIT>' ) ) <EOL> self . set_output ( "<STR_LIT:value>" , d ) <EOL> class Null ( Module ) : <EOL> """<STR_LIT>""" <EOL> _settings = ModuleSettings ( hide_descriptor = True ) <EOL> def compute ( self ) : <EOL> self . set_output ( "<STR_LIT:value>" , None ) <EOL> class Unpickle ( Module ) : <EOL> """<STR_LIT>""" <EOL> _settings = ModuleSettings ( hide_descriptor = True ) <EOL> _input_ports = [ IPort ( '<STR_LIT:input>' , '<STR_LIT>' ) ] <EOL> _output_ports = [ OPort ( '<STR_LIT:result>' , '<STR_LIT>' ) ] <EOL> def compute ( self ) : <EOL> value = self . get_input ( '<STR_LIT:input>' ) <EOL> self . set_output ( '<STR_LIT:result>' , pickle . loads ( value ) ) <EOL> class CodeRunnerMixin ( object ) : <EOL> def __init__ ( self ) : <EOL> self . output_ports_order = [ ] <EOL> super ( CodeRunnerMixin , self ) . __init__ ( ) <EOL> def transfer_attrs ( self , module ) : <EOL> Module . transfer_attrs ( self , module ) <EOL> self . output_ports_order = [ p . name for p in module . output_port_specs ] <EOL> self . output_ports_order . reverse ( ) <EOL> def run_code ( self , code_str , <EOL> use_input = False , <EOL> use_output = False ) : <EOL> """<STR_LIT>""" <EOL> import vistrails . core . packagemanager <EOL> def fail ( msg ) : <EOL> raise ModuleError ( self , msg ) <EOL> def cache_this ( ) : <EOL> self . is_cacheable = lambda * args , ** kwargs : True <EOL> locals_ = locals ( ) <EOL> if use_input : <EOL> for k in self . inputPorts : <EOL> locals_ [ k ] = self . get_input ( k ) <EOL> if use_output : <EOL> for output_portname in self . output_ports_order : <EOL> if output_portname not in self . inputPorts : <EOL> locals_ [ output_portname ] = None <EOL> _m = vistrails . core . packagemanager . get_package_manager ( ) <EOL> reg = get_module_registry ( ) <EOL> locals_ . update ( { '<STR_LIT>' : fail , <EOL> '<STR_LIT>' : _m , <EOL> '<STR_LIT>' : cache_this , <EOL> '<STR_LIT>' : reg , <EOL> '<STR_LIT>' : self } ) <EOL> if '<STR_LIT:source>' in locals_ : <EOL> del locals_ [ '<STR_LIT:source>' ] <EOL> exec code_str + '<STR_LIT:\n>' in locals_ , locals_ <EOL> if use_output : <EOL> for k in self . output_ports_order : <EOL> if locals_ . get ( k ) is not None : <EOL> self . set_output ( k , locals_ [ k ] ) <EOL> class PythonSource ( CodeRunnerMixin , NotCacheable , Module ) : <EOL> """<STR_LIT>""" <EOL> _settings = ModuleSettings ( <EOL> configure_widget = ( "<STR_LIT>" <EOL> "<STR_LIT>" ) ) <EOL> _input_ports = [ IPort ( '<STR_LIT:source>' , '<STR_LIT>' , optional = True , default = "<STR_LIT>" ) ] <EOL> _output_pors = [ OPort ( '<STR_LIT>' , '<STR_LIT>' ) ] <EOL> def compute ( self ) : <EOL> s = urllib . unquote ( str ( self . get_input ( '<STR_LIT:source>' ) ) ) <EOL> self . run_code ( s , use_input = True , use_output = True ) <EOL> def zip_extract_file ( archive , filename_in_archive , output_filename ) : <EOL> z = zipfile . ZipFile ( archive ) <EOL> try : <EOL> fileinfo = z . getinfo ( filename_in_archive ) <EOL> output_dirname , output_filename = os . path . split ( output_filename ) <EOL> fileinfo . filename = output_filename <EOL> z . extract ( fileinfo , output_dirname ) <EOL> finally : <EOL> z . close ( ) <EOL> def zip_extract_all_files ( archive , output_path ) : <EOL> z = zipfile . ZipFile ( archive ) <EOL> try : <EOL> z . extractall ( output_path ) <EOL> finally : <EOL> z . close ( ) <EOL> class Unzip ( Module ) : <EOL> """<STR_LIT>""" <EOL> _input_ports = [ IPort ( '<STR_LIT>' , '<STR_LIT>' ) , <EOL> IPort ( '<STR_LIT>' , '<STR_LIT>' ) ] <EOL> _output_ports = [ OPort ( '<STR_LIT:file>' , '<STR_LIT>' ) ] <EOL> def compute ( self ) : <EOL> self . check_input ( "<STR_LIT>" ) <EOL> self . check_input ( "<STR_LIT>" ) <EOL> filename_in_archive = self . get_input ( "<STR_LIT>" ) <EOL> archive_file = self . get_input ( "<STR_LIT>" ) <EOL> if not os . path . isfile ( archive_file . name ) : <EOL> raise ModuleError ( self , "<STR_LIT>" ) <EOL> suffix = self . interpreter . filePool . guess_suffix ( filename_in_archive ) <EOL> output = self . interpreter . filePool . create_file ( suffix = suffix ) <EOL> zip_extract_file ( archive_file . name , <EOL> filename_in_archive , <EOL> output . name ) <EOL> self . set_output ( "<STR_LIT:file>" , output ) <EOL> class UnzipDirectory ( Module ) : <EOL> """<STR_LIT>""" <EOL> _input_ports = [ IPort ( '<STR_LIT>' , '<STR_LIT>' ) ] <EOL> _output_ports = [ OPort ( '<STR_LIT>' , '<STR_LIT>' ) ] <EOL> def compute ( self ) : <EOL> self . check_input ( "<STR_LIT>" ) <EOL> archive_file = self . get_input ( "<STR_LIT>" ) <EOL> if not os . path . isfile ( archive_file . name ) : <EOL> raise ModuleError ( self , "<STR_LIT>" ) <EOL> output = self . interpreter . filePool . create_directory ( ) <EOL> zip_extract_all_files ( archive_file . name , <EOL> output . name ) <EOL> self . set_output ( "<STR_LIT>" , output ) <EOL> class Round ( Converter ) : <EOL> """<STR_LIT>""" <EOL> _settings = ModuleSettings ( hide_descriptor = True ) <EOL> _input_ports = [ IPort ( '<STR_LIT>' , '<STR_LIT>' ) , <EOL> IPort ( '<STR_LIT>' , '<STR_LIT>' , optional = True , default = "<STR_LIT:True>" ) ] <EOL> _output_ports = [ OPort ( '<STR_LIT>' , '<STR_LIT>' ) ] <EOL> def compute ( self ) : <EOL> fl = self . get_input ( '<STR_LIT>' ) <EOL> floor = self . get_input ( '<STR_LIT>' ) <EOL> if floor : <EOL> integ = int ( fl ) <EOL> else : <EOL> integ = int ( fl + <NUM_LIT:0.5> ) <EOL> self . set_output ( '<STR_LIT>' , integ ) <EOL> class TupleToList ( Converter ) : <EOL> """<STR_LIT>""" <EOL> _settings = ModuleSettings ( hide_descriptor = True ) <EOL> _input_ports = [ IPort ( '<STR_LIT>' , '<STR_LIT>' ) ] <EOL> _output_ports = [ OPort ( '<STR_LIT>' , '<STR_LIT>' ) ] <EOL> @ classmethod <EOL> def can_convert ( cls , sub_descs , super_descs ) : <EOL> if len ( sub_descs ) <= <NUM_LIT:1> : <EOL> return False <EOL> reg = get_module_registry ( ) <EOL> return super_descs == [ reg . get_descriptor ( List ) ] <EOL> def compute ( self ) : <EOL> tu = self . get_input ( '<STR_LIT>' ) <EOL> if not isinstance ( tu , tuple ) : <EOL> raise ModuleError ( self , "<STR_LIT>" ) <EOL> self . set_output ( '<STR_LIT>' , list ( tu ) ) <EOL> class Variant ( Module ) : <EOL> """<STR_LIT>""" <EOL> _settings = ModuleSettings ( abstract = True ) <EOL> class Generator ( object ) : <EOL> """<STR_LIT>""" <EOL> _settings = ModuleSettings ( abstract = True ) <EOL> generators = [ ] <EOL> def __init__ ( self , size = None , module = None , generator = None , port = None , <EOL> accumulated = False ) : <EOL> self . module = module <EOL> self . generator = generator <EOL> self . port = port <EOL> self . size = size <EOL> self . accumulated = accumulated <EOL> if generator and module not in Generator . generators : <EOL> module . generator = generator <EOL> Generator . generators . append ( module ) <EOL> def next ( self ) : <EOL> """<STR_LIT>""" <EOL> value = self . module . get_output ( self . port ) <EOL> if isinstance ( value , Generator ) : <EOL> value = value . all ( ) <EOL> return value <EOL> def all ( self ) : <EOL> """<STR_LIT>""" <EOL> items = [ ] <EOL> item = self . next ( ) <EOL> while item is not None : <EOL> items . append ( item ) <EOL> item = self . next ( ) <EOL> return items <EOL> @ staticmethod <EOL> def stream ( ) : <EOL> """<STR_LIT>""" <EOL> result = True <EOL> if not Generator . generators : <EOL> return <EOL> while result is not None : <EOL> for g in Generator . generators : <EOL> result = g . next ( ) <EOL> Generator . generators = [ ] <EOL> class Assert ( Module ) : <EOL> """<STR_LIT>""" <EOL> _input_ports = [ IPort ( '<STR_LIT>' , '<STR_LIT>' ) ] <EOL> def compute ( self ) : <EOL> condition = self . get_input ( '<STR_LIT>' ) <EOL> if not condition : <EOL> raise ModuleError ( self , "<STR_LIT>" , <EOL> abort = True ) <EOL> class AssertEqual ( Module ) : <EOL> """<STR_LIT>""" <EOL> _input_ports = [ IPort ( '<STR_LIT>' , '<STR_LIT>' ) , <EOL> IPort ( '<STR_LIT>' , '<STR_LIT>' ) ] <EOL> def compute ( self ) : <EOL> values = ( self . get_input ( '<STR_LIT>' ) , <EOL> self . get_input ( '<STR_LIT>' ) ) <EOL> if values [ <NUM_LIT:0> ] != values [ <NUM_LIT:1> ] : <EOL> reprs = tuple ( repr ( v ) for v in values ) <EOL> reprs = tuple ( '<STR_LIT>' % v [ : <NUM_LIT> ] if len ( v ) > <NUM_LIT:20> else v <EOL> for v in reprs ) <EOL> raise ModuleError ( self , "<STR_LIT>" <EOL> "<STR_LIT>" % reprs , <EOL> abort = True ) <EOL> class StringFormat ( Module ) : <EOL> """<STR_LIT>""" <EOL> _settings = ModuleSettings ( configure_widget = <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> _input_ports = [ IPort ( '<STR_LIT>' , String ) ] <EOL> _output_ports = [ OPort ( '<STR_LIT:value>' , String ) ] <EOL> @ staticmethod <EOL> def list_placeholders ( fmt ) : <EOL> placeholders = set ( ) <EOL> nb = <NUM_LIT:0> <EOL> i = <NUM_LIT:0> <EOL> n = len ( fmt ) <EOL> while i < n : <EOL> if fmt [ i ] == '<STR_LIT:{>' : <EOL> i += <NUM_LIT:1> <EOL> if fmt [ i ] == '<STR_LIT:{>' : <EOL> i += <NUM_LIT:1> <EOL> continue <EOL> e = fmt . index ( '<STR_LIT:}>' , i ) <EOL> f = e <EOL> for c in ( '<STR_LIT::>' , '<STR_LIT:!>' , '<STR_LIT:[>' , '<STR_LIT:.>' ) : <EOL> c = fmt . find ( c , i ) <EOL> if c != - <NUM_LIT:1> : <EOL> f = min ( f , c ) <EOL> if i == f : <EOL> nb += <NUM_LIT:1> <EOL> else : <EOL> arg = fmt [ i : f ] <EOL> try : <EOL> arg = int ( arg ) <EOL> except ValueError : <EOL> placeholders . add ( arg ) <EOL> else : <EOL> nb = max ( nb , arg + <NUM_LIT:1> ) <EOL> i = e <EOL> i += <NUM_LIT:1> <EOL> return nb , placeholders <EOL> def compute ( self ) : <EOL> fmt = self . get_input ( '<STR_LIT>' ) <EOL> args , kwargs = StringFormat . list_placeholders ( fmt ) <EOL> f_args = [ self . get_input ( '<STR_LIT>' % n ) <EOL> for n in xrange ( args ) ] <EOL> f_kwargs = dict ( ( n , self . get_input ( n ) ) <EOL> for n in kwargs ) <EOL> self . set_output ( '<STR_LIT:value>' , fmt . format ( * f_args , ** f_kwargs ) ) <EOL> def init_constant ( m ) : <EOL> reg = get_module_registry ( ) <EOL> reg . add_module ( m ) <EOL> reg . add_input_port ( m , "<STR_LIT:value>" , m ) <EOL> reg . add_output_port ( m , "<STR_LIT:value>" , m ) <EOL> _modules = [ Module , Converter , Constant , Boolean , Float , Integer , String , List , <EOL> Path , File , Directory , OutputPath , <EOL> FileSink , DirectorySink , WriteFile , ReadFile , StandardOutput , <EOL> Tuple , Untuple , ConcatenateString , Not , Dictionary , Null , Variant , <EOL> Unpickle , PythonSource , Unzip , UnzipDirectory , Color , <EOL> Round , TupleToList , Assert , AssertEqual , StringFormat ] <EOL> def initialize ( * args , ** kwargs ) : <EOL> import vistrails . core . modules . sub_module <EOL> import vistrails . core . modules . output_modules <EOL> _modules . extend ( vistrails . core . modules . sub_module . _modules ) <EOL> _modules . extend ( vistrails . core . modules . output_modules . _modules ) <EOL> def handle_module_upgrade_request ( controller , module_id , pipeline ) : <EOL> from vistrails . core . upgradeworkflow import UpgradeWorkflowHandler <EOL> reg = get_module_registry ( ) <EOL> def outputName_remap ( old_conn , new_module ) : <EOL> ops = [ ] <EOL> old_src_module = pipeline . modules [ old_conn . source . moduleId ] <EOL> op_desc = reg . get_descriptor ( OutputPath ) <EOL> new_x = ( old_src_module . location . x + new_module . location . x ) / <NUM_LIT> <EOL> new_y = ( old_src_module . location . y + new_module . location . y ) / <NUM_LIT> <EOL> op_module = controller . create_module_from_descriptor ( op_desc , new_x , new_y ) <EOL> ops . append ( ( '<STR_LIT>' , op_module ) ) <EOL> create_new_connection = UpgradeWorkflowHandler . create_new_connection <EOL> new_conn_1 = create_new_connection ( controller , <EOL> old_src_module , <EOL> old_conn . source , <EOL> op_module , <EOL> "<STR_LIT:name>" ) <EOL> ops . append ( ( '<STR_LIT>' , new_conn_1 ) ) <EOL> new_conn_2 = create_new_connection ( controller , <EOL> op_module , <EOL> "<STR_LIT:value>" , <EOL> new_module , <EOL> "<STR_LIT>" ) <EOL> ops . append ( ( '<STR_LIT>' , new_conn_2 ) ) <EOL> return ops <EOL> module_remap = { '<STR_LIT>' : <EOL> [ ( None , '<STR_LIT>' , None , <EOL> { '<STR_LIT>' : <EOL> { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : outputName_remap } , <EOL> '<STR_LIT>' : <EOL> { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } } ) ] , <EOL> '<STR_LIT>' : <EOL> [ ( None , '<STR_LIT>' , '<STR_LIT>' , <EOL> { '<STR_LIT>' : <EOL> { '<STR_LIT>' : '<STR_LIT:value>' } , <EOL> '<STR_LIT>' : <EOL> { '<STR_LIT>' : '<STR_LIT>' } , <EOL> } ) ] , <EOL> '<STR_LIT>' : <EOL> [ ( None , '<STR_LIT>' , None , <EOL> { '<STR_LIT>' : { '<STR_LIT>' : None } } ) ] , <EOL> '<STR_LIT>' : <EOL> [ ( None , '<STR_LIT>' , None , <EOL> { '<STR_LIT>' : { '<STR_LIT>' : None } } ) ] , <EOL> '<STR_LIT>' : <EOL> [ ( None , '<STR_LIT>' , None , { } ) ] , <EOL> '<STR_LIT>' : <EOL> [ ( None , '<STR_LIT>' , None , { } ) ] , <EOL> '<STR_LIT>' : <EOL> [ ( None , '<STR_LIT>' , None , { } ) ] , <EOL> '<STR_LIT>' : <EOL> [ ( None , '<STR_LIT>' , None , { } ) ] , <EOL> '<STR_LIT>' : <EOL> [ ( None , '<STR_LIT>' , None , { } ) ] , <EOL> '<STR_LIT>' : <EOL> [ ( None , '<STR_LIT>' , None , { } ) ] , <EOL> } <EOL> return UpgradeWorkflowHandler . remap_module ( controller , module_id , pipeline , <EOL> module_remap ) <EOL> class NewConstant ( Constant ) : <EOL> """<STR_LIT>""" <EOL> def setValue ( self , v ) : <EOL> self . set_output ( "<STR_LIT:value>" , v ) <EOL> self . upToDate = True <EOL> def create_constant ( value ) : <EOL> """<STR_LIT>""" <EOL> constant = NewConstant ( ) <EOL> constant . setValue ( value ) <EOL> return constant <EOL> def get_module ( value , signature = None ) : <EOL> """<STR_LIT>""" <EOL> if isinstance ( value , Constant ) : <EOL> return type ( value ) <EOL> elif isinstance ( value , bool ) : <EOL> return Boolean <EOL> elif isinstance ( value , str ) : <EOL> return String <EOL> elif isinstance ( value , int ) : <EOL> return Integer <EOL> elif isinstance ( value , float ) : <EOL> return Float <EOL> if isinstance ( value , list ) : <EOL> return List <EOL> elif isinstance ( value , tuple ) : <EOL> if signature is None or ( len ( signature ) == <NUM_LIT:1> and signature [ <NUM_LIT:0> ] [ <NUM_LIT:0> ] == Variant ) : <EOL> return ( Variant , ) * len ( value ) <EOL> v_modules = ( ) <EOL> for element in xrange ( len ( value ) ) : <EOL> v_modules += ( get_module ( value [ element ] , signature [ element ] ) , ) <EOL> if None in v_modules : <EOL> return None <EOL> return v_modules <EOL> else : <EOL> debug . warning ( "<STR_LIT>" ) <EOL> debug . warning ( "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> return None <EOL> import sys <EOL> import unittest <EOL> class TestConcatenateString ( unittest . TestCase ) : <EOL> @ staticmethod <EOL> def concatenate ( ** kwargs ) : <EOL> from vistrails . tests . utils import execute , intercept_result <EOL> with intercept_result ( ConcatenateString , '<STR_LIT:value>' ) as results : <EOL> errors = execute ( [ <EOL> ( '<STR_LIT>' , '<STR_LIT>' , [ <EOL> ( name , [ ( '<STR_LIT>' , value ) ] ) <EOL> for name , value in kwargs . iteritems ( ) <EOL> ] ) , <EOL> ] ) <EOL> if errors : <EOL> return None <EOL> return results <EOL> def test_concatenate ( self ) : <EOL> """<STR_LIT>""" <EOL> self . assertEqual ( self . concatenate ( <EOL> str1 = "<STR_LIT>" , str2 = "<STR_LIT>" ) , <EOL> [ "<STR_LIT>" ] ) <EOL> self . assertEqual ( self . concatenate ( <EOL> str3 = "<STR_LIT>" ) , <EOL> [ "<STR_LIT>" ] ) <EOL> self . assertEqual ( self . concatenate ( <EOL> str2 = "<STR_LIT>" , str4 = "<STR_LIT>" ) , <EOL> [ "<STR_LIT>" ] ) <EOL> self . assertEqual ( self . concatenate ( <EOL> str1 = "<STR_LIT:hello>" , str3 = "<STR_LIT:U+0020>" , str4 = "<STR_LIT>" ) , <EOL> [ "<STR_LIT>" ] ) <EOL> def test_empty ( self ) : <EOL> """<STR_LIT>""" <EOL> self . assertEqual ( self . concatenate ( ) , [ "<STR_LIT>" ] ) <EOL> class TestNot ( unittest . TestCase ) : <EOL> def run_pipeline ( self , functions ) : <EOL> from vistrails . tests . utils import execute , intercept_result <EOL> with intercept_result ( Not , '<STR_LIT:value>' ) as results : <EOL> errors = execute ( [ <EOL> ( '<STR_LIT>' , '<STR_LIT>' , <EOL> functions ) , <EOL> ] ) <EOL> return errors , results <EOL> def test_true ( self ) : <EOL> errors , results = self . run_pipeline ( [ <EOL> ( '<STR_LIT:input>' , [ ( '<STR_LIT>' , '<STR_LIT:True>' ) ] ) ] ) <EOL> self . assertFalse ( errors ) <EOL> self . assertEqual ( len ( results ) , <NUM_LIT:1> ) <EOL> self . assertIs ( results [ <NUM_LIT:0> ] , False ) <EOL> def test_false ( self ) : <EOL> errors , results = self . run_pipeline ( [ <EOL> ( '<STR_LIT:input>' , [ ( '<STR_LIT>' , '<STR_LIT:False>' ) ] ) ] ) <EOL> self . assertFalse ( errors ) <EOL> self . assertEqual ( len ( results ) , <NUM_LIT:1> ) <EOL> self . assertIs ( results [ <NUM_LIT:0> ] , True ) <EOL> def test_notset ( self ) : <EOL> errors , results = self . run_pipeline ( [ ] ) <EOL> self . assertTrue ( errors ) <EOL> class TestList ( unittest . TestCase ) : <EOL> @ staticmethod <EOL> def build_list ( value = None , head = None , tail = None ) : <EOL> from vistrails . tests . utils import execute , intercept_result <EOL> with intercept_result ( List , '<STR_LIT:value>' ) as results : <EOL> functions = [ ] <EOL> def add ( n , v , t ) : <EOL> if v is not None : <EOL> for e in v : <EOL> functions . append ( <EOL> ( n , [ ( t , e ) ] ) <EOL> ) <EOL> add ( '<STR_LIT:value>' , value , '<STR_LIT>' ) <EOL> add ( '<STR_LIT>' , head , '<STR_LIT>' ) <EOL> add ( '<STR_LIT>' , tail , '<STR_LIT>' ) <EOL> errors = execute ( [ <EOL> ( '<STR_LIT>' , '<STR_LIT>' , functions ) , <EOL> ] ) <EOL> if errors : <EOL> return None <EOL> return results [ - <NUM_LIT:1> ] <EOL> def test_simple ( self ) : <EOL> """<STR_LIT>""" <EOL> self . assertEqual ( self . build_list ( <EOL> value = [ '<STR_LIT>' ] ) , <EOL> [ "<STR_LIT:a>" , "<STR_LIT:b>" , "<STR_LIT:c>" ] ) <EOL> self . assertEqual ( self . build_list ( <EOL> head = [ "<STR_LIT:d>" ] , <EOL> value = [ '<STR_LIT>' ] ) , <EOL> [ "<STR_LIT:d>" , "<STR_LIT:a>" , "<STR_LIT:b>" , "<STR_LIT:c>" ] ) <EOL> self . assertEqual ( self . build_list ( <EOL> head = [ "<STR_LIT:d>" ] , <EOL> value = [ '<STR_LIT>' ] , <EOL> tail = [ '<STR_LIT>' ] ) , <EOL> [ "<STR_LIT:d>" , "<STR_LIT:a>" , "<STR_LIT:b>" , "<STR_LIT:c>" , "<STR_LIT:e>" , "<STR_LIT:f>" ] ) <EOL> self . assertEqual ( self . build_list ( <EOL> value = [ '<STR_LIT>' ] , <EOL> tail = [ '<STR_LIT>' ] ) , <EOL> [ ] ) <EOL> def test_multiple ( self ) : <EOL> """<STR_LIT>""" <EOL> self . assertEqual ( self . build_list ( <EOL> head = [ "<STR_LIT:a>" , "<STR_LIT:b>" ] ) , <EOL> [ "<STR_LIT:a>" , "<STR_LIT:b>" ] ) <EOL> self . assertEqual ( self . build_list ( <EOL> head = [ "<STR_LIT:a>" , "<STR_LIT:b>" ] , <EOL> value = [ '<STR_LIT>' ] ) , <EOL> [ "<STR_LIT:a>" , "<STR_LIT:b>" , "<STR_LIT:c>" , "<STR_LIT:d>" ] ) <EOL> res = self . build_list ( value = [ '<STR_LIT>' , '<STR_LIT>' ] ) <EOL> self . assertEqual ( res , [ "<STR_LIT:a>" , "<STR_LIT:b>" , "<STR_LIT:c>" , "<STR_LIT:d>" ] ) <EOL> def test_items ( self ) : <EOL> """<STR_LIT>""" <EOL> from vistrails . tests . utils import execute , intercept_result <EOL> def list_with_items ( nb_items , ** kwargs ) : <EOL> with intercept_result ( List , '<STR_LIT:value>' ) as results : <EOL> errors = execute ( [ <EOL> ( '<STR_LIT>' , '<STR_LIT>' , [ <EOL> ( k , [ ( '<STR_LIT>' , v ) ] ) <EOL> for k , v in kwargs . iteritems ( ) <EOL> ] ) , <EOL> ] , <EOL> add_port_specs = [ <EOL> ( <NUM_LIT:0> , '<STR_LIT:input>' , '<STR_LIT>' % i , <EOL> '<STR_LIT>' ) <EOL> for i in xrange ( nb_items ) <EOL> ] ) <EOL> if errors : <EOL> return None <EOL> return results [ - <NUM_LIT:1> ] <EOL> self . assertEqual ( <EOL> list_with_items ( <NUM_LIT:2> , head = "<STR_LIT>" , item0 = "<STR_LIT>" , item1 = "<STR_LIT>" ) , <EOL> [ "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" ] ) <EOL> self . assertIsNone ( <EOL> list_with_items ( <NUM_LIT:3> , head = "<STR_LIT>" , item0 = "<STR_LIT>" , item2 = "<STR_LIT>" ) ) <EOL> class TestPythonSource ( unittest . TestCase ) : <EOL> def test_simple ( self ) : <EOL> """<STR_LIT>""" <EOL> import urllib2 <EOL> from vistrails . tests . utils import execute , intercept_result <EOL> source = '<STR_LIT>' <EOL> source = urllib2 . quote ( source ) <EOL> with intercept_result ( PythonSource , '<STR_LIT>' ) as results : <EOL> self . assertFalse ( execute ( [ <EOL> ( '<STR_LIT>' , '<STR_LIT>' , [ <EOL> ( '<STR_LIT:source>' , [ ( '<STR_LIT>' , source ) ] ) , <EOL> ( '<STR_LIT>' , [ ( '<STR_LIT>' , '<STR_LIT>' ) ] ) <EOL> ] ) , <EOL> ( '<STR_LIT>' , '<STR_LIT>' , [ ] ) , <EOL> ] , <EOL> [ <EOL> ( <NUM_LIT:0> , '<STR_LIT>' , <NUM_LIT:1> , '<STR_LIT:value>' ) , <EOL> ] , <EOL> add_port_specs = [ <EOL> ( <NUM_LIT:0> , '<STR_LIT:input>' , '<STR_LIT>' , <EOL> '<STR_LIT>' ) , <EOL> ( <NUM_LIT:0> , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' ) , <EOL> ] ) ) <EOL> self . assertEqual ( results [ - <NUM_LIT:1> ] , "<STR_LIT>" ) <EOL> class TestNumericConversions ( unittest . TestCase ) : <EOL> def test_full ( self ) : <EOL> from vistrails . tests . utils import execute , intercept_result <EOL> with intercept_result ( Round , '<STR_LIT>' ) as results : <EOL> self . assertFalse ( execute ( [ <EOL> ( '<STR_LIT>' , '<STR_LIT>' , [ <EOL> ( '<STR_LIT:value>' , [ ( '<STR_LIT>' , '<STR_LIT:5>' ) ] ) <EOL> ] ) , <EOL> ( '<STR_LIT>' , '<STR_LIT>' , [ ] ) , <EOL> ( '<STR_LIT>' , '<STR_LIT>' , [ <EOL> ( '<STR_LIT>' , [ ( '<STR_LIT>' , '<STR_LIT>' ) ] ) , <EOL> ( '<STR_LIT>' , [ ( '<STR_LIT>' , '<STR_LIT:+>' ) ] ) , <EOL> ] ) , <EOL> ( '<STR_LIT>' , '<STR_LIT>' , [ <EOL> ( '<STR_LIT>' , [ ( '<STR_LIT>' , '<STR_LIT:True>' ) ] ) , <EOL> ] ) , <EOL> ] , <EOL> [ <EOL> ( <NUM_LIT:0> , '<STR_LIT:value>' , <NUM_LIT:1> , '<STR_LIT:value>' ) , <EOL> ( <NUM_LIT:1> , '<STR_LIT:value>' , <NUM_LIT:2> , '<STR_LIT>' ) , <EOL> ( <NUM_LIT:2> , '<STR_LIT:value>' , <NUM_LIT:3> , '<STR_LIT>' ) , <EOL> ] ) ) <EOL> self . assertEqual ( results , [ <NUM_LIT:7> ] ) <EOL> class TestUnzip ( unittest . TestCase ) : <EOL> def test_unzip_file ( self ) : <EOL> from vistrails . tests . utils import execute , intercept_result <EOL> from vistrails . core . system import vistrails_root_directory <EOL> zipfile = os . path . join ( vistrails_root_directory ( ) , <EOL> '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' ) <EOL> with intercept_result ( Unzip , '<STR_LIT:file>' ) as outfiles : <EOL> self . assertFalse ( execute ( [ <EOL> ( '<STR_LIT>' , '<STR_LIT>' , [ <EOL> ( '<STR_LIT>' , [ ( '<STR_LIT>' , zipfile ) ] ) , <EOL> ( '<STR_LIT>' , [ ( '<STR_LIT>' , '<STR_LIT>' ) ] ) , <EOL> ] ) , <EOL> ] ) ) <EOL> self . assertEqual ( len ( outfiles ) , <NUM_LIT:1> ) <EOL> with open ( outfiles [ <NUM_LIT:0> ] . name , '<STR_LIT:rb>' ) as outfile : <EOL> self . assertEqual ( outfile . read ( ) , "<STR_LIT>" ) <EOL> def test_unzip_all ( self ) : <EOL> from vistrails . tests . utils import execute , intercept_result <EOL> from vistrails . core . system import vistrails_root_directory <EOL> zipfile = os . path . join ( vistrails_root_directory ( ) , <EOL> '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' ) <EOL> with intercept_result ( UnzipDirectory , '<STR_LIT>' ) as outdir : <EOL> self . assertFalse ( execute ( [ <EOL> ( '<STR_LIT>' , '<STR_LIT>' , [ <EOL> ( '<STR_LIT>' , [ ( '<STR_LIT>' , zipfile ) ] ) , <EOL> ] ) , <EOL> ] ) ) <EOL> self . assertEqual ( len ( outdir ) , <NUM_LIT:1> ) <EOL> self . assertEqual ( <EOL> [ ( d , f ) for p , d , f in os . walk ( outdir [ <NUM_LIT:0> ] . name ) ] , <EOL> [ ( [ '<STR_LIT>' ] , [ '<STR_LIT>' ] ) , <EOL> ( [ ] , [ '<STR_LIT>' ] ) ] ) <EOL> from vistrails . core . configuration import get_vistrails_configuration <EOL> class TestTypechecking ( unittest . TestCase ) : <EOL> @ classmethod <EOL> def setUpClass ( cls ) : <EOL> conf = get_vistrails_configuration ( ) <EOL> cls . error_all = conf . showConnectionErrors <EOL> cls . error_variant = conf . showVariantErrors <EOL> @ classmethod <EOL> def tearDownClass ( cls ) : <EOL> conf = get_vistrails_configuration ( ) <EOL> conf . showConnectionErrors = cls . error_all <EOL> conf . showVariantErrors = cls . error_variant <EOL> @ staticmethod <EOL> def set_settings ( error_all , error_variant ) : <EOL> conf = get_vistrails_configuration ( ) <EOL> conf . showConnectionErrors = error_all <EOL> conf . showVariantErrors = error_variant <EOL> def run_test_pipeline ( self , result , expected_results , * args , ** kwargs ) : <EOL> from vistrails . tests . utils import execute , intercept_result <EOL> for error_all , error_variant , expected in expected_results : <EOL> self . set_settings ( error_all , error_variant ) <EOL> with intercept_result ( * result ) as results : <EOL> error = execute ( * args , ** kwargs ) <EOL> if not expected : <EOL> self . assertTrue ( error ) <EOL> else : <EOL> self . assertFalse ( error ) <EOL> self . assertEqual ( results , expected ) <EOL> def test_basic ( self ) : <EOL> import urllib2 <EOL> self . run_test_pipeline ( <EOL> ( PythonSource , '<STR_LIT:r>' ) , <EOL> [ ( False , False , [ "<STR_LIT:test>" ] ) , <EOL> ( True , True , [ "<STR_LIT:test>" ] ) ] , <EOL> [ <EOL> ( '<STR_LIT>' , '<STR_LIT>' , [ <EOL> ( '<STR_LIT:source>' , [ ( '<STR_LIT>' , urllib2 . quote ( '<STR_LIT>' ) ) ] ) , <EOL> ] ) , <EOL> ( '<STR_LIT>' , '<STR_LIT>' , [ <EOL> ( '<STR_LIT:source>' , [ ( '<STR_LIT>' , urllib2 . quote ( '<STR_LIT>' ) ) ] ) <EOL> ] ) , <EOL> ] , <EOL> [ <EOL> ( <NUM_LIT:0> , '<STR_LIT:o>' , <NUM_LIT:1> , '<STR_LIT:i>' ) , <EOL> ] , <EOL> add_port_specs = [ <EOL> ( <NUM_LIT:0> , '<STR_LIT>' , '<STR_LIT:o>' , <EOL> '<STR_LIT>' ) , <EOL> ( <NUM_LIT:1> , '<STR_LIT:input>' , '<STR_LIT:i>' , <EOL> '<STR_LIT>' ) , <EOL> ( <NUM_LIT:1> , '<STR_LIT>' , '<STR_LIT:r>' , <EOL> '<STR_LIT>' ) <EOL> ] ) <EOL> def test_fake ( self ) : <EOL> import urllib2 <EOL> self . run_test_pipeline ( <EOL> ( PythonSource , '<STR_LIT:r>' ) , <EOL> [ ( False , False , [ <NUM_LIT> ] ) , <EOL> ( False , True , [ <NUM_LIT> ] ) , <EOL> ( True , True , False ) ] , <EOL> [ <EOL> ( '<STR_LIT>' , '<STR_LIT>' , [ <EOL> ( '<STR_LIT:source>' , [ ( '<STR_LIT>' , urllib2 . quote ( '<STR_LIT>' ) ) ] ) , <EOL> ] ) , <EOL> ( '<STR_LIT>' , '<STR_LIT>' , [ <EOL> ( '<STR_LIT:source>' , [ ( '<STR_LIT>' , urllib2 . quote ( '<STR_LIT>' ) ) ] ) <EOL> ] ) , <EOL> ] , <EOL> [ <EOL> ( <NUM_LIT:0> , '<STR_LIT:o>' , <NUM_LIT:1> , '<STR_LIT:i>' ) , <EOL> ] , <EOL> add_port_specs = [ <EOL> ( <NUM_LIT:0> , '<STR_LIT>' , '<STR_LIT:o>' , <EOL> '<STR_LIT>' ) , <EOL> ( <NUM_LIT:1> , '<STR_LIT:input>' , '<STR_LIT:i>' , <EOL> '<STR_LIT>' ) , <EOL> ( <NUM_LIT:1> , '<STR_LIT>' , '<STR_LIT:r>' , <EOL> '<STR_LIT>' ) <EOL> ] ) <EOL> def test_inputport ( self ) : <EOL> import urllib2 <EOL> self . run_test_pipeline ( <EOL> ( PythonSource , '<STR_LIT:r>' ) , <EOL> [ ( False , False , [ <NUM_LIT> ] ) , <EOL> ( False , True , [ <NUM_LIT> ] ) , <EOL> ( True , True , [ <NUM_LIT> ] ) ] , <EOL> [ <EOL> ( '<STR_LIT>' , '<STR_LIT>' , [ <EOL> ( '<STR_LIT>' , [ ( '<STR_LIT>' , '<STR_LIT>' ) ] ) , <EOL> ] ) , <EOL> ( '<STR_LIT>' , '<STR_LIT>' , [ <EOL> ( '<STR_LIT:source>' , [ ( '<STR_LIT>' , urllib2 . quote ( '<STR_LIT>' ) ) ] ) <EOL> ] ) , <EOL> ] , <EOL> [ <EOL> ( <NUM_LIT:0> , '<STR_LIT>' , <NUM_LIT:1> , '<STR_LIT:i>' ) , <EOL> ] , <EOL> add_port_specs = [ <EOL> ( <NUM_LIT:1> , '<STR_LIT:input>' , '<STR_LIT:i>' , <EOL> '<STR_LIT>' ) , <EOL> ( <NUM_LIT:1> , '<STR_LIT>' , '<STR_LIT:r>' , <EOL> '<STR_LIT>' ) , <EOL> ] ) <EOL> class TestStringFormat ( unittest . TestCase ) : <EOL> def test_list_placeholders ( self ) : <EOL> fmt = '<STR_LIT>' <EOL> self . assertEqual ( StringFormat . list_placeholders ( fmt ) , <EOL> ( <NUM_LIT:2> , set ( [ '<STR_LIT:c>' ] ) ) ) <EOL> def run_format ( self , fmt , expected , ** kwargs ) : <EOL> from vistrails . tests . utils import execute , intercept_result <EOL> functions = [ ( '<STR_LIT>' , [ ( '<STR_LIT>' , fmt ) ] ) ] <EOL> functions . extend ( ( n , [ ( t , v ) ] ) <EOL> for n , ( t , v ) in kwargs . iteritems ( ) ) <EOL> with intercept_result ( StringFormat , '<STR_LIT:value>' ) as results : <EOL> self . assertFalse ( execute ( [ <EOL> ( '<STR_LIT>' , '<STR_LIT>' , <EOL> functions ) , <EOL> ] , <EOL> add_port_specs = [ <EOL> ( <NUM_LIT:0> , '<STR_LIT:input>' , n , t ) <EOL> for n , ( t , v ) in kwargs . iteritems ( ) <EOL> ] ) ) <EOL> self . assertEqual ( results , [ expected ] ) <EOL> def test_format ( self ) : <EOL> self . run_format ( '<STR_LIT>' , '<STR_LIT>' , <EOL> a = ( '<STR_LIT>' , '<STR_LIT>' ) , <EOL> c = ( '<STR_LIT>' , '<STR_LIT>' ) ) <EOL> @ unittest . skipIf ( sys . version_info < ( <NUM_LIT:2> , <NUM_LIT:7> ) , "<STR_LIT>" ) <EOL> def test_format_27 ( self ) : <EOL> self . run_format ( '<STR_LIT>' , '<STR_LIT>' , <EOL> _0 = ( '<STR_LIT>' , '<STR_LIT:a>' ) , _1 = ( '<STR_LIT>' , '<STR_LIT:b>' ) ) <EOL> self . run_format ( '<STR_LIT>' , '<STR_LIT>' , <EOL> a = ( '<STR_LIT>' , '<STR_LIT>' ) , _0 = ( '<STR_LIT>' , '<STR_LIT:b>' ) , <EOL> b = ( '<STR_LIT>' , '<STR_LIT>' ) ) <EOL> self . run_format ( '<STR_LIT>' , "<STR_LIT>" , <EOL> _0 = ( '<STR_LIT>' , '<STR_LIT:hello>' ) , _1 = ( '<STR_LIT>' , '<STR_LIT>' ) , <EOL> _2 = ( '<STR_LIT>' , '<STR_LIT>' ) , _3 = ( '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ponc = ( '<STR_LIT>' , '<STR_LIT:!>' ) ) <EOL> class TestConstantMetaclass ( unittest . TestCase ) : <EOL> def test_meta ( self ) : <EOL> """<STR_LIT>""" <EOL> mod1_in = [ ( '<STR_LIT:value>' , '<STR_LIT>' ) , IPort ( '<STR_LIT>' , '<STR_LIT>' ) ] <EOL> mod1_out = [ ( '<STR_LIT>' , '<STR_LIT>' ) ] <EOL> class Mod1 ( Constant ) : <EOL> _input_ports = mod1_in <EOL> _output_ports = mod1_out <EOL> self . assertEqual ( Mod1 . _input_ports , mod1_in ) <EOL> self . assertEqual ( Mod1 . _output_ports , [ ( '<STR_LIT:value>' , Mod1 ) ] + mod1_out ) <EOL> mod2_in = [ ( '<STR_LIT>' , '<STR_LIT>' ) ] <EOL> class Mod2 ( Mod1 ) : <EOL> _input_ports = mod2_in <EOL> self . assertEqual ( Mod2 . _input_ports , [ ( '<STR_LIT:value>' , Mod2 ) ] + mod2_in ) <EOL> self . assertEqual ( Mod2 . _output_ports , [ ( '<STR_LIT:value>' , Mod2 ) ] ) <EOL> class Mod3 ( Mod1 ) : <EOL> _output_ports = [ ] <EOL> self . assertEqual ( Mod3 . _input_ports , [ ( '<STR_LIT:value>' , Mod3 ) ] ) <EOL> self . assertEqual ( Mod3 . _output_ports , [ ( '<STR_LIT:value>' , Mod3 ) ] ) </s>
