Added no reset support

z3gi/console.py

@@ -50,9 +50,10 @@ aut2suttype={
 if __name__ == '__main__':
     parser = argparse.ArgumentParser(description='Inference tool which can:\n'
                                                  '(1) passively learn from traces file\n'
-                                                 '(2) actively learn a system described by a .dot file\n'
-                                                 '(3) actively learn a scalable SUT (system) of a given size')
-    parser.add_argument('-m', '--mode', type=str, choices=['traces', 'dot', 'scalable'], required=True)
+                                                 '(2) actively learn a Mealy Machine system described by a .dot file\n'
+                                                 '(3) actively learn a Mealy Machine system described by a .dot file without resets\n'
+                                                 '(4) actively learn a scalable SUT (system) of a given size')
+    parser.add_argument('-m', '--mode', type=str, choices=['traces', 'dot', 'dotnorst', 'scalable'], required=True)
     parser.add_argument('-f', '--file', type=str, help='the file location to the dot/traces file')
     parser.add_argument('-a', '--aut', type=str, choices=list(model.defined_formalisms().keys()), required=True,
                         help='the type of automaton (formalism) described in the file or '
@@ -64,6 +65,9 @@ if __name__ == '__main__':
                                                           '(i.e. the time limit given to the solver to solve the constraints)')
 
     # test parameters
