aaac_util.py

### utility for T5-aaac
import re
import ast
import logging
from string import Template
import random

import pyparsing as pp
import z3 

PREM_ARG_SCHEMES = {
 'modus ponens': 2,
 'chain rule': 2,
 'adjunction': 2,
 'case analysis': 3,
 'disjunctive syllogism': 2,
 'biconditional elimination': 1,
 'instantiation': 1,
 'hypothetical syllogism': 2,
 'generalized biconditional elimination': 1,
 'generalized adjunction': 2,
 'generalized dilemma': 3,
 'generalized disjunctive syllogism': 2
}


util_logger = logging.getLogger('transformer_tools.util.t5_util')



#######################################
#         Layouter class              #
#######################################

# Defines how to present AAAC raw data to model (as text)
class AAACLayouter:

    PRED_CHARS = "FGHIJKLMNOPQRSTUVWABCDE"
    ENT_CHARS = "abcdeklmnopqrstuwfgh"
    def substitutions():
        substitutions = {"F"+str(i+1):AAACLayouter.PRED_CHARS[i]+" " for i in range(20)}
        substitutions.update({"a"+str(i+1):AAACLayouter.ENT_CHARS[i] for i in range(20)})
        return substitutions
    

    MASK_STRING = "??"

    # defines how to present reason and conclusion statements to the model
    def format_statements_list(statements:list, mask_prob:float=0.0) -> str:
        if len(statements)==0:
            return "None"
        def ref_reco(sdict):
            r = "(%s)" % sdict['ref_reco'] if random.random()>mask_prob else AAACLayouter.MASK_STRING
            return r
        list_as_string = ["%s (ref: %s)" % (sdict['text'].lower(),ref_reco(sdict)) for sdict in statements]
        list_as_string = " | ".join(list_as_string)
        return list_as_string

    # defines how to present argdown premise and conclusion statements to the model
    def format_ad_statements_list(statements:list, mask_prob:float=0.0) -> str:
        if len(statements)==0:
            return "None"
        def ref_reco(sdict):
            r = "(%s)" % sdict['ref_reco'] if random.random()>mask_prob else AAACLayouter.MASK_STRING
            return r
        def explicit(sdict):
            r = str(sdict['explicit']) if random.random()>mask_prob else AAACLayouter.MASK_STRING
            return r
        list_as_string = ["%s (ref: %s explicit: %s)" % (sdict['text'].lower(),ref_reco(sdict),explicit(sdict)) for sdict in statements]
        list_as_string = " | ".join(list_as_string)
        return list_as_string

    # defines how to present formalizations to the model
    def format_formalizations_list(formalizations:list, mask_prob:float=0.0) -> str:
        if len(formalizations)==0:
            return "None"
        def ref_reco(sdict):
            r = "(%s)" % sdict['ref_reco'] if random.random()>mask_prob else AAACLayouter.MASK_STRING
            return r
        def fform(sdict):
            t = Template(sdict['form'])
            r = t.substitute(AAACLayouter.substitutions()) 
            r = r.replace("¬","not ")
            return r
        list_as_string = ["%s (ref: %s)" % (fform(sdict),ref_reco(sdict)) for sdict in formalizations]
        list_as_string = " | ".join(list_as_string)
        return list_as_string

    # defines how to present formalizations to the model
    def format_plcd_subs(plcd_subs:dict, mask_prob:float=0.0) -> str:
        if len(plcd_subs.keys())==0:
            return "None"
        def mask(s):
            return s if random.random()>mask_prob else AAACLayouter.MASK_STRING
        list_as_string = ["%s: %s" % (AAACLayouter.substitutions()[k],mask(v.lower())) for k,v in plcd_subs.items()]
        list_as_string = " | ".join(list_as_string)
        return list_as_string



    # defines how to present argdown-snippet to the model
    def format_argdown(argdown: str, mask_prob:float=0.0) -> str:
        # pattern = r"({.*uses: \[[\s\d,]*\]})" # matches yaml metadata inline blocks in inference patterns 
        pattern = r"--\nwith ([^{}]*)({[^{}]*})"
        matches = re.findall(pattern, argdown)
        for match in matches:
            m = match[1].replace('uses:','"uses":')
            m = m.replace('variant:','"variant":')
            d = ast.literal_eval(m)
            subst = ""
            mask_b = random.random()<mask_prob
            if mask_b:
                subst= "?? "           
            elif "variant" in d:
                subst = "(%s) " % ", ".join(d['variant'])
            subst = subst + "from " + " ".join(["(%s)" % i for i in d['uses']])
            if mask_b:
                argdown = argdown.replace(match[0]+match[1],subst) 
            else:
                argdown = argdown.replace(match[1],subst) 
        argdown = argdown.replace("\n"," ") # replace line breaks
        argdown = argdown.lower()
        return argdown

