class BuildProcessor:
    """
    Build programs
    """

    block_scheme = Dict({
        "attributes": ATTRIBUTES_TYPE,
        "run_config": Dict(unknown_keys=True) // Description("Run configuration for this program block"),
        "build_config": Dict({
            "cmd": Str() // Default("") // Description("Command to build this program block, might randomize it"),
            "number": (PositiveInt() | NonExistent()) // Default(1)
                      // Description("Number of times to build this program"),
            "working_dir": (DirName() | NonExistent()) // Default(".")
                // Description("Working directory in which the build command is run"),
            "revision": (Str() | Int() | NonExistent()) // Default(-1)
                // Description("Used version control system revision of the program (-1 is the current revision)"),
            "branch": (Str() | NonExistent()) // Default("")
                // Description("Used version control system branch (default is the current branch)"),
            "base_dir": (DirName() | NonExistent()) // Default(".")
                // Description("Base directory that contains everything to build an run the program")
        }, unknown_keys=True) // Description("Build configuration for this program block")
    })   # type: Dict
    """ Type scheme of the program block configurations """

    def __init__(self, build_blocks: t.Optional[t.List[t.Dict[str, t.Any]]] = None):
        """
        Creates a build processor for the passed build block configurations.

        :param build_blocks: passed build block configurations
        """
        if build_blocks is None:
            typecheck(Settings()["build/in"], ValidYamlFileName())
            with open(Settings()["build/in"], "r") as f:
                build_blocks = yaml.safe_load(f)
        self.build_blocks = self.preprocess_build_blocks(build_blocks)
        typecheck(Settings()["build/out"], FileName())
        typecheck_locals(build_blocks=List())
        self.out = Settings()["build/out"]  # type. str
        """ Temporary directory in which the building takes place """

    @classmethod
    def preprocess_build_blocks(cls, blocks: t.List[t.Dict[str, t.Any]]) -> t.List[t.Dict[str, t.Any]]:
        """
        Pre process and check build blocks

        :return: pre processed copy
        """
        typecheck(blocks, List(cls.block_scheme))
        build_blocks = [cls.block_scheme.get_default() for _ in range(len(blocks))]
        """
        Build block configurations.
        type: t.Optional[t.List[t.Dict[str, t.Any]]]
        """
        for i, block in enumerate(blocks):
            for key in block.keys():
                build_blocks[i][key].update(block[key])
            typecheck(build_blocks[i], cls.block_scheme, "build block {}".format(i))
        return build_blocks

    def build(self):
        """
        Build the configured programs.
        """
        run_blocks = []
        try:
            for id, block in enumerate(self.build_blocks):
                error = None
                try:
                    block_builder = Builder(id, block["build_config"]["working_dir"],
                                            block["build_config"]["cmd"], block["build_config"]["revision"],
                                            block["build_config"]["number"],
                                            block["build_config"]["base_dir"], block["build_config"]["branch"])
                    working_dirs = block_builder.build()
                except BuilderKeyboardInterrupt as err:
                    working_dirs = err.result
                    error = err.error
                except BuildError as err:
                    err.log()
                    raise err
                block["run_config"]["cwd"] = working_dirs
                run_blocks.append({
                    "attributes": block["attributes"],
                    "run_config": block["run_config"]
                })

                if error:
                    raise error
        except KeyboardInterrupt as err:
            with open(self.out, "w") as f:
                yaml.dump(run_blocks, f)
            raise err
        with open(self.out, "w") as f:
            yaml.dump(run_blocks, f)

    @classmethod
    def store_example_config(cls, file: str, comment_out_defaults: bool = False):
        import click
        with click.open_file(file, "w") as f:
            print(List(cls.block_scheme).get_default_yaml(comment_out_defaults=comment_out_defaults),
                  file=f) ATTRIBUTES_TYPE = Dict({
            "tags": ListOrTuple(Str()) // Default([]) // Description("Tags of this block"),
            "description": Optional(Str()) // Default("")
        }, unknown_keys=True, key_type=Str(), value_type=Any()) \
                  // Default({"tags": []}) \
                  // Description("Optional attributes that describe the block") class Settings(metaclass=Singleton):
    """
    Manages the Settings.
    The settings keys and sub keys are combined by a slash, e.g. "report/in".

    The current settings are:

    .. code: yaml


    """

    config_file_name = "temci.yaml"  # type: str
    """ Default name of the configuration files """
    type_scheme = Dict({
        "settings": Str() // Description("Additional settings file")
                        // Default(config_file_name if os.path.exists(config_file_name) else "")
                        // CompletionHint(zsh=YAML_FILE_COMPLETION_HINT),
        "config": Str() // Description("Alias for settings")
                    // Default(config_file_name if os.path.exists(config_file_name) else "")
                    // CompletionHint(zsh=YAML_FILE_COMPLETION_HINT),
        "tmp_dir": Str() // Default("/tmp/temci") // Description("Used temporary directory"),
        "log_level": ExactEither("debug", "info", "warn", "error", "quiet") // Default("info")
                     // Description("Logging level"),
        "stats": Dict({
            "properties": ListOrTuple(Str()) // Default(["all"])
                        // CompletionHint(zsh="(" + " ".join(["__ov-time", "cache-misses", "cycles", "task-clock",
                                                              "instructions", "branch-misses", "cache-references",
                                                              "all"])
                                              + ")")
                        // Description("Properties to use for reporting and null hypothesis tests, "
                                       "can be regular expressions"),
            "uncertainty_range": Tuple(Float(lambda x: x >= 0), Float(lambda x: x >= 0)) // Default([0.05, 0.15])
                        // Description("Range of p values that allow no conclusion.")
        }, unknown_keys=True),
        "report": Dict({
            #  "reporter": Str() // Default("console") // Description(),
            "in": Either(Str(), ListOrTuple(Str())) // Default("run_output.yaml")
                  // Description("Files that contain the benchmarking results")
                  // CompletionHint(zsh=YAML_FILE_COMPLETION_HINT),
            "excluded_properties": ListOrTuple(Str()) // Default(["__ov-time"])
                    // Description("Properties that aren't shown in the report."),
            "exclude_invalid": BoolOrNone() // Default(True)
                    // Description("Exclude all data sets that contain only NaNs."),
            "long_properties": BoolOrNone() // Default(False)
                    // Description("Replace the property names in reports with longer more descriptive versions?"),
            "xkcd_like_plots": BoolOrNone() // Default(False)
                    // Description("Produce xkcd like plots (requires the humor sans font to be installed)"),
            "number": FNumber.settings_format,
            "included_blocks": ListOrTuple(Str()) // Default(["all"])
                               // Description("List of included run blocks (all: include all), "
                                              "identified by their description or tag attribute, "
                                              "can be regular expressions"),
        }, unknown_keys=True),
        "run": Dict({
            "discarded_runs": NaturalNumber() // Description("First n runs that are discarded") // Default(1),
            "min_runs": NaturalNumber() // Default(20) // Description("Minimum number of benchmarking runs"),
            "max_runs": NaturalNumber() // Default(100) // Description("Maximum number of benchmarking runs"),
            "max_runs_per_tag": Dict(unknown_keys=True, key_type=Str() // Description("Tag"), value_type=NaturalNumber() // Description("Max runs"))
                                 // Default({}) // Description("Maximum runs per tag (block attribute 'tag'), min('max_runs', 'per_tag') is used"),
            "min_runs_per_tag": Dict(unknown_keys=True, key_type=Str() // Description("Tag"),
                                     value_type=NaturalNumber() // Description("Min runs"))
                                // Default({}) // Description(
                                "Minimum runs per tag (block attribute 'tag'), max('min_runs', 'per_tag') is used"),
            "runs_per_tag": Dict(unknown_keys=True, key_type=Str() // Description("Tag"),
                                     value_type=NaturalNumber() // Description("Runs"))
                                // Default({}) // Description(
                                 "Runs per tag (block attribute 'tag'), max('runs', 'per_tag') is used"),
            "runs": Int(lambda x: x >= -1) // Default(-1) // Description("if != -1 sets max and min runs to its value"),
            "max_time": ValidTimeSpan() // Default("-1") // Description("Maximum time the whole benchmarking should take, "
                                                                        "-1 == no timeout, supports normal time span expressions"), # in seconds
            "max_block_time": ValidTimeSpan() // Default("-1") // Description("Maximum time one run block should take, "
                                                                              "-1 == no timeout, supports normal time span expressions"),
            "run_block_size": PositiveInt() // Default(1)
                              // Description("Number of benchmarking runs that are done together"),
            "in": Str() // Default("input.exec.yaml")
                  // Description("Input file with the program blocks to benchmark")
                  // CompletionHint(zsh=YAML_FILE_COMPLETION_HINT),
            "out": Str() // Default("run_output.yaml") // Description("Output file for the benchmarking results")
                    // CompletionHint(zsh=YAML_FILE_COMPLETION_HINT),
            "store_often": Bool() // Default(False)
                           // Description("Store the result file after each set of blocks is benchmarked"),
            "exec_plugins": Dict({

            }),
            "included_blocks" : ListOrTuple(Str()) // Default(["all"])
                              // Description("List of included run blocks (all: include all), "
                                             "or their tag attribute "
                                             "or their number in the file (starting with 0), "
                                             "can be regular expressions"),
            "cpuset": Dict({
                "active": Bool() // Description("Use cpuset functionality?") // Default(False),
                "base_core_number": ValidCPUCoreNumber()
                                    // Description("Number of cpu cores for the base (remaining part of the) system") // Default(1),
                "parallel": Int(lambda x: x >= -1) // Description("0: benchmark sequential, "
                                                      "> 0: benchmark parallel with n instances, "
                                                      "-1: determine n automatically") // Default(0),
                "sub_core_number": ValidCPUCoreNumber() // Description("Number of cpu cores per parallel running program.")
                                   // Default(1),
                "temci_in_base_set": Bool() // Default(True)
                                     // Description("place temci in the same cpu set as the rest of the system?")
            }),
            "disable_hyper_threading": Bool() // Default(False)
                                       // Description("Disable the hyper threaded cores. Good for cpu bound programs."),
            "show_report": Bool() // Default(True)
                // Description("Print console report if log_level=info"),
            "append": Bool() // Default(False)
                      // Description("Append to the output file instead of overwriting by adding new run data blocks"),
            "shuffle": Bool() // Default(True) // Description("Randomize the order in which the program blocks are "
                                                              "benchmarked."),
            "send_mail": Str() // Default("")
                         // Description("If not empty, recipient of a mail after the benchmarking finished."),
            "discard_all_data_for_block_on_error": Bool() // Default(False)
                         // Description("Discard all run data for the failing program on error"),
            "record_errors_in_file": Bool() // Default(True)
                         // Description("Record the caught errors in the run_output file"),
            "no_build": Bool() // Default(False)
                         // Description("Do not build if build configs are present, only works if the working directory "
                                        "of the blocks does not change"),
            "only_build": Bool() // Default(False) // Description("Only build"),
            "abort_after_build_error": Bool() // Default(True)
                                    // Description("Abort after the first failing build"),
            "watch": Bool() // Default(False) // Description("Show the report continuously"),
            "watch_every": PositiveInt() // Default(1)
                            // Description("Update the screen nth run (less updates are better for benchmarks)")
        }),
        "build": Dict({
            "in": Str() // Default("build.yaml") // Description("Input file with the program blocks to build")
                // CompletionHint(zsh=YAML_FILE_COMPLETION_HINT),
            "out": Str() // Default("run_config.yaml") // Description("Resulting run config file"),
            "threads": PositiveInt() // Default(1) // Description("Number of simultaneous builds for a specific "
                                                                  "program block, only makes sense when build_config/number > 1, "
                                                                  "and if the build commands create a different binary every "
                                                                  "time they are run"),
            "rand": Obsolete("Removed builder randomization", "0.8") // Description("Obsolete randomization configuration")
        }, unknown_keys=True),
        "package": Obsolete("Removed temci package", "0.8"),
        "env": Dict({"USER": Str(), "PATH": Str()}, unknown_keys=True)
               // Default({"USER": "", "PATH": ""})
               // Description("Environment variables for the benchmarked programs, includes the user used for "
                              "generated files"),
        "sudo": Bool() // Default(False) // Description("Acquire sudo privileges and run benchmark programs with "
                                                        "non-sudo user. Only supported on the command line.")
    }, unknown_keys=True)  # type: Dict
    """ Type scheme of the settings """

    def __init__(self):
        """
        Initializes a Settings singleton object and thereby loads the Settings files.
        It loads the settings files from the app folder (config.yaml) and
        the current working directory (temci.yaml) if they exist.

        :raises: SettingsError if some of the settings aren't in the format described via the type_scheme class property
        """
        self.prefs = copy.deepcopy(self.type_scheme.get_default())  # type: t.Dict[str, t.Any]
        """ The set sonfigurations """
        res = self._validate_settings_dict(self.prefs, "default settings")
        if not res:
            raise SettingsError(str(res))
        self._setup()

    def load_files(self):
        """ Loads the configuration files from the current and the config directory """
        self.load_from_config_dir()
        self.load_from_current_dir()
        self._setup()

    def _setup(self):
        """
        Simple setup method that checks if basic directories exist and creates them if necessary.
        """
        if not os.path.exists(self.prefs["tmp_dir"]):
            os.mkdir(self.prefs["tmp_dir"])
        log_level = self["log_level"]
        logging.Logger.disabled = log_level == "quiet"
        logger = logging.getLogger()
        mapping = {
            "debug": logging.DEBUG,
            "info": logging.INFO,
            "warn": logging.WARNING,
            "error": logging.ERROR,
            "quiet": logging.ERROR
        }
        logger.setLevel(mapping[log_level])
        self._update_doc()
        self.apply_override_actions()

    def _update_doc(self):
        """
        Update the class documentation
        """
        if sphinx_doc():
            self.__doc__ = self.__doc__.split(".. code: yaml")[0] + """.. code: yaml 

        """ + "\n        ".join(self.type_scheme.get_default_yaml().split("\n"))

    def reset(self):
        """
        Resets the current settings to the defaults.
        """
        self.prefs = copy.deepcopy(self.type_scheme.get_default())

    def _validate_settings_dict(self, data: t.Dict[str, t.Any], description: str = None):
        """
        Check whether the passed dictionary matches the settings type scheme.

        :param data: passed dictionary
        :param description: short description of the passed dictionary
        :return: True like object if valid, else string like object which is the error message
        """
        return verbose_isinstance(data, self.type_scheme, description or "Settings")

    def load_file(self, file: str):
        """
        Loads the configuration from the configuration YAML file.

        :param file: path to the file
        :raises: SettingsError if the settings file is incorrect or doesn't exist
        """
        self.prefs = self.type_scheme.get_default()
        tmp = copy.deepcopy(self.prefs)
        try:
            with open(file, 'r') as stream:
                map = yaml.safe_load(stream.read().replace("!!python/tuple", ""))

                def func(key, path, value):
                    if value is not None or self.get_type_scheme(path).check(value):
                        self._set_default(path, value)
                        self._set(path, value)

                recursive_exec_for_leafs(map, func)
        except (yaml.YAMLError, IOError) as err:
            self.prefs = tmp
            raise SettingsError(str(err))
        res = self._validate_settings_dict(self.prefs, "settings with ones from file '{}'".format(file))
        if not res:
            self.prefs = tmp
            raise SettingsError(str(res))
        self._setup()

    def load_from_dict(self, config_dict: t.Dict[str, t.Any]):
        """
        Load the configuration from the passed dictionary.

        :param config_dict: passed configuration dictionary
        """
        self.prefs = self.type_scheme.get_default()
        tmp = copy.deepcopy(self.prefs)

        def func(key, path, value):
            self._set_default(path, value)

        recursive_exec_for_leafs(config_dict, func)
        res = self._validate_settings_dict(self.prefs, "settings with ones config dict")
        if not res:
            self.prefs = tmp
            raise SettingsError(str(res))
        self._setup()

    def load_from_dir(self, dir: str):
        """
        Load the configuration from the configuration file inside the passed directory.

        :param dir: path of the directory
        """
        self.load_file(os.path.join(dir, "config.yaml"))

    def load_from_config_dir(self):
        """
        Load the config file from the application directory (e.g. in the users home folder) if it exists.
        """
        conf = os.path.join(click.get_app_dir("temci"), "config.yaml")
        if os.path.exists(conf) and os.path.isfile(conf):
            self.load_file(conf)

    def load_from_current_dir(self):
        """
        Load the configuration from the `configuration file in the current working directory if it exists.
        """
        if os.path.exists(self.config_file_name) and os.path.isfile(self.config_file_name):
            self.load_file(self.config_file_name)

    def get(self, key: t.Union[str, t.List[str]]) -> t.Any:
        """
        Get the setting with the given key.

        :param key: name of the setting
        :return: value of the setting
        :raises: SettingsError if the setting doesn't exist
        """
        if self.is_obsolete(key):
            raise SettingsError("Using obsolete setting {!r}: {}".format(key, self.obsoleteness_reason(key)))
        path = key.split("/") if isinstance(key, str) else key
        if not self.validate_key_path(path):
            raise SettingsError("No such setting {}".format(key))
        data = self.prefs
        for sub in path:
            data = data[sub]
        return data

    def __getitem__(self, key: str) -> t.Any:
        """
        Alias for self.get(self, key).
        """
        return self.get(key)

    def _set(self, path: t.List[str], value):
        """
        Set the setting at the passed path.

        :param path: passed key path
        :param value: new value
        """
        if self.is_obsolete(path):
            return
        tmp_pref = self.prefs
        tmp_type = self.type_scheme
        for key in path[0:-1]:
            if key not in tmp_pref:
                tmp_pref[key] = {}
                tmp_type[key] = Dict(unknown_keys=True, key_type=Str())
            tmp_pref = tmp_pref[key]
            tmp_type = tmp_type[key]
        tmp_pref[path[-1]] = value
        if path[-1] not in tmp_type.data:
            tmp_type[path[-1]] = Any() // Default(value)
        if (path == ["config"] or path == ["settings"]) and value != "":
            self.load_file(value)
        self._update_doc()

    def validate(self):
        """
        Validate this settings object

        :raises: SettingsError if the setting isn't valid
        """
        self._validate_settings_dict(self.prefs)

    def set(self, key: str, value, validate: bool = True, setup: bool = True):
        """
        Sets the setting key to the passed new value

        :param key: settings key
        :param value: new value
        :param validate: validate after the setting operation
        :param setup: call the setup function
        :raises: SettingsError if the setting isn't valid
        """
        tmp = copy.deepcopy(self.prefs)
        path = key.split("/")
        self._set(path, value)
        if validate:
            res = self._validate_settings_dict(self.prefs, "settings with new setting ({}={!r})".format(key, value))
            if not res:
                self.prefs = tmp
                raise SettingsError(str(res))
        if setup:
            self._setup()

    def __setitem__(self, key: str, value):
        """
        Alias for self.set(key, value).
        """
        self.set(key, value)

    def validate_key_path(self, path: t.List[str]) -> bool:
        """
        Validates a path into in to the settings trees,
        :param path: list of sub keys
        :return: Is this key path valid?
        """
        if self.is_obsolete(path):
            return True
        tmp = self.prefs
        for item in path:
            if item not in tmp:
                return False
            tmp = tmp[item]
        return True

    def has_key(self, key: str) -> bool:
        """ Does the passed key exist? """
        return self.validate_key_path(key.split("/"))

    def _set_default(self, path: t.List[str], value):
        """
        Set the default value of the setting with the passed path

        :param path: passed key path
        :param value: new default value
        """
        self.modify_type_scheme("/".join(path), lambda t: t // Default(value))
        self._update_doc()

    def modify_setting(self, key: str, type_scheme: Type):
        """
        Modifies the setting with the given key and adds it if it doesn't exist.

        :param key: key of the setting
        :param type_scheme: Type of the setting
        :param default_value: default value of the setting
        :raises: SettingsError if the settings domain (the key without the last element) doesn't exist
        :raises: TypeError if the default value doesn't adhere the type scheme
        """
        if self.is_obsolete(key):
            logging.info("Using obsolete setting {!r}: {}".format(key, self.obsoleteness_reason(key)))
            return
        path = key.split("/")
        domain = "/".join(path[:-1])
        if len(path) > 1 and not self.validate_key_path(path[:-1]) \
                and not isinstance(self.get(domain), dict):
            raise SettingsError("Setting domain {} doesn't exist".format(domain))
        tmp_typ = self.type_scheme
        tmp_prefs = self.prefs
        for subkey in path[:-1]:
            tmp_typ = tmp_typ[subkey]
            tmp_prefs = tmp_prefs[subkey]
        tmp_typ[path[-1]] = type_scheme
        if path[-1] in tmp_prefs:
            if type_scheme.typecheck_default:
                typecheck(tmp_prefs[path[-1]], type_scheme)
            tmp_typ[path[-1]] = type_scheme
        else:
            tmp_prefs[path[-1]] = type_scheme.get_default()

    def get_type_scheme(self, key: t.Union[str, t.List[str]]) -> Type:
        """
        Returns the type scheme of the given key.

        :param key: given key
        :return: type scheme
        :raises: SettingsError if the setting with the given key doesn't exist
        """
        key = key.split("/") if isinstance(key, str) else key
        if not self.validate_key_path(key):
            raise SettingsError("Setting {} doesn't exist".format("/".join(key)))
        tmp_typ = self.type_scheme
        for subkey in key:
            tmp_typ = tmp_typ[subkey]
        return tmp_typ

    def modify_type_scheme(self, key: str, modificator: t.Callable[[Type], Type]):
        """
        Modifies the type scheme of the given key via a modificator function.

        :param key: given key
        :param modificator: gets the type scheme and returns its modified version
        :raises: SettingsError if the setting with the given key doesn't exist
        """
        if self.is_obsolete(key):
            return
        if not self.validate_key_path(key.split("/")):
            raise SettingsError("Setting {} doesn't exist".format(key))
        tmp_typ = self.type_scheme
        subkeys = key.split("/")
        for subkey in subkeys[:-1]:
            tmp_typ = tmp_typ[subkey]
        tmp_typ[subkeys[-1]] = modificator(tmp_typ[subkeys[-1]])
        assert isinstance(tmp_typ[subkeys[-1]], Type)

    def default(self, value: t.Optional[t.Any], key: str):
        """
        Returns the passed value if isn't None else the settings value under the passed key.

        :param value: passed value
        :param key: passed settings key
        """
        if value is None:
            return self[key]
        typecheck(value, self.get_type_scheme(key))
        return value

    def store_into_file(self, file_name: str, comment_out_defaults: bool = False):
        """
        Stores the current settings into a yaml file with comments.

        :param file_name: name of the resulting file
        :param comment_out_defaults: comment out the default values
        """
        with open(file_name, "w") as f:
            print(self.type_scheme.get_default_yaml(defaults=self.prefs,
                                                    comment_out_defaults=comment_out_defaults), file=f)

    def has_log_level(self, level: str) -> bool:
        """
        Is the current log level the passed level?

        :param level: passed level (in ["error", "warn", "info", "debug"])
        """
        levels = ["error", "warn", "info", "debug"]
        return levels.index(level) <= levels.index(self["log_level"])

    def is_obsolete(self, key: t.Union[str, t.List[str]]) -> bool:
        """
        Is the setting with the passed key obsolete?

        :param key: key or key path
        :return: obsolete setting?
        """
        return self.obsoleteness_reason(key) is not None

    def obsoleteness_reason(self, key: t.Union[str, t.List[str]]) -> t.Optional[Obsolete]:
        """
        Returns the obsolete type object for obsolete settings

        :param key: key or path
        :return: object that contains information on the obsoleteness or None
        """
        path = key.split("/") if isinstance(key, str) else key
        tmp_type = self.type_scheme
        for subkey in path[:-1]:
            if tmp_type.is_obsolete(subkey):
                return tmp_type.obsoleteness_reason(subkey)
            if subkey not in tmp_type:
                return None
            if isinstance(tmp_type[subkey], Obsolete):
                return tmp_type[subkey]
            tmp_type = tmp_type[subkey]
        if path[-1] in tmp_type and isinstance(tmp_type[path[-1]], Obsolete):
            return tmp_type[subkey]
        return None

    def apply_override_actions(self):
        """ Applies actions like overriding max_runs with runs """
        if self["run/runs"] > -1:
            self.set("run/max_runs", self["run/runs"], setup=False)
            self.set("run/min_runs", self["run/runs"], setup=False) def get_bench_user() -> str:
    user = Settings()["env"]["USER"]
    return os.getenv("USER", "") if user == "" else user def bench_as_different_user() -> bool:
    return get_bench_user() != os.getenv("USER", get_bench_user()) def get_env_setting() -> Dict[str, str]:
    env = Settings()["env"].copy()
    if env["USER"] == "":
        env["USER"] = get_bench_user()
    if env["PATH"] == "":
        env["PATH"] = os.getenv("PATH", "")
    return env class NoInfo(Info):
    """
    A dumb version of the information class that is used when meaningful error messages aren't needed.
    It has better performance characteristics as it doesn't store any values or create strings.
    """

    def __init__(self, value_name: str = None, _app_str: str = None, value=None):
        if False:
            super().__init__(value_name, _app_str, value)
        self.has_value = True

    def get_value(self) -> None:
        return None

    def set_value(self, value):
        pass

    def add_to_name(self, app_str: str) -> 'NoInfo':
        return self

    def errormsg(self, constraint: 'Type', *args) -> 'InfoMsg':
        return InfoMsg(False)

    def errormsg_cond(self, cond: bool, *args) -> 'InfoMsg':
        return InfoMsg(cond)

    def errormsg_key_non_existent(self, constraint: 'Type', key: str) -> 'InfoMsg':
        return InfoMsg(False)

    def errormsg_unexpected(self, key: str) -> 'InfoMsg':
        return InfoMsg(False)

    def wrap(self, result: bool) -> 'InfoMsg':
        return InfoMsg(result) def has_root_privileges() -> bool:
    """
    Has the current user root privileges?
    """
    return does_command_succeed("head /proc/1/stack") def join_strs(strs: t.List[str], last_word: str = "and") -> str:
    """
    Joins the passed strings together with ", " except for the last to strings that separated by the passed word.

    :param strs: strings to join
    :param last_word: passed word that is used between the two last strings
    """
    if not isinstance(strs, list):
        strs = list(strs)
    if len(strs) == 1:
        return strs[0]
    elif len(strs) > 1:
        return " {} ".format(last_word).join([", ".join(strs[0:-1]), strs[-1]]) def does_command_succeed(cmd: str) -> bool:
    """ Does the passed command succeed (when executed by /bin/sh)?  """
    try:
        subprocess.check_call(["/bin/sh", "-c", cmd], stdout=subprocess.DEVNULL,
                              stderr=subprocess.DEVNULL)
    except:
        return False
    return True def sphinx_doc() -> bool:
    """ Is the code only loaded to document it with sphinx? """
    return _sphinx_doc def on_apple_os() -> bool:
    """ Is the current operating system an apple OS X? """
    return sys.platform == 'darwin' def does_program_exist(program: str) -> bool:
    """ Does the passed program exist? """
    return shutil.which(program) is not None def document(**kwargs: t.Dict[str, str]):
    """
    Document

    :param kwargs: class attribute, documentation prefix
    """

    def dec(klass):
        if sphinx_doc():
            for k, v in kwargs.items():
                klass.__doc__ += """

    {}

    {}

                """.format(v, get_doc_for_type_scheme(klass.__dict__[k]))
        return klass

    return dec class proc_wait_with_rusage:
    """
    Each Popen object gets a field rusage
    """

    def __enter__(self):
        self.rusage = None
        self.old_try_wait = subprocess.Popen._try_wait

        def try_wait(self, wait_flags):
            """ Copied from subprocess._try_wait"""
            try:
                (pid, sts, _u) = os.wait4(self.pid, wait_flags)
                self.rusage = _u
                # instead of: (pid, sts) = os.waitpid(self.pid, wait_flags)
            except ChildProcessError:
                # This happens if SIGCLD is set to be ignored or waiting
                # for child processes has otherwise been disabled for our
                # process.  This child is dead, we can't get the status.
                pid = self.pid
                sts = 0
            return (pid, sts)

        subprocess.Popen._try_wait = try_wait

    def __exit__(self, exc_type, exc_val, exc_tb):
        subprocess.Popen._try_wait = self.old_try_wait @functools.lru_cache()
def rusage_header() -> str:
    from ..setup.setup import script_relative
    return open(script_relative("rusage/header.c")).read().split("\"")[1] class VCSDriver:
    """
    Abstract version control system driver class used to support different vcss.
    """

    id_type = Str()|Int()

    def __init__(self, dir: str = ".", branch: str = None):
        """
        Initializes the VCS driver for a given base directory.
        It also sets the current branch if it's defined in the Settings

        :param dir: base directory
        :param branch: used branch
        """
        typecheck_locals(dir=Str(), branch=Optional(Str()))
        self._exec_command_cache = {}
        self._exec_err_code_cache = {}
        self.dir = os.path.abspath(dir)  # type: str
        """ Base directory """
        self.branch = branch or self.get_branch()  # type: str
        """ Used branch """

    @classmethod
    def get_suited_vcs(cls, mode="auto", dir=".", branch: str = None) -> 'VCSDriver':
        """
        Chose the best suited vcs driver for the passed base directory and the passed mode.
        If mode is "auto" the best suited vcs driver is chosen. If mode is "git" or "file",
        the GitDriver or the FileDriver is chosen. If the chosen driver isn't applicable than
        a VCSError is raised.

        :param mode: passed mode
        :param dir: base directory
        :param branch: used branch
        :return: vcs driver for the base directory
        :raises: VCSError if the selected driver isn't applicable
        """
        typecheck_locals(mode=ExactEither("file", "git", "auto"), dir=Str(), branch=Optional(Str()))
        if mode == "file" and FileDriver.is_suited_for_dir(dir):
            return FileDriver(dir, branch)
        elif mode == "git" and GitDriver.is_suited_for_dir(dir):
            return GitDriver(dir, branch)
        elif mode == "auto" and FileDriver.is_suited_for_dir(dir):
            avcls = [cls for cls in [GitDriver, FileDriver] if cls.is_suited_for_dir(dir)]
            return avcls[0](dir, branch)
        else:
            raise NoSuchVCSError("No such vcs driver for mode {0} and directory {1}".format(mode, dir))

    @classmethod
    def is_suited_for_dir(cls, dir: str = ".") -> bool:
        """
        Checks whether or not this vcs driver can work with the passed base directory.

        :param dir: passed base directory path
        """
        raise NotImplementedError()

    def set_branch(self, new_branch: str):
        """
        Sets the current branch and throws an error if the branch doesn't exist.

        :param new_branch: new branch to set
        :raises: VCSError if new_branch doesn't exist
        """
        raise NotImplementedError()

    def get_branch(self) -> t.Optional[str]:
        """
        Gets the current branch.

        :return: current branch name
        :raises: VCSError if something goes terribly wrong
        """
        raise None

    def get_valid_branches(self) -> t.Optional[t.List[str]]:
        """
        Gets the valid branches for the associated repository or None if the vcs doesn't support branches.
        """
        return None

    def has_uncommitted(self) -> bool:
        """
        Check for uncommitted changes in the repository.
        """
        raise NotImplementedError()

    def number_of_revisions(self) -> int:
        """
        Number of committed revisions in the current branch (if branches are supported).
        :return: number of revisions
        """
        raise NotImplementedError()

    def validate_revision(self, id_or_num: t.Union[int, str]) -> bool:
        """
        Validate the existence of the referenced revision.

        :param id_or_num: id or number of the reverenced revision
        :return: does it exists?
        """
        raise NotImplementedError()

    def get_info_for_revision(self, id_or_num: t.Union[int, str]) -> dict:
        """
        Get an info dict for the given commit (-1 and 'HEAD' represent the uncommitted changes).
        Structure of the info dict::

            "commit_id"; …,
            "commit_message": …,
            "commit_number": …,
            "is_uncommitted": True/False,
            "is_from_other_branch": True/False,
            "branch": … # branch name or empty string if this commit belongs to no branch

        :param id_or_num: id or number of the commit
        :return: info dict
        :raises: VCSError if the number or id isn't valid
        """
        raise NotImplementedError()

    def get_info_for_all_revisions(self, max: int = -1) -> t.List[t.Dict[str, t.Any]]:
        """
        Get an info dict for all revisions.
        A single info dict has the following structure::

            "commit_id"; …,
            "commit_message": …,
            "commit_number": …,
            "is_uncommitted": True/False,
            "is_from_other_branch": True/False,
            "branch": … # branch name or empty string if this commit belongs to no branch

        :param max: if max isn't -1 it gives the maximum number of revision infos returned
        :return: list of info dicts
        """
        info_dicts = []
        if self.has_uncommitted() and (max >= 1 or max == -1):
            info_dicts.append(self.get_info_for_revision(-1))
            if max != -1:
                max -= 1
        num = self.number_of_revisions()
        if max != -1 and max < num:
            num = max
        for i in range(num):
            info_dicts.append(self.get_info_for_revision(i))

        return info_dicts


    def copy_revision(self, id_or_num: t.Union[int, str], sub_dir: str, dest_dirs: t.List[str]):
        """
        Copy the sub directory of the current vcs base directory into all of the destination directories.

        :param id_or_num: id or number of the revision (-1 and 'HEAD' represent the uncommitted changes)
        :param sub_dir: sub directory of the current vcs base directory relative to it
        :param dest_dirs: list of destination directories in which the content of the sub dir is placed or dest dir string
        :raises: VCSError if something goes wrong while copying the directories
        """
        raise NotImplementedError()

    def _copy_dir(self, src_dir: str, dest_dirs: t.List[str]):
        """
        Helper method to copy a directory too many destination directories.
        It also works if for files.

        :param src_dir: source directory relative to the current base directory
        :param dest_dirs: list of destination directories or just one destination directory string
        """
        typecheck_locals(src_dir=Str(), dest_dirs=List(Str())|Str())
        src_dir_path = os.path.abspath(os.path.join(self.dir, src_dir))
        dest_dir_paths = []
        if type(dest_dirs) is str:
            dest_dir_paths = [os.path.abspath(dest_dirs)]
        else:
            dest_dir_paths = [os.path.abspath(dest) for dest in dest_dirs]
        for dest in dest_dir_paths:
            try:
                shutil.rmtree(dest)
                shutil.copytree(src_dir_path, dest)
            except OSError as exc:
                try:
                    if exc.errno == errno.ENOTDIR:
                        shutil.copy(src_dir_path, dest)
                    else:
                        raise
                except OSError as exc2:
                    raise VCSError(str(exc2))

    def _exec_command(self, command: str, error: str = "Error executing {cmd}: {err}", cacheable: bool = False):
        """
        Executes the given external command and returns the resulting output.

        :param command: given external command, list or string (uses /bin/sh)
        :param error: error message with can have a placeholder `cmd` for the command and `èrr` for stderr
        :param cacheable: can the result of the command be cached to reduce the number of needed calls?
        :return: output as string
        :raises: VCSError if the external command hasn't exit code 0
        """
        typecheck_locals(command=List()|Str(), error=Str(), cacheable=Bool())
        args = command
        if isinstance(command, Str()):
            args = ["/bin/sh", "-c", command]

        args_str = "#~#".join(args)
        if cacheable and args_str in self._exec_command_cache:
            return self._exec_command_cache[args_str]
        proc = subprocess.Popen(args, cwd=abspath(self.dir), stdout=subprocess.PIPE,
                            stderr=subprocess.PIPE,
                            universal_newlines=True)
        out, err = proc.communicate()
        if proc.poll() > 0:
            msg = error.format(cmd=command, err=err)
            raise VCSError(msg)

        if cacheable:
            self._exec_command_cache[args_str] = str(out)

        return str(out)

    def _exec_err_code(self, command: str, cacheable: bool = False):
        """
        Executes the given external command and returns its error code.

        :param command: given external command (as string or list)
        :param cacheable: can the result of the command be cached to reduce the number of needed calls?
        :return: error code of the command (or 0 if no error occurred)
        """
        typecheck_locals(command=List(Str())|Str(), cacheable=Bool())
        args = []
        if isinstance(command, list):
            args = command
        else:
            args = ["/bin/sh", "-c", command]

        args_str = "#~#".join(args)
        if cacheable and args_str in self._exec_err_code_cache:
            return self._exec_err_code_cache[args_str]

        proc = subprocess.Popen(args, cwd=abspath(self.dir), universal_newlines=True,
                                stderr=subprocess.DEVNULL, stdout=subprocess.DEVNULL)
        out, err = proc.communicate()

        err_code = proc.poll()

        if cacheable:
            self._exec_err_code_cache[args_str] = err_code

        return err_code class CPUSet:
    """
    This class allows the usage of cpusets (see `man cpuset`) and therefore requires root privileges.
    It uses the program cset to modify the cpusets.
    This class needs root privileges to operate properly. Warns if not.
    """

    def __init__(self, active: bool = True, base_core_number: int = None,
                 parallel: int = None, sub_core_number: int = None,
                 temci_in_base_set: bool = True):
        """
        Initializes the cpu sets an determines the number of parallel programs (parallel_number variable).

        :param active: are cpu sets actually used?
        :param base_core_number: number of cpu cores for the base (remaining part of the) system
        :param parallel: 0: benchmark sequential, > 0: benchmark parallel with n instances, -1: determine n automatically
        :param sub_core_number: number of cpu cores per parallel running program
        :param temci_in_base_set: place temci in the same cpu set as the rest of the system?
        :raises ValueError: if the passed parameters don't work together on the current platform
        :raises EnvironmentError: if the environment can't be setup properly (e.g. no root privileges)
        """
        # self.bench_set = "bench.set"
        self.active = active and has_root_privileges()  # type: bool
        """ Are cpu sets actually used? """
        self.base_core_number = Settings().default(base_core_number, "run/cpuset/base_core_number")  # type: int
        """ Number of cpu cores for the base (remaining part of the) system """
        self.parallel = Settings().default(parallel, "run/cpuset/parallel")  # type: int
        """ 0: benchmark sequential, > 0: benchmark parallel with n instances, -1: determine n automatically """
        self.sub_core_number = Settings().default(sub_core_number, "run/cpuset/sub_core_number")  # type: int
        """ Number of cpu cores per parallel running program """
        self.av_cores = len(self._cpus_of_set("")) if active else multiprocessing.cpu_count()  # type: int
        """ Number of available cpu cores """
        self.parallel_number = 0  # type: int
        """ Number of used parallel instances, zero if the benchmarking is done sequentially """
        self.temci_in_base_set = Settings().default(temci_in_base_set, "run/cpuset/temci_in_base_set")
        """ Place temci in the same cpu set as the rest of the system? """
        if self.parallel != 0:
            if self.parallel == -1:
                self.parallel_number = self._number_of_parallel_sets(self.base_core_number,
                                                                     True, self.sub_core_number)
            else:
                self.parallel_number = self.parallel
                if self.parallel > self._number_of_parallel_sets(self.base_core_number, True, self.sub_core_number) \
                        and self.active:
                    raise ValueError("Invalid values for base_core_number and sub_core_number "
                                     "on system with just {} cores. Note: The benchmark controller "
                                     "needs a cpuset too.".format(self.av_cores))
            self.base_core_number = self.av_cores - self.sub_core_number * self.parallel_number - 1
        if not self.active:
            if active and not has_root_privileges():
                logging.warning("CPUSet functionality is disabled because root privileges are missing.")
            return
        logging.info("Initialize CPUSet")
        typecheck(self.base_core_number, PositiveInt())
        typecheck(self.parallel_number, NaturalNumber())
        self.own_sets = [SUB_BENCH_SET.format(i) for i in range(0, self.parallel_number)] \
                        + [CONTROLLER_SUB_BENCH_SET, NEW_ROOT_SET, BENCH_SET]
        try:
            self._init_cpuset()
        except BaseException:
            logging.error("Forced teardown of CPUSet")
            self.teardown()
            raise
        logging.info("Finished initializing CPUSet")

    def move_process_to_set(self, pid: int, set_id: int):
        """
        Moves the process with the passed id to the parallel sub cpuset with the passed id.

        :param pid: passed process id
        :param set_id: passed parallel sub cpuset id
        """
        if not self.active:
            return
        try:
            typecheck(pid, Int())
            typecheck(set_id, Int(range=range(0, self.parallel_number)))
            self._move_process_to_set(SUB_BENCH_SET.format(set_id), pid)
        except BaseException:
            logging.error("Forced teardown of CPUSet")
            self.teardown()
            raise

    def get_sub_set(self, set_id: int) -> str:
        """ Gets the name of the benchmarking cpu set with the given id / number (starting at zero). """
        if self.parallel == 0:
            return CONTROLLER_SUB_BENCH_SET
        if self.active:
            typecheck(set_id, Int(range=range(0, self.parallel_number)))
        return SUB_BENCH_SET.format(set_id)

    def teardown(self):
        """
        Tears the created cpusets down and makes the system usable again.
        """
        if not self.active:
            return
        for set in self.own_sets:
            try:
                self._delete_set(set)
            except EnvironmentError as ex:
                pass
                # logging.error(str(ex))
            except BaseException:
                raise

    def _number_of_parallel_sets(self, base_core_number: int, parallel: bool, sub_core_number: int) -> int:
        """
        Calculates the number of possible parallel sets.
        """
        typecheck([base_core_number, parallel, sub_core_number], List(Int()))
        if base_core_number + (0 if self.temci_in_base_set else 1) + sub_core_number > self.av_cores and self.active:
            raise ValueError("Invalid values for base_core_number and sub_core_number "
                             "on system with just {} cores.{}".format(self.av_cores,
                                                                      ""
                                                                      if self.temci_in_base_set
                                                                      else "Note: temci needs a cpuset too."))
        av_cores_for_par = self.av_cores - base_core_number - (0 if self.temci_in_base_set else 1)
        if parallel:
            return av_cores_for_par // sub_core_number
        return 1

    def _init_cpuset(self):
        """
        Mounts the cpuset pseudo filesystem at ``CPUSET_DIR`` and creates the necessary cpusets.
        """
        if not os.path.exists(CPUSET_DIR + "/cgroup.procs"):
            if not os.path.exists(CPUSET_DIR):
                os.mkdir(CPUSET_DIR)
            proc = subprocess.Popen(["bash", "-c", "mount -t cpuset none /cpuset/"],
                                    stdout=subprocess.DEVNULL,
                                    stderr=subprocess.PIPE,
                                    universal_newlines=True)
            out, err = proc.communicate()
            if proc.poll() > 0:
                raise EnvironmentError(
                    "Cannot mount /cpuset. " +
                    "Probably you're not in root mode or you've already mounted cpuset elsewhere.", str(err))
        self._create_cpuset(NEW_ROOT_SET, self._get_av_cpus()[0: self.base_core_number])
        logging.info("Move all processes to new root cpuset")
        self._move_all_to_new_root()
        if self.parallel == 0:  # just use all available cores, as the benchmarked program also runs in it
            self._create_cpuset(CONTROLLER_SUB_BENCH_SET, self._get_av_cpus()[self.base_core_number:self.av_cores])
        else:
            self._create_cpuset(CONTROLLER_SUB_BENCH_SET, self._get_av_cpus()[self.base_core_number:1])
        self._move_process_to_set(NEW_ROOT_SET if self.temci_in_base_set else CONTROLLER_SUB_BENCH_SET)
        for i in range(0, self.parallel_number):
            start = self.base_core_number + (0 if self.temci_in_base_set else 1) + (i * self.sub_core_number)
            self._create_cpuset(SUB_BENCH_SET.format(i), self._get_av_cpus()[start:start + self.sub_core_number])

    def _cpus_of_set(self, name: str) -> t.Optional[t.List[int]]:
        """ Gets all cpu cores that are assigned to the set with the passed name. """
        name = self._relname(name)
        if self._has_set(name):
            res = self._cset("set {}".format(name))
            arr = res.split("\n")[3].strip().split(" ")
            arr = [x for x in arr if x != ""]
            if "-" in arr[1]:
                start, end = map(int, arr[1].split("-"))
                return list(range(start, end + 1))
            elif "," in arr[1]:
                return list(map(int, arr[1].split(",")))
            else:
                return int(arr[1])
        return None

    def _get_av_cpus(self) -> t.List[int]:
        """ Gets the number of available cpu cores """
        return self._cpus_of_set("")

    def _ints_to_str(self, ints: t.List[int]) -> str:
        """ Turns a list of integers comma separated into a string """
        return ",".join(map(str, ints))

    def _has_set(self, name: str):
        """ Does the set with the given name exist? """
        name = self._relname(name)
        return name + "   " in self._cset("set -rl")

    def _delete_set(self, name: str):
        """ Delete the set with the given name """
        self._cset("set -r --force -d %s" % NEW_ROOT_SET)

    def _move_all_to_new_root(self, name: str = 'root', _count: int = 100):
        """
        Move all process from the root cpu set into the ``NEW_ROOT_SET``

        :param name: name of the root cpu set
        :param _count: maximum cpu set tree depth
        """
        cpus = self._get_av_cpus()[0:self.base_core_number]
        self._set_cpu_affinity_of_set(name, cpus)
        if _count > 0:
            for child in self._child_sets(name):
                if len(child) > 1:
                    # print("moved from {child} to {root}".format(child=child, root=NEW_ROOT_SET))
                    try:
                        self._move_all_to_new_root(child, _count - 1)
                    except EnvironmentError as err:
                        pass
                        # logging.warning(str(err))
        self._move_processes(name, NEW_ROOT_SET)
        # if _count == 100:
        #    self._cset("proc --move -k --force --threads --pid=0-100000 --toset={}".format(NEW_ROOT_SET))

    def _move_processes(self, from_set: str, to_set: str):
        """
        Move all processes from the first to the second cpuset.
        Only some kernel threads are left behind.

        :param from_set: name of the first cpuset
        :param to_set: name of the second cpuset
        """
        from_set, to_set = (self._relname(from_set), self._relname(to_set))
        self._cset("proc --move --kthread --force --threads --fromset %s --toset %s" % (from_set, to_set))

    def _move_process_to_set(self, cpuset: str, pid: int = os.getpid()):
        """
        Move the process with the given id into the passed cpu set.

        :param cpuset: name of the passed cpu set
        :param pid: id of the process to move, default is the own process
        """
        self._cset("proc --move --force --pid %d --threads %s" % (pid, cpuset))

    def _absname(self, relname: str):
        """ Get the absolute set name for the given relative """
        if "/" in relname:
            return relname
        res = self._cset("set %s" % relname)
        arr = res.split("\n")[-1].strip().split(" ")
        arr = [x for x in arr if x != ""]
        return arr[7]

    def _relname(self, absname: str):
        """ Get the realtive set name for the given absolute """
        if not "/" in absname:
            return absname
        return absname.split("/")[-1]

    def _child_sets(self, name: str) -> t.List[str]:
        """ Get the list of child set for the set with the given name """
        name = self._relname(name)
        res = self._cset("set %s" % name)
        arr = []
        for line in res.split("\n")[4:]:
            line = line.strip()
            arr.append(line.split(" ")[0])
        return arr

    def _create_cpuset(self, name: str, cpus: t.List[int]):
        """ Create the cpuset with the given name and assign the given cpu cores to it """
        typecheck(cpus, List(Int()))
        cpu_range = self._ints_to_str(cpus)
        path = []
        for part in name.split("/"):
            path.append(part)
            self._cset("set --cpu {} {} ".format(cpu_range, "/".join(path)))

    def _set_cpu_affinity_of_set(self, set: str, cpus: t.List[int]):
        """ Set the cpu affinity for all processes that belong to the given set """
        if set == "root":
            set = ""
        app = "cgroup.procs" if set == "" else set + "/cgroup.procs"
        with open(os.path.join(CPUSET_DIR + "/" + app), "r") as f:
            for line in f.readlines():
                try:
                    self._set_cpu_affinity(int(line.strip()), cpus)
                    # logging.info("success {}".format(line))
                except EnvironmentError as err:
                    pass
                    # logging.error(str(err))

    def _set_cpu_affinity(self, pid: int, cpus: t.List[int]):
        """ Set the cpu affinity for the given process to the given cpu cores """
        cmd = "taskset --all-tasks --cpu-list -p {} {}; nice".format(self._ints_to_str(cpus), pid)
        proc = subprocess.Popen(["/bin/sh", "-c", cmd],
                                stdout=subprocess.PIPE,
                                stderr=subprocess.PIPE,
                                universal_newlines=True)
        out, err = proc.communicate()
        if proc.poll() > 0:
            raise EnvironmentError(
                "taskset error (cmd = '{}'): ".format(cmd) + str(err) + str(out)
            )
        return str(out)

    def _cset(self, argument: str):
        """
        Execute the passed argument with the cset tool.

        :param passed argument for the tool
        :return: output of executing the combined command
        :raises EnvironmentError: if something goes wrong
        """
        # cmd = ["/bin/sh", "-c", "sudo cset {}".format(argument)]
        cmd = ["/bin/sh", "-c", "python3 -c 'import cpuset.main; print(cpuset.main.main())' " + argument]
        proc = subprocess.Popen(cmd,
                                stdout=subprocess.PIPE,
                                stderr=subprocess.PIPE,
                                universal_newlines=True)
        out, err = proc.communicate()
        if proc.poll() > 0:
            raise EnvironmentError(
                "Error with cset tool. "
                " More specific error (cmd = 'cset {}'): ".format(argument) + str(err) + str(out)
            )
        return str(out) def script_relative(file: str) -> str:
    def __init__(self, cmd: str, out: str, err: str):
    def __str__(self) -> str:
def exec(dir: str, cmd: str):
def make_scripts(build_kernel_modules: bool = False):
class ExecError(Exception): class AbstractRegistry:
    """
    An abstract registry.
    To create an own registry set the settings_key_path (type str),
    the use_key (type str), the use_list (type bool) and the default
    attribute (type (use_list ? list of strings : str).

    Important: Be sure to have a "register = {}" line in your extending class.
    """

    settings_key_path = ""  # type: str
    """ Used settings key path """
    use_key = None  # type: t.Optional[str]
    """ Used key that sets which registered class is currently used """
    use_list = False  # type: bool
    """ Allow more than one class to used at a specific moment in time """
    default = None  # type: t.Optional[t.Union[str, t.List[str]]]
    """ Name(s) of the class(es) used by default. Type depends on the `use_list` property."""

    registry = {}  # type: t.Dict[str, type]
    """ Registered classes (indexed by their name) """
    plugin_synonym = ("plugin", "plugins")  # type: t.Tuple[str, str]
    """ Singular and plural version of the word that is used in the documentation for the registered entities """

    @classmethod
    def get_for_name(cls, name: str, *args, **kwargs) -> t.Any:
        """
        Creates a plugin with the given name.

        :param name: name of the registered class
        :return: object of the registered class
        :raises: ValueError if there isn't such a class
        """
        if name not in cls.registry:
            raise ValueError("No such registered class {}".format(name))
        misc_settings = Settings()["/".join([cls.settings_key_path, name + "_misc"])]
        return cls.registry[name](misc_settings, *args, **kwargs)

    @classmethod
    def get_tester(cls) -> 'Tester':
        """
        Returns the tester that is configured in the settings

        :return: tester instance
        """
        return cls.get_for_name(cls.get_used(),
                                Settings()["stats/tester"],
                                Settings()["stats/uncertainty_range"])

    @classmethod
    def get_used(cls) -> t.Union[str, t.List[str]]:
        """
        Get the list of name of the used plugins (use_list=True)
        or the names of the used plugin (use_list=False).
        """
        key = "/".join([cls.settings_key_path, cls.use_key])
        if not Settings().has_key(key):
            return [] if cls.use_list else None
        if cls.use_list:
            plugin_allow_vals = {}
            active_list = Settings()[key].split(",") if not isinstance(Settings()[key], list) else Settings()[key]
            ret_list = []
            for name in sorted(cls.registry.keys()):
                active_path = "{}_active".format("/".join([cls.settings_key_path, name]))
                active = Settings()[active_path]
                if active is None and name in active_list:
                    ret_list.append(name)
                if active is True:
                    ret_list.append(name)
            return ret_list
        else:
            return Settings()[key]

    @classmethod
    def register(cls, name: str, klass: type, misc_type: Type, activate_by_default: bool = None,
                 deprecated: bool = False):
        """
        Registers a new class.
        The constructor of the class gets as first argument the misc settings.

        :param name: common name of the registered class
        :param klass: actual class
        :param misc_type: type scheme of the {name}_misc settings
        :param misc_default: default value of the {name}_misc settings
        :param deprecated: is the registered class deprecated and should not be used?
        """

        if deprecated:
            Settings().get_type_scheme(cls.settings_key_path).unknown_keys = True
            return

        def format_str_list(val: t.List[str], sep: str = "and") -> str:
            return join_strs(list(map(repr, val)), sep)

        misc_type_empty = misc_type == Dict() or misc_type == Dict({})
        misc_default = misc_type.get_default()
        description = None
        use_key_path = "/".join([cls.settings_key_path, cls.use_key])
        misc_key = "{}_misc".format("/".join([cls.settings_key_path, name]))

        if klass.__doc__ is not None:
            header = ""# "Description of {} (class {}):\n".format(name, klass.__qualname__)
            lines = str(klass.__doc__.strip()).split("\n")
            lines = map(lambda x: "  " + x.strip(), lines)
            description = Description(header + "\n".join(lines))
            klass.__description__ = description.description
            misc_type //= description
        else:
            klass.__description__ = ""
            logging.error("Class level documentation for {} is missing".format(klass.__name__))
        Settings().modify_setting(misc_key, misc_type)

        if cls.use_list:
            if not Settings().validate_key_path(use_key_path.split("/")) \
                    or isinstance(Settings().get_type_scheme(use_key_path), Any):
                use_key_type = (StrList() | Exact(name))
                use_key_type.typecheck_default = False
                Settings().modify_setting(use_key_path,
                                          use_key_type // Default(cls.default) if cls.default is not None else use_key_type)
            else:
                use_key_list = Settings().get_type_scheme(use_key_path)
                assert isinstance(use_key_list, StrList)
                use_key_list |= Exact(name)
            use_key_list = Settings().get_type_scheme(use_key_path)
            use_key_list // Description("Possible {} are {}".format(cls.plugin_synonym[1],
                                                                    format_str_list(use_key_list.allowed_values)))
            active_path = "{}_active".format("/".join([cls.settings_key_path, name]))
            if not Settings().validate_key_path(active_path.split("/")):
                Settings().modify_setting(active_path, BoolOrNone() // Default(activate_by_default))
            Settings().get_type_scheme(active_path) // Description("Enable: " + klass.__description__)
        else:
            if not Settings().validate_key_path(use_key_path.split("/")) \
                    or not isinstance(Settings().get_type_scheme(use_key_path), ExactEither):
                use_key_type = ExactEither(name)
                use_key_type.typecheck_default = False
                Settings().modify_setting(use_key_path,
                                          use_key_type // Default(cls.default) if cls.default else use_key_type)
            else:
                Settings().modify_setting(use_key_path, Settings().get_type_scheme(use_key_path) | Exact(name))
            use_key_type = Settings().get_type_scheme(use_key_path)
            use_key_type // Description("Possible {} are {}".format(cls.plugin_synonym[1],
                                                                    format_str_list(use_key_type.exp_values)))
        cls.registry[name] = klass

        if not sphinx_doc():
            return

        use_text = ""
        cls_use_text = ""
        if cls.use_list:
            use_text = "To use this {plugin} add its name (`{name}`) to the list at settings key `{key}` or " \
                       "set `{active}` to true."\
                .format(plugin=cls.plugin_synonym[0], name=name, key=use_key_path,
                        active="{}_active".format("/".join([cls.settings_key_path, name])))
            cls_use_text = "The used {plugins} can be configured by editing the list under the settings key `{key}`."\
                .format(plugins=cls.plugin_synonym[1], key=use_key_path)
        else:
            use_text = "To use this {plugin} set the currently used {plugin} (at key `{key}`) to its name (`{name}`)."\
                .format(plugin=cls.plugin_synonym[0], name=name, key=use_key_path)
            cls_use_text = "The used {plugin} can be configured by editing the settings key `{key}`."\
                .format(plugin=cls.plugin_synonym[0], key=use_key_path)
        other_plugins_text = ""
        used_type = Settings().get_type_scheme(use_key_path)
        possible_plugins = used_type.allowed_values if cls.use_list else used_type.exp_values  # type: t.List[str]
        possbible_p_wo_self = [x for x in possible_plugins if x != name]
        if len(possible_plugins) == 2:
            other_plugins_text = "Another usable {plugin} is `{p}`.".format(plugin=cls.plugin_synonym[0],
                                                                            p=possbible_p_wo_self[0])
        if len(possible_plugins) > 2:
            other_plugins_text = "Other usable {plugins} are {p}.".format(plugins=cls.plugin_synonym[1],
                                                                          p=join_strs(["`{}`".format(x)
                                                                                       for x in possbible_p_wo_self]))
        default_plugins_text = ""
        if cls.default:
            if cls.use_list and len(cls.default) > 1:
                p = join_strs(["`{}`".format(x) for x in cls.default])
                default_plugins_text = "The default {plugins} are {p}.".format(plugins=cls.plugin_synonym[1],
                                                                               p=p)
            else:
                default = cls.default[0] if cls.use_list else cls.default
                default_plugins_text = "The default {plugin} is `{p}`.".format(plugin=cls.plugin_synonym[0],
                                                                            p=default)
        if not misc_type_empty:
            klass.__doc__ += """

    Configuration format:

    .. code-block:: yaml

        {default_yaml}


    This {plugin} can be configured under the settings key `{misc_key}`.

        """.format(default_yaml="\n        ".join(misc_type.get_default_yaml().split("\n")),
                   misc_key=misc_key, plugin=cls.plugin_synonym[0])
            klass.__doc__ += """

    {use_text}
    {other}
    {default}
        """.format(use_text=use_text, other=other_plugins_text, default=default_plugins_text)
        if not hasattr(cls, "__old_documentation__"):
            cls.__old_documentation__ = cls.__doc__ or ""
        cls.__doc__ = cls.__old_documentation__
        cls.__doc__ += """

    {use_text}
    {possible}
    """.format(
            use_text=cls_use_text,
            possible=Settings().get_type_scheme(use_key_path).description
        )


    @classmethod
    def __getitem__(cls, name: str) -> type:
        """
        Alias for get_for_name(name).
        """
        return cls.get_for_name(name)

    @classmethod
    def get_class(cls, name: str) -> type:
        return cls.registry[name] @classmethod
