|
1 |
| -import logging |
2 | 1 | import subprocess
|
3 | 2 | import tempfile
|
4 | 3 | from typing import Dict, List, Tuple
|
5 | 4 |
|
6 | 5 | CHROM_INDEX = 0
|
7 | 6 | START_INDEX = 1
|
8 | 7 | STOP_INDEX = 2
|
9 |
| -LOG = logging.getLogger(__name__) |
10 | 8 |
|
11 | 9 |
|
12 | 10 | def get_d4tools_intervals_completeness(
|
13 | 11 | d4_file_path: str, bed_file_path: str, completeness_thresholds: List[int]
|
14 | 12 | ) -> List[Dict]:
|
15 | 13 | """Return coverage completeness over all intervals of a bed file using the perc_cov d4tools command."""
|
16 | 14 | threshold_stats = []
|
17 |
| - try: |
18 |
| - d4tools_stats_perc_cov: str = subprocess.check_output( |
19 |
| - [ |
20 |
| - "d4tools", |
21 |
| - "stat", |
22 |
| - "-s", |
23 |
| - f"perc_cov={','.join(str(threshold) for threshold in completeness_thresholds)}", |
24 |
| - "--region", |
25 |
| - bed_file_path, |
26 |
| - d4_file_path, |
27 |
| - ], |
28 |
| - text=True, |
29 |
| - ) |
30 |
| - |
31 |
| - for line in d4tools_stats_perc_cov.splitlines(): |
32 |
| - stats_dict: Dict = dict( |
33 |
| - ( |
34 |
| - zip( |
35 |
| - completeness_thresholds, |
36 |
| - [float(stat) for stat in line.rstrip().split("\t")[3:]], |
37 |
| - ) |
| 15 | + d4tools_stats_perc_cov: str = subprocess.check_output( |
| 16 | + [ |
| 17 | + "d4tools", |
| 18 | + "stat", |
| 19 | + "-s", |
| 20 | + f"perc_cov={','.join(str(threshold) for threshold in completeness_thresholds)}", |
| 21 | + "--region", |
| 22 | + bed_file_path, |
| 23 | + d4_file_path, |
| 24 | + ], |
| 25 | + text=True, |
| 26 | + ) |
| 27 | + for line in d4tools_stats_perc_cov.splitlines(): |
| 28 | + stats_dict: Dict = dict( |
| 29 | + ( |
| 30 | + zip( |
| 31 | + completeness_thresholds, |
| 32 | + [float(stat) for stat in line.rstrip().split("\t")[3:]], |
38 | 33 | )
|
39 | 34 | )
|
40 |
| - threshold_stats.append(stats_dict) |
41 |
| - except subprocess.CalledProcessError as e: |
42 |
| - LOG.error(f"d4tools stat -s perc_cov failed with the following error:{e}") |
| 35 | + ) |
| 36 | + threshold_stats.append(stats_dict) |
43 | 37 |
|
44 | 38 | return threshold_stats
|
45 | 39 |
|
|
0 commit comments