comparison data_manager/fetch_mothur_reference_data.py @ 4:0db22932bc39 draft

"planemo upload for repository https://github.com/galaxyproject/tools-iuc/tree/master/data_managers/data_manager_mothur_toolsuite/ commit f845716f6ac93500f143a30abef97eaba406344e"

3:9d09724f2bf1

#!/usr/bin/env python
#
# Data manager for reference data for the 'mothur_toolsuite' Galaxy tools
import json
import optparse
import os
import shutil
import sys
import tarfile
import tempfile
import urllib2
import zipfile
from functools import reduce

# When extracting files from archives, skip names that
# start with the following strings

    "lookup_gs20": {
        "GS20": ["https://mothur.s3.us-east-2.amazonaws.com/wiki/lookup_gs20.zip", ]
    },
    # RDP reference files
    # http://www.mothur.org/wiki/RDP_reference_files
    "RDP_v16": {
        "16S rRNA RDP training set 16":
        ["https://mothur.s3.us-east-2.amazonaws.com/wiki/trainset16_022016.rdp.tgz", ],
        "16S rRNA PDS training set 16":
        ["https://mothur.s3.us-east-2.amazonaws.com/wiki/trainset16_022016.pds.tgz", ],

        "RDP training set 6":
        ["https://mothur.s3.us-east-2.amazonaws.com/wiki/rdptrainingset.zip", ],
    },
    # Silva reference files
    # http://www.mothur.org/wiki/Silva_reference_files
    "silva_release_128": {
        "SILVA release 128":
        ["https://mothur.s3.us-east-2.amazonaws.com/wiki/silva.nr_v128.tgz",
         "https://mothur.s3.us-east-2.amazonaws.com/wiki/silva.seed_v128.tgz", ],
    },

    system.

    Returns the name that the file is saved with.

    """
    print("Downloading %s" % url)
    if not target:
        target = os.path.basename(url)
    if wd:
        target = os.path.join(wd, target)
    print("Saving to %s" % target)
    with open(target, 'wb') as fh:
        fh.write(urllib2.urlopen(url).read())
    return target


def unpack_zip_archive(filen, wd=None):
    """Extract files from a ZIP archive

    Once all the files are extracted the ZIP archive
    file is deleted from the file system.

    """
    if not zipfile.is_zipfile(filen):
        print("%s: not ZIP formatted file")
        return [filen]
    file_list = []
    with zipfile.ZipFile(filen) as z:
        for name in z.namelist():
            if reduce(lambda x, y: x or name.startswith(y), IGNORE_PATHS, False):
                print("Ignoring %s" % name)
                continue
            if wd:
                target = os.path.join(wd, name)
            else:
                target = name
            if name.endswith('/'):
                # Make directory
                print("Creating dir %s" % target)
                try:
                    os.makedirs(target)
                except OSError:
                    pass
            else:
                # Extract file
                print("Extracting %s" % name)
                try:
                    os.makedirs(os.path.dirname(target))
                except OSError:
                    pass
                with open(target, 'wb') as fh:
                    fh.write(z.read(name))
                file_list.append(target)
    print("Removing %s" % filen)
    os.remove(filen)
    return file_list


def unpack_tar_archive(filen, wd=None):

    file is deleted from the file system.

    """
    file_list = []
    if not tarfile.is_tarfile(filen):
        print("%s: not TAR file")
        return [filen]
    with tarfile.open(filen) as t:
        for name in t.getnames():
            # Check for unwanted files
            if reduce(lambda x, y: x or name.startswith(y), IGNORE_PATHS, False):
                print("Ignoring %s" % name)
                continue
            # Extract file
            print("Extracting %s" % name)
            t.extract(name, wd)
            if wd:
                target = os.path.join(wd, name)
            else:
                target = name
            file_list.append(target)
    print("Removing %s" % filen)
    os.remove(filen)
    return file_list


