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

from t2t.consensus import get_consensus_stats, taxa_score, hash_cons, \
        taxa_score_hash, merge_taxa_strings_and_scores
from cogent.util.unit_test import TestCase, main
from numpy import nan, array

class ConsensusTests(TestCase):
    def setUp(self):
        pass

    def test_merge_taxa_strings_and_scores(self):
        """merge taxa scores and the strings"""
        master = {'a':['k__k1','p__p1','c__c1','o__o1','f__f1','g__g1','s__s1'],
                 'b':['k__k1','p__p1','c__c2',None,'f__f2','g__g1','s__s2'],
                 'c':[None,None,None,None,None,None,None],
                 'd':['k__k2','p__x','c__c1','o__o10','f__f2','g__g2','s__s3']}
        rep1 = {'a':['k__k1','p__p1','c__c1','o__o1','f__f1','g__g1','s__s1'],
                 'c':[None,None,None,None,None,None,None],
                 'd':['k__k2','','c__c1','f__f2','f__f2','g__g2','s__s3']}
        rep2 = {'a':['k__k1','p__p1','c__c1','o__o1','f__f1','g__g1','s__s1'],
                 'b':['k__k1','p__p1','c__c2',None,'f__f2','g__g1','s__s9'],
                 'c':[None,None,None,None,None,None,None]}
        rep3 = {'a':['k__k1','p__p1','c__c1','o__o1','f__f1','g__g1','s__s1'],
                 'b':['k__k1','p__p1','c__c2',None,'f__f2','g__g1','s__s2'],
                 'd':['k__k2','','c__c1','f__f2','f__f2','g__g2','s__s3']}
        rep4 = {'a':['k__k1','p__p1','c__c1','o__o1','f__f1','g__g1','s__s1'],
                 'b':['k__k1','p__p1','c__c10',None,'f__f2','g__g1','s__s2'],
                 'c':[None,None,None,None,None,None,None],
                 'd':['k__k1','','c__c1','f__f2','f__f2','g__g2','s__s3']}
        scores = taxa_score(master, [rep1, rep2, rep3, rep4])
        exp = {'a':[('k__k1',1.0), ('p__p1',1.0), ('c__c1',1.0), ('o__o1',1.0),
                    ('f__f1',1.0), ('g__g1',1.0), ('s__s1',1.0)],
               'b':[('k__k1',1.0), ('p__p1',1.0), ('c__c2',0.75), (None,1.0), 
                    ('f__f2',1.0), ('g__g1',1.0), ('s__s2',0.75)],
               'c':[(None,1.0), (None,1.0), (None,1.0), (None,1.0), (None,1.0),
                    (None,1.0), (None,1.0)],
               'd':[('k__k2',0.75), ('p__x',0.25), ('c__c1',1.0), 
                    ('o__o10',0.25), ('f__f2',1.0), ('g__g2',1.0), 
                    ('s__s3',1.0)]}
        obs = merge_taxa_strings_and_scores(master, scores)
        self.assertEqual(obs,exp)

    def test_taxa_score(self):
        """score taxa strings in the face of replicates"""
        master = {'a':['k__k1','p__p1','c__c1','o__o1','f__f1','g__g1','s__s1'],
                 'b':['k__k1','p__p1','c__c2',None,'f__f2','g__g1','s__s2'],
                 'c':[None,None,None,None,None,None,None],
                 'd':['k__k2','p__x','c__c1','o__o10','f__f2','g__g2','s__s3']}
        rep1 = {'a':['k__k1','p__p1','c__c1','o__o1','f__f1','g__g1','s__s1'],
                 'c':[None,None,None,None,None,None,None],
                 'd':['k__k2','','c__c1','f__f2','f__f2','g__g2','s__s3']}
        rep2 = {'a':['k__k1','p__p1','c__c1','o__o1','f__f1','g__g1','s__s1'],
                 'b':['k__k1','p__p1','c__c2',None,'f__f2','g__g1','s__s9'],
                 'c':[None,None,None,None,None,None,None]}
        rep3 = {'a':['k__k1','p__p1','c__c1','o__o1','f__f1','g__g1','s__s1'],
                 'b':['k__k1','p__p1','c__c2',None,'f__f2','g__g1','s__s2'],
                 'd':['k__k2','','c__c1','f__f2','f__f2','g__g2','s__s3']}
        rep4 = {'a':['k__k1','p__p1','c__c1','o__o1','f__f1','g__g1','s__s1'],
                 'b':['k__k1','p__p1','c__c10',None,'f__f2','g__g1','s__s2'],
                 'c':[None,None,None,None,None,None,None],
                 'd':['k__k1','','c__c1','f__f2','f__f2','g__g2','s__s3']}
        # if malformed, as in rep3 rep4 for d, count as contridiction
        exp = {'a':[1.0, 1.0, 1.0, 1.0, 1.0, 1.0, 1.0],
               'b':[1.0, 1.0, 0.75, 1.0, 1.0, 1.0, 0.75],
               'c':[1.0, 1.0, 1.0, 1.0, 1.0, 1.0, 1.0],
               'd':[0.75, 0.25, 1.0, 0.25, 1.0, 1.0, 1.0]}
        obs = taxa_score(master, [rep1,rep2,rep3,rep4])
        self.assertEqual(obs, exp)

    def test_taxa_score_hash(self):
        """test hash based consensus scoring"""
        master = {'a':['k__k1','p__p1','c__c1','o__o1','f__f1','g__g1','s__s1'],
                 'b':['k__k1','p__p1','c__c2',None,'f__f2','g__g1','s__s2'],
                 'c':[None,None,None,None,None,None,None],
                 'd':['k__k2','p__x','c__c1','o__o10','f__f2','g__g2','s__s3']}
        rep1 = {'a':['k__k1','p__p1','c__c1','o__o1','f__f1','g__g1','s__s1'],
                 'c':[None,None,None,None,None,None,None],
                 'd':['k__k2','p__','c__c1','f__f2','f__f2','g__g2','s__s3']}
        rep2 = {'a':['k__k1','p__p1','c__c1','o__o1','f__f1','g__g1','s__s1'],
                 'b':['k__k1','p__p1','c__c2',None,'f__f2','g__g1','s__s9'],
                 'c':[None,None,None,None,None,None,None]}
        rep3 = {'a':['k__k1','p__p1','c__c1','o__o1','f__f1','g__g1','s__s1'],
                 'b':['k__k1','p__p1','c__c2',None,'f__f2','g__g1','s__s2'],
                 'd':['k__k2','p__','c__c1','f__f2','f__f2','g__g2','s__s3']}
        rep4 = {'a':['k__k1','p__p1','c__c1','o__o1','f__f1','g__g1','s__s1'],
                 'b':['k__k1','p__p1','c__c10',None,'f__f2','g__g1','s__s2'],
                 'c':[None,None,None,None,None,None,None],
                 'd':['k__k1','p__','c__c1','f__f2','f__f2','g__g2','s__s3']}
        # if malformed, as in rep3 rep4 for d, count as contridiction
        exp = {'a':[1.0, 1.0, 1.0, 1.0, 1.0, 1.0, 1.0],
               'b':[1.0, 1.0, 0.75, 1.0, 1.0, 1.0, 0.75],
               'c':[1.0, 1.0, 1.0, 1.0, 1.0, 1.0, 1.0],
               'd':[0.75, 0.25, 1.0, 0.25, 1.0, 1.0, 1.0]}
        obs = taxa_score_hash(master, [rep1,rep2,rep3,rep4])
        self.assertEqual(obs, exp)

    def test_hash_cons(self):
        """test turning consensus strings into hashes"""
        input = {'a':['k__k1','p__p1','c__c1','o__o1','f__f1','g__g1','s__s1'],
                 'b':['k__k1','p__p1','c__c2',None,'f__f2','g__g1','s__s2'],
                 'c':[None,None,None,None,None,None,None],
                 'd':['k__k2','p__x','c__c1','o__o10','f__f2','g__g2','s__s3']}
        exp = array([map(hash, input['a']),
                     map(hash, input['d']),
                     map(hash, input['c']),
                     map(hash, input['b'])])
        obs = hash_cons(input, ['a','d','c','b'], 7)
        self.assertEqual(obs, exp)

        exp = array([map(hash, input['a']),
                     map(hash, input['d']),
                     [0,0,0,0,0,0,0],
                     map(hash, input['c']),
                     map(hash, input['b'])])
        obs = hash_cons(input, ['a','d','e','c','b'], 7)
        self.assertFloatEqual(obs,exp)

    def test_get_consensus_stats(self):
        """Produces the correct stats"""
        input = {'a':['k__k1','p__p1','c__c1','o__o1','f__f1','g__g1','s__s1'],
                 'b':['k__k1','p__p1','c__c2',None,'f__f2','g__g1','s__s2'],
                 'c':[None,None,None,None,None,None,None],
                 'd':['k__k2','','c__c1','f__f2','f__f2','g__g2','s__s3']}<commit_msg>removing cogent<commit_after>#!/usr/bin/env python

