<commit_before># Probability Tables Implementation
# ---------------------------------
# @author <NAME>
# @author <NAME>

from itertools import product, combinations
from copy import deepcopy, copy
from warnings import warn
from functools import reduce

class Factor(object):
    """ Description:
        Usage:
            -
            -
            -
    """

    def __init__(self, unconditioned_vars, conditioned_vars, variable_domains):
        """ Creates a Factor object that encodes probabilities in a table """
        self._domains = variable_domains
        self._variables = set(conditioned_vars) | set(unconditioned_vars)
        self._unconditioned = unconditioned_vars
        self._conditioned = conditioned_vars
        entree_temp = [[(var, domain) for domain in self._domains[var]] for var in self._variables]
        self._entrees = list(sorted([element for element in product(*entree_temp)]))
        self._table = {}
        for entree in self._entrees:
            self._table[entree] = 0
        warn("Probabilities not set. Default = 0")

    def get_domains(self):
        return deepcopy(self._domains)

    def get_entrees(self):
        """ Retrieves all possible entrees to the table """
        return self._entrees

    def get_probability(self, **variables):
        """ Gets probability of a specific entree in table """
        variables = variables.items()
        if hasattr(variables, "__iter__"):
            to_sum = [entree for entree in self._entrees if set(entree) == set(variables)]
            if len(to_sum) != 1:
                raise UserWarning("Error in retrieving probability. Invalid method call.")
            return self._table[to_sum[0]]
        raise UserWarning("Error in retrieving probability. Invalid method call.")
        return 0

    def set_probability(self, val, **variables):
        """ Sets probability of a specific entree in table """
        assert set(variables) ^ set(self._variables) == set(), "Variables for assignment not Valid for JPT"
        variables = variables.items()
        entrees = [entree for entree in self._entrees if set(entree) | set(variables) == set(entree)]
        if len(entrees) > 1:
            print("ERROR IN SET PROBABILITY JPT")
        else:
            entree = entrees[0]
            self._table[entree] = val


    def __str__(self):
        """ Returns human readable Probability Table """
        longest_var = max(map(len,self._variables))
        def str_helper(s): return str(s).rjust(longest_var, " ").ljust(longest_var, " ")
        header = "P({unc}{given}{cond})".format(unc=", ".join(self._unconditioned),
                                            given=" | " if self._conditioned else "",
                                            cond=", ".join(self._conditioned))
        var_header = str_helper("{vars}".format(vars=" | ".join(self._variables)))
        lst_of_entrees = [" | ".join(map(str_helper, map(lambda x: x[1], k))) + " | " + str(v) for k, v in sorted(self._table.items())]
        table = "\n".join(lst_of_entrees)
        return "{header}\n{rule}\n{var_header}\n{table}".format(table=table,
                                                                header=header,
                                                                var_header = var_header,
                                                                rule="-" * len(header))

    def load_factor(self, string):
        pass

    def loadable_string(self):
        pass

class JPT(Factor):

    def __init__(self, variables, variable_domains):
        """Initializes a Joint Probability Table """
        super().__init__(variables, set(), variable_domains)

    def valid_table(self):
        """Checks if Probability Rules are followed """
        assert sum(self._table[entree] for entree in self.get_entrees()) == 1, "JPT does not sum to 1"
        return True if sum(self._table[entree] for entree in self.get_entrees()) == 1 else False

    def get_probability(self, **variables):
        """ Gets probability of certain variables. Includes summing out over vars """
        # variables = variables.items()
        if len(variables) == 1:
            var = tuple(list(variables.items())[0])
            to_sum = [entree for entree in self._entrees if var in entree]
            return sum(self._table[entree] for entree in to_sum)
        return super().get_probability(**variables)

