def compute_and_save_predictions(config_file_or_obj_or_dict,
                                 output_file,
                                 feats_file=None,
                                 logger=None):
    """
    Run rsmpredict using the given configuration.

    Generate predictions using given configuration file, object, or
    dictionary. Predictions are saved in ``output_file``. Optionally,
    pre-processed feature values are saved in ``feats_file``,
    if specified.

    Parameters
    ----------
    config_file_or_obj_or_dict : str or pathlib.Path or dict or Configuration
        Path to the experiment configuration file either a a string
        or as a ``pathlib.Path`` object. Users can also pass a
        ``Configuration`` object that is in memory or a Python dictionary
        with keys corresponding to fields in the configuration file. Given a
        configuration file, any relative paths in the configuration file
        will be interpreted relative to the location of the file. Given a
        ``Configuration`` object, relative paths will be interpreted
        relative to the ``configdir`` attribute, that _must_ be set. Given
        a dictionary, the reference path is set to the current directory.
    output_file : str
        The path to the output file.
    feats_file : str, optional
        Path to the output file for saving preprocessed feature values.
    logger : logging object, optional
        A logging object. If ``None`` is passed, get logger from ``__name__``.
        Defaults to ``None``.

    Raises
    ------
    FileNotFoundError
        If any of the files contained in ``config_file_or_obj_or_dict``
        cannot be located.
    FileNotFoundError
        If ``experiment_dir`` does not exist.
    FileNotFoundError
        If ``experiment_dir`` does not contain the required output
        needed from an rsmtool experiment.
    RuntimeError
        If the name of the output file does not end in
        ".csv", ".tsv", or ".xlsx".
    """
    logger = logger if logger else logging.getLogger(__name__)

    configuration = configure('rsmpredict', config_file_or_obj_or_dict)

    # get the experiment ID
    experiment_id = configuration['experiment_id']

    # Get output format
    file_format = configuration.get('file_format', 'csv')

    # Get DataWriter object
    writer = DataWriter(experiment_id)

    # get the input file containing the feature values
    # for which we want to generate the predictions
    input_features_file = DataReader.locate_files(configuration['input_features_file'],
                                                  configuration.configdir)
    if not input_features_file:
        raise FileNotFoundError('Input file {} does not exist'
                                ''.format(configuration['input_features_file']))

    experiment_dir = DataReader.locate_files(configuration['experiment_dir'],
                                             configuration.configdir)
    if not experiment_dir:
        raise FileNotFoundError('The directory {} does not exist.'
                                ''.format(configuration['experiment_dir']))
    else:
        experiment_output_dir = normpath(join(experiment_dir, 'output'))
        if not exists(experiment_output_dir):
            raise FileNotFoundError('The directory {} does not contain '
                                    'the output of an rsmtool experiment.'.format(experiment_dir))

    # find all the .model files in the experiment output directory
    model_files = glob.glob(join(experiment_output_dir, '*.model'))
    if not model_files:
        raise FileNotFoundError('The directory {} does not contain any rsmtool models.'
                                ''.format(experiment_output_dir))

    experiment_ids = [splitext(basename(mf))[0] for mf in model_files]
    if experiment_id not in experiment_ids:
        raise FileNotFoundError('{} does not contain a model for the experiment "{}". '
                                'The following experiments are contained in this '
                                'directory: {}'.format(experiment_output_dir,
                                                       experiment_id,
                                                       experiment_ids))

    # check that the directory contains outher required files
    required_file_types = ['feature', 'postprocessing_params']
    for file_type in required_file_types:
        expected_file_name = "{}_{}.csv".format(experiment_id, file_type)
        if not exists(join(experiment_output_dir, expected_file_name)):
            raise FileNotFoundError('{} does not contain the required file '
                                    '{} that was generated during the '
                                    'original model training'.format(experiment_output_dir,
                                                                     expected_file_name))

    logger.info('Reading input files.')

    feature_info = join(experiment_output_dir,
                        '{}_feature.csv'.format(experiment_id))

    post_processing = join(experiment_output_dir,
                           '{}_postprocessing_params.csv'.format(experiment_id))

    file_paths = [input_features_file, feature_info, post_processing]
    file_names = ['input_features',
                  'feature_info',
                  'postprocessing_params']

    converters = {'input_features': configuration.get_default_converter()}

    # Initialize the reader
    reader = DataReader(file_paths, file_names, converters)
    data_container = reader.read(kwargs_dict={'feature_info': {'index_col': 0}})

    # load the Modeler to generate the predictions
    model = Modeler.load_from_file(join(experiment_output_dir,
                                        '{}.model'.format(experiment_id)))

    # Add the model to the configuration object
    configuration['model'] = model

    # Initialize the processor
    processor = FeaturePreprocessor(logger=logger)

    (processed_config,
     processed_container) = processor.process_data(configuration,
                                                   data_container,
                                                   context='rsmpredict')

    # save the pre-processed features to disk if we were asked to
    if feats_file is not None:
        logger.info('Saving pre-processed feature values to {}'.format(feats_file))

        feats_dir = dirname(feats_file)

        # create any directories needed for the output file
        os.makedirs(feats_dir, exist_ok=True)

        _, feats_filename = split(feats_file)
        feats_filename, _ = splitext(feats_filename)

        # Write out files
        writer.write_experiment_output(feats_dir,
                                       processed_container,
                                       include_experiment_id=False,
                                       dataframe_names=['features_processed'],
                                       new_names_dict={'features_processed':
                                                       feats_filename},
                                       file_format=file_format)

    if (output_file.lower().endswith('.csv') or
            output_file.lower().endswith('.xlsx')):

        output_dir = dirname(output_file)
        _, filename = split(output_file)
        filename, _ = splitext(filename)

    else:
        output_dir = output_file
        filename = 'predictions_with_metadata'

    # create any directories needed for the output file
    os.makedirs(output_dir, exist_ok=True)

    # save the predictions to disk
    logger.info('Saving predictions.')

    # Write out files
    writer.write_experiment_output(output_dir,
                                   processed_container,
                                   include_experiment_id=False,
                                   dataframe_names=['predictions_with_metadata'],
                                   new_names_dict={'predictions_with_metadata':
                                                   filename},
                                   file_format=file_format)

    # save excluded responses to disk
    if not processed_container.excluded.empty:

        # save the predictions to disk
        logger.info('Saving excluded responses to {}'.format(join(output_dir,
                                                                  '{}_excluded_responses.csv'
                                                                  ''.format(filename))))

        # Write out files
        writer.write_experiment_output(output_dir,
                                       processed_container,
                                       include_experiment_id=False,
                                       dataframe_names=['excluded'],
                                       new_names_dict={'excluded':
                                                       '{}_excluded_responses'
                                                       ''.format(filename)},
                                       file_format=file_format) class Configuration:
    """
    Configuration class.

    Encapsulates all of the configuration parameters and methods to 
    access these parameters.
    """

    def __init__(self,
                 configdict,
                 *,
                 configdir=None,
                 context='rsmtool',
                 logger=None):
        """
        Create an object of the `Configuration` class.

        This method can be used to directly instantiate a Configuration
        object.

        Parameters
        ----------
        configdict : dict
            A dictionary of configuration parameters.
            The dictionary must be a valid configuration dictionary
            with default values filled as necessary.
        configdir : str, optional, keyword-only
            The reference path used to resolve any relative paths
            in the configuration object. When ``None``, will be set during
            initialization to the current working directory.
            Defaults to ``None``,
        context : str
            The context of the tool. One of {"rsmtool", "rsmeval", "rsmcompare",
            "rsmpredict", "rsmsummarize"}.
            Defaults to "rsmtool".
        logger : logging object, optional
            A logging object. If ``None`` is passed, get logger from ``__name__``.
            Defaults to ``None``.
        """
        if not isinstance(configdict, dict):
            raise TypeError('The input must be a dictionary.')

        # create a logger instance
        self.logger = logger if logger else logging.getLogger(__name__)

        # process and validate the configuration dictionary
        configdict = ConfigurationParser.process_config(configdict)
        configdict = ConfigurationParser.validate_config(configdict, context=context)

        # set configdir to `cwd` if not given and let the user know
        if configdir is None:
            configdir = Path(getcwd())
            self.logger.info("Configuration directory will be set to {}".format(configdir))
        else:
            configdir = Path(configdir).resolve()

        self._config = configdict
        self._configdir = configdir
        self._context = context

    def __contains__(self, key):
        """
        Check if the configuration object contains a given key.

        Parameters
        ----------
        key : str
            Key to check in the Configuration object.

        Returns
        -------
        key_check : bool
            ``True`` if key is in configuration object, else ``False``.
        """
        return key in self._config

    def __getitem__(self, key):
        """
        Get configuration value for the given key.

        Parameters
        ----------
        key : str
            Key to check in the configuration object

        Returns
        -------
        value
            The value in the configuration object dictionary.
        """
        return self._config[key]

    def __setitem__(self, key, value):
        """
        Set configuration value for the given key.

        Parameters
        ----------
        key : str
            Key to set in the configuration object.
        value
            The value to be set for the key.
        """
        self._config[key] = value

    def __len__(self):
        """
        Return the size of the configuration dictionary.

        Returns
        -------
        size : int
            The number of elements in the configuration dictionary.
        """
        return len(self._config)

    def __str__(self):
        """
        Return string representation of the configuration dictionary.

        Returns
        -------
        config_string : str
            A string representation of the underlying configuration
            dictionary as encoded by ``json.dumps()``. It only
            includes the configuration options that can be set by
            the user.
        """
        expected_fields = (CHECK_FIELDS[self._context]['required'] +
                           CHECK_FIELDS[self._context]['optional'])

        output_config = {k: v for k, v in self._config.items() if k in expected_fields}
        return json.dumps(output_config, indent=4, separators=(',', ': '))

    def __iter__(self):
        """
        Iterate through configuration object keys.

        Yields
        ------
        key : str
            A key in the config dictionary
        """
        for key in self.keys():
            yield key

    @property
    def configdir(self):
        """
        Get the path to configuration directory.

        Get the path to the configuration reference directory that
        will be used to resolve any relative paths in
        the configuration.

        Returns
        -------
        configdir : str
            The path to the configuration reference directory.
        """
        return str(self._configdir)

    @configdir.setter
    def configdir(self, new_path):
        """
        Set a new configuration reference directory.

        Parameters
        ----------
        new_path : str
            Path to the new configuration reference
            directory used to resolve any relative paths
            in the configuration object.
        """
        if new_path is None:
            raise ValueError("The `configdir` attribute cannot be set to `None` ")

        # TODO: replace `Path(abspath(new_path))` with `Path(new_path).resolve()
        # once this Windows bug is fixed: https://bugs.python.org/issue38671
        self._configdir = Path(abspath(new_path))

    @property
    def context(self):
        """Get the context."""
        return self._context

    @context.setter
    def context(self, new_context):
        """
        Set a new context.

        Parameters
        ----------
        new_context : str
            A new context for the configuration object.
        """
        self._context = new_context

    def get(self, key, default=None):
        """
        Get value or default for the given key.

        Parameters
        ----------
        key : str
            Key to check in the Configuration object.
        default, optional
            The default value to return, if no key exists.
            Defaults to ``None``.

        Returns
        -------
        value
            The value in the configuration object dictionary.
        """
        return self._config.get(key, default)

    def to_dict(self):
        """
        Get a dictionary representation of the configuration object.

        Returns
        -------
        config : dict
            The configuration dictionary.
        """
        return self._config

    def keys(self):
        """
        Return keys as a list.

        Returns
        -------
        keys : list of str
            A list of keys in the configuration object.
        """
        return [k for k in self._config.keys()]

    def values(self):
        """
        Return configuration values as a list.

        Returns
        -------
        values : list
            A list of values in the configuration object.
        """
        return [v for v in self._config.values()]

    def items(self):
        """
        Return configuration items as a list of tuples.

        Returns
        -------
        items : list of tuples
            A list of (key, value) tuples in the
            configuration object.
        """
        return [(k, v) for k, v in self._config.items()]

    def pop(self, key, default=None):
        """
        Remove and return an element from the object having the given key.

        Parameters
        ----------
        key : str
            Key to pop in the configuration object.
        default, optional
            The default value to return, if no key exists.
            Defaults to ``None``.

        Returns
        -------
        value
            The value removed from the object.
        """
        return self._config.pop(key, default)

    def copy(self, deep=True):
        """
        Return a copy of the object.

        Parameters
        ----------
        deep : bool, optional
            Whether to perform a deep copy.
            Defaults to ``True``.

        Returns
        -------
        copy : Configuration
            A new configuration object.
        """
        if deep:
            return deepcopy(self)
        return copy(self)

    def save(self, output_dir=None):
        """
        Save the configuration file to the output directory specified.

        Parameters
        ----------
        output_dir : str
            The path to the output directory. If ``None``, the current
            directory is used.
            Defaults to ``None``.
        """
        # save a copy of the main config into the output directory
        if output_dir is None:
            output_dir = Path(getcwd())

        # Create output directory, if it does not exist
        output_dir = Path(output_dir).resolve() / 'output'
        output_dir.mkdir(parents=True, exist_ok=True)

        id_field = ID_FIELDS[self._context]
        experiment_id = self._config[id_field]
        context = self._context
        outjson = output_dir / f"{experiment_id}_{context}.json"

        with outjson.open(mode='w') as outfile:
            outfile.write(str(self))

    def check_exclude_listwise(self):
        """
        Check for candidate exclusion.

        Check if we are excluding candidates based on number of responses, and
        add this to the configuration file.

        Returns
        -------
        exclude_listwise : bool
            Whether to exclude list-wise.
        """
        exclude_listwise = False
        if self._config.get('min_items_per_candidate'):
            exclude_listwise = True
        return exclude_listwise

    def check_flag_column(self,
                          flag_column='flag_column',
                          partition='unknown'):
        """
        Make sure the column name in ``flag_column`` is correctly specified.

        Get flag columns and values for filtering, if any, and convert single
        values to lists. Raises an exception if the column name in ``flag_column``
        is not correctly specified in the configuration file.

        Parameters
        ----------
        flag_column : str
            The flag column name to check. Currently used names are
            "flag_column" or "flag_column_test".
            Defaults to "flag_column".

        partition: str
            The data partition which is filtered based on the flag column name.
            One of {"train", "test", "both", "unknown"}.
            Defaults to "both".

        Returns
        -------
        new_filtering_dict : dict
            Properly formatted dictionary for the column
            name in ``flag_column``.

        Raises
        ------
        ValueError
            If the specified value of the column name in ``flag_column``
            is not a dictionary.
        ValueError
            If the value of ``partition`` is not in the expected list.
        ValueError
            If the value of ``partition`` does not match the ``flag_column``.
        """
        config = self._config

        new_filter_dict = {}

        flag_message = {"train": "training",
                        "test": "evaluating",
                        "both": "training and evaluating",
                        "unknown": "training and/or evaluating"}

        if partition not in flag_message:
            raise ValueError("Unknown value for partition: {} "
                             "This must be one of the following: {}."
                             "".format(partition, ','.join(flag_message.keys())))

        if flag_column == "flag_column_test":
            if partition in ["both", "train"]:
                raise ValueError("Conditions specified in 'flag_column_test' "
                                 "can only be applied to the evaluation partition.")

        if config.get(flag_column):

            original_filter_dict = config[flag_column]

            # first check that the value is a dictionary
            if not isinstance(original_filter_dict, dict):
                raise ValueError("`flag_column` must be a dictionary. "
                                 "Please refer to the documentation for "
                                 "further information.")

            for column in original_filter_dict:

                # if we were given a single value, convert it to list
                if not isinstance(original_filter_dict[column], list):
                    new_filter_dict[column] = [original_filter_dict[column]]
                    self.logger.warning("The filtering condition {}"
                                        " for column {} was converted "
                                        "to list. Only responses where "
                                        "{} == {} will be used for "
                                        "{} the "
                                        "model. You can ignore this "
                                        "warning if this is the correct "
                                        "interpretation of your "
                                        "configuration settings"
                                        ".".format(original_filter_dict[column],
                                                   column,
                                                   column,
                                                   original_filter_dict[column],
                                                   flag_message[partition])
                                        )
                else:
                    new_filter_dict[column] = original_filter_dict[column]

                    model_eval = ', '.join(map(str,
                                               original_filter_dict[column]))
                    self.logger.info("Only responses where "
                                     "{} equals one of the following values "
                                     "will be used for {} the model: "
                                     "{}.".format(column,
                                                  flag_message[partition],
                                                  model_eval))
        return new_filter_dict

    def get_trim_min_max_tolerance(self):
        """
        Get trim min, trim max, and tolerance values.

        Get the specified trim min and max, and trim_tolerance if any,
        and make sure they are numeric.

        Returns
        -------
        spec_trim_min : float
            Specified trim min value.
        spec_trim_max : float
            Specified trim max value.
        spec_trim_tolerance: float
            Specified trim tolerance value.
        """
        config = self._config

        spec_trim_min = config.get('trim_min', None)
        spec_trim_max = config.get('trim_max', None)
        spec_trim_tolerance = config.get('trim_tolerance', None)

        if spec_trim_min:
            spec_trim_min = float(spec_trim_min)
        if spec_trim_max:
            spec_trim_max = float(spec_trim_max)
        if spec_trim_tolerance:
            spec_trim_tolerance = float(spec_trim_tolerance)
        return (spec_trim_min, spec_trim_max, spec_trim_tolerance)

    def get_rater_error_variance(self):
        """
        Get specified rater error variance, if any, and make sure it's numeric.

        Returns
        -------
        rater_error_variance : float
            Specified rater error variance.
        """
        config = self._config

        rater_error_variance = config.get('rater_error_variance', None)

        if rater_error_variance:
            rater_error_variance = float(rater_error_variance)

        return rater_error_variance

    def get_default_converter(self):
        """
        Get default converter dictionary for data reader.

        Returns
        -------
        default_converter : dict
            The default converter for a train or test file.
        """
        string_columns = [self._config['id_column']]

        candidate = self._config.get('candidate_column')
        if candidate is not None:
            string_columns.append(candidate)

        subgroups = self._config.get('subgroups')
        if subgroups:
            string_columns.extend(subgroups)

        return dict([(column, str) for column in string_columns if column])

    def get_names_and_paths(self, keys, names):
        """
        Get a list of values for the given keys.

        Remove any values that are ``None``.

        Parameters
        -------
        keys : list
            A list of keys whose values to retrieve.
        names : list
            The default value to use if key cannot be found.
            Defaults to ``None``.

        Returns
        -------
        values : list
            The list of values.

        Raises
        ------
        ValueError
            If there are any duplicate keys or names.
        """
        assert len(keys) == len(names)

        # Make sure keys are not duplicated
        if not len(set(keys)) == len(keys):
            raise ValueError('The ``keys`` must be unique. However, the '
                             'following duplicate keys were found: {}.'
                             ''.format(', '.join([key for key, val in Counter(keys).items()
                                                  if val > 1])))
        # Make sure names are not duplicated
        if not len(set(names)) == len(names):
            raise ValueError('The``names`` must be unique. However, the '
                             'following duplicate names were found: {}.'
                             ''.format(', '.join([name for name, val in Counter(names).items()
                                                  if val > 1])))
        existing_names = []
        existing_paths = []
        for idx, key in enumerate(keys):

            path = self._config.get(key)

            # if the `features` field is a list,
            # do not include it in the container
            if key == 'features':
                if isinstance(path, list):
                    continue

            if path is not None:
                existing_paths.append(path)
                existing_names.append(names[idx])

        return existing_names, existing_paths def check_file_output(file1, file2, file_format='csv'):
    """
    Check if the two given tabular files contain matching values.

    This function checks if two experiment files have values that are
    the same to within 3 decimal places. It raises an AssertionError if
    they are not.

    Parameters
    ----------
    file1 : str
        Path to the first file.
    file2 : str
        Path to the second files.
    file_format : str, optional
        The format of the output files.
        Defaults to "csv".
    """
    # make sure that the main id columns are read as strings since
    # this may affect merging in custom notebooks
    string_columns = ['spkitemid', 'candidate']

    converter_dict = {column: str for column in string_columns}

    df1 = DataReader.read_from_file(file1, converters=converter_dict)
    df2 = DataReader.read_from_file(file2, converters=converter_dict)

    # convert all column names to strings
    # we do this to avoid any errors during sorting.
    for df in [df1, df2]:
        df.columns = df.columns.map(str)

    # if the first column is numeric, just force the index to string;
    # however, if it is non-numeric, assume that it is an index and
    # force it to string. We do this to ensure string indices are
    # preserved as such
    for df in [df1, df2]:
        if np.issubdtype(df[df.columns[0]].dtype, np.number):
            df.index = df.index.map(str)
        else:
            df.index = df[df.columns[0]]
            df.index = df.index.map(str)

    # sort all the indices alphabetically
    df1.sort_index(inplace=True)
    df2.sort_index(inplace=True)

    # sort all columns alphabetically
    df1.sort_index(axis=1, inplace=True)
    df2.sort_index(axis=1, inplace=True)

    # convert any integer columns to floats in either data frame
    for df in [df1, df2]:
        for c in df.columns:
            if df[c].dtype == np.int64:
                df[c] = df[c].astype(np.float64)

    # for pca and factor correlations convert all values to absolutes
    # because the sign may not always be the same
    if (file1.endswith('pca.{}'.format(file_format)) or
            file1.endswith('factor_correlations.{}'.format(file_format))):
        for df in [df1, df2]:
            msk = df.dtypes == np.float64
            df.loc[:, msk] = df.loc[:, msk].abs()

    try:
        assert_frame_equal(df1,
                           df2,
                           check_exact=False,
                           rtol=1e-03)
    except AssertionError as e:
        message = e.args[0]
        new_message = 'File {} - {}'.format(basename(file1), message)
        e.args = (new_message, )
        raise def check_generated_output(generated_files, experiment_id, model_source, file_format='csv'):
    """
    Check that all necessary output files have been generated.

    Parameters
    ----------
    generated_files : list of str
        List of files generated by a test.
    experiment_id : str
        The experiment ID.
    model_source : str
        One of "rsmtool" or "skll".
    file_format : str, optional
        The format of the output files.
        Defaults to "csv".
    """
    file_must_have_both = ["_confMatrix.{}".format(file_format),
                           "_cors_orig.{}".format(file_format),
                           "_cors_processed.{}".format(file_format),
                           "_eval.{}".format(file_format),
                           "_eval_short.{}".format(file_format),
                           "_feature.{}".format(file_format),
                           "_feature_descriptives.{}".format(file_format),
                           "_feature_descriptivesExtra.{}".format(file_format),
                           "_feature_outliers.{}".format(file_format),
                           "_margcor_score_all_data.{}".format(file_format),
                           "_pca.{}".format(file_format),
                           "_pcavar.{}".format(file_format),
                           "_pcor_score_all_data.{}".format(file_format),
                           "_pred_processed.{}".format(file_format),
                           "_pred_train.{}".format(file_format),
                           "_score_dist.{}".format(file_format),
                           "_train_preprocessed_features.{}".format(file_format),
                           "_test_preprocessed_features.{}".format(file_format),
                           "_postprocessing_params.{}".format(file_format)
                           ]

    file_must_have_rsmtool = ["_betas.{}".format(file_format),
                              "_coefficients.{}".format(file_format)]
    if model_source == 'rsmtool':
        file_must_have = file_must_have_both + file_must_have_rsmtool
    else:
        file_must_have = file_must_have_both

    file_must_with_id = [experiment_id + file_name for file_name in file_must_have]
    file_exist = [basename(file_name) for file_name in generated_files]
    missing_file = set(file_must_with_id).difference(set(file_exist))
    assert_equal(len(missing_file), 0, "Missing files: {}".format(','.join(missing_file))) def check_report(html_file,
                 raise_errors=True,
                 raise_warnings=True):
    """
    Raise ``AssertionError`` if given HTML report contains errors or warnings.

    Parameters
    ----------
    html_file : str
        Path to the HTML report file on disk.
    raise_errors : bool, optional
        Whether to raise an ``AssertionError`` if there
        are any errors in the report.
        Defaults to ``True``.
    raise_warnings : bool, optional
        Whether to raise an ``AssertionError`` if there
        are any warnings in the report.
        Defaults to ``True``.
    """
    report_errors = 0
    report_warnings = 0

    # Setting raise_warnings to false if not in STRICT mode
    if IGNORE_WARNINGS:
        raise_warnings = False

    with open(html_file, 'r') as htmlf:
        for line in htmlf:
            m_error = html_error_regexp.search(line)
            if m_error:
                report_errors += 1
            m_warning = html_warning_regexp.search(line)
            if m_warning:
                # actual text of warning is in the next line of HTML file
                warning_text = htmlf.readline()

                # NOTE: there is a separate function
                # ``collect_warning_messages_from_the_report`` that once again
                # checks for warnings. The warnings filtered here might still
                # be flagged by that function.
                # See https://github.com/EducationalTestingService/rsmtool/issues/539

                # we do not want to flag matlplotlib font cache warning
                if not re.search(r'font\s*cache', warning_text, flags=re.IGNORECASE):
                    report_warnings += 1

    if raise_errors:
        assert_equal(report_errors, 0)

    if raise_warnings:
        assert_equal(report_warnings, 0) def check_run_prediction(source,
                         excluded=False,
                         file_format='csv',
                         given_test_dir=None,
                         config_obj_or_dict=None,
                         suppress_warnings_for=[]):
    """
    Run a parameterized rsmpredict experiment test.

    Parameters
    ----------
    source : str
        The name of the source directory containing the experiment
        configuration.
    excluded : bool, optional
        Whether to check the excluded responses file as part of the test.
        Defaults to ``False``.
    file_format : str, optional
        Which file format is being used for the output files of the experiment.
        Defaults to "csv".
    given_test_dir : str, optional
        Path where the test experiments are located. Unless specified, the
        rsmtool test directory is used. This can be useful when using these
        experiments to run tests for RSMExtra.
        Defaults to ``None``.
    config_obj_or_dict: configuration_parser.Configuration or dict, optional
        Configuration object or dictionary to use as an input.
        If ``None``, the function will construct a path to the config file
        using ``source`` and ``experiment_id``.
        Defaults to ``None``.
    suppress_warnings_for : list, optional
        Categories for which warnings should be suppressed when running the
        experiments.
        Defaults to ``[]``.
    """
    # use the test directory from this file unless it's been overridden
    test_dir = given_test_dir if given_test_dir else rsmtool_test_dir

    if config_obj_or_dict is None:
        config_input = join(test_dir,
                            'data',
                            'experiments',
                            source,
                            'rsmpredict.json')
    else:
        config_input = config_obj_or_dict

    do_run_prediction(source,
                      config_input,
                      suppress_warnings_for=suppress_warnings_for)

    output_dir = join('test_outputs', source, 'output')
    expected_output_dir = join(test_dir, 'data', 'experiments', source, 'output')

    output_files = ['predictions.{}'.format(file_format),
                    'preprocessed_features.{}'.format(file_format)]
    if excluded:
        output_files.append('predictions_excluded_responses.{}'.format(file_format))
    for output_file in output_files:
        generated_output_file = join(output_dir, output_file)
        expected_output_file = join(expected_output_dir, output_file)

        check_file_output(generated_output_file, expected_output_file) def check_scaled_coefficients(output_dir, experiment_id, file_format='csv'):
    """
    Check that predictions using scaled coefficients match scaled scores.

    Parameters
    ----------
    output_dir : str
         Path to the experiment output directory for a test.
    experiment_id : str
        The experiment ID.
    file_format : str, optional
        The format of the output files.
        Defaults to "csv".
        """
    preprocessed_test_file = join(output_dir,
                                  '{}_test_preprocessed_features.{}'.format(experiment_id,
                                                                            file_format))
    scaled_coefficients_file = join(output_dir,
                                    '{}_coefficients_scaled.{}'.format(experiment_id,
                                                                       file_format))
    predictions_file = join(output_dir,
                            '{}_pred_processed.{}'.format(experiment_id,
                                                          file_format))

    postprocessing_params_file = join(output_dir,
                                      '{}_postprocessing_params.{}'.format(experiment_id,
                                                                           file_format))

    postproc_params = DataReader.read_from_file(postprocessing_params_file).loc[0]
    df_preprocessed_test_data = DataReader.read_from_file(preprocessed_test_file)
    df_old_predictions = DataReader.read_from_file(predictions_file)
    df_old_predictions = df_old_predictions[['spkitemid', 'sc1', 'scale']]

    # create fake skll objects with new coefficients
    df_coef = DataReader.read_from_file(scaled_coefficients_file)
    learner = Modeler().create_fake_skll_learner(df_coef)
    modeler = Modeler.load_from_learner(learner)

    # generate new predictions and rename the prediction column to 'scale'
    df_new_predictions = modeler.predict(df_preprocessed_test_data,
                                         postproc_params['trim_min'],
                                         postproc_params['trim_max'])
    df_new_predictions.rename(columns={'raw': 'scale'}, inplace=True)

    # check that new predictions match the scaled old predictions
    assert_frame_equal(df_new_predictions.sort_index(axis=1),
                       df_old_predictions.sort_index(axis=1),
                       check_exact=False,
                       rtol=1e-03) def copy_data_files(temp_dir_name,
                    input_file_dict,
                    given_test_dir):
    """
    Copy files from given test directory to a temporary directory.

    Useful for tests where the current directory is to be used as the
    reference for resolving paths in the configuration.

    Parameters
    ----------
    temp_dir_name : str
        Name of the temporary directory.
    input_file_dict : dict
        A dictionary of files/directories to copy with keys as the
        file type and the values are their paths relative to the ``tests``
        directory.
    given_test_dir : str
        Directory where the the test experiments are located. This can be
        useful when using these experiments to run tests for RSMExtra.

    Returns
    -------
    output_file_dict : dict
        The dictionary with the same keys as
        ``input_file_dict`` and values being the copied paths.
    """
    temp_dir = Path(temp_dir_name)
    if not temp_dir.exists():
        temp_dir.mkdir()

    output_file_dict = {}
    for file in input_file_dict:
        filepath = Path(input_file_dict[file])
        filename = filepath.name
        old_filepath = given_test_dir / filepath
        new_filepath = temp_dir / filename
        if old_filepath.is_dir():
            copytree(old_filepath, new_filepath)
        else:
            copyfile(old_filepath, new_filepath)
        output_file_dict[file] = str(new_filepath)

    return output_file_dict def do_run_experiment(source,
                      experiment_id,
                      config_input,
                      suppress_warnings_for=[]):
    """
    Run rsmtool experiment automatically.

    Use the given experiment configuration file located in the
    given source directory and use the given experiment ID.

    Parameters
    ----------
    source : str
        Path to where the test experiment is located on disk.
    experiment_id : str
        Experiment ID to use when running.
    config_input : str or Configuration or dict
        Path to the experiment configuration file,
        or a ``configuration_parser.Configuration`` object
        or a dictionary with keys corresponding to fields in the
        configuration file.
    suppress_warnings_for : list, optional
        Categories for which warnings should be suppressed
        when running the experiments. Note that ``RuntimeWarning``s
        are always suppressed.
        Defaults to ``[]``.
    """
    source_output_dir = 'test_outputs'
    experiment_dir = join(source_output_dir, source)

    # remove all previously created files
    for output_subdir in ['output', 'figure', 'report']:
        files = glob(join(source_output_dir, source, output_subdir, '*'))
        for f in files:
            remove(f)

    with warnings.catch_warnings():

        # always suppress runtime warnings
        warnings.filterwarnings('ignore', category=RuntimeWarning)

        # suppress additional warning types if specified
        for warning_type in suppress_warnings_for:
            warnings.filterwarnings('ignore', category=warning_type)

        run_experiment(config_input, experiment_dir) def do_run_prediction(source,
                      config_input,
                      suppress_warnings_for=[]):
    """
    Run rsmpredict experiment automatically.

    Use the given experiment configuration file located in the given
    source directory.

    Parameters
    ----------
    source : str
        Path to where the test experiment is located on disk.
    config_input : str or Configuration or dict
        Path to the experiment configuration file,
        or a ``configuration_parser.Configuration`` object,
        or a dictionary with keys corresponding to fields in the
        configuration file
    suppress_warnings_for : list, optional
        Categories for which warnings should be suppressed when running the
        experiments.
        Defaults to ``[]``.
    """
    source_output_dir = 'test_outputs'

    # The `csv` file extension is ultimately dropped by the `rsmpredict.py`
    # script, so these arguments can be used for CSV, TSV, or XLSX output
    output_file = join(source_output_dir, source, 'output', 'predictions.csv')
    feats_file = join(source_output_dir, source, 'output', 'preprocessed_features.csv')

    # remove all previously created files
    files = glob(join(source_output_dir, 'output', '*'))
    for f in files:
        remove(f)

    with warnings.catch_warnings():

        # always suppress runtime warnings
        warnings.filterwarnings('ignore', category=RuntimeWarning)

        # suppress additional warning types if specified
        for warning_type in suppress_warnings_for:
            warnings.filterwarnings('ignore', category=warning_type)

        compute_and_save_predictions(config_input, output_file, feats_file)
