WassimTenachi / PhySO / #16

Committed 28 Jul 2025 07:07AM UTC coverage: 70.145% (-10.8%) from 80.984%

Build # #16

Build Type

push

coveralls-python

Committed by

WassimTenachi

Commit Message

fix

Run Details

5963 of 8501 relevant lines covered (70.14%)

0.7 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

76.51

/physo/benchmark/ClassDataset/ClassProblem.py

import pandas as pd
import numpy as np
import pathlib
import sympy
import matplotlib.pyplot as plt

# Internal imports
from physo.benchmark.utils import symbolic_utils as su

# Dataset paths
PARENT_FOLDER = pathlib.Path(__file__).parents[0]
PATH_CLASS_EQS_CSV = PARENT_FOLDER / "ClassEquations.csv"


# ---------------------------------------------------------------------------------------------------------------------
# --------------------------------------------------- LOADING CSVs  ---------------------------------------------------
# ---------------------------------------------------------------------------------------------------------------------

def load_class_equations_csv (filepath_eqs ="ClassEquations.csv"):
    """
    Loads ClassEquations.csv into a pd.DataFrame.
    Parameters
    ----------
    filepath_eqs : str
        Path to ClassEquations.csv.
    Returns
    -------
    eqs_class_df : pd.DataFrame
    """

    eqs_class_df = pd.read_csv(filepath_eqs, sep=";")
    # Max nb of variables columns
    max_n_vars = int(eqs_class_df.columns.to_numpy()[-1].split('_')[0][1:])
    # Number of equations
    n_eqs = len(eqs_class_df)

    # Set types for int columns
    eqs_class_df = eqs_class_df.astype({'Number': int, '# variables': int, '# spe': int})
    # Set types for str columns
    eqs_class_df = eqs_class_df.astype({'v%i_type'%(i):str for i in range(1,max_n_vars+1)})

    # ---- Verifying number of variables for safety ----
    # Checking the number of variables declared in the file for each problem
    # Expected number of variables for each problem
    expected_n_vars = (~eqs_class_df[["v%i_name" % (i) for i in range(1, max_n_vars+1)]].isnull().to_numpy()).sum(axis=1) # (n_eqs,)
    # Declared number of variables for each problem
    n_vars = eqs_class_df["# variables"].to_numpy() + eqs_class_df["# spe"].to_numpy()                                    # (n_eqs,)
    # Is nb of declared variable consistent with variables columns ?
    is_consistent = np.equal(expected_n_vars, n_vars)                                                                     # (n_eqs,)
    assert is_consistent.all(), "Nb. of filled variables columns not consistent with declared nb. of variables for " \
                                "problems:\n %s"%(str(eqs_class_df.loc[~is_consistent]))

    return eqs_class_df


EQS_CLASS_DF = load_class_equations_csv (filepath_eqs = PATH_CLASS_EQS_CSV,)

# Number of equations in dataset
N_EQS = EQS_CLASS_DF.shape[0]

# Size of units vector
CLASS_UNITS_VECTOR_SIZE = 7

# ---------------------------------------------------------------------------------------------------------------------
# --------------------------------------------------- UNITS UTILS  ----------------------------------------------------
# ---------------------------------------------------------------------------------------------------------------------

# Gets units from variable name
def get_units (i_eq, i_var = 0, output_var = False):
    """
    Gets units of variable.
    Parameters
    ----------
    i_eq : int
        Equation number in the set of equations.
    i_var : int
        Variable id in its equation line.
    output_var : bool
        If True, returns units of output variable, otherwise returns units of input variable specified by i_var.
    Returns
    -------
    units : numpy.array of shape (CLASS_UNITS_VECTOR_SIZE,) of floats
        Units of variable.
    """

    units = np.zeros(CLASS_UNITS_VECTOR_SIZE)                                               # (CLASS_UNITS_VECTOR_SIZE,)

    if output_var:
        res = EQS_CLASS_DF.iloc[i_eq]["Output_units"]
    else:
        res = EQS_CLASS_DF.iloc[i_eq]["v%i_units"%(i_var)]

    res = res[1:-1]  # Removing brackets
    res = np.fromstring(res, dtype=float, sep=',')
    units[:len(res)] = res  # Replacing units by elements in res

    return units



# ---------------------------------------------------------------------------------------------------------------------
# -------------------------------------------------- CLASS PROBLEM  ---------------------------------------------------
# ---------------------------------------------------------------------------------------------------------------------
CONST_LOCAL_DICT = {"pi" : np.pi}

