<repo_name>cdpowell/test-project<file_sep>/mwFileStatusWebsite/__init__.py
from . import validator, constructor, compare


__version__ = '1.0.0'
<file_sep>/setup.py
#!/usr/bin/env python3
# -*- coding: utf-8 -*-

import os
import sys
import re
from setuptools import setup, find_packages


if sys.argv[-1] == 'publish':
    os.system('python3 setup.py sdist')
    os.system('twine upload dist/*')
    sys.exit()


def readme():
    with open('README.md') as readme_file:
        return readme_file.read()


def find_version():
    with open('mwFileStatusWebsite/__init__.py', 'r') as fd:
        version = re.search(r'^__version__\s*=\s*[\'"]([^\'"]*)[\'"]',
                            fd.read(), re.MULTILINE).group(1)
    if not version:
        raise RuntimeError('Cannot find version information')
    return version


REQUIRES = [
    "mwtab >= 1.2.4",
    "pytest"
]


setup(
    name='mwFileStatusWebsite',
    version=find_version(),
    author='<NAME>',
    author_email='<EMAIL>',
    description='Supplemental database for the Metabolomics Workbench data repository',
    keywords='metabolomics workbench',
    license='BSD',
    packages=find_packages(),
    platforms='any',
    long_description=readme(),
    install_requires=REQUIRES,
    classifiers=[
        'Development Status :: 4 - Beta',
        'Environment :: Console',
        'Intended Audience :: Developers',
        'Intended Audience :: Science/Research',
        'License :: OSI Approved :: BSD License',
        'Operating System :: OS Independent',
        'Programming Language :: Python :: 3.5',
        'Programming Language :: Python :: 3.6',
        'Programming Language :: Python :: 3.7',
        'Programming Language :: Python :: 3.8',
        'Programming Language :: Python :: 3.9',
        'Programming Language :: Python :: 3.10',
        'Topic :: Scientific/Engineering :: Bio-Informatics',
        'Topic :: Software Development :: Libraries :: Python Modules',
    ],
)
<file_sep>/mwFileStatusWebsite/constructor.py
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
constructor.py
~~~~~~~~~~~~~~~~

