Refactor Teacher base classes

firstpass_teacher.py

@@ -6,31 +6,12 @@ import numpy as np
 
 import z3
 
-from teacher_base import TeacherBase
+from teacher_base import GurobiTeacherBase
 
 
-class FirstpassTeacher(TeacherBase):
-    def __init__(self, name: str,
-                 state_dim: int, perc_dim: int, ctrl_dim: int) -> None:
-        super().__init__()
-
-        self._gp_model = gp.Model(name)
-        # Old state variables
-        self._old_state = self._gp_model.addMVar(shape=state_dim, name='x', vtype=GRB.CONTINUOUS, lb=-np.inf, ub=np.inf)
-        # New state variables
-        self._new_state = self._gp_model.addMVar(shape=state_dim, name="x'", vtype=GRB.CONTINUOUS, lb=-np.inf, ub=np.inf)
-        # Perception variables
-        self._percept = self._gp_model.addMVar(shape=perc_dim, name='z', vtype=GRB.CONTINUOUS, lb=-np.inf, ub=np.inf)
-        # Control variables
-        self._control = self._gp_model.addMVar(shape=ctrl_dim, name='u', vtype=GRB.CONTINUOUS, lb=-np.inf, ub=np.inf)
-
-        # FIXME Replace hardcoded contraints for this particular example
-        self._add_system()
-        self._add_unsafe()
-
-    def set_old_state_bound(self, lb, ub) -> None:
-        self._old_state.lb = lb
-        self._old_state.ub = ub
+class FirstpassTeacher(GurobiTeacherBase):
+    def __init__(self) -> None:
+        super().__init__("firstpass", 2, 2, 2)
 
     def _add_system(self) -> None:
         # Controller
@@ -39,7 +20,7 @@ class FirstpassTeacher(TeacherBase):
         # Dynamics
         self._gp_model.addConstr(self._new_state == self._old_state + self._control)
 
-    def _add_unsafe(self):
+    def _add_unsafe(self) -> None:
         old_state, new_state = self._old_state.tolist(), self._new_state.tolist()
         assert len(old_state) == len(new_state)
         m = self._gp_model
@@ -57,7 +38,12 @@ class FirstpassTeacher(TeacherBase):
             new_V += new_abs_xi
         m.addConstr(new_V >= old_V, name="Unstable")  # Tracking error is non-decreasing (UNSAFE)
 
-    def _add_candidate(self, coeff, intercept, radius) -> None:
+    def _add_candidate(self, candidate) -> None:
+        # TODO parse candidate
+        coeff, intercept, radius = candidate
+        assert coeff.shape == (self._percept.shape[0], self._old_state.shape[0])
+        assert intercept.shape == self._percept.shape
+
         m = self._gp_model
         x = self._old_state
         z = self._percept
@@ -69,66 +55,31 @@ class FirstpassTeacher(TeacherBase):
         # L2-norm objective
         m.setObjective(z_diff @ z_diff, GRB.MINIMIZE)
 
-    def add_constraints(self, *args) -> None:
-        # TODO replace hardcoded system with add_constraint
-        return super().add_constraints(*args)
-
-    def check(self, candidate) -> z3.CheckSatResult:
-        # TODO parse candidate
-        coeff, intercept, radius = candidate
-        assert coeff.shape == (self._percept.shape[0], self._old_state.shape[0])
-        assert intercept.shape == self._percept.shape
-
-        self._add_candidate(coeff, intercept, radius)
-
-        self._gp_model.write("%s.lp" % self._gp_model.ModelName)  # Dump optimization problem in LP format
-        self._gp_model.optimize()
-        if self._gp_model.status in [GRB.OPTIMAL, GRB.SUBOPTIMAL]:
-            return z3.sat
-        elif self._gp_model.status == GRB.INFEASIBLE:
-            return z3.unsat
-        else:
-            return z3.unknown
-
-    def model(self) -> z3.ModelRef:
-        if self._gp_model.status == GRB.OPTIMAL:
-            # TODO return satisfiable instance
-            m = self._gp_model
-            x = self._old_state
-            z = self._percept
-            print('Obj: %g, ObjBound: %g, x: %s, z: %s\n'
-                   % (m.objVal, m.objBound, x.x, z.x))
-        else:
-            raise z3.Z3Exception("model is not available")
-
-    def reason_unknown(self) -> str:
-        if self._gp_model.status == GRB.INF_OR_UNBD:
-            return '(model is infeasible or unbounded)'
-        elif self._gp_model.status == GRB.UNBOUNDED:
-            return '(model is unbounded)'
-        else:  # TODO other status code
-            return '(gurobi model status code %d)' % self._gp_model.status
-
 
 CAND0_A = np.array([[0.0, -1.0],
                     [0.0,  1.0]])
 CAND0_b = np.array([0.0, 0.0])
 CAND0_r = 2.0
