<commit_before>#!/usr/bin/env python

__author__ = "<NAME>"
__copyright__ = "Copyright 2011, The QIIME Project"
__credits__ = ["<NAME>", "<NAME>"]
__license__ = "GPL"
__version__ = "1.8.0-dev"
__maintainer__ = "<NAME>"
__email__ = "<EMAIL>"

from unittest import TestCase, main

import numpy
from numpy import array
from numpy.testing import assert_almost_equal, assert_allclose


from biom import load_table, parse_table

from decontaminate import *

def assertDeepAlmostEqual(test_case, expected, actual, *args, **kwargs):
    """
    Assert that two complex structures have almost equal contents.

    Compares lists, dicts and tuples recursively. Checks numeric values
    using test_case's :py:meth:`unittest.TestCase.assertAlmostEqual` and
    checks all other values with :py:meth:`unittest.TestCase.assertEqual`.
    Accepts additional positional and keyword arguments and pass those
    intact to assertAlmostEqual() (that's how you specify comparison
    precision).

    :param test_case: TestCase object on which we can call all of the basic
    'assert' methods.
    :type test_case: :py:class:`unittest.TestCase` object

    test from https://github.com/larsbutler/oq-engine/blob/master/tests/utils/helpers.py
    """
    is_root = not '__trace' in kwargs
    trace = kwargs.pop('__trace', 'ROOT')
    try:
        if isinstance(expected, (int, float, long, complex)):
            test_case.assertAlmostEqual(expected, actual, *args, **kwargs)
        elif isinstance(expected, (list, tuple, numpy.ndarray)):
            test_case.assertEqual(len(expected), len(actual))
            for index in xrange(len(expected)):
                v1, v2 = expected[index], actual[index]
                assertDeepAlmostEqual(test_case, v1, v2,
                                      __trace=repr(index), *args, **kwargs)
        elif isinstance(expected, dict):
            test_case.assertEqual(set(expected), set(actual))
            for key in expected:
                assertDeepAlmostEqual(test_case, expected[key], actual[key],
                                      __trace=repr(key), *args, **kwargs)
        else:
            test_case.assertEqual(expected, actual)
    except AssertionError as exc:
        exc.__dict__.setdefault('traces', []).append(trace)
        if is_root:
            trace = ' -> '.join(reversed(exc.traces))
            exc = AssertionError("%s\nTRACE: %s" % (exc.message, trace))
        raise exc


