compare_perf_tests.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

# ===--- compare_perf_tests.py -------------------------------------------===//
#
#  This source file is part of the Swift.org open source project
#
#  Copyright (c) 2014 - 2017 Apple Inc. and the Swift project authors
#  Licensed under Apache License v2.0 with Runtime Library Exception
#
#  See https://swift.org/LICENSE.txt for license information
#  See https://swift.org/CONTRIBUTORS.txt for the list of Swift project authors
#
# ===---------------------------------------------------------------------===//
"""
This script compares performance test logs and issues a formatted report.

Invoke `$ compare_perf_tests.py -h ` for complete list of options.

class `Sample` is single benchmark measurement.
class `PerformanceTestSamples` is collection of `Sample`s and their statistics.
class `PerformanceTestResult` is a summary of performance test execution.
class `LogParser` converts log files into `PerformanceTestResult`s.
class `ResultComparison` compares new and old `PerformanceTestResult`s.
class `TestComparator` analyzes changes between the old and new test results.
class `ReportFormatter` creates the test comparison report in specified format.

"""

import argparse
import functools
import re
import sys
from bisect import bisect, bisect_left, bisect_right
from collections import namedtuple
from math import ceil, sqrt


class Sample(namedtuple("Sample", "i num_iters runtime")):
    u"""Single benchmark measurement.

    Initialized with:
    `i`: ordinal number of the sample taken,
    `num-num_iters`:  number or iterations used to compute it,
    `runtime`: in microseconds (μs).
    """

    def __repr__(self):
        """Shorter Sample formatting for debugging purposes."""
        return "s({0.i!r}, {0.num_iters!r}, {0.runtime!r})".format(self)


class Yield(namedtuple("Yield", "before_sample after")):
    u"""Meta-measurement of when the Benchmark_X voluntarily yielded process.

    `before_sample`: index of measurement taken just after returning from yield
    `after`: time elapsed since the previous yield in microseconds (μs)
    """


