htsworkflow/submission/submission.py

   1 """Common submission elements
   2 """
   3 import logging
   4 import os
   5 import re
   6
   7 import RDF
   8
   9 from htsworkflow.util.rdfhelp import \
  10      blankOrUri, \
  11      dafTermOntology, \
  12      dump_model, \
  13      get_model, \
  14      libraryOntology, \
  15      owlNS, \
  16      rdfNS, \
  17      submissionLog, \
  18      submissionOntology, \
  19      toTypedNode, \
  20      fromTypedNode
  21 from htsworkflow.util.hashfile import make_md5sum
  22
  23 from htsworkflow.submission.daf import \
  24      MetadataLookupException, \
  25      get_submission_uri
  26
  27 logger = logging.getLogger(__name__)
  28
  29 class Submission(object):
  30     def __init__(self, name, model):
  31         self.name = name
  32         self.model = model
  33
  34         self.submissionSet = get_submission_uri(self.name)
  35         self.submissionSetNS = RDF.NS(str(self.submissionSet) + '/')
  36         self.libraryNS = RDF.NS('http://jumpgate.caltech.edu/library/')
  37
  38         self.__view_map = None
  39
  40     def scan_submission_dirs(self, result_map):
  41         """Examine files in our result directory
  42         """
  43         for lib_id, result_dir in result_map.items():
  44             logger.info("Importing %s from %s" % (lib_id, result_dir))
  45             try:
  46                 self.import_analysis_dir(result_dir, lib_id)
  47             except MetadataLookupException, e:
  48                 logger.error("Skipping %s: %s" % (lib_id, str(e)))
  49
  50     def import_analysis_dir(self, analysis_dir, library_id):
  51         """Import a submission directories and update our model as needed
  52         """
  53         #attributes = get_filename_attribute_map(paired)
  54         libNode = self.libraryNS[library_id + "/"]
  55
  56         self._add_library_details_to_model(libNode)
  57
  58         submission_files = os.listdir(analysis_dir)
  59         for filename in submission_files:
  60             self.construct_file_attributes(analysis_dir, libNode, filename)
  61
  62     def construct_file_attributes(self, analysis_dir, libNode, pathname):
  63         """Looking for the best extension
  64         The 'best' is the longest match
  65
  66         :Args:
  67         filename (str): the filename whose extention we are about to examine
  68         """
  69         path, filename = os.path.split(pathname)
  70
  71         logger.debug("Searching for view")
  72         file_classification = self.find_best_match(filename)
  73         if file_classification is None:
  74             logger.warn("Unrecognized file: {0}".format(pathname))
  75             return None
  76         if str(file_classification) == str(libraryOntology['ignore']):
  77             return None
  78
  79         an_analysis_name = self.make_submission_name(analysis_dir)
  80         an_analysis = self.get_submission_node(analysis_dir)
  81         an_analysis_uri = str(an_analysis.uri)
  82
  83         self.model.add_statement(RDF.Statement(an_analysis,
  84                                                submissionOntology['name'],
  85                                                toTypedNode(an_analysis_name)))
  86         self.model.add_statement(
  87             RDF.Statement(an_analysis,
  88                           rdfNS['type'],
  89                           submissionOntology['submission']))
  90         self.model.add_statement(RDF.Statement(an_analysis,
  91                                                submissionOntology['library'],
  92                                                libNode))
  93
  94         logger.debug("Adding statements to {0}".format(str(an_analysis)))
  95         # add track specific information
  96         self.model.add_statement(
  97             RDF.Statement(an_analysis,
  98                           dafTermOntology['paired'],
  99                           toTypedNode(self._is_paired(libNode))))
 100         self.model.add_statement(
 101             RDF.Statement(an_analysis,
 102                           dafTermOntology['submission'],
 103                           an_analysis))
 104
 105         # add file specific information
 106         fileNode = self.link_file_to_classes(filename,
 107                                              an_analysis,
 108                                              an_analysis_uri,
 109                                              analysis_dir)
 110         self.add_md5s(filename, fileNode, analysis_dir)
 111
 112         logger.debug("Done.")
 113
 114     def link_file_to_classes(self, filename, submissionNode, submission_uri, analysis_dir):
 115         # add file specific information
 116         fileNode = RDF.Node(RDF.Uri(submission_uri + '/' + filename))
 117         self.model.add_statement(
 118             RDF.Statement(submissionNode,
 119                           dafTermOntology['has_file'],
 120                           fileNode))
 121         self.model.add_statement(
 122             RDF.Statement(fileNode,
 123                           dafTermOntology['filename'],
 124                           filename))
 125         return fileNode
 126
 127     def add_md5s(self, filename, fileNode, analysis_dir):
 128         logger.debug("Updating file md5sum")
 129         submission_pathname = os.path.join(analysis_dir, filename)
 130         md5 = make_md5sum(submission_pathname)
 131         if md5 is None:
 132             errmsg = "Unable to produce md5sum for {0}"
 133             logger.warning(errmsg.format(submission_pathname))
 134         else:
 135             self.model.add_statement(
 136                 RDF.Statement(fileNode, dafTermOntology['md5sum'], md5))
 137
 138     def _add_library_details_to_model(self, libNode):
 139         parser = RDF.Parser(name='rdfa')
 140         new_statements = parser.parse_as_stream(libNode.uri)
 141         for s in new_statements:
 142             # don't override things we already have in the model
 143             targets = list(self.model.get_targets(s.subject, s.predicate))
 144             if len(targets) == 0:
 145                 self.model.append(s)
 146
 147
 148     def find_best_match(self, filename):
 149         """Search through potential filename matching patterns
 150         """
 151         if self.__view_map is None:
 152             self.__view_map = self._get_filename_view_map()
 153
 154         results = []
 155         for pattern, view in self.__view_map.items():
 156             if re.match(pattern, filename):
 157                 results.append(view)
 158
 159         if len(results) > 1:
 160             msg = "%s matched multiple views %s" % (
 161                 filename,
 162                 [str(x) for x in results])
 163             raise ModelException(msg)
 164         elif len(results) == 1:
 165             return results[0]
 166         else:
 167             return None
 168
 169     def _get_filename_view_map(self):
 170         """Query our model for filename patterns
 171
 172         return a dictionary of compiled regular expressions to view names
 173         """
 174         filename_query = RDF.Statement(
 175             None, dafTermOntology['filename_re'], None)
 176
 177         patterns = {}
 178         for s in self.model.find_statements(filename_query):
 179             view_name = s.subject
 180             literal_re = s.object.literal_value['string']
 181             logger.debug("Found: %s" % (literal_re,))
 182             try:
 183                 filename_re = re.compile(literal_re)
 184             except re.error, e:
 185                 logger.error("Unable to compile: %s" % (literal_re,))
 186             patterns[literal_re] = view_name
 187         return patterns
 188
 189     def make_submission_name(self, analysis_dir):
 190         analysis_dir = os.path.normpath(analysis_dir)
 191         analysis_dir_name = os.path.split(analysis_dir)[1]
 192         if len(analysis_dir_name) == 0:
 193             raise RuntimeError(
 194                 "Submission dir name too short: {0}".format(analysis_dir))
 195         return analysis_dir_name
 196
 197     def get_submission_node(self, analysis_dir):
 198         """Convert a submission directory name to a submission node
 199         """
 200         submission_name = self.make_submission_name(analysis_dir)
 201         return self.submissionSetNS[submission_name]
 202
 203     def _get_library_attribute(self, libNode, attribute):
 204         if not isinstance(attribute, RDF.Node):
 205             attribute = libraryOntology[attribute]
 206
 207         targets = list(self.model.get_targets(libNode, attribute))
 208         if len(targets) > 0:
 209             return self._format_library_attribute(targets)
 210         else:
 211             return None
 212
 213         #targets = self._search_same_as(libNode, attribute)
 214         #if targets is not None:
 215         #    return self._format_library_attribute(targets)
 216
 217         # we don't know anything about this attribute
 218         self._add_library_details_to_model(libNode)
 219
 220         targets = list(self.model.get_targets(libNode, attribute))
 221         if len(targets) > 0:
 222             return self._format_library_attribute(targets)
 223
 224         return None
 225
 226     def _format_library_attribute(self, targets):
 227         if len(targets) == 0:
 228             return None
 229         elif len(targets) == 1:
 230             return fromTypedNode(targets[0])
 231         elif len(targets) > 1:
 232             return [fromTypedNode(t) for t in targets]
 233
 234     def _is_paired(self, libNode):
 235         """Determine if a library is paired end"""
 236         library_type = self._get_library_attribute(libNode, 'library_type')
 237         if library_type is None:
 238             errmsg = "%s doesn't have a library type"
 239             raise ModelException(errmsg % (str(libNode),))
 240
 241         single = ['CSHL (lacking last nt)',
 242                   'Single End (non-multiplexed)',
 243                   'Small RNA (non-multiplexed)',]
 244         paired = ['Barcoded Illumina',
 245                   'Multiplexing',
 246                   'Nextera',
 247                   'Paired End (non-multiplexed)',]
 248         if library_type in single:
 249             return False
 250         elif library_type in paired:
 251             return True
 252         else:
 253             raise MetadataLookupException(
 254                 "Unrecognized library type %s for %s" % \
 255                 (library_type, str(libNode)))
 256