Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_hisat2(genome, force):
"""Create hisat2 index."""
assert os.path.exists(genome.filename)
force = True if force == "overwrite" else False
if cmd_ok("hisat2-build"):
p = Hisat2Plugin()
p.after_genome_download(genome)
dirname = os.path.dirname(genome.filename)
index_dir = os.path.join(dirname, "index", "hisat2")
fname = os.path.join(index_dir, "{}.1.ht2".format(genome.name))
assert os.path.exists(index_dir)
assert os.path.exists(fname)
force_test(p, fname, genome, force)
def test_gmap(genome, force):
"""Create gmap index."""
assert os.path.exists(genome.filename)
force = True if force == "overwrite" else False
if cmd_ok("gmap"):
p = GmapPlugin()
p.after_genome_download(genome, force=force)
dirname = os.path.dirname(genome.filename)
index_dir = os.path.join(dirname, "index", "gmap")
fname = os.path.join(index_dir, "{}.maps".format(genome.name))
assert os.path.exists(index_dir)
assert os.path.exists(fname)
force_test(p, fname, genome, force)
def test_bowtie2(genome, force):
"""Create bbowtie2 index."""
assert os.path.exists(genome.filename)
force = True if force == "overwrite" else False
if cmd_ok("bowtie2"):
p = Bowtie2Plugin()
p.after_genome_download(genome, force=force)
dirname = os.path.dirname(genome.filename)
index_dir = os.path.join(dirname, "index", "bowtie2")
fname = os.path.join(index_dir, "{}.1.bt2".format(genome.name))
assert os.path.exists(index_dir)
assert os.path.exists(fname)
force_test(p, fname, genome, force)
def test_bwa(genome, force):
"""Create bwa index."""
assert os.path.exists(genome.filename)
force = True if force == "overwrite" else False
if cmd_ok("bwa"):
p = BwaPlugin()
p.after_genome_download(genome, force=force)
dirname = os.path.dirname(genome.filename)
index_dir = os.path.join(dirname, "index", "bwa")
fname = os.path.join(index_dir, "{}.fa.sa".format(genome.name))
assert os.path.exists(index_dir)
assert os.path.exists(fname)
force_test(p, fname, genome, force)
def test_minimap2(genome, force):
"""Create minimap2 index."""
assert os.path.exists(genome.filename)
force = True if force == "overwrite" else False
if cmd_ok("minimap2"):
p = Minimap2Plugin()
p.after_genome_download(genome, force=force)
dirname = os.path.dirname(genome.filename)
index_dir = os.path.join(dirname, "index", "minimap2")
fname = os.path.join(index_dir, "{}.mmi".format(genome.name))
assert os.path.exists(index_dir)
assert os.path.exists(fname)
force_test(p, fname, genome, force)
def after_genome_download(self, genome, threads=1, force=False):
if not cmd_ok("bowtie2-build"):
return
# Create index dir
index_dir = genome.props["bowtie2"]["index_dir"]
index_name = genome.props["bowtie2"]["index_name"]
if force:
# Start from scratch
rmtree(index_dir, ignore_errors=True)
mkdir_p(index_dir)
if not any(fname.endswith(".bt2") for fname in os.listdir(index_dir)):
# Create index
cmd = "bowtie2-build --threads {} {} {}".format(
threads, genome.filename, index_name
)
run_index_cmd("bowtie2", cmd)
def after_genome_download(self, genome, threads=1, force=False):
if not cmd_ok("hisat2-build"):
return
# Create index dir
index_dir = genome.props["hisat2"]["index_dir"]
index_name = genome.props["hisat2"]["index_name"]
if force:
# Start from scratch
rmtree(index_dir, ignore_errors=True)
mkdir_p(index_dir)
if not any(fname.endswith(".ht2") for fname in os.listdir(index_dir)):
# unzip genome if zipped and return up-to-date genome name
bgzip, fname = bgunzip_and_name(genome)
# Create index
cmd = "hisat2-build -p {} {} {}".format(threads, fname, index_name)
def after_genome_download(self, genome, threads=1, force=False):
if not cmd_ok("bwa"):
return
# Create index dir
index_dir = genome.props["bwa"]["index_dir"]
index_name = genome.props["bwa"]["index_name"]
if force:
# Start from scratch
rmtree(index_dir, ignore_errors=True)
mkdir_p(index_dir)
if not any(fname.endswith(".bwt") for fname in os.listdir(index_dir)):
# Create index
if not os.path.exists(index_name):
os.symlink(genome.filename, index_name)
cmd = "bwa index {}".format(index_name)
def after_genome_download(self, genome, threads=1, force=False):
if not cmd_ok("STAR"):
return
# Create index dir
index_dir = genome.props["star"]["index_dir"]
index_name = genome.props["star"]["index_name"]
if force:
# Start from scratch
rmtree(index_dir, ignore_errors=True)
mkdir_p(index_dir)
if not os.path.exists(index_name):
# unzip genome if zipped and return up-to-date genome name
bgzip, fname = bgunzip_and_name(genome)
# Create index
cmd = "STAR --runMode genomeGenerate --runThreadN {} --genomeFastaFiles {} --genomeDir {} --outFileNamePrefix {}".format(
def after_genome_download(self, genome, threads=1, force=False):
if not cmd_ok("gmap_build"):
return
# Create index dir
index_dir = genome.props["gmap"]["index_dir"]
if force:
# Start from scratch
rmtree(index_dir, ignore_errors=True)
if not os.path.exists(index_dir):
# unzip genome if zipped and return up-to-date genome name
bgzip, fname = bgunzip_and_name(genome)
# gmap outputs a folder named genome.name
# its content is moved to index dir, consistent with other plugins
with TemporaryDirectory() as tmpdir:
# Create index