class PerformanceTestSamples(object):
    """Collection of runtime samples from the benchmark execution.

    Computes the sample population statistics.
    """

    def __init__(self, name, samples=None):
        """Initialize with benchmark name and optional list of Samples."""
        self.name = name  # Name of the performance test
        self.samples = []
        self.outliers = []
        self._runtimes = []
        self.mean = 0.0
        self.S_runtime = 0.0  # For computing running variance
        for sample in samples or []:
            self.add(sample)

    def __str__(self):
        """Text summary of benchmark statistics."""
        return (
            "{0.name!s} n={0.count!r} "
            "Min={0.min!r} Q1={0.q1!r} M={0.median!r} Q3={0.q3!r} "
            "Max={0.max!r} "
            "R={0.range!r} {0.spread:.2%} IQR={0.iqr!r} "
            "Mean={0.mean:.0f} SD={0.sd:.0f} CV={0.cv:.2%}".format(self)
            if self.samples
            else "{0.name!s} n=0".format(self)
        )

    def add(self, sample):
        """Add sample to collection and recompute statistics."""
        assert isinstance(sample, Sample)
        self._update_stats(sample)
        i = bisect(self._runtimes, sample.runtime)
        self._runtimes.insert(i, sample.runtime)
        self.samples.insert(i, sample)

    def _update_stats(self, sample):
        old_stats = (self.count, self.mean, self.S_runtime)
        _, self.mean, self.S_runtime = self.running_mean_variance(
            old_stats, sample.runtime
        )

    def exclude_outliers(self, top_only=False):
        """Exclude outliers by applying Interquartile Range Rule.

        Moves the samples outside of the inner fences
        (Q1 - 1.5*IQR and Q3 + 1.5*IQR) into outliers list and recomputes
        statistics for the remaining sample population. Optionally apply
        only the top inner fence, preserving the small outliers.

        Experimentally, this rule seems to perform well-enough on the
        benchmark runtimes in the microbenchmark range to filter out
        the environment noise caused by preemptive multitasking.
        """
        lo = (
            0
            if top_only
            else bisect_left(self._runtimes, int(self.q1 - 1.5 * self.iqr))
        )
        hi = bisect_right(self._runtimes, int(self.q3 + 1.5 * self.iqr))

        outliers = self.samples[:lo] + self.samples[hi:]
        samples = self.samples[lo:hi]

        self.__init__(self.name)  # re-initialize
        for sample in samples:  # and
            self.add(sample)  # re-compute stats
        self.outliers = outliers

    @property
    def count(self):
        """Number of samples used to compute the statistics."""
        return len(self.samples)

    @property
    def num_samples(self):
        """Number of all samples in the collection."""
        return len(self.samples) + len(self.outliers)

    @property
    def all_samples(self):
        """List of all samples in ascending order."""
        return sorted(self.samples + self.outliers, key=lambda s: s.i or -1)

    @property
    def min(self):
        """Minimum sampled value."""
        return self.samples[0].runtime

    @property
    def max(self):
        """Maximum sampled value."""
        return self.samples[-1].runtime

    def quantile(self, q):
        """Return runtime for given quantile.

        Equivalent to quantile estimate type R-1, SAS-3. See:
        https://en.wikipedia.org/wiki/Quantile#Estimating_quantiles_from_a_sample
        """
        index = max(0, int(ceil(self.count * float(q))) - 1)
        return self.samples[index].runtime

    @property
    def median(self):
        """Median sampled value."""
        return self.quantile(0.5)

    @property
    def q1(self):
        """First Quartile (25th Percentile)."""
        return self.quantile(0.25)

    @property
    def q3(self):
        """Third Quartile (75th Percentile)."""
        return self.quantile(0.75)

    @property
    def iqr(self):
        """Interquartile Range."""
        return self.q3 - self.q1

    @property
    def sd(self):
        u"""Standard Deviation (μs)."""
        return 0 if self.count < 2 else sqrt(self.S_runtime / (self.count - 1))

    @staticmethod
    def running_mean_variance(stats, x):
        """Compute running variance, B. P. Welford's method.

        See Knuth TAOCP vol 2, 3rd edition, page 232, or
        https://www.johndcook.com/blog/standard_deviation/
        M is mean, Standard Deviation is defined as sqrt(S/k-1)
        """

        (k, M_, S_) = stats

        k = float(k + 1)
        M = M_ + (x - M_) / k
        S = S_ + (x - M_) * (x - M)
        return (k, M, S)

    @property
    def cv(self):
        """Coefficient of Variation (%)."""
        return (self.sd / self.mean) if self.mean else 0

    @property
    def range(self):
        """Range of samples values (Max - Min)."""
        return self.max - self.min

    @property
    def spread(self):
        """Sample Spread; i.e. Range as (%) of Min."""
        return self.range / float(self.min) if self.min else 0


