# Path: denovoFilter/preliminary_filtering.py
# def fix_maf(max_af):
#     """ cleans up the max AF entries in the de novo dataframe
#     
#     Args:
#         max_af: pandas Series of maximum allele frequencies from the table of
#             candidate de novos.
#     
#     Returns:
#         modified pandas Series of max_af values
#     """
#     
#     max_af = max_af.copy()
#     
#     # some de novos have comma-separated lists of maf values. We select the
#     # first value (which correspononds to the alternate allele, the additional
#     # alleles are from denovogear selecting all possibly alternates at a
#     # candidate de novo site)
#     max_af = [ str(x).split(',')[0] for x in max_af ]
#     
#     # fix the null max AF values, and convert everything to floats
#     missing = set(['', '.', 'missing', 'nan', 'None', 'NA'])
#     max_af = [ float(x) if x not in missing else 0.0 for x in max_af ]
#     
#     return Series(max_af)
# 
# def preliminary_filtering(de_novos, sample_fails=None, maf_cutoff=0.01):
#     """run some preliminary filtering of de novos.
#     
#     We want to filter out de novos with high MAF, where they are not present in
#     the child VCF, or are present in the parental VCFs
#     
#     Args:
#         de_novos: dataframe of de novo variants
#         sample_fails: list of IDs for samples who have failed QC checks (due to
#             having too many de novo calls).
#         maf_cutoff: sites have to have a minor allele frequency below this.
#     
#     Returns:
#         data frame of de novos, but where we have excluded sites with high
#         allele frequency and where the variant appears in one or more parents.
#     """
#     
#     max_af = fix_maf(de_novos['max_af'])
#     passes_maf = (max_af <= maf_cutoff) & max_af.notnull()
#     
#     # keep sites in child vcf, and not in parental vcfs
#     appears_de_novo = (de_novos["in_child_vcf"] == 1) & (de_novos["in_father_vcf"] == 0) & (de_novos["in_mother_vcf"] == 0)
#     
#     good_samples = [True] * len(de_novos)
#     if sample_fails is not None:
#         # remove samples with too many candidates
#         good_samples = ~de_novos["person_stable_id"].isin(sample_fails)
#     
#     return passes_maf & appears_de_novo & good_samples
# 
# def check_coding(de_novos, cq_name="consequence"):
#     """check whether the consequence is for a coding consequence
#     
#     Args:
#         de_novos: dataframe of de novo variants
#         cq_name: name of the column for the consequence
#     
#     Returns:
#         data frame with an additional column to show if the site is coding
#     """
#     
#     # annotate candidates within coding exons or splice sites
#     coding_splicing = ["coding_sequence_variant", "frameshift_variant",
#         "inframe_deletion", "inframe_insertion", "initiator_codon_variant",
#         "missense_variant", "protein_altering_variant", "splice_acceptor_variant",
#         "splice_donor_variant", "splice_region_variant", "start_lost",
#         "stop_gained", "stop_lost", "synonymous_variant",
#         'conserved_exon_terminus_variant']
#     
#     return de_novos[cq_name].isin(coding_splicing)


import unittest
import time
from pandas import DataFrame, Series
from denovoFilter.preliminary_filtering import fix_maf, preliminary_filtering, \
    check_coding