def register(cls, name: str, klass: type, misc_type: Type, activate_by_default: bool = None,
             deprecated: bool = False):
    """
    Registers a new class.
    The constructor of the class gets as first argument the misc settings.

    :param name: common name of the registered class
    :param klass: actual class
    :param misc_type: type scheme of the {name}_misc settings
    :param misc_default: default value of the {name}_misc settings
    :param deprecated: is the registered class deprecated and should not be used?
    """

    if deprecated:
        Settings().get_type_scheme(cls.settings_key_path).unknown_keys = True
        return

    def format_str_list(val: t.List[str], sep: str = "and") -> str:
        return join_strs(list(map(repr, val)), sep)

    misc_type_empty = misc_type == Dict() or misc_type == Dict({})
    misc_default = misc_type.get_default()
    description = None
    use_key_path = "/".join([cls.settings_key_path, cls.use_key])
    misc_key = "{}_misc".format("/".join([cls.settings_key_path, name]))

    if klass.__doc__ is not None:
        header = ""# "Description of {} (class {}):\n".format(name, klass.__qualname__)
        lines = str(klass.__doc__.strip()).split("\n")
        lines = map(lambda x: "  " + x.strip(), lines)
        description = Description(header + "\n".join(lines))
        klass.__description__ = description.description
        misc_type //= description
    else:
        klass.__description__ = ""
        logging.error("Class level documentation for {} is missing".format(klass.__name__))
    Settings().modify_setting(misc_key, misc_type)

    if cls.use_list:
        if not Settings().validate_key_path(use_key_path.split("/")) \
                or isinstance(Settings().get_type_scheme(use_key_path), Any):
            use_key_type = (StrList() | Exact(name))
            use_key_type.typecheck_default = False
            Settings().modify_setting(use_key_path,
                                      use_key_type // Default(cls.default) if cls.default is not None else use_key_type)
        else:
            use_key_list = Settings().get_type_scheme(use_key_path)
            assert isinstance(use_key_list, StrList)
            use_key_list |= Exact(name)
        use_key_list = Settings().get_type_scheme(use_key_path)
        use_key_list // Description("Possible {} are {}".format(cls.plugin_synonym[1],
                                                                format_str_list(use_key_list.allowed_values)))
        active_path = "{}_active".format("/".join([cls.settings_key_path, name]))
        if not Settings().validate_key_path(active_path.split("/")):
            Settings().modify_setting(active_path, BoolOrNone() // Default(activate_by_default))
        Settings().get_type_scheme(active_path) // Description("Enable: " + klass.__description__)
    else:
        if not Settings().validate_key_path(use_key_path.split("/")) \
                or not isinstance(Settings().get_type_scheme(use_key_path), ExactEither):
            use_key_type = ExactEither(name)
            use_key_type.typecheck_default = False
            Settings().modify_setting(use_key_path,
                                      use_key_type // Default(cls.default) if cls.default else use_key_type)
        else:
            Settings().modify_setting(use_key_path, Settings().get_type_scheme(use_key_path) | Exact(name))
        use_key_type = Settings().get_type_scheme(use_key_path)
        use_key_type // Description("Possible {} are {}".format(cls.plugin_synonym[1],
                                                                format_str_list(use_key_type.exp_values)))
    cls.registry[name] = klass

    if not sphinx_doc():
        return

    use_text = ""
    cls_use_text = ""
    if cls.use_list:
        use_text = "To use this {plugin} add its name (`{name}`) to the list at settings key `{key}` or " \
                   "set `{active}` to true."\
            .format(plugin=cls.plugin_synonym[0], name=name, key=use_key_path,
                    active="{}_active".format("/".join([cls.settings_key_path, name])))
        cls_use_text = "The used {plugins} can be configured by editing the list under the settings key `{key}`."\
            .format(plugins=cls.plugin_synonym[1], key=use_key_path)
    else:
        use_text = "To use this {plugin} set the currently used {plugin} (at key `{key}`) to its name (`{name}`)."\
            .format(plugin=cls.plugin_synonym[0], name=name, key=use_key_path)
        cls_use_text = "The used {plugin} can be configured by editing the settings key `{key}`."\
            .format(plugin=cls.plugin_synonym[0], key=use_key_path)
    other_plugins_text = ""
    used_type = Settings().get_type_scheme(use_key_path)
    possible_plugins = used_type.allowed_values if cls.use_list else used_type.exp_values  # type: t.List[str]
    possbible_p_wo_self = [x for x in possible_plugins if x != name]
    if len(possible_plugins) == 2:
        other_plugins_text = "Another usable {plugin} is `{p}`.".format(plugin=cls.plugin_synonym[0],
                                                                        p=possbible_p_wo_self[0])
    if len(possible_plugins) > 2:
        other_plugins_text = "Other usable {plugins} are {p}.".format(plugins=cls.plugin_synonym[1],
                                                                      p=join_strs(["`{}`".format(x)
                                                                                   for x in possbible_p_wo_self]))
    default_plugins_text = ""
    if cls.default:
        if cls.use_list and len(cls.default) > 1:
            p = join_strs(["`{}`".format(x) for x in cls.default])
            default_plugins_text = "The default {plugins} are {p}.".format(plugins=cls.plugin_synonym[1],
                                                                           p=p)
        else:
            default = cls.default[0] if cls.use_list else cls.default
            default_plugins_text = "The default {plugin} is `{p}`.".format(plugin=cls.plugin_synonym[0],
                                                                        p=default)
    if not misc_type_empty:
        klass.__doc__ += """

Configuration format:

.. code-block:: yaml

    {default_yaml}


This {plugin} can be configured under the settings key `{misc_key}`.

    """.format(default_yaml="\n        ".join(misc_type.get_default_yaml().split("\n")),
               misc_key=misc_key, plugin=cls.plugin_synonym[0])
        klass.__doc__ += """

{use_text}
{other}
{default}
    """.format(use_text=use_text, other=other_plugins_text, default=default_plugins_text)
    if not hasattr(cls, "__old_documentation__"):
        cls.__old_documentation__ = cls.__doc__ or ""
    cls.__doc__ = cls.__old_documentation__
    cls.__doc__ += """

{use_text}
{possible}
""".format(
        use_text=cls_use_text,
        possible=Settings().get_type_scheme(use_key_path).description
    )