import os
import tempfile
from glob import glob
from os import getcwd
from os.path import basename, exists, join
from nose.tools import raises
from parameterized import param, parameterized
from rsmtool import compute_and_save_predictions
from rsmtool.configuration_parser import Configuration
from rsmtool.test_utils import (check_file_output,
                                check_generated_output,
                                check_report,
                                check_run_prediction,
                                check_scaled_coefficients,
                                copy_data_files,
                                do_run_experiment,
                                do_run_prediction)
    from rsmtool.test_utils import rsmtool_test_dir

    old_file_dict = {'feature_file': 'data/files/test.csv',
                     'experiment_dir': 'data/experiments/lr-predict-dict/existing_experiment'}

    new_file_dict = copy_data_files(temp_dir.name,
                                    old_file_dict,
                                    rsmtool_test_dir)

    config_dict = {"id_column": "ID",
                   "input_features_file": new_file_dict['feature_file'],
                   "experiment_dir": new_file_dict['experiment_dir'],
                   "experiment_id": "lr"}

    check_run_prediction(source,
                         given_test_dir=rsmtool_test_dir,
                         config_obj_or_dict=config_dict)


@raises(ValueError)
def test_run_experiment_lr_predict_with_repeated_ids():

    # rsmpredict experiment with non-unique ids
    source = 'lr-predict-with-repeated-ids'
    config_file = join(rsmtool_test_dir,
                       'data',
                       'experiments',
                       source,
                       'rsmpredict.json')
    do_run_prediction(source, config_file)