This script contains methods for generating HTML pages from validation dictionaries.
"""
import json
from datetime import datetime
import pkgutil


MESSAGE_COLOR = {
    "Passing": "brightgreen",
    "Validation Error": "orange",
    "Parsing Error": "red",
    "Missing/Blank": "lightgrey",
    'Consistent': 'brightgreen',
    'Inconsistent': 'orange',
    'Not Checked': 'lightgrey'
}
MESSAGE_TO_LEVEL = {
    "Passing": 0,
    "Validation Error": 1,
    "Parsing Error": 2,
    "Missing/Blank": 3,
}
LEVEL_TO_MESSAGE = {
    MESSAGE_TO_LEVEL[k]: k for k in MESSAGE_TO_LEVEL
}
# load in all the necessary HTML templates
INDEX_HEADER_TEMPLATE = pkgutil.get_data(__name__, 'templates/index_header_template.txt').decode('utf-8')
INDEX_TEMPLATE = pkgutil.get_data(__name__, 'templates/index_template.txt').decode('utf-8')
VAL_STATS_TEMPLATE = pkgutil.get_data(__name__, 'templates/statistics_template.txt').decode('utf-8')
COMP_STATS_TEMPLATE = pkgutil.get_data(__name__, 'templates/comparison_stats_template.txt').decode('utf-8')
HEADER_TEMPLATE = pkgutil.get_data(__name__, 'templates/header_template.txt').decode('utf-8')
GRID_TEMPLATE = pkgutil.get_data(__name__, 'templates/grid_template.txt').decode('utf-8')
GRID_ITEM_TEMPLATE = pkgutil.get_data(__name__, 'templates/grid_item_template.txt').decode('utf-8')
BADGE_TEMPLATE = pkgutil.get_data(__name__, 'templates/badge_template.txt').decode('utf-8')
DESC_TEMPLATE = "<div class=\"desc__grid__item\"{0}>{1}</div>"


def load_json(filepath):
    """Help function for loading in JSON data files.

    :param filepath: Path to JSON file to be loaded.
    :type filepath: str
    :return: JSON dictionary object.
    :rtype: dict
    """
    with open(filepath, "r") as fh:
        json_dict = json.loads(fh.read())
        fh.close()

    return json_dict


def generate_validation_stats_summary(validation_dict):
    """Method for generating the statistics to filling the HTML template with the current Metabolomics Workbench mwTab
    files validation data.

    :param validation_dict: Dictionary object containing the validation statuses for all validated mwTab data files.
    :type validation_dict: dict
    :return: Tuple containing the number of validated studies, number of validated analyses, and the dictionary
    containing the validation data (eg. number Passing, number with Validation Errors, etc.).
    :rtype: tuple
    """
    num_studies = 0
    num_analyses = 0
    error_num_dict = {
        key: {"txt": 0, "json": 0} for key in ["Passing", "Parsing Error", "Validation Error", "Missing/Blank"]
    }

    for study_id in validation_dict:
        num_studies += 1

        for analysis_id in validation_dict[study_id]["analyses"]:
            num_analyses += 1

            for file_format in ('txt', 'json'):
                error_num_dict[validation_dict[study_id]["analyses"][analysis_id]["status"][file_format]][file_format] += 1

    return num_studies, num_analyses, error_num_dict


def generate_comparison_stats_summary(validation_dict):
    """Method for generating the statistics to filling the HTML template with the current Metabolomics Workbench mwTab
    files comparison data.

    :param validation_dict: Dictionary object containing the validation statuses for all validated mwTab data files.
    :type validation_dict: dict
    :return: Tuple containing the number of validated studies, number of validated analyses, and the dictionary
    containing the validation data (eg. number Passing, number with Validation Errors, etc.).
    :rtype: tuple
    """
    count_dict = {
        'Consistent': 0,
        'Inconsistent': 0,
        'Not Checked': 0,
    }

    for study_id in validation_dict:
        for analysis_id in validation_dict[study_id]['analyses']:
            count_dict[validation_dict[study_id]['analyses'][analysis_id]['status']['comparison']] += 1

    return count_dict['Consistent'], count_dict['Inconsistent'], count_dict['Not Checked']


def create_desc(params, tabs="\t"*6):
    """Method for parsing out STUDY or ANALYSIS description items from the generated validation nested dictionary and
    returning a formatted string.

    :param params: Nested validation dictionary or section of nested validation dictionary.
    :type params: dict
    :param tabs: White spacing for tabs.
    :type tabs: str
    :return: String containing formatted
    :rtype: str
    """
    desc_items = list()
    for k in params:
        desc_items.append(tabs + DESC_TEMPLATE.format("", k))
        desc_items.append(tabs + DESC_TEMPLATE.format(" style=\"white-space:nowrap;overflow:hidden;text-overflow:ellipsis;width:calc(100%);\"", params.get(k)))

    return "\n".join(desc_items)


def create_html(validation_dict, config_dict, output_filename):
    """Creates and saves HTML file based on given validation and config dictionaries.

    :param validation_dict: Structured dictionary containing analyses statuses and other study information.
    :type validation_dict: dict
    :param config_dict: Dictionary containing GitHub repository information.
    :type config_dict: dict
    :param output_filename: Filename of HTML file to be created.
    :type output_filename: str
    :return: None
    """
    with open(output_filename, "w") as fh:

        #####################################
        # write the HTML header information #
        #####################################
        fh.write(INDEX_HEADER_TEMPLATE.format(
            config_dict['owner'],
            config_dict['repo'],
            str(datetime.now()),
        ))

        #####################################################
        # collect and write validation and comparison stats #
        #####################################################
        # collect general statistics for the run (number of available studies and analyses).
        num_studies, num_analyses, error_dict = generate_validation_stats_summary(validation_dict)

        # Fill out the statistics_template and comparison_stats_template.
        num_errors = list()
        for error_type in error_dict:
            for file_format in error_dict[error_type]:
                num_errors.append(error_dict[error_type][file_format])

        # writes the validation and comparison stats sections to the HTML file
        fh.write(VAL_STATS_TEMPLATE.format(num_studies, num_analyses, *num_errors, config_dict['owner'], config_dict['repo']))
        fh.write(COMP_STATS_TEMPLATE.format(*generate_comparison_stats_summary(validation_dict)))

        ################################
        # generate file status section #
        ################################
        # file_status_list = []
        for study_id in validation_dict:

            # # Add space between grid items
            # if file_status_list:
            #     file_status_list.append()

            # Add study header
            # Adds header line (grid)
            # Adds study meta data
            height = 1*len(validation_dict[study_id]["params"])
            fh.write(HEADER_TEMPLATE.format(
                study_id,
                validation_dict[study_id]["params"].get("STUDY_TITLE"),
                validation_dict[study_id]["params"].get("INSTITUTE"),
                validation_dict[study_id]["params"].get("LAST_NAME"),
                validation_dict[study_id]["params"].get("FIRST_NAME"),
                " style=\"height:" + str(height) + "em;max-height:" + str(height) + "\"",
                create_desc(validation_dict[study_id]["params"])
            ))

            grid_item_list = []
            for analysis_id in validation_dict[study_id]["analyses"]:

                badge_list = []
                for format_type in validation_dict[study_id]["analyses"][analysis_id]["status"]:

                    badge_list.append(BADGE_TEMPLATE.format(
                        analysis_id,
                        format_type,
                        MESSAGE_COLOR[validation_dict[study_id]["analyses"][analysis_id]["status"][format_type]],
                        validation_dict[study_id]["analyses"][analysis_id]["status"][format_type],
                        config_dict['owner'],
                        config_dict['repo']
                    ))

                # adds the colored analysis button
                grid_item_list.append(GRID_ITEM_TEMPLATE.format(
                    analysis_id,
                    MESSAGE_COLOR[LEVEL_TO_MESSAGE[max([
                        MESSAGE_TO_LEVEL[value] for value in validation_dict[study_id]["analyses"][analysis_id]["status"].values() if value in MESSAGE_TO_LEVEL.keys()
                    ])]],
                    "\n".join(badge_list),
                    create_desc(validation_dict[study_id]["analyses"][analysis_id]["params"])
                ))

            fh.write(GRID_TEMPLATE.format("\n".join(grid_item_list)))

            fh.write("\t\t\t<br>")

        # close file
        fh.write("\t\t</div>\n\t</body>\n</html>\n")


def create_error_dicts(validation_dict, status_str, file_format=None):
    """Method for creating a dictionary containing the validation status and additional parameters of analyses with
    indicated validation status.

    :param validation_dict: Structured dictionary containing analyses statuses and other study information.
    :type validation_dict: dict
    :param status_str: Analysis validation status to be searched for.
    :type status_str: str
    :param file_format: Indicates which file format to be searched (if only one).
    :type file_format: str or bool
    :return: Structured dictionary containing analyses statuses and other study information for analyses with indicated
    validation status.
    :rtype: dict
    """
    status_dict = dict()

    for study_id in validation_dict:
        for analysis_id in validation_dict[study_id]["analyses"]:

            # both file formats have the same validation status
            if not file_format:
                file_format_status_set = {validation_dict[study_id]["analyses"][analysis_id]["status"]['json'],
                                          validation_dict[study_id]["analyses"][analysis_id]["status"]['txt']}
                if {status_str} == file_format_status_set:
                    status_dict.setdefault(study_id, dict()).setdefault("params", validation_dict[study_id]["params"])
                    status_dict[study_id].setdefault("analyses", dict())[analysis_id] = \
                        validation_dict[study_id]["analyses"][analysis_id]

            # TODO: Allow specifying which file format to be checked

            # only one file format has a given status
            else:
                if status_str in set(validation_dict[study_id]["analyses"][analysis_id]["status"].values()):
                    status_dict.setdefault(study_id, dict()).setdefault("params", validation_dict[study_id]["params"])
                    status_dict[study_id].setdefault("analyses", dict())[analysis_id] =  \
                        validation_dict[study_id]["analyses"][analysis_id]

    return status_dict
<file_sep>/mwFileStatusWebsite/validator.py
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
validator.py
~~~~~~~~~~~~

This script contains methods for validating all available mwTab data files, creating log files, and generating a JSON
file containing the validation metadata.
"""
import mwFileStatusWebsite.compare
import mwtab
import json
import re
from datetime import datetime
from os import walk
from os.path import join, isdir, isfile
from time import sleep


