|
24 | 24 | class AltPercentageDepthFilter(Filter):
|
25 | 25 | min_percentage = 0
|
26 | 26 |
|
27 |
| - def __init__(self, args: argparse.Namespace, header: vcfpy.Header) -> "AltPercentageDepthFilter": |
| 27 | + def __init__( |
| 28 | + self, args: argparse.Namespace, header: vcfpy.Header |
| 29 | + ) -> "AltPercentageDepthFilter": |
28 | 30 | super().__init__(args, header)
|
29 | 31 | if (
|
30 | 32 | hasattr(args, "min_percentage_alt_filter")
|
@@ -71,34 +73,46 @@ def __call__(self, record: vcfpy.Record) -> Union[vcfpy.Record, None]:
|
71 | 73 | alt_percentage = record.INFO["AF1"] * 100
|
72 | 74 | elif self.header.has_header_line("INFO", "DP4"):
|
73 | 75 | (fwd_ref, rev_ref, fwd_alt, rev_alt) = record.INFO["DP4"]
|
74 |
| - alt_percentage = (fwd_alt + rev_alt) / (fwd_ref + rev_ref + fwd_alt + rev_alt) * 100 |
| 76 | + alt_percentage = ( |
| 77 | + (fwd_alt + rev_alt) / (fwd_ref + rev_ref + fwd_alt + rev_alt) * 100 |
| 78 | + ) |
75 | 79 | if alt_percentage is not None:
|
76 | 80 | retain.append(not alt_percentage < self.min_percentage)
|
77 |
| - elif self.header.has_header_line("INFO", "AF") or (self.header.has_header_line("INFO", "AO") and self.header.has_header_line("INFO", "DP")): |
| 81 | + elif self.header.has_header_line("INFO", "AF") or ( |
| 82 | + self.header.has_header_line("INFO", "AO") |
| 83 | + and self.header.has_header_line("INFO", "DP") # noqa: W503 |
| 84 | + ): |
78 | 85 | for i, _ in enumerate(record.ALT):
|
79 | 86 | if self.header.has_header_line("INFO", "AF"):
|
80 | 87 | alt_percentage = record.INFO["AF"][i] * 100
|
81 |
| - elif self.header.has_header_line("INFO", "AO") and self.header.has_header_line("INFO", "DP"): |
| 88 | + elif self.header.has_header_line( |
| 89 | + "INFO", "AO" |
| 90 | + ) and self.header.has_header_line("INFO", "DP"): |
82 | 91 | alt_percentage = (
|
83 | 92 | float(record.INFO["AO"][i]) / float(record.INFO["DP"]) * 100
|
84 | 93 | )
|
85 | 94 | retain.append(not alt_percentage < self.min_percentage)
|
86 | 95 | else:
|
87 | 96 | # we've got nothing to add to retain - leave it as an empty list
|
88 |
| - print("No alt allele depth information found in VCF, disabling alt allele percentage filter", file=sys.stderr) |
| 97 | + print( |
| 98 | + "No alt allele depth information found in VCF, disabling alt allele percentage filter", |
| 99 | + file=sys.stderr, |
| 100 | + ) |
89 | 101 | if not any(retain):
|
90 | 102 | return None
|
91 |
| - |
| 103 | + |
92 | 104 | new_ALT = [alt for i, alt in enumerate(record.ALT) if retain[i]]
|
93 | 105 | new_INFO = OrderedDict()
|
94 | 106 | # these are produced by snpEff and keys occur once per implicated gene
|
95 | 107 | # the simplest solution is to copy them all across
|
96 | 108 | for key in record.INFO:
|
97 |
| - if self.header.get_info_field_info(key).number == 'A': |
| 109 | + if self.header.get_info_field_info(key).number == "A": |
98 | 110 | # 'A' fields have one entry for each alternative allele - copy only those for retained alleles
|
99 | 111 | field_len = len(record.INFO[key])
|
100 | 112 | retain_len = len(retain)
|
101 |
| - assert field_len == retain_len, f"Length of array-type INFO field ({field_len}) does not match length of retain ({retain_len})" |
| 113 | + assert ( |
| 114 | + field_len == retain_len |
| 115 | + ), f"Length of array-type INFO field ({field_len}) does not match length of retain ({retain_len})" |
102 | 116 | new_INFO[key] = [
|
103 | 117 | # retain all ANN records and the only those other records that correspond to alts that we retain
|
104 | 118 | el
|
|
0 commit comments