def unpack_archive(filen, wd=None):

    'wd' specifies the working directory to extract
    the files to, otherwise they are extracted to the
    current working directory.

    """
    print("Unpack %s" % filen)
    ext = os.path.splitext(filen)[1]
    print("Extension: %s" % ext)
    if ext == ".zip":
        return unpack_zip_archive(filen, wd=wd)
    elif ext == ".tgz":
        return unpack_tar_archive(filen, wd=wd)
    else:

    """
    ext = os.path.splitext(filen)[1]
    try:
        return MOTHUR_FILE_TYPES[ext]
    except KeyError:
        print("WARNING: unknown file type for " + filen + ", skipping")
        return None


def get_name(filen):
    """Generate a descriptive name based on the file name

      datasets: a list of dataset names corresponding to keys in
        the MOTHUR_REFERENCE_DATA dictionary
    """
    # Make working dir
    wd = tempfile.mkdtemp(suffix=".mothur", dir=os.getcwd())
    print("Working dir %s" % wd)
    # Iterate over all requested reference data URLs
    for dataset in datasets:
        print("Handling dataset '%s'" % dataset)
        for name in MOTHUR_REFERENCE_DATA[dataset]:
            for f in fetch_files(MOTHUR_REFERENCE_DATA[dataset][name], wd=wd):
                type_ = identify_type(f)
                entry_name = "%s (%s)" % (os.path.splitext(os.path.basename(f))[0], name)
                print("%s\t\'%s'\t.../%s" % (type_, entry_name, os.path.basename(f)))
                if type_ is not None:
                    # Move to target dir
                    ref_data_file = os.path.basename(f)
                    f1 = os.path.join(target_dir, ref_data_file)
                    print("Moving %s to %s" % (f, f1))
                    os.rename(f, f1)
                    # Add entry to data table
                    table_name = "mothur_%s" % type_
                    add_data_table_entry(data_tables, table_name, dict(name=entry_name, value=ref_data_file))
    # Remove working dir
    print("Removing %s" % wd)
    shutil.rmtree(wd)


def files_from_filesystem_paths(paths):
    """Return list of file paths from arbitrary input paths

    """
    # Collect files to add
    files = []
    for path in paths:
        path = os.path.abspath(path)
        print("Examining '%s'..." % path)
        if os.path.isfile(path):
            # Store full path for file
            files.append(path)
        elif os.path.isdir(path):
            # Descend into directory and collect the files

    files = files_from_filesystem_paths(paths)
    # Handle each file individually
    for f in files:
        type_ = identify_type(f)
        if type_ is None:
            print("%s: unrecognised type, skipped" % f)
            continue
        ref_data_file = os.path.basename(f)
        target_file = os.path.join(target_dir, ref_data_file)
        entry_name = "%s" % os.path.splitext(ref_data_file)[0]
        if description:
            entry_name += " (%s)" % description
        print("%s\t\'%s'\t.../%s" % (type_, entry_name, ref_data_file))
        # Link to or copy the data
        if link_to_data:
            os.symlink(f, target_file)
        else:
            shutil.copyfile(f, target_file)
        # Add entry to data table
        table_name = "mothur_%s" % type_
        add_data_table_entry(data_tables, table_name, dict(name=entry_name, value=ref_data_file))


if __name__ == "__main__":
    print("Starting...")

    parser.add_option('--datasets', action='store', dest='datasets', default='')
    parser.add_option('--paths', action='store', dest='paths', default=[])
    parser.add_option('--description', action='store', dest='description', default='')
    parser.add_option('--link', action='store_true', dest='link_to_data')
    options, args = parser.parse_args()
    print("options: %s" % options)
    print("args   : %s" % args)

    # Check for JSON file
    if len(args) != 1:
        sys.stderr.write("Need to supply JSON file name")
        sys.exit(1)

    # Read the input JSON
    params, target_dir = read_input_json(jsonfile)

    # Make the target directory
    print("Making %s" % target_dir)
    os.mkdir(target_dir)

    # Set up data tables dictionary
    data_tables = create_data_tables_dict()
    add_data_table(data_tables, 'mothur_lookup')

4:0db22932bc39

#!/usr/bin/env python3
#
# Data manager for reference data for the 'mothur_toolsuite' Galaxy tools
import io
import json
import optparse
import os
import shutil
import sys
import tarfile
import tempfile
import urllib.error
import urllib.parse
import urllib.request
import zipfile
from functools import reduce

