<jupyter_start><jupyter_text>LTL-f BASED-TRACE ALIGNMENT<jupyter_code>import re
from typing import Match, cast
import copy

from ltlf2dfa.parser.ltlf import LTLfParser
from dataclasses import dataclass
from typing import Dict, Match, Set, Tuple, cast<jupyter_output><empty_output><jupyter_text>Constraint automata<jupyter_code>parser = LTLfParser()<jupyter_output><empty_output><jupyter_text>Select constraints as LTL-f/Declare formulas     constraint_formulas = {"existence":"F(a)",                       "absence2":"!(F((b & X(F(b)))))",                        "response":"G(k -> F(l))",                        "precedence":"((!(j) U i) | G(!(j)))",                       "chain_response":"G((n -> X(o)))",                        "chain_precedence": "(!(q) & G((X(q) -> p)))",                       "responded_existence":"(F(e) -> F(f))",                        "not_coexistence":"!((F(s) & F(t)))",                        "not_succession":"G((t -> !(F(u))))",                        "not_chain_succession":"G((v  !(X(w))))",                       "formula_inventata":"(F(a) & F(b)) -> G(c)",                                              "alternative_not_chain_succession": "G((!v & !w)|((v -> X(!(w))) & (w -> X(!(v)))))"}<jupyter_code>constraints = ["(!(q) & G((X(q) -> p)))", 
                "F(a)"]

constraints_10 = ["(!(q) & G((X(q) -> p)))", 
                    "F(a)", 
                    "((!(j) U i) | G(!(j)))", 
                    "(F(e) -> F(f))", 
                    "G((n -> X(o)))",
                    "!((F(s) & F(t)))",
                    "G((t -> !(F(u))))",
                    "!G((v <-> (X(w))))",
                    "G(k -> F(l))",
                    "!(F((b & X(F(b)))))"]

constraints_15 = ["(!(q) & G((X(q) -> p)))", 
                    "F(a)", 
                    "((!(j) U i) | G(!(j)))", 
                    "(F(e) -> F(f))", 
                    "G((n -> X(o)))",
                    "!((F(s) & F(t)))",
                    "G((t -> !(F(u))))",
                    "!G((v <-> (X(w))))",
                    "G(k -> F(l))",
                    "!(F((b & X(F(b)))))",
                    "(!(r) & G((X(r) -> q)))",
                    "G(h -> F(i))",
                    "F(j)",
                    "F(c)",
                    "G((o -> X(p)))"]
  
constraints_20 = ["(!(q) & G((X(q) -> p)))", 
                    "F(a)", 
                    "((!(j) U i) | G(!(j)))", 
                    "(F(e) -> F(f))", 
                    "G((n -> X(o)))",
                    "!((F(s) & F(t)))",
                    "G((t -> !(F(u))))",
                    "!G((v <-> (X(w))))",
                    "G(k -> F(l))",
                    "!(F((b & X(F(b)))))",
                    "(!(r) & G((X(r) -> q)))",
                    "G(h -> F(i))",
                    "F(j)",
                    "F(c)",
                    "G((o -> X(p)))",
                    "((!(y) U x) | G(!(y)))",
                    "F(x)",
                    "G(c -> F(d))",
                    "!(F((g & X(F(g)))))",
                    "((!(m) U l) | G(!(m)))"]    

constraint_formulas = constraints<jupyter_output><empty_output><jupyter_text>Parse DFA with regular expressions<jupyter_code>class StructureOutput():
    def __init__(self, nb_states, variable_names, initial_state, accepting_states, rejecting_states, transitions): 
        self.nb_states = nb_states
        self.variable_names = variable_names
        self.initial_state = initial_state
        self.accepting_states = accepting_states
        self.rejecting_states = rejecting_states
        self.transitions = transitions


