distrcmp.py

#!/usr/bin/env python3

import argparse
import logging
import statistics
import sys
from typing import Optional

from respdiff import cli
from respdiff.stats import Stats, SummaryStatistics

DEFAULT_COEF = 2


def belongs_to_distribution(
    ref: Optional[Stats], new: Optional[Stats], coef: float
) -> bool:
    if ref is None or new is None:
        return False
    median = statistics.median(new.samples)
    if median > statistics.mean(ref.samples) + coef * statistics.stdev(ref.samples):
        return False
    if median < statistics.mean(ref.samples) - coef * statistics.stdev(ref.samples):
        return False
    return True


def belongs_to_all(ref: SummaryStatistics, new: SummaryStatistics, coef: float) -> bool:
    if not belongs_to_distribution(
        ref.target_disagreements, new.target_disagreements, coef
    ):
        return False
    if not belongs_to_distribution(ref.upstream_unstable, new.upstream_unstable, coef):
        return False
    if not belongs_to_distribution(ref.not_reproducible, new.not_reproducible, coef):
        return False
    if ref.fields is not None and new.fields is not None:
        for field in ref.fields:
            if not belongs_to_distribution(
                ref.fields[field].total, new.fields[field].total, coef
            ):
                return False
    return True


def main():
    cli.setup_logging()
    parser = argparse.ArgumentParser(
        description="Check if new samples belong to reference "
        "distribution. Ends with exitcode 0 if belong, "
        "1 if not,"
    )
    parser.add_argument(
        "-r",
        "--reference",
        type=cli.read_stats,
        help="json statistics file with reference data",
    )
    cli.add_arg_report_filename(parser)
    parser.add_argument(
        "-c",
        "--coef",
        type=float,
        default=DEFAULT_COEF,
        help=(
            "coeficient for comparation (new belongs to refference if "
            "its median is closer than COEF * standart deviation of reference "
            "from reference mean) (default: {})".format(DEFAULT_COEF)
        ),
    )
    args = parser.parse_args()
    reports = cli.get_reports_from_filenames(args)

    try:
        newstats = SummaryStatistics(reports)
    except ValueError as exc:
        logging.critical(exc)
        sys.exit(2)

    if not belongs_to_all(args.reference, newstats, args.coef):
        sys.exit(1)
    sys.exit(0)


if __name__ == "__main__":
    main()