MW_REST_URL = "https://www.metabolomicsworkbench.org/rest/study/analysis_id/{}/mwtab/{}"
SLEEP_TIME = 1
NUM_TRIES = 3


def retrieve_mwtab_files(verbose=False):
    """Method for retrieving a dictionary of Metabolomics Workbench file identifiers.

    Example:
    {
        'ST000001': [
            'AN000001',
            ...
        ],
        ...
    }

    :param bool verbose: Run in verbose mode.
    :return: Dictionary of study IDs (keys) and their associated lists of analysis IDs (values).
    """
    try:
        study_analysis_dict = mwtab.mwrest._pull_study_analysis()

        # count the number of present studies and analyses
        num_studies = len(study_analysis_dict)
        num_analyses = sum([len(study_analysis_dict[k]) for k in study_analysis_dict])
        if verbose:
            print("{} studies found".format(num_studies))
            print("{} analyses found".format(num_analyses))

        return study_analysis_dict

    # cannot access Metabolomics Workbench REST server
    except Exception as e:
        print("Unable to access https://www.metabolomicsworkbench.org/")
        print(e)
        exit()


def create_validation_dict(study_analysis_dict):
    """Method for creating a null structured validation dictionary.

    :param study_analysis_dict: Dictionary of Metabolomics study IDs (key) and lists of their associated analysis IDs
    (value).
    :type study_analysis_dict: dict

    :return: Returns a structured dictionary to contain analysis validation statuses and associated study data.
    :rtype: dict
    """
    # setup some variables for collecting validation statuses
    validation_dict = {
        study_id: {
            "params": {},  # TODO:Change to STUDY_params
            "analyses": {
                analysis_id: {
                    "params": {
                        "ANALYSIS_ID": analysis_id
                    },
                    "status": {
                        "txt": None,
                        "json": None
                    }
                } for analysis_id in sorted(study_analysis_dict[study_id])
            }
        } for study_id in sorted(study_analysis_dict.keys())
    }
    return validation_dict


def _validate(validation_dict, study_id, analysis_id, file_format, save_path=None):
    """Helper function for performing validation of a specified mwTab data file given the files; study ID, analysis ID,
    and file format (.txt or .json).

    :param validation_dict: Structured dictionary containing analyses statuses and other study information.
    :type validation_dict: dict
    :param study_id: Metabolomics Workbench study ID string (eg. ST000001).
    :type study_id: str
    :param analysis_id: Metabolomics Workbench analysis ID string (eg. AN000001).
    :type analysis_id: str
    :param file_format: File format extension string (either: 'txt' or 'json').
    :type file_format: str
    :param save_path: Boolean value indicating if retrieved analyses should be save.
    :type save_path: bool or str

    :return: Tuple containing of the validated mwtab file object and the string validation log.
    :rtype: tuple
    """
    mwtabfile = next(mwtab.read_files(MW_REST_URL.format(analysis_id, file_format)))

    # allows saving out the retrieved non-validated mwTab analysis files.
    if save_path:
        with open(join(save_path, analysis_id + '.' + file_format), 'w') as fh:
            mwtabfile.write(fh, 'mwtab' if file_format == 'txt' else 'json')

    sleep(SLEEP_TIME)

    # TODO: Allow validating METABOLITES section
    validated_mwtabfile, validation_log = mwtab.validate_file(mwtabfile, metabolites=False)

    # parse validation status (e.g. "Passing") from validation log
    status_str = re.search(r'Status.*', validation_log).group(0).split(': ')[1]
    if status_str == 'Passing':
        validation_dict[study_id]["analyses"][analysis_id]["status"][file_format] = "Passing"
    elif status_str == 'Contains Validation Errors':
        validation_dict[study_id]["analyses"][analysis_id]["status"][file_format] = "Validation Error"

    # parse out STUDY block parameters
    if not validation_dict[study_id]["params"]:
        validation_dict[study_id]["params"] = mwtabfile["STUDY"]

    return validated_mwtabfile, validation_log


def validate(validation_dict, study_id, analysis_id, file_format, save_path=None):
    """Method for validating a given Metabolomics Workbench mwTab file.

    Creates a validation log and adds validation status to the given validation_dict dictionary. Fetches files using the
    ``mwtab`` Python3 packages built in methods for accessing Metabolomics Workbench's REST API. Performs three attempts
    to retrieve a file before labeling it as "Missing/Blank".

    :param validation_dict: Structured dictionary containing analyses statuses and other study information.
    :type validation_dict: dict
    :param study_id: Metabolomics Workbench study ID string (eg. ST000001).
    :type study_id: str
    :param analysis_id: Metabolomics Workbench analysis ID string (eg. AN000001).
    :type analysis_id: str
    :param file_format: File format extension string (either: 'txt' or 'json').
    :type file_format: str
    :param save_path: Boolean value indicating if retrieved analyses should be save.
    :type save_path: bool or str

    :return: Tuple containing of the validated mwtab file object and the string validation log.
    :rtype: tuple
    """
    error = False

    try:
        validated_mwtabfile, validation_log = _validate(validation_dict, study_id, analysis_id, file_format, save_path)

    except Exception as e:
        # error is one of; 1) temporary server error, 2) source is blank, or 3) source cannot be parsed

        # check to see if temporary server error
        error = True
        for x in range(NUM_TRIES):  # try three times to see if there is a temporary server error
            try:
                validated_mwtabfile, validation_log = _validate(validation_dict, study_id, analysis_id, file_format)
                error = False
                break
            except Exception:
                pass

        # not temporary server error, either source is blank or source cannot be parsed
        if error:
            # blank source given
            if type(e) == ValueError and e.args[0] == "Blank input string retrieved from source.":
                validation_dict[study_id]["analyses"][analysis_id]["status"][file_format] = "Missing/Blank"
            else:
                validation_dict[study_id]["analyses"][analysis_id]["status"][file_format] = "Parsing Error"

            # create validation log for missing or unparsable files.
            validation_log = mwtab.validator.VALIDATION_LOG_HEADER.format(
                str(datetime.now()),
                mwtab.__version__,
                MW_REST_URL.format(analysis_id, file_format),
                study_id,
                analysis_id,
                file_format
            )
            validation_log += \
                "\nStatus:" + \
                validation_dict[study_id]["analyses"][analysis_id]["status"][file_format] + \
                "\n" + str(e)

    if not error:
        return validated_mwtabfile, validation_log
    else:
        return {}, validation_log