#######################################
#         Parser classes              #
#######################################

class AAACParser:
    def parse_proposition_block(ad_raw:str,inf_args:dict=None):
        if not ad_raw:
            return []
        ad_raw = ad_raw
        if ad_raw[0]!=" ":
            ad_raw = " "+ad_raw
        regex = r" \(([0-9]*)\) " # match labels
        proposition_list = []
        if not re.match(regex,ad_raw):
            return proposition_list
        matches = re.finditer(regex, ad_raw, re.MULTILINE) 
        label = -1
        pointer = -1
        for matchNum, match in enumerate(matches, start=1):        
            if label>-1:
                proposition = {
                    "text":ad_raw[pointer:match.start()].strip(),
                    "label":label,
                    "uses": [],
                    "scheme": "",
                    "variants":[]
                }
                proposition_list.append(proposition)
            label = int(match.group(1))
            pointer = match.end()
        if label>-1:
            proposition = {'text':ad_raw[pointer:].strip() ,'label':label,"uses": [],"scheme": "","variants":[]}
            proposition_list.append(proposition)
        if proposition_list and inf_args:
            proposition_list[0].update(inf_args)
        return proposition_list

    def parse_variants(variants_raw)->list:
        if not variants_raw:
            return []
        regex = r"(?! )[^\(\),]+"
        matches = re.finditer(regex, str(variants_raw), re.MULTILINE)
        return [match.group() for match in matches]
                
    def parse_uses(uses_raw)->list:
        if not uses_raw:
            return []
        regex = r"\(([0-9]+)\)"
        matches = re.finditer(regex, str(uses_raw), re.MULTILINE)
        return [int(match.group(1)) for match in matches]
                
    def preprocess_ad(ad_raw:str):
        ad_raw = ad_raw.replace("\n"," ")
        ad_raw = ad_raw.replace("  "," ")
        ad_raw = ad_raw.replace("with?? ","with ?? ")        
        return ad_raw

    def parse_argdown_block(ad_raw:str):
        ad_raw = AAACParser.preprocess_ad(ad_raw)
        regex = r"-- with ([^\(\)]*)( \([^-\(\))]*\))? from ([\(\), 0-9]+) --" # matches inference patterns
        proposition_list = []
        inf_args = None
        pointer = 0
        matches = re.finditer(regex, ad_raw, re.MULTILINE)
        for matchNum, match in enumerate(matches, start=1):   
            # parse all propositions before inference matched     
            propositions = AAACParser.parse_proposition_block(ad_raw[pointer:match.start()],inf_args=inf_args)
            if not propositions:
                return None
            proposition_list.extend(propositions)
            # update pointer and inf_args to be used for parsing next propositions block
            pointer = match.end()
            inf_args = {    
                'scheme': match.group(1),
                'variants': AAACParser.parse_variants(match.group(2)),
                'uses': AAACParser.parse_uses(match.group(3))
            }
        if pointer > 0:
            propositions = AAACParser.parse_proposition_block(ad_raw[pointer:],inf_args=inf_args)
            proposition_list.extend(propositions)
        return proposition_list

    def parse_statements(statements_raw:str):
        if not statements_raw:
            return None
        statements = []
        if statements_raw.strip()=="None":
            return statements
        list_raw = statements_raw.split(" | ")
        regex = r" \(ref: (?:\(([0-9]+)\)|\?\?)\)$"
        for s in list_raw:
            match = re.search(regex, s)
            if not match:
                return None
            item = {
                'text':s[:match.start()],
                'ref_reco':int(match.group(1)) if match.group(1) else match.group(1)
            }
            statements.append(item)
        return statements

    def parse_formalizations(forms_raw:str):
        parsed = AAACParser.parse_statements(forms_raw)
        if not parsed:
            return None
        formalizations = []
        for d in parsed:
            d['form'] = d.pop('text')
            formalizations.append(d)

        # post-process: cleanup "⁇"
        for f in formalizations:
            form = f['form']
            form = form.replace("⁇","") 
            form = form.replace("  "," ")
            form = form.strip()
            f['form'] = form

        return formalizations

    def parse_plcd_subs(subs_raw:str):
        if not subs_raw:
            return None
        plcd_subs = []
        if subs_raw.strip()=="None":
            return plcd_subs
        list_raw = subs_raw.split(" | ")
        regex = r"^(..?):\s(.+)"
        for s in list_raw:
            match = re.search(regex, s)
            if not match:
                return None
            k = match.group(1)
            # comment out reverse substitution
            #if k in AAACLayouter.PRED_CHARS:
            #    k = 'F'+str(1+AAACLayouter.PRED_CHARS.index(k))
            #if k in AAACLayouter.ENT_CHARS:
            #    k = 'a'+str(1+AAACLayouter.ENT_CHARS.index(k))
            item = {k: match.group(2)}
            plcd_subs.append(item)
        return plcd_subs




