"""Top-level package for genai_evaluation."""

__author__ = """Dr. Vincent Granville, Rajiv Iyer"""
__email__ = 'vincentg@mltechniques.com, raju.rgi@gmail.com'
__version__ = '0.1.5'
from .genai_evaluation import *

genai_evaluation/__init__.py
rajiviyer/genai_evaluation
"""Contains functions for evaluating Generative AI Models"""


import pandas as pd
import numpy as np
from typing import List, Tuple
import re


def multivariate_ecdf(data_a: pd.DataFrame,
                      data_b: pd.DataFrame,
                      n_nodes: int = 1000,
                      verbose: bool = True,
                      random_seed: int = None) -> Tuple:
    """
    Args:
        data_a (pd.DataFrame): Pandas DataFrame
        data_b (pd.DataFrame): Pandas DataFrame
        n_nodes (int, optional):Specifies the number of nodes 
                                i.e. the query strings to be generated. 
                                This is a hyperparameter and can be tuned.
                                Defaults to 1000.
        verbose (bool): Flag to display progress of the operations. 
                        Defaults to True 
        random_seed (int, optional): random seed to be set before
                                    operations. If set random seed is set using `np.random.seed(random_seed)`. Defaults to None

    Raises:
        TypeError: Throws error if Input Datasets are not Pandas DataFrames
        ValueError: Throws error if any Input Dataset is empty
        ValueError: Thows value error if one or more column names between both the  
                    Input DataFrames do not match


    Returns:
        List: Returns Tuple of query_string & computed ECDFs of both Input Datasets
    """

    if not isinstance(data_a, pd.DataFrame) or not isinstance(data_b, pd.DataFrame):
        raise TypeError("Input Datasets should be Pandas DataFrames!!")
    
    if data_a.empty or data_b.empty:
        raise ValueError("Input Datasets should not be empty!!")
    
    data_a = data_a.copy()
    data_b = data_b.copy()
    
    if re.search(r'[^a-zA-Z0-9_]', "".join(data_a.columns)):
        if verbose:
            print("Dataset A columns have special characters. Will be cleaned before processing!!")
        data_a_cols_cleaned = [re.sub(r'[^a-zA-Z0-9_]', '', col).lower() 
                               for col in data_a.columns]
        data_a.columns = data_a_cols_cleaned
        
    if re.search(r'[^a-zA-Z0-9_]', "".join(data_a.columns)):
        if verbose:
            print("Dataset B columns have special characters. Will be cleaned before processing!!")
        data_b_cols_cleaned = [re.sub(r'[^a-zA-Z0-9_]', '', col).lower() 
                               for col in data_b.columns]
        data_b.columns = data_b_cols_cleaned
    
    if not data_a.columns.equals(data_b.columns):
        raise ValueError("One or more column names do not match in both DataFrames!!")    

    eps = 0.0000000001
    query_val = []
    features = data_a.columns
    n_features = len(features)
    if random_seed:
        np.random.seed(random_seed)
    
    for point in range(n_nodes):
        if point % 100 == 0 and verbose:
            print(f"Sampling ecdf, location = {point}")
        
        # Get random percentiles
        percentiles = np.random.uniform(0, 1, n_features)
        percentiles = percentiles**(1/n_features)

        # Get the percentile values from the dataset a for each column
        perc_vals = [eps + np.quantile(data_a.iloc[:, k], perc)
                     for k, perc in enumerate(percentiles)]

        # Create the query string combined for each column
        query_str = " and ".join([f"{features[k]} <= {perc_val}"
                                  for k, perc_val in enumerate(perc_vals)])

        # From dataset a, get the counts of rows which
        # satisfy the conditions in the query string
        filter_count_a = len(data_a.query(query_str))

        # For counts > 0, create key: str of the list of perc_vals
        # Append key, query_str & the normalized filter count for dataset
        if filter_count_a > 0:
            key = ', '.join(map(str, perc_vals))
            query_val.append((key, query_str, filter_count_a/len(data_a)))

    # Sort the list based on the items (third element of each tuple)
    query_val.sort(key=lambda item: item[2])

    query_lst = []
    ecdf_a = []
    ecdf_b = []

    # for each entry in the query_val list
    # Retrieve the query_str and run on both datasets to get the filter counts
    # Normalize the filter count for dataset b

    for e_val in query_val:
        query_str = e_val[1]
        value_data_a = e_val[2]
        filter_count_b = len(data_b.query(query_str))
        value_data_b = filter_count_b / len(data_b)
        query_lst.append(query_str)
        ecdf_a.append(value_data_a)
        ecdf_b.append(value_data_b)

    return query_lst, ecdf_a, ecdf_b


def ks_statistic(ecdf_a: List, ecdf_b: List) -> float:
    """
    Args:
        ecdf_a (List): ECDF Generated through the Multivariate ECDF function
        ecdf_a (List): ECDF Generated through the Multivariate ECDF function

    Raises:
        ValueEror: Throws error if the input ECDFs' are empty or their lengths are not equal

    Returns:
        float: Returns KS Statistic
    """
    if len(ecdf_a) != len(ecdf_b):
        raise ValueError("Both Input ECDFs should be of the same length!!")

    if len(ecdf_a) == 0 or len(ecdf_b) == 0:
        raise ValueError("ECDFs should not be empty!!")

    np_ecdf_a = np.array(ecdf_a)
    np_ecdf_b = np.array(ecdf_b)
    
    # Compute the absolute difference between ECDFs
    abs_diff = np.abs(np_ecdf_a - np_ecdf_b)

    # Find the maximum absolute difference (KS statistic)
    ks_statistic = np.max(abs_diff)

    return ks_statistic

