Merge pull request #71 from steve-anunknown/improve-performance-of-w-methods

optimized the W and Wp method oracles

Author: emuskardin

aalpy/oracles/WMethodEqOracle.py

@@ -2,8 +2,7 @@
 
 from aalpy.base.Oracle import Oracle
 from aalpy.base.SUL import SUL
-from aalpy.utils.HelperFunctions import product_with_possible_empty_iterable
-
+from itertools import product
 
 class WMethodEqOracle(Oracle):
     """
@@ -26,34 +25,55 @@ def __init__(self, alphabet: list, sul: SUL, max_number_of_states, shuffle_test_
         self.shuffle = shuffle_test_set
         self.cache = set()
 
+    def test_suite(self, cover, depth, char_set):
+        """
+        Construct the test suite for the W Method using
+        the provided state cover and characterization set,
+        exploring up to a given depth.
+        Args:
+
+            cover: list of states to cover
+            depth: maximum length of middle part
+            char_set: characterization set
+        """
+        # fix the length of the middle part per loop
+        # to avoid generating large sequences early on
+        char_set = char_set or [()]
+        for d in range(depth):
+            middle = product(self.alphabet, repeat=d)
+            for m in middle:
+                for (s, c) in product(cover, char_set):
+                    yield s + m + c
+
+
     def find_cex(self, hypothesis):
 
         if not hypothesis.characterization_set:
             hypothesis.characterization_set = hypothesis.compute_characterization_set()
 
         # covers every transition of the specification at least once.
-        transition_cover = [state.prefix + (letter,) for state in hypothesis.states for letter in self.alphabet]
-
-        middle = []
-        for i in range(self.m + 1 - len(hypothesis.states)):
-            middle.extend(list(product_with_possible_empty_iterable(self.alphabet, repeat=i)))
-
-        for seq in product_with_possible_empty_iterable(transition_cover, middle, hypothesis.characterization_set):
-            inp_seq = tuple([i for sub in seq for i in sub])
-            if inp_seq not in self.cache:
+        transition_cover = [
+                state.prefix + (letter,)
+                for state in hypothesis.states
+                for letter in self.alphabet
+                ]
+
+        depth = self.m + 1 - len(hypothesis.states)
+        for seq in self.test_suite(transition_cover, depth, hypothesis.characterization_set):
+            if seq not in self.cache:
                 self.reset_hyp_and_sul(hypothesis)
                 outputs = []
 
-                for ind, letter in enumerate(inp_seq):
+                for ind, letter in enumerate(seq):
                     out_hyp = hypothesis.step(letter)
                     out_sul = self.sul.step(letter)
                     self.num_steps += 1
 
                     outputs.append(out_sul)
                     if out_hyp != out_sul:
                         self.sul.post()
-                        return inp_seq[:ind + 1]
-                self.cache.add(inp_seq)
+                        return seq[:ind + 1]
+                self.cache.add(seq)
 
         return None
 

aalpy/oracles/WpMethodEqOracle.py

@@ -1,8 +1,7 @@
-from itertools import chain, tee
+from itertools import chain, product
 
 from aalpy.base.Oracle import Oracle
 from aalpy.base.SUL import SUL
-from aalpy.utils.HelperFunctions import product_with_possible_empty_iterable
 
 
 def state_characterization_set(hypothesis, alphabet, state):
@@ -23,34 +22,45 @@ def state_characterization_set(hypothesis, alphabet, state):
     return result
 
 
-def i_star(alphabet, max_seq_len):
+def first_phase_it(alphabet, state_cover, depth, char_set):
     """
-    Return an iterator that generates all possible sequences of length upto from the given alphabet.
+    Return an iterator that generates all possible sequences for the first phase of the Wp-method.
     Args:
         alphabet: input alphabet
-        max_seq_len: maximum length of the sequences
+        state_cover: list of states to cover
+        depth: maximum length of middle part
+        char_set: characterization set
     """
-    return chain(*(product_with_possible_empty_iterable(alphabet, repeat=i) for i in range(max_seq_len)))
+    char_set = char_set or [()]
+    for d in range(depth):
+        middle = product(alphabet, repeat=d)
+        for m in middle:
+            for s in state_cover:
+                for c in char_set:
+                    yield s + m + c
 
 
-def second_phase_test_case_generator(hyp, alphabet, difference, middle):
+def second_phase_it(hyp, alphabet, difference, depth):
     """
     Return an iterator that generates all possible sequences for the second phase of the Wp-method.
     Args:
         hyp: hypothesis automaton
         alphabet: input alphabet
         difference: set of sequences that are in the transition cover but not in the state cover
