evaluate.py

Go to the documentation of this file.

 
 
 
 
from abc import abstractmethod
import re
import allogate as logging
import statistics
import os
from copy import copy
import glob
import re
 
 
 
class Evaluator:
    """!
    @brief Abstract class to evaluate and score integrity/validty
    @belongsto Python::Markopy::Evaluation
    """
    def __init__(self, filename: str) -> None:
        """! 
        @brief default constructor for evaluator
        @param filename filename to evaluate. Can be a pattern
        """
        self.filenamefilename = filename
        self.checkschecks = []
        self.TEST_PASS_SYMBOLTEST_PASS_SYMBOL = b"\xe2\x9c\x85".decode()
        self.TEST_FAIL_SYMBOLTEST_FAIL_SYMBOL = b"\xe2\x9d\x8c".decode()
        self.all_checks_passedall_checks_passed = True
        self.filesfiles = []
        if("*" in filename):
            self.filesfiles = glob.glob(filename)
        else:
            self.filesfiles.append(filename)
        return True
 
    def evaluate(self) -> bool:
        "! @brief base evaluation function"
        for file in self.filesfiles:
            self._evaluate_evaluate(file)
        
        self.check_funcscheck_funcs = [func for func in dir(self) if (callable(getattr(self, func)) and func.startswith("check_"))]
 
    @abstractmethod
    def _evaluate(self, file) -> list:
        """! 
        @brief internal evaluation function for a single file
        @param file filename to evaluate
        """
        if(not os.path.isfile(file)):
            logging.pprint(f"Given file {file} is not a valid filename")
            return False
        else:
             return open(file, "rb").read().split(b"\n")
        
        
 
    def success(self, checkname):
        """!
        @brief pass a test
        @param checkname text to display with the check
        """
        self.checkschecks.append((checkname, self.TEST_PASS_SYMBOLTEST_PASS_SYMBOL))
 
    def fail(self, checkname):
        """!
        @brief fail a test
        @param checkname text to display with the check
        """
 
        self.all_checks_passedall_checks_passed = False
        self.checkschecks.append((checkname, self.TEST_FAIL_SYMBOLTEST_FAIL_SYMBOL))
 
    def finalize(self):
        "! @brief finalize an evaluation and print checks"
        print("\n################ Checks ################ ")
        for test in self.checkschecks:
            logging.pprint(f"{test[0]:30}:{test[1]} ")
        print("\n")
        self.checkschecks = []
        return self.all_checks_passedall_checks_passed
 