class PerformanceTestResult(object):
    u"""Result from executing an individual Swift Benchmark Suite benchmark.

    Reported by the test driver (Benchmark_O, Benchmark_Onone, Benchmark_Osize
    or Benchmark_Driver).

    It supports 2 log formats emitted by the test driver. Legacy format with
    statistics for normal distribution (MEAN, SD):
        #,TEST,SAMPLES,MIN(μs),MAX(μs),MEAN(μs),SD(μs),MEDIAN(μs),MAX_RSS(B)
    And new quantiles format with variable number of columns:
        #,TEST,SAMPLES,QMIN(μs),MEDIAN(μs),MAX(μs)
        #,TEST,SAMPLES,QMIN(μs),Q1(μs),Q2(μs),Q3(μs),MAX(μs),MAX_RSS(B)
    The number of columns between MIN and MAX depends on the test driver's
    `--quantile`parameter. In both cases, the last column, MAX_RSS is optional.
    """

    def __init__(self, csv_row, quantiles=False, memory=False, delta=False, meta=False):
        """Initialize from a row of multiple columns with benchmark summary.

        The row is an iterable, such as a row provided by the CSV parser.
        """
        self.test_num = csv_row[0]  # Ordinal number of the test
        self.name = csv_row[1]  # Name of the performance test
        self.num_samples = int(csv_row[2])  # Number of measurements taken

        mem_index = (-1 if memory else 0) + (-3 if meta else 0)
        if quantiles:  # Variable number of columns representing quantiles
            runtimes = csv_row[3:mem_index] if memory or meta else csv_row[3:]
            last_runtime_index = mem_index - 1
            if delta:
                runtimes = [int(x) if x else 0 for x in runtimes]
                runtimes = functools.reduce(
                    lambda l, x: l.append(l[-1] + x) or l if l else [x],  # runnin
                    runtimes,
                    None,
                )  # total
            num_values = len(runtimes)
            if self.num_samples < num_values:  # remove repeated samples
                quantile = num_values - 1
                qs = [float(i) / float(quantile) for i in range(0, num_values)]
                indices = [
                    max(0, int(ceil(self.num_samples * float(q))) - 1) for q in qs
                ]
                runtimes = [
                    runtimes[indices.index(i)] for i in range(0, self.num_samples)
                ]

            self.samples = PerformanceTestSamples(
                self.name, [Sample(None, None, int(runtime)) for runtime in runtimes]
            )
            self.samples.exclude_outliers(top_only=True)
            sams = self.samples
            self.min, self.max, self.median, self.mean, self.sd = (
                sams.min,
                sams.max,
                sams.median,
                sams.mean,
                sams.sd,
            )
        else:  # Legacy format with statistics for normal distribution.
            self.min = int(csv_row[3])  # Minimum runtime (μs)
            self.max = int(csv_row[4])  # Maximum runtime (μs)
            self.mean = float(csv_row[5])  # Mean (average) runtime (μs)
            self.sd = float(csv_row[6])  # Standard Deviation (μs)
            self.median = int(csv_row[7])  # Median runtime (μs)
            last_runtime_index = 7
            self.samples = None

        self.max_rss = (  # Maximum Resident Set Size (B)
            int(csv_row[mem_index]) if (
                memory and len(csv_row) > (last_runtime_index + 1)
            ) else None
        )

        # Optional measurement metadata. The number of:
        # memory pages used, involuntary context switches and voluntary yields
        self.mem_pages, self.involuntary_cs, self.yield_count = (
            [int(x) for x in csv_row[-3:]] if meta else (None, None, None)
        )
        self.yields = None
        self.setup = None

    def __repr__(self):
        """Short summary for debugging purposes."""
        return (
            "<PerformanceTestResult name:{0.name!r} "
            "samples:{0.num_samples!r} min:{0.min!r} max:{0.max!r} "
            "mean:{0.mean:.0f} sd:{0.sd:.0f} median:{0.median!r}>".format(self)
        )

    def merge(self, r):
        """Merge two results.

        Recomputes min, max and mean statistics. If all `samples` are
        available, it recomputes all the statistics.
        The use case here is comparing test results parsed from concatenated
        log files from multiple runs of benchmark driver.
        """
        # Statistics
        if self.samples and r.samples:
            for sample in r.samples.samples:
                self.samples.add(sample)
            sams = self.samples
            self.num_samples = sams.num_samples
            self.min, self.max, self.median, self.mean, self.sd = (
                sams.min,
                sams.max,
                sams.median,
                sams.mean,
                sams.sd,
            )
        else:
            self.min = min(self.min, r.min)
            self.max = max(self.max, r.max)
            self.mean = (  # pooled mean is the weighted sum of means
                (self.mean * self.num_samples) + (r.mean * r.num_samples)
            ) / float(self.num_samples + r.num_samples)
            self.num_samples += r.num_samples
            self.median, self.sd = None, None

        # Metadata
        def minimum(a, b):  # work around None being less than everything
            return min(filter(lambda x: x is not None, [a, b])) if any([a, b]) else None

        self.max_rss = minimum(self.max_rss, r.max_rss)
        self.setup = minimum(self.setup, r.setup)


class ResultComparison(object):
    """ResultComparison compares MINs from new and old PerformanceTestResult.

    It computes speedup ratio and improvement delta (%).
    """

    def __init__(self, old, new):
        """Initialize with old and new `PerformanceTestResult`s to compare."""
        self.old = old
        self.new = new
        assert old.name == new.name
        self.name = old.name  # Test name, convenience accessor

        # Speedup ratio
        self.ratio = (old.min + 0.001) / (new.min + 0.001)

        # Test runtime improvement in %
        ratio = (new.min + 0.001) / (old.min + 0.001)
        self.delta = (ratio - 1) * 100

        # Indication of dubious changes: when result's MIN falls inside the
        # (MIN, MAX) interval of result they are being compared with.
        self.is_dubious = (old.min < new.min and new.min < old.max) or (
            new.min < old.min and old.min < new.max
        )


