Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def __repr__(self):
return self._SAMheader_raw.decode().rstrip() if self._SAMheader_raw else str(self.refs)
def __str__(self):
""" Used for printing the header
Note:
Preferentially prints out the SAM header (if present). Otherwise, it will print
the string representation of the BAM header dictionary
"""
return self._SAMheader_raw.decode().rstrip() if self._SAMheader_raw else str(self.refs)
class BamReader(bgzf.BgzfReader):
""" The BAM reader. Heavily modified from Peter Cock's BgzfReader.
Attributes:
header: representation of header data (if present)
lengths (:py:obj:`list` of :py:obj:`int`): lengths of references listed in header
nocoordinate (int): number of reads that have no coordinates
nreferences (int): number of references in header
ref2tid (:py:obj:`dict` of :py:obj:`str`, :py:obj:`int`): refernce names and refID dictionary
references (:py:obj:`list` of :py:obj:`str`): names of references listed in header
text (str): SAM header (if present)
unmapped (int): number of unmapped reads
Note:
This implementation is likely to change. While the API was meant to
mirror `pysam`, it makes sense to include the `pysam`-like API in an extension
that will wrap the core reader. This would be a major refactor, and therefore
pass
if referencenames is not None or referencelengths is not None:
assert len(referencenames) == len(referencelengths)
references = zip(referencenames, referencelengths)
alt_refs = _sam_header_to_ref_list(header)
if not alt_refs:
# Append minimal @SQ lines to the header
header += _ref_list_to_sam_header(references)
elif alt_refs != references:
raise ValueError("Reference names and lengths inconsistent with header @SQ lines")
else:
references = _sam_header_to_ref_list(header)
return header, references
class BamWriter(bgzf.BgzfWriter):
def __init__(self, filepath_or_object, mode = 'xb', compresslevel = 6, ignore_overwrite = False,
copy_header = None, header = b'', reference_names = None, reference_lengths = None):
"""Useful class for writing BAM files.
Note:
As of right now, it is tuned for working within a pipeline where reads
are being filtered or modified from another BAM file.
Args:
filepath_or_object (str | :py:obj:`file`): the path or file object of the BAM file
mode (str): Mode for writing. BAM files are binary by nature (default: 'xb')
compresslevel (int): desired level of Gzip compression (default: 6)
ignore_overwrite (bool): whether or not to ignore overwrite detection
copy_header (filepath_or_object): copy the header from another BAM file
header (bytes): SAM-style header
Every BAM file should contain an EOF signature within the last
28 bytes of the file. This function checks for that signature.
Returns:
(bool): True if truncated, else False
Warns:
BytesWarning: if no EOF signature found.
"""
temp_pos = self._handle.tell()
self._handle.seek(-28, 2)
eof = self._handle.read()
self._handle.seek(temp_pos)
if eof == bgzf._bgzf_eof:
return False
else:
warnings.warn('No EOF character found. File may be truncated', BytesWarning)
return True
def to_bam(bam_file):
"""Writes the alignment record to a BAM file
Args:
bam_file (string or :py:obj:`bamnostic.bam.BamWriter`): BAM file path or open bam file in a write mode
"""
if type(bam_file) == str:
# Natively go into append mode
if os.path.isfile(bam_file):
with bam.BamWriter(bam_file, mode = 'ab') as bam_out:
bam_out.write(self._raw_stream)
else:
raise IOError('BAM file must already exist, or be initilaized through bamnostic.bam.BamWriter')
elif isinstance(bam_file, bgzf.BgzfWriter):
bam_file.write(self._raw_stream)
else:
raise IOError('BAM file must already exist, or be initilaized through bamnostic.bam.BamWriter')