def test_merge_and_switch():
test_bam1 = bam.BamFile(
os.path.join(dir, "chr19_window.bam"), "samtools", no_initial_index=True)
test_bam2 = bam.BamFile(
os.path.join(dir, "chrX_window1.bam"), "samtools", no_initial_index=True)
test_bam3 = bam.BamFile(
os.path.join(dir, "chrX_window2.bam"), "samtools", no_initial_index=True)
merged = bam.samtools_merge(
"samtools", [test_bam1.filepath, test_bam2.filepath],
os.path.join(dir, "merged1"), 1)
merged = bam.BamFile(
os.path.join(dir, "merged1.merged.bam"), "samtools", no_initial_index=True)
a = pysam.idxstats(test_bam1.filepath)
test1_reads = sum([
int(k[2]) + int(k[3]) for k in [
x.split("\t") for x in a.split("\n")] if len(k) > 3])
a = pysam.idxstats(test_bam2.filepath)
test2_reads = sum([
int(k[2]) + int(k[3]) for k in [
x.split("\t") for x in a.split("\n")] if len(k) > 3])
a = pysam.idxstats(test_bam3.filepath)
test3_reads = sum([
int(k[2]) + int(k[3]) for k in [
x.split("\t") for x in a.split("\n")] if len(k) > 3])
a = pysam.idxstats(merged.filepath)
merged1_reads = sum([
int(k[2]) + int(k[3]) for k in [
x.split("\t") for x in a.split("\n")] if len(k) > 3])
assert merged1_reads == test1_reads + test2_reads
swapped = bam.switch_sex_chromosomes_sambamba(
"samtools", "sambamba", merged.filepath, test_bam3.filepath,
"chrX", dir, "swapped", 1, {"CL": ["foo"], "ID": "xyalign"})
swapped = bam.BamFile(
os.path.join(dir, "swapped.merged.bam"), "samtools", no_initial_index=True)
a = pysam.idxstats(swapped.filepath)
swapped_reads = sum([
int(k[2]) + int(k[3]) for k in [
x.split("\t") for x in a.split("\n")] if len(k) > 3])
assert swapped_reads == test1_reads + test3_reads
header = read_bed(os.path.join(dir, "swapped.header.sam"))
assert ["@PG", "ID:xyalign", "CL:foo"] in header
评论列表
文章目录