class LogParser(object):
    """Converts log outputs into `PerformanceTestResult`s.

    Supports various formats produced by the `Benchmark_Driver` and
    `Benchmark_O`('Onone', 'Osize'). It can also merge together the
    results from concatenated log files.
    """

    def __init__(self):
        """Create instance of `LogParser`."""
        self.results = []
        self.quantiles, self.delta, self.memory = False, False, False
        self.meta = False
        self._reset()

    def _reset(self):
        """Reset parser to the default state for reading a new result."""
        self.samples, self.yields, self.num_iters = [], [], 1
        self.setup, self.max_rss, self.mem_pages = None, None, None
        self.voluntary_cs, self.involuntary_cs = None, None

    # Parse lines like this
    # #,TEST,SAMPLES,MIN(μs),MAX(μs),MEAN(μs),SD(μs),MEDIAN(μs)
    results_re = re.compile(
        r"( *\d+[, \t]+[\w.\-\?!]+[, \t]+"
        + r"[, \t]+".join([r"\d+"] * 2)  # #,TEST
        + r"(?:[, \t]+\d*)*)"  # at least 2...
    )  # ...or more numeric columns

    def _append_result(self, result):
        columns = result.split(",") if "," in result else result.split()
        r = PerformanceTestResult(
            columns,
            quantiles=self.quantiles,
            memory=self.memory,
            delta=self.delta,
            meta=self.meta,
        )
        r.setup = self.setup
        r.max_rss = r.max_rss or self.max_rss
        r.mem_pages = r.mem_pages or self.mem_pages
        r.voluntary_cs = self.voluntary_cs
        r.involuntary_cs = r.involuntary_cs or self.involuntary_cs
        if self.samples:
            r.samples = PerformanceTestSamples(r.name, self.samples)
            r.samples.exclude_outliers()
        self.results.append(r)
        r.yields = self.yields or None
        self._reset()

    def _store_memory_stats(self, max_rss, mem_pages):
        self.max_rss = int(max_rss)
        self.mem_pages = int(mem_pages)

    def _configure_format(self, header):
        self.quantiles = "QMIN" in header
        self.memory = "MAX_RSS" in header
        self.meta = "PAGES" in header
        self.delta = "𝚫" in header

    # Regular expression and action to take when it matches the parsed line
    state_actions = {
        results_re: _append_result,
        # Verbose mode adds new productions:
        # Adaptively determined N; test loop multiple adjusting runtime to ~1s
        re.compile(r"\s+Measuring with scale (\d+)."): (
            lambda self, num_iters: setattr(self, "num_iters", num_iters)
        ),
        re.compile(r"\s+Sample (\d+),(\d+)"): (
            lambda self, i, runtime: self.samples.append(
                Sample(int(i), int(self.num_iters), int(runtime))
            )
        ),
        re.compile(r"\s+SetUp (\d+)"): (
            lambda self, setup: setattr(self, "setup", int(setup))
        ),
        re.compile(r"\s+Yielding after ~(\d+) μs"): (
            lambda self, since_last_yield: self.yields.append(
                Yield(len(self.samples), int(since_last_yield))
            )
        ),
        re.compile(r"( *#[, \t]+TEST[, \t]+SAMPLES[, \t].*)"): _configure_format,
        # Environmental statistics: memory usage and context switches
        re.compile(
            r"\s+MAX_RSS \d+ - \d+ = (\d+) \((\d+) pages\)"
        ): _store_memory_stats,
        re.compile(r"\s+VCS \d+ - \d+ = (\d+)"): (
            lambda self, vcs: setattr(self, "voluntary_cs", int(vcs))
        ),
        re.compile(r"\s+ICS \d+ - \d+ = (\d+)"): (
            lambda self, ics: setattr(self, "involuntary_cs", int(ics))
        ),
    }

    def parse_results(self, lines):
        """Parse results from the lines of the log output from Benchmark*.

        Returns a list of `PerformanceTestResult`s.
        """
        for line in lines:
            for regexp, action in LogParser.state_actions.items():
                match = regexp.match(line)
                if match:
                    action(self, *match.groups())
                    break  # stop after 1st match
            else:  # If none matches, skip the line.
                # print('skipping: ' + line.rstrip('\n'))
                continue
        return self.results

    @staticmethod
    def _results_from_lines(lines):
        tests = LogParser().parse_results(lines)

        def add_or_merge(names, r):
            if r.name not in names:
                names[r.name] = r
            else:
                names[r.name].merge(r)
            return names

        return functools.reduce(add_or_merge, tests, dict())

    @staticmethod
    def results_from_string(log_contents):
        """Parse `PerformanceTestResult`s from the supplied string.

        Returns dictionary of test names and `PerformanceTestResult`s.
        """
        return LogParser._results_from_lines(log_contents.splitlines())

    @staticmethod
    def results_from_file(log_file):
        """Parse `PerformanceTestResult`s from the log file.

        Returns dictionary of test names and `PerformanceTestResult`s.
        """
        with open(log_file) as f:
            return LogParser._results_from_lines(f.readlines())


