gaworkflow/pipeline/bustard.py

   1
   2 from datetime import datetime
   3 from glob import glob
   4 import logging
   5 import os
   6 import time
   7 import re
   8
   9 from gaworkflow.pipeline.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 = datetime.now()
  70         self.user = None
  71         self.phasing = {}
  72
  73         if xml is not None:
  74             self.set_elements(xml)
  75
  76     def _get_time(self):
  77         return time.mktime(self.date.timetuple())
  78     time = property(_get_time, doc='return run time as seconds since epoch')
  79
  80     def dump(self):
  81         print "Bustard version:", self.version
  82         print "Run date", self.date
  83         print "user:", self.user
  84         for lane, tree in self.phasing.items():
  85             print lane
  86             print tree
  87
  88     def get_elements(self):
  89         root = ElementTree.Element('Bustard',
  90                                    {'version': str(Bustard.XML_VERSION)})
  91         version = ElementTree.SubElement(root, Bustard.SOFTWARE_VERSION)
  92         version.text = self.version
  93         run_date = ElementTree.SubElement(root, Bustard.DATE)
  94         run_date.text = str(self.time)
  95         user = ElementTree.SubElement(root, Bustard.USER)
  96         user.text = self.user
  97         params = ElementTree.SubElement(root, Bustard.PARAMETERS)
  98         for p in self.phasing.values():
  99             params.append(p.get_elements())
 100         return root
 101
 102     def set_elements(self, tree):
 103         if tree.tag != Bustard.BUSTARD:
 104             raise ValueError('Expected "Bustard" SubElements')
 105         xml_version = int(tree.attrib.get('version', 0))
 106         if xml_version > Bustard.XML_VERSION:
 107             logging.warn('Bustard XML tree is a higher version than this class')
 108         for element in list(tree):
 109             if element.tag == Bustard.SOFTWARE_VERSION:
 110                 self.version = element.text
 111             elif element.tag == Bustard.DATE:
 112                 self.date = datetime.fromtimestamp(float(element.text))
 113             elif element.tag == Bustard.USER:
 114                 self.user = element.text
 115             elif element.tag == Bustard.PARAMETERS:
 116                 for param in element:
 117                     p = Phasing(xml=param)
 118                     self.phasing[p.lane] = p
 119             else:
 120                 raise ValueError("Unrecognized tag: %s" % (element.tag,))
 121
 122
 123
 124 def bustard(pathname):
 125     """
 126     Construct a Bustard object from pathname
 127     """
 128     b = Bustard()
 129     path, name = os.path.split(pathname)
 130     groups = name.split("_")
 131     version = re.search(VERSION_RE, groups[0])
 132     b.version = version.group(1)
 133     b.date = datetime.strptime(groups[1], EUROPEAN_STRPTIME)
 134     b.user = groups[2]
 135     paramfiles = glob(os.path.join(pathname, "params?.xml"))
 136     for paramfile in paramfiles:
 137         phasing = Phasing(paramfile)
 138         assert (phasing.lane >= 1 and phasing.lane <= 8)
 139         b.phasing[phasing.lane] = phasing
 140     return b
 141
 142 def fromxml(tree):
 143     b = Bustard()
 144     b.set_elements(tree)
 145     return b