1 # Copyright (c) 2011 The Chromium OS Authors. All rights reserved. 2 # Use of this source code is governed by a BSD-style license that can be 3 # found in the LICENSE file. 4 """The experiment file module. It manages the input file of crosperf.""" 5 6 from __future__ import print_function 7 import os.path 8 import re 9 from settings_factory import SettingsFactory 10 11 12 class ExperimentFile(object): 13 """Class for parsing the experiment file format. 14 15 The grammar for this format is: 16 17 experiment = { _FIELD_VALUE_RE | settings } 18 settings = _OPEN_SETTINGS_RE 19 { _FIELD_VALUE_RE } 20 _CLOSE_SETTINGS_RE 21 22 Where the regexes are terminals defined below. This results in an format 23 which looks something like: 24 25 field_name: value 26 settings_type: settings_name { 27 field_name: value 28 field_name: value 29 } 30 """ 31 32 # Field regex, e.g. "iterations: 3" 33 _FIELD_VALUE_RE = re.compile(r'(\+)?\s*(\w+?)(?:\.(\S+))?\s*:\s*(.*)') 34 # Open settings regex, e.g. "label {" 35 _OPEN_SETTINGS_RE = re.compile(r'(?:([\w.-]+):)?\s*([\w.-]+)\s*{') 36 # Close settings regex. 37 _CLOSE_SETTINGS_RE = re.compile(r'}') 38 39 def __init__(self, experiment_file, overrides=None): 40 """Construct object from file-like experiment_file. 41 42 Args: 43 experiment_file: file-like object with text description of experiment. 44 overrides: A settings object that will override fields in other settings. 45 46 Raises: 47 Exception: if invalid build type or description is invalid. 48 """ 49 self.all_settings = [] 50 self.global_settings = SettingsFactory().GetSettings('global', 'global') 51 self.all_settings.append(self.global_settings) 52 53 self._Parse(experiment_file) 54 55 for settings in self.all_settings: 56 settings.Inherit() 57 settings.Validate() 58 if overrides: 59 settings.Override(overrides) 60 61 def GetSettings(self, settings_type): 62 """Return nested fields from the experiment file.""" 63 res = [] 64 for settings in self.all_settings: 65 if settings.settings_type == settings_type: 66 res.append(settings) 67 return res 68 69 def GetGlobalSettings(self): 70 """Return the global fields from the experiment file.""" 71 return self.global_settings 72 73 def _ParseField(self, reader): 74 """Parse a key/value field.""" 75 line = reader.CurrentLine().strip() 76 match = ExperimentFile._FIELD_VALUE_RE.match(line) 77 append, name, _, text_value = match.groups() 78 return (name, text_value, append) 79 80 def _ParseSettings(self, reader): 81 """Parse a settings block.""" 82 line = reader.CurrentLine().strip() 83 match = ExperimentFile._OPEN_SETTINGS_RE.match(line) 84 settings_type = match.group(1) 85 if settings_type is None: 86 settings_type = '' 87 settings_name = match.group(2) 88 settings = SettingsFactory().GetSettings(settings_name, settings_type) 89 settings.SetParentSettings(self.global_settings) 90 91 while reader.NextLine(): 92 line = reader.CurrentLine().strip() 93 94 if not line: 95 continue 96 elif ExperimentFile._FIELD_VALUE_RE.match(line): 97 field = self._ParseField(reader) 98 settings.SetField(field[0], field[1], field[2]) 99 elif ExperimentFile._CLOSE_SETTINGS_RE.match(line): 100 return settings 101 102 raise EOFError('Unexpected EOF while parsing settings block.') 103 104 def _Parse(self, experiment_file): 105 """Parse experiment file and create settings.""" 106 reader = ExperimentFileReader(experiment_file) 107 settings_names = {} 108 try: 109 while reader.NextLine(): 110 line = reader.CurrentLine().strip() 111 112 if not line: 113 continue 114 elif ExperimentFile._OPEN_SETTINGS_RE.match(line): 115 new_settings = self._ParseSettings(reader) 116 if new_settings.name in settings_names: 117 raise SyntaxError("Duplicate settings name: '%s'." % 118 new_settings.name) 119 settings_names[new_settings.name] = True 120 self.all_settings.append(new_settings) 121 elif ExperimentFile._FIELD_VALUE_RE.match(line): 122 field = self._ParseField(reader) 123 self.global_settings.SetField(field[0], field[1], field[2]) 124 else: 125 raise IOError('Unexpected line.') 126 except Exception, err: 127 raise RuntimeError('Line %d: %s\n==> %s' % (reader.LineNo(), str(err), 128 reader.CurrentLine(False))) 129 130 def Canonicalize(self): 131 """Convert parsed experiment file back into an experiment file.""" 132 res = '' 133 board = '' 134 for field_name in self.global_settings.fields: 135 field = self.global_settings.fields[field_name] 136 if field.assigned: 137 res += '%s: %s\n' % (field.name, field.GetString()) 138 if field.name == 'board': 139 board = field.GetString() 140 res += '\n' 141 142 for settings in self.all_settings: 143 if settings.settings_type != 'global': 144 res += '%s: %s {\n' % (settings.settings_type, settings.name) 145 for field_name in settings.fields: 146 field = settings.fields[field_name] 147 if field.assigned: 148 res += '\t%s: %s\n' % (field.name, field.GetString()) 149 if field.name == 'chromeos_image': 150 real_file = ( 151 os.path.realpath(os.path.expanduser(field.GetString()))) 152 if real_file != field.GetString(): 153 res += '\t#actual_image: %s\n' % real_file 154 if field.name == 'build': 155 chromeos_root_field = settings.fields['chromeos_root'] 156 if chromeos_root_field: 157 chromeos_root = chromeos_root_field.GetString() 158 value = field.GetString() 159 autotest_field = settings.fields['autotest_path'] 160 autotest_path = '' 161 if autotest_field.assigned: 162 autotest_path = autotest_field.GetString() 163 image_path, autotest_path = settings.GetXbuddyPath(value, 164 autotest_path, 165 board, 166 chromeos_root, 167 'quiet') 168 res += '\t#actual_image: %s\n' % image_path 169 if not autotest_field.assigned: 170 res += '\t#actual_autotest_path: %s\n' % autotest_path 171 172 res += '}\n\n' 173 174 return res 175 176 177 class ExperimentFileReader(object): 178 """Handle reading lines from an experiment file.""" 179 180 def __init__(self, file_object): 181 self.file_object = file_object 182 self.current_line = None 183 self.current_line_no = 0 184 185 def CurrentLine(self, strip_comment=True): 186 """Return the next line from the file, without advancing the iterator.""" 187 if strip_comment: 188 return self._StripComment(self.current_line) 189 return self.current_line 190 191 def NextLine(self, strip_comment=True): 192 """Advance the iterator and return the next line of the file.""" 193 self.current_line_no += 1 194 self.current_line = self.file_object.readline() 195 return self.CurrentLine(strip_comment) 196 197 def _StripComment(self, line): 198 """Strip comments starting with # from a line.""" 199 if '#' in line: 200 line = line[:line.find('#')] + line[-1] 201 return line 202 203 def LineNo(self): 204 """Return the current line number.""" 205 return self.current_line_no 206