class TestComparator(object):
    """Analyzes changes between the old and new test results.

    It determines which tests were `added`, `removed` and which can be
    compared. It then splits the `ResultComparison`s into 3 groups according to
    the `delta_threshold` by the change in performance: `increased`,
    `descreased` and `unchanged`. Whole computation is performed during
    initialization and results are provided as properties on this object.

    The lists of `added`, `removed` and `unchanged` tests are sorted
    alphabetically. The `increased` and `decreased` lists are sorted in
    descending order by the amount of change.
    """

    def __init__(self, old_results, new_results, delta_threshold):
        """Initialize with dictionaries of old and new benchmark results.

        Dictionary keys are benchmark names, values are
        `PerformanceTestResult`s.
        """
        old_tests = set(old_results.keys())
        new_tests = set(new_results.keys())
        comparable_tests = new_tests.intersection(old_tests)
        added_tests = new_tests.difference(old_tests)
        removed_tests = old_tests.difference(new_tests)

        self.added = sorted([new_results[t] for t in added_tests], key=lambda r: r.name)
        self.removed = sorted(
            [old_results[t] for t in removed_tests], key=lambda r: r.name
        )

        def compare(name):
            return ResultComparison(old_results[name], new_results[name])

        comparisons = list(map(compare, comparable_tests))

        def partition(items, p):
            return functools.reduce(
                lambda x, y: x[not p(y)].append(y) or x, items, ([], [])
            )

        decreased, not_decreased = partition(
            comparisons, lambda c: c.ratio < (1 - delta_threshold)
        )
        increased, unchanged = partition(
            not_decreased, lambda c: c.ratio > (1 + delta_threshold)
        )

        # sorted partitions
        names = [c.name for c in comparisons]
        comparisons = dict(zip(names, comparisons))
        self.decreased = [
            comparisons[c.name] for c in sorted(decreased, key=lambda c: -c.delta)
        ]
        self.increased = [
            comparisons[c.name] for c in sorted(increased, key=lambda c: c.delta)
        ]
        self.unchanged = [
            comparisons[c.name] for c in sorted(unchanged, key=lambda c: c.name)
        ]


