Learning from Failures¶

# from bookutils import YouTubeVideo
# YouTubeVideo("w4u5gCgPlmg")

import bookutils.setup

from typing import List, Tuple, Dict, Any, Optional

from fuzzingbook.Grammars import Grammar, EXPR_GRAMMAR, reachable_nonterminals, is_valid_grammar
from fuzzingbook.GrammarFuzzer import GrammarFuzzer, expansion_to_children, DerivationTree, tree_to_string, display_tree, is_nonterminal
from fuzzingbook.Parser import EarleyParser

from math import tan as rtan
from math import cos as rcos
from math import sin as rsin

"""
This file contains the code under test for the example bug.
The sqrt() method fails on x <= 0.
"""
def task_sqrt(x):
    """Computes the square root of x, using the Newton-Raphson method"""
    if x <= -12 and x >= -42:
        x = 0  # Guess where the bug is :-)
    else:
        x = 1
    x = max(x, 0)
    approx = None
    guess = x / 2
    while approx != guess:
        approx = guess
        guess = (approx + x / approx) / 2
    return approx

def task_tan(x):
    return rtan(x)

def task_cos(x):
    return rcos(x)

def task_sin(x):
    return rsin(x)

CALC_GRAMMAR: Grammar = {
    "<start>":
        ["<function>(<term>)"],

    "<function>":
        ["sqrt", "tan", "cos", "sin"],

    "<term>": ["-<value>", "<value>"],

    "<value>":
        ["<integer>.<digits>",
         "<integer>"],

    "<integer>":
        ["<lead-digit><digits>", "<digit>"],

    "<digits>":
        ["<digit><digits>", "<digit>"],

    "<lead-digit>":  # First digit cannot be zero
        ["1", "2", "3", "4", "5", "6", "7", "8", "9"],

    "<digit>":
        ["0", "1", "2", "3", "4", "5", "6", "7", "8", "9"],
}

# Load initial input files
initial_sample_list = ['sqrt(-16)', 'sqrt(4)']

from enum import Enum

class OracleResult(Enum):
    BUG = "BUG"
    NO_BUG = "NO_BUG"
    UNDEF = "UNDEF"

    def __str__(self):
        return self.value

import sys

SUBJECT = "calculator"

def sample_runner(sample):
    testcode = sample

    try:
        # Simply execute the calculator code, with the functions replaced
        exec(testcode, {"sqrt": task_sqrt, "tan": task_tan, "sin": task_sin, "cos": task_cos}, {})
        return OracleResult.NO_BUG
    except ZeroDivisionError:
        return OracleResult.BUG
    except Exception as e:
        print(e, file=sys.stderr)
        return OracleResult.UNDEF

sample = "sqrt(-16)"
sample_runner(sample)

assert sample_runner("sqrt(-23)") == OracleResult.BUG
assert sample_runner("sqrt(44)") == OracleResult.NO_BUG
assert sample_runner("cos(-9)") == OracleResult.NO_BUG

sample_runner("undef_function(QUERY)")

import pandas
import numpy
import matplotlib

# Executes a list of samples and return the execution outcome (label)
# The function returns a pandas dataframe
def execute_samples(sample_list):
    data = []
    for sample in sample_list:
        result = sample_runner(sample)
        data.append({"oracle": result })

    return pandas.DataFrame.from_records(data)

sample_list = ["sqrt(-20)", "cos(2)", "sqrt(-100)", "undef_function(foo)"]

labels = execute_samples(sample_list)
labels

for i, row in enumerate(labels['oracle']): print(sample_list[i].ljust(30) + str(row))

clean_data = labels.drop(labels[labels.oracle.astype(str) == "UNDEF"].index)
clean_data

oracle = execute_samples(sample_list)
for i, row in enumerate(oracle['oracle']):
    print(sample_list[i].ljust(30) + str(row))

guess_samples = ['cos(-16)', 'tan(-16)', 'sqrt(-100)', 'sqrt(-20.23412431234123)']

guess_oracle = execute_samples(guess_samples)

for i, row in enumerate(guess_oracle['oracle']):
    print(guess_samples[i].ljust(30) + str(row))

from abc import ABC, abstractmethod