# When extracting files from archives, skip names that
# start with the following strings

    "lookup_gs20": {
        "GS20": ["https://mothur.s3.us-east-2.amazonaws.com/wiki/lookup_gs20.zip", ]
    },
    # RDP reference files
    # http://www.mothur.org/wiki/RDP_reference_files
    "RDP_v18": {
        "16S rRNA RDP training set 18":
            [
                "https://mothur.s3.us-east-2.amazonaws.com/wiki/trainset18_062020.rdp.tgz", ],
        "16S rRNA PDS training set 18":
            [
                "https://mothur.s3.us-east-2.amazonaws.com/wiki/trainset18_062020.pds.tgz", ],
    },
    "RDP_v16": {
        "16S rRNA RDP training set 16":
        ["https://mothur.s3.us-east-2.amazonaws.com/wiki/trainset16_022016.rdp.tgz", ],
        "16S rRNA PDS training set 16":
        ["https://mothur.s3.us-east-2.amazonaws.com/wiki/trainset16_022016.pds.tgz", ],

        "RDP training set 6":
        ["https://mothur.s3.us-east-2.amazonaws.com/wiki/rdptrainingset.zip", ],
    },
    # Silva reference files
    # http://www.mothur.org/wiki/Silva_reference_files
    "silva_release_138.1": {
        "SILVA release 138.1":
            [
                "https://mothur.s3.us-east-2.amazonaws.com/wiki/silva.nr_v138_1.tgz",
                "https://mothur.s3.us-east-2.amazonaws.com/wiki/silva.seed_v138_1.tgz", ],
    },
    "silva_release_128": {
        "SILVA release 128":
        ["https://mothur.s3.us-east-2.amazonaws.com/wiki/silva.nr_v128.tgz",
         "https://mothur.s3.us-east-2.amazonaws.com/wiki/silva.seed_v128.tgz", ],
    },

    system.

    Returns the name that the file is saved with.

    """
    print(f"Downloading {url}")
    if not target:
        target = os.path.basename(url)
    if wd:
        target = os.path.join(wd, target)
    print(f"Saving to {target}")
    with open(target, 'wb') as fh:
        url_h = urllib.request.urlopen(url)
        while True:
            buffer = url_h.read(io.DEFAULT_BUFFER_SIZE)
            if buffer == b"":
                break
            fh.write(buffer)
    return target


def unpack_zip_archive(filen, wd=None):
    """Extract files from a ZIP archive

    Once all the files are extracted the ZIP archive
    file is deleted from the file system.

    """
    if not zipfile.is_zipfile(filen):
        print(f"{filen}: not ZIP formatted file")
        return [filen]
    file_list = []
    with zipfile.ZipFile(filen) as z:
        for name in z.namelist():
            if reduce(lambda x, y: x or name.startswith(y), IGNORE_PATHS, False):
                print(f"Ignoring {name}")
                continue
            if wd:
                target = os.path.join(wd, name)
            else:
                target = name
            if name.endswith('/'):
                # Make directory
                print(f"Creating dir {target}")
                try:
                    os.makedirs(target)
                except OSError:
                    pass
            else:
                # Extract file
                print("Extracting {target}")
                try:
                    os.makedirs(os.path.dirname(target))
                except OSError:
                    pass
                with open(target, 'wb') as fh:
                    fh.write(z.read(name))
                file_list.append(target)
    print(f"Removing {filen}")
    os.remove(filen)
    return file_list


def unpack_tar_archive(filen, wd=None):

    file is deleted from the file system.

    """
    file_list = []
    if not tarfile.is_tarfile(filen):
        print(f"{filen}: not TAR file")
        return [filen]
    with tarfile.open(filen) as t:
        for name in t.getnames():
            # Check for unwanted files
            if reduce(lambda x, y: x or name.startswith(y), IGNORE_PATHS, False):
                print(f"Ignoring {name}")
                continue
            # Extract file
            print(f"Extracting {name}")
            t.extract(name, wd)
            if wd:
                target = os.path.join(wd, name)
            else:
                target = name
            file_list.append(target)
    print(f"Removing {filen}")
    os.remove(filen)
    return file_list


