htsworkflow/submission/daf.py

   1 """Parse UCSC DAF File
   2 """
   3 import logging
   4 import os
   5 import re
   6 import string
   7 from StringIO import StringIO
   8 import types
   9 import urlparse
  10
  11 import RDF
  12 from htsworkflow.util.rdfhelp import \
  13      blankOrUri, \
  14      dafTermOntology, \
  15      get_model, \
  16      libraryOntology, \
  17      owlNS, \
  18      rdfNS, \
  19      submissionLog, \
  20      submissionOntology, \
  21      toTypedNode, \
  22      fromTypedNode
  23 from htsworkflow.util.hashfile import make_md5sum
  24
  25 logger = logging.getLogger(__name__)
  26
  27 DAF_VARIABLE_NAMES = ("variables", "extraVariables")
  28 VARIABLES_TERM_NAME = 'variables'
  29
  30 class ModelException(RuntimeError):
  31     """Assumptions about the RDF model failed"""
  32     pass
  33
  34
  35 class MetadataLookupException(RuntimeError):
  36     """Problem accessing metadata"""
  37     pass
  38
  39
  40 # STATES
  41 DAF_HEADER = 1
  42 DAF_VIEW = 2
  43
  44
  45 def parse_into_model(model, subject, filename):
  46     """Read a DAF into RDF Model
  47
  48     requires a subject node to attach statements to
  49     """
  50     attributes = parse(filename)
  51     add_to_model(model, attributes, subject)
  52
  53
  54 def fromstream_into_model(model, subject, daf_stream):
  55     """Load daf stream into model attached to node subject
  56     """
  57     attributes = parse_stream(daf_stream)
  58     add_to_model(model, attributes, subject)
  59
  60
  61 def fromstring_into_model(model, subject, daf_string):
  62     """Read a string containing a DAF into RDF Model
  63
  64     requires a short submission name
  65     """
  66     attributes = fromstring(daf_string)
  67     add_to_model(model, attributes, subject)
  68
  69
  70 def parse(filename):
  71     """Parse daf from a file
  72     """
  73     stream = open(filename, 'r')
  74     attributes = parse_stream(stream)
  75     stream.close()
  76     return attributes
  77
  78
  79 def fromstring(daf_string):
  80     """Parse UCSC daf from a provided string"""
  81     stream = StringIO(daf_string)
  82     return parse_stream(stream)
  83
  84
  85 def parse_stream(stream):
  86     """Parse UCSC dat stored in a stream"""
  87     comment_re = re.compile("#.*$")
  88
  89     state = DAF_HEADER
  90     attributes = {'views': {}}
  91     view_name = None
  92     view_attributes = {}
  93     for line in stream:
  94         #remove comments
  95         line = comment_re.sub("", line)
  96         nstop = _extract_name_index(line)
  97         name = line[0:nstop]
  98         sstop = _consume_whitespace(line, start=nstop)
  99         vstop = _extract_value_index(line, start=sstop)
 100         value = line[sstop:vstop]
 101
 102         if value.lower() in ('yes',):
 103             value = True
 104         elif value.lower() in ('no',):
 105             value = False
 106
 107         if len(name) == 0:
 108             if view_name is not None:
 109                 attributes['views'][view_name] = view_attributes
 110                 view_name = None
 111                 view_attributes = {}
 112             state = DAF_HEADER
 113         elif state == DAF_HEADER and name in DAF_VARIABLE_NAMES:
 114             attributes[name] = [x.strip() for x in value.split(',')]
 115         elif state == DAF_HEADER and name == 'view':
 116             view_name = value
 117             view_attributes['view'] = value
 118             state = DAF_VIEW
 119         elif state == DAF_HEADER:
 120             attributes[name] = value
 121         elif state == DAF_VIEW:
 122             view_attributes[name] = value
 123
 124     # save last block
 125     if view_name is not None:
 126         attributes['views'][view_name] = view_attributes
 127
 128     return attributes
 129
 130
 131 def _consume_whitespace(line, start=0):
 132     """return index of next non whitespace character
 133
 134     returns length of string if it can't find anything
 135     """
 136     for i in xrange(start, len(line)):
 137         if line[i] not in string.whitespace:
 138             return i
 139
 140     return len(line)
 141
 142
 143 def _extract_name_index(line, start=0):
 144     """Used to find end of word by looking for a whitespace character
 145
 146     returns length of string if nothing matches
 147     """
 148     for i in xrange(start, len(line)):
 149         if line[i] in string.whitespace:
 150             return i
 151
 152     return len(line)
 153
 154
 155 def _extract_value_index(line, start=0):
 156     """Returns position of last non-whitespace character
 157     """
 158     shortline = line.rstrip()
 159     return len(shortline)
 160
 161
 162 def convert_to_rdf_statements(attributes, subject):
 163     """Convert dictionary of DAF attributes into rdf statements
 164
 165     The statements are attached to the provided subject node
 166     """
 167     variables_term = dafTermOntology[VARIABLES_TERM_NAME]
 168     statements = []
 169     for daf_key in attributes:
 170         predicate = dafTermOntology[daf_key]
 171         if daf_key == 'views':
 172             statements.extend(_views_to_statements(subject,
 173                                                    dafTermOntology,
 174                                                    attributes[daf_key]))
 175         elif daf_key in DAF_VARIABLE_NAMES:
 176             for var in attributes.get(daf_key, []):
 177                 obj = toTypedNode(var)
 178                 statements.append(RDF.Statement(subject, variables_term, obj))
 179         else:
 180             value = attributes[daf_key]
 181             obj = toTypedNode(value)
 182             statements.append(RDF.Statement(subject, predicate, obj))
 183
 184     return statements
 185
 186
 187 def _views_to_statements(subject, dafNS, views):
 188     """Attach view attributes to new view nodes atached to provided subject
 189     """
 190     viewNS = get_view_namespace(subject)
 191
 192     statements = []
 193     for view_name in views:
 194         view_attributes = views[view_name]
 195         viewSubject = viewNS[view_name]
 196         statements.append(RDF.Statement(subject, dafNS['views'], viewSubject))
 197         statements.append(
 198             RDF.Statement(viewSubject, dafNS['name'], toTypedNode(view_name)))
 199         for view_attribute_name in view_attributes:
 200             predicate = dafNS[view_attribute_name]
 201             obj = toTypedNode(view_attributes[view_attribute_name])
 202             statements.append(RDF.Statement(viewSubject, predicate, obj))
 203
 204         #statements.extend(convert_to_rdf_statements(view, viewNode))
 205     return statements
 206
 207
 208 def add_to_model(model, attributes, subject):
 209     for statement in convert_to_rdf_statements(attributes, subject):
 210         model.add_statement(statement)
 211
 212
 213 def get_submission_uri(name):
 214     return submissionLog[name].uri
 215
 216
 217 def submission_uri_to_string(submission_uri):
 218     if isinstance(submission_uri, RDF.Node):
 219         submission_uri = str(submission_uri.uri)
 220     elif isinstance(submission_uri, RDF.Uri):
 221         submission_uri = str(submission_uri)
 222     if submission_uri[-1] != '/':
 223         submission_uri += '/'
 224     return submission_uri
 225
 226
 227 def get_view_namespace(submission_uri):
 228     submission_uri = submission_uri_to_string(submission_uri)
 229     view_uri = urlparse.urljoin(submission_uri, 'view/')
 230     viewNS = RDF.NS(view_uri)
 231     return viewNS
 232
 233
 234 class DAFMapper(object):
 235     """Convert filenames to views in the UCSC Daf
 236     """
 237     def __init__(self, name, daf_file=None, model=None):
 238         """Construct a RDF backed model of a UCSC DAF
 239
 240         :args:
 241           name (str): the name of this submission (used to construct DAF url)
 242           daf_file (str, stream, or None):
 243              if str, use as filename
 244              if stream, parse as stream
 245              if none, don't attempt to load the DAF into our model
 246           model (RDF.Model or None):
 247              if None, construct a memory backed model
 248              otherwise specifies model to use
 249         """
 250         if daf_file is None and model is None:
 251             logger.error("We need a DAF or Model containing a DAF to work")
 252
 253         self.name = name
 254         self.submissionSet = get_submission_uri(self.name)
 255         self.viewNS = get_view_namespace(self.submissionSet)
 256
 257         if model is not None:
 258             self.model = model
 259         else:
 260             self.model = get_model()
 261
 262         if hasattr(daf_file, 'next'):
 263             # its some kind of stream
 264             fromstream_into_model(self.model, self.submissionSet, daf_file)
 265         else:
 266             # file
 267             parse_into_model(self.model, self.submissionSet, daf_file)
 268
 269         self.libraryNS = RDF.NS('http://jumpgate.caltech.edu/library/')
 270         self.submissionSetNS = RDF.NS(str(self.submissionSet) + '/')
 271         self.__view_map = None
 272
 273     def add_pattern(self, view_name, filename_pattern):
 274         """Map a filename regular expression to a view name
 275         """
 276         obj = toTypedNode(filename_pattern)
 277         self.model.add_statement(
 278             RDF.Statement(self.viewNS[view_name],
 279                           dafTermOntology['filename_re'],
 280                           obj))
 281
 282     def import_submission_dir(self, submission_dir, library_id):
 283         """Import a submission directories and update our model as needed
 284         """
 285         #attributes = get_filename_attribute_map(paired)
 286         libNode = self.libraryNS[library_id + "/"]
 287
 288         self._add_library_details_to_model(libNode)
 289
 290         submission_files = os.listdir(submission_dir)
 291         for filename in submission_files:
 292             self.construct_track_attributes(submission_dir, libNode, filename)
 293
 294     def construct_track_attributes(self, submission_dir, libNode, pathname):
 295         """Looking for the best extension
 296         The 'best' is the longest match
 297
 298         :Args:
 299         filename (str): the filename whose extention we are about to examine
 300         """
 301         path, filename = os.path.split(pathname)
 302
 303         logger.debug("Searching for view")
 304         view = self.find_view(filename)
 305         if view is None:
 306             logger.warn("Unrecognized file: {0}".format(pathname))
 307             return None
 308         if str(view) == str(libraryOntology['ignore']):
 309             return None
 310
 311         submission_name = self.make_submission_name(submission_dir)
 312         submissionNode = self.get_submission_node(submission_dir)
 313         submission_uri = str(submissionNode.uri)
 314         view_name = fromTypedNode(self.model.get_target(view,
 315                                        dafTermOntology['name']))
 316         if view_name is None:
 317             errmsg = 'Could not find view name for {0}'
 318             logging.warning(errmsg.format(str(view)))
 319             return
 320
 321         view_name = str(view_name)
 322         submissionView = RDF.Node(RDF.Uri(submission_uri + '/' + view_name))
 323
 324         self.model.add_statement(
 325             RDF.Statement(self.submissionSet,
 326                           dafTermOntology['has_submission'],
 327                           submissionNode))
 328         logger.debug("Adding statements to {0}".format(str(submissionNode)))
 329         self.model.add_statement(RDF.Statement(submissionNode,
 330                                                submissionOntology['has_view'],
 331                                                submissionView))
 332         self.model.add_statement(RDF.Statement(submissionNode,
 333                                                submissionOntology['name'],
 334                                                toTypedNode(submission_name)))
 335         self.model.add_statement(
 336             RDF.Statement(submissionNode,
 337                           rdfNS['type'],
 338                           submissionOntology['submission']))
 339         self.model.add_statement(RDF.Statement(submissionNode,
 340                                                submissionOntology['library'],
 341                                                libNode))
 342
 343         logger.debug("Adding statements to {0}".format(str(submissionView)))
 344         # add track specific information
 345         self.model.add_statement(
 346             RDF.Statement(submissionView, dafTermOntology['view'], view))
 347         self.model.add_statement(
 348             RDF.Statement(submissionView,
 349                           dafTermOntology['paired'],
 350                           toTypedNode(self._is_paired(libNode))))
 351         self.model.add_statement(
 352             RDF.Statement(submissionView,
 353                           dafTermOntology['submission'],
 354                           submissionNode))
 355
 356         # extra information
 357         terms = [dafTermOntology['type'],
 358                  dafTermOntology['filename_re'],
 359                  ]
 360         terms.extend((dafTermOntology[v] for v in self.get_daf_variables()))
 361
 362         # add file specific information
 363         self.create_file_attributes(filename, submissionView, submission_uri, submission_dir)
 364
 365         logger.debug("Done.")
 366
 367     def create_file_attributes(self, filename, submissionView, submission_uri, submission_dir):
 368         # add file specific information
 369         logger.debug("Updating file md5sum")
 370         fileNode = RDF.Node(RDF.Uri(submission_uri + '/' + filename))
 371         submission_pathname = os.path.join(submission_dir, filename)
 372         self.model.add_statement(
 373             RDF.Statement(submissionView,
 374                           dafTermOntology['has_file'],
 375                           fileNode))
 376         self.model.add_statement(
 377             RDF.Statement(fileNode,
 378                           dafTermOntology['filename'],
 379                           filename))
 380
 381         md5 = make_md5sum(submission_pathname)
 382         if md5 is None:
 383             errmsg = "Unable to produce md5sum for {0}"
 384             logging.warning(errmsg.format(submission_pathname))
 385         else:
 386             self.model.add_statement(
 387                 RDF.Statement(fileNode, dafTermOntology['md5sum'], md5))
 388
 389     def _add_library_details_to_model(self, libNode):
 390         parser = RDF.Parser(name='rdfa')
 391         new_statements = parser.parse_as_stream(libNode.uri)
 392         for s in new_statements:
 393             # don't override things we already have in the model
 394             targets = list(self.model.get_targets(s.subject, s.predicate))
 395             if len(targets) == 0:
 396                 self.model.append(s)
 397
 398     def get_daf_variables(self):
 399         """Returns simple variables names that to include in the ddf
 400         """
 401         variables_term = dafTermOntology[VARIABLES_TERM_NAME]
 402         results = ['view']
 403         if self.need_replicate():
 404             results.append('replicate')
 405
 406         for obj in self.model.get_targets(self.submissionSet, variables_term):
 407             value = str(fromTypedNode(obj))
 408             results.append(value)
 409         results.append('labVersion')
 410         return results
 411
 412     def make_submission_name(self, submission_dir):
 413         submission_dir = os.path.normpath(submission_dir)
 414         submission_dir_name = os.path.split(submission_dir)[1]
 415         if len(submission_dir_name) == 0:
 416             raise RuntimeError(
 417                 "Submission dir name too short: {0}".format(submission_dir))
 418         return submission_dir_name
 419
 420     def get_submission_node(self, submission_dir):
 421         """Convert a submission directory name to a submission node
 422         """
 423         submission_name = self.make_submission_name(submission_dir)
 424         return self.submissionSetNS[submission_name]
 425
 426     def _get_library_attribute(self, libNode, attribute):
 427         if not isinstance(attribute, RDF.Node):
 428             attribute = libraryOntology[attribute]
 429
 430         targets = list(self.model.get_targets(libNode, attribute))
 431         if len(targets) > 0:
 432             return self._format_library_attribute(targets)
 433         else:
 434             return None
 435
 436         #targets = self._search_same_as(libNode, attribute)
 437         #if targets is not None:
 438         #    return self._format_library_attribute(targets)
 439
 440         # we don't know anything about this attribute
 441         self._add_library_details_to_model(libNode)
 442
 443         targets = list(self.model.get_targets(libNode, attribute))
 444         if len(targets) > 0:
 445             return self._format_library_attribute(targets)
 446
 447         return None
 448
 449     def _format_library_attribute(self, targets):
 450         if len(targets) == 0:
 451             return None
 452         elif len(targets) == 1:
 453             return fromTypedNode(targets[0])
 454         elif len(targets) > 1:
 455             return [fromTypedNode(t) for t in targets]
 456
 457     def _search_same_as(self, subject, predicate):
 458         # look for alternate names
 459         other_predicates = self.model.get_targets(predicate, owlNS['sameAs'])
 460         for other in other_predicates:
 461             targets = list(self.model.get_targets(subject, other))
 462             if len(targets) > 0:
 463                 return targets
 464         return None
 465
 466     def find_view(self, filename):
 467         """Search through potential DAF filename patterns
 468         """
 469         if self.__view_map is None:
 470             self.__view_map = self._get_filename_view_map()
 471
 472         results = []
 473         for pattern, view in self.__view_map.items():
 474             if re.match(pattern, filename):
 475                 results.append(view)
 476
 477         if len(results) > 1:
 478             msg = "%s matched multiple views %s" % (
 479                 filename,
 480                 [str(x) for x in results])
 481             raise ModelException(msg)
 482         elif len(results) == 1:
 483             return results[0]
 484         else:
 485             return None
 486
 487     def get_view_name(self, view):
 488         view_term = submissionOntology['view_name']
 489         names = list(self.model.get_targets(view, view_term))
 490         if len(names) == 1:
 491             return fromTypedNode(names[0])
 492         else:
 493             msg = "Found wrong number of view names for {0} len = {1}"
 494             msg = msg.format(str(view), len(names))
 495             logger.error(msg)
 496             raise RuntimeError(msg)
 497
 498     def _get_filename_view_map(self):
 499         """Query our model for filename patterns
 500
 501         return a dictionary of compiled regular expressions to view names
 502         """
 503         filename_query = RDF.Statement(
 504             None, dafTermOntology['filename_re'], None)
 505
 506         patterns = {}
 507         for s in self.model.find_statements(filename_query):
 508             view_name = s.subject
 509             literal_re = s.object.literal_value['string']
 510             logger.debug("Found: %s" % (literal_re,))
 511             try:
 512                 filename_re = re.compile(literal_re)
 513             except re.error, e:
 514                 logger.error("Unable to compile: %s" % (literal_re,))
 515             patterns[literal_re] = view_name
 516         return patterns
 517
 518     def _get_library_url(self):
 519         return str(self.libraryNS[''].uri)
 520
 521     def _set_library_url(self, value):
 522         self.libraryNS = RDF.NS(str(value))
 523
 524     library_url = property(_get_library_url, _set_library_url)
 525
 526     def _is_paired(self, libNode):
 527         """Determine if a library is paired end"""
 528         library_type = self._get_library_attribute(libNode, 'library_type')
 529         if library_type is None:
 530             errmsg = "%s doesn't have a library type"
 531             raise ModelException(errmsg % (str(libNode),))
 532
 533         #single = (1,3,6)
 534         single = ['Single End', 'Small RNA', 'CSHL (lacking last nt)']
 535         paired = ['Paired End', 'Multiplexing', 'Barcoded']
 536         if library_type in single:
 537             return False
 538         elif library_type in paired:
 539             return True
 540         else:
 541             raise MetadataLookupException(
 542                 "Unrecognized library type %s for %s" % \
 543                 (library_type, str(libNode)))
 544
 545     def need_replicate(self):
 546         viewTerm = dafTermOntology['views']
 547         replicateTerm = dafTermOntology['hasReplicates']
 548
 549         views = self.model.get_targets(self.submissionSet, viewTerm)
 550
 551         for view in views:
 552             replicate = self.model.get_target(view, replicateTerm)
 553             if fromTypedNode(replicate):
 554                 return True
 555
 556         return False