# 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
copies or substantial portions of the Software.

THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER
IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN
CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
'''




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'])
        

# 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

copies or substantial portions of the Software.

THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER
IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN
CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
'''




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,