class ReportFormatter(object):
    """Creates the report from performance test comparison in specified format.

    `ReportFormatter` formats the `PerformanceTestResult`s and
    `ResultComparison`s provided by `TestComparator` into report table.
    Supported formats are: `markdown` (used for displaying benchmark results on
    GitHub), `git` and `html`.
    """

    def __init__(self, comparator, changes_only, single_table=False):
        """Initialize with `TestComparator` and names of branches."""
        self.comparator = comparator
        self.changes_only = changes_only
        self.single_table = single_table

    PERFORMANCE_TEST_RESULT_HEADER = ("TEST", "MIN", "MAX", "MEAN", "MAX_RSS")
    RESULT_COMPARISON_HEADER = ("TEST", "OLD", "NEW", "DELTA", "RATIO")

    @staticmethod
    def header_for(result):
        """Column labels for header row in results table."""
        return (
            ReportFormatter.PERFORMANCE_TEST_RESULT_HEADER
            if isinstance(result, PerformanceTestResult)
            else
            # isinstance(result, ResultComparison)
            ReportFormatter.RESULT_COMPARISON_HEADER
        )

    @staticmethod
    def values(result):
        """Format values from PerformanceTestResult or ResultComparison.

        Returns tuple of strings to display in the results table.
        """
        return (
            (
                result.name,
                str(result.min),
                str(result.max),
                str(int(result.mean)),
                str(result.max_rss) if result.max_rss else "—",
            )
            if isinstance(result, PerformanceTestResult)
            else
            # isinstance(result, ResultComparison)
            (
                result.name,
                str(result.old.min),
                str(result.new.min),
                "{0:+.1f}%".format(result.delta),
                "{0:.2f}x{1}".format(result.ratio, " (?)" if result.is_dubious else ""),
            )
        )

    def markdown(self):
        """Report results of benchmark comparisons in Markdown format."""
        return self._formatted_text(
            label_formatter=lambda s: ("**" + s + "**"),
            COLUMN_SEPARATOR=" | ",
            DELIMITER_ROW=([":---"] + ["---:"] * 4),
            SEPARATOR="&nbsp; | | | | \n",
            SECTION="""
<details {3}>
  <summary>{0} ({1})</summary>
  {2}
</details>
""",
        )

    def git(self):
        """Report results of benchmark comparisons in 'git' format."""
        return self._formatted_text(
            label_formatter=lambda s: s.upper(),
            COLUMN_SEPARATOR="   ",
            DELIMITER_ROW=None,
            SEPARATOR="\n",
            SECTION="""
{0} ({1}): \n{2}""",
        )

    def _column_widths(self):
        changed = self.comparator.decreased + self.comparator.increased
        results = changed if self.changes_only else changed + self.comparator.unchanged
        results += self.comparator.added + self.comparator.removed

        widths = [
            map(len, columns)
            for columns in [
                ReportFormatter.PERFORMANCE_TEST_RESULT_HEADER,
                ReportFormatter.RESULT_COMPARISON_HEADER,
            ]
            + [ReportFormatter.values(r) for r in results]
        ]

        def max_widths(maximum, widths):
            return map(max, zip(maximum, widths))

        return list(functools.reduce(max_widths, widths, [0] * 5))

    def _formatted_text(
        self, label_formatter, COLUMN_SEPARATOR, DELIMITER_ROW, SEPARATOR, SECTION
    ):
        widths = self._column_widths()
        self.header_printed = False

        def justify_columns(contents):
            return [c.ljust(w) for w, c in zip(widths, contents)]

        def row(contents):
            return (
                ""
                if not contents
                else COLUMN_SEPARATOR.join(justify_columns(contents)) + "\n"
            )

        def header(title, column_labels):
            labels = (
                column_labels
                if not self.single_table
                else map(label_formatter, (title,) + column_labels[1:])
            )
            h = (
                ("" if not self.header_printed else SEPARATOR)
                + row(labels)
                + (row(DELIMITER_ROW) if not self.header_printed else "")
            )
            if self.single_table and not self.header_printed:
                self.header_printed = True
            return h

        def format_columns(r, is_strong):
            return r if not is_strong else r[:-1] + ("**" + r[-1] + "**",)

        def table(title, results, is_strong=False, is_open=False):
            if not results:
                return ""
            rows = [
                row(format_columns(ReportFormatter.values(r), is_strong))
                for r in results
            ]
            table = header(
                title if self.single_table else "",
                ReportFormatter.header_for(results[0]),
            ) + "".join(rows)
            return (
                table
                if self.single_table
                else SECTION.format(
                    title, len(results), table, "open" if is_open else ""
                )
            )

        return "\n" + "".join(
            [
                table("Regression", self.comparator.decreased, True, True),
                table("Improvement", self.comparator.increased, True),
                (
                    ""
                    if self.changes_only
                    else table("No Changes", self.comparator.unchanged)
                ),
                table("Added", self.comparator.added, is_open=True),
                table("Removed", self.comparator.removed, is_open=True),
            ]
        )

    HTML = """
<!DOCTYPE html>
<html>
<head>
    <meta http-equiv="Content-Type" content="text/html; charset=utf-8" />
    <style>
        body {{ font-family: -apple-system, sans-serif; font-size: 14px; }}
        table {{ border-spacing: 2px; border-color: gray; border-spacing: 0;
                border-collapse: collapse; }}
        table tr {{ background-color: #fff; border-top: 1px solid #c6cbd1; }}
        table th, table td {{ padding: 6px 13px; border: 1px solid #dfe2e5; }}
        th {{ text-align: center; padding-top: 130px; }}
        td {{ text-align: right; }}
        table td:first-child {{ text-align: left; }}
        tr:nth-child(even) {{ background-color: #000000; }}
        tr:nth-child(2n) {{ background-color: #f6f8fa; }}
    </style>
</head>
<body>
<table>
{0}
</table>
</body>
</html>"""

    HTML_HEADER_ROW = """
        <tr>
                <th align='left'>{0} ({1})</th>
                <th align='left'>{2}</th>
                <th align='left'>{3}</th>
                <th align='left'>{4}</th>
                <th align='left'>{5}</th>
        </tr>
"""

    HTML_ROW = """
        <tr>
                <td align='left'>{0}</td>
                <td align='left'>{1}</td>
                <td align='left'>{2}</td>
                <td align='left'>{3}</td>
                <td align='left'><font color='{4}'>{5}</font></td>
        </tr>
"""

    def html(self):
        """Report results of benchmark comparisons in HTML format."""

        def row(name, old, new, delta, speedup, speedup_color):
            return self.HTML_ROW.format(name, old, new, delta, speedup_color, speedup)

        def header(contents):
            return self.HTML_HEADER_ROW.format(*contents)

        def table(title, results, speedup_color):
            rows = [
                row(*(ReportFormatter.values(r) + (speedup_color,))) for r in results
            ]
            return (
                ""
                if not rows
                else header(
                    (title, len(results)) + ReportFormatter.header_for(results[0])[1:]
                )
                + "".join(rows)
            )

        return self.HTML.format(
            "".join(
                [
                    table("Regression", self.comparator.decreased, "red"),
                    table("Improvement", self.comparator.increased, "green"),
                    (
                        ""
                        if self.changes_only
                        else table("No Changes", self.comparator.unchanged, "black")
                    ),
                    table("Added", self.comparator.added, ""),
                    table("Removed", self.comparator.removed, ""),
                ]
            )
        )