@raises(ValueError)
def test_compute_predictions_wrong_input_format():
    config_list = [('experiment_id', 'AAAA'),
                   ('train_file', 'some_path')]
    with tempfile.TemporaryDirectory() as temp_dir:
        compute_and_save_predictions(config_list, temp_dir)


@raises(FileNotFoundError)
def test_run_experiment_lr_predict_missing_model_file():
    """Run rsmpredict experiment with missing model file."""
    source = 'lr-predict-missing-model-file'
    config_file = join(rsmtool_test_dir,
                       'data',
                       'experiments',
                       source,
                       'rsmpredict.json')
    do_run_prediction(source, config_file)


@raises(FileNotFoundError)
def test_run_experiment_lr_predict_missing_feature_file():
    """Run rsmpredict experiment with missing feature file."""
    source = 'lr-predict-missing-feature-file'
    config_file = join(rsmtool_test_dir,
                       'data',
                       'experiments',
                       source,
                       'rsmpredict.json')
    do_run_prediction(source, config_file)


@raises(FileNotFoundError)
def test_run_experiment_lr_predict_missing_postprocessing_file():
    """Run rsmpredict experiment with missing post-processing file."""
    source = 'lr-predict-missing-postprocessing-file'
    config_file = join(rsmtool_test_dir,
                       'data',
                       'experiments',
                       source,
                       'rsmpredict.json')
    do_run_prediction(source, config_file)