def validate_mwtab_rest(input_dict=None, logs_path='docs/validation_logs', output_file="tmp.json",
                        verbose=False, save_path=None):
    """Method for validating all available Metabolomics Workbench mwTab formatted data files.

    :param input_dict: Dictionary of Metabolomics study IDs (key) and lists of their associated analysis IDs (value).
    :type input_dict: dict
    :param logs_path: File path to the directory validation logs are to be saved to.
    :type logs_path: str
    :param output_file: File path for the structured dictionary containing analyses statuses and other study information
    to be saved to.
    :type output_file: str
    :param verbose: Run in verbose mode.
    :type verbose: bool
    :param save_path: Directory path for retrieved Metabolomics Workbench analysis data files to be saved in.
    :type save_path: str
    :return: Structured dictionary containing analyses statuses and other study information.
    :rtype: dict
    """
    if verbose:
        print("Running mwTab file validation.")
        print("\tmwtab version:", mwtab.__version__)

    # collect dictionary of studies and their analyses
    if input_dict:
        study_analysis_dict = input_dict
    else:
        study_analysis_dict = retrieve_mwtab_files(verbose)

    # create the validation dict
    validation_dict = create_validation_dict(study_analysis_dict)

    for study_id in sorted(study_analysis_dict.keys()):

        if verbose:
            print("Validating study:", study_id)

        for analysis_id in study_analysis_dict[study_id]:

            if verbose:
                print("\t", analysis_id)

            # retrieve file in both its 'txt' and 'json' formats
            txt_mwtab_file, txt_validation_log = validate(validation_dict, study_id, analysis_id, 'txt', save_path=save_path)
            json_mwtab_file, json_validation_log = validate(validation_dict, study_id, analysis_id, 'json', save_path=save_path)

            # if both formats are available and parsable, compare the two files
            validation_dict[study_id]["analyses"][analysis_id]["status"]['comparison'] = 'Not Checked'
            if txt_mwtab_file and json_mwtab_file:  # both files passed validation and can be compared
                comparison_list = mwFileStatusWebsite.compare.compare(txt_mwtab_file, json_mwtab_file)

                if comparison_list:
                    comparison_status = 'Inconsistent'
                    error_str = '\n'.join([str(error) for error in comparison_list])
                else:
                    comparison_status = 'Consistent'
                    error_str = ''

                validation_dict[study_id]["analyses"][analysis_id]["status"]['comparison'] = comparison_status
                comparison_log = mwFileStatusWebsite.compare.COMPARISON_LOG.format(
                    str(datetime.now()),
                    mwtab.__version__,
                    MW_REST_URL.format(analysis_id, '...'),
                    study_id,
                    analysis_id,
                    comparison_status
                ) + error_str

                with open(join(logs_path, '{}_comparison.log').format(analysis_id), 'w') as fh:
                    fh.write(comparison_log)

            # save out each files validation log
            with open(join(logs_path, '{}_{}.log'.format(analysis_id, 'txt')), 'w') as fh:
                fh.write(txt_validation_log)
            with open(join(logs_path, '{}_{}.log'.format(analysis_id, 'json')), 'w') as fh:
                fh.write(json_validation_log)

    # export validation status dictionary
    with open(output_file, "w") as fh:
        fh.write(json.dumps(validation_dict, indent=4))

    return validation_dict


def validate_mwtab_files(file_path=None, logs_path='docs/validation_logs', output_file="tmp.json", verbose=False,):
    """Validate and create a validation dictionary for a single Metabolomics Workbench analysis data file (mwTab or
    JSON) or a directory of data files.

    :param file_path: File path to directory containing Metabolomics Workbench analysis data files to be validated.
    :type file_path: str
    :param logs_path: File path to the directory validation logs are to be saved to.
    :type logs_path: str
    :param output_file: File path for the structured dictionary containing analyses statuses and other study information
    to be saved to.
    :type output_file: str
    :param verbose: Run in verbose mode.
    :type verbose: bool
    :return: Structured dictionary containing analyses statuses and other study information.
    :rtype: dict
    """
    # check to see if a directory or file was passed in
    if isdir(file_path):
        _, _, filenames = next(walk(file_path))
        analysis_ids = sorted({filename.split('.')[0] for filename in filenames})
    elif isfile(file_path):
        analysis_ids = []
    else:
        raise ValueError("'file_path' is not a valid directory or file.")

    validation_dict = dict()
    for analysis_id in analysis_ids:
        pass
<file_sep>/mwFileStatusWebsite/cli.py
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
The mwFileStatusWebsite command-line interface
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~

Usage:
    mwFileStatusWebsite -h | --help
    mwFileStatusWebsite --version
    mwFileStatusWebsite validate [--to-path=<path>]
    mwFileStatusWebsite generate

Options:
    -h, --help                      Show this screen.
    --version                       Show version.
    --verbose                       Enable verbose processing.
    --to-path=<path>                Path to directory for retrieved analyses to be saved in.