'''




class TestPreliminaryFiltering(unittest.TestCase):
    
    def setUp(self):
        self.variants = DataFrame({'person_stable_id': ['a', 'b'],
            'max_af': ['0.001', 'missing'],
            'in_child_vcf': [1, 1],
            'in_father_vcf': [0, 0],
            'in_mother_vcf': [0, 0],
            })
    
    def test_fix_maf(self):
        ''' test that fix_maf works correctly
        '''
        
        values = Series(['0.001', '0.2,0.3', '', '.', 'missing', 0.5,
            None, float('nan'), 'NA'])
        
        self.assertTrue(all(fix_maf(values) == Series([0.001, 0.2, 0, 0, 0,
            0.5, 0, 0, 0])))
    
    def test_preliminary_filtering(self):
        ''' test that preliminary_filtering works correctly.
        '''
        
        # for a clean table, all variants should pass

# Path: denovoFilter/preliminary_filtering.py
# def fix_maf(max_af):
#     """ cleans up the max AF entries in the de novo dataframe
#     
#     Args:
#         max_af: pandas Series of maximum allele frequencies from the table of
#             candidate de novos.
#     
#     Returns:
#         modified pandas Series of max_af values
#     """
#     
#     max_af = max_af.copy()
#     
#     # some de novos have comma-separated lists of maf values. We select the
#     # first value (which correspononds to the alternate allele, the additional
#     # alleles are from denovogear selecting all possibly alternates at a
#     # candidate de novo site)
#     max_af = [ str(x).split(',')[0] for x in max_af ]
#     
#     # fix the null max AF values, and convert everything to floats
#     missing = set(['', '.', 'missing', 'nan', 'None', 'NA'])
#     max_af = [ float(x) if x not in missing else 0.0 for x in max_af ]
#     
#     return Series(max_af)
# 
# def preliminary_filtering(de_novos, sample_fails=None, maf_cutoff=0.01):
#     """run some preliminary filtering of de novos.
#     
#     We want to filter out de novos with high MAF, where they are not present in
#     the child VCF, or are present in the parental VCFs
#     
#     Args:
#         de_novos: dataframe of de novo variants
#         sample_fails: list of IDs for samples who have failed QC checks (due to
#             having too many de novo calls).
#         maf_cutoff: sites have to have a minor allele frequency below this.
#     
#     Returns:
#         data frame of de novos, but where we have excluded sites with high
#         allele frequency and where the variant appears in one or more parents.
#     """
#     
#     max_af = fix_maf(de_novos['max_af'])
#     passes_maf = (max_af <= maf_cutoff) & max_af.notnull()
#     
#     # keep sites in child vcf, and not in parental vcfs
#     appears_de_novo = (de_novos["in_child_vcf"] == 1) & (de_novos["in_father_vcf"] == 0) & (de_novos["in_mother_vcf"] == 0)
#     
#     good_samples = [True] * len(de_novos)
#     if sample_fails is not None:
#         # remove samples with too many candidates
#         good_samples = ~de_novos["person_stable_id"].isin(sample_fails)
#     
#     return passes_maf & appears_de_novo & good_samples
# 
# def check_coding(de_novos, cq_name="consequence"):
#     """check whether the consequence is for a coding consequence
#     
#     Args:
#         de_novos: dataframe of de novo variants
#         cq_name: name of the column for the consequence
#     
#     Returns:
#         data frame with an additional column to show if the site is coding
#     """
#     
#     # annotate candidates within coding exons or splice sites
#     coding_splicing = ["coding_sequence_variant", "frameshift_variant",
#         "inframe_deletion", "inframe_insertion", "initiator_codon_variant",
#         "missense_variant", "protein_altering_variant", "splice_acceptor_variant",
#         "splice_donor_variant", "splice_region_variant", "start_lost",
#         "stop_gained", "stop_lost", "synonymous_variant",
#         'conserved_exon_terminus_variant']
#     
#     return de_novos[cq_name].isin(coding_splicing)


# Path: tests/test_preliminary_filtering.py
import unittest
import time
from pandas import DataFrame, Series
from denovoFilter.preliminary_filtering import fix_maf, preliminary_filtering, \
    check_coding

'''




class TestPreliminaryFiltering(unittest.TestCase):
    
    def setUp(self):
        self.variants = DataFrame({'person_stable_id': ['a', 'b'],
            'max_af': ['0.001', 'missing'],
            'in_child_vcf': [1, 1],
            'in_father_vcf': [0, 0],
            'in_mother_vcf': [0, 0],
            })
    
    def test_fix_maf(self):
        ''' test that fix_maf works correctly
        '''
        
        values = Series(['0.001', '0.2,0.3', '', '.', 'missing', 0.5,
            None, float('nan'), 'NA'])
        
        self.assertTrue(all(fix_maf(values) == Series([0.001, 0.2, 0, 0, 0,
            0.5, 0, 0, 0])))
    
    def test_preliminary_filtering(self):
        ''' test that preliminary_filtering works correctly.
        '''
        
        # for a clean table, all variants should pass

        status = preliminary_filtering(self.variants)