htsworkflow/pipelines/bustard.py

   1 """
   2 Extract configuration from Illumina Bustard Directory.
   3
   4 This includes the version number, run date, bustard executable parameters, and
   5 phasing estimates.
   6 """
   7 from datetime import date
   8 from glob import glob
   9 import logging
  10 import os
  11 import time
  12 import re
  13
  14 from htsworkflow.pipelines.runfolder import \
  15    ElementTree, \
  16    VERSION_RE, \
  17    EUROPEAN_STRPTIME
  18
  19 # make epydoc happy
  20 __docformat__ = "restructuredtext en"
  21
  22 class Phasing(object):
  23     PHASING = 'Phasing'
  24     PREPHASING = 'Prephasing'
  25
  26     def __init__(self, fromfile=None, xml=None):
  27         self.lane = None
  28         self.phasing = None
  29         self.prephasing = None
  30
  31         if fromfile is not None:
  32             self._initialize_from_file(fromfile)
  33         elif xml is not None:
  34             self.set_elements(xml)
  35
  36     def _initialize_from_file(self, pathname):
  37         path, name = os.path.split(pathname)
  38         basename, ext = os.path.splitext(name)
  39         # the last character of the param base filename should be the
  40         # lane number
  41         tree = ElementTree.parse(pathname).getroot()
  42         self.set_elements(tree)
  43         self.lane = int(basename[-1])
  44
  45     def get_elements(self):
  46         root = ElementTree.Element(Phasing.PHASING, {'lane': str(self.lane)})
  47         phasing = ElementTree.SubElement(root, Phasing.PHASING)
  48         phasing.text = str(self.phasing)
  49         prephasing = ElementTree.SubElement(root, Phasing.PREPHASING)
  50         prephasing.text = str(self.prephasing)
  51         return root
  52
  53     def set_elements(self, tree):
  54         if tree.tag not in ('Phasing', 'Parameters'):
  55             raise ValueError('exptected Phasing or Parameters')
  56         lane = tree.attrib.get('lane', None)
  57         if lane is not None:
  58             self.lane = int(lane)
  59         for element in list(tree):
  60             if element.tag == Phasing.PHASING:
  61                 self.phasing = float(element.text)
  62             elif element.tag == Phasing.PREPHASING:
  63                 self.prephasing = float(element.text)
  64
  65 class Bustard(object):
  66     XML_VERSION = 1
  67
  68     # Xml Tags
  69     BUSTARD = 'Bustard'
  70     SOFTWARE_VERSION = 'version'
  71     DATE = 'run_time'
  72     USER = 'user'
  73     PARAMETERS = 'Parameters'
  74
  75     def __init__(self, xml=None):
  76         self.version = None
  77         self.date = date.today()
  78         self.user = None
  79         self.phasing = {}
  80         self.pathname = None
  81
  82         if xml is not None:
  83             self.set_elements(xml)
  84
  85     def _get_time(self):
  86         return time.mktime(self.date.timetuple())
  87     time = property(_get_time, doc='return run time as seconds since epoch')
  88
  89     def dump(self):
  90         print "Bustard version:", self.version
  91         print "Run date", self.date
  92         print "user:", self.user
  93         for lane, tree in self.phasing.items():
  94             print lane
  95             print tree
  96
  97     def get_elements(self):
  98         root = ElementTree.Element('Bustard',
  99                                    {'version': str(Bustard.XML_VERSION)})
 100         version = ElementTree.SubElement(root, Bustard.SOFTWARE_VERSION)
 101         version.text = self.version
 102         run_date = ElementTree.SubElement(root, Bustard.DATE)
 103         run_date.text = str(self.time)
 104         user = ElementTree.SubElement(root, Bustard.USER)
 105         user.text = self.user
 106         params = ElementTree.SubElement(root, Bustard.PARAMETERS)
 107         for p in self.phasing.values():
 108             params.append(p.get_elements())
 109         return root
 110
 111     def set_elements(self, tree):
 112         if tree.tag != Bustard.BUSTARD:
 113             raise ValueError('Expected "Bustard" SubElements')
 114         xml_version = int(tree.attrib.get('version', 0))
 115         if xml_version > Bustard.XML_VERSION:
 116             logging.warn('Bustard XML tree is a higher version than this class')
 117         for element in list(tree):
 118             if element.tag == Bustard.SOFTWARE_VERSION:
 119                 self.version = element.text
 120             elif element.tag == Bustard.DATE:
 121                 self.date = date.fromtimestamp(float(element.text))
 122             elif element.tag == Bustard.USER:
 123                 self.user = element.text
 124             elif element.tag == Bustard.PARAMETERS:
 125                 for param in element:
 126                     p = Phasing(xml=param)
 127                     self.phasing[p.lane] = p
 128             else:
 129                 raise ValueError("Unrecognized tag: %s" % (element.tag,))
 130
 131
 132
 133 def bustard(pathname):
 134     """
 135     Construct a Bustard object by analyzing an Illumina Bustard directory.
 136
 137     :Parameters:
 138       - `pathname`: A bustard directory
 139
 140     :Return:
 141       Fully initialized Bustard object.
 142     """
 143     b = Bustard()
 144     path, name = os.path.split(pathname)
 145     groups = name.split("_")
 146     version = re.search(VERSION_RE, groups[0])
 147     b.version = version.group(1)
 148     t = time.strptime(groups[1], EUROPEAN_STRPTIME)
 149     b.date = date(*t[0:3])
 150     b.user = groups[2]
 151     b.pathname = pathname
 152     paramfiles = glob(os.path.join(pathname, "params?.xml"))
 153     for paramfile in paramfiles:
 154         phasing = Phasing(paramfile)
 155         assert (phasing.lane >= 1 and phasing.lane <= 8)
 156         b.phasing[phasing.lane] = phasing
 157     return b
 158
 159 def fromxml(tree):
 160     """
 161     Reconstruct a htsworkflow.pipelines.Bustard object from an xml block
 162     """
 163     b = Bustard()
 164     b.set_elements(tree)
 165     return b