#######################################
#       Logic Evaluator Class         #
#######################################

class AAACLogicEvaluator():

    def __init__(
            self, 
            nl_schemes = None,
            domains = None,
            **kwargs
        ):

            self.nl_schemes_re = []
            if nl_schemes:
                self.nl_schemes_re = nl_schemes.copy()
                for item in self.nl_schemes_re:
                    item['scheme'] = [self.construct_regex(s) for s in item['scheme']]

            # construct de-paraphrasing rules from domain-config-file
            self.de_paraphrasing_rules = {}
            if domains:
                for domain in domains.get('domains'):
                    rules = {}
                    for k,v in domain.get("paraphrases",{}).items():
                        rules.update({repl.lower():k.lower() for repl in v}) # all paraphrasing rules are cast as lower case
                    self.de_paraphrasing_rules[domain['id']] = rules


    def construct_regex(self,statement:str):
        regex = r"( a )?\$\{([A-Za-z])\}"
        regex_template = ""
        pointer = 0
        matches = re.finditer(regex, statement, re.MULTILINE)
        for matchNum, match in enumerate(matches):
            regex_template += statement[pointer:match.start()].lower()
            neg_la = statement[match.end():match.end()+5] if match.end()+5<=len(statement) else statement[match.end():]
            if match.group(1):
                regex_template += " an? "  
            regex_template += "(?P<%s%s>.*(?!%s)*)"%(match.group(2),matchNum,neg_la)
            pointer = match.end()
        regex_template += statement[pointer:].lower()
        return regex_template


    def parse_inference_as_scheme(
        self,
        argument:list = None,
        nl_scheme_re:dict = None
    ):
        # recursively try to match premises to scheme, i.e. recursively construct a consistent mapping
        # mapping maps sentences in 
        # matching contains all matches found so far
        def match_premises(matching:list=None, mapping:dict=None):
            #print("match_premises:" + str(mapping))
            unmapped_formulas = [i for i in range(len(argument)) if not i in mapping.keys()]
            unmapped_premises = [j for j in range(len(argument)) if not j in mapping.values()]
            for i in unmapped_formulas:
                for j in unmapped_premises:
                    try:
                        match = re.match(nl_scheme_re['scheme'][i], argument[j])
                    except IndexError: 
                        match=False
                    if match:
                        matching[i]=match
                        mapping[i]=j
                        if any(m==None for m in matching):
                            full_match = match_premises(
                                matching = matching,
                                mapping = mapping
                            )
                        else:
                            full_match = matching_consistent(matching)
                        if full_match:
                            return True
                    else:
                        full_match = False

            return full_match

        # check whether a mapping is consistent with respect to placeholders
        def matching_consistent(matching:list):
            if any(m==None for m in matching): 
                return False
            def group_by_name(match=None, group_name=None):
                try:
                    g=match.group(group_name)
                except IndexError: 
                    g=None
                return g
            all_plcds = (nl_scheme_re["predicate-placeholders"]+nl_scheme_re["entity-placeholders"])
            for plcd in all_plcds:
                all_subst = []
                for i in range(10):
                    group_name = plcd+str(i)
                    subst = [group_by_name(match=match, group_name=group_name) for match in matching]
                    subst = [x for x in subst if x != None]
                    all_subst.extend(subst)
                if len(set(all_subst))>1:
                    return False
            return True

        c_match = re.match(nl_scheme_re['scheme'][-1], argument[-1])
        if c_match: 
            try:
                g=c_match.group("F")
            except IndexError: 
                g="None"

        if c_match:
            full_match = match_premises(
                matching = [None]*(len(argument)-1)+[c_match],
                mapping = {len(argument)-1:len(argument)-1}
            )
        else:
            full_match = False

        return full_match


    def parse_inference_as_base_scheme(
            self,
            argument:list = None,
            base_scheme_group:str = None,
            domain:str = None
        ):
            variant = None
            matches = False
            # make the entire argument lower case
            argument = [item.lower() for item in argument]

            argument = self.de_paraphrase(argument, domain=domain) if domain else argument
            for nl_scheme_re in self.nl_schemes_re:
                if nl_scheme_re['base_scheme_group'] == base_scheme_group:
                    matches = self.parse_inference_as_scheme(
                        argument = argument,
                        nl_scheme_re = nl_scheme_re
                    )
                    if matches:
                        variant = nl_scheme_re['scheme_variant']
                        break

            return matches, variant


    def de_paraphrase(self, argument:list = None, domain:str = None):
        rules = {}
        if domain in self.de_paraphrasing_rules:
            rules = self.de_paraphrasing_rules[domain]
        for i,statement in enumerate(argument):    
            s = statement
            for k,v in rules.items():
                s = s.replace(k,v)
            argument[i] = s
        return argument


    def parse_string_formula(self, formula:str):
        atom = pp.Regex("[A-Z]\s[a-u|w-z]").setName("atom")
        expr = pp.infixNotation(atom,[
                                    ("not", 1, pp.opAssoc.RIGHT, ),
                                    ("&", 2, pp.opAssoc.LEFT, ),
                                    ("v", 2, pp.opAssoc.LEFT, ),
                                    ("->", 2, pp.opAssoc.LEFT, ),
                                    ("<->", 2, pp.opAssoc.LEFT, )
                                    ])
        try: 
            parsed = expr.parseString(formula,parseAll=True)[0]
        except pp.ParseException as e:
            parsed = None
        return parsed


    def c_bf(self,parse_tree):
        if not parse_tree:
            return None
        functions = {}
        constants = {}
        Object = z3.DeclareSort('Object')
        bin_op = {
            "&": z3.And,
            "v": z3.Or,
            "->": z3.Implies
        }
        pt = parse_tree
        if pt[0]=="not":
            return z3.Not(self.c_bf(pt[1]))
        if pt[1]=="<->":
            return z3.And(z3.Implies(self.c_bf(pt[0]),self.c_bf(pt[2])),z3.Implies(self.c_bf(pt[2]),self.c_bf(pt[0])))
        if pt[1] in bin_op.keys():
            return bin_op[pt[1]](self.c_bf(pt[0]),self.c_bf(pt[2]))
        # atom
        pred = parse_tree[0]
        if not pred in functions.keys():     # add predicate to dict if necessary
            functions[pred] = z3.Function(pred, Object, z3.BoolSort())
        const = parse_tree[-1]
        if not const in constants.keys():    # add function to dict if necessary
            functions[const] = z3.Const(const, Object)
        return functions[pred](functions[const]) 

    def to_z3(self,str_f:str):
        if not str_f:
            return None
        f = str_f.strip()
        if f[:4] == "(x):":
            Object = z3.DeclareSort('Object')
            x = z3.Const('x', Object)
            parsed = self.parse_string_formula(f[4:].strip())
            if parsed:
                return (z3.ForAll(x,self.c_bf(parsed)))
            return None
        parsed = self.parse_string_formula(f)
        if parsed:
            return self.c_bf(parsed)
        return None

    def check_deductive_validity(self,scheme:list):
        premises = [self.to_z3(f) for f in scheme[:-1]] # premises
        conclusion = self.to_z3(scheme[-1]) # conclusion
        #print(theory)
        if any(p==None for p in premises) or (conclusion==None):
            return None
        theory = premises # premises
        theory.append(z3.Not(conclusion)) # negation of conclusion
        s = z3.Solver()
        s.add(theory)
        #print(s.check())
        valid = s.check()==z3.unsat
        return valid