"""
from . import validator, constructor


def cli(cmdargs):

    if cmdargs['validate']:
        validator.validate_mwtab_rest(save_path=cmdargs['--to-path'])

    elif cmdargs['generate']:
        # create the main webpage (index.html)
        validation_dict = constructor.load_json('tmp.json')
        config_dict = constructor.load_json('docs/config.json')
        constructor.create_html(validation_dict, config_dict, 'index.html')

        # create the passing.html page
        # contains only analyses which one or both formats (mwTab and JSON) are passing
        passing = constructor.create_error_dicts(validation_dict, 'Passing', True)
        constructor.create_html(passing, config_dict, 'passing.html')

        # create the validation_error.html page
        # contains analyses which both formats (mwTab and JSON) have validation errors.
        validation = constructor.create_error_dicts(validation_dict, 'Validation Error')
        constructor.create_html(validation, config_dict, 'validation_error.html')

        # create the parsing_error.html page
        # contains analyses which both formats (mwTab and JSON) have parsing errors.
        parsing = constructor.create_error_dicts(validation_dict, 'Parsing Error')
        constructor.create_html(parsing, config_dict, 'parsing_error.html')

        # create the missing.html page
        # contains analyses which both formats (mwTab and JSON) are missing.
        missing = constructor.create_error_dicts(validation_dict, 'Missing/Blank')
        constructor.create_html(missing, config_dict, 'missing.html')
<file_sep>/tests/test_compare.py
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
test_compare.py
~~~~~~~~~~~~~~~

This script contains methods comparing two Metabolomics Workbench file in `~mwtab.mwtab.MWTabFile` object and assert
that the files (objects) are the same.
"""
import pytest
import mwFileStatusWebsite
import mwtab
from urllib.request import urlopen
from time import sleep
import pkgutil
from collections import OrderedDict


@pytest.mark.parametrize('mwtabfile_1, mwtabfile_2, error', [
    ({"a": {"b": 1}}, {"a": {"b": 1}}, False),  # same blocks, same items
    ({"a": {"b": 1}}, {"a": {"c": 1}}, False),  # same blocks, different item blocks
    ({"a": {"b": 1}}, {"b": {"b": 1}}, True),  # different blocks
    ({"a": {"b": 1}}, {"a": {"b": 1}, "c": {"b": 1}}, True), # one file has an additional block
    ({"a": {"b": 1}}, {}, True),  # one file is null
])
def test_compare_blocks(mwtabfile_1, mwtabfile_2, error):
    try:
        mwFileStatusWebsite.compare.compare_blocks(mwtabfile_1, mwtabfile_2)
        if error:  # test should have raise an error
            assert False
    except Exception as e:
        if not error:
            assert False


@pytest.mark.parametrize('mwtabfile_1, mwtabfile_2, error', [
    ({"PROJECT": {"b": 1}}, {"PROJECT": {"b": 1}}, False),  # same blocks, same items
    ({"PROJECT": {"b": 1}}, {"PROJECT": {"c": 1}}, True),  # same blocks, different item blocks
    ({"PROJECT": {"b": 1}}, {"PROJECT": {}}, True),  # same blocks, one block is null
    ({"PROJECT": {"b": 1}}, {"PROJECT": {"b": 1}, "STUDY": {"a": 1}}, False),  # different blocks, same items in mutual blocks
])
def test_compare_block_items(mwtabfile_1, mwtabfile_2, error):
    errors = mwFileStatusWebsite.compare.compare_block_items(mwtabfile_1, mwtabfile_2)
    if bool(errors) != error:
        assert False


@pytest.mark.parametrize('mwtabfile_1, mwtabfile_2, error', [
    # same SSF section
    (
        {'SUBJECT_SAMPLE_FACTORS':[
            OrderedDict([('Subject ID', '-'), ('Sample ID', '1'), ('Factors', {'Factor1': 'Blah'})])
        ]},
        {'SUBJECT_SAMPLE_FACTORS': [
            OrderedDict([('Subject ID', '-'), ('Sample ID', '1'), ('Factors', {'Factor1': 'Blah'})])
        ]},
        False
    ),
    # different 'Sample ID'
    (
        {'SUBJECT_SAMPLE_FACTORS': [
            OrderedDict([('Subject ID', '-'), ('Sample ID', '1'), ('Factors', {'Factor1': 'Blah'})])
        ]},
        {'SUBJECT_SAMPLE_FACTORS': [
            OrderedDict([('Subject ID', '-'), ('Sample ID', '2'), ('Factors', {'Factor1': 'Blah'})])
        ]},
        True
    ),
    # different factor values
    (
        {'SUBJECT_SAMPLE_FACTORS': [
            OrderedDict([('Subject ID', '-'), ('Sample ID', '1'), ('Factors', {'Factor1': 'Blah'})])
        ]},
        {'SUBJECT_SAMPLE_FACTORS': [
            OrderedDict([('Subject ID', '-'), ('Sample ID', '1'), ('Factors', {'Factor1': 'Meh'})])
        ]},
        True
    )
])
def test_compare_block_items(mwtabfile_1, mwtabfile_2, error):
    try:
        mwFileStatusWebsite.compare.compare_subject_sample_factors(mwtabfile_1, mwtabfile_2)
        if error:  # test should have raise an error
            assert False
    except Exception as e:
        if not error:
            assert False