class DecontaminationTests(TestCase):

    def setUp(self):
        """Init variables for the tests """
        test_biom_fp = '/Users/jonsanders/Development/git_sw/qiime/qiime_test_data/decontaminate/test_otu_table.biom'

        self.test_biom = parse_table(test_biom_file)

    def test_pick_min_relabund_threshold(self):

        contamination_header = exp_contamination_header

        contamination_stats_dict = exp_contamination_stats_dict_prop

        exp_below_threshold_seqs = set(['contam1','contam4'])

        obs_below_threshold_seqs = pick_min_relabund_threshold(contamination_stats_dict, exp_contamination_header, 0.1)

        self.assertEqual(exp_below_threshold_seqs,obs_below_threshold_seqs)

    def test_pick_corr_contaminants(self):

        corr_data_dict = {'Blank1': 0, 'Blank2': 1, 'Sample1': 2, 'Sample2': 3}

        test_biom = self.test_biom

        exp_corr_contaminants = set(['contam1','contam2','contam4'])

        exp_corr_dict = {u'otu1': (0.8, 0.2),
                         u'otu2': (0.6, 0.4),
                         u'otu3': (-0.4, 0.6),
                         u'otu4': (0.2581988897471611, 0.7418011102528389),
                         u'contam1': (-0.8, 0.2),
                         u'contam2': (-1.0, 0.0),
                         u'contam3': (0.4, 0.6),
                         u'contam4': (-0.89442719099991586, 0.10557280900008413)}

        obs_corr_contaminants, obs_corr_dict = pick_corr_contaminants(test_biom,
                                                           corr_data_dict,
                                                           -0.6)

        self.assertEqual(obs_corr_contaminants, exp_corr_contaminants)
        assertDeepAlmostEqual(self, obs_corr_dict, exp_corr_dict)


    def test_reinstate_abund_seqs(self):

        putative_contaminants = set(['otu1','otu2','otu3','contam1','contam2','contam3','contam4'])

        contamination_stats_dict = exp_contamination_stats_dict

        contamination_stats_header = exp_contamination_header


        exp_abund_reinstated_seqs_max = set(['otu1','otu2','otu3'])

        obs_abund_reinstated_seqs_max = reinstate_abund_seqs(putative_contaminants, 
                         contamination_stats_dict, 
                         contamination_stats_header,
                         'maxS',
                         'maxB',
                         1)

        self.assertEqual(exp_abund_reinstated_seqs_max, obs_abund_reinstated_seqs_max)


        exp_abund_reinstated_seqs_avg = set(['otu1','otu2','contam3'])

        obs_abund_reinstated_seqs_avg = reinstate_abund_seqs(putative_contaminants, 
                         contamination_stats_dict, 
                         contamination_stats_header,
                         'avgS',
                         'avgB',
                         1)

        self.assertEqual(exp_abund_reinstated_seqs_avg, obs_abund_reinstated_seqs_avg)

    def test_reinstate_incidence_seqs(self):

        test_biom = self.test_biom

        blank_sample_ids = ['Blank1', 'Blank2']

        putative_contaminants = set(['otu1','otu2','otu3','contam1','contam2','contam3','contam4'])

        exp_incidence_reinstated_seqs = set(['otu1','otu2','contam3'])

        obs_incidence_reinstated_seqs = reinstate_incidence_seqs(
                         putative_contaminants,
                         test_biom,
                         blank_sample_ids,
                         2)

        self.assertEqual(exp_incidence_reinstated_seqs, obs_incidence_reinstated_seqs)


    def test_mothur_counts_to_biom(self):
        """Convert a mothur counts table to a biom object"""
        test_counts_fp = '/Users/jonsanders/Development/git_sw/decontaminate/qiime_scripts/qiime_test_data/decontaminate/test.count_table'

        test_counts_biom_fp = '/Users/jonsanders/Development/git_sw/decontaminate/qiime_scripts/qiime_test_data/decontaminate/test.count_table.biom'

        test_counts_biom = load_table(test_counts_biom_fp)

        obs_counts_biom = mothur_counts_to_biom(test_counts_fp)

        self.assertEqual(test_counts_biom, obs_counts_biom)


    def test_prescreen_libraries(self):
        """Test pre-screening libraries for contamination"""

        unique_seq_biom = self.test_biom

        exp_exclude_libs = set(['Blank1', 'Blank2', 'Sample2'])

        obs_exclude_libs = prescreen_libraries(unique_seq_biom,
                                               blank_sample_ids = ['Blank1','Blank2'],
                                               removal_stat_blank = 'maxB',
                                               removal_stat_sample = 'maxS',
                                               removal_differential = 1,
                                               presecreen_threshold = 0.65)

        self.assertEqual(exp_exclude_libs, obs_exclude_libs)

    def test_pick_ref_contaminants(self):
        """Test the reference-based contaminant search"""

        test_queries = set(['denovo47', 'denovo46', 'denovo27', 'denovo26', 'denovo25', 'denovo24', 'denovo23', 'denovo22', 'denovo21', 'denovo20', 'denovo49', 'denovo48', 'denovo29', 'denovo28', 'denovo0', 'denovo1', 'denovo2', 'denovo3', 'denovo4', 'denovo5', 'denovo6', 'denovo7', 'denovo8', 'denovo9', 'denovo18', 'denovo19', 'denovo12', 'denovo13', 'denovo10', 'denovo11', 'denovo16', 'denovo17', 'denovo14', 'denovo15', 'denovo34', 'denovo35', 'denovo36', 'denovo37', 'denovo30', 'denovo31', 'denovo32', 'denovo33', 'denovo38', 'denovo39', 'denovo50', 'denovo51', 'denovo41', 'denovo40', 'denovo43', 'denovo42', 'denovo52', 'denovo45', 'denovo44'])

        test_ref_db_fp = '/Users/jonsanders/Development/git_sw/decontaminate/qiime_scripts/qiime_test_data/decontaminate/blanks_rep_set.filtered.fna'

        test_input_fasta_fp = '/Users/jonsanders/Development/git_sw/decontaminate/qiime_scripts/qiime_test_data/decontaminate/unique_seqs_rep_set.fna'<commit_msg>Fixed bug in compare_blank_abundances method in which the wrong set of sequences were being returned (i.e. non-contams instead of contams)<commit_after>#!/usr/bin/env python

__author__ = "<NAME>"
__copyright__ = "Copyright 2011, The QIIME Project"
__credits__ = ["<NAME>", "<NAME>"]
__license__ = "GPL"
__version__ = "1.8.0-dev"
__maintainer__ = "<NAME>"
__email__ = "<EMAIL>"

from unittest import TestCase, main

