Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
global read_mappings
global reference_names
global matched_counts
global unmatched_counts
path = query_path.split("/")
query_file = path[-1]
print("Mapping query file: " + query_file)
i = 0
unmatched = 0
# barcode = default_barcode
for name, seq, qual, comment in mp.fastx_read(query_path, read_comment=True): # read one sequence
read_time = re.search(r'start_time=([^\s]+)', comment).group(1)
# if default_barcode is None:
barcode = re.search(r'barcode=([^\s]+)', comment).group(1)
try:
barcode_index = barcodes.index(barcode)
except ValueError:
generic = re.match("^barcode(\d\d)$", barcode)
if generic:
barcode_index = int(generic[1])-1
else:
raise ValueError('unknown barcode "' + barcode + '"... skipping')
time_stamp = datetime.strptime(read_time, "%Y-%m-%dT%H:%M:%SZ")
try:
count += 1
sys.exit(1)
preset = min_cnt = min_sc = k = w = bw = None
out_cs = False
for opt, arg in opts:
if opt == '-x': preset = arg
elif opt == '-n': min_cnt = int(arg)
elif opt == '-m': min_chain_score = int(arg)
elif opt == '-r': bw = int(arg)
elif opt == '-k': k = int(arg)
elif opt == '-w': w = int(arg)
elif opt == '-c': out_cs = True
a = mp.Aligner(args[0], preset=preset, min_cnt=min_cnt, min_chain_score=min_sc, k=k, w=w, bw=bw)
if not a: raise Exception("ERROR: failed to load/build index file '{}'".format(args[0]))
for name, seq, qual in mp.fastx_read(args[1]): # read one sequence
for h in a.map(seq, cs=out_cs): # traverse hits
print('{}\t{}\t{}'.format(name, len(seq), h))
print(" -w INT minimizer window length")
print(" -r INT band width")
sys.exit(1)
preset, min_cnt, min_sc, k, w, bw = None, None, None, None, None, None
for opt, arg in opts:
if opt == '-x': preset = arg
elif opt == '-n': min_cnt = int(arg)
elif opt == '-m': min_chain_score = int(arg)
elif opt == '-r': bw = int(arg)
elif opt == '-k': k = int(arg)
elif opt == '-w': w = int(arg)
a = mp.Aligner(args[0], preset=preset, min_cnt=min_cnt, min_chain_score=min_sc, k=k, w=w, bw=bw)
if not a: raise Exception("ERROR: failed to load/build index file '{}'".format(args[0]))
for name, seq, qual in mp.fastx_read(args[1]): # read one sequence
for h in a.map(seq): # traverse hits
print('{}\t{}\t{}'.format(name, len(seq), h))
def create_index(reference_file):
aligner = mp.Aligner(reference_file, best_n = 1)
for name, seq, qual in mp.fastx_read(reference_file, read_comment=False):
reference_names.append(name)
reference_lengths[name] = len(seq)
if not aligner:
raise Exception("ERROR: failed to load/build index file '{}'".format(reference_file))
return aligner
def get_fasta_names(path):
fasta_names = []
for name, seq, qual in mp.fastx_read(path, read_comment=False):
fasta_names.append(name)
return fasta_names
def _method_mappy(self, *args, **kwargs):
with open(self.outfile, "w") as fasta:
for (name, seq, _) in fastx_read(self.infile):
fasta.write(">{}\n{}\n".format(name, seq))
def create_coordinate_index(coordinate_reference):
"""Initialise minimap2 for 2 different reference FASTAs"""
## Parse the (single) reference sequence used to find co-ordinates:
coordinate_aligner = mp.Aligner(coordinate_reference, best_n=1)
name, seq, qual = next(mp.fastx_read(coordinate_reference, read_comment=False))
coordinate_reference_length = len(seq)
if not coordinate_aligner:
raise Exception("ERROR: failed to load/build index file '{}'".format(coordinate_reference))
return (coordinate_aligner, coordinate_reference_length)