def parse_args(args):
    """Parse command line arguments and set default values."""
    parser = argparse.ArgumentParser(description="Compare Performance tests.")
    parser.add_argument(
        "--old-file", help="Baseline performance test suite (csv file)", required=True
    )
    parser.add_argument(
        "--new-file", help="New performance test suite (csv file)", required=True
    )
    parser.add_argument(
        "--format",
        choices=["markdown", "git", "html"],
        help="Output format. Default is markdown.",
        default="markdown",
    )
    parser.add_argument("--output", help="Output file name")
    parser.add_argument(
        "--changes-only", help="Output only affected tests", action="store_true"
    )
    parser.add_argument(
        "--single-table",
        help="Combine data in a single table in git and markdown formats",
        action="store_true",
    )
    parser.add_argument(
        "--delta-threshold",
        help="Delta threshold. Default 0.05.",
        type=float,
        default=0.05,
    )
    return parser.parse_args(args)


def create_report(
    old_results,
    new_results,
    delta_threshold,
    format,
    changes_only=True,
    single_table=True,
):
    comparator = TestComparator(old_results, new_results, delta_threshold)
    formatter = ReportFormatter(comparator, changes_only, single_table)
    formats = {
        "markdown": formatter.markdown,
        "git": formatter.git,
        "html": formatter.html,
    }

    report = formats[format]()
    return report


def main():
    """Compare benchmarks for changes in a formatted report."""
    args = parse_args(sys.argv[1:])
    report = create_report(
        LogParser.results_from_file(args.old_file),
        LogParser.results_from_file(args.new_file),
        args.delta_threshold,
        args.format,
        args.changes_only,
        args.single_table,
    )
    print(report)

    if args.output:
        with open(args.output, "w") as f:
            f.write(report)


if __name__ == "__main__":
    sys.exit(main())