def unpack_archive(filen, wd=None):

    'wd' specifies the working directory to extract
    the files to, otherwise they are extracted to the
    current working directory.

    """
    print(f"Unpack {filen}")
    ext = os.path.splitext(filen)[1]
    print(f"Extension: {ext}")
    if ext == ".zip":
        return unpack_zip_archive(filen, wd=wd)
    elif ext == ".tgz":
        return unpack_tar_archive(filen, wd=wd)
    else:

    """
    ext = os.path.splitext(filen)[1]
    try:
        return MOTHUR_FILE_TYPES[ext]
    except KeyError:
        print(f"WARNING: unknown file type for {filen}, skipping")
        return None


def get_name(filen):
    """Generate a descriptive name based on the file name

      datasets: a list of dataset names corresponding to keys in
        the MOTHUR_REFERENCE_DATA dictionary
    """
    # Make working dir
    wd = tempfile.mkdtemp(suffix=".mothur", dir=os.getcwd())
    print(f"Working dir {wd}")
    # Iterate over all requested reference data URLs
    for dataset in datasets:
        print(f"Handling dataset '{dataset}'")
        for name in MOTHUR_REFERENCE_DATA[dataset]:
            for f in fetch_files(MOTHUR_REFERENCE_DATA[dataset][name], wd=wd):
                type_ = identify_type(f)
                name_from_file = os.path.splitext(os.path.basename(f))[0]
                entry_name = f"{name_from_file} ({name})"
                print(f"{type_}\t\'{entry_name}'\t.../{os.path.basename(f)}")
                if type_ is not None:
                    # Move to target dir
                    ref_data_file = os.path.basename(f)
                    f1 = os.path.join(target_dir, ref_data_file)
                    print(f"Moving {f} to {f1}")
                    shutil.move(f, f1)
                    # Add entry to data table
                    table_name = f"mothur_{type_}"
                    add_data_table_entry(data_tables, table_name, dict(name=entry_name, value=ref_data_file))
    # Remove working dir
    print(f"Removing {wd}")
    shutil.rmtree(wd)


def files_from_filesystem_paths(paths):
    """Return list of file paths from arbitrary input paths

    """
    # Collect files to add
    files = []
    for path in paths:
        path = os.path.abspath(path)
        print(f"Examining '{path}'...")
        if os.path.isfile(path):
            # Store full path for file
            files.append(path)
        elif os.path.isdir(path):
            # Descend into directory and collect the files

    files = files_from_filesystem_paths(paths)
    # Handle each file individually
    for f in files:
        type_ = identify_type(f)
        if type_ is None:
            print(f"{f}: unrecognised type, skipped")
            continue
        ref_data_file = os.path.basename(f)
        target_file = os.path.join(target_dir, ref_data_file)
        entry_name = "%s" % os.path.splitext(ref_data_file)[0]
        if description:
            entry_name += " (%s)" % description
        print(f"{type_}\t\'{entry_name}'\t.../{ref_data_file}")
        # Link to or copy the data
        if link_to_data:
            os.symlink(f, target_file)
        else:
            shutil.copyfile(f, target_file)
        # Add entry to data table
        table_name = f"mothur_{type_}"
        add_data_table_entry(data_tables, table_name, dict(name=entry_name, value=ref_data_file))


if __name__ == "__main__":
    print("Starting...")

    parser.add_option('--datasets', action='store', dest='datasets', default='')
    parser.add_option('--paths', action='store', dest='paths', default=[])
    parser.add_option('--description', action='store', dest='description', default='')
    parser.add_option('--link', action='store_true', dest='link_to_data')
    options, args = parser.parse_args()
    print(f"options: {options}")
    print(f"args   : {args}")

    # Check for JSON file
    if len(args) != 1:
        sys.stderr.write("Need to supply JSON file name")
        sys.exit(1)

    # Read the input JSON
    params, target_dir = read_input_json(jsonfile)

    # Make the target directory
    print(f"Making {target_dir}")
    os.mkdir(target_dir)

    # Set up data tables dictionary
    data_tables = create_data_tables_dict()
    add_data_table(data_tables, 'mothur_lookup')