#######################################
#         Main eval function          #
#######################################


def ad_valid_syntax(argdown):
    check = False
    if argdown:
        # consecutive labeling?
        check = all(p['label']==i+1 for i,p in enumerate(argdown))
        # no "--" as statement
        check = check and all(item['text']!="--" for item in argdown)
    return 1 if check else 0

def ad_last_st_concl(argdown):
    check = len(argdown[-1]['uses'])>0 if argdown else False
    return 1 if check else 0

# do all statements referenced in inference exist and do they occur before inference is drawn?
def used_prem_exist(argdown):
    if not argdown: return None
    check = True
    previous_labels = []
    for p in argdown:
        if p['uses']:
            exist = all((l in previous_labels) for l in p['uses'])
            check = check and exist
        if not check: return 0
        previous_labels.append(p['label'])
    return 1 if check else 0

def ad_redundant_prem(argdown):
    if not argdown: return None
    premises = [item['text'].strip() for item in argdown if not item['uses']] 
    check = len(premises)!=len(set(premises))
    return 1 if check else 0

def ad_petitio(argdown):
    if not argdown: return None
    premises = [item['text'].strip() for item in argdown if not item['uses']] 
    conclusions = [item['text'].strip() for item in argdown if item['uses']] 
    check = any(p==c for p in premises for c in conclusions)
    return 1 if check else 0

def prem_non_used(argdown):
    if not argdown: return None
    used_total = [l for p in argdown if p['uses'] for l in p['uses']]
    non_used = [p['label'] for p in argdown if not p['label'] in used_total]
    return len(non_used)-1



#######################################
#    Evaluating reason_statements     #
#######################################

def s_valid_syntax(output:list,raw:str=""):
    return 1 if output!=None or raw.strip()=="None" else 0

def s_not_verb_quotes(output:list,source):
    l = [s for s in output if not s['text'] in source]
    return len(l)

def s_ord_me_subsseq(output:list,source):
    source = source
    check = True
    for reason in output:
        text = reason['text']
        check = check and (text in source)
        if not check: return check
        source = source.replace(text,"",1)
    return check



#######################################
#    Evaluating r-c-a consistency     #
#######################################

# test: no reason statements is contained in a conclusion statement and vice versa
def reason_concl_mutually_exclusive(reasons,concl):
    check = True
    for r in reasons:
        for c in concl:
            check = bool(not c['text'] in r['text'] and not r['text'] in c['text'])
            if not check: return check
    return check