class CPT(Factor):
    def __init__(self, unconditioned_var, conditioned_vars, variable_domains):
        super().__init__(list([unconditioned_var]), list([conditioned_vars]), variable_domains)

    @staticmethod
    def cpts2jpt(cpts):
        """Converts mutliple CPTs to JPT """
        all_domains = {}
        for cpt in cpts:
            all_domains.update(cpt.get_domains())
        jpt = JPT(all_domains.keys(), all_domains)
        jpt_entrees = jpt.get_entrees()
        for entree in jpt_entrees:
            prob = 1
            for cpt in cpts:
                for cpt_entree in cpt.get_entrees():
                    if set(cpt_entree) | set(entree) == set(entree): # Union of CPT and JPT does not yield new Vars
                        prob *= cpt.get_probability(**dict(cpt_entree)) #shouldnt have to put entree in dict
            jpt.set_probability(prob, **dict(entree))#shouldnt have to put entree in dict
        return jpt




def variable_elimination(elimination_var, factors):
    pass

# var1 = ["hi", "bye"]
# dom1 = {"hi": ["friend", "parent"], "bye": ["friend", "parent"]}
# var2 = ["mom", "dad"]
# dom2 = {"mom": ["scold", "praise"], "dad": ["scold", "praise"]}
#
# cpt1 = CPT("hi", "bye", dom1)
# entrees = cpt1.get_entrees()
# # print(entrees)
# for entree in entrees:
#     # print(dict(entree))
#     cpt1.set_probability(.2, **dict(entree))
#
# cpt2 = CPT("mom", "dad", dom2)
# entrees = cpt2.get_entrees()
# # print(entrees)
# for entree in entrees:
#     # print(dict(entree))
#     cpt2.set_probability(.1, **dict(entree))
#
# print(cpt1)
# print(cpt2)
#
# jpt = CPT.cpts2jpt([cpt1, cpt2])
# print(jpt)
<commit_msg>Pushing All Updates<commit_after># Probability Tables Implementation
# ---------------------------------
# @author <NAME>
# @author <NAME>

from itertools import product
from copy import deepcopy, copy
from warnings import warn

class Factor(object):
    """ Description:
        Usage:
            -
            -
            -
    """

    def __init__(self, unconditioned_vars, conditioned_vars, variable_domains):
        """ Creates a Factor object that encodes probabilities in a table """
        self._domains = variable_domains
        self._variables = set(conditioned_vars) | set(unconditioned_vars)
        self._unconditioned = unconditioned_vars
        self._conditioned = conditioned_vars
        entree_temp = [[(var, domain) for domain in self._domains[var]] for var in self._variables]
        self._entrees = list(sorted([element for element in product(*entree_temp)]))
        self._table = {}
        for entree in self._entrees:
            self._table[entree] = 0
        warn("Probabilities not set. Default = 0")

    def get_domains(self):
        return deepcopy(self._domains)

    def get_entrees(self):
        """ Retrieves all possible entrees to the table """
        return self._entrees

    def get_probability(self, **variables):
        """ Gets probability of a specific entree in table """
        variables = variables.items()
        if hasattr(variables, "__iter__"):
            to_sum = [entree for entree in self._entrees if set(entree) == set(variables)]
            if len(to_sum) != 1:
                raise UserWarning("Error in retrieving probability. Invalid method call.")
            return self._table[to_sum[0]]
        raise UserWarning("Error in retrieving probability. Invalid method call.")
        return 0

    def set_probability(self, val, **variables):
        """ Sets probability of a specific entree in table """
        assert set(variables) ^ set(self._variables) == set(), "Variables for assignment not Valid for JPT"
        variables = variables.items()
        entrees = [entree for entree in self._entrees if set(entree) | set(variables) == set(entree)]
        if len(entrees) > 1:
            print("ERROR IN SET PROBABILITY JPT")
        else:
            entree = entrees[0]
            self._table[entree] = val

    def __str__(self):
        """ Returns human readable Probability Table """
        length = len(max(self._variables, key = len))
        def str_beautify(s): return str(s).rjust(length + 2, " ").ljust(length + 3, " ")
        probability = "P({unc}{given}{cond})".format(unc=", ".join(self._unconditioned),
                                                     given=" | " if self._conditioned else "",
                                                     cond=", ".join(self._conditioned))
        var_header = str_beautify(" {vars}".format(vars=" | ".join(map(str_beautify,self._variables))))
        lst_of_entrees = [" | ".join(map(str_beautify, map(lambda x: x[1], k))) + " | " + str(v) for k, v in sorted(self._table.items())]
        table = " \n ".join(lst_of_entrees)
        border_size = (len(var_header) + len(probability) + 5)
        return "{border}\n{var_header} | {probability} \n{rule}\n {table} \n{border}\n\n".format(table=table,
                                                                    probability=probability,
                                                                    var_header = var_header,
                                                                    rule="-" * border_size,
                                                                    border = "*" * border_size
                                                                    )

    def __eq__(self, object2):
        return type(self) == type(object2) and \
               self._unconditioned == object2._unconditioned and \
               self._conditioned == object2._conditioned and \
               self._entrees == object2._entrees and \
               all([self._table[entree] == object2._table[entree] for entree in self._entrees])

    def load_factor(self, string):
        pass

    def loadable_string(self):
        pass