class StructureOutputRegex():

    def __init__(self, output):
        self.output = output

    def total_function (self):
        def create_transitions(self) -> Dict[int, Dict[int, Set[str]]]:
            raw_transitions: Dict[int, Dict[int, Set[str]]] = {}
            lines = self.output.splitlines()
            # from the 8th line, the output specifies the transitions.
            transition_strings = lines[7:]
            for t in transition_strings:
                match = cast(
                    Match, re.search("State ([0-9]+): ([01X]+|) -> state ([0-9]+)", t)
                )
                if match is None:
                    continue
                start_state = int(match.group(1))
                guard = match.group(2)
                end_state = int(match.group(3))
                raw_transitions.setdefault(start_state, {}).setdefault(
                    end_state, set()
                ).add(guard)
            return raw_transitions

        variable_names : Tuple[str, ...] = tuple(cast(Match, re.search("DFA for formula with free variables: (.*)", self.output),).group(1).split())
        initial_state : int = int(cast(Match, re.search("Initial state: (.*)\n", self.output)).group(1))
        accepting_states : Set[int] = set(map(int,cast(Match, re.search("Accepting states: (.*)\n", self.output)).group(1).split(),))
        rejecting_states : Set[int] = set(map(int,cast(Match, re.search("Rejecting states: (.*)\n", self.output)).group(1).split(),))
        nb_states : int = int(cast(Match,re.search(r"Automaton has ([0-9]+) state(\(?s\)?)? and .* BDD-node(\(?s\)?)?",self.output,),).group(1))
        raw_transitions : Dict[int, Dict[int, Set[str]]] = create_transitions(self)
        

        return StructureOutput(nb_states, variable_names, initial_state, accepting_states, rejecting_states, raw_transitions)<jupyter_output><empty_output><jupyter_text>Build vocabulary of constraint automata<jupyter_code>index = 1

all_automata = {}

for f in constraint_formulas:
    build_automaton = {}
    build_automaton["formula"] = f

    all_states_constr = []
    final_states_constr = []
    init_states_constr = []
    rejecting_states_constr = []
    rho_constr_basic = []
    rho_tobe_negated = []
    rho_tobe_positive = []

    ## Parser + dfa #####################################################
    formula = parser(f)       # returns an LTLfFormula
    dfa = formula.to_dfa(mona_dfa_out=True)
    mona_output = StructureOutputRegex(output=dfa).total_function() 
   
    ## save all what is needed ##########################################
    init_states_constr = 's1'
    
    states = set.union(mona_output.rejecting_states, mona_output.accepting_states)
    states.remove(0)
    states = list(states)
    for elem in states:
        all_states_constr.append('s'+str(elem))

    for elem in list(mona_output.accepting_states):
        final_states_constr.append('s'+str(elem))

    alphabet = mona_output.variable_names
    alphabet = [x.lower() for x in alphabet]
    print (alphabet)

    for key,elem in mona_output.transitions.items():
        s_start = key
        if(s_start != 0):
            for k,e in elem.items():
                s_end = k
                sum = 0
                if(s_end != s_start):
                    comb = list(e)[0]
                    for char in comb:
                        if(char!='X'):
                            sum += int(char)
                    if(sum > 1):
                        continue
                    elif(sum == 1):
                        index1 = comb.find('1')
                        rho_constr_basic.append("s"+str(s_start)+" "+alphabet[index1]+" s"+str(s_end))
                    elif(sum == 0):
                        indeces0 = [i for i in range(len(comb)) if comb[i] in '0']
                        indeces0 = [i for i in range(len(comb)) if comb[i] in '0']
                        if (len(indeces0) == 0):
                            indeces0 = [i for i in range(len(comb)) if comb[i] in 'X']
                            res_list = [alphabet[i] for i in indeces0]
                            rho_tobe_positive.append([s_start,s_end,res_list])
                        else:
                            res_list = [alphabet[i] for i in indeces0]
                            rho_tobe_negated.append([s_start,s_end,res_list])


    build_automaton["all_states"] = all_states_constr
    build_automaton["final_states"] = final_states_constr
    build_automaton["init_state"] = init_states_constr
    build_automaton["transitions"] = rho_constr_basic
    build_automaton["symbols_constr"] = alphabet
    build_automaton["negated_transitions"] = rho_tobe_negated
    build_automaton["positive_transitions"] = rho_tobe_positive

    all_automata["a"+str(index)] = build_automaton

    index += 1<jupyter_output>['q', 'p']