import collections
import datetime
import gc
import logging
import os
import random
import re
import shlex
import shutil
import stat
import subprocess
import tempfile
import time
import typing as t
import humanfriendly
import yaml
        import click
from collections import namedtuple
from copy import deepcopy
from temci.build.build_processor import BuildProcessor
from temci.utils.config_utils import ATTRIBUTES_TYPE
from temci.utils.settings import Settings
from temci.utils.sudo_utils import get_bench_user, bench_as_different_user, get_env_setting
from temci.utils.typecheck import NoInfo
from temci.utils.util import has_root_privileges, join_strs, does_command_succeed, sphinx_doc, on_apple_os, \
    does_program_exist, document, proc_wait_with_rusage, rusage_header
from temci.utils.vcs import VCSDriver
from .cpuset import CPUSet
from ..setup import setup
from ..utils.registry import AbstractRegistry, register
from ..utils.typecheck import *
        from temci.report.rundata import get_for_tags
        from temci.report.rundata import RecordedProgramError
        from temci.report.rundata import RecordedInternalError
    from temci.report.rundata import RecordedProgramError
        from temci.report.rundata import RecordedProgramError
        self.error = error  # type: t.Optional[BaseException]
        """ Exception object if something went wrong during benchmarking """
        self.recorded_error = recorded_error

    def properties(self) -> t.List[str]:
        """ Get a list of the measured properties """
        return list(self.data.keys())

    def add_run_data(self, data: t.Dict[str, t.Union[Number, t.List[Number]]]):
        """
        Add data.

        :param data: data to be added (measured data per property)
        """
        typecheck(data, Dict(unknown_keys=True, key_type=Str(), value_type=Int() | Float() | List(Int() | Float())))
        for prop in data:
            if isinstance(data[prop], list):
                self.data[prop].extend(data[prop])
            else:
                self.data[prop].append(data[prop])

                # def _to_dict(self):
                #    """
                #    Serializes this instance into a data structure that is accepted by the ``from_dict`` method.
                #    """
                #    return {
                #        "properties": self.properties(),
                #        "data": self.data
                #    }
                #
                # @classmethod
                # def _from_dict(cls, source: dict):
                #    typecheck(source, Dict({
                #        "data": Dict(unknown_keys=True)
                #    }, unknown_keys=True))
                #    return BenchmarkingResultBlock(source["data"])


