#!/usr/bin/env python
from ConfigParser import SafeConfigParser
+import fnmatch
from glob import glob
import json
import logging
-from optparse import OptionParser
+import netrc
+from optparse import OptionParser, OptionGroup
import os
from pprint import pprint, pformat
import shlex
from StringIO import StringIO
-import time
+import stat
import sys
+import time
import types
import urllib
import urllib2
import urlparse
+import RDF
from htsworkflow.util import api
-from htsworkflow.pipelines.sequences import \
- create_sequence_table, \
- scan_for_sequences
-
+from htsworkflow.util.rdfhelp import \
+ dafTermOntology, \
+ fromTypedNode, \
+ get_model, \
+ get_serializer, \
+ load_into_model, \
+ sparql_query, \
+ submissionOntology
+from htsworkflow.submission.daf import \
+ DAFMapper, \
+ MetadataLookupException, \
+ get_submission_uri
+from htsworkflow.submission.condorfastq import CondorFastqExtract
+
+logger = logging.getLogger('ucsc_gather')
def main(cmdline=None):
parser = make_parser()
elif opts.verbose:
logging.basicConfig(level = logging.INFO )
else:
- logging.basicConfig(level = logging.WARNING )
-
+ logging.basicConfig(level = logging.WARNING )
- apidata = {'apiid': opts.apiid, 'apikey': opts.apikey }
+ apidata = api.make_auth_from_opts(opts, parser)
- if opts.host is None or opts.apiid is None or opts.apikey is None:
- parser.error("Please specify host url, apiid, apikey")
+ model = get_model(opts.load_model)
+ if opts.name:
+ mapper = DAFMapper(opts.name, opts.daf, model)
+ submission_uri = get_submission_uri(opts.name)
- if len(args) == 0:
- parser.error("I need at least one library submission-dir input file")
+ if opts.library_url is not None:
+ mapper.library_url = opts.library_url
+ if opts.load_rdf is not None:
+ load_into_model(model, 'turtle', opts.load_rdf, submission_uri)
+
+ if opts.make_ddf and opts.daf is None:
+ parser.error("Please specify your daf when making ddf files")
+
library_result_map = []
for a in args:
library_result_map.extend(read_library_result_map(a))
- if opts.daf is not None:
+ if opts.make_tree_from is not None:
+ make_tree_from(opts.make_tree_from, library_result_map)
+
+ if opts.link_daf:
link_daf(opts.daf, library_result_map)
if opts.fastq:
- build_fastqs(opts.host,
- apidata,
- opts.sequence,
- library_result_map,
- force=opts.force)
+ extractor = CondorFastqExtract(opts.host, apidata, opts.sequence,
+ force=opts.force)
+ extractor.build_fastqs(library_result_map)
- if opts.ini:
- make_submission_ini(opts.host, apidata, library_result_map)
+ if opts.scan_submission:
+ scan_submission_dirs(mapper, library_result_map)
- if opts.makeddf:
- make_all_ddfs(library_result_map, opts.daf)
+ if opts.make_ddf:
+ make_all_ddfs(mapper, library_result_map, opts.daf, force=opts.force)
-def make_parser():
- # Load defaults from the config files
- config = SafeConfigParser()
- config.read([os.path.expanduser('~/.htsworkflow.ini'), '/etc/htsworkflow.ini'])
-
- sequence_archive = None
- apiid = None
- apikey = None
- apihost = None
- SECTION = 'sequence_archive'
- if config.has_section(SECTION):
- sequence_archive = config.get(SECTION, 'sequence_archive',sequence_archive)
- sequence_archive = os.path.expanduser(sequence_archive)
- apiid = config.get(SECTION, 'apiid', apiid)
- apikey = config.get(SECTION, 'apikey', apikey)
- apihost = config.get(SECTION, 'host', apihost)
+ if opts.sparql:
+ sparql_query(model, opts.sparql)
+
+ if opts.print_rdf:
+ writer = get_serializer()
+ print writer.serialize_model_to_string(model)
+
+def make_parser():
parser = OptionParser()
+ model = OptionGroup(parser, 'model')
+ model.add_option('--name', help="Set submission name")
+ model.add_option('--load-model', default=None,
+ help="Load model database")
+ model.add_option('--load-rdf', default=None,
+ help="load rdf statements into model")
+ model.add_option('--sparql', default=None, help="execute sparql query")
+ model.add_option('--print-rdf', action="store_true", default=False,
+ help="print ending model state")
+ parser.add_option_group(model)
# commands
- parser.add_option('--fastq', help="generate scripts for making fastq files",
- default=False, action="store_true")
-
- parser.add_option('--ini', help="generate submission ini file", default=False,
- action="store_true")
-
- parser.add_option('--makeddf', help='make the ddfs', default=False,
+ commands = OptionGroup(parser, 'commands')
+ commands.add_option('--make-tree-from',
+ help="create directories & link data files",
+ default=None)
+ commands.add_option('--fastq', default=False, action="store_true",
+ help="generate scripts for making fastq files")
+ commands.add_option('--scan-submission', default=False, action="store_true",
+ help="Import metadata for submission into our model")
+ commands.add_option('--link-daf', default=False, action="store_true",
+ help="link daf into submission directories")
+ commands.add_option('--make-ddf', help='make the ddfs', default=False,
action="store_true")
+ parser.add_option_group(commands)
- parser.add_option('--daf', default=None, help='specify daf name')
parser.add_option('--force', default=False, action="store_true",
help="Force regenerating fastqs")
-
- # configuration options
- parser.add_option('--apiid', default=apiid, help="Specify API ID")
- parser.add_option('--apikey', default=apikey, help="Specify API KEY")
- parser.add_option('--host', default=apihost,
- help="specify HTSWorkflow host",)
- parser.add_option('--sequence', default=sequence_archive,
- help="sequence repository")
-
+ parser.add_option('--daf', default=None, help='specify daf name')
+ parser.add_option('--library-url', default=None,
+ help="specify an alternate source for library information")
# debugging
parser.add_option('--verbose', default=False, action="store_true",
help='verbose logging')
parser.add_option('--debug', default=False, action="store_true",
help='debug logging')
+ api.add_auth_options(parser)
+
return parser
-def build_fastqs(host, apidata, sequences_path, library_result_map,
- force=False ):
+def make_tree_from(source_path, library_result_map):
+ """Create a tree using data files from source path.
"""
- Generate condor scripts to build any needed fastq files
+ for lib_id, lib_path in library_result_map:
+ if not os.path.exists(lib_path):
+ logging.info("Making dir {0}".format(lib_path))
+ os.mkdir(lib_path)
+ source_lib_dir = os.path.abspath(os.path.join(source_path, lib_path))
+ if os.path.exists(source_lib_dir):
+ pass
+ for filename in os.listdir(source_lib_dir):
+ source_pathname = os.path.join(source_lib_dir, filename)
+ target_pathname = os.path.join(lib_path, filename)
+ if not os.path.exists(source_pathname):
+ raise IOError("{0} does not exist".format(source_pathname))
+ if not os.path.exists(target_pathname):
+ os.symlink(source_pathname, target_pathname)
+ logging.info(
+ 'LINK {0} to {1}'.format(source_pathname, target_pathname))
- Args:
- host (str): root of the htsworkflow api server
- apidata (dict): id & key to post to the server
- sequences_path (str): root of the directory tree to scan for files
- library_result_map (list): [(library_id, destination directory), ...]
- """
- qseq_condor_header = """
-Universe=vanilla
-executable=/woldlab/rattus/lvol0/mus/home/diane/proj/solexa/gaworkflow/scripts/qseq2fastq
-error=log/qseq2fastq.err.$(process).log
-output=log/qseq2fastq.out.$(process).log
-log=log/qseq2fastq.log
-
-"""
- qseq_condor_entries = []
- srf_condor_header = """
-Universe=vanilla
-executable=/woldlab/rattus/lvol0/mus/home/diane/proj/solexa/gaworkflow/scripts/srf2named_fastq.py
-output=log/srf_pair_fastq.out.$(process).log
-error=log/srf_pair_fastq.err.$(process).log
-log=log/srf_pair_fastq.log
-environment="PYTHONPATH=/home/diane/lib/python2.6/site-packages:/home/diane/proj/solexa/gaworkflow PATH=/woldlab/rattus/lvol0/mus/home/diane/bin:/usr/bin:/bin"
-
-"""
- srf_condor_entries = []
- fastq_paired_template = '%(lib_id)s_%(flowcell)s_c%(cycle)s_l%(lane)s_r%(read)s.fastq'
- fastq_single_template = '%(lib_id)s_%(flowcell)s_c%(cycle)s_l%(lane)s.fastq'
- lib_db = find_archive_sequence_files(host,
- apidata,
- sequences_path,
- library_result_map)
-
- # find what targets we're missing
- needed_targets = {}
- for lib_id, result_dir in library_result_map:
- lib = lib_db[lib_id]
- lane_dict = make_lane_dict(lib_db, lib_id)
-
- for lane_key, sequences in lib['lanes'].items():
- for seq in sequences:
- paired = lane_dict[seq.flowcell]['paired_end']
- if paired and seq.read is None:
- seq.read = 1
- filename_attributes = {
- 'flowcell': seq.flowcell,
- 'lib_id': lib_id,
- 'lane': seq.lane,
- 'read': seq.read,
- 'cycle': seq.cycle
- }
- # throw out test runs
- # FIXME: this should probably be configurable
- if seq.cycle < 50:
- continue
- if seq.flowcell == '30CUUAAXX':
- # 30CUUAAXX run sucked
- continue
- if seq.flowcell == '30DY0AAXX':
- # 30DY0 only ran for 151 bases instead of 152
- # it is actually 76 1st read, 75 2nd read
- seq.mid_point = 76
-
- # end filters
- if paired:
- target_name = fastq_paired_template % filename_attributes
- else:
- target_name = fastq_single_template % filename_attributes
-
- target_pathname = os.path.join(result_dir, target_name)
- if force or not os.path.exists(target_pathname):
- t = needed_targets.setdefault(target_pathname, {})
- t[seq.filetype] = seq
-
- for target_pathname, available_sources in needed_targets.items():
- logging.debug(' target : %s' % (target_pathname,))
- logging.debug(' candidate sources: %s' % (available_sources,))
- if available_sources.has_key('qseq'):
- source = available_sources['qseq']
- qseq_condor_entries.append(
- condor_qseq_to_fastq(source.path,
- target_pathname,
- source.flowcell,
- force=force)
- )
- elif available_sources.has_key('srf'):
- source = available_sources['srf']
- mid = getattr(source, 'mid_point', None)
- srf_condor_entries.append(
- condor_srf_to_fastq(source.path,
- target_pathname,
- paired,
- source.flowcell,
- mid,
- force=force)
- )
- else:
- print " need file", target_pathname
-
- if len(srf_condor_entries) > 0:
- make_submit_script('srf.fastq.condor',
- srf_condor_header,
- srf_condor_entries)
-
- if len(qseq_condor_entries) > 0:
- make_submit_script('qseq.fastq.condor',
- qseq_condor_header,
- qseq_condor_entries)
def link_daf(daf_path, library_result_map):
if not os.path.exists(daf_path):
base_daf = os.path.basename(daf_path)
for lib_id, result_dir in library_result_map:
+ if not os.path.exists(result_dir):
+ raise RuntimeError("Couldn't find target directory %s" %(result_dir,))
submission_daf = os.path.join(result_dir, base_daf)
if not os.path.exists(submission_daf):
+ if not os.path.exists(daf_path):
+ raise RuntimeError("Couldn't find daf: %s" %(daf_path,))
os.link(daf_path, submission_daf)
-def make_submission_ini(host, apidata, library_result_map, paired=True):
- # ma is "map algorithm"
- ma = 'TH1014'
- if paired:
- aligns = "Paired"
- else:
- aligns = "Aligns"
-
- attributes = {
- # bam index
- '.bai': {'view': None, 'MapAlgorithm': 'NA'},
- '.bam': {'view': aligns, 'MapAlgorithm': ma},
- '.splices.bam': {'view': 'Splices', 'MapAlgorithm': ma},
- '.jnct': {'view': 'Junctions', 'MapAlgorithm': ma},
- '.plus.bigwig': {'view': 'PlusSignal', 'MapAlgorithm': ma},
- '.minus.bigwig': {'view': 'MinusSignal', 'MapAlgorithm': ma},
- '.bigwig': {'view': 'Signal', 'MapAlgorithm': ma},
- '.tar.bz2': {'view': None},
- '.condor': {'view': None},
- '.daf': {'view': None},
- '.ddf': {'view': None},
- 'denovo.genes.expr': {'view': 'GeneDeNovo', 'MapAlgorithm': ma},
- 'denovo.transcripts.expr': {'view': 'TranscriptDeNovo', 'MapAlgorithm': ma},
- 'novel.genes.expr': {'view': 'GeneDeNovo', 'MapAlgorithm': ma},
- 'novel.transcripts.expr': {'view': 'TranscriptDeNovo', 'MapAlgorithm': ma},
- '.genes.expr': {'view': 'GeneFPKM', 'MapAlgorithm': ma},
- '.transcripts.expr': {'view': 'TranscriptFPKM', 'MapAlgorithm': ma},
- '.transcript.expr': {'view': 'TranscriptFPKM', 'MapAlgorithm': ma},
- '.fastq': {'view': 'Fastq', 'MapAlgorithm': 'NA' },
- '_r1.fastq': {'view': 'FastqRd1', 'MapAlgorithm': 'NA'},
- '_r2.fastq': {'view': 'FastqRd2', 'MapAlgorithm': 'NA'},
- '.gtf': {'view': 'GeneModel', 'MapAlgorithm': ma},
- '.ini': {'view': None},
- '.log': {'view': None},
- '.stats.txt': {'view': 'InsLength', 'MapAlgorithm': ma},
- '.srf': {'view': None},
- '.wig': {'view': None},
- '.zip': {'view': None},
- }
-
- candidate_fastq_src = {}
-
- for lib_id, result_dir in library_result_map:
- order_by = ['order_by=files', 'view', 'replicate', 'cell',
- 'readType', 'mapAlgorithm', 'insertLength' ]
- inifile = ['[config]']
- inifile += [" ".join(order_by)]
- inifile += ['']
- line_counter = 1
- lib_info = get_library_info(host, apidata, lib_id)
- result_ini = os.path.join(result_dir, result_dir+'.ini')
-
- if lib_info['cell_line'].lower() == 'unknown':
- logging.warn("Library %s missing cell_line" % (lib_id,))
-
- standard_attributes = {'cell': lib_info['cell_line'],
- 'replicate': lib_info['replicate'],
- }
- if paired:
- if lib_info['insert_size'] is None:
- errmsg = "Library %s is missing insert_size, assuming 200"
- logging.warn(errmsg % (lib_id,))
- insert_size = 200
- else:
- insert_size = lib_info['insert_size']
- standard_attributes['insertLength'] = insert_size
- standard_attributes['readType'] = '2x75'
- else:
- standard_attributes['insertLength'] = 'ilNA'
- standard_attributes['readType'] = '1x75D'
-
- # write other lines
- submission_files = os.listdir(result_dir)
- fastqs = {}
- for f in submission_files:
- best_ext = find_best_extension(attributes, f)
-
- if best_ext is not None:
- if attributes[best_ext]['view'] is None:
-
- continue
- elif best_ext.endswith('fastq'):
- fastqs.setdefault(best_ext, set()).add(f)
- else:
- inifile.extend(
- make_submission_section(line_counter,
- [f],
- standard_attributes,
- attributes[best_ext]
- )
- )
- inifile += ['']
- line_counter += 1
- else:
- raise ValueError("Unrecognized file: %s" % (f,))
-
- # add in fastqs on a single line.
- for extension, fastq_set in fastqs.items():
- inifile.extend(
- make_submission_section(line_counter,
- fastq_set,
- standard_attributes,
- attributes[extension])
- )
- inifile += ['']
- line_counter += 1
-
- f = open(result_ini,'w')
- f.write(os.linesep.join(inifile))
-
-def make_lane_dict(lib_db, lib_id):
- """
- Convert the lane_set in a lib_db to a dictionary
- indexed by flowcell ID
+def scan_submission_dirs(view_map, library_result_map):
+ """Look through our submission directories and collect needed information
"""
- result = []
- for lane in lib_db[lib_id]['lane_set']:
- result.append((lane['flowcell'], lane))
- return dict(result)
-
-def make_all_ddfs(library_result_map, daf_name, make_condor=True):
+ for lib_id, result_dir in library_result_map:
+ logging.info("Importing %s from %s" % (lib_id, result_dir))
+ try:
+ view_map.import_submission_dir(result_dir, lib_id)
+ except MetadataLookupException, e:
+ logging.error("Skipping %s: %s" % (lib_id, str(e)))
+
+def make_all_ddfs(view_map, library_result_map, daf_name, make_condor=True, force=False):
dag_fragment = []
for lib_id, result_dir in library_result_map:
- ininame = result_dir+'.ini'
- inipathname = os.path.join(result_dir, ininame)
- if os.path.exists(inipathname):
- dag_fragment.extend(
- make_ddf(ininame, daf_name, True, make_condor, result_dir)
- )
+ submissionNode = view_map.get_submission_node(result_dir)
+ dag_fragment.extend(
+ make_ddf(view_map, submissionNode, daf_name, make_condor, result_dir)
+ )
if make_condor and len(dag_fragment) > 0:
dag_filename = 'submission.dagman'
- if os.path.exists(dag_filename):
+ if not force and os.path.exists(dag_filename):
logging.warn("%s exists, please delete" % (dag_filename,))
else:
f = open(dag_filename,'w')
f.close()
-def make_ddf(ininame, daf_name, guess_ddf=False, make_condor=False, outdir=None):
+def make_ddf(view_map, submissionNode, daf_name, make_condor=False, outdir=None):
"""
Make ddf files, and bonus condor file
"""
+ query_template = """PREFIX libraryOntology: <http://jumpgate.caltech.edu/wiki/LibraryOntology#>
+PREFIX submissionOntology: <http://jumpgate.caltech.edu/wiki/UcscSubmissionOntology#>
+PREFIX ucscDaf: <http://jumpgate.caltech.edu/wiki/UcscDaf#>
+
+select ?submitView ?filename ?md5sum ?view ?cell ?antibody ?sex ?control ?controlId ?labExpId ?labVersion ?treatment ?protocol
+WHERE {
+ ?file ucscDaf:filename ?filename ;
+ ucscDaf:md5sum ?md5sum .
+ ?submitView ucscDaf:has_file ?file ;
+ ucscDaf:view ?dafView ;
+ ucscDaf:submission %(submission)s .
+ ?dafView ucscDaf:name ?view .
+ %(submission)s submissionOntology:library ?library .
+
+ OPTIONAL { ?submitView ucscDaf:antibody ?antibody }
+ OPTIONAL { ?submitView ucscDaf:cell ?cell }
+ OPTIONAL { ?submitView ucscDaf:control ?control }
+ OPTIONAL { ?library ucscDaf:controlId ?controlId }
+ OPTIONAL { ?submitView ucscDaf:sex ?sex }
+ OPTIONAL { ?submitView ucscDaf:labVersion ?labExpId }
+ OPTIONAL { ?submitView ucscDaf:labVersion ?labVersion }
+ OPTIONAL { ?library ucscDaf:treatment ?treatment }
+ OPTIONAL { ?submitView ucscDaf:protocol ?protocol }
+}
+ORDER BY ?submitView"""
dag_fragments = []
- curdir = os.getcwd()
+
+ name = fromTypedNode(view_map.model.get_target(submissionNode, submissionOntology['name']))
+ if name is None:
+ logging.error("Need name for %s" % (str(submissionNode)))
+ return []
+
+ ddf_name = name + '.ddf'
if outdir is not None:
- os.chdir(outdir)
- output = sys.stdout
- ddf_name = None
- if guess_ddf:
- ddf_name = make_ddf_name(ininame)
- print ddf_name
- output = open(ddf_name,'w')
+ outfile = os.path.join(outdir, ddf_name)
+ output = open(outfile,'w')
+ else:
+ output = sys.stdout
+
+ formatted_query = query_template % {'submission': str(submissionNode)}
+
+ query = RDF.SPARQLQuery(formatted_query)
+ results = query.execute(view_map.model)
- file_list = read_ddf_ini(ininame, output)
+ variables = ['filename']
+ # filename goes first
+ variables.extend(view_map.get_daf_variables())
+ variables += ['controlId', 'labExpId', 'md5sum']
+ output.write('\t'.join(variables))
+ output.write(os.linesep)
+
+ all_views = {}
+ all_files = []
+ for row in results:
+ viewname = fromTypedNode(row['view'])
+ current = all_views.setdefault(viewname, {})
+ for variable_name in variables:
+ value = str(fromTypedNode(row[variable_name]))
+ if variable_name in ('filename', 'md5sum'):
+ current.setdefault(variable_name,[]).append(value)
+ else:
+ current[variable_name] = value
+
+ for view in all_views.keys():
+ line = []
+ for variable_name in variables:
+ if variable_name in ('filename', 'md5sum'):
+ line.append(','.join(all_views[view][variable_name]))
+ else:
+ line.append(all_views[view][variable_name])
+ output.write("\t".join(line))
+ output.write(os.linesep)
+ all_files.extend(all_views[view]['filename'])
+
+ logging.info(
+ "Examined {0}, found files: {1}".format(
+ str(submissionNode), ", ".join(all_files)))
- file_list.append(daf_name)
- if ddf_name is not None:
- file_list.append(ddf_name)
+ all_files.append(daf_name)
+ all_files.append(ddf_name)
if make_condor:
- archive_condor = make_condor_archive_script(ininame, file_list)
- upload_condor = make_condor_upload_script(ininame)
+ archive_condor = make_condor_archive_script(name, all_files, outdir)
+ upload_condor = make_condor_upload_script(name, outdir)
dag_fragments.extend(
- make_dag_fragment(ininame, archive_condor, upload_condor)
+ make_dag_fragment(name, archive_condor, upload_condor)
)
- os.chdir(curdir)
-
return dag_fragments
-def read_ddf_ini(filename, output=sys.stdout):
- """
- Read a ini file and dump out a tab delmited text file
- """
- file_list = []
- config = SafeConfigParser()
- config.read(filename)
- order_by = shlex.split(config.get("config", "order_by"))
-
- output.write("\t".join(order_by))
- output.write(os.linesep)
- sections = config.sections()
- sections.sort()
- for section in sections:
- if section == "config":
- # skip the config block
- continue
- values = []
- for key in order_by:
- v = config.get(section, key)
- values.append(v)
- if key == 'files':
- file_list.extend(parse_filelist(v))
-
- output.write("\t".join(values))
- output.write(os.linesep)
- return file_list
-
def read_library_result_map(filename):
"""
Read a file that maps library id to result directory.
results = []
for line in stream:
- if not line.startswith('#'):
- library_id, result_dir = line.strip().split()
+ line = line.rstrip()
+ if not line.startswith('#') and len(line) > 0 :
+ library_id, result_dir = line.split()
results.append((library_id, result_dir))
return results
-
-def make_condor_archive_script(ininame, files):
+
+
+def make_condor_archive_script(name, files, outdir=None):
script = """Universe = vanilla
Executable = /bin/tar
-arguments = czvf ../%(archivename)s %(filelist)s
+arguments = czvhf ../%(archivename)s %(filelist)s
-Error = compress.err.$(Process).log
+Error = compress.out.$(Process).log
Output = compress.out.$(Process).log
-Log = /tmp/submission-compress.log
+Log = /tmp/submission-compress-%(user)s.log
initialdir = %(initialdir)s
+environment="GZIP=-3"
+request_memory = 20
queue
"""
+ if outdir is None:
+ outdir = os.getcwd()
for f in files:
- if not os.path.exists(f):
+ pathname = os.path.join(outdir, f)
+ if not os.path.exists(pathname):
raise RuntimeError("Missing %s" % (f,))
- context = {'archivename': make_submission_name(ininame),
+ context = {'archivename': make_submission_name(name),
'filelist': " ".join(files),
- 'initialdir': os.getcwd()}
+ 'initialdir': os.path.abspath(outdir),
+ 'user': os.getlogin()}
- condor_script = make_condor_name(ininame, 'archive')
+ condor_script = os.path.join(outdir, make_condor_name(name, 'archive'))
condor_stream = open(condor_script,'w')
condor_stream.write(script % context)
condor_stream.close()
return condor_script
-def make_condor_upload_script(ininame):
+
+def make_condor_upload_script(name, outdir=None):
script = """Universe = vanilla
Executable = /usr/bin/lftp
-arguments = -c put ../%(archivename)s -o ftp://detrout@encodeftp.cse.ucsc.edu/
+arguments = -c put ../%(archivename)s -o ftp://%(ftpuser)s:%(ftppassword)s@%(ftphost)s/%(archivename)s
-Error = upload.err.$(Process).log
+Error = upload.out.$(Process).log
Output = upload.out.$(Process).log
-Log = /tmp/submission-upload.log
+Log = /tmp/submission-upload-%(user)s.log
initialdir = %(initialdir)s
queue
"""
- context = {'archivename': make_submission_name(ininame),
- 'initialdir': os.getcwd()}
-
- condor_script = make_condor_name(ininame, 'upload')
+ if outdir is None:
+ outdir = os.getcwd()
+
+ auth = netrc.netrc(os.path.expanduser("~diane/.netrc"))
+
+ encodeftp = 'encodeftp.cse.ucsc.edu'
+ ftpuser = auth.hosts[encodeftp][0]
+ ftppassword = auth.hosts[encodeftp][2]
+ context = {'archivename': make_submission_name(name),
+ 'initialdir': os.path.abspath(outdir),
+ 'user': os.getlogin(),
+ 'ftpuser': ftpuser,
+ 'ftppassword': ftppassword,
+ 'ftphost': encodeftp}
+
+ condor_script = os.path.join(outdir, make_condor_name(name, 'upload'))
condor_stream = open(condor_script,'w')
condor_stream.write(script % context)
condor_stream.close()
+ os.chmod(condor_script, stat.S_IREAD|stat.S_IWRITE)
+
return condor_script
+
def make_dag_fragment(ininame, archive_condor, upload_condor):
"""
Make the couple of fragments compress and then upload the data.
return fragments
+
def get_library_info(host, apidata, library_id):
url = api.library_url(host, library_id)
contents = api.retrieve_info(url, apidata)
return contents
-def condor_srf_to_fastq(srf_file, target_pathname, paired, flowcell=None,
- mid=None, force=False):
- args = [ srf_file, ]
- if paired:
- args.extend(['--left', target_pathname])
- # this is ugly. I did it because I was pregenerating the target
- # names before I tried to figure out what sources could generate
- # those targets, and everything up to this point had been
- # one-to-one. So I couldn't figure out how to pair the
- # target names.
- # With this at least the command will run correctly.
- # however if we rename the default targets, this'll break
- # also I think it'll generate it twice.
- args.extend(['--right',
- target_pathname.replace('_r1.fastq', '_r2.fastq')])
- else:
- args.extend(['--single', target_pathname ])
- if flowcell is not None:
- args.extend(['--flowcell', flowcell])
-
- if mid is not None:
- args.extend(['-m', str(mid)])
-
- if force:
- args.extend(['--force'])
-
- script = """
-arguments="%s"
-queue
-""" % (" ".join(args),)
-
- return script
-
-def condor_qseq_to_fastq(qseq_file, target_pathname, flowcell=None, force=False):
- args = ['-i', qseq_file, '-o', target_pathname ]
- if flowcell is not None:
- args.extend(['-f', flowcell])
- script = """
-arguments="%s"
-queue
-""" % (" ".join(args))
-
- return script
-def find_archive_sequence_files(host, apidata, sequences_path,
- library_result_map):
- """
- Find all the archive sequence files possibly associated with our results.
-
- """
- logging.debug("Searching for sequence files in: %s" %(sequences_path,))
-
- lib_db = {}
- seq_dirs = set()
- #seq_dirs = set(os.path.join(sequences_path, 'srfs'))
- candidate_lanes = {}
- for lib_id, result_dir in library_result_map:
- lib_info = get_library_info(host, apidata, lib_id)
- lib_db[lib_id] = lib_info
-
- for lane in lib_info['lane_set']:
- lane_key = (lane['flowcell'], lane['lane_number'])
- candidate_lanes[lane_key] = lib_id
- seq_dirs.add(os.path.join(sequences_path,
- 'flowcells',
- lane['flowcell']))
- logging.debug("Seq_dirs = %s" %(unicode(seq_dirs)))
- candidate_seq_list = scan_for_sequences(seq_dirs)
-
- # at this point we have too many sequences as scan_for_sequences
- # returns all the sequences in a flowcell directory
- # so lets filter out the extras
-
- for seq in candidate_seq_list:
- lane_key = (seq.flowcell, seq.lane)
- lib_id = candidate_lanes.get(lane_key, None)
- if lib_id is not None:
- lib_info = lib_db[lib_id]
- lanes = lib_info.setdefault('lanes', {})
- lanes.setdefault(lane_key, set()).add(seq)
-
- return lib_db
-
-def find_best_extension(extension_map, filename):
- """
- Search through extension_map looking for the best extension
- The 'best' is the longest match
-
- :Args:
- extension_map (dict): '.ext' -> { 'view': 'name' or None }
- filename (str): the filename whose extention we are about to examine
- """
- best_ext = None
- path, last_ext = os.path.splitext(filename)
-
- for ext in extension_map.keys():
- if filename.endswith(ext):
- if best_ext is None:
- best_ext = ext
- elif len(ext) > len(best_ext):
- best_ext = ext
- return best_ext
-
-def make_submission_section(line_counter, files, standard_attributes, file_attributes):
+def make_submission_section(line_counter, files, attributes):
"""
Create a section in the submission ini file
"""
- inifile = [ '[line%s]' % (line_counter,) ]
+ inifile = [ "[line%s]" % (line_counter,) ]
inifile += ["files=%s" % (",".join(files))]
- cur_attributes = {}
- cur_attributes.update(standard_attributes)
- cur_attributes.update(file_attributes)
-
- for k,v in cur_attributes.items():
+
+ for k,v in attributes.items():
inifile += ["%s=%s" % (k,v)]
return inifile
+
def make_base_name(pathname):
base = os.path.basename(pathname)
name, ext = os.path.splitext(base)
return name
+
def make_submission_name(ininame):
name = make_base_name(ininame)
- return name + '.tgz'
+ return name + ".tgz"
+
def make_ddf_name(pathname):
name = make_base_name(pathname)
- return name + '.ddf'
+ return name + ".ddf"
+
def make_condor_name(pathname, run_type=None):
name = make_base_name(pathname)
elements = [name]
if run_type is not None:
elements.append(run_type)
- elements.append('condor')
+ elements.append("condor")
return ".".join(elements)
-
-def make_submit_script(target, header, body_list):
- """
- write out a text file
- this was intended for condor submit scripts
-
- Args:
- target (str or stream):
- if target is a string, we will open and close the file
- if target is a stream, the caller is responsible.
-
- header (str);
- header to write at the beginning of the file
- body_list (list of strs):
- a list of blocks to add to the file.
- """
- if type(target) in types.StringTypes:
- f = open(target,'w')
- else:
- f = target
- f.write(header)
- for entry in body_list:
- f.write(entry)
- if type(target) in types.StringTypes:
- f.close()
def parse_filelist(file_string):
- return file_string.split(',')
+ return file_string.split(",")
+
def validate_filelist(files):
"""
for f in files:
if not os.path.exists(f):
raise RuntimeError("%s does not exist" % (f,))
-
+
if __name__ == "__main__":
main()
projects / htsworkflow.git / blobdiff