@pytest.mark.parametrize('mwtabfile_1, mwtabfile_2, error', [
    # same "_DATA" section
    (
        {'MS_METABOLITE_DATA': {
            'Units': 'test',
            'Data': [OrderedDict([('Metabolite', 'Test'), ('Subject_1', 1000)])],
            'Metabolites': [OrderedDict([('Metabolite', 'Test'),('Value_1', 'Test')])],
        }},
        {'MS_METABOLITE_DATA': {
            'Units': 'test',
            'Data': [OrderedDict([('Metabolite', 'Test'), ('Subject_1', 1000)])],
            'Metabolites': [OrderedDict([('Metabolite', 'Test'), ('Value_1', 'Test')])],
        }},
        False
    ),

    # different "_DATA" sections
    (
        {'MS_METABOLITE_DATA': {
            'Units': 'test',
            'Data': [OrderedDict([('Metabolite', 'Test'), ('Subject_1', 1000)])],
            'Metabolites': [OrderedDict([('Metabolite', 'Test'), ('Value_1', 'Test')])],
        }},
        {'MNMR_BINNED_DATA': {
            'Units': 'test',
            'Data': [OrderedDict([('Metabolite', 'Test'), ('Subject_1', 1000)])],
            'Metabolites': [OrderedDict([('Metabolite', 'Test'), ('Value_1', 'Test')])],
        }},
        True
    ),

    # different "Units" subsections
    (
        {'MS_METABOLITE_DATA': {
            'Units': 'test',
            'Data': [OrderedDict([('Metabolite', 'Test'), ('Subject_1', 1000)])],
            'Metabolites': [OrderedDict([('Metabolite', 'Test'), ('Value_1', 'Test')])],
        }},
        {'MNMR_BINNED_DATA': {
            'Units': 'test2',
            'Data': [OrderedDict([('Metabolite', 'Test'), ('Subject_1', 1000)])],
            'Metabolites': [OrderedDict([('Metabolite', 'Test'), ('Value_1', 'Test')])],
        }},
        True
    ),

    # different "Data" sections
    (
        {'MS_METABOLITE_DATA': {
            'Units': 'test',
            'Data': [OrderedDict([('Metabolite', 'Test'), ('Subject_1', 1000)])],
            'Metabolites': [OrderedDict([('Metabolite', 'Test'), ('Value_1', 'Test')])],
        }},
        {'MNMR_BINNED_DATA': {
            'Units': 'test',
            'Data': [OrderedDict([('Metabolite', 'Test2'), ('Subject_1', 1000)])],
            'Metabolites': [OrderedDict([('Metabolite', 'Test'), ('Value_1', 'Test')])],
        }},
        True
    ),

    # different "Metabolites" sections
    (
        {'MS_METABOLITE_DATA': {
            'Units': 'test',
            'Data': [OrderedDict([('Metabolite', 'Test'), ('Subject_1', 1000)])],
            'Metabolites': [OrderedDict([('Metabolite', 'Test'), ('Value_1', 'Test')])],
        }},
        {'MNMR_BINNED_DATA': {
            'Units': 'test',
            'Data': [OrderedDict([('Metabolite', 'Test'), ('Subject_1', 1000)])],
            'Metabolites': [OrderedDict([('Metabolite', 'Test2'), ('Value_1', 'Test')])],
        }},
        True
    ),
])
def test_compare_data(mwtabfile_1, mwtabfile_2, error):
    errors = mwFileStatusWebsite.compare.compare_data(mwtabfile_1, mwtabfile_2)

    if errors and not error:
        assert False

    if not errors and error:
        assert False
<file_sep>/mwFileStatusWebsite/compare.py
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
compare.py
~~~~~~~~~~

This script contains methods comparing two Metabolomics Workbench file in `~mwtab.mwtab.MWTabFile` object and assert
that the files (objects) are the same.

TODO: This should be migrated into the mwtab Python library as a ~mwtab.mwtab.MWTabFile.equals() functionality or
TODO: similar.
"""
import operator


ITEM_SECTIONS = {
    # "METABOLOMICS WORKBENCH",
    "PROJECT",
    "STUDY",
    "ANALYSIS",
    "SUBJECT",
    "COLLECTION",
    "TREATMENT",
    "SAMPLEPREP",
    "CHROMATOGRAPHY",
    "MS",
    "NM"
}

DATA_BLOCKS = {
    'MS_METABOLITE_DATA',
    'NMR_BINNED_DATA',
    'NMR_METABOLITE_DATA'
}

COMPARISON_LOG = \
"""Comparison Log
{}
mwtab Python Library Version: {}
Source:      {}
Study ID:    {}
Analysis ID: {}
Status:      {}

