blob: d2831bda27284d04ebb29835224ffcc28c9e0226 [file] [log] [blame]
# -*- coding: utf-8 -*-
# Copyright (c) 2011 The Chromium OS Authors. All rights reserved.
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.
"""The experiment file module. It manages the input file of crosperf."""
from __future__ import print_function
import os.path
import re
from settings_factory import SettingsFactory
class ExperimentFile(object):
"""Class for parsing the experiment file format.
The grammar for this format is:
experiment = { _FIELD_VALUE_RE | settings }
settings = _OPEN_SETTINGS_RE
Where the regexes are terminals defined below. This results in an format
which looks something like:
field_name: value
settings_type: settings_name {
field_name: value
field_name: value
# Field regex, e.g. "iterations: 3"
_FIELD_VALUE_RE = re.compile(r'(\+)?\s*(\w+?)(?:\.(\S+))?\s*:\s*(.*)')
# Open settings regex, e.g. "label {"
_OPEN_SETTINGS_RE = re.compile(r'(?:([\w.-]+):)?\s*([\w.-]+)\s*{')
# Close settings regex.
_CLOSE_SETTINGS_RE = re.compile(r'}')
def __init__(self, experiment_file, overrides=None):
"""Construct object from file-like experiment_file.
experiment_file: file-like object with text description of experiment.
overrides: A settings object that will override fields in other settings.
Exception: if invalid build type or description is invalid.
self.all_settings = []
self.global_settings = SettingsFactory().GetSettings('global', 'global')
for settings in self.all_settings:
if overrides:
def GetSettings(self, settings_type):
"""Return nested fields from the experiment file."""
res = []
for settings in self.all_settings:
if settings.settings_type == settings_type:
return res
def GetGlobalSettings(self):
"""Return the global fields from the experiment file."""
return self.global_settings
def _ParseField(self, reader):
"""Parse a key/value field."""
line = reader.CurrentLine().strip()
match = ExperimentFile._FIELD_VALUE_RE.match(line)
append, name, _, text_value = match.groups()
return (name, text_value, append)
def _ParseSettings(self, reader):
"""Parse a settings block."""
line = reader.CurrentLine().strip()
match = ExperimentFile._OPEN_SETTINGS_RE.match(line)
settings_type =
if settings_type is None:
settings_type = ''
settings_name =
settings = SettingsFactory().GetSettings(settings_name, settings_type)
while reader.NextLine():
line = reader.CurrentLine().strip()
if not line:
if ExperimentFile._FIELD_VALUE_RE.match(line):
field = self._ParseField(reader)
settings.SetField(field[0], field[1], field[2])
elif ExperimentFile._CLOSE_SETTINGS_RE.match(line):
return settings, settings_type
raise EOFError('Unexpected EOF while parsing settings block.')
def _Parse(self, experiment_file):
"""Parse experiment file and create settings."""
reader = ExperimentFileReader(experiment_file)
settings_names = {}
while reader.NextLine():
line = reader.CurrentLine().strip()
if not line:
if ExperimentFile._OPEN_SETTINGS_RE.match(line):
new_settings, settings_type = self._ParseSettings(reader)
# We will allow benchmarks with duplicated settings name for now.
# Further decision will be made when parsing benchmark details in
# ExperimentFactory.GetExperiment().
if settings_type != 'benchmark':
if in settings_names:
raise SyntaxError(
"Duplicate settings name: '%s'." %
settings_names[] = True
elif ExperimentFile._FIELD_VALUE_RE.match(line):
field = self._ParseField(reader)
self.global_settings.SetField(field[0], field[1], field[2])
raise IOError('Unexpected line.')
except Exception as err:
raise RuntimeError('Line %d: %s\n==> %s' % (reader.LineNo(), str(err),
def Canonicalize(self):
"""Convert parsed experiment file back into an experiment file."""
res = ''
board = ''
for field_name in self.global_settings.fields:
field = self.global_settings.fields[field_name]
if field.assigned:
res += '%s: %s\n' % (, field.GetString())
if == 'board':
board = field.GetString()
res += '\n'
for settings in self.all_settings:
if settings.settings_type != 'global':
res += '%s: %s {\n' % (settings.settings_type,
for field_name in settings.fields:
field = settings.fields[field_name]
if field.assigned:
res += '\t%s: %s\n' % (, field.GetString())
if == 'chromeos_image':
real_file = (
if real_file != field.GetString():
res += '\t#actual_image: %s\n' % real_file
if == 'build':
chromeos_root_field = settings.fields['chromeos_root']
if chromeos_root_field:
chromeos_root = chromeos_root_field.GetString()
value = field.GetString()
autotest_field = settings.fields['autotest_path']
autotest_path = ''
if autotest_field.assigned:
autotest_path = autotest_field.GetString()
debug_field = settings.fields['debug_path']
debug_path = ''
if debug_field.assigned:
debug_path = autotest_field.GetString()
# Do not download the debug symbols since this function is for
# canonicalizing experiment file.
downlad_debug = False
image_path, autotest_path, debug_path = settings.GetXbuddyPath(
value, autotest_path, debug_path, board, chromeos_root,
'quiet', downlad_debug)
res += '\t#actual_image: %s\n' % image_path
if not autotest_field.assigned:
res += '\t#actual_autotest_path: %s\n' % autotest_path
if not debug_field.assigned:
res += '\t#actual_debug_path: %s\n' % debug_path
res += '}\n\n'
return res
class ExperimentFileReader(object):
"""Handle reading lines from an experiment file."""
def __init__(self, file_object):
self.file_object = file_object
self.current_line = None
self.current_line_no = 0
def CurrentLine(self, strip_comment=True):
"""Return the next line from the file, without advancing the iterator."""
if strip_comment:
return self._StripComment(self.current_line)
return self.current_line
def NextLine(self, strip_comment=True):
"""Advance the iterator and return the next line of the file."""
self.current_line_no += 1
self.current_line = self.file_object.readline()
return self.CurrentLine(strip_comment)
def _StripComment(self, line):
"""Strip comments starting with # from a line."""
if '#' in line:
line = line[:line.find('#')] + line[-1]
return line
def LineNo(self):
"""Return the current line number."""
return self.current_line_no