-        middle: iterator that generates all possible sequences of length upto from the given alphabet
+        depth: maximum length of middle part
     """
     state_mapping = {}
-    for t, mid in product_with_possible_empty_iterable(difference, middle):
-        _ = hyp.execute_sequence(hyp.initial_state, t + mid)
-        state = hyp.current_state
-        if state not in state_mapping:
-            state_mapping[state] = state_characterization_set(hyp, alphabet, state)
+    for d in range(depth):
+        middle = product(alphabet, repeat=d)
+        for mid in middle:
+            for t in difference:
+                _ = hyp.execute_sequence(hyp.initial_state, t + mid)
+                state = hyp.current_state
+                if state not in state_mapping:
+                    state_mapping[state] = state_characterization_set(hyp, alphabet, state)
 
-        for sm in state_mapping[state]:
-            yield t + mid + sm
+                for sm in state_mapping[state]:
+                    yield t + mid + sm
 
 
 class WpMethodEqOracle(Oracle):
@@ -63,21 +73,6 @@ def __init__(self, alphabet: list, sul: SUL, max_number_of_states=4):
         self.m = max_number_of_states
         self.cache = set()
 
-    def test_sequance(self, hypothesis, seq_under_test):
-        self.reset_hyp_and_sul(hypothesis)
-
-        for ind, letter in enumerate(seq_under_test):
-            out_hyp = hypothesis.step(letter)
-            out_sul = self.sul.step(letter)
-            self.num_steps += 1
-
-            if out_hyp != out_sul:
-                self.sul.post()
-                return seq_under_test[: ind + 1]
-        self.cache.add(seq_under_test)
-
-        return None
-
     def find_cex(self, hypothesis):
         if not hypothesis.characterization_set:
             hypothesis.characterization_set = hypothesis.compute_characterization_set()
@@ -90,31 +85,27 @@ def find_cex(self, hypothesis):
 
         state_cover = set(state.prefix for state in hypothesis.states)
         difference = transition_cover.difference(state_cover)
-
-        # two views of the same iterator
-        middle_1, middle_2 = tee(i_star(self.alphabet, self.m - hypothesis.size + 1), 2)
-
+        depth = self.m + 1 - len(hypothesis.states)
         # first phase State Cover * Middle * Characterization Set
-        state_cover = state_cover or [()]
-        char_set = hypothesis.characterization_set or [()]
-
-        for sc in state_cover:
-            for m in middle_1:
-                for cs in char_set:
-                    test_seq = sc + m + cs
-                    if test_seq not in self.cache:
-                        counterexample = self.test_sequance(hypothesis, test_seq)
-                        if counterexample:
-                            return counterexample
+        first_phase = first_phase_it(self.alphabet, state_cover, depth, hypothesis.characterization_set)
 
         # second phase (Transition Cover - State Cover) * Middle * Characterization Set
         # of the state that the prefix leads to
-        second_phase = second_phase_test_case_generator(hypothesis, self.alphabet, difference, middle_2)
-
-        for test_seq in second_phase:
-            if test_seq not in self.cache:
-                counterexample = self.test_sequance(hypothesis, test_seq)
-                if counterexample:
-                    return counterexample
+        second_phase = second_phase_it(hypothesis, self.alphabet, difference, depth)
+        test_suite = chain(first_phase, second_phase)
+
+        for seq in test_suite:
+            if seq not in self.cache:
+                self.reset_hyp_and_sul(hypothesis)
+
+                for ind, letter in enumerate(seq):
+                    out_hyp = hypothesis.step(letter)
+                    out_sul = self.sul.step(letter)
+                    self.num_steps += 1
+
+                    if out_hyp != out_sul:
+                        self.sul.post()
+                        return seq[: ind + 1]
+                self.cache.add(seq)
 
         return None

aalpy/oracles/WpMethodOther.py

@@ -0,0 +1,119 @@
+from itertools import chain, tee, product
+
+from aalpy.base.Oracle import Oracle
+from aalpy.base.SUL import SUL
+from aalpy.utils.HelperFunctions import product_with_possible_empty_iterable
+
+
+def state_characterization_set(hypothesis, alphabet, state):
+    """
+    Return a list of sequences that distinguish the given state from all other states in the hypothesis.
+    Args:
+        hypothesis: hypothesis automaton
+        alphabet: input alphabet
+        state: state for which to find distinguishing sequences
+    """
+    result = []
+    for i in range(len(hypothesis.states)):
+        if hypothesis.states[i] == state:
+            continue
+        seq = hypothesis.find_distinguishing_seq(state, hypothesis.states[i], alphabet)
+        if seq:
+            result.append(tuple(seq))
+    return result
+
+
+def i_star(alphabet, max_seq_len):
+    """
+    Return an iterator that generates all possible sequences of length upto from the given alphabet.
+    Args:
+        alphabet: input alphabet
+        max_seq_len: maximum length of the sequences
+    """
+    return chain(*(product_with_possible_empty_iterable(alphabet, repeat=i) for i in range(max_seq_len)))
+
+
+def second_phase_test_case_generator(hyp, alphabet, difference, middle):
+    """
+    Return an iterator that generates all possible sequences for the second phase of the Wp-method.
+    Args:
+        hyp: hypothesis automaton
+        alphabet: input alphabet
+        difference: set of sequences that are in the transition cover but not in the state cover
+        middle: iterator that generates all possible sequences of length upto from the given alphabet
+    """
+    state_mapping = {}
+    for t, mid in product_with_possible_empty_iterable(difference, middle):
+        _ = hyp.execute_sequence(hyp.initial_state, t + mid)
+        state = hyp.current_state
+        if state not in state_mapping:
+            state_mapping[state] = state_characterization_set(hyp, alphabet, state)
+
+        for sm in state_mapping[state]:
+            yield t + mid + sm
+
+
+class WpMethodOtherEqOracle(Oracle):
+    """
+    Implements the Wp-method equivalence oracle.
+    """
+
+    def __init__(self, alphabet: list, sul: SUL, max_number_of_states=4):
+        super().__init__(alphabet, sul)
+        self.m = max_number_of_states
+        self.cache = set()
+
+    def test_sequance(self, hypothesis, seq_under_test):
+        self.reset_hyp_and_sul(hypothesis)
+
+        for ind, letter in enumerate(seq_under_test):
+            out_hyp = hypothesis.step(letter)
+            out_sul = self.sul.step(letter)
+            self.num_steps += 1
+
+            if out_hyp != out_sul:
+                self.sul.post()
+                return seq_under_test[: ind + 1]
+        self.cache.add(seq_under_test)
+
+        return None
+
+    def find_cex(self, hypothesis):
+        if not hypothesis.characterization_set:
+            hypothesis.characterization_set = hypothesis.compute_characterization_set()
+
+        transition_cover = set(
+            state.prefix + (letter,)
+            for state in hypothesis.states
+            for letter in self.alphabet
+        )
+
+        state_cover = set(state.prefix for state in hypothesis.states)
+        difference = transition_cover.difference(state_cover)
+
+        # two views of the same iterator
+        middle_1, middle_2 = tee(i_star(self.alphabet, self.m - hypothesis.size + 1), 2)
+
+        # first phase State Cover * Middle * Characterization Set
+        state_cover = state_cover or [()]
+        char_set = hypothesis.characterization_set or [()]
+
+        for sc in state_cover:
+            for m in middle_1:
+                for cs in char_set:
+                    test_seq = sc + m + cs
+                    if test_seq not in self.cache:
+                        counterexample = self.test_sequance(hypothesis, test_seq)
+                        if counterexample:
+                            return counterexample
+        # second phase (Transition Cover - State Cover) * Middle * Characterization Set
+        # of the state that the prefix leads to
+        second_phase = second_phase_test_case_generator(hypothesis, self.alphabet, difference, middle_2)
+
+        for test_seq in second_phase:
+            if test_seq not in self.cache:
+                counterexample = self.test_sequance(hypothesis, test_seq)
+                if counterexample:
+                    return counterexample
+
+        return None