-
Notifications
You must be signed in to change notification settings - Fork 243
/
plink_to_vcf.py
158 lines (142 loc) · 5.85 KB
/
plink_to_vcf.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
#!/usr/bin/env python
"""Convert Plink ped/map files into VCF format using plink and Plink/SEQ.
Requires:
plink: http://pngu.mgh.harvard.edu/~purcell/plink/
PLINK/SEQ: http://atgu.mgh.harvard.edu/plinkseq/
bx-python: https://bitbucket.org/james_taylor/bx-python/wiki/Home
You also need the genome reference file in 2bit format:
http://genome.ucsc.edu/FAQ/FAQformat.html#format7
using faToTwoBit:
http://hgdownload.cse.ucsc.edu/admin/exe/linux.x86_64/
Usage:
plink_to_vcf.py <ped file> <map file> <UCSC reference file in 2bit format)
"""
import os
import sys
import subprocess
from bx.seq import twobit
def main(ped_file, map_file, ref_file):
base_dir = os.getcwd()
pbed_prefix = convert_to_plink_bed(ped_file, map_file, base_dir)
vcf_file = convert_bed_to_vcf(pbed_prefix, ped_file, base_dir)
fix_nonref_positions(vcf_file, ref_file)
def convert_to_plink_bed(ped_file, map_file, base_dir):
# from ubuntu package, 'plink' otherwise
for plink_cl in ["p-link", "plink"]:
try:
subprocess.check_call([plink_cl, "--help"])
break
except:
pass
plink_prefix = os.path.splitext(os.path.basename(ped_file))[0]
work_dir = os.path.join(base_dir, "vcfconvert")
if not os.path.exists(work_dir):
os.makedirs(work_dir)
out_base = os.path.join(work_dir, plink_prefix)
if not os.path.exists("{0}.bed".format(out_base)):
subprocess.check_call([plink_cl, "--noweb",
"--ped", ped_file, "--map", map_file,
"--make-bed", "--out", out_base])
return out_base
def convert_bed_to_vcf(pbed_prefix, ped_file, base_dir):
out_file = os.path.join(base_dir,
"{0}.vcf".format(os.path.splitext(os.path.basename(ped_file))[0]))
if not os.path.exists(out_file):
subprocess.check_call(["pseq", pbed_prefix, "new-project"])
subprocess.check_call(["pseq", pbed_prefix, "load-plink",
"--file", pbed_prefix, "--id", "vcfconvert"])
with open(out_file, "w") as out_handle:
subprocess.check_call(["pseq", pbed_prefix, "write-vcf"],
stdout=out_handle)
return out_file
def fix_line_problems(parts):
"""Fix problem alleles and reference/variant bases in VCF line.
"""
varinfo = parts[:9]
genotypes = []
# replace haploid calls
for x in parts[9:]:
if len(x) == 1:
x = "./."
genotypes.append(x)
if varinfo[3] == "0": varinfo[3] = "N"
if varinfo[4] == "0": varinfo[4] = "N"
return varinfo, genotypes
def fix_vcf_line(parts, ref_base):
"""Orient VCF allele calls with respect to reference base.
Handles cases with ref and variant swaps. strand complements.
"""
swap = {"1/1": "0/0", "0/1": "0/1", "0/0": "1/1", "./.": "./."}
complements = {"G": "C", "A": "T", "C": "G", "T": "A", "N": "N"}
varinfo, genotypes = fix_line_problems(parts)
ref, var = varinfo[3:5]
# non-reference regions or non-informative, can't do anything
if ref_base in [None, "N"] or set(genotypes) == set(["./."]):
varinfo = None
# matching reference, all good
elif ref_base == ref:
assert ref_base == ref, (ref_base, parts)
# swapped reference and alternate regions
elif ref_base == var or ref in ["N", "0"]:
varinfo[3] = var
varinfo[4] = ref
genotypes = [swap[x] for x in genotypes]
# reference is on alternate strand
elif ref_base != ref and complements[ref] == ref_base:
varinfo[3] = complements[ref]
varinfo[4] = complements[var]
# unspecified alternative base
elif ref_base != ref and var in ["N", "0"]:
varinfo[3] = ref_base
varinfo[4] = ref
genotypes = [swap[x] for x in genotypes]
# swapped and on alternate strand
elif ref_base != ref and complements[var] == ref_base:
varinfo[3] = complements[var]
varinfo[4] = complements[ref]
genotypes = [swap[x] for x in genotypes]
else:
print "Did not associate ref {0} with line: {1}".format(
ref_base, varinfo)
if varinfo is not None:
return varinfo + genotypes
def fix_nonref_positions(in_file, ref_file):
"""Fix Genotyping VCF positions where the bases are all variants.
The plink/pseq output does not handle these correctly, and
has all reference/variant bases reversed.
"""
ignore_chrs = ["."]
ref2bit = twobit.TwoBitFile(open(ref_file))
out_file = apply("{0}-fix{1}".format, os.path.splitext(in_file))
with open(in_file) as in_handle:
with open(out_file, "w") as out_handle:
for line in in_handle:
if line.startswith("#"):
out_handle.write(line)
else:
parts = line.rstrip("\r\n").split("\t")
pos = int(parts[1])
# handle chr/non-chr naming
if parts[0] not in ref2bit.keys():
parts[0] = parts[0].replace("chr", "")
ref_base = None
if parts[0] not in ignore_chrs:
try:
ref_base = ref2bit[parts[0]].get(pos-1, pos).upper()
except Exception, msg:
# off the end of the chromosome
if str(msg).startswith("end before start"):
print msg
else:
print parts
raise
parts = fix_vcf_line(parts, ref_base)
if parts is not None:
out_handle.write("\t".join(parts) + "\n")
return out_file
if __name__ == "__main__":
if len(sys.argv) != 4:
print "Incorrect arguments"
print __doc__
sys.exit(1)
main(*sys.argv[1:])