['a']<jupyter_text>Trace automata  Load xes files<jupyter_code>log_paths = {}
base_opt = "try_opt"
syn_real = "synthetic"
n_constr = ["10"]#, "15", "20"]
choice = "2"
constr_inverted = ["3", "4", "6"]
len_traces = ["1-50", "51-100", "101-150", "151-200"]

for i in range(len(n_constr)):
    for j in range(len(constr_inverted)):
        for k in range(len(len_traces)):
            key = choice+"/"+constr_inverted[j]+"/"+len_traces[k]
            log_paths[key] = "dataset/logs/"+syn_real+"-logs/"+n_constr[i]+"constraints/"+constr_inverted[j]+" constraints inverted/log-from-"+n_constr[i]+"constr-model-"+constr_inverted[j]+"constr_inverted-"+len_traces[k]+".xes"<jupyter_output><empty_output><jupyter_text>From xes files to traces as strings<jupyter_code>alphabet = "ABCDEFGHIJKLMNOPQRSTUVWXYZ".lower()
voc = {}
for i in range(len(alphabet)):
    voc[i+1] = alphabet[i]

def convertNumberToChar (val):
    return voc[val]
def readLog (log_path):
    '''
        flag:           True when we have to read an event from the file
        trace:          List of events as integers
        traceChar:      List of events as strings
        traceString:    String of events
        log:            List of all traces (as strings)
    '''

    flag = False
    trace = []
    traceChar = []
    traceString = ""
    log = []

    f = open  (log_path)
    f1 = f.readlines()
    
    for x in f1:
        # if there is an event, we activate the flag
        if (x.__contains__("<event>")):
            flag = True

        # If flag is True and the name of the event is present,
        # we extract the event name and append it to "trace"
        if (flag and x.__contains__('<string key="concept:name"')):
            val = x.split('value="activity ',1)[1]
            val = val.split('"')[0]                
            trace.append(val)
            flag = False

        # When the trace is terminated, we can convert the numbers in strings
        if (x.__contains__("</trace>")):
            for event in trace:
                traceChar.append(convertNumberToChar(int(event)))
            traceString = "".join(traceChar)
            log.append(traceString)

            # Reset variables
            trace = []
            traceChar = []
            traceString = ""
    
    return log<jupyter_output><empty_output><jupyter_text>Build vocabulary of trace automata Build traces automata + add positive transitions for the negated ones<jupyter_code>all_traces_log = copy.deepcopy(log_paths)
all_automata_log = {}

