######################################################## ######################################################## ## Laboratoire Specification et Verification ## TOOL_NAME = 'IMITATOR' ## ## Author: Etienne ANDRE ## ## Created : 03/2008 LAST_MODIF = '2009/02/24' VERSION = '1.3' ######################################################## ######################################################## ######################################################## ## IMPORTED MODULES ######################################################## import commands, copy, getopt, os, random, re, sys, time ######################################################## ## Computation time ######################################################## start_proc_time = time.clock() start_date = time.time() ######################################################## ## CONSTANTS TO BE CUSTOMIZED IN HYTECH FILES ######################################################## # Characteristics in the HyTech file INIT_REG = 'init_reg' FINAL_REG = 'post_reg' POST = 'post' # Tags in HyTech file START_PIZERO = '---START PI0---' END_PIZERO = '---END PI0---' # Tags in HYTECH LOG START_LOG = '---START LOG---' END_LOG = '---END LOG---' ######################################################## ## LOCAL CONSTANTS ######################################################## NOT_FOUND = -1 # Debug mode ; see signification below RESULT_ONLY = 0 NO_DEBUG = 1 LOW_DEBUG = 2 MEDIUM_DEBUG = 3 HIGH_DEBUG = 4 TOTAL_DEBUG = 5 # Syntax for debug in command line parameters {CODE : (commande line arg, signification)} DEBUG_SYNTAX = { RESULT_ONLY : ('result_only', 'only the constraints'), NO_DEBUG : ('no', 'no debugging information'), LOW_DEBUG : ('low', 'some debugging information'), MEDIUM_DEBUG : ('medium', 'quiet a lot of debugging information'), HIGH_DEBUG : ('high', 'much debugging information'), TOTAL_DEBUG : ('total', 'really too much debugging information') } # Operators OP_G = '>' OP_GEQ = '>=' OP_EQ = '=' OP_LEQ = '<=' OP_L = '<' # '=' is here replaced with '>' but in the code this is replaced by '<' and '>' INVERSE_OPERATORS = { OP_LEQ: '>', OP_GEQ: '<', OP_G: '<=', OP_L: '>=', OP_EQ : '>' } # To replace > with < and so on RETURNED_OPERATORS = { OP_LEQ: OP_GEQ, OP_GEQ: OP_LEQ, OP_G: OP_L, OP_L: OP_G, OP_EQ : OP_EQ } # Expression depth in Hytech (officially 40) HYTECH_EXPRESSION_DEPTH = 35 # HyTech tags HYTECH_POSTDEF_START = "\n" + '-- Start of post definition' + "\n" HYTECH_POSTDEF_END = "\n" + '-- End of post definition' + "\n" END_K = '---END K---' ######################################################## ## DATA STRUCTURE FOR CONSTRAINTS ######################################################## # A set of objects : equality "==" is performed using "==" and not simple address comparison class Set_of_objects(object): "A set of objects, compared using (possibly redefined) '==' on objects" # Convert a list of objects to a set of objects #def list_to_set(l): #result = set([]) #for element in l: #add_to_set(element, result) #return result def __init__(self, list_of_objects = []): self.set_of_objects = set([]) for element in list_of_objects: self.add_to_set(element) # Redefine length def __len__(self): return len(self.set_of_objects) # Copy the set def copy(self): return copy.deepcopy(self) # Return a new set made of the elements of the union of two sets #def union_copy(self, other): #result = self.copy() #for a in other.getObjects(): #result.add_to_set(a) #return result # Add the elements of other to self def add_all(self, other): for a in other.getObjects(): self.add_to_set(a) # Add the elements of a list to self def add_list(self, other_list): for a in other_list: self.add_to_set(a) # Return True if an element belongs to a set, False otherwise def in_set(self, element): for a in self.set_of_objects: if element == a: return True return False # Add an element to a set def add_to_set(self, element): if not self.in_set(element): (self.set_of_objects).add(element) # Return True if self is included in other, False otherwise def included(self, other): # First check size if len(self) > len(other): return False for a in self.set_of_objects: if not other.in_set(a): return False return True # Redefine equality '==' def __eq__(self, other): # TODO : il faudrait en fait verifier que "other" est une instance de (sous-classe de) Set_of_objects if other == None: return False return len(self) == len(other) and self.included(other) and other.included(self) # Return a list containing all the elements of the set, in no particular order def toList(self): result = [] for a in self.set_of_objects: result.append(a) return result # Return a random element of the set def random(self): l = self.toList() index = random.randint(0, len(l) - 1) return l[index] # Return the (normal) set of objects (useful for iterations) def getObjects(self): return self.set_of_objects # Redefine "print" def __str__(self): result = '[' first = True for a in self.set_of_objects: if first: first = False else: result += ', ' result += a.toString() result += ']' return result # An atom is of the form (i * X), where i is an integer, and X a variable name class Atom(object): "A variable name and a coefficient" def __init__(self, coeff, variable_name): self.coeff = coeff self.variable_name = variable_name # Redefinition of "==" def __eq__(self, atom): return self.coeff == atom.coeff and self.variable_name == atom.variable_name # Return a string in the specified format def toString(self): var = self.variable_name # Case no coeff if (self.coeff == 1): return var else: sep = '' return str(self.coeff) + sep + var # Redefine "print" (for debug) def __str__(self): return self.toString() # A sum of atoms and integers class Sum_of_atoms(object): "A set of atoms and a possible integer" def __init__(self, atoms, integer): self.atoms = atoms self.integer = integer # Return True if the inequality is only an integer, False otherwise def isInteger(self): return len(self.atoms) == 0 # Return a string in the specified format def toString(self): start = True result = '' # Convert the atoms for a in (self.atoms).getObjects(): if start: start = False else: result += ' + ' result += a.toString() # Write the integer if non null or alone if start: result += str(self.integer) else: if self.integer <> 0: result += ' + ' + str(self.integer) return result # Redefine equality def __eq__(self, other): return self.atoms == other.atoms and self.integer == other.integer # Redefine "print" (for debug) def __str__(self): return self.toString() # An inequality in normal form, i.e., a sum of atoms and integers, an operator (=, <, <=) and a sum of atoms and integers. Instanciation constraints var = integer must be in this order class Inequality(object): "A constraint in normal form" def __init__(self, left, op, right): # Case operator belonging to (<, <=) if op == OP_L or op == OP_LEQ: self.left = left self.op = op self.right = right # Case operator belonging to (=) else: if op == OP_EQ: # Put the instanciation inequality (var = integer) in normal form if len(left.atoms) == 0: self.left = right self.op = op self.right = left else: self.left = left self.op = op self.right = right # Case operator belonging to (>, >=) : invert the constraint else: if op == OP_G or op == OP_GEQ: self.left = right self.op = RETURNED_OPERATORS[op] self.right = left else: abort_program('Operator not understood in the construction of object Constraint') # Return True if the inequality is of the form '... = integer' or 'integer = ....', False otherwise def isInstanciation(self): return self.op == OP_EQ and (self.left.isInteger() or self.right.isInteger()) # Return True if the inequality is an equality def isEquality(self): return self.op == OP_EQ # Redefine equality def __eq__(self, other): # TODO : il faudrait verifier que other est du bon type if other == None: return False return self.op == other.op and self.left == other.left and self.right == other.right # Return True if self is more specific than the other inequality, False otherwise # (For example, "x + y + 5 < a + 5" is more specific than "x + 4 <= a + b + 8") def moreSpecific(self, other): atomsL1 = (self.left).atoms intL1 = (self.left).integer atomsR1 = (self.right).atoms intR1 = (self.right).integer op1 = self.op atomsL2 = (other.left).atoms intL2 = (other.left).integer atomsR2 = (other.right).atoms intR2 = (other.right).integer op2 = other.op if (op1 == OP_L and (op2 == OP_L or op2 == OP_LEQ)) and atomsL2.included(atomsL1) and intL2 <= intL1 and atomsR1.included(atomsR2) and intR1 <= intR2: return True return False # Return a string in the specified format def toString(self): op = self.op return (self.left).toString() + ' ' + op + ' ' + (self.right).toString() # Redefine "print" (for debug) def __str__(self): return self.toString() class Set_of_inequalities(Set_of_objects): "A disjunction is a set of inequalities in normal form" # Return a string in the specified format def toString(self): result = '' sep1 = ' & ' sep2 = "\n" for c in self.set_of_objects: result += sep1 + c.toString() + sep2 return result # Return a triple : (Set_of_inequalities of the instanciation constraints of the form "var = integer", Set_of_inequalities of the instanciation constraints of the form "... = ...", and a Set_of_inequalities of the other constraints) def getInstanciation(self): instanciation = Set_of_inequalities() equalities = Set_of_inequalities() others = Set_of_inequalities() for inequality in self.set_of_objects: # Look for an inequality of the form '... = integer' (or vice versa) if inequality.isInstanciation(): instanciation.add_to_set(inequality) else: if inequality.isEquality(): equalities.add_to_set(inequality) else: others.add_to_set(inequality) return (instanciation, equalities, others) def __str__(self): return self.toString() ######################################################## ## FUNCTIONS ######################################################## #******************************************************* #* Exit functions #******************************************************* # Aborting def abort_program(string): # TO DO !!!!!!! delete temp files print ' ***************** Fatal error *****************' print ' ' + string print '' # Recall the selected atoms for information if DEBUG_MODE >= LOW_DEBUG and len(selected_atoms) > 0: print ' Inequalities selected: (current constraint K)' print list_of_inequalities_toString(selected_atoms) # Print finalization information finalize() def finalize(): # Remove all tmp files unless specified by an option if not KEEPLOG: # Remove tmp Hytech log files for num_file in range(0, (nb_hytech)): remove_file(HYTECH_LOG_FILE_WITHOUT_EXT + '_' + str(num_file) + '.log') # Remove tmp Hytech file remove_file(HYTECH_TMP_FILE) # Remove HyTech simplification file remove_file(HYTECH_SIMPLIFICATION_FILE) # Remove HyTech simplification log file remove_file(HYTECH_SIMPLIFICATION_LOG_FILE) # Remove result file remove_file(RESULT_FILE) # Remove tmp log dir remove_directory(DIR_NAME) # Computation time end_proc_time = time.clock() end_date = time.time() script_duration = end_date - start_date proc_time = end_proc_time - start_proc_time # Print the computation time if asked if DEBUG_MODE >= NO_DEBUG: print ' ' + TOOL_NAME + ' ended after ' + str(round(script_duration, 2)) + ' seconds' print ' Python execution time : ' + str(round(proc_time, 2)) + 's' print ' HyTech execution time : ' + str(round(hytech_time, 2)) + 's (' + str(nb_hytech) + ' calls)' print ' ***********************************************' sys.exit(0) #******************************************************* #* Functions on files #******************************************************* # Read a file and return the content as a string def read_from_file(file_name): # Open file try: f = open(file_name, 'r') except IOError: abort_program('File ' + file_name + ' could not be opened.') # Read file file_content = f.read() f.close() if DEBUG_MODE >= HIGH_DEBUG: print ' File ' + file_name + ' successfully read.' return file_content # Write a string to a file (the string will replace the possible content) def write_to_file(file_name, file_content): # Open log file try: f = open(file_name, 'w') except IOError: abort_program('File ' + file_name + ' could not be written.') # Write to file f.write(file_content) f.close() if DEBUG_MODE >= HIGH_DEBUG: print ' File ' + file_name + ' successfully written.' # Remove a file def remove_file(file_name): # Try to remove file try: os.remove(file_name) except OSError: print ' WARNING : File ' + file_name + ' could not be deleted.' return if DEBUG_MODE >= HIGH_DEBUG: print ' File ' + file_name + ' successfully deleted.' # Remove a directory def remove_directory(dir_name): # Try to remove directory try: os.rmdir(dir_name) except OSError: print ' WARNING : Directory ' + dir_name + ' could not be deleted.' return if DEBUG_MODE >= HIGH_DEBUG: print ' Directory ' + dir_name + ' successfully deleted.' #******************************************************* #* Functions on HyTech files #******************************************************* # Count the number of locations in a set of constraints def get_nb_locations(constraints): return len(constraints) # Count the number of disjunctions in a set of constraints def get_nb_disjunctions(constraints): result = 0 for location in constraints.itervalues(): result += len(location) return result ## Check that the constraints section exists in a HyTech file (string) #def check_constraints_section(string): #check = re.search(START_K #+ "((.|\n)*)" #+ END_K, #string) #if check == None: #abort_program('Constraints section not found in ' + HYTECH_ORIG_FILE) #else: #if DEBUG_MODE >= HIGH_DEBUG: #print ' Constraints section found in ' + HYTECH_ORIG_FILE # Get the constraints from the instanciation section in a HyTech file (string) def get_instanciation_constraints(string): search = re.search(START_PIZERO + "(?P(.|\n)*)" + END_PIZERO, string) # If no constraints if search == None: if DEBUG_MODE >= LOW_DEBUG: print ' No instanciation found in ' + HYTECH_ORIG_FILE return [] # Remove HyTech comments "--" constraints = re.sub("--+", '', search.group('constraints')) # Remove new lines constraints = re.sub("\n", ' ', constraints) # Split the constraints with "&" constraints = re.split("\s*&\s*", constraints) # Remove blank constraints def is_not_blank(x): if re.search("\S+", x): return True else: return False constraints = filter(is_not_blank, constraints) # Make atoms of the form [a, op, b] constraints = map(make_atom, constraints) # Convert to a set of objects return Set_of_inequalities(constraints) # Get the reference valuation pi_0 from the instanciation section in a HyTech file (string) # Return a list of couples (name, value) def get_pi_0(string): search = re.search(START_PIZERO + "(?P(.|\n)*)" + END_PIZERO, string) # If no valuations if search == None: if DEBUG_MODE >= LOW_DEBUG: print ' No instanciation found in ' + HYTECH_ORIG_FILE return [] # Remove HyTech comments "--" valuations = re.sub("--+", '', search.group('valuations')) # Remove new lines valuations = re.sub("\n", ' ', valuations) # Split the constraints with "&" valuations = re.split("\s*&\s*", valuations) # Remove blank "valuations" def is_not_blank(x): if re.search("\S+", x): return True else: return False valuations = filter(is_not_blank, valuations) # Create pi_0 under the form of a dictionary pi_0 = dict() # Fill in the dictionary with couples variable --> value for valuation in valuations: # Check that the equality is under the form "variable = value" search = re.match("(\s*)(?P\w+)(\s*)=(\s*)(?P\d+)(\s*)$", valuation) # If not : abort if search == None: abort_program('When parsing pi_0, the equality "' + valuation + ' could not be recognized. An inequality should be of the form "variable = value"') else: # Confirmation message if DEBUG_MODE >= HIGH_DEBUG: print ' Following value added to pi_0 : "' + search.group('variable') + " := " + search.group('value') + '"' # Add the valuation to the dictionary pi_0[search.group('variable')] = search.group('value') # At that point, we could verify that every value is defined only once return pi_0 # Check that the final_reg is defined in the HYTECH_TMP_FILE and tag it def tag_final_reg(hytech_file): # Check the final_reg definition search = re.search("(?P(.|\n)*)" + "(?P(" + FINAL_REG + "(\s*):=(\s*)([^;]+);))" + "(?P(.|\n)*)", hytech_file) if search == None: abort_program('Final region definition "' + FINAL_REG + ' := [...];" not found in ' + HYTECH_ORIG_FILE) # Print a successful message if DEBUG_MODE >= HIGH_DEBUG: print ' Final region definition found in ' + HYTECH_ORIG_FILE # Tag the final_reg definition hytech_file = search.group('start') + HYTECH_POSTDEF_START + search.group('final') + HYTECH_POSTDEF_END + search.group('end') # Return file return hytech_file # Update the temp HyTech file by replacing the final_reg definition in the file with its current step definition def update_final_reg(step): # Definition of post* final_reg_def = FINAL_REG + ' := ' + INIT_REG + ';' + "\n" # Need to do several lines if step > HYTECH_EXPRESSION_DEPTH nb_expr = (step // HYTECH_EXPRESSION_DEPTH) + 1 # Definition of a standard maximum line of HYTECH_EXPRESSION_DEPTH post standard_line = FINAL_REG for i in range(0, HYTECH_EXPRESSION_DEPTH): standard_line = POST + '(' + standard_line + ')' standard_line = FINAL_REG + ' := ' + standard_line + ';' for count_expr in range(0, nb_expr): # Case of a maximum line of HYTECH_EXPRESSION_DEPTH post if (count_expr < nb_expr - 1): final_reg_def += standard_line + ' -- ' + POST + '^(' + str(count_expr+1) + '*' + str(HYTECH_EXPRESSION_DEPTH) + ")\n"; # Case of the last line of the form final_reg := post(...(post(init_reg))...); -- post^i else: nb_post_to_do = step - ((nb_expr-1) * HYTECH_EXPRESSION_DEPTH) last_line = FINAL_REG for useless in range(0, nb_post_to_do): last_line = POST + '(' + last_line + ')' final_reg_def += FINAL_REG + ' := ' + last_line + '; -- ' + POST + '^' + str(step) + "\n"; # Open temp HyTech file tmp_file_string = read_from_file(HYTECH_TMP_FILE) # Replace old final_reg definition by new one computed above tmp_file_string = re.sub(HYTECH_POSTDEF_START + "(.|\n)*" + HYTECH_POSTDEF_END, HYTECH_POSTDEF_START + final_reg_def + HYTECH_POSTDEF_END, tmp_file_string) # Write and close file write_to_file(HYTECH_TMP_FILE, tmp_file_string) # Simplify a set of inequalities by letting HyTech perform preprocessing on it def simplify_constraint(inequalities): # Read original file simplification_file = read_from_file(HYTECH_ORIG_FILE) # Remove comments in HyTech file simplification_file = re.sub("--(.*)", "", simplification_file) # Remove (some) blank lines in HyTech file simplification_file = re.sub("(\s*\n){2,}", "\n", simplification_file) # The end of the file is a trivial automaton and a trivial region print new_end_of_file = ' automaton aa' + "\n" + 'synclabs: ;' + "\n" + 'initially A1;' + "\n" + 'loc A1: while True wait {}' + "\n" + 'when True do {} goto A1;' + "\n" + 'end -- aa' + "\n" + 'var init_reg: region;' + "\n" + 'init_reg := True' + "\n" + (inequalities.toString()) + "\n" + ';' + "\n" + 'prints "' + START_LOG + '";' + "\n" + 'print(hide non_parameters in init_reg endhide );' + "\n" + 'prints "' + END_LOG + '";' + "\n" # Replace everything from the first occurrence of "automaton" until the end of the file with our new end of file simplification_file = '-- FILE GENERATED BY ' + TOOL_NAME + ' ' + now_is() + "\n\n" + re.sub("automaton((.|\n)*)", new_end_of_file, simplification_file) # Save it to a new HyTech file write_to_file(HYTECH_SIMPLIFICATION_FILE, simplification_file) # Perform HyTech computation cmd = 'hytech ' + HYTECH_SIMPLIFICATION_FILE + ' > ' + HYTECH_SIMPLIFICATION_LOG_FILE if DEBUG_MODE >= HIGH_DEBUG: print ' cmd?> ' + cmd # Execute command commands.getoutput(cmd) # Find constraints string_simplified_constraints = get_log_constraints(HYTECH_SIMPLIFICATION_LOG_FILE) # Remove 'Location: A1' string_simplified_constraints = re.sub("Location: A1", "", string_simplified_constraints) # Normalize form of the constraints simplified_constraints = parse_inequalities(string_simplified_constraints) return simplified_constraints #******************************************************* #* Functions on log file #******************************************************* #*------------------------------------------------------ # Data structure for constraints in log file : #*------------------------------------------------------ # dictionary of # Location_name --> List_of_inequalities # Get the constraints from the log file def get_log_constraints(log_file_name): # Read log file log_file = read_from_file(log_file_name) # Looking for the constraint zone search = re.search(START_LOG + "(?P(.|\n)*)" + END_LOG, log_file) if search == None: print log_file abort_program('Constraints zone not found in ' + log_file_name) return search.group('constraints') # Make a data structure [set(var), op, set(var)] from an atom under the form of a string def make_atom(atom): # STUPID SIMPLIFICATION HERE : 'True' becomes '0 = 0' (!) if(re.match("(\s*)True(\s*)", atom) ): a = Sum_of_atoms(Set_of_inequalities(), 0) return Inequality(a, OP_EQ, a) search = re.split("(" + OP_GEQ + "|" + OP_G + "|" + OP_EQ + "|" + OP_LEQ + "|" + OP_L + ")", atom) if len(search) != 3: abort_program('The atom is not of the form "A op B" in make_atom(' + atom + ')') a = search[0] op = search[1] b = search[2] # Function to split sum of variables in a set of variables def sum2set_of_atoms(sum_of_vars): # Split with respect to "+" selection = re.split("\+", sum_of_vars) # Remove blanks def remove_blanks(a): return re.sub("\s*", '', a) selection = map(remove_blanks, selection) # Result under the form of a Sum_of_inequalities atoms = Set_of_inequalities() integer = 0 # Iterate on the variable names for a in selection: # Look for an integer search = re.match("(?P\d+)$", a) if search <> None: integer += int(search.group('integer')) else: # Case of a variable with possible coeff coeff = 1 name = a # Look for the coeff search = re.match("(?P\d+)(?P.*)$", a) if search <> None: coeff = search.group('coeff') name = search.group('name') atoms.add_to_set(Atom(coeff, name)) # Return return Sum_of_atoms(atoms, integer) a = sum2set_of_atoms(a) b = sum2set_of_atoms(b) # Return the inequality return Inequality(a, op, b) # Parses a string representing inequalities into a Set_of_inequalities structure def parse_inequalities(string_inequalities): # Remove newlines string_inequalities = re.sub("\n", ' ', string_inequalities) # Split inequalities with "&" string_inequalities = re.split("\s*\&\s*", string_inequalities) # Making inequalities of the form [a, op, b] inequalities = Set_of_inequalities(map(make_atom, string_inequalities)) # Returns return inequalities # Make a data structure constraints from a string def make_constraints(string): # Creating the result data structure result = dict([]) # Looking for the different locations locations = re.finditer('Location: ' + "(?P(\S+))(\s|\n)*(?P(((.|\n)(?!Location))*))", string) for location in locations: if location == None: abort_program('Parse error in ' + HYTECH_LOG_FILE + ': "Location" zone found but constraints were not found.') disjunctions = re.split("\s*\|\s*", location.group('constraints')) location_name = location.group('location') #result[location_name] = parse_inequalities(disjunctions) result[location_name] = [] ## Looking for the disjunctions in locations for disjunction in disjunctions: # Add the list of inequalities to the location result[location_name].append(parse_inequalities(disjunction)) return result # Print constraints for debugging def print_constraints(constraints): iterator = constraints.iteritems() for (location_name, disjunctions) in iterator: print ' Location: ' + location_name i = 1 for disjunction in disjunctions: print disjunction if i < len(disjunctions): print ' |' i = i + 1 # Compute the set of reachable states and outputs a constraint def compute_reachable_states(): # Find the constraints zone log_file = get_log_constraints(HYTECH_LOG_FILE) # Convert the constraints constraints = make_constraints(log_file) if DEBUG_MODE >= TOTAL_DEBUG: print ' -----------------------------' print ' The constraints read are:' print_constraints(constraints) print ' -----------------------------' # Return the constraints return constraints #******************************************************* #* Functions to select a pi_0-incompatible inequality in a Set_of_inequalities #******************************************************* # Select a Select an atom in the constraints data structure or return None if no suitable atom is found def select_pi_0_incompatible_inequality(inequalities): # List of all pi_0-incompatible (negated) inequalities, under the form of a Set_of_inequalities pi_0_incompatible_inequalities = Set_of_inequalities() # Iteration on 'locations' (set of states) for (location_name, location) in inequalities.iteritems(): # Print location name if DEBUG_MODE >= MEDIUM_DEBUG: print ' Considering location "' + location_name + '" :' # Iteration on states for state in location: # Count the number of pi-incomp inequalities in a single state nb_random_ineq = 0 # Print the state if DEBUG_MODE >= MEDIUM_DEBUG: print ' Considering the following state in this location :' print state # Iteration on the inequalities ('state' is a Set_of_inequalities object) for inequality in state.getObjects(): # Perform the negation of the inequality (there might be 2 in the case of "=") for negated_inequality in neg_inequality(inequality): # Check if the inequality was already selected if (pi_0_incompatible_inequalities.in_set(negated_inequality)): if DEBUG_MODE >= HIGH_DEBUG: print ' Found an inequality which has already been selected: skip.' else: # Check if this negated inequality is pi_0-compatible pi_0_compatible = check_pi_0_compatibility(negated_inequality) # If no, it means that the inequality is pi_0-compatible : skip if not pi_0_compatible: if DEBUG_MODE >= HIGH_DEBUG: print ' Skip a pi_0-compatible inequality (' + inequality.toString() + ')' # If pi_0-incompatible : keep it else: if DEBUG_MODE >= LOW_DEBUG: print ' Found a pi_0-incompatible inequality : ' + inequality.toString() # Add the inequality to the set of pi_0-incompatible (negated) inequalities pi_0_incompatible_inequalities.add_to_set(negated_inequality) nb_random_ineq = nb_random_ineq + 1 # [optimization] If not random : return immediately the pi_0-incompatible inequality if not RANDOM: return negated_inequality if (DEBUG_MODE >= LOW_DEBUG and nb_random_ineq > 0): print ' Number of pi-incompatible inequalities in this pi-incompatible state : ' + str(nb_random_ineq) # If no pi_0-incompatible inequality found : return None if len(pi_0_incompatible_inequalities) == 0: return None # Choose randomly a pi_0-incompatible inequality among the set of pi_0-incompatible inequalities if len(pi_0_incompatible_inequalities) > 1 and DEBUG_MODE >= NO_DEBUG: print ' Randomly choosing an inequality among ' + str(len(pi_0_incompatible_inequalities)) + ' possible pi_0-incompatible inequalities' return pi_0_incompatible_inequalities.random() # negate an inequality and return a LIST of inequalities (because of case '=' where the negation has two results) def neg_inequality(inequality): a = inequality.left b = inequality.right op = inequality.op if op not in INVERSE_OPERATORS: abort_program('The operator "' + op + '" is not correct in inequality to negate "' + inequality.toString()) # Case 'a = b' : return ['a > b', 'a < b'] if op == OP_EQ: return [Inequality(a, OP_G, b), Inequality(a, OP_L, b)] # Else return ['a inv(op) b'] else: return [Inequality(a, INVERSE_OPERATORS[op], b)] #******************************************************* #* Functions with interaction between inequalities and pi_0 #******************************************************* # Return the value associated to a variable name def get_value_from_pi_0(variable_name): # Check that this variable is in pi_0 if variable_name not in pi_0: abort_program('The variable name "' + variable_name + '" is used but was not given a reference value in pi_0. You should probably hide it in the HyTech file.') # Return the reference value associated to this variable return int(pi_0[variable_name]) # Compute the real value of an object Sum_of_atoms using the reference valuation pi_0 def compute_sum_with_pi_0(sum_of_atoms): # Start with the possible integer sum = sum_of_atoms.integer # Compute the sum of the variables using pi_0 for atom in (sum_of_atoms.atoms).getObjects(): # An atom if a coeff and a variable name coeff = int(atom.coeff) variable_name = atom.variable_name # Check the value of the variable in pi_0 value = get_value_from_pi_0(variable_name) # Perform the sum sum += (coeff * value) return sum # Perform a comparison between two integers and a comparison operator def perform_comparison(left, op, right): # Check the operator and perform the comparison if (op == OP_G): return left > right else: if (op == OP_GEQ): return left >= right else: if (op == OP_EQ): return left == right else: if (op == OP_LEQ): return left <= right else: if (op == OP_L): return left < right # If other: error else: abort_program('The operator "' + op + '" is not correct in perform_inequality("' + str(left) + "," + op + "," + str(right) + ')') # Check that an inequality is pi_0-compatible def check_pi_0_compatibility(inequality): # Left sum of atoms left = inequality.left # Operator op = inequality.op # Right sum of atoms right = inequality.right # Compute left operand value using pi_0 left_value = compute_sum_with_pi_0(left) # Compute right operand value using pi_0 right_value = compute_sum_with_pi_0(right) # Check inequality result = perform_comparison(left_value, op, right_value) # Print some message if result and DEBUG_MODE >= TOTAL_DEBUG: print ' ' + inequality.toString() + ' is pi_0-compatible (' + str(left_value) + op + str(right_value) + ')' if not result and DEBUG_MODE >= TOTAL_DEBUG: print ' ' + inequality.toString() + ' is pi_0-incompatible (' + str(left_value) + op + str(right_value) + ' is not verified)' return result #******************************************************* #* Functions on list of inequalities #******************************************************* # Redef this function because the native '==' is on object addresses... def in_list(inequality, list_of_inequalities): for i in list_of_inequalities: if i == inequality: return True return False # Copy a list def copyList(list_of_inequalities): result = [] for inequality in list_of_inequalities: result.append(inequality) return result # Return a new list made of the elements of the old list in the same order, plus the new inequality, minus the redundant constraints in the list def add_to_list_and_reduce(new_inequality, list_of_inequalities): # Check if the new inequality is redundant with the old inequalities (very unlikely) for inequality in list_of_inequalities: if inequality.moreSpecific(new_inequality): if DEBUG_MODE >= NO_DEBUG: print ' Unlikely case: removing the last inequality found because it is less specific than another one ' + new_inequality.toString() # Stop and do not change the list return list_of_inequalities # Remove the redundant inequalities compared to the new inequality new_list = [] for inequality in list_of_inequalities: if new_inequality.moreSpecific(inequality): if DEBUG_MODE >= NO_DEBUG: print ' Removing redundant inequality ' + inequality.toString() else: new_list.append(inequality) # Add the new_inequality at the end new_list.append(new_inequality) return new_list # Convert a list of inequalities to a string def list_of_inequalities_toString(list_of_inequalities): result = '' sep1 = ' & ' sep2 = "\n" for i in list_of_inequalities: result += sep1 + i.toString() + sep2 return result # Make a structure def log_to_set_of_inequalities(log): result = Set_of_inequalities() # Iterate on the locations of the log file for location in log.itervalues(): # Iterate on the disjunctions for one location for disjunction in location: # A disjunction is a Set_of_inequalities result.add_all(disjunction) return result #******************************************************* #* Functions on system #******************************************************* # Return the absolute date and time under the form of a string def now_is(): return time.strftime("%a, %d %b %Y %H:%M:%S", time.gmtime()) # Return the time under the form of a string "t = xxxx s", starting from t=0 def time_info(): time_info = '' if TIMED_MODE: time_info = ' (at t = ' + str(round(time.time() - start_date, 2)) + 's)' return time_info # Call to HyTech def hytech(): global hytech_time, nb_hytech, HYTECH_LOG_FILE HYTECH_LOG_FILE = HYTECH_LOG_FILE_WITHOUT_EXT + '_' + str(nb_hytech) + '.log' cmd = 'hytech ' + HYTECH_TMP_FILE + ' > ' + HYTECH_LOG_FILE if DEBUG_MODE >= HIGH_DEBUG: print ' cmd?> ' + cmd # Compute time tmp_start_date = time.time() # Execute command commands.getoutput(cmd) # Compute execution time tmp_end_date = time.time() hytech_time += tmp_end_date - tmp_start_date # Increment number of executions nb_hytech += 1 #(status, _) = commands.getstatusoutput(cmd) #if status != 0: #abort_program('HyTech exited abnormally with error ' + status) ######################################################## ## START ######################################################## #******************************************************* #* SOME GLOBAL VARIABLES #******************************************************* # That is the most important thing we are looking for !! (ordered list of inequalities) selected_atoms = [] # The atoms which are already listed as not compatible in Prolog inconsistent_atoms = Set_of_inequalities() # Computation times hytech_time = 0 nb_hytech = 0 #******************************************************* #* COMMAND LINE PARAMETERS #******************************************************* # Get the parameters as an array args = sys.argv # Python script name PYTHON_FILE_NAME = args[0] # Print the syntax of the parameters def print_help(): print ' Syntax is the following:' print ' python ' + PYTHON_FILE_NAME + ' [Hytech_file_without_extension]' print '' print ' Options' print ' -v | --version print version information' print ' -h | --help print some help' print '' print ' --log_dir=[dir_name] create the temporary log files in the [dir_name] directory' print ' (default: [dir_name] = [Hytech_file_without_extension])' print ' --keeplog keep the log directory at the end of the computation' print ' (default: no, i.e., remove all temp files)' print ' --norandom choose deterministically the first pi_0-incompatible inequality' print ' (default: no, i.e., find all and choose one randomly)' print ' --timed print time on screen for each action performed' print ' (default: no)' print ' --debug= ' for (debug_opt, debug_meaning) in DEBUG_SYNTAX.values(): print ' ' + debug_opt + ' : ' + debug_meaning print ' (default: no)' print '' print ' Examples:' print ' python ' + PYTHON_FILE_NAME + ' my_hytech_file' print ' python ' + PYTHON_FILE_NAME + ' my_circuit --norandom --debug=result_only' print ' python ' + PYTHON_FILE_NAME + ' my_circuit --debug=low --timed' print ' python ' + PYTHON_FILE_NAME + ' my_protocol --log_dir=protocols --keeplog' #******************************************************* #* SET PARAMETERS AND OPTIONS #******************************************************* # Get parameters and options try: options, parameters = getopt.gnu_getopt(args[1:], 'vhr', ['help', 'log_dir=', 'norandom', 'keeplog', 'timed', 'version', 'debug=']) except getopt.GetoptError, error: print ' ** FATAL ERROR **' print ' ' + error.msg print '' print_help() sys.exit(0) # Debug mode DEBUG_MODE = NOT_FOUND for (o, v) in options: if o == '--debug': for debug_code, (debug_opt, unuseful) in DEBUG_SYNTAX.iteritems(): if v == debug_opt: DEBUG_MODE = debug_code break # If not found: error if DEBUG_MODE == NOT_FOUND: print ' Debug mode "' + v + '" not recognized.' print_help() sys.exit(0) # Default : no debug if DEBUG_MODE == NOT_FOUND: DEBUG_MODE = NO_DEBUG # Print version information if ('-v', '') in options or ('--version', '') in options: print ' Version ' + VERSION print ' Last modified: ' + LAST_MODIF #print '' #print ' To know more about the script syntax, use option -h' sys.exit(0) # Print help if ('-h', '') in options or ('--help', '') in options: print_help() sys.exit(0) # Randomization (default: yes) if ('--norandom', '') in options: RANDOM = False if DEBUG_MODE >= LOW_DEBUG: print ' Random mode set to FALSE' else: if DEBUG_MODE >= LOW_DEBUG: print ' Random mode set to TRUE' RANDOM = True # Keep log (default: no) if ('--keeplog', '') in options: if DEBUG_MODE >= LOW_DEBUG: print ' Keeplog mode set to TRUE' KEEPLOG = True else: if DEBUG_MODE >= LOW_DEBUG: print ' Keeplog mode set to FALSE' KEEPLOG = False # Print time information (default: no) if ('--timed', '') in options: if DEBUG_MODE >= LOW_DEBUG: print ' Time mode set to TRUE' TIMED_MODE = True else: if DEBUG_MODE >= LOW_DEBUG: print ' Time mode set to FALSE' TIMED_MODE = False # Check the number of parameters if len(parameters) > 1: print ' *WARNING* There are useless parameters, they will be ignored.' print ' For more information about the syntax, use option -h.' print ' ' if len(parameters) < 1: print 'Error: no HyTech file name was found. For more information about the syntax, use option -h.' print ' ' sys.exit(0) # Original Hytech file without extension HYTECH_ORIG_FILE_WITHOUT_EXT = parameters[0] # Log file directory # default : hytech file without ext DIR_NAME = HYTECH_ORIG_FILE_WITHOUT_EXT for (op, value) in options: if op == '--log_dir': DIR_NAME = value # Create the directory if not os.path.isdir(DIR_NAME): try: os.mkdir(DIR_NAME) except OSError: abort_program('Directory "' + DIR_NAME + '" could not be created.') if DEBUG_MODE >= HIGH_DEBUG: print ' Directory ' + DIR_NAME + ' successfully created.' else: if DEBUG_MODE >= MEDIUM_DEBUG: print ' Directory ' + DIR_NAME + ' was not created because it already exists.' # HyTech original file HYTECH_ORIG_FILE = HYTECH_ORIG_FILE_WITHOUT_EXT + '.hy' # HyTech tmp file (copy) HYTECH_TMP_FILE = DIR_NAME + '/' + HYTECH_ORIG_FILE_WITHOUT_EXT + '_tmp.hy' # HyTech simplification file HYTECH_SIMPLIFICATION_FILE = DIR_NAME + '/' + HYTECH_ORIG_FILE_WITHOUT_EXT + '_simplify.hy' # HyTech log file with directory without extension HYTECH_LOG_FILE_WITHOUT_EXT = DIR_NAME + '/' + HYTECH_ORIG_FILE_WITHOUT_EXT.replace('.', '_') # Log file with extension : defined in hytech() function #HYTECH_LOG_FILE = HYTECH_LOG_FILE_WITHOUT_EXT + '.log' # HyTech simplification log file HYTECH_SIMPLIFICATION_LOG_FILE = HYTECH_LOG_FILE_WITHOUT_EXT + '_simplify.log' # Result file RESULT_FILE = DIR_NAME + '/' + HYTECH_ORIG_FILE_WITHOUT_EXT + '.result.txt' #******************************************************* #* WELCOME MESSAGE #******************************************************* if DEBUG_MODE >= NO_DEBUG: print ' ***********************************************' print ' * ' + TOOL_NAME + ' *' print ' * Etienne ANDRE *' print ' * 2008 - 2009 *' print ' * Laboratoire Specification et Verification *' print ' * ENS de Cachan & CNRS, France *' print ' ***********************************************' #+ "\n" print ' ' + now_is() if DEBUG_MODE >= LOW_DEBUG: print ' Treating file ' + HYTECH_ORIG_FILE #******************************************************* #* OPERATIONS ON ORIGINAL HYTECH FILE #******************************************************* # Read original file original_file_string = read_from_file(HYTECH_ORIG_FILE) # Add a tag END_K just after END_PIZERO to insert at that point the constraint K original_file_string = original_file_string.replace(END_PIZERO, END_PIZERO + "\n\t" + END_K) # Check that the final_reg is defined and tag it original_file_string = tag_final_reg(original_file_string) # Get the reference valuation pi_0 pi_0 = get_pi_0(original_file_string) # Copy Hytech original file to the tmp file write_to_file(HYTECH_TMP_FILE, '-- FILE GENERATED BY ' + TOOL_NAME + ' ' + now_is() + "\n\n" + original_file_string) original_file_string = '' # free memory ######################################################## ## MAIN LOOP (algorithm InverseMethod) ######################################################## # Post^i step = 1 # Number of locations at the beginning of the step nb_locations = 0 new_nb_locations = 0 # Number of states nb_disjunctions = 0 new_nb_disjunctions = 0 # Whether we are still growing more_locations = True # Stop when the number of locations is not growing anymore while more_locations: #******************************************************* #* UPDATE HYTECH FILE WITH CURRENT STEP #******************************************************* update_final_reg(step) # Print blank lines for debugging if DEBUG_MODE >= LOW_DEBUG: print '' if DEBUG_MODE >= MEDIUM_DEBUG: print '' # Print step number if DEBUG_MODE >= NO_DEBUG: print ' Step ' + str(step) + time_info() # First Hytech computation in the current step ? first_computation_in_step = True while True: #******************************************************* #* LAUNCH HYTECH #******************************************************* hytech() #******************************************************* #* COMPUTE THE ATOM TO NEGATE FROM THE LOG FILE #******************************************************* constraints = compute_reachable_states() # Count the number of locations and disjunctions new_nb_locations = get_nb_locations(constraints) new_nb_disjunctions = get_nb_disjunctions(constraints) if(new_nb_locations == 0): abort_program('No more constraint available in the log file at step ' + str(step) + ' :\'(') # Update number of locations #tmp_nb_locations = new_nb_locations if DEBUG_MODE >= LOW_DEBUG: print ' Reached ' + str(new_nb_disjunctions) + ' states (from ' + str(new_nb_locations) + ' different locations)' if first_computation_in_step: # If the number of locations is stable, then fixpoint and END if(new_nb_locations == nb_locations and new_nb_disjunctions == nb_disjunctions): if DEBUG_MODE >= NO_DEBUG: print '' print ' Post* was reached at step ' + str(step-1) + ' with ' + str(nb_disjunctions) + ' states (from ' + str(nb_locations) + ' different locations)' # Print the selected inequalities (current K) if DEBUG_MODE >= LOW_DEBUG and len(selected_atoms) > 0: print ' Inequalities selected: (current constraint K)' print list_of_inequalities_toString(selected_atoms) # Get the inequalities associated to all states of post* post_star = log_to_set_of_inequalities(constraints) if DEBUG_MODE >= TOTAL_DEBUG: print '' print ' States in post* are:' print_constraints(constraints) if DEBUG_MODE >= LOW_DEBUG: print '' print ' Intersection of the inequalities of all states in post* (brute non-simplified version):' print post_star # Simplify this list of inequalities using HyTech final_constraint = simplify_constraint(post_star) if DEBUG_MODE >= NO_DEBUG: print '' print ' Intersection of the inequalities of all states in post* (after simplification):' if DEBUG_MODE >= RESULT_ONLY: print final_constraint # Create a log file with the constraint write_to_file(RESULT_FILE, final_constraint.toString()) # Quit the script more_locations = False break else: first_computation_in_step = False # Find a (negated) pi_0-incompatible inequality pi_0_incompatible_inequality = select_pi_0_incompatible_inequality(constraints) if pi_0_incompatible_inequality == None: if DEBUG_MODE >= LOW_DEBUG: print ' No pi_0-incompatible inequality was found' # Go to next step (post^{i+1}) break ## Perform the negation (note that there might be 2 possible negations in the case where the inequality is an equality using '=') #inequality = negate_pi_incompatible_inequality(pi_0_incompatible_inequality) # Nice message printed on screen if DEBUG_MODE >= NO_DEBUG: print ' Adding inequality ' + pi_0_incompatible_inequality.toString() + time_info() # Add the pi_0-incompatible inequality to the list, and remove possibly redundant inequalities selected_atoms = add_to_list_and_reduce(pi_0_incompatible_inequality, selected_atoms) #******************************************************* #* ADD THE INEQUALITY TO THE HYTECH FILE #******************************************************* # Read tmp file tmp_file_string = read_from_file(HYTECH_TMP_FILE) # Add the inequality (no verification to check that END_K is present) tmp_file_string = tmp_file_string.replace(END_K, '& ' + pi_0_incompatible_inequality.toString() + "\n\t" + END_K) # Write and close file write_to_file(HYTECH_TMP_FILE, tmp_file_string) tmp_file_string = '' # Free memory # Update number of locations nb_locations = new_nb_locations nb_disjunctions = new_nb_disjunctions # Go to the following step step += 1 ######################################################## ## THE END ######################################################## if DEBUG_MODE >= NO_DEBUG: print ' ' + TOOL_NAME + ' ended with success :-)' print '' # Finalisation finalize()