Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def buildRead( self ):
'''build an example read.'''
a = pysam.AlignedRead()
a.qname = "read_12345"
a.seq="ACGT" * 10
a.flag = 0
a.rname = 0
a.pos = 20
a.mapq = 20
a.cigar = ( (0,10), (2,1), (0,9), (1,1), (0,20) )
a.mrnm = 0
a.mpos=200
a.isize=167
a.qual="1234" * 10
# todo: create tags
return a
def testEmpty( self ):
a = pysam.AlignedRead()
self.assertEqual( a.qname, None )
self.assertEqual( a.seq, None )
self.assertEqual( a.qual, None )
self.assertEqual( a.flag, 0 )
self.assertEqual( a.rname, 0 )
self.assertEqual( a.mapq, 0 )
self.assertEqual( a.cigar, None )
self.assertEqual( a.tags, [] )
self.assertEqual( a.mrnm, 0 )
self.assertEqual( a.mpos, 0 )
self.assertEqual( a.isize, 0 )
with open(args.contigs_fastq, 'r') as f, open(out_fastq_name, 'w') as out_fq:
fq_iter = tk_fasta.read_generator_fastq(f)
for (name, seq, quals) in fq_iter:
if name in sel_contigs:
if name == best_contig:
best_contig_seq = seq
best_contig_quals = quals
header = cr_fastq.AugmentedFastqHeader(name)
# Create a pseudo-UMI for each input contig
header.set_tag(PROCESSED_UMI_TAG, name)
# Put all reads on the same "barcode". This is important, so
# the assembler assembles all of them together.
header.set_tag(PROCESSED_BARCODE_TAG, clonotype_name)
record = pysam.AlignedRead()
record.reference_start = 0
record.reference_id = 0
# Wrap with str() or pysam will crash when given unicode
record.qname = str(header.to_string())
record.seq = seq
record.qual = quals
record.flag = MAPPED_UNPAIRED_FLAG
out_bam.write(record)
# Now change the tags. The final bam concatenation code will pull
# the tags out of the header, so we want these to be meaningful.
# Put the real barcode in the barcode tag. The alignment-base-qual
# code will ignore it anyway.
header.set_tag(PROCESSED_BARCODE_TAG, name.split('_')[0])
a = pysam.AlignedRead()
a.qname = "read_28833_29006_6945"
a.seq="AGCTTAGCTAGCTACCTATATCTTGGTCTTGGCCG"
a.flag = 99
a.rname = 0
a.pos = 32
a.mapq = 20
a.cigar = ( (0,10), (2,1), (0,25) )
a.mrnm = 0
a.mpos=199
a.isize=167
a.qual="<<<<<<<<<<<<<<<<<<<<<:<9/,&,22;;<<<"
a.tags = ( ("NM", 1),
("RG", "L1") )
"""
cAR = pysam.AlignedRead()
aR = chainedAlignedReads[0]
cAR.qname = aR.qname
#Parameters we don't and therefore set properly
#cAR.flag = aR.flag
#cAR.mapq = aR.mapq
#cAR.mrnm = 0
#cAR.mpos=0
#cAR.isize=0
#cAR.qual = "<" * len(readSequence)
#cAR.tags = aR.tags
cAR.rnext = -1
cAR.pos = 0
cAR.is_reverse = aR.is_reverse
if cAR.is_reverse:
cAR.seq = reverseComplement(readSequence)
def export_bam(outbam, read1, read2, quiet=False):
if read2:
def gen():
for r1, r2 in itertools.izip(read1.fetch(quiet=quiet), read2.fetch(quiet=True)):
yield (r1, r2)
else:
def gen():
for r1 in read1.fetch(quiet=quiet):
yield (r1, None)
for r1, r2 in gen():
record1 = pysam.AlignedRead()
record1.qname = r1.name
record1.seq = r1.seq
record1.qual = r1.qual
if r2:
record1.is_paired = True
record1.is_read1 = True
record2 = pysam.AlignedRead()
record2.qname = r1.name
record2.seq = r1.seq
record2.qual = r1.qual
record2.is_paired = True
record2.is_read2 = True
outbam.write(record1)
status_failed = 0
for k, v in map_statistics_pair.items():
if k.startswith('success'):
status_success += v
elif k.startswith('fail'):
status_failed += v
LOG.info("Processed {0:,} reads, {1:,} successful, {2:,} failed".format(total, status_success, status_failed))
if compat.is_py2:
alignment = sam_file.next()
else:
alignment = next(sam_file)
alignment_new = pysam.AlignedRead()
read_chr = sam_file.getrname(alignment.tid)
# READ ONLY
# aend aligned reference position of the read on the reference genome
# alen aligned length of the read on the reference genome.
# positions a list of reference positions that this read aligns to
# qend end index of the aligned query portion of the sequence (0-based, exclusive)
# qlen Length of the aligned query sequence
# qqual aligned query sequence quality values
# qstart start index of the aligned query portion of the sequence (0-based, inclusive)
# query aligned portion of the read and excludes any flanking bases that were soft clipped
# rlen length of the read
# TRUE / FALSE (setting effects flag)
def write_read(bam_file, name, seq, qual, tid, pos, mapq=0, cigar=None, reverse=False, paired=False, read1=True,
mate_tid=None, mate_pos=None, mate_reverse=False):
""" Creates an alignedRead object and writes it to the corresponding bam file.
If cigar is unspecified, the alignment is defaulted to a perfect alignment.
Chrom
"""
r = pysam.AlignedRead()
r.qname = name
r.seq = seq
r.qual = qual
r.tid = tid
r.pos = pos
r.mapq = mapq
if pos == -1:
r.is_unmapped=True
if cigar:
r.cigar = cigar
else:
r.cigar = [(0, len(seq))]
r.is_reverse = reverse
def writeBAMentry(chrom,pos,seq,length,strand):
global BAMoutfile,rcounter
global options
chromTID = BAMoutfile.gettid(chrom)
if chromTID == -1: return
qual = "I"*len(seq)
if length > options.readlength:
forward = pysam.AlignedRead()
forward.qname = "SIM-%s-%09d"%(chrom,rcounter)
forward.is_paired = True
forward.is_proper_pair = True
forward.is_read1 = True
forward.tid = chromTID
forward.rnext = chromTID
reverse = pysam.AlignedRead()
reverse.qname = "SIM-%s-%09d"%(chrom,rcounter)
reverse.is_paired = True
reverse.is_proper_pair = True
reverse.is_read2 = True
reverse.tid = chromTID
reverse.rnext = chromTID
if strand: