Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
all_files = []
for wc, filename in inputmap:
try:
wc = eval(wc)
except:
pass
wc = update_wildcard_constraints(wc, wildcard_constraints, {})
all_wc.append(wc)
if filename is None:
continue
if isinstance(filename, str):
filename = [filename]
all_files = all_files + filename
for f in all_files:
for wc in all_wc:
wildcards = glob_wildcards(wc, [os.path.basename(f)])
for k, v in wildcards._asdict().items():
if len(v) > 0:
d[k] = v[0]
except:
logger.debug("Failed to get wildcards for inputmap ", inputmap)
raise
return d
def get_batch_ids_raw(runname, config, tag=None, checkpoints=None):
tag_barcode = get_tag_barcode(tag, runname, config) if tag else None
if tag_barcode and checkpoints:
if hasattr(checkpoints, config['demux_default'] + '_barcode'):
barcode_batch_dir = getattr(checkpoints, config['demux_default'] + '_barcode').get(runname=runname).output.barcodes
else:
raise NotImplementedError("Demultiplexing with {} is not implemented.".format(config['demux_default']))
barcode_batch = os.path.join(barcode_batch_dir, tag_barcode, '{id}.txt')
batches_txt, = glob_wildcards(barcode_batch)
return batches_txt
else:
batches_tar, = glob_wildcards("{datadir}/{runname}/reads/{{id}}.tar".format(datadir=config["storage_data_raw"], runname=runname))
batches_fast5, = glob_wildcards("{datadir}/{runname}/reads/{{id}}.fast5".format(datadir=config["storage_data_raw"], runname=runname))
return batches_tar + batches_fast5
def rename_genomes(input_folder,mapfile_genomes,mapfile_contigs,output_dir):
file_name = f"{input_folder}/{{binid}}.fasta"
bin_ids, = glob_wildcards(file_name)
old2new_name= dict(zip(bin_ids,utils.gen_names_for_range(len(bin_ids),prefix='MAG')))
os.makedirs(output_dir)
with open(mapfile_contigs,'w') as out_contigs, open(mapfile_genomes,'w') as old2new_mapping_file :
old2new_mapping_file.write(f"BinID\tMAG\n")
for binid in bin_ids:
fasta_in = file_name.format(binid=binid)
new_name= old2new_name[binid]
old2new_mapping_file.write(f"{binid}\t{new_name}\n")
fasta_out = os.path.join(output_dir,f"{new_name}.fasta")
def predict_genes_genomes(in_dir,out_dir,log):
path= os.path.join(in_dir,'{genome}.fasta')
os.makedirs(out_dir,exist_ok=True)
for genome in glob_wildcards(path).genome:
predict_genes(genome,path.format(genome=genome),out_dir,log)
def get_batch_ids_raw(runname, config, tag=None, checkpoints=None):
tag_barcode = get_tag_barcode(tag, runname, config) if tag else None
if tag_barcode and checkpoints:
if hasattr(checkpoints, config['demux_default'] + '_barcode'):
barcode_batch_dir = getattr(checkpoints, config['demux_default'] + '_barcode').get(runname=runname).output.barcodes
else:
raise NotImplementedError("Demultiplexing with {} is not implemented.".format(config['demux_default']))
barcode_batch = os.path.join(barcode_batch_dir, tag_barcode, '{id}.txt')
batches_txt, = glob_wildcards(barcode_batch)
return batches_txt
else:
batches_tar, = glob_wildcards("{datadir}/{runname}/reads/{{id}}.tar".format(datadir=config["storage_data_raw"], runname=runname))
batches_fast5, = glob_wildcards("{datadir}/{runname}/reads/{{id}}.fast5".format(datadir=config["storage_data_raw"], runname=runname))
return batches_tar + batches_fast5