New subject: Illumina Export Format ( subject was: why is my lastz process showing it is not finished?)

10 Nov 2010

def get_available_tracks(self):
    return self.available_tracks

def get_mimetype_by_extension(self, ext, default = 'application/octet-stream' ):
    """Returns a mimetype based on an extension"""
    try:
        mimetype = self.mimetypes_by_extension[ext]
    except KeyError:
        #datatype was never declared
        mimetype = default
        self.log.warning('unknown mimetype in data factory %s' % ext)
    return mimetype

def get_datatype_by_extension(self, ext ):
    """Returns a datatype based on an extension"""
    try:
        builder = self.datatypes_by_extension[ext]
    except KeyError:
        builder = data.Text()
    return builder

def change_datatype(self, data, ext, set_meta = True ):
    data.extension = ext
    # call init_meta and copy metadata from itself.  The datatype
    # being converted *to* will handle any metadata copying and
    # initialization.
    if data.has_data():
        data.set_size()
        data.init_meta( copy_from=data )
        if set_meta:
            #metadata is being set internally
            data.set_meta( overwrite = False )
            data.set_peek()
    return data

def old_change_datatype(self, data, ext):
    """Creates and returns a new datatype based on an existing data and an extension"""
    newdata = factory(ext)(id=data.id)
    for key, value in data.__dict__.items():
        setattr(newdata, key, value)
    newdata.ext = ext
    return newdata

def load_datatype_converters( self, toolbox ):
    """Adds datatype converters from self.converters to the calling app's toolbox"""     
    for elem in self.converters:
        tool_config = elem[0]
        source_datatype = elem[1]
        target_datatype = elem[2]
        converter_path = os.path.join( self.datatype_converters_path, tool_config )
        try:
            converter = toolbox.load_tool( converter_path )
            toolbox.tools_by_id[converter.id] = converter
            if source_datatype not in self.datatype_converters:
                self.datatype_converters[source_datatype] = odict()
            self.datatype_converters[source_datatype][target_datatype] = converter
            self.log.debug( "Loaded converter: %s", converter.id )
        except:
            self.log.exception( "error reading converter from path: %s" % converter_path )

def load_external_metadata_tool( self, toolbox ):
    """Adds a tool which is used to set external metadata"""
    #we need to be able to add a job to the queue to set metadata. The queue will currently only accept jobs with an associated tool.
    #We'll create a special tool to be used for Auto-Detecting metadata; this is less than ideal, but effective
    #Properly building a tool without relying on parsing an XML file is near impossible...so we'll create a temporary file 
    tool_xml_text = """
        <tool id="__SET_METADATA__" name="Set External Metadata" version="1.0.1" tool_type="set_metadata">
          <type class="SetMetadataTool" module="galaxy.tools"/>
          <action module="galaxy.tools.actions.metadata" class="SetMetadataToolAction"/>
          <command>$__SET_EXTERNAL_METADATA_COMMAND_LINE__</command>
          <inputs>
            <param format="data" name="input1" type="data" label="File to set metadata on."/>
            <param name="__ORIGINAL_DATASET_STATE__" type="hidden" value=""/>
            <param name="__SET_EXTERNAL_METADATA_COMMAND_LINE__" type="hidden" value=""/>
          </inputs>
        </tool>
        """
    tmp_name = tempfile.NamedTemporaryFile()
    tmp_name.write( tool_xml_text )
    tmp_name.flush()
    set_meta_tool = toolbox.load_tool( tmp_name.name )
    toolbox.tools_by_id[ set_meta_tool.id ] = set_meta_tool
    self.set_external_metadata_tool = set_meta_tool
    self.log.debug( "Loaded external metadata tool: %s", self.set_external_metadata_tool.id )

def load_datatype_indexers( self, toolbox ):
    """Adds indexers from self.indexers to the toolbox from app"""
    for elem in self.indexers:
        tool_config = elem[0]
        datatype = elem[1]
        indexer = toolbox.load_tool( os.path.join( self.datatype_indexers_path, tool_config ) )
        toolbox.tools_by_id[indexer.id] = indexer
        self.datatype_indexers[datatype] = indexer
        self.log.debug( "Loaded indexer: %s", indexer.id )

def get_converters_by_datatype(self, ext):
    """Returns available converters by source type"""
    converters = odict()
    source_datatype = type(self.get_datatype_by_extension(ext))
    for ext2, dict in self.datatype_converters.items():
        converter_datatype = type(self.get_datatype_by_extension(ext2))
        if issubclass(source_datatype, converter_datatype):
            converters.update(dict)
    #Ensure ext-level converters are present
    if ext in self.datatype_converters.keys():
        converters.update(self.datatype_converters[ext])
    return converters

def get_indexers_by_datatype( self, ext ):
    """Returns indexers based on datatype"""
    class_chain = list()
    source_datatype = type(self.get_datatype_by_extension(ext))
    for ext_spec in self.datatype_indexers.keys():
        datatype = type(self.get_datatype_by_extension(ext_spec))
        if issubclass( source_datatype, datatype ):
            class_chain.append( ext_spec )
    # Prioritize based on class chain
    ext2type = lambda x: self.get_datatype_by_extension(x)
    class_chain = sorted(class_chain, lambda x,y: issubclass(ext2type(x),ext2type(y)) and -1 or 1)
    return [self.datatype_indexers[x] for x in class_chain]

def get_converter_by_target_type(self, source_ext, target_ext):
    """Returns a converter based on source and target datatypes"""
    converters = self.get_converters_by_datatype(source_ext)
    if target_ext in converters.keys():
        return converters[target_ext]
    return None

def find_conversion_destination_for_dataset_by_extensions( self, dataset, accepted_formats, converter_safe = True ):
    """Returns ( target_ext, existing converted dataset )"""
    for convert_ext in self.get_converters_by_datatype( dataset.ext ):
        if isinstance( self.get_datatype_by_extension( convert_ext ), accepted_formats ):
            dataset = dataset.get_converted_files_by_type( convert_ext )
            if dataset:
                ret_data = dataset
            elif not converter_safe:
                continue
            else:
                ret_data = None
            return ( convert_ext, ret_data )
    return ( None, None )

def get_composite_extensions( self ):
    return [ ext for ( ext, d_type ) in self.datatypes_by_extension.iteritems() if d_type.composite_type is not None ]

def get_upload_metadata_params( self, context, group, tool ):
    """Returns dict of case value:inputs for metadata conditional for upload tool"""
    rval = {}
    for ext, d_type in self.datatypes_by_extension.iteritems():
        inputs = []
        for meta_name, meta_spec in d_type.metadata_spec.iteritems():
            if meta_spec.set_in_upload:
                help_txt = meta_spec.desc
                if not help_txt or help_txt == meta_name:
                    help_txt = ""
                inputs.append( '<param type="text" name="%s" label="Set metadata value for &quot;%s&quot;" value="%s" help="%s"/>' % ( meta_name, meta_name, meta_spec.default, help_txt ) )
        rval[ ext ] = "\n".join( inputs )
    if 'auto' not in rval and 'txt' in rval: #need to manually add 'auto' datatype
        rval[ 'auto' ] = rval[ 'txt' ]
    return rval