Blame for src/benchmarks_analysis.py

"""
Business Source License 1.1

Copyright (C) 2025 Raman Marozau, raman@worktif.com
Use of this software is governed by the Business Source License included in the LICENSE file and at www.mariadb.com/bsl11.

Change Date: Never
On the date above, in accordance with the Business Source License, use of this software will be governed by the open source license specified in the LICENSE file.
Additional Use Grant: Free for personal and non-commercial research use only.

SPDX-License-Identifier: BUSL-1.1
"""

import datetime
import os

import json
import warnings
from pathlib import Path
from typing import List

import numpy as np
import pandas as pd
from matplotlib import pyplot as plt
from pandas.core.interchange.dataframe_protocol import DataFrame
from scipy.signal import savgol_filter


class BenchmarksAnalysis:
    """
    Class for managing the analysis of benchmarks.

    This class provides functionality to handle benchmark data including loading,
    processing into a DataFrame, and generating various visual plots for performance
    analysis. It ensures proper organization of benchmark files and provides specific
    plot types to assess performance metrics and acceleration factors across a range
    of input sizes.

    :ivar benchmarks_plot_dir: Directory for storing generated benchmark plots.
    :type benchmarks_plot_dir: pathlib.Path
    :ivar test_number: The integer representation of the test identifier.
    :type test_number: int
    """

    def __init__(self, test_id: str) -> None:
        self.benchmarks_plot_dir = self.benchmarks_setup()
        self.test_number = int(test_id)

    @staticmethod
    def benchmarks_setup() -> Path:
        """
        Creates and initializes a directory for storing benchmark plots.

        This static method generates a uniquely named directory based on the current
        date and time in the "purem_benchmarks" directory relative to the script's
        location. If the directory does not already exist, it will be created. The
        directory name includes a timestamp in the format "YYYY-MM-DD_HH_MM_SS__FFF".

        :return: A `Path` object pointing to the created directory.
        :rtype: Path
        """
        current_time = datetime.datetime.now()
        benchmarks_id: str = f"""purem_benchmarks_{current_time.strftime("%Y-%m-%d_%H_%M_%S__%f")[:-4]}"""
        current_dir: str = os.path.dirname(__file__)
        benchmarks_plot_dir: Path = Path(current_dir) / "purem_benchmarks" / benchmarks_id
        os.makedirs(benchmarks_plot_dir, exist_ok=True)

        return benchmarks_plot_dir

    @staticmethod
    def extract_size_from_filepath(filepath: str) -> int:
        """
        Extract the `size` value from the given file path.
        Assumes the size is encoded as the last part before the file extension.
        """
        split_filepath = filepath.split("_")
        return int(split_filepath[-1].split(".")[0])

    def run(self) -> None:
        """
        Executes the process of generating benchmarks and acceleration plots, as well as creating
        and saving Markdown-formatted data frames. This function orchestrates the pipeline by
        calling several helper methods to handle plotting and data frame construction.

        :raises AnyRelevantError: Description of why and when it might be raised.

        :return: None
        """
        benchmarks_df = self.benchmarks_data_frame()
        self.benchmarks_plot(benchmarks_df)
        self.acceleration_plot(benchmarks_df)

        markdown_df = self.markdown_data_frame()
        self.markdown_save(markdown_df)

    def load_benchmarks_files(self) -> List[str]:
        """
        Loads benchmark files matching a specific test number from a designated directory.
        The function identifies files in the benchmarks directory that match a given test number,
        extracts a unique timestamp from the matching files, and retrieves all files that share
        the same timestamp. This ensures that all related benchmark files for a specific test are
        collected efficiently.

        :param test_number: An integer representing the test number used to filter and locate
            benchmark files. The test number will be zero-padded to ensure a four-digit format.
        :return: A list of file paths for all benchmark files associated with the specified test
            number and same timestamp.
        :rtype: List[str]
        :raises FileNotFoundError: If no benchmark files matching the specified test number
            are found in the benchmarks directory.
        """
        benchmarks_dir = ".benchmarks/Darwin-CPython-3.11-64bit"
        search_pattern = str(self.test_number).zfill(4)

        matched_files = [f for f in os.listdir(benchmarks_dir) if f.startswith(search_pattern) and f.endswith(".json")]

        if not matched_files:
            raise FileNotFoundError(f"No benchmark files found starting with {search_pattern} in {benchmarks_dir}")

        file_stamp = matched_files[0].split("_")[2]
        return [
            os.path.join(benchmarks_dir, f)
            for f in os.listdir(benchmarks_dir)
            if f.split("_")[2] == file_stamp and f.endswith(".json")
        ]

    def benchmarks_data_frame(self) -> pd.DataFrame:
        """
        Generates a Pandas DataFrame containing benchmark data extracted
        from all related files. The method processes multiple files, loads
        their content, extracts benchmark statistics, and organizes them
        into a structured tabular format.

        :param self: The instance of the class containing this method.

        :returns: A pandas DataFrame where each row represents a single
            benchmark entry. The columns include 'size', 'func', 'min',
            'max', 'mean', 'stddev', and 'ops', corresponding to the
            benchmark statistics and other metadata.
        :rtype: pd.DataFrame
        """
        files = self.load_benchmarks_files()

        records = []

        for filepath in files:
            with open(filepath, "r") as f:
                data = json.load(f)

            filepath_split = filepath.split("_")

            for entry in data["benchmarks"]:
                size = self.extract_size_from_filepath(filepath)
                func = entry["params"]["func_name"]
                stats = entry["stats"]
                records.append({
                    "size": size,
                    "func": func,
                    "min": stats["min"],
                    "max": stats["max"],
                    "mean": stats["mean"],
                    "stddev": stats["stddev"],
                    "ops": stats["ops"]
                })

        return pd.DataFrame(records)

    def benchmarks_plot(self, df: DataFrame) -> None:
        """
        Generates and saves benchmarking plots for various performance metrics.

        This function iterates over a defined list of metrics such as "min", "max", "mean",
        "stddev", and "ops", and generates two types of plots for each metric:
        one for the full range of input sizes and another for large input sizes only.
        Each plot visualizes the relationship between input sizes and metric values for all
        functions provided in the DataFrame. The x-axis is set to log scale for both
        plots. The y-axis is also set to log scale for all metrics except "stddev".
        The generated plots are saved as PNG files in a directory defined by
        `self.benchmarks_plot_dir`.

        :param df: A DataFrame containing benchmarking data. The DataFrame must have the
            following columns:
            - "func": Names of the functions being benchmarked.
            - "size": Input sizes corresponding to the benchmarks.
            - Metric columns (e.g., "min", "max", "mean", "stddev", "ops"): Values of the
              respective performance metrics.
        """
        metrics = ["min", "max", "mean", "stddev", "ops"]

        for metric in metrics:
            plt.figure(figsize=(10, 6))
            for func_name in df["func"].unique():
                data_by_func = df[df["func"] == func_name].sort_values("size")
                plt.plot(data_by_func["size"], data_by_func[metric], label=func_name, marker='o')
            plt.title(f"{metric.upper()} vs Size (Full Range)")
            plt.xlabel("Input Size")
            plt.ylabel(metric)
            plt.legend()
            plt.grid(True)
            plt.xscale("log")
            if metric != "stddev":
                plt.yscale("log")
            plt.tight_layout()
            plt.savefig(f"{self.benchmarks_plot_dir}/benchmark_{metric}_full.png")
            plt.close()

            plt.figure(figsize=(10, 6))
            for func_name in df["func"].unique():
                data_by_func = df[(df["func"] == func_name) & (df["size"] > 1e5)].sort_values("size")
                if not data_by_func.empty:
                    plt.plot(data_by_func["size"], data_by_func[metric], label=func_name, marker='o')
            plt.title(f"{metric.upper()} vs Size (Large Inputs Only)")
            plt.xlabel("Input Size (>1e5)")
            plt.ylabel(metric)
            plt.legend()
            plt.grid(True)
            plt.xscale("log")
            if metric != "stddev":
                plt.yscale("log")
            plt.tight_layout()
            plt.savefig(f"{self.benchmarks_plot_dir}/benchmark_{metric}_large.png")
            plt.close()

    def acceleration_plot(self, df: pd.DataFrame) -> None:
        """
        Plots the acceleration of the "Purem" implementation relative to other functions/libraries
        based on their performance for different input sizes. The plot shows acceleration as a
        function of the input size as a logarithmic graph, detailing the relative performance
        gain or loss.

        This function is intended to visualize and compare performance metrics for computational
        functions with respect to a "Purem" baseline. It ensures that shared input sizes
        are used for comparison and applies smoothing (Savitzky-Golay filter) to the acceleration
        curve for better interpretability.

        :param df: A pandas DataFrame containing benchmarking results with the following
            columns:
            - "func": The name of the function or library being benchmarked.
            - "size": The size of the input data for the benchmark.
            - "ops": The respective operations per second or performance metric for each
              function.

        :return: None
        """
        func_names = df["func"].unique()

        assert any("Purem" in name for name in func_names), "Purem is missed!"

        df_purem = df[df["func"].str.contains("Purem")].sort_values("size")

        plt.figure(figsize=(14, 8))

        for func_name in func_names:
            if "Purem" in func_name:
                continue

            df_other = df[df["func"] == func_name].sort_values("size")
            common_sizes = np.intersect1d(df_purem["size"], df_other["size"])

            if len(common_sizes) == 0:
                continue

            purem_ops = df_purem[df_purem["size"].isin(common_sizes)]["ops"].values
            other_ops = df_other[df_other["size"].isin(common_sizes)]["ops"].values

            acceleration = purem_ops / other_ops

            try:
                if len(acceleration) >= 5:
                    acceleration_smooth = savgol_filter(acceleration, window_length=5, polyorder=2)
                else:
                    acceleration_smooth = acceleration
            except Exception as e:
                warnings.warn(f"Smoothing skipped: {e}")
                acceleration_smooth = acceleration

            plt.plot(common_sizes, acceleration_smooth, marker='o', label=f"Purem vs {func_name.partition(':')[2].strip()}")

        plt.xscale('log')
        plt.yscale('log')
        plt.axhline(y=1, color='black', linestyle='--', linewidth=1, label='Equal Performance (x1)')
        plt.xlabel('Input Size', fontsize=14)
        plt.ylabel('Acceleration (Purem / Other)', fontsize=14)
        plt.title('Purem Acceleration Relative to Other Libraries (Production Level)', fontsize=16)
        plt.grid(True, which="both", linestyle='--', linewidth=0.6)
        plt.legend(fontsize=12)
        plt.tight_layout()
        plt.savefig(f"{self.benchmarks_plot_dir}/benchmark_acceleration_large.png")
        plt.close()

    def markdown_data_frame(self) -> pd.DataFrame:
        """
        Converts benchmark data from files into a structured pandas DataFrame.

        This method processes a collection of JSON files that contain benchmark
        results. It reads the files, extracts relevant statistical information
        about the benchmarking process, and formats this data into a structured
        pandas DataFrame. Each record in the resultant DataFrame represents a
        specific benchmark test and includes its relevant metadata and statistical
        metrics.

        :return:
            A pandas DataFrame where each row corresponds to a benchmark with
            columns that include details such as 'Elements', 'Function', 'OPS',
            'Min Time (s)', 'Max Time (s)', 'Mean Time (s)', and 'Std Dev'.
        :rtype:
            pd.DataFrame
        """
        files = self.load_benchmarks_files()

        records_markdown = []

        for filepath in files:
            with open(filepath, "r") as f:
                data = json.load(f)

            filepath_split = filepath.split("_")

            for b in data['benchmarks']:
                size = self.extract_size_from_filepath(filepath)
                func_name = b['params']['func_name']
                stats = b['stats']
                records_markdown.append({
                    "Elements": size,
                    "Function": func_name,
                    "OPS": stats["ops"],
                    "Min Time (s)": stats["min"],
                    "Max Time (s)": stats["max"],
                    "Mean Time (s)": stats["mean"],
                    "Std Dev": stats["stddev"]
                })

        return pd.DataFrame(records_markdown)

    def markdown_save(self, df: pd.DataFrame) -> None:
        """
        Saves a markdown file summarizing benchmark results in a human-readable format. The function processes
        a pandas DataFrame containing benchmark data, formats specific columns for scientific notation or fixed-point
        notation, and groups the results by the "Elements" column. Each group is written as a separate section
        in the markdown file, where the file is saved to the benchmarks directory.

        :param df: A pandas DataFrame containing benchmark results. Expected columns include "Elements",
            "OPS", "Min Time (s)", "Max Time (s)", "Mean Time (s)", and "Std Dev".
        :return: None
        """
        df = df.sort_values(by=["Elements", "OPS"], ascending=[True, False])

        df["Min Time (s)"] = df["Min Time (s)"].map(lambda x: f"{x:.2e}")
        df["Max Time (s)"] = df["Max Time (s)"].map(lambda x: f"{x:.2e}")
        df["Mean Time (s)"] = df["Mean Time (s)"].map(lambda x: f"{x:.2e}")
        df["Std Dev"] = df["Std Dev"].map(lambda x: f"{x:.2e}")
        df["OPS"] = df["OPS"].map(lambda x: f"{x:.2f}")

        grouped = df.groupby('Elements')

        with open(f"{self.benchmarks_plot_dir}/benchmarks_table.md", "w") as f:
            f.write("# Benchmark Results\n\n")
            for size, group in grouped:
                f.write(f"### Elements: {size}\n\n")
                f.write(group.drop(columns=["Elements"]).to_markdown(index=False))
                f.write("\n\n")