class JPT(Factor):

    def __init__(self, variables, variable_domains):
        """Initializes a Joint Probability Table """
        super().__init__(variables, set(), variable_domains)
        if not self.valid_table():
            warn("JPT is not valid!")

    def valid_table(self):
        """Checks if Probability Rules are followed """
        return True if sum(self._table[entree] for entree in self.get_entrees()) == 1 else False

    def get_probability(self, **variables):
        """ Gets probability of certain variables. Includes summing out over vars """
        # variables = variables.items()
        if len(variables) == 1:
            var = tuple(list(variables.items())[0])
            to_sum = [entree for entree in self._entrees if var in entree]
            return sum(self._table[entree] for entree in to_sum)
        return super().get_probability(**variables)

class CPT(Factor):
    def __init__(self, unconditioned_var, conditioned_vars, variable_domains):
        super().__init__(list([unconditioned_var]), list([conditioned_vars]), variable_domains)

    @staticmethod
    def cpts2jpt(cpts):
        """Converts mutliple CPTs to JPT """
        all_domains = {}
        for cpt in cpts:
            all_domains.update(cpt.get_domains())
        jpt = JPT(all_domains.keys(), all_domains)
        jpt_entrees = jpt.get_entrees()
        for entree in jpt_entrees:
            prob = 1
            for cpt in cpts:
                for cpt_entree in cpt.get_entrees():
                    if set(cpt_entree) | set(entree) == set(entree): # Union of CPT and JPT does not yield new Vars
                        prob *= cpt.get_probability(**dict(cpt_entree)) #shouldnt have to put entree in dict
            jpt.set_probability(prob, **dict(entree))#shouldnt have to put entree in dict
        return jpt

def variable_elimination(elimination_var, factors):
    pass

var1 = ["hi", "bye"]
dom1 = {"hi": ["friend", "parent"], "bye": ["friend", "parent"]}
var2 = ["mom", "dad"]
dom2 = {"mom": ["scold", "praise"], "dad": ["scold", "praise"]}

cpt1 = CPT("hi", "bye", dom1)
entrees = cpt1.get_entrees()
# print(entrees)
for entree in entrees:
    # print(dict(entree))
    cpt1.set_probability(.2, **dict(entree))


cpt2 = CPT("mom", "dad", dom2)
entrees = cpt2.get_entrees()
# print(entrees)
for entree in entrees:
    # print(dict(entree))
    cpt2.set_probability(.1, **dict(entree))

print(cpt1)
print(cpt2)

jpt = CPT.cpts2jpt([cpt1, cpt2])
print(jpt)

{'fraction_non_alphanumeric': 0.12640958914912073, 'fraction_numerical': 0.005914360066240833, 'mean_word_length': 2.288047705470573, 'pattern_counts': {'":': 8, '<': 7, '<?xml version=': 0, '>': 9, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 24, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1312472', 'n_tokens_mistral': 3767, 'n_tokens_neox': 3593, 'n_words': 961}