<s> from __future__ import division <EOL> from vistrails . core . modules . config import ModuleSettings <EOL> from vistrails . core . modules . module_registry import get_module_registry <EOL> from vistrails . core . modules . vistrails_module import Module , ModuleError , ModuleSuspended <EOL> from vistrails . core . system import current_user <EOL> from vistrails . core . upgradeworkflow import UpgradeWorkflowHandler <EOL> from vistrails . core . vistrail . job import JobMixin <EOL> from remoteq . pipelines . shell import FileCommander as BQMachine <EOL> from remoteq . core . stack import select_machine , end_machine , use_machine , current_machine <EOL> from remoteq . batch . commandline import PBS , PBSScript , Subshell <EOL> from remoteq . batch . directories import CreateDirectory <EOL> from remoteq . batch . files import TransferFiles <EOL> class Machine ( Module ) : <EOL> _input_ports = [ ( '<STR_LIT>' , '<STR_LIT>' , True ) , <EOL> ( '<STR_LIT:port>' , '<STR_LIT>' , True ) , <EOL> ( '<STR_LIT:username>' , '<STR_LIT>' , True ) , <EOL> ( '<STR_LIT:password>' , '<STR_LIT>' , True ) , <EOL> ] <EOL> _output_ports = [ ( '<STR_LIT:value>' , '<STR_LIT>' ) ] <EOL> def compute ( self ) : <EOL> server = self . get_input ( '<STR_LIT>' ) if self . has_input ( '<STR_LIT>' ) else '<STR_LIT:localhost>' <EOL> port = self . get_input ( '<STR_LIT:port>' ) if self . has_input ( '<STR_LIT:port>' ) else <NUM_LIT> <EOL> username = self . get_input ( '<STR_LIT:username>' ) if self . has_input ( '<STR_LIT:username>' ) else current_user ( ) <EOL> password = self . get_input ( '<STR_LIT:password>' ) if self . has_input ( '<STR_LIT:password>' ) else '<STR_LIT>' <EOL> self . machine = Machine . create_machine ( server , username , password , port ) <EOL> self . set_output ( "<STR_LIT:value>" , self . machine ) <EOL> @ staticmethod <EOL> def validate ( v ) : <EOL> return isinstance ( v , BQMachine ) <EOL> @ staticmethod <EOL> def create_machine ( server , username , password , port ) : <EOL> machine = BQMachine ( server , username , password , port , <EOL> accept_fingerprint = True ) <EOL> machine . params = { } <EOL> machine . params [ '<STR_LIT>' ] = server <EOL> machine . params [ '<STR_LIT:port>' ] = port <EOL> machine . params [ '<STR_LIT:username>' ] = username <EOL> machine . params [ '<STR_LIT:password>' ] = bool ( password ) <EOL> select_machine ( machine ) <EOL> end_machine ( ) <EOL> return machine <EOL> class RQModule ( JobMixin , Module ) : <EOL> """<STR_LIT>""" <EOL> _settings = ModuleSettings ( abstract = True ) <EOL> default_machine = None <EOL> machine = None <EOL> def get_machine ( self ) : <EOL> if self . has_input ( '<STR_LIT>' ) : <EOL> return self . get_input ( '<STR_LIT>' ) <EOL> default = False <EOL> machine = self . get_job_machine ( ) <EOL> if not machine : <EOL> machine = self . get_default_machine ( ) <EOL> default = True <EOL> if not machine : <EOL> raise ModuleError ( self , '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> server , port , username , password = machine <EOL> if RQModule . default_machine and ( server , port , username ) == RQModule . default_machine [ : <NUM_LIT:3> ] : <EOL> return RQModule . default_machine [ <NUM_LIT:3> ] <EOL> if password : <EOL> text = '<STR_LIT>' % ( username , server ) <EOL> from PyQt4 import QtGui <EOL> ( password , ok ) = QtGui . QInputDialog . getText ( None , text , text , <EOL> QtGui . QLineEdit . Password ) <EOL> if not ok : <EOL> raise ModuleError ( self , "<STR_LIT>" ) <EOL> machine = Machine . create_machine ( server , username , password , port ) <EOL> if default : <EOL> RQModule . default_machine = ( server , port , username , machine ) <EOL> return machine <EOL> def get_job_machine ( self ) : <EOL> """<STR_LIT>""" <EOL> jm = self . job_monitor ( ) <EOL> if jm . hasJob ( self . signature ) : <EOL> params = jm . getJob ( self . signature ) . parameters <EOL> if '<STR_LIT>' in params : <EOL> return ( params [ '<STR_LIT>' ] , <EOL> params [ '<STR_LIT:port>' ] , <EOL> params [ '<STR_LIT:username>' ] , <EOL> params [ '<STR_LIT:password>' ] ) <EOL> def set_job_machine ( self , params , machine ) : <EOL> """<STR_LIT>""" <EOL> params [ '<STR_LIT>' ] = machine . params [ '<STR_LIT>' ] <EOL> params [ '<STR_LIT:port>' ] = machine . params [ '<STR_LIT:port>' ] <EOL> params [ '<STR_LIT:username>' ] = machine . params [ '<STR_LIT:username>' ] <EOL> params [ '<STR_LIT:password>' ] = machine . params [ '<STR_LIT:password>' ] <EOL> def get_default_machine ( self ) : <EOL> """<STR_LIT>""" <EOL> server = username = port = password = '<STR_LIT>' <EOL> if configuration . check ( '<STR_LIT>' ) : <EOL> server = configuration . server <EOL> if not server : <EOL> return None <EOL> if configuration . check ( '<STR_LIT:username>' ) : <EOL> username = configuration . username <EOL> if not username : <EOL> username = current_user ( ) <EOL> if configuration . check ( '<STR_LIT:port>' ) is not None : <EOL> port = configuration . port <EOL> if configuration . check ( '<STR_LIT:password>' ) : <EOL> password = configuration . password <EOL> self . annotate ( { '<STR_LIT>' : server , <EOL> '<STR_LIT>' : username , <EOL> '<STR_LIT>' : port } ) <EOL> return server , port , username , password <EOL> class RunCommand ( RQModule ) : <EOL> """<STR_LIT>""" <EOL> _input_ports = [ ( '<STR_LIT>' , Machine ) , <EOL> ( '<STR_LIT>' , '<STR_LIT>' , True ) , <EOL> ] <EOL> _output_ports = [ ( '<STR_LIT>' , Machine ) , <EOL> ( '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ] <EOL> def compute ( self ) : <EOL> machine = self . get_machine ( ) <EOL> jm = self . job_monitor ( ) <EOL> cache = jm . getCache ( self . signature ) <EOL> if cache : <EOL> result = cache . parameters [ '<STR_LIT:result>' ] <EOL> else : <EOL> if not self . has_input ( '<STR_LIT>' ) : <EOL> raise ModuleError ( self , "<STR_LIT>" ) <EOL> command = self . get_input ( '<STR_LIT>' ) . strip ( ) <EOL> use_machine ( machine ) <EOL> m = current_machine ( ) <EOL> result = m . remote . send_command ( command ) <EOL> exitcode = m . remote . last_exitcode ( ) <EOL> end_machine ( ) <EOL> if exitcode != <NUM_LIT:0> : <EOL> raise ModuleError ( self , <EOL> "<STR_LIT>" % <EOL> ( exitcode , result ) ) <EOL> d = { '<STR_LIT:result>' : result } <EOL> self . set_job_machine ( d , machine ) <EOL> jm . setCache ( self . signature , d , self . job_name ( ) ) <EOL> self . set_output ( "<STR_LIT>" , result ) <EOL> self . set_output ( "<STR_LIT>" , machine ) <EOL> class RunJob ( RQModule ) : <EOL> """<STR_LIT>""" <EOL> _input_ports = [ ( '<STR_LIT>' , Machine ) , <EOL> ( '<STR_LIT>' , '<STR_LIT>' , True ) , <EOL> ( '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ] <EOL> _output_ports = [ ( '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ( '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ] <EOL> job = None <EOL> def job_read_inputs ( self ) : <EOL> d = { } <EOL> if not self . has_input ( '<STR_LIT>' ) : <EOL> raise ModuleError ( self , "<STR_LIT>" ) <EOL> d [ '<STR_LIT>' ] = self . get_input ( '<STR_LIT>' ) . strip ( ) <EOL> d [ '<STR_LIT>' ] = self . get_input ( '<STR_LIT>' ) if self . has_input ( '<STR_LIT>' ) else '<STR_LIT:.>' <EOL> return d <EOL> def job_start ( self , params ) : <EOL> work_dir = params [ '<STR_LIT>' ] <EOL> self . machine = self . get_machine ( ) <EOL> use_machine ( self . machine ) <EOL> self . job = Subshell ( "<STR_LIT>" , params [ '<STR_LIT>' ] , work_dir ) <EOL> self . job . run ( ) <EOL> ret = self . job . _ret <EOL> if ret : <EOL> try : <EOL> job_id = int ( ret . split ( '<STR_LIT:\n>' ) [ <NUM_LIT:0> ] ) <EOL> except ValueError : <EOL> end_machine ( ) <EOL> raise ModuleError ( self , "<STR_LIT>" % ret ) <EOL> self . set_job_machine ( params , self . machine ) <EOL> return params <EOL> def job_get_handle ( self , params ) : <EOL> if not self . job : <EOL> self . job_start ( params ) <EOL> return self . job <EOL> def job_finish ( self , params ) : <EOL> params [ '<STR_LIT>' ] = self . job . standard_output ( ) <EOL> params [ '<STR_LIT>' ] = self . job . standard_error ( ) <EOL> if self . job . failed ( ) : <EOL> self . job . _pushw ( ) <EOL> code = self . job . terminal . cat ( "<STR_LIT>" % <EOL> self . job . _identifier_filename ) <EOL> self . job . _popw ( ) <EOL> end_machine ( ) <EOL> raise ModuleError ( self , <EOL> "<STR_LIT>" % <EOL> ( code . strip ( ) , params [ '<STR_LIT>' ] . strip ( ) ) ) <EOL> end_machine ( ) <EOL> return params <EOL> def job_set_results ( self , params ) : <EOL> self . set_output ( '<STR_LIT>' , params [ '<STR_LIT>' ] ) <EOL> self . set_output ( '<STR_LIT>' , params [ '<STR_LIT>' ] ) <EOL> class PBSJob ( RQModule ) : <EOL> _input_ports = [ ( '<STR_LIT>' , Machine ) , <EOL> ( '<STR_LIT>' , '<STR_LIT>' , True ) , <EOL> ( '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ( '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ( '<STR_LIT>' , '<STR_LIT>' , True ) , <EOL> ( '<STR_LIT:time>' , '<STR_LIT>' , True ) , <EOL> ( '<STR_LIT>' , '<STR_LIT>' , True ) , <EOL> ( '<STR_LIT>' , '<STR_LIT>' , True ) , <EOL> ( '<STR_LIT>' , '<STR_LIT>' , True ) , <EOL> ( '<STR_LIT>' , '<STR_LIT>' , True ) , <EOL> ] <EOL> _output_ports = [ ( '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ( '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ( '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ] <EOL> def compute ( self ) : <EOL> machine = self . get_machine ( ) <EOL> if not self . has_input ( '<STR_LIT>' ) : <EOL> raise ModuleError ( self , "<STR_LIT>" ) <EOL> command = self . get_input ( '<STR_LIT>' ) . strip ( ) <EOL> working_directory = self . get_input ( '<STR_LIT>' ) if self . has_input ( '<STR_LIT>' ) else '<STR_LIT:.>' <EOL> if not self . has_input ( '<STR_LIT>' ) : <EOL> raise ModuleError ( self , "<STR_LIT>" ) <EOL> input_directory = self . get_input ( '<STR_LIT>' ) . strip ( ) <EOL> additional_arguments = { '<STR_LIT>' : <NUM_LIT:1> , '<STR_LIT:time>' : - <NUM_LIT:1> , '<STR_LIT>' : False , <EOL> '<STR_LIT>' : <NUM_LIT:1> , '<STR_LIT>' : - <NUM_LIT:1> , '<STR_LIT>' : - <NUM_LIT:1> } <EOL> for k in additional_arguments : <EOL> if self . has_input ( k ) : <EOL> additional_arguments [ k ] = self . get_input ( k ) <EOL> use_machine ( machine ) <EOL> cdir = CreateDirectory ( "<STR_LIT>" , working_directory ) <EOL> trans = TransferFiles ( "<STR_LIT>" , input_directory , working_directory , <EOL> dependencies = [ cdir ] ) <EOL> job = PBS ( "<STR_LIT>" , command , working_directory , dependencies = [ trans ] , <EOL> ** additional_arguments ) <EOL> job . run ( ) <EOL> ret = job . _ret <EOL> if ret : <EOL> try : <EOL> job_id = int ( ret ) <EOL> except ValueError : <EOL> end_machine ( ) <EOL> raise ModuleError ( self , "<STR_LIT>" % ret ) <EOL> finished = job . finished ( ) <EOL> job_info = job . get_job_info ( ) <EOL> if job_info : <EOL> self . annotate ( { '<STR_LIT>' : job . get_job_info ( ) } ) <EOL> if not finished : <EOL> status = job . status ( ) <EOL> if job_info : <EOL> comment = [ line for line in job_info . split ( '<STR_LIT:\n>' ) if line . startswith ( '<STR_LIT>' ) ] <EOL> if comment : <EOL> status += '<STR_LIT>' + comment [ <NUM_LIT:10> : ] <EOL> end_machine ( ) <EOL> raise ModuleSuspended ( self , '<STR_LIT:%s>' % status , handle = job ) <EOL> get_result = TransferFiles ( "<STR_LIT>" , input_directory , working_directory , <EOL> dependencies = [ cdir ] ) <EOL> get_result . run ( ) <EOL> end_machine ( ) <EOL> self . set_output ( "<STR_LIT>" , job . standard_output ( ) ) <EOL> self . set_output ( "<STR_LIT>" , job . standard_error ( ) ) <EOL> files = machine . local . send_command ( "<STR_LIT>" % input_directory ) <EOL> self . set_output ( "<STR_LIT>" , <EOL> [ f . split ( '<STR_LIT:U+0020>' ) [ - <NUM_LIT:1> ] for f in files . split ( '<STR_LIT:\n>' ) [ <NUM_LIT:1> : ] ] ) <EOL> class RunPBSScript ( RQModule ) : <EOL> """<STR_LIT>""" <EOL> _input_ports = [ ( '<STR_LIT>' , Machine ) , <EOL> ( '<STR_LIT>' , '<STR_LIT>' , True ) , <EOL> ( '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ( '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ( '<STR_LIT>' , '<STR_LIT>' , True ) , <EOL> ( '<STR_LIT:time>' , '<STR_LIT>' , True ) , <EOL> ( '<STR_LIT>' , '<STR_LIT>' , True ) , <EOL> ( '<STR_LIT>' , '<STR_LIT>' , True ) , <EOL> ( '<STR_LIT>' , '<STR_LIT>' , True ) , <EOL> ( '<STR_LIT>' , '<STR_LIT>' , True ) , <EOL> ] <EOL> _output_ports = [ ( '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ( '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ] <EOL> job = None <EOL> def job_read_inputs ( self ) : <EOL> d = { } <EOL> if not self . has_input ( '<STR_LIT>' ) : <EOL> raise ModuleError ( self , "<STR_LIT>" ) <EOL> d [ '<STR_LIT>' ] = self . get_input ( '<STR_LIT>' ) . strip ( ) <EOL> d [ '<STR_LIT>' ] = self . get_input ( '<STR_LIT>' ) if self . has_input ( '<STR_LIT>' ) else '<STR_LIT:.>' <EOL> if not self . has_input ( '<STR_LIT>' ) : <EOL> raise ModuleError ( self , "<STR_LIT>" ) <EOL> d [ '<STR_LIT>' ] = self . get_input ( '<STR_LIT>' ) . strip ( ) <EOL> d [ '<STR_LIT>' ] = { '<STR_LIT>' : <NUM_LIT:1> , '<STR_LIT:time>' : - <NUM_LIT:1> , '<STR_LIT>' : False , <EOL> '<STR_LIT>' : <NUM_LIT:1> , '<STR_LIT>' : - <NUM_LIT:1> , '<STR_LIT>' : - <NUM_LIT:1> } <EOL> for k in d [ '<STR_LIT>' ] : <EOL> if self . has_input ( k ) : <EOL> d [ '<STR_LIT>' ] [ k ] = self . get_input ( k ) <EOL> return d <EOL> def job_start ( self , params ) : <EOL> work_dir = params [ '<STR_LIT>' ] <EOL> self . machine = self . get_machine ( ) <EOL> use_machine ( self . machine ) <EOL> self . cdir = CreateDirectory ( "<STR_LIT>" , work_dir ) <EOL> trans = TransferFiles ( "<STR_LIT>" , params [ '<STR_LIT>' ] , work_dir , <EOL> dependencies = [ self . cdir ] ) <EOL> self . job = PBSScript ( "<STR_LIT>" , params [ '<STR_LIT>' ] , work_dir , <EOL> dependencies = [ trans ] , ** params [ '<STR_LIT>' ] ) <EOL> self . job . run ( ) <EOL> ret = self . job . _ret <EOL> if ret : <EOL> try : <EOL> job_id = int ( ret . split ( '<STR_LIT:\n>' ) [ <NUM_LIT:0> ] ) <EOL> except ValueError : <EOL> end_machine ( ) <EOL> raise ModuleError ( self , "<STR_LIT>" % ret ) <EOL> self . set_job_machine ( params , self . machine ) <EOL> return params <EOL> def job_get_handle ( self , params ) : <EOL> if not self . job : <EOL> self . job_start ( params ) <EOL> return self . job <EOL> def job_finish ( self , params ) : <EOL> job_info = self . job . get_job_info ( ) <EOL> if job_info : <EOL> self . annotate ( { '<STR_LIT>' : job_info } ) <EOL> get_result = TransferFiles ( "<STR_LIT>" , params [ '<STR_LIT>' ] , <EOL> params [ '<STR_LIT>' ] , <EOL> dependencies = [ self . cdir ] ) <EOL> get_result . run ( ) <EOL> end_machine ( ) <EOL> params [ '<STR_LIT>' ] = self . job . standard_output ( ) <EOL> params [ '<STR_LIT>' ] = self . job . standard_error ( ) <EOL> return params <EOL> def job_set_results ( self , params ) : <EOL> self . set_output ( '<STR_LIT>' , params [ '<STR_LIT>' ] ) <EOL> self . set_output ( '<STR_LIT>' , params [ '<STR_LIT>' ] ) <EOL> class SyncDirectories ( RQModule ) : <EOL> """<STR_LIT>""" <EOL> _input_ports = [ ( '<STR_LIT>' , Machine ) , <EOL> ( '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ( '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ( '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ] <EOL> _output_ports = [ ( '<STR_LIT>' , Machine ) , <EOL> ] <EOL> def compute ( self ) : <EOL> machine = self . get_machine ( ) <EOL> jm = self . job_monitor ( ) <EOL> cache = jm . getCache ( self . signature ) <EOL> if not cache : <EOL> if not self . has_input ( '<STR_LIT>' ) : <EOL> raise ModuleError ( self , "<STR_LIT>" ) <EOL> local_directory = self . get_input ( '<STR_LIT>' ) . strip ( ) <EOL> if not self . has_input ( '<STR_LIT>' ) : <EOL> raise ModuleError ( self , "<STR_LIT>" ) <EOL> remote_directory = self . get_input ( '<STR_LIT>' ) . strip ( ) <EOL> whereto = '<STR_LIT>' <EOL> if self . has_input ( '<STR_LIT>' ) and self . get_input ( '<STR_LIT>' ) : <EOL> whereto = '<STR_LIT>' <EOL> use_machine ( machine ) <EOL> to_dir = local_directory if whereto == '<STR_LIT>' else remote_directory <EOL> cdir = CreateDirectory ( whereto , to_dir ) <EOL> job = TransferFiles ( whereto , local_directory , remote_directory , <EOL> dependencies = [ cdir ] ) <EOL> job . run ( ) <EOL> end_machine ( ) <EOL> d = { } <EOL> self . set_job_machine ( d , machine ) <EOL> cache = jm . setCache ( self . signature , d , self . job_name ( ) ) <EOL> self . set_output ( "<STR_LIT>" , machine ) <EOL> class CopyFile ( RQModule ) : <EOL> """<STR_LIT>""" <EOL> _input_ports = [ ( '<STR_LIT>' , Machine ) , <EOL> ( '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ( '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ( '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ] <EOL> _output_ports = [ ( '<STR_LIT>' , Machine ) , <EOL> ( '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ] <EOL> def compute ( self ) : <EOL> machine = self . get_machine ( ) <EOL> jm = self . job_monitor ( ) <EOL> cache = jm . getCache ( self . signature ) <EOL> if cache : <EOL> result = cache . parameters [ '<STR_LIT:result>' ] <EOL> else : <EOL> if not self . has_input ( '<STR_LIT>' ) : <EOL> raise ModuleError ( self , "<STR_LIT>" ) <EOL> local_file = self . get_input ( '<STR_LIT>' ) . strip ( ) <EOL> if not self . has_input ( '<STR_LIT>' ) : <EOL> raise ModuleError ( self , "<STR_LIT>" ) <EOL> remote_file = self . get_input ( '<STR_LIT>' ) . strip ( ) <EOL> whereto = '<STR_LIT>' <EOL> if self . has_input ( '<STR_LIT>' ) and self . get_input ( '<STR_LIT>' ) : <EOL> whereto = '<STR_LIT>' <EOL> command = machine . getfile if whereto == '<STR_LIT>' else machine . sendfile <EOL> result = command ( local_file , remote_file ) <EOL> d = { '<STR_LIT:result>' : result } <EOL> self . set_job_machine ( d , machine ) <EOL> jm . setCache ( self . signature , d , self . job_name ( ) ) <EOL> self . set_output ( "<STR_LIT>" , machine ) <EOL> self . set_output ( "<STR_LIT>" , result ) <EOL> def handle_module_upgrade_request ( controller , module_id , pipeline ) : <EOL> reg = get_module_registry ( ) <EOL> hadoop_remaps = [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] <EOL> module_remap = { } <EOL> for name in hadoop_remaps : <EOL> module_remap [ name ] = [ ( None , '<STR_LIT>' , "<STR_LIT>" % name , { } ) ] <EOL> return UpgradeWorkflowHandler . remap_module ( controller , <EOL> module_id , <EOL> pipeline , <EOL> module_remap ) <EOL> def initialize ( ) : <EOL> global _modules <EOL> _modules = [ Machine , RQModule , RunPBSScript , RunCommand , RunJob , <EOL> SyncDirectories , CopyFile ] <EOL> import base <EOL> import hdfs <EOL> import streaming <EOL> _modules . extend ( base . register ( ) ) <EOL> _modules . extend ( hdfs . register ( ) ) <EOL> _modules . extend ( streaming . register ( ) ) <EOL> _modules = [ ] </s>