class Feature(ABC):
    '''
    The abstract base class for grammar features.

    Args:
        name : A unique identifier name for this feature. Should not contain Whitespaces.
               e.g., 'type(<feature>@1)'
        rule : The production rule (e.g., '<function>' or '<value>').
        key  : The feature key (e.g., the chosen alternative or rule itself).
    '''

    def __init__(self, name: str, rule: str, key: str, /,
                 friendly_name: str = None) -> None:
        self.name = name
        self.rule = rule
        self.key = key
        self._friendly_name = friendly_name or name
        super().__init__()

    def __repr__(self) -> str:
        '''Returns a printable string representation of the feature.'''
        return self.name_rep()

    @abstractmethod
    def name_rep(self) -> str:
        pass

    def friendly_name(self) -> str:
        return self._friendly_name

    @abstractmethod
    def get_feature_value(self, derivation_tree) -> float:
        '''Returns the feature value for a given derivation tree of an input.'''
        pass

    def replace(self, new_key: str) -> 'Feature':
        '''Returns a new feature with the same name but a different key.'''
        return self.__class__(self.name, self.rule, new_key)

class ExistenceFeature(Feature):
    '''
    This class represents existence features of a grammar. Existence features indicate
    whether a particular production rule was used in the derivation sequence of an input.
    For a given production rule P -> A | B, a production existence feature for P and
    alternative existence features for each alternative (i.e., A and B) are defined.

    name : A unique identifier name for this feature. Should not contain Whitespaces.
           e.g., 'exist(<digit>@1)'
    rule : The production rule.
    key  : The feature key, equal to the rule attribute for production features,
           or equal to the corresponding alternative for alternative features.
    '''
    def __init__(self, name: str, rule: str, key: str,
                 friendly_name: str = None) -> None:
        super().__init__(name, rule, key, friendly_name=friendly_name)

    def name_rep(self) -> str:
        if self.rule == self.key:
            return f"exists({self.rule})"
        else:
            return f"exists({self.rule} == {self.key})"

    def get_feature_value(self, derivation_tree) -> float:
        '''Returns the feature value for a given derivation tree of an input.'''
        raise NotImplementedError

    def get_feature_value(self, derivation_tree: DerivationTree) -> float:
        '''Counts the number of times this feature was matched in the derivation tree.'''
        (node, children) = derivation_tree

        # The local match count (1 if the feature is matched for the current node, 0 if not)
        count = 0

        # First check if the current node can be matched with the rule
        if node == self.rule:

            # Production existance feature
            if self.rule == self.key:
                count = 1

            # Production alternative existance feature
            # We compare the children of the expansion with the actual children
            else:
                expansion_children = list(map(lambda x: x[0], expansion_to_children(self.key)))
                node_children = list(map(lambda x: x[0], children))
                if expansion_children == node_children:
                    count= 1

        # Recursively compute the counts for all children and return the sum for the whole tree
        for child in children:
            count = max(count, self.get_feature_value(child)) 

        return count

from numpy import nanmax, isnan

class NumericInterpretation(Feature):
    '''
    This class represents numeric interpretation features of a grammar. These features
    are defined for productions that only derive words composed of the characters
    [0-9], '.', and '-'. The returned feature value corresponds to the maximum
    floating-point number interpretation of the derived words of a production.

    name : A unique identifier name for this feature. Should not contain Whitespaces.
           e.g., 'num(<integer>)'
    rule : The production rule.
    '''
    def __init__(self, name: str, rule: str, /, 
                 friendly_name: str = None) -> None:
        super().__init__(name, rule, rule, friendly_name=friendly_name)

    def name_rep(self) -> str:
        return f"num({self.key})"

    def get_feature_value(self, derivation_tree) -> float:
        '''Returns the feature value for a given derivation tree of an input.'''
        raise NotImplementedError

    def get_feature_value(self, derivation_tree: DerivationTree) -> float:
        '''Determines the maximum float of this feature in the derivation tree.'''
        (node, children) = derivation_tree

        value = float('nan')
        if node == self.rule:
            try:
                #print(self.name, float(tree_to_string(derivation_tree)))
                value = float(tree_to_string(derivation_tree))
            except ValueError:
                #print(self.name, float(tree_to_string(derivation_tree)), "err")
                pass

        # Return maximum value encountered in tree, ignoring all NaNs
        tree_values = [value] + [self.get_feature_value(c) for c in children]
        if all(isnan(tree_values)):
            return value
        else:
            return nanmax(tree_values)