genai_evaluation/genai_evaluation.py
rajiviyer/genai_evaluation
#!/usr/bin/env python

"""The setup script."""

from setuptools import setup, find_packages

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

with open('HISTORY.md') as history_file:
    history = history_file.read()

requirements = []

test_requirements = ['pytest>=3', ]

setup(
    author="Rajiv Iyer",
    author_email='raju.rgi@gmail.com',
    python_requires='>=3.6',
    classifiers=[
        'Development Status :: 2 - Pre-Alpha',
        'Intended Audience :: Developers',
        'License :: OSI Approved :: MIT License',
        'Natural Language :: English',
        'Programming Language :: Python :: 3',
        'Programming Language :: Python :: 3.6',
        'Programming Language :: Python :: 3.7',
        'Programming Language :: Python :: 3.8',
    ],
    description="Evaluation of Generative AI Models",
    install_requires=requirements,
    license="MIT license",
    long_description=readme + '\n\n' + history,
    long_description_content_type="text/markdown",
    include_package_data=True,
    keywords='genai_evaluation',
    name='genai_evaluation',
    packages=find_packages(include=['genai_evaluation', 'genai_evaluation.*']),
    test_suite='tests',
    tests_require=test_requirements,
    url='https://github.com/rajiviyer/genai_evaluation',
    version='0.1.5',
    zip_safe=False,
)

setup.py
rajiviyer/genai_evaluation
"""Unit test package for genai_evaluation."""

tests/__init__.py
rajiviyer/genai_evaluation
"""Tests for `genai_evaluation` package."""

import numpy as np
import pytest
from genai_evaluation import ks_statistic, multivariate_ecdf
import pandas as pd

def test_multivariate_ecdf():
    
    # Dataset is a list
    with pytest.raises(TypeError):    
        real_data = [1,2,3,4]
        
        # Simulate a synthetic dataset
        synthetic_data = pd.DataFrame(np.random.rand(100, 5), columns=['x1', 'x2', 'x3', 'x4', 'x5'])

        # Calculate ECDFs 
        _, ecdf_val1, ecdf_synth = \
                    multivariate_ecdf(real_data, 
                                    synthetic_data, 
                                    n_nodes = 1000,
                                    verbose = False,
                                    random_seed=42)
    # Empty Dataset
    with pytest.raises(ValueError):    
        real_data = pd.DataFrame()
        
        # Simulate a synthetic dataset
        synthetic_data = pd.DataFrame(np.random.rand(100, 5), columns=['x1', 'x2', 'x3', 'x4', 'x5'])

        # Calculate ECDFs 
        _, ecdf_val1, ecdf_synth = \
                    multivariate_ecdf(real_data, 
                                    synthetic_data, 
                                    n_nodes = 1000,
                                    verbose = False,
                                    random_seed=42)
    # Column Names do not Match
    with pytest.raises(ValueError):     
        real_data = pd.DataFrame(np.random.rand(100, 5), columns=['x1', 'x_2', 'x3', 'x4', 'x5'])
        
        # Simulate a synthetic dataset
        synthetic_data = pd.DataFrame(np.random.rand(100, 5), columns=['x1', 'x2', 'x_3', 'x4', 'x5'])

        # Calculate ECDFs 
        _, ecdf_val1, ecdf_synth = \
                    multivariate_ecdf(real_data, 
                                    synthetic_data, 
                                    n_nodes = 1000,
                                    verbose = False,
                                    random_seed=42)                              

# Define test cases
def test_ks_statistic():
    # Basic test cases
    a = np.array([1.0, 2.0, 3.0])
    b = np.array([1.0, 2.0, 3.0])
    assert ks_statistic(a, b) == 0.0

    a = np.array([1.0, 2.0, 3.0])
    b = np.array([4.0, 5.0, 6.0])
    assert ks_statistic(a, b) == 3.0

    a = np.array([1.0, 2.0, 3.0])
    b = np.array([1.0, 2.0, 2.5])
    assert ks_statistic(a, b) == 0.5  # Approximate value

    assert ks_statistic(np.array([1.0]), np.array([1.0])) == 0.0
    # Single-element arrays

    # Large arrays
    a = np.arange(1, 1001)
    b = np.arange(1001, 2001)
    assert ks_statistic(a, b) == 1000

    # Mismatched length arrays (Expecting a ValueError)
    with pytest.raises(ValueError):
        a = np.array([1.0, 2.0, 3.0])
        b = np.array([1.0, 2.0])
        ks_statistic(a, b)

    # Edge cases. Empty Arrays
    with pytest.raises(ValueError):
        a = np.array([])
        b = np.array([])
        ks_statistic(a, b)

tests/test_genai_evaluation.py
rajiviyer/genai_evaluation