+CAND0 = (CAND0_A, CAND0_b, CAND0_r)
 
 CAND1_A = np.array([[1.0, 0.0],
                    [0.0, 1.0]])
 CAND1_b = np.array([0.0, 0.0])
 CAND1_r = 2.0
+CAND1 = (CAND1_A, CAND1_b, CAND1_r)
 
 
 def test_firstpass_teacher():
-    teacher = FirstpassTeacher("firstpass", 2, 2, 2)
+    teacher = FirstpassTeacher()
     teacher.set_old_state_bound(lb=[6.0, 5.0], ub=[11.0, 10.0])
-    result = teacher.check((CAND1_A, CAND1_b, CAND1_r))
+    teacher.dump_model()
+    result = teacher.check(CAND1)
     print(result)
     if result == z3.sat:
         teacher.model()
-    elif result == z3.unknown:
+    elif result == z3.unsat:
+        print("candidate is verified")
+    else:
         print(teacher.reason_unknown())
 
 

teacher_base.py

 import abc
 
+import gurobipy as gp
+import numpy as np
 import z3
 
 
@@ -7,10 +9,6 @@ class TeacherBase(abc.ABC):
     def __init__(self) -> None:
         pass
 
-    @abc.abstractmethod
-    def add_constraints(self, *args) -> None:
-        raise NotImplementedError
-
     @abc.abstractmethod
     def check(self, candidate) -> z3.CheckSatResult:
         raise NotImplementedError
@@ -22,3 +20,74 @@ class TeacherBase(abc.ABC):
     @abc.abstractmethod
     def reason_unknown(self) -> str:
         raise NotImplementedError
+
+
+class GurobiTeacherBase(TeacherBase):
+    def __init__(self, name: str,
+                 state_dim: int, perc_dim: int, ctrl_dim: int) -> None:
+        super().__init__()
+
+        self._gp_model = gp.Model(name)
+        # Old state variables
+        self._old_state = self._gp_model.addMVar(shape=state_dim, name='x', vtype=gp.GRB.CONTINUOUS, lb=-np.inf, ub=np.inf)
+        # New state variables
+        self._new_state = self._gp_model.addMVar(shape=state_dim, name="x'", vtype=gp.GRB.CONTINUOUS, lb=-np.inf, ub=np.inf)
+        # Perception variables
+        self._percept = self._gp_model.addMVar(shape=perc_dim, name='z', vtype=gp.GRB.CONTINUOUS, lb=-np.inf, ub=np.inf)
+        # Control variables
+        self._control = self._gp_model.addMVar(shape=ctrl_dim, name='u', vtype=gp.GRB.CONTINUOUS, lb=-np.inf, ub=np.inf)
+
+        # FIXME Replace hardcoded contraints for this particular example
+        self._add_system()
+        self._add_unsafe()
+
+    @abc.abstractmethod
+    def _add_system(self) -> None:
+        raise NotImplementedError
+
+    @abc.abstractmethod
+    def _add_unsafe(self) -> None:
+        raise NotImplementedError
+
+    @abc.abstractmethod
+    def _add_candidate(candidate) -> None:
+        raise NotImplementedError
+
+    def set_old_state_bound(self, lb, ub) -> None:
+        self._old_state.lb = lb
+        self._old_state.ub = ub
+
+    def check(self, candidate) -> z3.CheckSatResult:
+        self._add_candidate(candidate)
+        self._gp_model.optimize()
+        if self._gp_model.status in [gp.GRB.OPTIMAL, gp.GRB.SUBOPTIMAL]:
+            return z3.sat
+        elif self._gp_model.status == gp.GRB.INFEASIBLE:
+            return z3.unsat
+        else:
+            return z3.unknown
+
+    def dump_model(self, basename:str = "") -> None:
+        """ Dump optimization problem in LP format """
+        if not basename:
+            basename = self._gp_model.ModelName
+        self._gp_model.write(basename + ".lp")
+
+    def model(self) -> z3.ModelRef:
+        if self._gp_model.status == gp.GRB.OPTIMAL:
+            # TODO return satisfiable instance
+            m = self._gp_model
+            x = self._old_state
+            z = self._percept
+            print('Obj: %g, ObjBound: %g, x: %s, z: %s\n'
+                   % (m.objVal, m.objBound, x.x, z.x))
+        else:
+            raise z3.Z3Exception("model is not available")
+
+    def reason_unknown(self) -> str:
+        if self._gp_model.status == gp.GRB.INF_OR_UNBD:
+            return '(model is infeasible or unbounded)'
+        elif self._gp_model.status == gp.GRB.UNBOUNDED:
+            return '(model is unbounded)'
+        else:  # TODO other status code
+            return '(gurobi model status code %d)' % self._gp_model.status