def extract_existence_features(grammar: Grammar) -> List[ExistenceFeature]:
    '''
        Extracts all existence features from the grammar and returns them as a list.
        grammar : The input grammar.
    '''

    features = []

    for rule in grammar:
        # add the rule
        features.append(ExistenceFeature(f"exists({rule})", rule, rule))
        # add all alternatives
        for count, expansion in enumerate(grammar[rule]):
            name = f"exists({rule}@{count})"
            friendly_name = f"{rule} == {repr(expansion)}"
            feature = ExistenceFeature(name, rule, expansion,
                                       friendly_name=friendly_name)
            features.append(feature)

    return features

from collections import defaultdict
import re

# Regex for non-terminal symbols in expansions
RE_NONTERMINAL = re.compile(r'(<[^<> ]*>)')

def extract_numeric_features(grammar: Grammar) -> List[NumericInterpretation]:
    '''
        Extracts all numeric interpretation features from the grammar and returns them as a list.

        grammar : The input grammar.
    '''

    features = []

    # Mapping from non-terminals to derivable terminal chars
    derivable_chars = defaultdict(set)

    for rule in grammar:
        for expansion in grammar[rule]:
            # Remove non-terminal symbols and whitespace from expansion
            terminals = re.sub(RE_NONTERMINAL, '', expansion).replace(' ', '')

            # Add each terminal char to the set of derivable chars
            for c in terminals:
                derivable_chars[rule].add(c)

    # Repeatedly update the mapping until convergence
    while True:
        updated = False
        for rule in grammar:
            for r in reachable_nonterminals(grammar, rule):
                before = len(derivable_chars[rule])
                derivable_chars[rule].update(derivable_chars[r])
                after = len(derivable_chars[rule])

                # Set of derivable chars was updated
                if after > before:
                    updated = True

        if not updated:
            break

    numeric_chars = set(['0','1','2','3','4','5','6','7','8','9','.','-'])

    for key in derivable_chars:
        # Check if derivable chars contain only numeric chars
        if len(derivable_chars[key] - numeric_chars) == 0:
            name = f"num({key})"
            friendly_name = f"{key}"

            features.append(NumericInterpretation(f"num({key})", key,
                                                  friendly_name=friendly_name))

    return features

def extract_all_features(grammar: Grammar) -> List[Feature]:
    return (extract_existence_features(grammar)
            + extract_numeric_features(grammar))

extract_all_features(CALC_GRAMMAR)

[f.friendly_name() for f in extract_all_features(CALC_GRAMMAR)]

def collect_features(sample_list: List[str],
                     grammar: Grammar) -> pandas.DataFrame:

    data = []

    # parse grammar and extract features
    all_features = extract_all_features(grammar)

    # iterate over all samples
    for sample in sample_list:
        parsed_features = {}
        parsed_features["sample"] = sample
        # initate dictionary
        for feature in all_features:
            parsed_features[feature.name] = 0

        # Obtain the parse tree for each input file
        earley = EarleyParser(grammar)
        for tree in earley.parse(sample):

            for feature in all_features:
                parsed_features[feature.name] = feature.get_feature_value(tree)

        data.append(parsed_features)

    return pandas.DataFrame.from_records(data)

sample_list = ["sqrt(-900)", "sin(24)", "cos(-3.14)"]
collect_features(sample_list, CALC_GRAMMAR)

# TODO: handle multiple trees
def compute_feature_values(sample: str, grammar: Grammar, features: List[Feature]) -> Dict[str, float]:
    '''
        Extracts all feature values from an input.

        sample   : The input.
        grammar  : The input grammar.
        features : The list of input features extracted from the grammar.

    '''
    earley = EarleyParser(CALC_GRAMMAR)

    features = {}
    for tree in earley.parse(sample):
        for feature in extract_all_features(CALC_GRAMMAR):
            features[feature.name_rep()] = feature.get_feature_value(tree)
    return features

all_features = extract_all_features(CALC_GRAMMAR)
for sample in sample_list:
    print(f"Features of {sample}:")
    features = compute_feature_values(sample, CALC_GRAMMAR, all_features)
    for feature, value in features.items():
        print(f"    {feature}: {value}")