class ClassProblem:
    """
    Represents a single Class SR benchmark problem.
    (See https://arxiv.org/abs/2312.01816 for details).

    Attributes
    ----------
    i_eq : int
        Equation number in the set of equations.
    eq_name : str
        Equation name in the set of equations (e.g. 'Harmonic Oscillator').
    n_vars : int
        Number of input variables.
    n_spe : int
        Number of realization specific free constants.
    eq_df : pandas.core.series.Series
        Underlying pandas dataframe line of this equation.
    original_var_names : bool
        Using original variable names (e.g. theta, sigma etc.) and original output variable name (e.g. f, E etc.) if
        True, using x0, x1 ... as input variable names k0, k1,... as spe free consts names and y as output variable name
        otherwise.

    y_name_original : str
        Name of output variable as in the Class dataset.
    y_name : str
        Name of output variable.
    y_units : array_like of shape (CLASS_UNITS_VECTOR_SIZE,) of floats
        Units of output variables.

    X_names_original : array_like of shape (n_vars,) of str
        Names of input variables as in the Class dataset.
    X_names : array_like of shape (n_vars,) of str
        Names of input variables.
    X_lows : array_like of shape (n_vars,) of floats
        Lowest values taken by input variables.
    X_highs : array_like of shape (n_vars,) of floats
        Highest values taken by input variables.
    X_units :  array_like of shape (n_vars, CLASS_UNITS_VECTOR_SIZE,) of floats
        Units of input variables.

    K_names_original : array_like of shape (n_spe,) of str
        Names of spe free consts as in the Class dataset.
    K_names : array_like of shape (n_spe,) of str
        Names of spe free consts.
    K_lows : array_like of shape (n_spe,) of floats
        Lowest values taken by spe free consts.
    K_highs : array_like of shape (n_spe,) of floats
        Highest values taken by spe free consts.
    K_units :  array_like of shape (n_spe, CLASS_UNITS_VECTOR_SIZE,) of floats
        Units of spe free consts.

    formula_original : str
        Formula as in the Class dataset.

    X_sympy_symbols : array_like of shape (n_vars,) of sympy.Symbol
        Sympy symbols representing each input variables with assumptions (negative, positive etc.).
    sympy_X_symbols_dict : dict of {str : sympy.Symbol}
        Input variables names to sympy symbols (w assumptions), can be passed to sympy.parsing.sympy_parser.parse_expr
        as local_dict.
    K_sympy_symbols : array_like of shape (n_spe,) of sympy.Symbol
        Sympy symbols representing each spe free consts with assumptions (negative, positive etc.).
    sympy_K_symbols_dict : dict of {str : sympy.Symbol}
        Spe free consts names to sympy symbols (w assumptions), can be passed to sympy.parsing.sympy_parser.parse_expr

    local_dict : dict of {str : sympy.Symbol or float}
        Input variables names and spe free consts to sympy symbols (w assumptions) and constants (eg. pi : np.pi etc.),
        can be passed to sympy.parsing.sympy_parser.parse_expr as local_dict.
    formula_sympy : sympy expression
        Formula in sympy.
    formula_sympy_eval : sympy expression
        Formula in sympy with evaluated fixed constants (eg. pi -> 3.14... etc).
    formula_latex : str
        Formula in latex.

    """

    def __init__(self, i_eq = None, eq_name = None, original_var_names = False):
        """
        Loads a Class problem based on its number in the set or its equation name.
        Parameters
        ----------
        i_eq : int
            Equation number in the set of equations.
        eq_name : str
            Equation name in the set of equations (e.g. 'Harmonic Oscillator').
        original_var_names : bool
            Using original variable names (e.g. theta, sigma etc.) and original output variable name (e.g. f, E etc.) if
            True, using x0, x1 ... as input variable names and y as output variable name otherwise.
        """
        # Selecting equation line in dataframe
        if i_eq is not None:
            self.eq_df  = EQS_CLASS_DF.iloc[i_eq]                                       # pandas.core.series.Series
        elif eq_name is not None:
            self.eq_df = EQS_CLASS_DF[EQS_CLASS_DF ["Name"] == eq_name ].iloc[0]        # pandas.core.series.Series
        else:
            raise ValueError("At least one of equation number (i_eq) or equation name (eq_name) should be specified to select a Class problem.")


        # Equation number
        self.i_eq = i_eq                                                     # int
        # Code name of equation (eg. 'Harmonic Oscillator')
        self.eq_name = self.eq_df["Name"]                                    # str
        # SRBench style name
        self.SRBench_name = "class_%i"%(self.i_eq)                           # str
        # Number of input variables
        self.n_vars = int(self.eq_df["# variables"])                         # int
        # Number of realization specific free constants
        self.n_spe  = int(self.eq_df["# spe"])                               # int

        # --------- Handling input variables vs spe free consts ---------
        # Total number of variables
        n_v     = self.n_vars + self.n_spe                                                      # int
        v_ids   = np.array( [i_var for i_var in range(1, n_v +1)] )                             # (n_v,)
        v_types = self.eq_df[np.array( [ "v%i_type"%(i_var) for i_var in v_ids ] )].to_numpy()  # (n_v,)
        is_var  = v_types == "var"                                                              # (n_v,)
        is_spe  = v_types == "spe"                                                              # (n_v,)


        # Using x0, x1 ... and y names or original names (e.g. theta, sigma, f etc.)
        self.original_var_names = original_var_names                         # bool

        # ----------- y : output variable -----------
        # Name of output variable
        self.y_name_original = self.eq_df["Output_name"]                         # str
        # Name of output variable : y or original name (eg. f, E etc.)
        self.y_name = self.y_name_original if self.original_var_names else 'y'   # str
        # Units of output variables
        self.y_units = get_units(i_eq=self.i_eq, output_var=True)                # (CLASS_UNITS_VECTOR_SIZE,)

         # ----------- X : input variables -----------
        var_ids     = v_ids[is_var]                                                                                        # (n_vars,)
        var_ids_str = np.array( [ "v%i"%(i_var) for i_var in var_ids ]   ).astype(str)                                     # (n_vars,)
        # Names of input variables
        self.X_names_original = np.array( [ self.eq_df[ id + "_name" ] for id in var_ids_str  ]   ).astype(str)            # (n_vars,)
        X_names_xi_style      = np.array( [ "x%i"%(i_var) for i_var in range(self.n_vars)     ]   ).astype(str)            # (n_vars,)
        self.X_names          = self.X_names_original if self.original_var_names else X_names_xi_style                     # (n_vars,)
        # Lowest values taken by input variables
        self.X_lows           = np.array( [ self.eq_df[ id + "_low"  ] for id in var_ids_str ]    ).astype(float)          # (n_vars,)
        # Highest values taken by input variables
        self.X_highs          = np.array( [ self.eq_df[ id + "_high" ] for id in var_ids_str  ]   ).astype(float)          # (n_vars,)
        # Units of input variables
        self.X_units          = np.array( [ get_units(i_eq=self.i_eq, i_var=i_var) for i_var in var_ids ] ).astype(float)  # (n_vars, CLASS_UNITS_VECTOR_SIZE,)

        # Input variables as sympy symbols
        self.X_sympy_symbols = []
        for i in range(self.n_vars):
            self.X_sympy_symbols.append (su.sympy_symbol_with_assumptions_from_range(name = self.X_names[i],
                                                                                     low  = self.X_lows [i],
                                                                                     high = self.X_highs[i],
                                                                                    ))
        # Input variables names to sympy symbols dict
        self.sympy_X_symbols_dict = {self.X_names[i] : self.X_sympy_symbols[i] for i in range(self.n_vars)}                      #  (n_vars,)
        # Dict to use to read original class dataset formula
        # Original names to symbols in usage (i.e. symbols having original names or not)
        # eg. 'theta' -> theta symbol etc. (if original_var_names=True) or 'theta' -> x0 symbol etc. (else)
        self.sympy_original_to_X_symbols_dict = {self.X_names_original[i] : self.X_sympy_symbols[i] for i in range(self.n_vars)} #  (n_vars,)
        # NB: if original_var_names=True, then self.sympy_X_symbols_dict = self.sympy_original_to_X_symbols_dict

         # ----------- K : spe free consts -----------

        spe_ids     = v_ids[is_spe]                                                                                        # (n_spe,)
        spe_ids_str = np.array( [ "v%i"%(i_spe) for i_spe in spe_ids ]   ).astype(str)                                     # (n_spe,)
        # Names of spe free consts
        self.K_names_original = np.array( [ self.eq_df[ id + "_name" ] for id in spe_ids_str  ]   ).astype(str)            # (n_spe,)
        K_names_ki_style      = np.array( [ "k%i"%(i_spe) for i_spe in range(self.n_spe)     ]   ).astype(str)             # (n_spe,)
        self.K_names          = self.K_names_original if self.original_var_names else K_names_ki_style                     # (n_spe,)
        # Lowest values taken by spe free consts
        self.K_lows           = np.array( [ self.eq_df[ id + "_low"  ] for id in spe_ids_str ]    ).astype(float)          # (n_spe,)
        # Highest values taken by spe free consts
        self.K_highs          = np.array( [ self.eq_df[ id + "_high" ] for id in spe_ids_str  ]   ).astype(float)          # (n_spe,)
        # Units of spe free consts
        self.K_units          = np.array( [ get_units(i_eq=self.i_eq, i_var=i_var) for i_var in spe_ids ] ).astype(float)  # (n_spe, CLASS_UNITS_VECTOR_SIZE,)

        # Spe free consts as sympy symbols
        self.K_sympy_symbols = []
        for i in range(self.n_spe):
            self.K_sympy_symbols.append (su.sympy_symbol_with_assumptions_from_range(name = self.K_names[i],
                                                                                     low  = self.K_lows [i],
                                                                                     high = self.K_highs[i],
                                                                                    ))
        # Spe free consts names to sympy symbols dict
        self.sympy_K_symbols_dict = {self.K_names[i] : self.K_sympy_symbols[i] for i in range(self.n_spe)}                      #  (n_spe,)
        # Dict to use to read original class dataset formula
        # Original names to symbols in usage (i.e. symbols having original names or not)
        # eg. 'theta' -> theta symbol etc. (if original_var_names=True) or 'theta' -> k0 symbol etc. (else)
        self.sympy_original_to_K_symbols_dict = {self.K_names_original[i] : self.K_sympy_symbols[i] for i in range(self.n_spe)} #  (n_spe,)
        # NB: if original_var_names=True, then self.sympy_K_symbols_dict = self.sympy_original_to_K_symbols_dict

        # ----------- Formula -----------
        self.formula_original = self.eq_df["Formula"]  # (str)

        self.v_local_dict = {}
        self.v_local_dict.update(self.sympy_original_to_X_symbols_dict)
        self.v_local_dict.update(self.sympy_original_to_K_symbols_dict)


        # Declaring input variables via local_dict to avoid confusion
        # Eg. So sympy knows that we are referring to gamma as a variable and not the function etc.
        # evaluate = False avoids eg. sin(theta) = 0 when theta domain = [0,5] ie. nonzero=False, but no need for this
        # if nonzero assumption is not used
        evaluate = False
        self.formula_sympy   = sympy.parsing.sympy_parser.parse_expr(self.formula_original,
                                                                     local_dict = self.v_local_dict,
                                                                     evaluate   = evaluate)

        # Local dict : dict of input variables (sympy_original_to_X_symbols_dict) and fixed constants (pi -> 3.14.. etc)
        self.local_dict = {}
        self.local_dict.update(self.v_local_dict)
        self.local_dict.update(CONST_LOCAL_DICT)
        self.formula_sympy_eval = sympy.parsing.sympy_parser.parse_expr(self.formula_original,
                                                                     local_dict = self.local_dict,
                                                                     evaluate   = evaluate)
        # Latex formula
        self.formula_latex   = sympy.printing.latex(self.formula_sympy)
        return None


    def target_function(self, X, K):
        """
        Evaluates X with target function, using K values.
        Parameters
        ----------
        X : numpy.array of shape (n_vars, ?,) of floats
            Input variables.
        K : numpy.array of shape (n_spe,) of floats
            Spe free consts.
        Returns
        -------
        y : numpy.array of shape (?,) of floats
        """
        # Getting sympy function
        sympy_symbols = self.X_sympy_symbols + self.K_sympy_symbols
        f = sympy.lambdify(sympy_symbols, self.formula_sympy, "numpy")
        mapping_vals = {}
        # Mapping between variables names and their data value
        mapping_vals.update({self.X_names[i]: X[i] for i in range(self.n_vars)})
        # Mapping between spe free consts names and their data value
        mapping_vals.update({self.K_names[i]: K[i] for i in range(self.n_spe)})
        # Evaluation
        # Forcing float type so if some symbols are not evaluated as floats (eg. if some variables are not declared
        # properly in source file) resulting partly symbolic expressions will not be able to be converted to floats
        # and an error can be raised).
        # This is also useful for detecting issues such as sin(theta) = 0 because theta.is_nonzero = False -> the result
        # is just an int of float
        y = f(**mapping_vals).astype(float)
        return y

    def generate_data_points (self, n_samples = 1_000, n_realizations = 10, return_K = False):
        """
        Generates data points accordingly for this Class problem.
        Parameters
        ----------
        n_samples : int
            Number of samples to draw. By default, 1e3.
        n_realizations : int
            Number of realizations to draw. By default, 10.
        return_K : bool
            If True, returns K values used to generate data as well.
        Returns
        -------
        multi_X : numpy.array of shape (n_realizations, n_vars, n_samples,) of floats,
        multi_y : numpy.array of shape (n_realizations, n_samples,) of floats
        (multi_K : numpy.array of shape (n_realizations, n_spe,) of floats)
        """
        multi_X = []
        multi_y = []
        multi_K = []

        for i_real in range(n_realizations):
            # Random K sample
            K = np.stack([np.random.uniform(self.K_lows[i_var], self.K_highs[i_var], ) for i_var in range(self.n_spe)])           # (n_spe,)
            # Random X sample
            X = np.stack([np.random.uniform(self.X_lows[i_var], self.X_highs[i_var], n_samples) for i_var in range(self.n_vars)]) # (n_vars, n_samples)
            # Evaluating formula
            y = self.target_function(X=X, K=K)                                                                                    # (n_samples,)

            multi_X.append(X)
            multi_y.append(y)
            multi_K.append(K)

        multi_X = np.stack(multi_X)                                                                                     # (n_realizations, n_vars, n_samples)
        multi_y = np.stack(multi_y)                                                                                     # (n_realizations, n_samples)
        multi_K = np.stack(multi_K)                                                                                     # (n_realizations, n_spe)

        if return_K:
            return multi_X, multi_y, multi_K
        else:
            return multi_X, multi_y

    def get_sympy(self, K_vals=None):
        """
        Gets sympy expression of the formula evaluated with spe free consts.
        Parameters
        ----------
        K_vals : numpy.array of shape (?, n_spe,) of floats or None
            Values to evaluate spe free consts with, if None, uses random values and returns only one realization.
        Returns
        -------
        sympy_expr : np.array of shape (?, n_spe,) of Sympy Expression
        """
        if K_vals is None:
            K  = np.stack([np.random.uniform(self.K_lows[i_var], self.K_highs[i_var], ) for i_var in range(self.n_spe)])        # (n_spe,)
            K_vals = np.stack([K,])                                                                                             # (?, n_spe) = (1, n_spe)

        sympy_expr = []
        for K in K_vals:
            K_dict = {self.K_sympy_symbols[i]: K[i] for i in range(self.n_spe)}
            expr   = self.formula_sympy.subs(K_dict)
            sympy_expr.append(expr)
        sympy_expr = np.array(sympy_expr)                                                                                        # (?,)

        return sympy_expr

    def show_sample(self, n_samples = 10_000, n_realizations = 10, do_show = True, save_path = None):
        multi_X, multi_y = self.generate_data_points(n_samples=n_samples, n_realizations=n_realizations)

        n_dim = multi_X.shape[1]
        fig, ax = plt.subplots(n_dim, 1, figsize=(15, n_dim * 6))
        fig.suptitle(self.formula_original)
        for i in range(n_dim):
            curr_ax = ax if n_dim == 1 else ax[i]
            curr_ax.set_xlabel("%s : %s" % (self.X_names[i], self.X_units[i]))
            curr_ax.set_ylabel("%s : %s" % (self.y_name    , self.y_units))
            for i_real in range(n_realizations):
                curr_ax.plot(multi_X[i_real,i], multi_y[i_real], '.', markersize=1.)
        fig.tight_layout()
        if save_path is not None:
            fig.savefig(save_path, dpi=200)
        if do_show:
            plt.show()

    def get_prefix_expression (self):
        """
        Gets the prefix expression of the formula.
        Returns
        -------
        dict :
            tokens_str : numpy.array of str
                List of tokens in the expression.
            arities : numpy.array of int
                List of arities of the tokens.
            tokens : numpy.array of sympy.core
                List of sympy tokens.
        """
        return su.sympy_to_prefix(self.formula_sympy)

    def __str__(self):
        return "ClassProblem : %s : %s\n%s"%(self.i_eq, self.eq_name, str(self.formula_sympy))

    def __repr__(self):
        return str(self)

