htsworkflow/pipelines/bustard.py

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