@raises(FileNotFoundError)
def test_run_experiment_predict_no_input_feature_file():
    """Run rsmpredict experiment with missing feature file."""
    source = 'lr-predict-no-input-feature-file'
    config_file = join(rsmtool_test_dir,
                       'data',
                       'experiments',
                       source,
                       'rsmpredict.json')
    do_run_prediction(source, config_file)


@raises(FileNotFoundError)
def test_run_experiment_predict_no_experiment_dir():
    """Run rsmpredict experiment with missing experiment dir."""
    source = 'lr-predict-no-experiment-dir'
    config_file = join(rsmtool_test_dir,
                       'data',
                       'experiments',
                       source,
                       'rsmpredict.json')
    do_run_prediction(source, config_file)


@raises(FileNotFoundError)
def test_run_experiment_predict_no_output_dir():
    """Run rsmpredict experiment with a missing "output" directory."""
    source = 'lr-predict-no-output-dir'
    config_file = join(rsmtool_test_dir,
                       'data',
                       'experiments',
                       source,
                       'rsmpredict.json')
    do_run_prediction(source, config_file)


@raises(FileNotFoundError)
def test_run_experiment_predict_no_experiment_id():
    """Run rsmpredict experiment with no experiment ID."""
    source = 'lr-predict-no-experiment-id'
    config_file = join(rsmtool_test_dir,
                       'data',
                       'experiments',
                       source,
                       'rsmpredict.json')

    do_run_prediction(source, config_file)