Mealy machine encoding

tests/mealy_testscenario.py

+
+
+import collections
+
+from z3gi.model.ra import Action
+
+
+
+"""Test scenarios contain a list of traces together with the number of locations and registers of the SUT generating
+ these traces.
+"""
+MealyTestCase = collections.namedtuple('MealyTestCase', 'description traces nr_states')
+
+sut_m1 = MealyTestCase("output '1' if the length of the trace so far is even, else output '0'",
+                        [(('a', '0'), ('a', '1'), ('a', '0'), ('a', '1')),
+                         (('a', '0'), ('a', '1'), ('a', '0'), ('b', '1')),
+                         (('a', '0'), ('a', '1'), ('b', '0'), ('a', '1')),
+                         (('a', '0'), ('b', '1'), ('a', '0'), ('a', '1')),
+                         (('b', '0'), ('a', '1'), ('a', '0'), ('a', '1')),
+                         (('a', '0'), ('a', '1'), ('b', '0'), ('b', '1')),
+                         (('a', '0'), ('b', '1'), ('b', '0'), ('a', '1')),
+                         (('b', '0'), ('b', '1'), ('a', '0'), ('a', '1')),
+                         (('a', '0'), ('b', '1'), ('b', '0'), ('b', '1')),
+                         (('b', '0'), ('b', '1'), ('b', '0'), ('a', '1')),
+                         (('b', '0'), ('b', '1'), ('b', '0'), ('b', '1')),
+                         ], 2)
\ No newline at end of file

tests/test_dfalearner.py

@@ -38,10 +38,7 @@ class DFALearnerTest(unittest.TestCase):
                              "Register automaton output doesn't correspond to observation {0}".format(str(trace)))
 
     def learn_model(self, test_scenario):
-        labels = set()
-        for label, _ in test_scenario.traces:
-            labels.add(label)
-        learner = FALearner(list(labels), encoder=DFAEncoder(),  verbose=True)
+        learner = FALearner(encoder=DFAEncoder(),  verbose=True)
         for trace in test_scenario.traces:
             learner.add(trace)
 

tests/test_mealylearner.py

+import unittest
+
+from encode.fa import MealyEncoder
+from tests.mealy_testscenario import *
+from z3gi.learn.fa import FALearner
+
+num_exp = 1
+
+
+class MealyLearnerTest(unittest.TestCase):
+    def setUp(self):
+        pass
+
+    def test_sut1(self):
+        self.check_scenario(sut_m1)
+
+    def check_scenario(self, test_scenario):
+        print("Scenario " + test_scenario.description)
+
+        for i in range(0, num_exp):
+            (succ, fa, model) = self.learn_model(test_scenario)
+
+            self.assertTrue(succ, msg="Automaton could not be inferred")
+            self.assertEqual(len(fa.states), test_scenario.nr_states,
+                             "Wrong number of states.")
+            self.assertEqual(len(fa.states), test_scenario.nr_states,
+                             "Wrong number of registers.")
+            exported = fa.export(model)
+            print("Learned model:  \n", exported)
+            self.assertEqual(len(exported.states()), test_scenario.nr_states,
+                             "Wrong number of states in exported model. ")
+            self.check_against_obs(exported, test_scenario)
+
+    def check_against_obs(self, learned_fa, test_scenario):
+        """Checks if the learned RA corresponds to the scenario observations"""
+        for trace, acc in test_scenario.traces:
+            self.assertEqual(learned_fa.accepts(trace), acc,
+                             "Register automaton output doesn't correspond to observation {0}".format(str(trace)))
+
+    def learn_model(self, test_scenario):
+        # inputs = set()
+        # outputs = set()
+        # for trace in test_scenario.traces:
+        #     inputs.update([i for (i, _) in trace])
+        #     outputs.update([o for (_, o) in trace])
+        learner = FALearner(encoder=MealyEncoder(),  verbose=True)
+        for trace in test_scenario.traces:
+            learner.add(trace)
+
+        min_states = test_scenario.nr_states - 1
+        max_states = test_scenario.nr_states + 1
+
+        result = learner._learn_model(min_states, max_states)  #
+        return result
\ No newline at end of file

z3gi/encode/fa.py

@@ -3,7 +3,7 @@ import itertools
 from define.fa import DFA, MealyMachine, Mapper
 from encode import Encoder
 from utils import Tree
-from model.fa import MealyMachine
+#from model.fa import MealyMachine
 import z3
 
 
@@ -20,7 +20,7 @@ class DFAEncoder(Encoder):
         self.labels.update(seq)
 
     def build(self, num_states):
-        dfa = DFA(self.labels, num_states)
+        dfa = DFA(list(self.labels), num_states)
         mapper = Mapper(dfa)
         constraints = self.axioms(dfa, mapper)
         constraints += self.node_constraints(dfa, mapper)
@@ -48,7 +48,7 @@ class DFAEncoder(Encoder):
         return constraints
 
 class MealyEncoder(Encoder):
-    def __init__(self, input_labels, output_labels):
+    def __init__(self):
         self.tree = Tree(itertools.count(0))
         self.cache = {}
         self.input_labels = set()
@@ -56,11 +56,48 @@ class MealyEncoder(Encoder):
 
 
     def add(self, trace):
-        _ = self.tree[trace]
-        self.input_labels.update([input_label for input_label in [i for (i, _) in trace]])
-        self.output_labels.update([output_label for output_label in [o for (_, o) in trace]])
+        input_sequence = [input for (input, _) in trace]
+        output_sequence = [output for (_, output) in trace]
+        for i in range(len(input_sequence)):
+            node = self.tree[input_sequence[:i+1]]
+            self.cache[node] = output_sequence[i]
+        self.input_labels.update(input_sequence)
+        self.output_labels.update(output_sequence)
 
 
     def build(self, num_states) -> (MealyMachine, z3.ExprRef):
-        return None
+        mm = MealyMachine(list(self.input_labels), list(self.output_labels), num_states)
+        mapper = Mapper(mm)
+        constraints = self.axioms(mm, mapper)
+        constraints += self.node_constraints(mm, mapper)
+        constraints += self.transition_constraints(mm, mapper)
+        return mm, constraints
+
+    def axioms(self, mm: MealyMachine, mapper: Mapper):
+        return []
+
+    def node_constraints(self, mm, mapper):
+        constraints = []
+        for node in self.cache:
+            parent = node.parent
+            input = None
+            for label in parent.children:
+                if parent.children[label] is node:
+                    input = label
+                    break
+            output = self.cache[node]
+            n = mapper.element(parent.id)
+            i = mm.inputs[input]
+            o = mm.outputs[output]
+            constraints.append(mm.output(mapper.map(n), i) == o)
+        return constraints
+
+    def transition_constraints(self, mm, mapper):
+        constraints = [mm.start == mapper.map(mapper.start)]
+        for node, input, child in self.tree.transitions():
+            n = mapper.element(node.id)
+            i = mm.inputs[input]
+            c = mapper.element(child.id)
+            constraints.append(mm.transition(mapper.map(n), i) == mapper.map(c))
+        return constraints
 

z3gi/learn/fa.py

@@ -14,7 +14,7 @@ from model import Automaton
 import define.fa
 
 class FALearner(Learner):
-    def __init__(self, labels, encoder, solver=None, verbose=False):
+    def __init__(self, encoder, solver=None, verbose=False):
         if not solver:
             solver = z3.Solver()
         self.solver = solver