"""


def compare_blocks(mwtabfile_1, mwtabfile_2):
    """Method for comparing the section headers of two Metabolomics Workbench file in `~mwtab.mwtab.MWTabFile` object
    format. Raises AssertionError if false.

    :param mwtabfile_1: First Metabolomics Workbench file in mwtab format to be compared.
    :type mwtabfile_1: :py:class:`~mwtab.mwtab.MWTabFile`
    :param mwtabfile_2: Second Metabolomics Workbench file in mwtab format to be compared.
    :type mwtabfile_2: :py:class:`~mwtab.mwtab.MWTabFile`
    :return: None
    """
    if not mwtabfile_1.keys() == mwtabfile_2.keys():
        raise AssertionError(
            "mwTab files contain different blocks: \"{}\"".format(
                mwtabfile_1.keys() ^ mwtabfile_2.keys()  # symmetric difference
            )
        )


def compare_block_items(mwtabfile_1, mwtabfile_2):
    """Method for comparing the items (keys and values) present within the sections of two Metabolomics Workbench file
    in `~mwtab.mwtab.MWTabFile` object format.

    :param mwtabfile_1: First Metabolomics Workbench file in mwtab format to be compared.
    :type mwtabfile_1: :py:class:`~mwtab.mwtab.MWTabFile`
    :param mwtabfile_2: Second Metabolomics Workbench file in mwtab format to be compared.
    :type mwtabfile_2: :py:class:`~mwtab.mwtab.MWTabFile`
    :return: List of errors.
    :rtype: list
    """
    error_list = list()

    for section_key in set(mwtabfile_1.keys()) & set(mwtabfile_2.keys()) & ITEM_SECTIONS:
        try:
            if mwtabfile_1[section_key].items() ^ mwtabfile_2[section_key].items():
                error_list.append(
                    AssertionError(
                        "Sections \"{}\" contain missmatched items: {}".format(
                            section_key,
                            mwtabfile_1[section_key].items() ^ mwtabfile_2[section_key].items()
                        )
                    )
                )
        except Exception as e:
            error_list.append(e)

    return error_list


def compare_subject_sample_factors(mwtabfile_1, mwtabfile_2):
    """Method for comparing the items (keys and values) present within the sections of two Metabolomics Workbench file
    in `~mwtab.mwtab.MWTabFile` object format.

    :param mwtabfile_1: First Metabolomics Workbench file in mwtab format to be compared.
    :type mwtabfile_1: :py:class:`~mwtab.mwtab.MWTabFile`
    :param mwtabfile_2: Second Metabolomics Workbench file in mwtab format to be compared.
    :type mwtabfile_2: :py:class:`~mwtab.mwtab.MWTabFile`
    :return: None
    """
    new_mwtabfile_1_SSF_list = sorted(mwtabfile_1['SUBJECT_SAMPLE_FACTORS'], key=operator.itemgetter('Sample ID'))
    new_mwtabfile_2_SSF_list = sorted(mwtabfile_2['SUBJECT_SAMPLE_FACTORS'], key=operator.itemgetter('Sample ID'))

    if new_mwtabfile_1_SSF_list != new_mwtabfile_2_SSF_list:
        raise AssertionError("mwTab files contain different 'SUBJECT_SAMPLE_FACTORS' sections.")


def compare_data(mwtabfile_1, mwtabfile_2):
    """Method for comparing "_DATA" blocks present within two Metabolomics Workbench file in `~mwtab.mwtab.MWTabFile`
    object format.

    :param mwtabfile_1: First Metabolomics Workbench file in mwtab format to be compared.
    :type mwtabfile_1: :py:class:`~mwtab.mwtab.MWTabFile`
    :param mwtabfile_2: Second Metabolomics Workbench file in mwtab format to be compared.
    :type mwtabfile_2: :py:class:`~mwtab.mwtab.MWTabFile`
    :return: List of errors.
    :rtype: list
    """
    data_section = set(mwtabfile_1.keys()) & set(mwtabfile_2.keys()) & DATA_BLOCKS

    # data block is not consistently present across files
    if not data_section or type(data_section) != set:
        return [AssertionError("Unable to find '_DATA' block in given files.")]

    else:
        data_section = list(data_section)[0]
        error_list = list()
        subsections = set(mwtabfile_1[data_section].keys()) & set(mwtabfile_2[data_section].keys())

        if mwtabfile_1[data_section].keys() != mwtabfile_2[data_section].keys():
            error_list.append(AssertionError("'_DATA' blocks do not contain the same subsections: {}".format(
                set(mwtabfile_1[data_section].keys()) ^ set(mwtabfile_2[data_section].keys())
            )))

        # compare "Units"
        if 'Units' in subsections:
            if mwtabfile_1[data_section]['Units'] != mwtabfile_2[data_section]['Units']:
                error_list.append(AssertionError("'Units' section of '{}' block do not match.".format(data_section)))

        # compare "Metabolites"
        if 'Metabolites' in subsections:
            try:
                new_mwtabfile_1_metabolites_list = sorted(mwtabfile_1[data_section]['Metabolites'],
                                                          key=operator.itemgetter('Metabolite'))
                new_mwtabfile_2_metabolites_list = sorted(mwtabfile_2[data_section]['Metabolites'],
                                                          key=operator.itemgetter('Metabolite'))
                if new_mwtabfile_1_metabolites_list != new_mwtabfile_2_metabolites_list:
                    error_list.append(AssertionError("'Metabolites' section of '{}' block do not match.".format(data_section)))
            except Exception as e:
                error_list.append(e)

        # compare "Data"
        if 'Data' in subsections:
            try:
                new_mwtabfile_1_data_list = sorted(mwtabfile_1[data_section]['Data'],
                                                          key=operator.itemgetter('Metabolite'))
                new_mwtabfile_2_data_list = sorted(mwtabfile_2[data_section]['Data'],
                                                          key=operator.itemgetter('Metabolite'))
                if new_mwtabfile_1_data_list != new_mwtabfile_2_data_list:
                    error_list.append(AssertionError("'Data' section of '{}' block do not match.".format(data_section)))
            except Exception as e:
                error_list.append(e)

        return error_list


def compare(mwtabfile_1, mwtabfile_2):
    """Method for comparing two Metabolomics Workbench file in `~mwtab.mwtab.MWTabFile` object format to assert that
    they are the same. Raises AssertionError if false.

    :param mwtabfile_1: First Metabolomics Workbench file in mwtab format to be compared.
    :type mwtabfile_1: :py:class:`~mwtab.mwtab.MWTabFile`
    :param mwtabfile_2: Second Metabolomics Workbench file in mwtab format to be compared.
    :type mwtabfile_2: :py:class:`~mwtab.mwtab.MWTabFile`
    :return: Log of assertion errors.
    """
    error_list = list()

    # compare files to assert they have the same section headings.
    try:
        compare_blocks(mwtabfile_1, mwtabfile_2)
    except Exception as e:
        error_list.append(e)

    # compare files to assert they have the same items in each  appropriate block (e.g. PROJECT, STUDY, etc.).
    error_list.extend(compare_block_items(mwtabfile_1, mwtabfile_2))  # no issue if compare_block_keys() returns a null list.

    # check that the SUBJECT_SAMPLE_FACTORS sections match
    try:
        compare_subject_sample_factors(mwtabfile_1, mwtabfile_2)
    except Exception as e:
        error_list.append(e)

    error_list.extend(compare_data(mwtabfile_1, mwtabfile_2))

    return error_list
<file_sep>/mw_website.sh
#!/bin/bash
# A Shell Script to update the Metabolomics Workbench File Status website
# <NAME> - 2022/02/20

###########################
# change directory to tmp #
###########################
cd /tmp

###################
# start ssh agent #
###################
eval "$(ssh-agent -s)"
ssh-add /root/.ssh/id_ed25519_github

##############################
# clone  mw file status repo #
##############################
git clone <EMAIL>:MoseleyBioinformaticsLab/mwFileStatusWebsite.git
cd mwFileStatusWebsite/

###################################################
# install virtualenv and create a new environment #
###################################################
python3 -m pip install virtualenv
python3 -m virtualenv venv
source venv/bin/activate
python3 -m pip install mwtab
python3 -m pip install -e ../mwFilesStatusWebsite

########################
# update website files #
########################
#python3 -m validator/validator.py
#python3 -m validator/constructor.py
python3 -m mwFileStatusWebsite validate
python3 -m mwFileStatusWebsite generate

###################
# start ssh agent #
###################
eval "$(ssh-agent -s)"
ssh-add /root/.ssh/id_ed25519_github

#############################
# add updated files to repo #
#############################
git add docs/validation_logs index.html missing.html parsing_error.html passing.html validation_error.html
now=$(date +'%Y/%m/%d')
git commit -m "Weekly update for $now"
git push

#####################
# remove everything #
#####################
cd ..
rm -rf mwFileStatusWebsite/
<file_sep>/README.md
# Metabolomics Workbench File Status Website

This repository holds the scripts required for creating 
and updating the Metabolomics Workbench File Status 
website along with the validation longs for each 
available analysis in both `JSON` and `mwTab` file 
formats.


## Usage

The Metabolomics Workbench File Status website can be 
accessed through it's GitHub Pages hosted webpage at: 

https://moseleybioinformaticslab.github.io/mwFileStatusWebsite/

Individual validation files can be accessed in the 
[docs/validation_logs/](https://github.com/moseleybioinformaticslab/mwFileStatusWebsite/tree/master/docs/validation_logs) 
directory in this repository.


## Installation

The `mwFilesStatus` website can be run locally but is intended to be setup on a Virtual Machine for automated updating.


### Basic Installation

Clone the repository
```bash
git clone <EMAIL>:MoseleyBioinformaticsLab/mwFileStatusWebsite.git
cd mwFileStatusWebsite
```

Create a virtual environment and install the `mwFileStatusWebsite` package locally through `pip`.
```python
python3 -m virtualenv venv
source venv/bin/activate
python3 -m pip install -e ../mwFileStatusWebsite
```


### Installation for Virtual Machine Usage

Coming soon


## Usage

### Commandline Usage

```bash
mwFilesStatusWebsite validate
mwFilesStatusWebsite generate
```

### Python3 Usage

```bash
python3 -m mwFilesStatusWebsite validate
python3 -m mwFilesStatusWebsite generate
```

## License

This package is distributed under the [BSD](https://choosealicense.com/licenses/bsd-3-clause-clear/) `license`.
<file_sep>/tests/test_validator.py
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
test_validator.py
~~~~~~~~~~~~~~~~~

This script contains methods for validating all available mwTab data files, creating log files, and generating a JSON
file containing the validation metadata.
"""
import pytest
import mwFileStatusWebsite
import mwtab
from urllib.request import urlopen
from time import sleep
import pkgutil