for k, log_path in log_paths.items():
    traces_log = readLog(log_path) # list of traces

    all_traces = {}
    variabili = {}

    idx_trace = 1

    for t in traces_log:
        trace = {}
        variabile = copy.deepcopy(all_automata)
        
        symbols_trace = list(set(t))
        
        ## Build trace automaton ##############################################
        rho_trace_basic = []
        Q_trace = []
        for i in range(len(t)):
            Q_trace.append('t'+str(i))
            rho_trace_basic.append('t'+str(i)+" "+t[i] +" "+ 't'+str(i+1))

        Q_trace.append('t'+str(len(t)))
        init_state_trace = Q_trace[0]
        final_state_trace = Q_trace[-1]

        trace["symbols_trace"] = symbols_trace
        trace["init_state_trace"] = init_state_trace
        trace["final_state_trace"] = final_state_trace
        trace["Q_trace"] = Q_trace
        trace["rho_trace_basic"] = rho_trace_basic

        all_traces[idx_trace] = trace

        ## Now we add the transitions for the negated symbols ###################

        for ID,automaton in all_automata.items():
            symb_constr = set(variabile[ID]["symbols_constr"])
            symb_trace = set(symbols_trace)
            all_symbs = list(set.union(symb_constr, symb_trace))

            trans = copy.deepcopy(automaton["transitions"])
            for t_neg in variabile[ID]["negated_transitions"]:
                symbs = copy.deepcopy(all_symbs)
                for t in t_neg[2]:
                    symbs.remove(t)
                for elem in symbs:
                    if ("s"+str(t_neg[0])+" "+elem+" s"+str(t_neg[1]) not in trans):
                        trans.append("s"+str(t_neg[0])+" "+elem+" s"+str(t_neg[1]))

            for t_pos in variabile[ID]["positive_transitions"]:
                    symbs = copy.deepcopy(all_symbs)
                    for elem in symbs:
                        if ("s"+str(t_pos[0])+" "+elem+" s"+str(t_pos[1]) not in trans):
                            trans.append("s"+str(t_pos[0])+" "+elem+" s"+str(t_pos[1]))

            variabile[ID]["transitions"] = trans
        
        variabili[idx_trace] = variabile

        idx_trace += 1

    all_automata_log[k] = variabili
    
    all_traces_log[k] = all_traces<jupyter_output><empty_output><jupyter_text>PDDL<jupyter_code>for key,all_traces in all_traces_log.items():
    n_cons, inv, lenght = key.split('/')
    type_db = n_cons+"_"+inv+"_"+lenght
    
    for id_trace, trace in all_traces.items():
        all_automata = copy.deepcopy(all_automata_log[key][id_trace])

        M = 0
        for ID,a in all_automata.items():
            i = len(a['all_states'])
            if (i >= M):
                M = i
                index_max = ID
        all_states_constr_max = all_automata[index_max]['all_states'] 


        symbols_tot = []
        for ID,a in all_automata.items():
            for s in a["symbols_constr"]:
                symbols_tot.append(s)
        symbols_tot = list(set(symbols_tot).union(set(trace["symbols_trace"])))

        # DOMAIN ###########################################################################
        domain_name = "domain_"+base_opt+"_"+type_db+"_trace"+str(id_trace)
        pddl_domain_initial = "(define (domain "+domain_name+") "\
                        "(:requirements :strips :typing :action-costs) "\
                        "(:types trace_state automaton_state - state activity automaton_name) "

        pddl_domain_predicates = "(:predicates (trace ?t1 - trace_state "\
                                    "?e - activity "\
                                    "?t2 - trace_state) "\
                                    "(automaton ?a1 - automaton_name "\
                                    "?s1 - automaton_state "\
                                    "?e - activity "\
                                    "?s2 - automaton_state) "\
                                    "(cur_state ?a1 - automaton_name ?s - state) "\
                                    "(final_state ?a1 - automaton_name ?s - state)) "\
                                    "(:functions (total-cost)) "


        pddl_domain_actions =  "(:action sync "\
                                ":parameters (?t1 - trace_state ?e - activity ?t2 - trace_state ?a1 - automaton_name) "\
                                ":precondition (and (cur_state ?a1 ?t1) (trace ?t1 ?e ?t2)) "\
                                ":effect (and (not (cur_state ?a1 ?t1)) (cur_state ?a1 ?t2) "\
                                "(forall (?a1 - automaton_name ?s1 ?s2 - automaton_state) "\
                                "(when (and (cur_state ?a1 ?s1) "\
                                "(automaton ?a1 ?s1 ?e ?s2)) "\
                                "(and (not (cur_state ?a1 ?s1)) "\
                                "(cur_state ?a1 ?s2)))))) "\
                                "(:action add "\
                                ":parameters (?e - activity ?a1 - automaton_name ?s1 ?s2 - state) "\
                                ":effect (and (increase (total-cost) 1) "\
                                "(forall (?a1 - automaton_name ?s1 ?s2 - automaton_state) "\
                                "(when (and (cur_state ?a1 ?s1) "\
                                "(automaton ?a1 ?s1 ?e ?s2)) "\
                                "(and (not (cur_state ?a1 ?s1)) "\
                                "(cur_state ?a1 ?s2)))))) "\
                                "(:action del "\
                                ":parameters (?t1 - trace_state ?e - activity "\
                                "?t2 - trace_state ?a1 - automaton_name) "\
                                ":precondition (and (cur_state ?a1 ?t1) (trace ?t1 ?e ?t2)) "\
                                ":effect (and (increase (total-cost) 1) "\
                                "(not (cur_state ?a1 ?t1)) (cur_state ?a1 ?t2))))"

        pddl_domain = pddl_domain_initial + pddl_domain_predicates + pddl_domain_actions

        # PROBLEM ###########################################################################
        problem_name = "problem_"+base_opt+"_"+type_db+"_trace"+str(id_trace)
        pddl_problem_initial = "(define (problem "+problem_name+") (:domain "+domain_name+") "
        pddl_problem_objects = "(:objects "

        for q in trace["Q_trace"]:
            pddl_problem_objects += q+" "
        pddl_problem_objects += "- trace_state "


        for q in all_states_constr_max:
            pddl_problem_objects += q+" "
        pddl_problem_objects += "- automaton_state a0 "


        for ID, a in all_automata.items():
            pddl_problem_objects += ID+" "
        pddl_problem_objects += "- automaton_name "


        for s in symbols_tot:
            pddl_problem_objects += s+" "
        pddl_problem_objects += "- activity"

        pddl_problem_objects += ") "


        pddl_problem_init = "(:init (= (total-cost) 0) (cur_state a0 "+trace["init_state_trace"]+") "
        for t in trace["rho_trace_basic"]:
            pddl_problem_init += "(trace "+t+") "
        pddl_problem_init += "(final_state a0 "+trace["final_state_trace"]+") "

        for ID, a in all_automata.items():
            pddl_problem_init += "(cur_state "+ID+" "+a["init_state"]+") " 
            for t in a["transitions"]:
                pddl_problem_init += "(automaton "+ID+" "+t+") "

            for i in range(len(a["final_states"])):
                pddl_problem_init += "(final_state "+ID+" "+a["final_states"][i]+") "
        pddl_problem_init += ") "


        pddl_problem_goal = "(:goal (forall (?a - automaton_name ?s - state) "\
                            "(imply (cur_state ?a ?s) (final_state ?a ?s)))) "
        pddl_problem_metric = "(:metric minimize (total-cost)))"
        pddl_problem = pddl_problem_initial + pddl_problem_objects + pddl_problem_init + pddl_problem_goal + pddl_problem_metric

        # SAVE ########################################################################
        file1 = open("PDDL/"+syn_real+"/"+base_opt+"/"+domain_name+".pddl", "w")
        file1.write(pddl_domain)
        file1.close()

        file2 = open("PDDL/"+syn_real+"/"+base_opt+"/"+problem_name+".pddl", "w")
        file2.write(pddl_problem)
        file2.close()<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.16788928775818115, 'fraction_numerical': 0.008688413714166796, 'mean_word_length': 1.8536223277909738, 'pattern_counts': {'":': 20, '<': 42, '<?xml version=': 0, '>': 76, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 14, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '9', 'dir': 'jupyter-structured-clean-dedup', 'id': '212161', 'license': 'Xnet', 'n_tokens_mistral': 6302, 'n_tokens_neox': 5972, 'n_words': 1164, 'path': 'main_try_opt.ipynb', 'repo_name': 'pietro-nardelli/LTLf-based-trace-alignment', 'size': '19221'}