import sklearn
from sklearn.tree import DecisionTreeClassifier
from sklearn.feature_extraction import DictVectorizer

# Features for each input, one dict per input
features = [
    {'function-sqrt': 1, 'function-cos': 0, 'function-sin': 0, 'number': -900}, # sqrt(-900)
    {'function-sqrt': 0, 'function-cos': 1, 'function-sin': 0, 'number': 300}, # cos(300)
    {'function-sqrt': 1, 'function-cos': 0, 'function-sin': 0, 'number': -1}, # sqrt(-1)
    {'function-sqrt': 0, 'function-cos': 1, 'function-sin': 0, 'number': -10}, # cos(-10)
    {'function-sqrt': 0, 'function-cos': 0, 'function-sin': 1, 'number': 36}, # sin(36)
    {'function-sqrt': 0, 'function-cos': 0, 'function-sin': 1, 'number': -58}, # sin(-58)
    {'function-sqrt': 1, 'function-cos': 0, 'function-sin': 0, 'number': 27}, # sqrt(27)
]

# Labels for each input
oracle = [
    OracleResult.BUG,
    OracleResult.NO_BUG,
    OracleResult.BUG,
    OracleResult.NO_BUG,
    OracleResult.NO_BUG,
    OracleResult.NO_BUG,
    OracleResult.NO_BUG
]

# Transform to numpy array
vec = DictVectorizer()
X = vec.fit_transform(features).toarray()

# Fix the random state to produce a deterministic result (for illustration purposes only)
clf = DecisionTreeClassifier(random_state=10)

# sci-kit learn requires an array of strings
oracle_clean = [str(c) for c in oracle]
clf = clf.fit(X, oracle_clean)

import graphviz

def show_decision_tree(clf, feature_names):
    dot_data = sklearn.tree.export_graphviz(clf, out_file=None, 
                                    feature_names=feature_names,
                                    class_names=["BUG", "NO_BUG"],  
                                    filled=True, rounded=True)  
    return graphviz.Source(dot_data)

show_decision_tree(clf, vec.get_feature_names_out())

import math

def friendly_decision_tree(clf, feature_names,
                           class_names = ['NO_BUG', 'BUG'],
                           indent=0):
    def _tree(index, indent):
        s = ""
        feature = clf.tree_.feature[index]
        feature_name = feature_names[feature]
        threshold = clf.tree_.threshold[index]
        value = clf.tree_.value[index]
        class_ = int(value[0][0])
        class_name = class_names[class_]
        left = clf.tree_.children_left[index]
        right = clf.tree_.children_right[index]
        if left == right:
            # Leaf node
            s += " " * indent + class_name + "\n"
        else:
            if math.isclose(threshold, 0.5):
                s += " " * indent + f"if {feature_name}:\n"
                s += _tree(right, indent + 2)
                s += " " * indent + f"else:\n"
                s += _tree(left, indent + 2)
            else:
                s += " " * indent + f"if {feature_name} <= {threshold:.4f}:\n"
                s += _tree(left, indent + 2)
                s += " " * indent + f"else:\n"
                s += _tree(right, indent + 2)
        return s

    ROOT_INDEX = 0
    return _tree(ROOT_INDEX, indent)

print(friendly_decision_tree(clf, vec.get_feature_names_out()))

def train_tree(data):
    sample_bug_count = len(data[(data["oracle"].astype(str) == "BUG")])
    assert sample_bug_count > 0, "No bug samples found"
    sample_count = len(data)

    clf = DecisionTreeClassifier(min_samples_leaf=1,
                                     min_samples_split=2,  # minimal value
                                     max_features=None,
                                     max_depth=5, # max depth of the decision tree
                                     class_weight={str("BUG"): (1.0/sample_bug_count),
                                                   str("NO_BUG"):
                                                       (1.0/(sample_count - sample_bug_count))})
    clf = clf.fit(data.drop('oracle', axis=1), data['oracle'].astype(str))
    # MARTIN: This is optional, but is a nice extesion that results in nicer decision trees
    # clf = treetools.remove_infeasible(clf, features)
    return clf