FULL_STUDY_ANALYSIS_DICT = pkgutil.get_data(__name__, '/test/study_analysis_dict.json')


# @pytest.mark.parametrize('verbose', [
#     False,
#     True
# ])
# def test_retrieve_mwtab_files(verbose):
#     assert mwFileStatusWebsite.validator.retrieve_mwtab_files(verbose)


# def test_validate_mwtab_files():
#     """Method for testing the ``validate_mwtab_files()`` method in the ``validator`` module.
#
#     :return:
#     """
#     # search for analysis missing `txt` through Metabolomics Workbench REST API
#     txt_missing_ground_truth = set()
#     study_analysis_dict = mwtab.mwrest._pull_study_analysis()
#     for study_id in study_analysis_dict.keys():
#         for analysis_id in study_analysis_dict[study_id]:
#             response = urlopen(mwFileStatusWebsite.validator.MW_REST_URL.format(analysis_id, 'txt'))
#             sleep(1)  # don't piss off Metabolomics Workbench
#             response_str = response.read().decode('utf-8')
#             if response_str:
#                 pass
#             else:
#                 txt_missing_ground_truth.add(analysis_id)
#
#     # search for analysis missing `json` through Metabolomics Workbench REST API
#     json_missing_ground_truth = set()
#     study_analysis_dict = mwtab.mwrest._pull_study_analysis()
#     for study_id in study_analysis_dict.keys():
#         for analysis_id in study_analysis_dict[study_id]:
#             response = urlopen(mwFileStatusWebsite.validator.MW_REST_URL.format(analysis_id, 'json'))
#             sleep(1)  # don't piss off Metabolomics Workbench
#             response_str = response.read().decode('utf-8')
#             if response_str:
#                 pass
#             else:
#                 json_missing_ground_truth.add(analysis_id)
#
#     # test validator method
#     validation_dict = mwFileStatusWebsite.validator.validate_mwtab_files(study_analysis_dict)
#     txt_missing = set()
#     json_missing = set()
#     for study_id in validation_dict.keys():
#         for analysis_id in validation_dict[study_id]['analyses']:
#             if validation_dict[study_id]['analyses'][analysis_id]['status']['txt'] == 'Missing/Blank':
#                 txt_missing.add(analysis_id)
#             if validation_dict[study_id]['analyses'][analysis_id]['status']['json'] == 'Missing/Blank':
#                 json_missing.add(analysis_id)
#
#     # assert number of missing detected by ``validator.py`` is same as ground truth
#     assert txt_missing == txt_missing_ground_truth
#     assert json_missing == json_missing_ground_truth


@pytest.mark.parametrize('input_list', [
    [{'ST000001': ['AN000001']}, 'Passing', 'txt'],
    [{'ST000001': ['AN000001']}, 'Passing', 'json']
])
def test_validate_mwtab_files_hardcoded(input_list):
    """Hardcoded method for testing the ``validate_mwtab_files()`` method in the ``validator`` module.

    :return:
    """
    study_id = list(input_list[0].keys())[0]
    analysis_id = input_list[0][study_id][0]
    validation_dict = mwFileStatusWebsite.validator.validate_mwtab_files(input_dict=input_list[0])

    assert validation_dict[study_id]['analyses'][analysis_id]['status'][input_list[2]]

<file_sep>/main.py
import mwtab
from datetime import datetime
import re
import mwFileStatusWebsite
import json


VERBOSE = True


if __name__ == '__main__':
    # study_analysis_dict = validator.retrieve_mwtab_files(True)
    # with open('files.json', 'w') as fh:
    #     fh.write(json.dumps(study_analysis_dict))

    try:
        mwtabfile = next(mwtab.read_files(mwFileStatusWebsite.MW_REST_URL.format('AN000012', 'txt')))
        validated_mwtabfile, validation_log = mwtab.validate_file(mwtabfile, metabolites=False)
    except Exception as e:
        print(e)
