htsworkflow/pipelines/ipar.py

   1 """
   2 Extract information about the IPAR run
   3
   4 IPAR
   5     class holding the properties we found
   6 ipar
   7     IPAR factory function initalized from a directory name
   8 fromxml
   9     IPAR factory function initalized from an xml dump from
  10     the IPAR object.
  11 """
  12 __docformat__ = "restructuredtext en"
  13
  14 import datetime
  15 from glob import glob
  16 import logging
  17 import os
  18 import re
  19 import stat
  20 import time
  21
  22 from htsworkflow.pipelines.runfolder import \
  23    ElementTree, \
  24    VERSION_RE, \
  25    EUROPEAN_STRPTIME
  26
  27 LOGGER = logging.getLogger(__name__)
  28 SOFTWARE_NAMES = ('IPAR_1.01', 'IPAR_1.3', 'Intensities')
  29
  30 class Tiles(object):
  31   def __init__(self, tree):
  32     self.tree = tree.find("TileSelection")
  33
  34   def keys(self):
  35     key_list = []
  36     for c in self.tree.getchildren():
  37       k = c.attrib.get('Index', None)
  38       if k is not None:
  39         key_list.append(k)
  40     return key_list
  41
  42   def values(self):
  43     value_list = []
  44     for lane in self.tree.getchildren():
  45       attributes = {}
  46       for child in lane.getchildren():
  47         if child.tag == "Sample":
  48           attributes['Sample'] = child.text
  49         elif child.tag == 'TileRange':
  50           attributes['TileRange'] = (int(child.attrib['Min']),int(child.attrib['Max']))
  51       value_list.append(attributes)
  52     return value_list
  53
  54   def items(self):
  55     return zip(self.keys(), self.values())
  56
  57   def __getitem__(self, key):
  58     # FIXME: this is inefficient. building the dictionary be rescanning the xml.
  59     v = dict(self.items())
  60     return v[key]
  61
  62 class IPAR(object):
  63     XML_VERSION=1
  64
  65     # xml tag names
  66     IPAR = 'IPAR'
  67     TIMESTAMP = 'timestamp'
  68     MATRIX = 'matrix'
  69     RUN = 'Run'
  70
  71     def __init__(self, xml=None):
  72         self.tree = None
  73         self.date = datetime.datetime.today()
  74         self._tiles = None
  75         if xml is not None:
  76             self.set_elements(xml)
  77
  78     def _get_time(self):
  79         return time.mktime(self.date.timetuple())
  80     def _set_time(self, value):
  81         mtime_tuple = time.localtime(value)
  82         self.date = datetime.datetime(*(mtime_tuple[0:7]))
  83     time = property(_get_time, _set_time,
  84                     doc='run time as seconds since epoch')
  85
  86     def _get_cycles(self):
  87         if self.tree is None:
  88           raise RuntimeError("get cycles called before xml tree initalized")
  89         cycles = self.tree.find("Cycles")
  90         assert cycles is not None
  91         if cycles is None:
  92           return None
  93         return cycles.attrib
  94
  95     def _get_start(self):
  96         """
  97         return cycle start
  98         """
  99         cycles = self._get_cycles()
 100         if cycles is not None:
 101           return int(cycles['First'])
 102         else:
 103           return None
 104     start = property(_get_start, doc="get cycle start")
 105
 106     def _get_stop(self):
 107         """
 108         return cycle stop
 109         """
 110         cycles = self._get_cycles()
 111         if cycles is not None:
 112           return int(cycles['Last'])
 113         else:
 114           return None
 115     stop = property(_get_stop, doc="get cycle stop")
 116
 117     def _get_tiles(self):
 118       if self._tiles is None:
 119         self._tiles = Tiles(self.tree)
 120       return self._tiles
 121     tiles = property(_get_tiles)
 122
 123     def _get_version(self):
 124       software = self.tree.find('Software')
 125       if software is not None:
 126         return software.attrib['Version']
 127     version = property(_get_version, "IPAR software version")
 128
 129
 130     def file_list(self):
 131         """
 132         Generate list of all files that should be generated by the IPAR unit
 133         """
 134         suffix_node = self.tree.find('RunParameters/CompressionSuffix')
 135         if suffix_node is None:
 136           print "find compression suffix failed"
 137           return None
 138         suffix = suffix_node.text
 139         files = []
 140         format = "%s_%s_%04d_%s.txt%s"
 141         for lane, attrib in self.tiles.items():
 142           for file_type in ["int","nse"]:
 143             start, stop = attrib['TileRange']
 144             for tile in range(start, stop+1):
 145               files.append(format % (attrib['Sample'], lane, tile, file_type, suffix))
 146         return files
 147
 148     def dump(self):
 149         print "Matrix:", self.matrix
 150         print "Tree:", self.tree
 151
 152     def get_elements(self):
 153         attribs = {'version': str(IPAR.XML_VERSION) }
 154         root = ElementTree.Element(IPAR.IPAR, attrib=attribs)
 155         timestamp = ElementTree.SubElement(root, IPAR.TIMESTAMP)
 156         timestamp.text = str(int(self.time))
 157         root.append(self.tree)
 158         matrix = ElementTree.SubElement(root, IPAR.MATRIX)
 159         matrix.text = self.matrix
 160         return root
 161
 162     def set_elements(self, tree):
 163         if tree.tag != IPAR.IPAR:
 164             raise ValueError('Expected "IPAR" SubElements')
 165         xml_version = int(tree.attrib.get('version', 0))
 166         if xml_version > IPAR.XML_VERSION:
 167             LOGGER.warn('IPAR XML tree is a higher version than this class')
 168         for element in list(tree):
 169             if element.tag == IPAR.RUN:
 170                 self.tree = element
 171             elif element.tag == IPAR.TIMESTAMP:
 172                 self.time = int(element.text)
 173             elif element.tag == IPAR.MATRIX:
 174                 self.matrix = element.text
 175             else:
 176                 raise ValueError("Unrecognized tag: %s" % (element.tag,))
 177
 178 def load_ipar_param_tree(paramfile):
 179     """
 180     look for a .param file and load it if it is an IPAR tree
 181     """
 182
 183     tree = ElementTree.parse(paramfile).getroot()
 184     run = tree.find('Run')
 185     if run.attrib.has_key('Name') and run.attrib['Name'] in SOFTWARE_NAMES:
 186         return run
 187     else:
 188         LOGGER.info("No run found")
 189         return None
 190
 191 def ipar(pathname):
 192     """
 193     Examine the directory at pathname and initalize a IPAR object
 194     """
 195     LOGGER.info("Searching IPAR directory %s" % (pathname,))
 196     i = IPAR()
 197     i.pathname = pathname
 198
 199     # parse firecrest directory name
 200     path, name = os.path.split(pathname)
 201     groups = name.split('_')
 202     if not (groups[0] == 'IPAR' or groups[0] == 'Intensities'):
 203       raise ValueError('ipar can only process IPAR directories')
 204
 205     bustard_pattern = os.path.join(pathname, 'Bustard*')
 206     # contents of the matrix file?
 207     matrix_pathname = os.path.join(pathname, 'Matrix', 's_matrix.txt')
 208     if os.path.exists(matrix_pathname):
 209         # this is IPAR_1.01
 210         i.matrix = open(matrix_pathname, 'r').read()
 211     elif glob(bustard_pattern) > 0:
 212         i.matrix = None
 213         # its still live.
 214
 215     # look for parameter xml file
 216     paramfiles = [os.path.join(pathname, 'config.xml'),
 217                   os.path.join(path, '.params')]
 218     for paramfile in paramfiles:
 219         if os.path.exists(paramfile):
 220             LOGGER.info("Found IPAR Config file at: %s" % ( paramfile, ))
 221             i.tree = load_ipar_param_tree(paramfile)
 222             mtime_local = os.stat(paramfile)[stat.ST_MTIME]
 223             i.time = mtime_local
 224             return i
 225
 226     return i
 227
 228 def fromxml(tree):
 229     """
 230     Initialize a IPAR object from an element tree node
 231     """
 232     f = IPAR()
 233     f.set_elements(tree)
 234     return f
 235
 236 #if __name__ == "__main__":
 237   #i = ipar(os.path.expanduser('~/gec/081021_HWI-EAS229_0063_30HKUAAXX/Data/IPAR_1.01'))
 238   #x = i.get_elements()
 239   #j = fromxml(x)
 240   #ElementTree.dump(x)
 241   #print j.date
 242   #print j.start
 243   #print j.stop
 244   #print i.tiles.keys()
 245   #print j.tiles.keys()
 246   #print j.tiles.items()
 247   #print j.file_list()