xref: /aosp_15_r20/external/toolchain-utils/crosperf/experiment_file.py (revision 760c253c1ed00ce9abd48f8546f08516e57485fe)

# -*- coding: utf-8 -*-
# Copyright 2011 The ChromiumOS Authors
# 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."""


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
               { _FIELD_VALUE_RE }
               _CLOSE_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.

        Args:
          experiment_file: file-like object with text description of experiment.
          overrides: A settings object that will override fields in other settings.

        Raises:
          Exception: if invalid build type or description is invalid.
        """
        self.all_settings = []
        self.global_settings = SettingsFactory().GetSettings("global", "global")
        self.all_settings.append(self.global_settings)

        self._Parse(experiment_file)

        for settings in self.all_settings:
            settings.Inherit()
            settings.Validate()
            if overrides:
                settings.Override(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:
                res.append(settings)
        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 = match.group(1)
        if settings_type is None:
            settings_type = ""
        settings_name = match.group(2)
        settings = SettingsFactory().GetSettings(settings_name, settings_type)
        settings.SetParentSettings(self.global_settings)

        while reader.NextLine():
            line = reader.CurrentLine().strip()

            if not line:
                continue

            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 = {}
        try:
            while reader.NextLine():
                line = reader.CurrentLine().strip()

                if not line:
                    continue

                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 new_settings.name in settings_names:
                            raise SyntaxError(
                                "Duplicate settings name: '%s'."
                                % new_settings.name
                            )
                        settings_names[new_settings.name] = True
                    self.all_settings.append(new_settings)
                elif ExperimentFile._FIELD_VALUE_RE.match(line):
                    field = self._ParseField(reader)
                    self.global_settings.SetField(field[0], field[1], field[2])
                else:
                    raise IOError("Unexpected line.")
        except Exception as err:
            raise RuntimeError(
                "Line %d: %s\n==> %s"
                % (reader.LineNo(), str(err), reader.CurrentLine(False))
            )

    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.name, field.GetString())
            if field.name == "board":
                board = field.GetString()
        res += "\n"

        for settings in self.all_settings:
            if settings.settings_type != "global":
                res += "%s: %s {\n" % (settings.settings_type, settings.name)
                for field_name in settings.fields:
                    field = settings.fields[field_name]
                    if field.assigned:
                        res += "\t%s: %s\n" % (field.name, field.GetString())
                        if field.name == "chromeos_image":
                            real_file = os.path.realpath(
                                os.path.expanduser(field.GetString())
                            )
                            if real_file != field.GetString():
                                res += "\t#actual_image: %s\n" % real_file
                        if field.name == "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