+    parser.add_argument('-mi', '--max_inputs', type=int, default=10, help='the max number of inputs executed on a '
+                                                                          'hypothesis before it is judged to be correct, '
+                                                                          'only considered for learning with no resets')
     parser.add_argument('-mt', '--max_tests', type=int, default=1000, help='the max number of tests executed on a '
                                                                            'hypothesis before it is judged to be correct')
     parser.add_argument('-rl', '--rand_length', type=int, default=5, help='the maximum length of the random sequence '
@@ -86,22 +90,29 @@ if __name__ == '__main__':
         traces = parse.extract_traces_from_file(args.file, formalism)
         (automaton, statistics) = alg.learn(learner, aut2testcls[aut_type], traces)
     else:
-        if args.mode == 'dot':
+        if args.mode == 'dotnorst':
             dot_file = args.file
             aut_to_learn = parse.build_automaton_from_dot(formalism, dot_file)
-            sut_to_learn = sut.get_simulation(aut_to_learn)
-        elif args.mode == 'scalable':
-            sut_class_name = args.sut_class
-            sut_size = args.size
-            sut_to_learn = sut.get_scalable_sut(sut_class_name, aut2suttype[aut_type], sut_size)
+            sut_to_learn = sut.get_no_rst_simulation(aut_to_learn)
+            max_inputs = args.max_inputs
+            (automaton, statistics) = alg.learn_no_reset(sut_to_learn, learner, max_inputs)
         else:
-            print("Invalid mode ", args.mode)
-            exit(1)
+            if args.mode == 'dot':
+                dot_file = args.file
+                aut_to_learn = parse.build_automaton_from_dot(formalism, dot_file)
+                sut_to_learn = sut.get_simulation(aut_to_learn)
+            elif args.mode == 'scalable':
+                sut_class_name = args.sut_class
+                sut_size = args.size
+                sut_to_learn = sut.get_scalable_sut(sut_class_name, aut2suttype[aut_type], sut_size)
+            else:
+                print("Invalid mode ", args.mode)
+                exit(1)
 
-        num_tests = args.max_tests
-        rand_test_length = args.rand_length
-        reset_prob = args.reset_prob
-        test_generator = aut2rwalkcls[aut_type](sut_to_learn, rand_test_length, reset_prob)
-        (automaton, statistics) = alg.learn_mbt(sut_to_learn, learner, test_generator, num_tests)
+            num_tests = args.max_tests
+            rand_test_length = args.rand_length
+            reset_prob = args.reset_prob
+            test_generator = aut2rwalkcls[aut_type](sut_to_learn, rand_test_length, reset_prob)
+            (automaton, statistics) = alg.learn_mbt(sut_to_learn, learner, test_generator, num_tests)
 
     print("Learned\n", automaton, "\nWith stats\n", statistics)
\ No newline at end of file

z3gi/learn/algorithm.py

@@ -4,14 +4,16 @@ from typing import List, Tuple, Union,cast
 from model import Automaton
 from learn import Learner
 from model.ra import Action
-from sut import StatsTracker, SUT
+from sut import StatsTracker, SUT, NoRstSUT
 from sut.scalable import ActionSignature
+from sut.simulation import get_no_rst_simulation
 from test import TestGenerator, Test
 import time
 
 __all__ = [
     "learn",
-    "learn_mbt"
+    "learn_mbt",
+    "learn_no_reset"
     ]
 
 
@@ -77,6 +79,42 @@ def learn(learner:Learner, test_type:type, traces: List[object]) -> Tuple[Automa
                     break
         return (model, statistics)
 
+
+import random
+rand = random.Random(0)
+
+def rand_sel(l:List):
+    return l[rand.randint(0, len(l)-1)]
+
+def learn_no_reset(sut:NoRstSUT, learner:Learner, max_inputs:int) -> Tuple[Union[Automaton, None], Statistics]:
+    """ takes a non-reseting SUL, a learner, a test generator, and a bound on the number of inputs and generates a model"""
+    trace = []
+    alpha = sut.input_interface()
+    seq = gen_blind_seq(sut)
+    statistics = Statistics()
+    trace.extend(sut.steps(seq))
+    learner.add(list(trace))
+    (hyp, definition) = learner.model()
+    done = False
+    while not done:
+        sim = get_no_rst_simulation(hyp)
+        sim.steps([inp for (inp,_) in trace])
+        done = True
+        for _ in range(0, max_inputs):
+            rand_inp = rand_sel(alpha)
+            out_sut = sut.step(rand_inp)
+            trace.append((rand_inp, out_sut))
+            out_hyp = sim.step(rand_inp)
+            if out_sut != out_hyp:
+                learner.add(list(trace))
+                (hyp, definition) = learner.model(old_definition=definition)
+                done = False
+        if hyp is None:
+            return None, None
+    statistics.inputs = len(trace) - max_inputs
+    return hyp, statistics
+
+
 def learn_mbt(sut:SUT, learner:Learner, test_generator:TestGenerator, max_tests:int, stats_tracker:StatsTracker=None) -> Tuple[Union[Automaton, None], Statistics]:
     """ takes learner, a test generator, and bound on the number of tests and generates a model"""
     next_test = gen_blind_test(sut)
@@ -162,4 +200,23 @@ def gen_blind_test(sut:SUT):
         else:
             raise Exception("Unrecognized type")
     obs = sut.run(seq)
-    return obs.to_test()
\ No newline at end of file
+    return obs.to_test()
+
+def gen_blind_seq(sut):
+    """generates a sequence covering all input elements in the sut interface"""
+    seq = []
+    for abs_inp in sut.input_interface():
+        cnt = 0
+        # if it's RA stuff
+        if isinstance(abs_inp, ActionSignature):
+            if abs_inp.num_params == 0:
+                val = None
+            else:
+                val = cnt
+                cnt += 1
+            seq.append(Action(abs_inp.label, val))
+        elif isinstance(abs_inp, str):
+            seq.append(abs_inp)
+        else:
+            raise Exception("Unrecognized type")
+    return seq
\ No newline at end of file

z3gi/sut/__init__.py

@@ -9,6 +9,7 @@ from test import MealyTest, AcceptorTest, IORATest
 __all__ = [
     "get_simulation",
     "get_scalable_sut",
+    "get_no_rst_simulation",
     "scalable_sut_classes"
     ]
 
@@ -134,6 +135,20 @@ class SUT(metaclass=ABCMeta):
         """Runs the list of inputs or input signatures comprising the input interface"""
         pass
 
+class NoRstSUT(metaclass=ABCMeta):
+    def steps(self, inputs:Symbol) -> Tuple[Symbol, Symbol]:
+        trace = []
+        for inp in inputs:
+            trace.append((inp, self.step(inp)))
+        return trace
+
+    def step(self, input:Symbol) -> Symbol:
+        pass
+
+    def input_interface(self) -> List[Symbol]:
+        """Runs the list of inputs or input signatures comprising the input interface"""
+        pass
+
 class SUTType(Enum):
     IORA = 1
     RA = 2
@@ -152,7 +167,7 @@ class SUTType(Enum):
 
 # some useful functions
 from sut.scalable import scalable_sut_classes, get_scalable_sut
-from sut.simulation import get_simulation
+from sut.simulation import get_simulation, get_no_rst_simulation
 
 
 class StatsTracker(object):

z3gi/sut/simulation.py

@@ -4,14 +4,15 @@ from typing import List, Type, Union
 from model import Acceptor, Transducer, Automaton
 from model.fa import MealyMachine, Symbol
 from . import SUT, MealyObservation, TransducerObservation, \
-    AcceptorObservation
+    AcceptorObservation, NoRstSUT
+
 
 class Simulation(SUT,metaclass=ABCMeta):
     pass
 
 
 class AcceptorSimulation(Simulation):
-    def __init__(self, model:Acceptor, sut_obs_type:Type[Union[AcceptorObservation]]):
+    def __init__(self, model:Acceptor, sut_obs_type:Type[AcceptorObservation]):
         self.model = model
         self.sut_obs_type = sut_obs_type
 
@@ -44,6 +45,33 @@ class MealyMachineSimulation(TransducerSimulation):
         #TODO remove non-deterministic behavior
         return list(sorted(self.model.input_labels()))
 
+class NoRstSimulation(NoRstSUT,metaclass=ABCMeta):
+    pass
+
+class NoRstMealyMachineSimulation(NoRstSimulation):
+    def __init__(self, model:MealyMachine):
+        self.model = model
+        self.crt_state = model.start_state()
+
+    def step(self, inp:Symbol):
+        transitions = self.model.transitions(self.crt_state, inp)
+        assert len(transitions) == 1
+        trans = transitions[0]
+        self.crt_state = trans.end_state
+        out = trans.output
+        return out
+
+    def input_interface(self):
+        return list(sorted(self.model.input_labels()))
+
+def get_no_rst_simulation(aut:Automaton) -> NoRstSimulation:
+    """builds a simulation for the model. The simulation acts like a deterministic sut."""
+    if isinstance(aut, MealyMachine):
+        return NoRstMealyMachineSimulation(aut)
+    else:
+        print("Simulation not yet supported for ", type(aut))
+        exit(0)
+
 # TODO replace not suported -> print exit by throwing an adequate exception
 def get_simulation(aut: Automaton) -> Simulation:
     """builds a simulation for the model. The simulation acts like a deterministic sut."""

z3gi/yan_benchmark.py

@@ -137,7 +137,7 @@ pdus = [ModelDesc("pdu" + str(i), "MealyMachine",
 b.add_experiment(biometric)
 #b.add_experiment(bankcards[1])
 # create a test description
-t_desc = TestDesc(max_tests=10000, max_k=3, rand_length=9)
+t_desc = TestDesc(max_tests=10000, max_k=3, rand_length=3)
 
 # give the smt timeout value (in ms)
 timeout = 60000