import numpy
from numpy import array
from numpy.testing import assert_almost_equal, assert_allclose


from biom import load_table, parse_table

from decontaminate import *

def assertDeepAlmostEqual(test_case, expected, actual, *args, **kwargs):
    """
    Assert that two complex structures have almost equal contents.

    Compares lists, dicts and tuples recursively. Checks numeric values
    using test_case's :py:meth:`unittest.TestCase.assertAlmostEqual` and
    checks all other values with :py:meth:`unittest.TestCase.assertEqual`.
    Accepts additional positional and keyword arguments and pass those
    intact to assertAlmostEqual() (that's how you specify comparison
    precision).

    :param test_case: TestCase object on which we can call all of the basic
    'assert' methods.
    :type test_case: :py:class:`unittest.TestCase` object

    test from https://github.com/larsbutler/oq-engine/blob/master/tests/utils/helpers.py
    """
    is_root = not '__trace' in kwargs
    trace = kwargs.pop('__trace', 'ROOT')
    try:
        if isinstance(expected, (int, float, long, complex)):
            test_case.assertAlmostEqual(expected, actual, *args, **kwargs)
        elif isinstance(expected, (list, tuple, numpy.ndarray)):
            test_case.assertEqual(len(expected), len(actual))
            for index in xrange(len(expected)):
                v1, v2 = expected[index], actual[index]
                assertDeepAlmostEqual(test_case, v1, v2,
                                      __trace=repr(index), *args, **kwargs)
        elif isinstance(expected, dict):
            test_case.assertEqual(set(expected), set(actual))
            for key in expected:
                assertDeepAlmostEqual(test_case, expected[key], actual[key],
                                      __trace=repr(key), *args, **kwargs)
        else:
            test_case.assertEqual(expected, actual)
    except AssertionError as exc:
        exc.__dict__.setdefault('traces', []).append(trace)
        if is_root:
            trace = ' -> '.join(reversed(exc.traces))
            exc = AssertionError("%s\nTRACE: %s" % (exc.message, trace))
        raise exc