from t2t.consensus import get_consensus_stats, taxa_score, hash_cons, \
        taxa_score_hash, merge_taxa_strings_and_scores
from unittest import TestCase, main
from numpy import nan, array, array_equal

class ConsensusTests(TestCase):
    def setUp(self):
        pass

    def test_merge_taxa_strings_and_scores(self):
        """merge taxa scores and the strings"""
        master = {'a':['k__k1','p__p1','c__c1','o__o1','f__f1','g__g1','s__s1'],
                 'b':['k__k1','p__p1','c__c2',None,'f__f2','g__g1','s__s2'],
                 'c':[None,None,None,None,None,None,None],
                 'd':['k__k2','p__x','c__c1','o__o10','f__f2','g__g2','s__s3']}
        rep1 = {'a':['k__k1','p__p1','c__c1','o__o1','f__f1','g__g1','s__s1'],
                 'c':[None,None,None,None,None,None,None],
                 'd':['k__k2','','c__c1','f__f2','f__f2','g__g2','s__s3']}
        rep2 = {'a':['k__k1','p__p1','c__c1','o__o1','f__f1','g__g1','s__s1'],
                 'b':['k__k1','p__p1','c__c2',None,'f__f2','g__g1','s__s9'],
                 'c':[None,None,None,None,None,None,None]}
        rep3 = {'a':['k__k1','p__p1','c__c1','o__o1','f__f1','g__g1','s__s1'],
                 'b':['k__k1','p__p1','c__c2',None,'f__f2','g__g1','s__s2'],
                 'd':['k__k2','','c__c1','f__f2','f__f2','g__g2','s__s3']}
        rep4 = {'a':['k__k1','p__p1','c__c1','o__o1','f__f1','g__g1','s__s1'],
                 'b':['k__k1','p__p1','c__c10',None,'f__f2','g__g1','s__s2'],
                 'c':[None,None,None,None,None,None,None],
                 'd':['k__k1','','c__c1','f__f2','f__f2','g__g2','s__s3']}
        scores = taxa_score(master, [rep1, rep2, rep3, rep4])
        exp = {'a':[('k__k1',1.0), ('p__p1',1.0), ('c__c1',1.0), ('o__o1',1.0),
                    ('f__f1',1.0), ('g__g1',1.0), ('s__s1',1.0)],
               'b':[('k__k1',1.0), ('p__p1',1.0), ('c__c2',0.75), (None,1.0),
                    ('f__f2',1.0), ('g__g1',1.0), ('s__s2',0.75)],
               'c':[(None,1.0), (None,1.0), (None,1.0), (None,1.0), (None,1.0),
                    (None,1.0), (None,1.0)],
               'd':[('k__k2',0.75), ('p__x',0.25), ('c__c1',1.0),
                    ('o__o10',0.25), ('f__f2',1.0), ('g__g2',1.0),
                    ('s__s3',1.0)]}
        obs = merge_taxa_strings_and_scores(master, scores)
        self.assertEqual(obs,exp)

    def test_taxa_score(self):
        """score taxa strings in the face of replicates"""
        master = {'a':['k__k1','p__p1','c__c1','o__o1','f__f1','g__g1','s__s1'],
                 'b':['k__k1','p__p1','c__c2',None,'f__f2','g__g1','s__s2'],
                 'c':[None,None,None,None,None,None,None],
                 'd':['k__k2','p__x','c__c1','o__o10','f__f2','g__g2','s__s3']}
        rep1 = {'a':['k__k1','p__p1','c__c1','o__o1','f__f1','g__g1','s__s1'],
                 'c':[None,None,None,None,None,None,None],
                 'd':['k__k2','','c__c1','f__f2','f__f2','g__g2','s__s3']}
        rep2 = {'a':['k__k1','p__p1','c__c1','o__o1','f__f1','g__g1','s__s1'],
                 'b':['k__k1','p__p1','c__c2',None,'f__f2','g__g1','s__s9'],
                 'c':[None,None,None,None,None,None,None]}
        rep3 = {'a':['k__k1','p__p1','c__c1','o__o1','f__f1','g__g1','s__s1'],
                 'b':['k__k1','p__p1','c__c2',None,'f__f2','g__g1','s__s2'],
                 'd':['k__k2','','c__c1','f__f2','f__f2','g__g2','s__s3']}
        rep4 = {'a':['k__k1','p__p1','c__c1','o__o1','f__f1','g__g1','s__s1'],
                 'b':['k__k1','p__p1','c__c10',None,'f__f2','g__g1','s__s2'],
                 'c':[None,None,None,None,None,None,None],
                 'd':['k__k1','','c__c1','f__f2','f__f2','g__g2','s__s3']}
        # if malformed, as in rep3 rep4 for d, count as contridiction
        exp = {'a':array([1.0, 1.0, 1.0, 1.0, 1.0, 1.0, 1.0]),
               'b':array([1.0, 1.0, 0.75, 1.0, 1.0, 1.0, 0.75]),
               'c':array([1.0, 1.0, 1.0, 1.0, 1.0, 1.0, 1.0]),
               'd':array([0.75, 0.25, 1.0, 0.25, 1.0, 1.0, 1.0])}
        obs = taxa_score(master, [rep1,rep2,rep3,rep4])
        self.assertEqual(obs.keys(), exp.keys())
        for k in exp:
            self.assertTrue(array_equal(obs[k], exp[k]))

    def test_taxa_score_hash(self):
        """test hash based consensus scoring"""
        master = {'a':['k__k1','p__p1','c__c1','o__o1','f__f1','g__g1','s__s1'],
                 'b':['k__k1','p__p1','c__c2',None,'f__f2','g__g1','s__s2'],
                 'c':[None,None,None,None,None,None,None],
                 'd':['k__k2','p__x','c__c1','o__o10','f__f2','g__g2','s__s3']}
        rep1 = {'a':['k__k1','p__p1','c__c1','o__o1','f__f1','g__g1','s__s1'],
                 'c':[None,None,None,None,None,None,None],
                 'd':['k__k2','p__','c__c1','f__f2','f__f2','g__g2','s__s3']}
        rep2 = {'a':['k__k1','p__p1','c__c1','o__o1','f__f1','g__g1','s__s1'],
                 'b':['k__k1','p__p1','c__c2',None,'f__f2','g__g1','s__s9'],
                 'c':[None,None,None,None,None,None,None]}
        rep3 = {'a':['k__k1','p__p1','c__c1','o__o1','f__f1','g__g1','s__s1'],
                 'b':['k__k1','p__p1','c__c2',None,'f__f2','g__g1','s__s2'],
                 'd':['k__k2','p__','c__c1','f__f2','f__f2','g__g2','s__s3']}
        rep4 = {'a':['k__k1','p__p1','c__c1','o__o1','f__f1','g__g1','s__s1'],
                 'b':['k__k1','p__p1','c__c10',None,'f__f2','g__g1','s__s2'],
                 'c':[None,None,None,None,None,None,None],
                 'd':['k__k1','p__','c__c1','f__f2','f__f2','g__g2','s__s3']}
        # if malformed, as in rep3 rep4 for d, count as contridiction
        exp = {'a':[1.0, 1.0, 1.0, 1.0, 1.0, 1.0, 1.0],
               'b':[1.0, 1.0, 0.75, 1.0, 1.0, 1.0, 0.75],
               'c':[1.0, 1.0, 1.0, 1.0, 1.0, 1.0, 1.0],
               'd':[0.75, 0.25, 1.0, 0.25, 1.0, 1.0, 1.0]}
        obs = taxa_score_hash(master, [rep1,rep2,rep3,rep4])
        self.assertEqual(obs.keys(), exp.keys())
        for k in exp:
            self.assertTrue(array_equal(obs[k], exp[k]))

    def test_hash_cons(self):
        """test turning consensus strings into hashes"""
        input = {'a':['k__k1','p__p1','c__c1','o__o1','f__f1','g__g1','s__s1'],
                 'b':['k__k1','p__p1','c__c2',None,'f__f2','g__g1','s__s2'],
                 'c':[None,None,None,None,None,None,None],
                 'd':['k__k2','p__x','c__c1','o__o10','f__f2','g__g2','s__s3']}
        exp = array([map(hash, input['a']),
                     map(hash, input['d']),
                     map(hash, input['c']),
                     map(hash, input['b'])])
        obs = hash_cons(input, ['a','d','c','b'], 7)
        self.assertTrue(array_equal(obs, exp))

        exp = array([map(hash, input['a']),
                     map(hash, input['d']),
                     [0,0,0,0,0,0,0],
                     map(hash, input['c']),
                     map(hash, input['b'])])
        obs = hash_cons(input, ['a','d','e','c','b'], 7)
        self.assertTrue(array_equal(obs, exp))

    def test_get_consensus_stats(self):
        """Produces the correct stats"""
        input = {'a':['k__k1','p__p1','c__c1','o__o1','f__f1','g__g1','s__s1'],
                 'b':['k__k1','p__p1','c__c2',None,'f__f2','g__g1','s__s2'],
                 'c':[None,None,None,None,None,None,None],
                 'd':['k__k2','','c__c1','f__f2','f__f2','g__g2','s__s3']}
{'fraction_non_alphanumeric': 0.26899187768752986, 'fraction_numerical': 0.07303255750460719, 'mean_word_length': 2.8146316063525125, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2848367', 'n_tokens_mistral': 6760, 'n_tokens_neox': 6515, 'n_words': 612}