#!/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 /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()