Source code for galaxy.datatypes.anvio

"""
Datatypes for Anvi'o
https://github.com/merenlab/anvio
"""
import glob
import logging
import os
import sys
from typing import Optional

from galaxy.datatypes.metadata import MetadataElement
from galaxy.datatypes.text import Html

log = logging.getLogger(__name__)


[docs]class AnvioComposite(Html):
    """
    Base class to use for Anvi'o composite datatypes.
    Generally consist of a sqlite database, plus optional additional files
    """
    file_ext = "anvio_composite"
    composite_type = 'auto_primary_file'

[docs]    def generate_primary_file(self, dataset=None):
        """
        This is called only at upload to write the html file
        cannot rename the datasets here - they come with the default unfortunately
        """
        defined_files = self.get_composite_files(dataset=dataset).items()
        rval = [f"<html><head><title>Files for Anvi'o Composite Dataset ({self.file_ext})</title></head>"]
        if defined_files:
            rval.append("<p/>This composite dataset is composed of the following defined files:<p/><ul>")
            for composite_name, composite_file in defined_files:
                opt_text = ''
                if composite_file.optional:
                    opt_text = ' (optional)'
                missing_text = ''
                if not os.path.exists(os.path.join(dataset.extra_files_path, composite_name)):
                    missing_text = ' (missing)'
                rval.append(f'<li><a href="{composite_name}">{composite_name}</a>{opt_text}{missing_text}</li>')
            rval.append("</ul>")
        defined_files = map(lambda x: x[0], defined_files)
        extra_files = []
        for dirpath, _dirnames, filenames in os.walk(dataset.extra_files_path, followlinks=True):
            for filename in filenames:
                rel_path = os.path.relpath(os.path.join(dirpath, filename), dataset.extra_files_path)
                if rel_path not in defined_files:
                    extra_files.append(rel_path)
        if extra_files:
            rval.append("<p/>This composite dataset contains these undefined files:<p/><ul>")
            for rel_path in extra_files:
                rval.append(f'<li><a href="{rel_path}">{rel_path}</a></li>')
            rval.append('</ul>')
        if not (defined_files or extra_files):
            rval.append("<p/>This composite dataset does not contain any files!<p/><ul>")
        rval.append('</html>')
        return "\n".join(rval)

[docs]    def get_mime(self):
        """Returns the mime type of the datatype"""
        return 'text/html'

[docs]    def set_peek(self, dataset, is_multi_byte=False):
        """Set the peek and blurb text"""
        if not dataset.dataset.purged:
            dataset.peek = 'Anvio database (multiple files)'
            dataset.blurb = 'Anvio database (multiple files)'
        else:
            dataset.peek = 'file does not exist'
            dataset.blurb = 'file purged from disk'

[docs]    def display_peek(self, dataset):
        """Create HTML content, used for displaying peek."""
        try:
            return dataset.peek
        except Exception:
            return "Anvio database (multiple files)"


[docs]class AnvioDB(AnvioComposite):
    """Class for AnvioDB database files."""
    _anvio_basename: Optional[str] = None
    MetadataElement(name="anvio_basename", default=_anvio_basename, desc="Basename", readonly=True)
    file_ext = 'anvio_db'

[docs]    def __init__(self, *args, **kwd):
        super().__init__(*args, **kwd)
        if self._anvio_basename is not None:
            self.add_composite_file(self._anvio_basename, is_binary=True, optional=False)

[docs]    def set_meta(self, dataset, **kwd):
        """
        Set the anvio_basename based upon actual extra_files_path contents.
        """
        super().set_meta(dataset, **kwd)
        if dataset.metadata.anvio_basename is not None and os.path.exists(os.path.join(dataset.extra_files_path, dataset.metadata.anvio_basename)):
            return
        found = False
        for basename in [dataset.metadata.anvio_basename, self._anvio_basename]:
            if found:
                break
            if basename is not None and not os.path.exists(os.path.join(dataset.extra_files_path, basename)):
                for name in glob.glob(os.path.join(dataset.extra_files_path, f"*{basename}")):
                    dataset.metadata.anvio_basename = os.path.basename(name)
                    found = True
                    break


[docs]class AnvioStructureDB(AnvioDB):
    """Class for Anvio Structure DB database files."""
    _anvio_basename = 'STRUCTURE.db'
    MetadataElement(name="anvio_basename", default=_anvio_basename, desc="Basename", readonly=True)
    file_ext = 'anvio_structure_db'


[docs]class AnvioGenomesDB(AnvioDB):
    """Class for Anvio Genomes DB database files."""
    _anvio_basename = '-GENOMES.db'
    MetadataElement(name="anvio_basename", default=_anvio_basename, desc="Basename", readonly=True)
    file_ext = 'anvio_genomes_db'


[docs]class AnvioContigsDB(AnvioDB):
    """Class for Anvio Contigs DB database files."""
    _anvio_basename = 'CONTIGS.db'
    MetadataElement(name="anvio_basename", default=_anvio_basename, desc="Basename", readonly=True)
    file_ext = 'anvio_contigs_db'

[docs]    def __init__(self, *args, **kwd):
        super().__init__(*args, **kwd)
        self.add_composite_file('CONTIGS.h5', is_binary=True, optional=True)


[docs]class AnvioProfileDB(AnvioDB):
    """Class for Anvio Profile DB database files."""
    _anvio_basename = 'PROFILE.db'
    MetadataElement(name="anvio_basename", default=_anvio_basename, desc="Basename", readonly=True)
    file_ext = 'anvio_profile_db'

[docs]    def __init__(self, *args, **kwd):
        super().__init__(*args, **kwd)
        self.add_composite_file('RUNINFO.cp', is_binary=True, optional=True)
        self.add_composite_file('RUNINFO.mcp', is_binary=True, optional=True)
        self.add_composite_file('AUXILIARY_DATA.db', is_binary=True, optional=True)
        self.add_composite_file('RUNLOG.txt', is_binary=False, optional=True)


[docs]class AnvioPanDB(AnvioDB):
    """Class for Anvio Pan DB database files."""
    _anvio_basename = 'PAN.db'
    MetadataElement(name="anvio_basename", default=_anvio_basename, desc="Basename", readonly=True)
    file_ext = 'anvio_pan_db'


[docs]class AnvioSamplesDB(AnvioDB):
    """Class for Anvio Samples DB database files."""
    _anvio_basename = 'SAMPLES.db'
    MetadataElement(name="anvio_basename", default=_anvio_basename, desc="Basename", readonly=True)
    file_ext = 'anvio_samples_db'


if __name__ == '__main__':
    import doctest
    doctest.testmod(sys.modules[__name__])