FFPE deamination QC pipeline
- ffpe_damage_v2.py: normalized C>T/G>A profiling (all 12 substitutions, R1/R2 and strand profiles, 5'/3' read-end distance, BED tracks, plots) - ffpe_compare.py: low/moderate/high classification vs control samples - make_test_data.py: synthetic BAM with known damage for validation
This commit is contained in:
@@ -0,0 +1,214 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Generate synthetic reference + paired-end BAM with a known damage profile.
|
||||
|
||||
Two scenarios:
|
||||
* damaged: plus-strand deamination C>T at the 5' end of forward reads (R1)
|
||||
and minus-strand deamination G>A at the 3' end of reverse
|
||||
reads (R2, fragment far end) -- the two strands of the same
|
||||
cytosine-deamination event, appearing at opposite read ends
|
||||
+ one "real" heterozygous SNP with no read-position / strand
|
||||
bias, which the pipeline must NOT mistake for damage
|
||||
* clean: background sequencing error only, no end-associated damage
|
||||
(used as a control sample)
|
||||
|
||||
Ground truth for end-to-end validation of ffpe_damage_v2.py / ffpe_compare.py.
|
||||
"""
|
||||
|
||||
import argparse
|
||||
import os
|
||||
import random
|
||||
|
||||
import pysam
|
||||
|
||||
|
||||
COMP = str.maketrans("ACGT", "TGCA")
|
||||
|
||||
|
||||
def rc(seq):
|
||||
return seq.translate(COMP)[::-1]
|
||||
|
||||
|
||||
def make_reference(outdir, n_chroms, length, seed):
|
||||
rng = random.Random(seed)
|
||||
fa_path = os.path.join(outdir, "synthetic_reference.fa")
|
||||
with open(fa_path, "w") as fh:
|
||||
for i in range(n_chroms):
|
||||
name = f"chr{i + 1}"
|
||||
seq = "".join(rng.choice("ACGT") for _ in range(length))
|
||||
fh.write(f">{name}\n")
|
||||
for j in range(0, length, 80):
|
||||
fh.write(seq[j:j + 80] + "\n")
|
||||
pysam.faidx(fa_path)
|
||||
return fa_path
|
||||
|
||||
|
||||
def mutate_sequence(seq, positions, base, alt, rng, p):
|
||||
"""Set bases at given query positions to alt with probability p."""
|
||||
s = list(seq)
|
||||
for pos in positions:
|
||||
if s[pos] == base and rng.random() < p:
|
||||
s[pos] = alt
|
||||
return "".join(s)
|
||||
|
||||
|
||||
def build_bam(fa_path, bam_path, rng, chroms, length, n_frags, read_len,
|
||||
frag_len, damage_rate, snp_pos, snp_alt, background_error,
|
||||
mapq, baseq):
|
||||
fasta = pysam.FastaFile(fa_path)
|
||||
|
||||
header = {
|
||||
"HD": {"VN": "1.6", "SO": "unsorted"},
|
||||
"SQ": [{"SN": c, "LN": length} for c in chroms],
|
||||
}
|
||||
|
||||
with pysam.AlignmentFile(bam_path, "wb", header=header) as out:
|
||||
read_id = 0
|
||||
for chrom in chroms:
|
||||
for _ in range(n_frags):
|
||||
start = rng.randint(0, length - frag_len - 1)
|
||||
frag_start = start
|
||||
frag_end = start + frag_len
|
||||
ref_r1 = fasta.fetch(chrom, frag_start, frag_start + read_len)
|
||||
ref_r2 = fasta.fetch(chrom, frag_end - read_len, frag_end)
|
||||
|
||||
q1 = ref_r1
|
||||
q2 = ref_r2
|
||||
|
||||
# --- damage injection: fragment ends ---
|
||||
# forward-strand read (R1): 5' end of stored read
|
||||
# (plus-strand deamination) -> ref C shows as read T
|
||||
q1 = mutate_sequence(q1, range(min(5, read_len)),
|
||||
"C", "T", rng, damage_rate)
|
||||
# reverse-strand read (R2): sequenced 5' end is the fragment's
|
||||
# far end, which lands on the 3' end of the stored read.
|
||||
# (minus-strand deamination) -> ref plus-strand G shows as read A
|
||||
q2 = mutate_sequence(q2, range(read_len - 5, read_len),
|
||||
"G", "A", rng, damage_rate)
|
||||
|
||||
# --- "real" SNP, heterozygous, no positional/strand bias ---
|
||||
if frag_start <= snp_pos < frag_end:
|
||||
if rng.random() < 0.5:
|
||||
qp_r1 = snp_pos - frag_start
|
||||
if 0 <= qp_r1 < read_len:
|
||||
q1 = q1[:qp_r1] + snp_alt + q1[qp_r1 + 1:]
|
||||
qp_r2 = snp_pos - (frag_end - read_len)
|
||||
if 0 <= qp_r2 < read_len:
|
||||
if rng.random() < 0.5:
|
||||
q2 = q2[:qp_r2] + snp_alt + q2[qp_r2 + 1:]
|
||||
|
||||
# --- background error ---
|
||||
q1 = _background_error(q1, rng, background_error)
|
||||
q2 = _background_error(q2, rng, background_error)
|
||||
|
||||
quals1 = pysam.qualitystring_to_array(
|
||||
bytes([baseq + 33]) * read_len).tobytes()
|
||||
quals2 = pysam.qualitystring_to_array(
|
||||
bytes([baseq + 33]) * read_len).tobytes()
|
||||
|
||||
r1 = pysam.AlignedSegment()
|
||||
r1.query_name = f"{chrom}_frag{read_id}_R1"
|
||||
r1.query_sequence = q1
|
||||
r1.flag = 99 # paired, proper, mate reverse, read1
|
||||
r1.reference_id = out.get_tid(chrom)
|
||||
r1.reference_start = frag_start
|
||||
r1.mapping_quality = mapq
|
||||
r1.cigar = ((0, read_len),)
|
||||
r1.query_qualities = quals1
|
||||
r1.next_reference_id = out.get_tid(chrom)
|
||||
r1.next_reference_start = frag_end - read_len
|
||||
r1.template_length = frag_len
|
||||
|
||||
r2 = pysam.AlignedSegment()
|
||||
r2.query_name = f"{chrom}_frag{read_id}_R2"
|
||||
r2.query_sequence = q2
|
||||
r2.flag = 147 # paired, proper, reverse, read2
|
||||
r2.reference_id = out.get_tid(chrom)
|
||||
r2.reference_start = frag_end - read_len
|
||||
r2.mapping_quality = mapq
|
||||
r2.cigar = ((0, read_len),)
|
||||
r2.query_qualities = quals2
|
||||
r2.next_reference_id = out.get_tid(chrom)
|
||||
r2.next_reference_start = frag_start
|
||||
r2.template_length = -frag_len
|
||||
|
||||
out.write(r1)
|
||||
out.write(r2)
|
||||
read_id += 1
|
||||
|
||||
fasta.close()
|
||||
|
||||
|
||||
def _background_error(seq, rng, p):
|
||||
s = list(seq)
|
||||
bases = "ACGT"
|
||||
for i in range(len(s)):
|
||||
if rng.random() < p:
|
||||
alt = rng.choice(bases.replace(s[i], ""))
|
||||
s[i] = alt
|
||||
return "".join(s)
|
||||
|
||||
|
||||
def main():
|
||||
ap = argparse.ArgumentParser(description=__doc__)
|
||||
ap.add_argument("--outdir", default="test_data")
|
||||
ap.add_argument("--bam-out", default=None,
|
||||
help="output BAM path (default <outdir>/sample.bam)")
|
||||
ap.add_argument("--seed", type=int, default=42)
|
||||
ap.add_argument("--n-chroms", type=int, default=3)
|
||||
ap.add_argument("--chrom-length", type=int, default=5000)
|
||||
ap.add_argument("--depth", type=int, default=30,
|
||||
help="approximate per-base read coverage")
|
||||
ap.add_argument("--read-length", type=int, default=100)
|
||||
ap.add_argument("--fragment-length", type=int, default=250)
|
||||
ap.add_argument("--damage-rate", type=float, default=0.5,
|
||||
help="0.5 = damaged FFPE-like, 0.0 = clean control")
|
||||
ap.add_argument("--background-error", type=float, default=0.001)
|
||||
ap.add_argument("--mapq", type=int, default=60)
|
||||
ap.add_argument("--baseq", type=int, default=35)
|
||||
args = ap.parse_args()
|
||||
|
||||
os.makedirs(args.outdir, exist_ok=True)
|
||||
rng = random.Random(args.seed)
|
||||
|
||||
n_chroms = args.n_chroms
|
||||
length = args.chrom_length
|
||||
chroms = [f"chr{i + 1}" for i in range(n_chroms)]
|
||||
|
||||
fa_path = make_reference(args.outdir, n_chroms, length, args.seed)
|
||||
|
||||
frags = args.depth * length // (2 * args.read_length)
|
||||
snp_pos = None
|
||||
snp_ref = None
|
||||
snp_alt = "T"
|
||||
with pysam.FastaFile(fa_path) as f:
|
||||
for offset in range(length // 2, length):
|
||||
base = f.fetch(chroms[0], offset, offset + 1).upper()
|
||||
if base == "C":
|
||||
snp_pos = offset
|
||||
snp_ref = base
|
||||
break
|
||||
if snp_ref is None:
|
||||
snp_pos = length // 2
|
||||
snp_ref = f.fetch(chroms[0], snp_pos, snp_pos + 1).upper()
|
||||
snp_alt = "A" if snp_ref != "A" else "G"
|
||||
|
||||
bam_path = args.bam_out or os.path.join(args.outdir, "sample.bam")
|
||||
build_bam(fa_path, bam_path, rng, chroms, length, frags,
|
||||
args.read_length, args.fragment_length, args.damage_rate,
|
||||
snp_pos, snp_alt, args.background_error, args.mapq, args.baseq)
|
||||
|
||||
sorted_bam = bam_path.replace(".bam", ".sorted.bam")
|
||||
pysam.sort("-o", sorted_bam, bam_path)
|
||||
pysam.index(sorted_bam)
|
||||
|
||||
print(f"reference : {fa_path}")
|
||||
print(f"bam : {sorted_bam}")
|
||||
print(f"fragments : {frags} per chromosome")
|
||||
print(f"real SNP : {chroms[0]}:{snp_pos} {snp_ref}>{snp_alt} (heterozygous)")
|
||||
print(f"damage : C>T at 5' of R1/forward (fragment 5' end), "
|
||||
f"G>A at 3' of R2/reverse (fragment far end)"
|
||||
f" (rate {args.damage_rate})")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user