# Features for each input, one dict per input
features = [
    {'function-sqrt': 1, 'function-cos': 0, 'function-sin': 0, 'number': -900},
    {'function-sqrt': 0, 'function-cos': 1, 'function-sin': 0, 'number': 300},
    {'function-sqrt': 1, 'function-cos': 0, 'function-sin': 0, 'number': -1},
    {'function-sqrt': 0, 'function-cos': 1, 'function-sin': 0, 'number': -10},
    {'function-sqrt': 0, 'function-cos': 0, 'function-sin': 1, 'number': 36},
    {'function-sqrt': 0, 'function-cos': 0, 'function-sin': 1, 'number': -58},
    {'function-sqrt': 1, 'function-cos': 0, 'function-sin': 0, 'number': 27},
]

# Labels for each input
oracle = [
    "BUG",
    "NO_BUG",
    "BUG",
    "NO_BUG",
    "NO_BUG",
    "NO_BUG",
    "NO_BUG"
]

# We can use the sklearn DictVectorizer to transform the features to numpy array:
# Notice: Use the correct labeling of the feature_names

# vec = DictVectorizer()
# X_vec = vec.fit_transform(features).toarray()
# feature_names = vec.get_feature_names_out()

# We can also use a pandas DataFrame and directly parse it to the decision tree learner
feature_names = ['function-sqrt', 'function-cos', 'function-sin', 'number']
X_data = pandas.DataFrame.from_records(features)

# Fix the random state to produce a deterministic result (for illustration purposes only)
clf = DecisionTreeClassifier(random_state=10)

# Train with DictVectorizer
# **Note:** The sklearn `DictVectorizer` uses an internal sort function as default. This will result in different feature_name indices. If you want to use the `Dictvectorizer` please ensure that you only access the feature_names with the function `vec.get_feature_names_out()`.
# We recommend that you use the `pandas` data frame, since this is also the format used in the feedback loop.
# clf = clf.fit(X_vec, oracle)

# Train with Pandas Dataframe
clf = clf.fit(X_data, oracle)

dot_data = sklearn.tree.export_graphviz(clf, out_file=None,
                                feature_names=feature_names,
                                class_names=["BUG", "NO BUG"],
                                filled=True, rounded=True)
graph = graphviz.Source(dot_data)

graph

print(friendly_decision_tree(clf, feature_names, class_names = ['NO_BUG', 'BUG']))

x = pandas.DataFrame.from_records(features)
bounds = pandas.DataFrame([{'feature': c, 'min': x[c].min(), 'max': x[c].max()}
                           for c in feature_names],
                          columns=['feature', 'min', 'max']).set_index(['feature']).transpose()

for count, path in enumerate(all_paths):
    negated_string_path = path.get(0).get_neg_ext(bounds)[0]
    for box_ in range(1, len(path)):
        negated_string_path += " " + str(path.get(box_).get_neg_ext(bounds)[0])
    print(f"Path {count}: {negated_string_path}, is_bug: {path.is_bug()}")

def extracting_prediction_paths(clf, feature_names, data):
    # determine the bounds
    bounds = pandas.DataFrame([{'feature': c, 'min': data[c].min(), 'max': data[c].max()}
                           for c in feature_names],
                          columns=['feature', 'min', 'max']).set_index(['feature']).transpose()

    # go through tree leaf by leaf
    all_reqs = set()
    for path in tree_to_paths(clf, feature_names):
        # generate conditions
        for i in range(0, len(path)+1):
            reqs_list = []
            bins = format(i, "#0{}b".format(len(path)+2))[2:]
            for p, b in zip(range(0, len(bins)), bins):
                r = path.get(p)
                if '1' == b:
                    reqs_list.append(r.get_neg_ext(bounds))
                else:
                    reqs_list.append([r.get_str_ext()])
            for reqs in all_combinations(reqs_list):
                all_reqs.add(", ".join(sorted(reqs)))
    return all_reqs

def all_combinations(reqs_lists):
    result = [[]]
    for reqs in reqs_lists:
        t = []
        for r in reqs:
            for i in result:
                t.append(i+[r])
        result = t
    return result

new_prediction_paths = extracting_prediction_paths(clf, feature_names, data=x)

for path in new_prediction_paths:
    print(path)