edited format to compare_and_merge_lines

Author: khetherin

convert_gvf_to_vcf/convertGVFtoVCF.py

@@ -286,6 +286,11 @@ def format_vcf_datalines(list_of_vcf_objects, list_of_sample_names):
     return formatted_vcf_datalines
 
 def get_bigger_dictionary(dict1, dict2):
+    """Determines the biggest of two dictionaries
+    :param: dictionary1
+    :param: dictinary2
+    :return: smallest, largest
+    """
     if len(dict1) > len(dict2):
         biggest_dict = dict1
         smallest_dict = dict2
@@ -298,13 +303,20 @@ def get_bigger_dictionary(dict1, dict2):
     return smallest_dict, biggest_dict
 
 def merge_and_add(previous_element, current_element, delimiter):
+    """ If same, use current element. If different, merge with delimiter.
+    :param: previous_element
+    :param: current_element
+    :param: delimiter
+    :return: merged element
+    """
     if previous_element == current_element:
         merged_element = current_element
     else:
         merged_element = delimiter.join((previous_element, current_element))
     return merged_element
 
 def compare_and_merge_lines(list_of_formatted_vcf_datalines, headerline):
+    merged_lines = []
     for previous, current in zip(list_of_formatted_vcf_datalines, list_of_formatted_vcf_datalines[1:]):
         # print(f"previous line:\n{previous}\ncurrent line:\n{current}\n")
         previous_tokens = previous.split("\t")
@@ -319,7 +331,7 @@ def compare_and_merge_lines(list_of_formatted_vcf_datalines, headerline):
                 and previous_data["POS"] == current_data["POS"]
                 and previous_data["REF"] == current_data["REF"]
         ):
-            # print("True - merge")
+            print("True - merge")
             merged_data["#CHROM"] = current_data["#CHROM"]
             merged_data["POS"] = current_data["POS"]
             merged_data["ID"] = merge_and_add(previous_data["ID"], current_data["ID"], ";")
@@ -401,9 +413,26 @@ def compare_and_merge_lines(list_of_formatted_vcf_datalines, headerline):
                 sample_format_string =':'.join(flat_values)
 
                 merged_data[sample_name] = sample_format_string
-        # else:
-        #     print("False - keep previous")
-        #     print("---")
+            merged_lines.append(merged_data)
+            print("---")
+        else:
+            print("False - keep previous")
+            merged_data["#CHROM"] = previous_data["#CHROM"]
+            merged_data["POS"] = previous_data["POS"]
+            merged_data["ID"] = previous_data["ID"]
+            merged_data["REF"] = previous_data["REF"]
+            merged_data["ALT"] = previous_data["ALT"]
+            merged_data["QUAL"] = previous_data["QUAL"]
+            merged_data["FILTER"] = previous_data["FILTER"]
+            merged_data["INFO"] = previous_data["INFO"]
+            merged_data["FORMAT"] = previous_data["FORMAT"]
+            sample_names = header_fields[9:]
+            for sample in sample_names:
+                merged_data[sample] = previous_data[sample]
+
+            merged_lines.append(merged_data)
+            print("---")
+    return merged_lines
 
 
 
@@ -478,9 +507,9 @@ def main():
         vcf_output.write(f"{header_fields}\n")
         logger.info("Generating the VCF datalines")
         formatted_vcf_datalines = format_vcf_datalines(list_of_vcf_objects, samples)
-        compare_and_merge_lines(formatted_vcf_datalines, header_fields)
-        for line in formatted_vcf_datalines:
-            vcf_output.write(f"{line}\n")
+        merged_lines = compare_and_merge_lines(formatted_vcf_datalines, header_fields)
+        for line in merged_lines:
+            vcf_output.write("\t".join(str(val) for val in line.values()) + "\n")
     vcf_output.close()
     logger.info("GVF to VCF conversion complete")
 

tests/test_convert_gvf_to_vcf.py

@@ -4,10 +4,9 @@
 #from convert_gvf_to_vcf.utils import read_file
 from convert_gvf_to_vcf.convertGVFtoVCF import generate_custom_unstructured_meta_line, read_in_gvf_file, \
     gvf_features_to_vcf_objects, format_vcf_datalines, \
-    generate_vcf_metainfo, generate_vcf_header_structured_lines,  \
-    generate_vcf_header_line,  \
-    format_sample_values, read_yaml, read_pragma_mapper, generate_symbolic_allele_dict
-
+    generate_vcf_metainfo, generate_vcf_header_structured_lines, \
+    generate_vcf_header_line, \
+    format_sample_values, read_yaml, read_pragma_mapper, generate_symbolic_allele_dict, get_bigger_dictionary, merge_and_add, compare_and_merge_lines
 from convert_gvf_to_vcf.vcfline import VcfLine
 from convert_gvf_to_vcf.gvffeature import GvfFeatureline
 
@@ -426,7 +425,21 @@ def test_format_sample_values(self):
         assert actual_number_of_tokens == number_of_tokens_should_have, f"must have {number_of_tokens_should_have}"
         assert sample_format_values_string == ".:.\t.:.\t.:.\t3:0:1", "String must match expected value"
 
-
+    def test_get_bigger_dictionary(self):
+        dictionary1 = {"key1": "value1"}
+        dictionary2 = {"key1": "value1", "key2": "value2"}
+        small, large = get_bigger_dictionary(dictionary1, dictionary2)
+        assert len(large) > len(small)
+
+    def merge_and_add(self):
+        previous="1"
+        current ="2"
+        delimiter =";"
+        merged_string = merge_and_add(previous, current, delimiter)
+        assert len(merged_string) > 1
+
+    def compare_and_merge_lines(self):
+        pass
 
     def test_format_vcf_datalines(self):
         gvf_pragmas, gvf_non_essential, gvf_lines_obj_list = read_in_gvf_file(self.input_file)