class DecontaminationTests(TestCase):

    def setUp(self):
        """Init variables for the tests """
        test_biom_fp = '/Users/jonsanders/Development/git_sw/qiime/qiime_test_data/decontaminate/test_otu_table.biom'

        self.test_biom = parse_table(test_biom_file)

    def test_pick_min_relabund_threshold(self):

        contamination_header = exp_contamination_header

        contamination_stats_dict = exp_contamination_stats_dict_prop

        exp_below_threshold_seqs = set(['contam1','contam4'])

        obs_below_threshold_seqs = pick_min_relabund_threshold(contamination_stats_dict, exp_contamination_header, 0.1)

        self.assertEqual(exp_below_threshold_seqs,obs_below_threshold_seqs)

    def test_pick_corr_contaminants(self):

        corr_data_dict = {'Blank1': 0, 'Blank2': 1, 'Sample1': 2, 'Sample2': 3}

        test_biom = self.test_biom

        exp_corr_contaminants = set(['contam1','contam2','contam4'])

        exp_corr_dict = {u'otu1': (0.8, 0.2),
                         u'otu2': (0.6, 0.4),
                         u'otu3': (-0.4, 0.6),
                         u'otu4': (0.2581988897471611, 0.7418011102528389),
                         u'contam1': (-0.8, 0.2),
                         u'contam2': (-1.0, 0.0),
                         u'contam3': (0.4, 0.6),
                         u'contam4': (-0.89442719099991586, 0.10557280900008413)}

        obs_corr_contaminants, obs_corr_dict = pick_corr_contaminants(test_biom,
                                                           corr_data_dict,
                                                           -0.6)

        self.assertEqual(obs_corr_contaminants, exp_corr_contaminants)
        assertDeepAlmostEqual(self, obs_corr_dict, exp_corr_dict)


    def test_reinstate_abund_seqs(self):

        putative_contaminants = set(['otu1','otu2','otu3','contam1','contam2','contam3','contam4'])

        contamination_stats_dict = exp_contamination_stats_dict

        contamination_stats_header = exp_contamination_header


        exp_abund_reinstated_seqs_max = set(['otu1','otu2','otu3'])

        obs_abund_reinstated_seqs_max = reinstate_abund_seqs(putative_contaminants, 
                         contamination_stats_dict, 
                         contamination_stats_header,
                         'maxS',
                         'maxB',
                         1)

        self.assertEqual(exp_abund_reinstated_seqs_max, obs_abund_reinstated_seqs_max)


        exp_abund_reinstated_seqs_avg = set(['otu1','otu2','contam3'])

        obs_abund_reinstated_seqs_avg = reinstate_abund_seqs(putative_contaminants, 
                         contamination_stats_dict, 
                         contamination_stats_header,
                         'avgS',
                         'avgB',
                         1)

        self.assertEqual(exp_abund_reinstated_seqs_avg, obs_abund_reinstated_seqs_avg)

    def test_reinstate_incidence_seqs(self):

        test_biom = self.test_biom

        blank_sample_ids = ['Blank1', 'Blank2']

        putative_contaminants = set(['otu1','otu2','otu3','contam1','contam2','contam3','contam4'])

        exp_incidence_reinstated_seqs = set(['otu1','otu2','contam3'])

        obs_incidence_reinstated_seqs = reinstate_incidence_seqs(
                         putative_contaminants,
                         test_biom,
                         blank_sample_ids,
                         2)

        self.assertEqual(exp_incidence_reinstated_seqs, obs_incidence_reinstated_seqs)


    def test_mothur_counts_to_biom(self):
        """Convert a mothur counts table to a biom object"""
        test_counts_fp = '/Users/jonsanders/Development/git_sw/decontaminate/qiime_scripts/qiime_test_data/decontaminate/test.count_table'

        test_counts_biom_fp = '/Users/jonsanders/Development/git_sw/decontaminate/qiime_scripts/qiime_test_data/decontaminate/test.count_table.biom'

        test_counts_biom = load_table(test_counts_biom_fp)

        obs_counts_biom = mothur_counts_to_biom(test_counts_fp)

        self.assertEqual(test_counts_biom, obs_counts_biom)


    def test_prescreen_libraries(self):
        """Test pre-screening libraries for contamination"""

        unique_seq_biom = self.test_biom

        exp_exclude_libs = set(['Blank1', 'Blank2', 'Sample1'])

        obs_exclude_libs = prescreen_libraries(unique_seq_biom,
                                               blank_sample_ids = ['Blank1','Blank2'],
                                               removal_stat_blank = 'maxB',
                                               removal_stat_sample = 'maxS',
                                               removal_differential = 1,
                                               presecreen_threshold = 0.65)

        self.assertEqual(exp_exclude_libs, obs_exclude_libs)


        exp_exclude_libs = set(['Blank1', 'Blank2'])

        obs_exclude_libs = prescreen_libraries(unique_seq_biom,
                                               blank_sample_ids = ['Blank1','Blank2'],
                                               removal_stat_blank = 'maxB',
                                               removal_stat_sample = 'maxS',
                                               removal_differential = 1,
                                               presecreen_threshold = 0.5)

        self.assertEqual(exp_exclude_libs, obs_exclude_libs)

    def test_pick_ref_contaminants(self):
        """Test the reference-based contaminant search"""

        test_queries = set(['denovo47', 'denovo46', 'denovo27', 'denovo26', 'denovo25', 'denovo24', 'denovo23', 'denovo22', 'denovo21', 'denovo20', 'denovo49', 'denovo48', 'denovo29', 'denovo28', 'denovo0', 'denovo1', 'denovo2', 'denovo3', 'denovo4', 'denovo5', 'denovo6', 'denovo7', 'denovo8', 'denovo9', 'denovo18', 'denovo19', 'denovo12', 'denovo13', 'denovo10', 'denovo11', 'denovo16', 'denovo17', 'denovo14', 'denovo15', 'denovo34', 'denovo35', 'denovo36', 'denovo37', 'denovo30', 'denovo31', 'denovo32', 'denovo33', 'denovo38', 'denovo39', 'denovo50', 'denovo51', 'denovo41', 'denovo40', 'denovo43', 'denovo42', 'denovo52', 'denovo45', 'denovo44'])

        test_ref_db_fp = '/Users/jonsanders/Development/git_sw/decontaminate/qiime_scripts/qiime_test_data/decontaminate/blanks_rep_set.filtered.fna'

        test_input_fasta_fp = '/Users/jonsanders/Development/git_sw/decontaminate/qiime_scripts/qiime_test_data/decontaminate/unique_seqs_rep_set.fna'
{'fraction_non_alphanumeric': 0.10602659298912105, 'fraction_numerical': 0.030679790479479652, 'mean_word_length': 2.5370521172638436, 'pattern_counts': {'":': 0, '<': 13, '<?xml version=': 0, '>': 15, 'https://': 2, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 52, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5989225', 'n_tokens_mistral': 5853, 'n_tokens_neox': 5263, 'n_words': 899}