1	import pandas as pd	1✔
2	import numpy as np	1✔
3	import pathlib	1✔
4	import sympy	1✔
5	import matplotlib.pyplot as plt	1✔
6
7	# Internal imports
8	from physo.benchmark.utils import symbolic_utils as su	1✔
9
10	# Dataset paths
11	PARENT_FOLDER = pathlib.Path(__file__).parents[0]	1✔
12	PATH_CLASS_EQS_CSV = PARENT_FOLDER / "ClassEquations.csv"	1✔
13
14
15	# ---------------------------------------------------------------------------------------------------------------------
16	# --------------------------------------------------- LOADING CSVs ---------------------------------------------------
17	# ---------------------------------------------------------------------------------------------------------------------
18
19	def load_class_equations_csv (filepath_eqs ="ClassEquations.csv"):	1✔
20	"""
21	Loads ClassEquations.csv into a pd.DataFrame.
22	Parameters
23	----------
24	filepath_eqs : str
25	Path to ClassEquations.csv.
26	Returns
27	-------
28	eqs_class_df : pd.DataFrame
29	"""
30
31	eqs_class_df = pd.read_csv(filepath_eqs, sep=";")	1✔
32	# Max nb of variables columns
33	max_n_vars = int(eqs_class_df.columns.to_numpy()[-1].split('_')[0][1:])	1✔
34	# Number of equations
35	n_eqs = len(eqs_class_df)	1✔
36
37	# Set types for int columns
38	eqs_class_df = eqs_class_df.astype({'Number': int, '# variables': int, '# spe': int})	1✔
39	# Set types for str columns
40	eqs_class_df = eqs_class_df.astype({'v%i_type'%(i):str for i in range(1,max_n_vars+1)})	1✔
41
42	# ---- Verifying number of variables for safety ----
43	# Checking the number of variables declared in the file for each problem
44	# Expected number of variables for each problem
45	expected_n_vars = (~eqs_class_df[["v%i_name" % (i) for i in range(1, max_n_vars+1)]].isnull().to_numpy()).sum(axis=1) # (n_eqs,)	1✔
46	# Declared number of variables for each problem
47	n_vars = eqs_class_df["# variables"].to_numpy() + eqs_class_df["# spe"].to_numpy() # (n_eqs,)	1✔
48	# Is nb of declared variable consistent with variables columns ?
49	is_consistent = np.equal(expected_n_vars, n_vars) # (n_eqs,)	1✔
50	assert is_consistent.all(), "Nb. of filled variables columns not consistent with declared nb. of variables for " \	1✔
51	"problems:\n %s"%(str(eqs_class_df.loc[~is_consistent]))
52
53	return eqs_class_df	1✔
54
55
56	EQS_CLASS_DF = load_class_equations_csv (filepath_eqs = PATH_CLASS_EQS_CSV,)	1✔
57
58	# Number of equations in dataset
59	N_EQS = EQS_CLASS_DF.shape[0]	1✔
60
61	# Size of units vector
62	CLASS_UNITS_VECTOR_SIZE = 7	1✔
63
64	# ---------------------------------------------------------------------------------------------------------------------
65	# --------------------------------------------------- UNITS UTILS ----------------------------------------------------
66	# ---------------------------------------------------------------------------------------------------------------------
67
68	# Gets units from variable name
69	def get_units (i_eq, i_var = 0, output_var = False):	1✔
70	"""
71	Gets units of variable.
72	Parameters
73	----------
74	i_eq : int
75	Equation number in the set of equations.
76	i_var : int
77	Variable id in its equation line.
78	output_var : bool
79	If True, returns units of output variable, otherwise returns units of input variable specified by i_var.
80	Returns
81	-------
82	units : numpy.array of shape (CLASS_UNITS_VECTOR_SIZE,) of floats
83	Units of variable.
84	"""
85
86	units = np.zeros(CLASS_UNITS_VECTOR_SIZE) # (CLASS_UNITS_VECTOR_SIZE,)	1✔
87
88	if output_var:	1✔
89	res = EQS_CLASS_DF.iloc[i_eq]["Output_units"]	1✔
90	else:
91	res = EQS_CLASS_DF.iloc[i_eq]["v%i_units"%(i_var)]	1✔
92
93	res = res[1:-1] # Removing brackets	1✔
94	res = np.fromstring(res, dtype=float, sep=',')	1✔
95	units[:len(res)] = res # Replacing units by elements in res	1✔
96
97	return units	1✔
98
99
100
101	# ---------------------------------------------------------------------------------------------------------------------
102	# -------------------------------------------------- CLASS PROBLEM ---------------------------------------------------
103	# ---------------------------------------------------------------------------------------------------------------------
104	CONST_LOCAL_DICT = {"pi" : np.pi}	1✔
105
106	class ClassProblem:	1✔
107	"""
108	Represents a single Class SR benchmark problem.
109	(See https://arxiv.org/abs/2312.01816 for details).
110
111	Attributes
112	----------
113	i_eq : int
114	Equation number in the set of equations.
115	eq_name : str
116	Equation name in the set of equations (e.g. 'Harmonic Oscillator').
117	n_vars : int
118	Number of input variables.
119	n_spe : int
120	Number of realization specific free constants.
121	eq_df : pandas.core.series.Series
122	Underlying pandas dataframe line of this equation.
123	original_var_names : bool
124	Using original variable names (e.g. theta, sigma etc.) and original output variable name (e.g. f, E etc.) if
125	True, using x0, x1 ... as input variable names k0, k1,... as spe free consts names and y as output variable name
126	otherwise.
127
128	y_name_original : str
129	Name of output variable as in the Class dataset.
130	y_name : str
131	Name of output variable.
132	y_units : array_like of shape (CLASS_UNITS_VECTOR_SIZE,) of floats
133	Units of output variables.
134
135	X_names_original : array_like of shape (n_vars,) of str
136	Names of input variables as in the Class dataset.
137	X_names : array_like of shape (n_vars,) of str
138	Names of input variables.
139	X_lows : array_like of shape (n_vars,) of floats
140	Lowest values taken by input variables.
141	X_highs : array_like of shape (n_vars,) of floats
142	Highest values taken by input variables.
143	X_units : array_like of shape (n_vars, CLASS_UNITS_VECTOR_SIZE,) of floats
144	Units of input variables.
145
146	K_names_original : array_like of shape (n_spe,) of str
147	Names of spe free consts as in the Class dataset.
148	K_names : array_like of shape (n_spe,) of str
149	Names of spe free consts.
150	K_lows : array_like of shape (n_spe,) of floats
151	Lowest values taken by spe free consts.
152	K_highs : array_like of shape (n_spe,) of floats
153	Highest values taken by spe free consts.
154	K_units : array_like of shape (n_spe, CLASS_UNITS_VECTOR_SIZE,) of floats
155	Units of spe free consts.
156
157	formula_original : str
158	Formula as in the Class dataset.
159
160	X_sympy_symbols : array_like of shape (n_vars,) of sympy.Symbol
161	Sympy symbols representing each input variables with assumptions (negative, positive etc.).
162	sympy_X_symbols_dict : dict of {str : sympy.Symbol}
163	Input variables names to sympy symbols (w assumptions), can be passed to sympy.parsing.sympy_parser.parse_expr
164	as local_dict.
165	K_sympy_symbols : array_like of shape (n_spe,) of sympy.Symbol
166	Sympy symbols representing each spe free consts with assumptions (negative, positive etc.).
167	sympy_K_symbols_dict : dict of {str : sympy.Symbol}
168	Spe free consts names to sympy symbols (w assumptions), can be passed to sympy.parsing.sympy_parser.parse_expr
169
170	local_dict : dict of {str : sympy.Symbol or float}
171	Input variables names and spe free consts to sympy symbols (w assumptions) and constants (eg. pi : np.pi etc.),
172	can be passed to sympy.parsing.sympy_parser.parse_expr as local_dict.
173	formula_sympy : sympy expression
174	Formula in sympy.
175	formula_sympy_eval : sympy expression
176	Formula in sympy with evaluated fixed constants (eg. pi -> 3.14... etc).
177	formula_latex : str
178	Formula in latex.
179
180	"""
181
182	def __init__(self, i_eq = None, eq_name = None, original_var_names = False):	1✔
183	"""
184	Loads a Class problem based on its number in the set or its equation name.
185	Parameters
186	----------
187	i_eq : int
188	Equation number in the set of equations.
189	eq_name : str
190	Equation name in the set of equations (e.g. 'Harmonic Oscillator').
191	original_var_names : bool
192	Using original variable names (e.g. theta, sigma etc.) and original output variable name (e.g. f, E etc.) if
193	True, using x0, x1 ... as input variable names and y as output variable name otherwise.
194	"""
195	# Selecting equation line in dataframe
196	if i_eq is not None:	1✔
197	self.eq_df = EQS_CLASS_DF.iloc[i_eq] # pandas.core.series.Series	1✔
198	elif eq_name is not None:	×
199	self.eq_df = EQS_CLASS_DF[EQS_CLASS_DF ["Name"] == eq_name ].iloc[0] # pandas.core.series.Series	×
200	else:
201	raise ValueError("At least one of equation number (i_eq) or equation name (eq_name) should be specified to select a Class problem.")	×
202
203
204	# Equation number
205	self.i_eq = i_eq # int	1✔
206	# Code name of equation (eg. 'Harmonic Oscillator')
207	self.eq_name = self.eq_df["Name"] # str	1✔
208	# SRBench style name
209	self.SRBench_name = "class_%i"%(self.i_eq) # str	1✔
210	# Number of input variables
211	self.n_vars = int(self.eq_df["# variables"]) # int	1✔
212	# Number of realization specific free constants
213	self.n_spe = int(self.eq_df["# spe"]) # int	1✔
214
215	# --------- Handling input variables vs spe free consts ---------
216	# Total number of variables
217	n_v = self.n_vars + self.n_spe # int	1✔
218	v_ids = np.array( [i_var for i_var in range(1, n_v +1)] ) # (n_v,)	1✔
219	v_types = self.eq_df[np.array( [ "v%i_type"%(i_var) for i_var in v_ids ] )].to_numpy() # (n_v,)	1✔
220	is_var = v_types == "var" # (n_v,)	1✔
221	is_spe = v_types == "spe" # (n_v,)	1✔
222
223
224	# Using x0, x1 ... and y names or original names (e.g. theta, sigma, f etc.)
225	self.original_var_names = original_var_names # bool	1✔
226
227	# ----------- y : output variable -----------
228	# Name of output variable
229	self.y_name_original = self.eq_df["Output_name"] # str	1✔
230	# Name of output variable : y or original name (eg. f, E etc.)
231	self.y_name = self.y_name_original if self.original_var_names else 'y' # str	1✔
232	# Units of output variables
233	self.y_units = get_units(i_eq=self.i_eq, output_var=True) # (CLASS_UNITS_VECTOR_SIZE,)	1✔
234
235	# ----------- X : input variables -----------
236	var_ids = v_ids[is_var] # (n_vars,)	1✔
237	var_ids_str = np.array( [ "v%i"%(i_var) for i_var in var_ids ] ).astype(str) # (n_vars,)	1✔
238	# Names of input variables
239	self.X_names_original = np.array( [ self.eq_df[ id + "_name" ] for id in var_ids_str ] ).astype(str) # (n_vars,)	1✔
240	X_names_xi_style = np.array( [ "x%i"%(i_var) for i_var in range(self.n_vars) ] ).astype(str) # (n_vars,)	1✔
241	self.X_names = self.X_names_original if self.original_var_names else X_names_xi_style # (n_vars,)	1✔
242	# Lowest values taken by input variables
243	self.X_lows = np.array( [ self.eq_df[ id + "_low" ] for id in var_ids_str ] ).astype(float) # (n_vars,)	1✔
244	# Highest values taken by input variables
245	self.X_highs = np.array( [ self.eq_df[ id + "_high" ] for id in var_ids_str ] ).astype(float) # (n_vars,)	1✔
246	# Units of input variables
247	self.X_units = np.array( [ get_units(i_eq=self.i_eq, i_var=i_var) for i_var in var_ids ] ).astype(float) # (n_vars, CLASS_UNITS_VECTOR_SIZE,)	1✔
248
249	# Input variables as sympy symbols
250	self.X_sympy_symbols = []	1✔
251	for i in range(self.n_vars):	1✔
252	self.X_sympy_symbols.append (su.sympy_symbol_with_assumptions_from_range(name = self.X_names[i],	1✔
253	low = self.X_lows [i],
254	high = self.X_highs[i],
255	))
256	# Input variables names to sympy symbols dict
257	self.sympy_X_symbols_dict = {self.X_names[i] : self.X_sympy_symbols[i] for i in range(self.n_vars)} # (n_vars,)	1✔
258	# Dict to use to read original class dataset formula
259	# Original names to symbols in usage (i.e. symbols having original names or not)
260	# eg. 'theta' -> theta symbol etc. (if original_var_names=True) or 'theta' -> x0 symbol etc. (else)
261	self.sympy_original_to_X_symbols_dict = {self.X_names_original[i] : self.X_sympy_symbols[i] for i in range(self.n_vars)} # (n_vars,)	1✔
262	# NB: if original_var_names=True, then self.sympy_X_symbols_dict = self.sympy_original_to_X_symbols_dict
263
264	# ----------- K : spe free consts -----------
265
266	spe_ids = v_ids[is_spe] # (n_spe,)	1✔
267	spe_ids_str = np.array( [ "v%i"%(i_spe) for i_spe in spe_ids ] ).astype(str) # (n_spe,)	1✔
268	# Names of spe free consts
269	self.K_names_original = np.array( [ self.eq_df[ id + "_name" ] for id in spe_ids_str ] ).astype(str) # (n_spe,)	1✔
270	K_names_ki_style = np.array( [ "k%i"%(i_spe) for i_spe in range(self.n_spe) ] ).astype(str) # (n_spe,)	1✔
271	self.K_names = self.K_names_original if self.original_var_names else K_names_ki_style # (n_spe,)	1✔
272	# Lowest values taken by spe free consts
273	self.K_lows = np.array( [ self.eq_df[ id + "_low" ] for id in spe_ids_str ] ).astype(float) # (n_spe,)	1✔
274	# Highest values taken by spe free consts
275	self.K_highs = np.array( [ self.eq_df[ id + "_high" ] for id in spe_ids_str ] ).astype(float) # (n_spe,)	1✔
276	# Units of spe free consts
277	self.K_units = np.array( [ get_units(i_eq=self.i_eq, i_var=i_var) for i_var in spe_ids ] ).astype(float) # (n_spe, CLASS_UNITS_VECTOR_SIZE,)	1✔
278
279	# Spe free consts as sympy symbols
280	self.K_sympy_symbols = []	1✔
281	for i in range(self.n_spe):	1✔
282	self.K_sympy_symbols.append (su.sympy_symbol_with_assumptions_from_range(name = self.K_names[i],	1✔
283	low = self.K_lows [i],
284	high = self.K_highs[i],
285	))
286	# Spe free consts names to sympy symbols dict
287	self.sympy_K_symbols_dict = {self.K_names[i] : self.K_sympy_symbols[i] for i in range(self.n_spe)} # (n_spe,)	1✔
288	# Dict to use to read original class dataset formula
289	# Original names to symbols in usage (i.e. symbols having original names or not)
290	# eg. 'theta' -> theta symbol etc. (if original_var_names=True) or 'theta' -> k0 symbol etc. (else)
291	self.sympy_original_to_K_symbols_dict = {self.K_names_original[i] : self.K_sympy_symbols[i] for i in range(self.n_spe)} # (n_spe,)	1✔
292	# NB: if original_var_names=True, then self.sympy_K_symbols_dict = self.sympy_original_to_K_symbols_dict
293
294	# ----------- Formula -----------
295	self.formula_original = self.eq_df["Formula"] # (str)	1✔
296
297	self.v_local_dict = {}	1✔
298	self.v_local_dict.update(self.sympy_original_to_X_symbols_dict)	1✔
299	self.v_local_dict.update(self.sympy_original_to_K_symbols_dict)	1✔
300
301
302	# Declaring input variables via local_dict to avoid confusion
303	# Eg. So sympy knows that we are referring to gamma as a variable and not the function etc.
304	# evaluate = False avoids eg. sin(theta) = 0 when theta domain = [0,5] ie. nonzero=False, but no need for this
305	# if nonzero assumption is not used
306	evaluate = False	1✔
307	self.formula_sympy = sympy.parsing.sympy_parser.parse_expr(self.formula_original,	1✔
308	local_dict = self.v_local_dict,
309	evaluate = evaluate)
310
311	# Local dict : dict of input variables (sympy_original_to_X_symbols_dict) and fixed constants (pi -> 3.14.. etc)
312	self.local_dict = {}	1✔
313	self.local_dict.update(self.v_local_dict)	1✔
314	self.local_dict.update(CONST_LOCAL_DICT)	1✔
315	self.formula_sympy_eval = sympy.parsing.sympy_parser.parse_expr(self.formula_original,	1✔
316	local_dict = self.local_dict,
317	evaluate = evaluate)
318	# Latex formula
319	self.formula_latex = sympy.printing.latex(self.formula_sympy)	1✔
320	return None	1✔
321
322
323	def target_function(self, X, K):	1✔
324	"""
325	Evaluates X with target function, using K values.
326	Parameters
327	----------
328	X : numpy.array of shape (n_vars, ?,) of floats
329	Input variables.
330	K : numpy.array of shape (n_spe,) of floats
331	Spe free consts.
332	Returns
333	-------
334	y : numpy.array of shape (?,) of floats
335	"""
336	# Getting sympy function
337	sympy_symbols = self.X_sympy_symbols + self.K_sympy_symbols	1✔
338	f = sympy.lambdify(sympy_symbols, self.formula_sympy, "numpy")	1✔
339	mapping_vals = {}	1✔
340	# Mapping between variables names and their data value
341	mapping_vals.update({self.X_names[i]: X[i] for i in range(self.n_vars)})	1✔
342	# Mapping between spe free consts names and their data value
343	mapping_vals.update({self.K_names[i]: K[i] for i in range(self.n_spe)})	1✔
344	# Evaluation
345	# Forcing float type so if some symbols are not evaluated as floats (eg. if some variables are not declared
346	# properly in source file) resulting partly symbolic expressions will not be able to be converted to floats
347	# and an error can be raised).
348	# This is also useful for detecting issues such as sin(theta) = 0 because theta.is_nonzero = False -> the result
349	# is just an int of float
350	y = f(**mapping_vals).astype(float)	1✔
351	return y	1✔
352
353	def generate_data_points (self, n_samples = 1_000, n_realizations = 10, return_K = False):	1✔
354	"""
355	Generates data points accordingly for this Class problem.
356	Parameters
357	----------
358	n_samples : int
359	Number of samples to draw. By default, 1e3.
360	n_realizations : int
361	Number of realizations to draw. By default, 10.
362	return_K : bool
363	If True, returns K values used to generate data as well.
364	Returns
365	-------
366	multi_X : numpy.array of shape (n_realizations, n_vars, n_samples,) of floats,
367	multi_y : numpy.array of shape (n_realizations, n_samples,) of floats
368	(multi_K : numpy.array of shape (n_realizations, n_spe,) of floats)
369	"""
370	multi_X = []	1✔
371	multi_y = []	1✔
372	multi_K = []	1✔
373
374	for i_real in range(n_realizations):	1✔
375	# Random K sample
376	K = np.stack([np.random.uniform(self.K_lows[i_var], self.K_highs[i_var], ) for i_var in range(self.n_spe)]) # (n_spe,)	1✔
377	# Random X sample
378	X = np.stack([np.random.uniform(self.X_lows[i_var], self.X_highs[i_var], n_samples) for i_var in range(self.n_vars)]) # (n_vars, n_samples)	1✔
379	# Evaluating formula
380	y = self.target_function(X=X, K=K) # (n_samples,)	1✔
381
382	multi_X.append(X)	1✔
383	multi_y.append(y)	1✔
384	multi_K.append(K)	1✔
385
386	multi_X = np.stack(multi_X) # (n_realizations, n_vars, n_samples)	1✔
387	multi_y = np.stack(multi_y) # (n_realizations, n_samples)	1✔
388	multi_K = np.stack(multi_K) # (n_realizations, n_spe)	1✔
389
390	if return_K:	1✔
391	return multi_X, multi_y, multi_K	×
392	else:
393	return multi_X, multi_y	1✔
394
395	def get_sympy(self, K_vals=None):	1✔
396	"""
397	Gets sympy expression of the formula evaluated with spe free consts.
398	Parameters
399	----------
400	K_vals : numpy.array of shape (?, n_spe,) of floats or None
401	Values to evaluate spe free consts with, if None, uses random values and returns only one realization.
402	Returns
403	-------
404	sympy_expr : np.array of shape (?, n_spe,) of Sympy Expression
405	"""
406	if K_vals is None:	×
407	K = np.stack([np.random.uniform(self.K_lows[i_var], self.K_highs[i_var], ) for i_var in range(self.n_spe)]) # (n_spe,)	×
408	K_vals = np.stack([K,]) # (?, n_spe) = (1, n_spe)	×
409
410	sympy_expr = []	×
411	for K in K_vals:	×
412	K_dict = {self.K_sympy_symbols[i]: K[i] for i in range(self.n_spe)}	×
413	expr = self.formula_sympy.subs(K_dict)	×
414	sympy_expr.append(expr)	×
415	sympy_expr = np.array(sympy_expr) # (?,)	×
416
417	return sympy_expr	×
418
419	def show_sample(self, n_samples = 10_000, n_realizations = 10, do_show = True, save_path = None):	1✔
420	multi_X, multi_y = self.generate_data_points(n_samples=n_samples, n_realizations=n_realizations)	×
421
422	n_dim = multi_X.shape[1]	×
423	fig, ax = plt.subplots(n_dim, 1, figsize=(15, n_dim * 6))	×
424	fig.suptitle(self.formula_original)	×
425	for i in range(n_dim):	×
426	curr_ax = ax if n_dim == 1 else ax[i]	×
427	curr_ax.set_xlabel("%s : %s" % (self.X_names[i], self.X_units[i]))	×
428	curr_ax.set_ylabel("%s : %s" % (self.y_name , self.y_units))	×
429	for i_real in range(n_realizations):	×
430	curr_ax.plot(multi_X[i_real,i], multi_y[i_real], '.', markersize=1.)	×
431	fig.tight_layout()	×
432	if save_path is not None:	×
433	fig.savefig(save_path, dpi=200)	×
434	if do_show:	×
435	plt.show()	×
436
437	def get_prefix_expression (self):	×
438	"""
439	Gets the prefix expression of the formula.
440	Returns
441	-------
442	dict :
443	tokens_str : numpy.array of str
444	List of tokens in the expression.
445	arities : numpy.array of int
446	List of arities of the tokens.
447	tokens : numpy.array of sympy.core
448	List of sympy tokens.
449	"""
450	return su.sympy_to_prefix(self.formula_sympy)	×
451
452	def __str__(self):	×
453	return "ClassProblem : %s : %s\n%s"%(self.i_eq, self.eq_name, str(self.formula_sympy))	×
454
455	def __repr__(self):	×
456	return str(self)	×

WassimTenachi / PhySO / #16

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous