<commit_before>#!/usr/bin/env python
# Description = Filters prteins that contains domains of interest

import os


class FilterPredictedDomains(object):
    '''Filter the predicted domain by
    parameters and domain of interest'''
    def __init__(self, prediction_method,
                 analysis_result_path,
                 domain_description_file,
                 go_path,
                 filtered_result_path,
                 all_prediction_output,
                 filter_parameters):
        self._prediction_method = prediction_method
        self._analysis_result_path = analysis_result_path
        self._domain_description_file = domain_description_file
        self._go_path = go_path
        self._filtered_result_path = filtered_result_path
        self._all_prediction_output = all_prediction_output
        self._filter_parameters = filter_parameters
        
        self._id_description_dict = {}
        self._domain_length = {}
        self._cdd_go_dict = {}
        self._interpro_go_dict = {}
        self._filter_parameters_dict = {}
        self._result_detail_dict = {}
        
    def streamline_filter_predicted_domains(self):
        '''To call from apricot'''
        self.read_id_description_file()
        self.read_filter_parameters()
        self.summarize_analysis_result_files()
        self.create_filtered_result_file()
        
    def read_id_description_file(self):
        '''read id file.'''
        with open(self._domain_description_file,
                  'r') as id_description_fh:
            for entry in id_description_fh:
                if 'ReferenceId' not in entry:
                    try:
                        length = entry.split('\t')[-2]
                        parent_id = entry.split('\t')[1]
                        if 'smart' in parent_id:
                            parent_id = parent_id.replace('smart', 'SM')
                        elif 'pfam' in parent_id:
                            parent_id = parent_id.replace('pfam', 'PF')
                        if parent_id not in set(self._domain_length.keys()):
                            self._domain_length[parent_id] = length
                        self._id_description_dict[entry.split(
                            '\t')[0]] = entry.strip().replace(
                            entry.split('\t')[1], parent_id)
                        self._id_description_dict[entry.split(
                            '\t')[1]] = entry.strip().replace(
                            entry.split('\t')[1], parent_id)
                    except IndexError:
                        pass
        return self._id_description_dict, self._domain_length
    
    def read_filter_parameters(self):
        '''Records the filtering parameter for cut-off'''
        for param in self._filter_parameters:
            param_key = param.split(':')[0]
            param_val = param.split(':')[1]    
            if not param_val == 'NA':
                self._filter_parameters_dict[param_key] = float(param_val)
        return self._filter_parameters_dict
    
    def summarize_analysis_result_files(self):
        '''summarize all the the results in dictionary'''
        if self._prediction_method == 'cdd':
            self.read_cdd_go_file()
            self.summarize_rps_analysis_result_files()
        elif self._prediction_method == 'ipr':
            self.read_interpro_go_file()
            self.summarize_ipr_analysis_result_files()
            
    def read_cdd_go_file(self):
        '''read mapped file for cdd id to go.'''
        with open(self._go_path+'/mapped_cdd_to_go.csv',
                  'r') as cdd_go_fh:
            for cdd_entry in cdd_go_fh:
                self._cdd_go_dict[cdd_entry.split(
                    '\t')[0]] = cdd_entry.split('\t')[-1].strip()
        return self._cdd_go_dict
        
    def read_interpro_go_file(self):
        '''read mapped file for interpro id to go.'''
        with open(self._go_path+'/mapped_interpro_to_go.csv',
                  'r') as interpro_go_fh:
            for interpro_entry in interpro_go_fh:
                self._interpro_go_dict[interpro_entry.split(
                    '\t')[0]] = interpro_entry.split('\t')[-1].strip()
        return self._interpro_go_dict
        
    def summarize_rps_analysis_result_files(self):
        '''summarize all the RPS-BLAST results in dictionary'''
        for individual_rps_result_file in os.listdir(
                self._analysis_result_path):
            protein_id = individual_rps_result_file.split('.')[0]
            with open(
                    self._analysis_result_path+'/'+individual_rps_result_file
            ) as individual_rps_result_fh:
                for individual_rps_result_section in \
                individual_rps_result_fh.read().split('>gnl'):
                    if individual_rps_result_section.startswith('|CDD|'):
                        stat_data = individual_rps_result_section.split(
                            "Score = ")
                        for individual_rps_result in stat_data[0].split('\n'):
                            if individual_rps_result.startswith('|CDD|'):
                                cdd_main = self._compile_cdd_main(
                                    individual_rps_result)
                            if individual_rps_result.startswith("Length"):
                                if " = " in individual_rps_result:
                                    individual_rps_result = individual_rps_result.replace(" = ", "=")
                                try:
                                    length = individual_rps_result.split(
                                        'Length=')[1].strip()
                                except:
                                    length = self._domain_length[
                                        cdd_main.split('\t')[1]]
                        for each_stat_group in stat_data[1:]:
                            check_list = set()
                            parameter_dict = self._compile_cdd_stat(
                                each_stat_group)
                            coverage_percent = "%.4f" % ((int(
                                parameter_dict[
                                    "cover_length"])/int(length)) * 100)
                            if float(coverage_percent) > 100:
                                coverage_percent = 100
                            similarity_percent = "%.4f" % ((int(
                                parameter_dict["similarity_value"].split(
                                    '/')[0])/int(length))*100)
                            identity_percent = "%.4f" % ((int(
                                parameter_dict["identity_value"].split(
                                    '/')[0])/int(length))*100)
                            if parameter_dict["gaps_value"] == 'None':
                                gap_percent = 0
                            else:
                                gap_percent = "%.4f" % ((int(
                                    parameter_dict["gaps_value"].split(
                                        '/')[0])/int(length)) * 100)
                            evalue = parameter_dict["evalue"]
                            bits = parameter_dict["bit"].split(' ')[0]
                            start = parameter_dict["start"]
                            stop = parameter_dict["stop"]
                            compiled_data = '\t'.join([
                                'CDD', protein_id, cdd_main,
                                str(length), str(start), str(stop),
                                str(evalue),
                                str(parameter_dict["bit"]),<commit_msg>Update filter_keyword_selected_result.py<commit_after>#!/usr/bin/env python
# Description = Filters prteins that contains domains of interest

import os


class FilterPredictedDomains(object):
    '''Filter the predicted domain by
    parameters and domain of interest'''
    def __init__(self, prediction_method,
                 analysis_result_path,
                 domain_description_file,
                 go_path,
                 filtered_result_path,
                 all_prediction_output,
                 filter_parameters):
        self._prediction_method = prediction_method
        self._analysis_result_path = analysis_result_path
        self._domain_description_file = domain_description_file
        self._go_path = go_path
        self._filtered_result_path = filtered_result_path
        self._all_prediction_output = all_prediction_output
        self._filter_parameters = filter_parameters
        
        self._id_description_dict = {}
        self._domain_length = {}
        self._cdd_go_dict = {}
        self._interpro_go_dict = {}
        self._filter_parameters_dict = {}
        self._result_detail_dict = {}
        
    def streamline_filter_predicted_domains(self):
        '''To call from apricot'''
        self.read_id_description_file()
        self.read_filter_parameters()
        self.summarize_analysis_result_files()
        self.create_filtered_result_file()
        
    def read_id_description_file(self):
        '''read id file.'''
        with open(self._domain_description_file,
                  'r') as id_description_fh:
            for entry in id_description_fh:
                if 'ReferenceId' not in entry:
                    try:
                        length = entry.split('\t')[-2]
                        parent_id = entry.split('\t')[1]
                        if 'smart' in parent_id:
                            parent_id = parent_id.replace('smart', 'SM')
                        elif 'pfam' in parent_id:
                            parent_id = parent_id.replace('pfam', 'PF')
                        if parent_id not in set(self._domain_length.keys()):
                            self._domain_length[parent_id] = length
                        self._id_description_dict[entry.split(
                            '\t')[0]] = entry.strip().replace(
                            entry.split('\t')[1], parent_id)
                        self._id_description_dict[entry.split(
                            '\t')[1]] = entry.strip().replace(
                            entry.split('\t')[1], parent_id)
                    except IndexError:
                        pass
        return self._id_description_dict, self._domain_length
    
    def read_filter_parameters(self):
        '''Records the filtering parameter for cut-off'''
        for param in self._filter_parameters:
            param_key = param.split(':')[0]
            param_val = param.split(':')[1]    
            if not param_val == 'NA':
                self._filter_parameters_dict[param_key] = float(param_val)
        return self._filter_parameters_dict
    
    def summarize_analysis_result_files(self):
        '''summarize all the the results in dictionary'''
        if self._prediction_method == 'cdd':
            self.read_cdd_go_file()
            self.summarize_rps_analysis_result_files()
        elif self._prediction_method == 'ipr':
            self.read_interpro_go_file()
            self.summarize_ipr_analysis_result_files()
            
    def read_cdd_go_file(self):
        '''read mapped file for cdd id to go.'''
        with open(self._go_path+'/mapped_cdd_to_go.csv',
                  'r') as cdd_go_fh:
            for cdd_entry in cdd_go_fh:
                self._cdd_go_dict[cdd_entry.split(
                    '\t')[0]] = cdd_entry.split('\t')[-1].strip()
        return self._cdd_go_dict
        
    def read_interpro_go_file(self):
        '''read mapped file for interpro id to go.'''
        with open(self._go_path+'/mapped_interpro_to_go.csv',
                  'r') as interpro_go_fh:
            for interpro_entry in interpro_go_fh:
                self._interpro_go_dict[interpro_entry.split(
                    '\t')[0]] = interpro_entry.split('\t')[-1].strip()
        return self._interpro_go_dict
        
    def summarize_rps_analysis_result_files(self):
        '''summarize all the RPS-BLAST results in dictionary'''
        for individual_rps_result_file in os.listdir(
                self._analysis_result_path):
            protein_id = individual_rps_result_file.split('.')[0]
            with open(
                    self._analysis_result_path+'/'+individual_rps_result_file
            ) as individual_rps_result_fh:
                for individual_rps_result_section in \
                individual_rps_result_fh.read().split('>gnl'):
                    if individual_rps_result_section.startswith('|CDD|'):
                        stat_data = individual_rps_result_section.split(
                            "Score = ")
                        for individual_rps_result in stat_data[0].split('\n'):
                            if individual_rps_result.startswith('|CDD|'):
                                cdd_main = self._compile_cdd_main(
                                    individual_rps_result)
                            try:
                                if individual_rps_result.startswith("Length"):
                                    if " = " in individual_rps_result:
                                        individual_rps_result = individual_rps_result.replace(" = ", "=")
                                        length = individual_rps_result.split(
                                        'Length=')[1].strip()
                            except:
                                length = self._domain_length[
                                    cdd_main.split('\t')[1]]
                        for each_stat_group in stat_data[1:]:
                            check_list = set()
                            parameter_dict = self._compile_cdd_stat(
                                each_stat_group)
                            coverage_percent = "%.4f" % ((int(
                                parameter_dict[
                                    "cover_length"])/int(length)) * 100)
                            if float(coverage_percent) > 100:
                                coverage_percent = 100
                            similarity_percent = "%.4f" % ((int(
                                parameter_dict["similarity_value"].split(
                                    '/')[0])/int(length))*100)
                            identity_percent = "%.4f" % ((int(
                                parameter_dict["identity_value"].split(
                                    '/')[0])/int(length))*100)
                            if parameter_dict["gaps_value"] == 'None':
                                gap_percent = 0
                            else:
                                gap_percent = "%.4f" % ((int(
                                    parameter_dict["gaps_value"].split(
                                        '/')[0])/int(length)) * 100)
                            evalue = parameter_dict["evalue"]
                            bits = parameter_dict["bit"].split(' ')[0]
                            start = parameter_dict["start"]
                            stop = parameter_dict["stop"]
                            compiled_data = '\t'.join([
                                'CDD', protein_id, cdd_main,
                                str(length), str(start), str(stop),
                                str(evalue),
                                str(parameter_dict["bit"]),
{'fraction_non_alphanumeric': 0.09216081054750926, 'fraction_numerical': 0.005715399103721504, 'mean_word_length': 1.3591236402635207, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 7, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 25, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '922', 'n_tokens_mistral': 4046, 'n_tokens_neox': 3893, 'n_words': 660}