Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def trace(args):
"""
%prog trace unitig{version}.{partID}.{unitigID}
Call `grep` to get the erroneous fragment placement.
"""
p = OptionParser(trace.__doc__)
opts, args = p.parse_args(args)
if len(args) != 1:
sys.exit(p.print_help())
s, = args
version, partID, unitigID = get_ID(s)
flist = glob("../5-consensus/*_{0:03d}.err".format(int(partID)))
assert len(flist) == 1
fp = open(flist[0])
instate = False
for row in fp:
if working in row and unitigID in row:
rows = []
instate = True
if instate:
rows.append(row)
if failed in row:
instate = False
if len(rows) > 20:
ignore_line = "... ({0} lines skipped)\n".format(len(rows) - 20)
rows = rows[:10] + [ignore_line] + rows[-10:]
def __init__(self, filenames=None, accessions=None, idfile=None):
self.accessions = accessions
self.idfile = idfile
if filenames is not None:
self.accessions = [op.basename(f).split(".")[0] for f in filenames]
d = dict(SeqIO.to_dict(SeqIO.parse(f, "gb")).items()[0] \
for f in filenames)
for (k, v) in d.iteritems():
self[k.split(".")[0]] = v
elif idfile is not None:
gbdir = self._get_records()
d = dict(SeqIO.to_dict(SeqIO.parse(f, "gb")).items()[0] \
for f in glob(gbdir+"/*.gb"))
for (k, v) in d.iteritems():
self[k.split(".")[0]] = v
else:
sys.exit("GenBank object is initiated from either gb files or "\
"accession IDs.")
def get_edges(weightsfiles=None):
if weightsfiles is None:
weightsfiles = glob("*.weights")
edges = {}
for row in must_open(weightsfiles):
a, b, c = row.split()
c = int(c)
edges[(a, b)] = c
edges[(b, a)] = c
return edges
help="GenBank accession IDs in a file. One ID per row, or all IDs" \
" in one row comma separated.")
p.add_option("--simple", default=None, type="string",
help="GenBank accession IDs comma separated " \
"(for lots of IDs please use --id instead).")
p.add_option("--individual", default=False, action="store_true",
help="parse gb accessions individually [default: %default]")
opts, args = p.parse_args(args)
accessions = opts.id
filenames = opts.gb_dir
if not (opts.gb_dir or opts.id or opts.simple):
sys.exit(not p.print_help())
if opts.gb_dir:
filenames = glob(opts.gb_dir+"/*.gb")
if opts.id:
rows = file(opts.id).readlines()
accessions = []
for row in rows:
accessions += map(str.strip, row.strip().split(","))
if opts.simple:
accessions = opts.simple.split(",")
if opts.id or opts.simple:
fw = must_open("GenBank_accession_IDs.txt", "w")
for atom in accessions:
print(atom, file=fw)
fw.close()
idfile = fw.name
def get_prefix(dir="../"):
"""
Look for prefix.gkpStore in the upper directory.
"""
prefix = glob(dir + "*.gkpStore")[0]
prefix = op.basename(prefix).rsplit(".", 1)[0]
return prefix
def assemble_dir(pf, target, ploidy="1"):
from jcvi.assembly.allpaths import prepare
logging.debug("Work on {0}".format(pf))
asm = [x.replace("final", pf) for x in target]
if not need_update(pf, asm):
logging.debug("Assembly found: {0}. Skipped.".format(asm))
return
cwd = os.getcwd()
os.chdir(pf)
prepare([pf] + sorted(glob("*.fastq") + glob("*.fastq.gz")) + \
["--ploidy={0}".format(ploidy)])
sh("./run.sh")
for a, t in zip(asm, target):
sh("cp allpaths/ASSEMBLIES/run/{0} ../{1}".format(t, a))
logging.debug("Assembly finished: {0}".format(asm))
os.chdir(cwd)
def set_human_axis(ax, formatter=human_formatter):
ax.xaxis.set_major_formatter(formatter)
ax.yaxis.set_major_formatter(formatter)
set_human_base_axis = partial(set_human_axis, formatter=human_base_formatter)
def set_helvetica_axis(ax):
ax.set_xticklabels([int(x) for x in ax.get_xticks()], family="Helvetica")
ax.set_yticklabels([int(x) for x in ax.get_yticks()], family="Helvetica")
available_fonts = [op.basename(x) for x in glob(datadir + "/*.ttf")]
def fontprop(ax, name, size=12):
assert name in available_fonts, "Font must be one of {0}.".format(available_fonts)
import matplotlib.font_manager as fm
fname = op.join(datadir, name)
prop = fm.FontProperties(fname=fname, size=size)
logging.debug("Set font to `{0}` (`{1}`).".format(name, prop.get_file()))
for text in ax.texts:
text.set_fontproperties(prop)
return prop
"""
Take one pair of reads and 'widow' reads after correction and run SOAP.
"""
from jcvi.assembly.soap import prepare
logging.debug("Work on {0} ({1})".format(pf, ','.join(p)))
asm = "{0}.closed.scafSeq".format(pf)
if not need_update(p, asm):
logging.debug("Assembly found: {0}. Skipped.".format(asm))
return
slink(p, pf, tag, extra)
cwd = os.getcwd()
os.chdir(pf)
prepare(sorted(glob("*.fastq") + glob("*.fastq.gz")) + \
["--assemble_1st_rank_only", "-K 31"])
sh("./run.sh")
sh("cp asm31.closed.scafSeq ../{0}".format(asm))
logging.debug("Assembly finished: {0}".format(asm))
os.chdir(cwd)