class ModelEvaluator(Evaluator):
    """!
    @brief evaluate a model
    @belongsto Python::Markopy::Evaluation
    @extends Python::Markopy::Evaluation::Evaluator
    """
    def __init__(self, filename: str) -> None:
        "! @brief default constructor"
        valid = super().__init__(filename)
        
        if not valid:
            return False
 
    def evaluate(self):
        "! @brief evaluate a model"
        logging.VERBOSITY=2
        logging.SHOW_STACK_THRESHOLD=3
        super().evaluate()
        for file in self.filesfiles:
            logging.pprint(f"Model: {file.split('/')[-1]}: ",2)
            edges = super()._evaluate(file)
            if not edges:
                continue
            self.lnodeslnodes = {}
            self.rnodesrnodes = {}
            self.ewsews = []
            self.edge_countedge_count = len(edges)
            for edge in edges:
                if(edge ==b''):
                    self.edge_countedge_count-=1
                    continue
                try:
                    e = edge.split(b',')
                    self.ewsews.append(int(edge[2:-2:1]))
                    if(e[0] not in self.lnodeslnodes):
                        self.lnodeslnodes[e[0]]=1
                    else:
                        self.lnodeslnodes[e[0]]+=1
                    if(e[-1] not in self.rnodesrnodes):
                        self.rnodesrnodes[e[-1]]=1
                    else:
                        self.rnodesrnodes[e[-1]]+=1
                except Exception as e:
                    print(e)
                    logging.pprint(f"Model file is corrupted.", 0)
                    continue
        
            self.lnode_countlnode_count = len(self.lnodeslnodes)
            self.rnode_countrnode_count = len(self.rnodesrnodes)
            logging.pprint(f"total edges: {self.edge_count}", 1)
            logging.pprint(f"unique left nodes: {self.lnode_count}", 1)
            logging.pprint(f"unique right nodes: {self.rnode_count}", 1)
 
            for check in self.check_funcscheck_funcs:
                try:
                    self.__getattribute__(check)()
                except Exception as e:
                    print(e)
                    self.failfail(f"Exceptionn in {check}")
            self.finalizefinalize()
        
    def check_dangling(self):
        "! @brief check if model has dangling nodes"
        if(self.lnode_countlnode_count == self.rnode_countrnode_count):
            self.successsuccess("No dangling nodes")
        else:
            logging.pprint(f"Dangling nodes found, lnodes and rnodes do not match", 0)
            self.failfail("No dangling nodes")
    
    def check_structure(self):
        "! @brief check model structure for validity"
        if((self.lnode_countlnode_count-1) * (self.rnode_countrnode_count-1) + 2*(self.lnode_countlnode_count-1)):  
            self.successsuccess("Model structure")
        else:
            logging.pprint(f"Model did not satisfy structural integrity check (lnode_count-1) * (rnode_count-1) + 2*(lnode_count-1)", 0)
            self.failfail("Model structure")
 
    def check_weight_deviation(self):
        "! @brief check model standart deviation between edge weights"
        mean = sum(self.ewsews) / len(self.ewsews)
        variance = sum([((x - mean) ** 2) for x in self.ewsews]) / len(self.ewsews)
        res = variance ** 0.5
        self.stdevstdev = res
        if(res==0):
            logging.pprint(f"Model seems to be untrained", 0)
            self.failfail("Model has any training")
        else:
            self.successsuccess("Model has any training")
        if(res<3000):
            logging.pprint(f"Model is not adequately trained. Might result in inadequate results", 1)
            self.failfail("Model has training")
            self.failfail(f"Model training score: {round(self.stdev,2)}")
        else:
            self.successsuccess("Model has training")
            self.successsuccess(f"Model training score: {round(self.stdev)}")
 
    def check_min(self):
        "! @brief check 0 edge weights distribution"
        count = 0
        for ew in self.ewsews:
            if ew==0:
                count+=1
        if(count > self.rnode_countrnode_count*0.8):
            self.failfail("Too many 0 edges")
            logging.pprint(f"0 weighted edges are dangerous and may halt the model.", 0)
        else:
            self.successsuccess("0 edges below threshold")
    
    def check_min_10percent(self):
        "! @brief check minimum 10% of the edges"
        sample = self.ewsews[int(self.edge_countedge_count*0.1)]
        #print(f"10per: {sample}")
        avg = sum(self.ewsews) / len(self.ewsews)
        #print(f"avg: {avg}")
        med = statistics.median(self.ewsews)
        #print(f"med: {med}")
 
    def check_lean(self):
        "! @brief check which way model is leaning. Left, or right"
        sample = self.ewsews[int(self.edge_countedge_count*0.1)]
        avg = sum(self.ewsews) / len(self.ewsews)
        med = statistics.median(self.ewsews)
        
        if(med*10<sample):
            logging.pprint("Median is too left leaning and might indicate high entropy")
            self.failfail("Median too left leaning")
        else:
            self.successsuccess("Median in expected ratio")
        pass
 
        if(sample*5>avg):
            logging.pprint("Least probable 10% too close to average, might indicate inadequate training")
            self.failfail("Bad bottom 10%")
        else:
            self.successsuccess("Good bottom 10%")
        pass
        
 
    def check_distrib(self):
        "! @deprecated"
        sorted_ews = copy(self.ewsews)
        sorted_ews.sort(reverse=True)
        ratio1 = sorted_ews[0]/sorted_ews[int(self.edge_countedge_count/2)]
        ratio2 = sorted_ews[int(self.edge_countedge_count/2)]/sorted_ews[int(self.edge_countedge_count*0.1)]
        #print(ratio1)
        #print(ratio2)
 
 
class CorpusEvaluator(Evaluator):
    """!
    @brief evaluate a corpus
    @belongsto Python::Markopy::Evaluation
    @extends Python::Markopy::Evaluation::Evaluator
    """
    def __init__(self, filename: str) -> None:
        """! 
        @brief default constructor
        @param filename filename or pattern to check
        """
        valid = super().__init__(filename)
        if not valid:
            return False
 
    def evaluate(self):
        "! @brief evalute a corpus. Might take a long time"
        logging.pprint("WARNING: This takes a while with larger corpus files", 2)
        logging.VERBOSITY=2
        logging.SHOW_STACK_THRESHOLD=3
        super().evaluate()
        for file in self.filesfiles:
 
            delimiter = ''
            sum=0
            max=0
            total_chars = 0
            lines_count = 0
            bDelimiterConflict=False
            logging.pprint(f"Corpus: {file.split('/')[-1]}: ",2)
            with open(file, "rb") as corpus:
                for line in corpus:
                    lines_count+=1
                    match = re.match(r"([0-9]+)(.)(.*)\n", line.decode()).groups()
                    if(delimiter and delimiter!=match[1]):
                        bDelimiterConflict = True
                        
                    elif(not delimiter):
                        delimiter = match[1]
                        logging.pprint(f"Delimiter is: {delimiter.encode()}")
                    sum +=int(match[0])
                    total_chars += len(match[2])
                    if(int(match[0])>max):
                        max=int(match[0])
 
                if(bDelimiterConflict):
                    self.failfail("Incorrect delimiter found")
                else:
                    self.successsuccess("No structural conflicts")
 
                logging.pprint(f"Total number of lines: {lines_count}")
                logging.pprint(f"Sum of all string weights: {sum}")
                logging.pprint(f"Character total: {total_chars}")
                logging.pprint(f"Average length: {total_chars/lines_count}")
                logging.pprint(f"Average weight: {sum/lines_count}")
 
            self.finalizefinalize()
    def _evaluate(self, file) -> list:
        """!
        @brief evaluate a single file. Remove reading file because it should be read line by line.
        @param file corpus filename to evaluate
        """
        if(not os.path.isfile(file)):
            logging.pprint(f"Given file {file} is not a valid filename")
            return False
        else:
            return True