class AbstractRunDriver(AbstractRegistry):
    """
    A run driver that does the actual benchmarking and supports plugins to modify the benchmarking environment.

    The constructor also calls the setup methods on all registered plugins. It calls the setup() method.
    """

    settings_key_path = "run/plugins"
    use_key = "active"
    use_list = True
    default = []
    registry = {}
    plugin_synonym = ("run driver plugin", "run driver plugins")
    block_type_scheme = Dict()
    """ Type scheme for the program block configuration """
    runs_benchmarks = True

    store_files = True

    def __init__(self, misc_settings: dict = None):
        """
        Creates an instance.
        Also calls the setup methods on all registered plugins.
        It calls the setup() method.

        :param misc_settings: further settings
        """
        self.misc_settings = misc_settings
        """ Further settings """
        self.used_plugins = []  # type: t.List[RunDriverPlugin]
        """ Used and active plugins """
        miss_root_plugins = []
        is_root = has_root_privileges()
        for used in self.get_used_plugins():
            klass = self.get_class(used)
            if klass.needs_root_privileges and not is_root:
                miss_root_plugins.append(used)
            else:
                self.used_plugins.append(self.get_for_name(used))
        if miss_root_plugins:
            logging.warning("The following plugins are disabled because they need root privileges "
                            "(consider using `--sudo`): %s", join_strs(miss_root_plugins))
        self.setup()

    def setup(self):
        """
        Call the setup() method on all used plugins for this driver.
        """
        for plugin in self.used_plugins:
            plugin.setup()

    def teardown(self):
        """
        Call the teardown() method on all used plugins for this driver.
        """
        for plugin in self.used_plugins:
            plugin.teardown()

    def _setup_block(self, block: RunProgramBlock):
        """
        Call the setup_block() method on all used plugins for this driver.
        """
        typecheck(block.attributes, self.get_full_block_typescheme()["attributes"],
                  value_name="attributes of {}".format(block))
        for plugin in self.used_plugins:
            plugin.setup_block(block)

    def _setup_block_run(self, block: RunProgramBlock):
        """
        Call the setup_block_run() method on all used plugins for this driver.
        """
        for plugin in self.used_plugins:
            plugin.setup_block_run(block)

    def _teardown_block(self, block: RunProgramBlock):
        """
        Call the teardown_block() method on all used plugins for this driver.
        """
        for plugin in self.used_plugins:
            plugin.teardown_block(block)

    def benchmark(self, block: RunProgramBlock, runs: int,

                  cpuset: CPUSet = None, set_id: int = 0,