2 Extract information about the IPAR run
5 class holding the properties we found
7 IPAR factory function initalized from a directory name
9 IPAR factory function initalized from an xml dump from
12 __docformat__ = "restructuredtext en"
22 from htsworkflow.pipelines.runfolder import \
27 LOGGER = logging.getLogger(__name__)
28 SOFTWARE_NAMES = ('IPAR_1.01', 'IPAR_1.3', 'Intensities')
31 def __init__(self, tree):
32 self.tree = tree.find("TileSelection")
36 for c in self.tree.getchildren():
37 k = c.attrib.get('Index', None)
44 for lane in self.tree.getchildren():
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)
55 return zip(self.keys(), self.values())
57 def __getitem__(self, key):
58 # FIXME: this is inefficient. building the dictionary be rescanning the xml.
59 v = dict(self.items())
67 TIMESTAMP = 'timestamp'
71 def __init__(self, xml=None):
73 self.date = datetime.datetime.today()
76 self.set_elements(xml)
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')
86 def _get_cycles(self):
88 raise RuntimeError("get cycles called before xml tree initalized")
89 cycles = self.tree.find("Cycles")
90 assert cycles is not None
99 cycles = self._get_cycles()
100 if cycles is not None:
101 return int(cycles['First'])
104 start = property(_get_start, doc="get cycle start")
110 cycles = self._get_cycles()
111 if cycles is not None:
112 return int(cycles['Last'])
115 stop = property(_get_stop, doc="get cycle stop")
117 def _get_tiles(self):
118 if self._tiles is None:
119 self._tiles = Tiles(self.tree)
121 tiles = property(_get_tiles)
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")
132 Generate list of all files that should be generated by the IPAR unit
134 suffix_node = self.tree.find('RunParameters/CompressionSuffix')
135 if suffix_node is None:
136 print "find compression suffix failed"
138 suffix = suffix_node.text
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))
149 print "Matrix:", self.matrix
150 print "Tree:", self.tree
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
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:
171 elif element.tag == IPAR.TIMESTAMP:
172 self.time = int(element.text)
173 elif element.tag == IPAR.MATRIX:
174 self.matrix = element.text
176 raise ValueError("Unrecognized tag: %s" % (element.tag,))
178 def load_ipar_param_tree(paramfile):
180 look for a .param file and load it if it is an IPAR tree
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:
188 LOGGER.info("No run found")
193 Examine the directory at pathname and initalize a IPAR object
195 LOGGER.info("Searching IPAR directory %s" % (pathname,))
197 i.pathname = pathname
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')
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):
210 i.matrix = open(matrix_pathname, 'r').read()
211 elif glob(bustard_pattern) > 0:
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]
230 Initialize a IPAR object from an element tree node
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()
244 #print i.tiles.keys()
245 #print j.tiles.keys()
246 #print j.tiles.items()