Merge pull request #156 from sandialabs/hmm_app

Updating application tests

Author: whart222

conin/hmm/hmm_application.py

@@ -14,6 +14,7 @@ def __init__(self, name="unknown"):
         self.name = name
 
         self._hidden_markov_model = None
+        self._simulations = None
 
         # Applicaton data used to initialize the HMM from simulations
         self._transition_prior = (None,)  # Nonzero values
@@ -29,6 +30,14 @@ def hidden_markov_model(self):
     def hidden_markov_model(self, hidden_markov_model):
         self._hidden_markov_model = hidden_markov_model
 
+    @property
+    def simulations(self):
+        return self._simulations
+
+    @simulations.setter
+    def simulations(self, simulations):
+        self._simulations = simulations
+
     def create_chmm(self, constraint_type=None):
         chmm = ConstrainedHiddenMarkovModel(hmm=self.hidden_markov_model)
         if constraint_type == "oracle":
@@ -45,6 +54,7 @@ def initialize(self, *args, **kwargs):
         """
         pass
 
+    # TODO - return an error if these methods are not defined
     def run_simulations(
         self, *, num=1, debug=False, with_observations=False, seed=None
     ):
@@ -54,40 +64,31 @@ def run_simulations(
 
         This method is defined by the application developer, and it provides a
         strategy for expressing domain knowledge regarding feasible hidden states.
+
+        This method returns the simulations generated
         """
-        return None
+        pass
 
     def initialize_hmm_from_simulations(
         self,
         *,
-        num=100,
-        debug=False,
-        seed=None,
         start_tolerance=None,
         transition_tolerance=None,
         emission_tolerance=None,
-        simulation_args=None,
+        simulations=None,
     ):
         assert (
             self._hidden_states is not None
         ), "HMMApplication.create_hmm_from_simulations must be run after the initialize() method is executed"
-        if simulation_args is None:
-            simulation_args = {}
-        simulation_args["num"] = num
-        simulation_args["debug"] = debug
-        simulation_args["seed"] = seed
-        simulation_args["with_observations"] = True
-        simulations = self.run_simulations(**simulation_args)
-        if debug:
-            for sim in simulations:
-                print("TSIM", sim.observations, sim.hidden)
 
+        if simulations is not None:
+            self.simulations = simulations
         assert (
-            simulations is not None
-        ), f"HMMApplication.create_hmm_from_simulations - Method run_simulations() has not been defined for the {self.name} application"
+            self.simulations is not None
+        ), f"HMMApplication.create_hmm_from_simulations - No simulations specified"
 
         self.hidden_markov_model = learning.supervised_learning(
-            simulations=simulations,
+            simulations=self.simulations,
             hidden_states=self._hidden_states,
             observable_states=self._observable_states,
             start_tolerance=start_tolerance,
@@ -97,8 +98,10 @@ def initialize_hmm_from_simulations(
             emission_prior=self._emission_prior,
         )
 
+    # TODO - return an error if these methods are not defined
     def get_oracle_constraints(self):
         return []
 
+    # TODO - return an error if these methods are not defined
     def get_pyomo_constraints(self):
         return []

conin/hmm/tests/test_application_chmm.py

@@ -50,11 +50,10 @@ def initialize(
                 transition_probs=transition_probs,
                 emission_probs=emission_probs,
             )
-            self.hmm = hmm
+            self.hidden_markov_model = hmm
         else:
-            self.hmm = hmm
+            self.hidden_markov_model = hmm
         self.num_zeros = num_zeros
-        self.generate_oracle_constraints()
         self._hidden_states = {0, 1}
         self._observable_states = {0, 1}
         self.time = time
@@ -65,32 +64,36 @@ def run_simulations(
         if seed is not None:
             random.seed(seed)
         output = []
+        oracle_chmm = self.create_chmm("oracle")
+
         for n in range(num):
             res = munch.Munch()
-            hidden = self.oracle.generate_hidden(self.time)
+            hidden = oracle_chmm.generate_hidden(self.time)
             if with_observations:
-                observed = self.oracle.generate_observed_from_hidden(hidden)
+                observed = oracle_chmm.generate_observed_from_hidden(hidden)
             res = munch.Munch(hidden=hidden, index=n)
             if with_observations:
                 res.observed = observed
             output.append(res)
         return output
 
-    def generate_oracle_constraints(self):
+    def get_oracle_constraints(self):
         constraint = has_exact_number_of_occurences_constraint(
             val=0, count=self.num_zeros
         )
-        self.oracle.set_constraints([constraint])
+        return [constraint]
 
-    def generate_pyomo_constraints(self, *, M):
-        index_sets = list(M.hmm.x.index_set().subsets())
-        T = list(index_sets[0])
+    def get_pyomo_constraints(self):
+        @pyomo_constraint_fn()
+        def constraint(M, data):
+            index_sets = list(M.hmm.x.index_set().subsets())
+            T = list(index_sets[0])
 
-        M.num_zeros = pyo.Constraint(
-            expr=sum(M.hmm.x[t, 0] for t in T) == self.num_zeros
-        )
+            M.num_zeros = pyo.Constraint(
+                expr=sum(M.hmm.x[t, 0] for t in T) == data.num_zeros
+            )
 
-        return M
+        return [constraint]
 
 
 @pytest.fixture
@@ -111,24 +114,29 @@ def app():
     return app
 
 
-class XTest_Application_CHMM:
+@pytest.fixture
+def oracle_chmm(app):
+    return app.create_chmm("oracle")
+
+
+class Test_Application_CHMM:
     def test_hmm(self, app):
-        assert app.hmm.transition_mat == [[0.6, 0.4], [0.4, 0.6]]
-        assert app.hmm.emission_mat == [[0.7, 0.3], [0.3, 0.7]]
-        assert app.hmm.start_vec == [0.5, 0.5]
+        assert app.hidden_markov_model.transition_mat == [[0.6, 0.4], [0.4, 0.6]]
+        assert app.hidden_markov_model.emission_mat == [[0.7, 0.3], [0.3, 0.7]]
+        assert app.hidden_markov_model.start_vec == [0.5, 0.5]
         assert app._hidden_states == {0, 1}
         assert app._observable_states == {0, 1}
         assert app.time == 20
 
     def test_oracle_type(self, app):
-        assert isinstance(app.oracle, type(Oracle_CHMM()))
+        chmm = app.create_chmm("oracle")
+        assert isinstance(chmm.chmm, Oracle_CHMM)
+        assert app.hidden_markov_model == chmm.hidden_markov_model
 
     def test_algebraic_type(self, app):
-        assert isinstance(app.algebraic, type(PyomoAlgebraic_CHMM()))
-
-    def test_hmm_equality(self, app):
-        assert app.hmm == app.oracle.hmm
-        assert app.hmm == app.algebraic.hmm
+        chmm = app.create_chmm("pyomo")
+        assert isinstance(chmm.chmm, PyomoAlgebraic_CHMM)
+        assert app.hidden_markov_model == chmm.hidden_markov_model
 
     def test_hmm_equality_setter(self, app):
         hmm = HiddenMarkovModel()
@@ -147,45 +155,49 @@ def test_hmm_equality_setter(self, app):
                 (1, 1): 0.6,
             },
         )
-        app.hmm = hmm
-        assert app.hmm == app.oracle.hmm
-        assert app.hmm == app.algebraic.hmm
+        app.hidden_markov_model = hmm
+        chmm = app.create_chmm("oracle")
+        assert app.hidden_markov_model == chmm.hidden_markov_model
+        chmm = app.create_chmm("pyomo")
+        assert app.hidden_markov_model == chmm.hidden_markov_model
 
     def test_get_internal_hmm(self, app):
-        assert app.get_internal_hmm().transition_mat == [
+        repn = app.hidden_markov_model.repn
+        assert repn.transition_mat == [
             [0.6, 0.4],
             [0.4, 0.6],
         ]
-        assert app.get_internal_hmm().emission_mat == [[0.7, 0.3], [0.3, 0.7]]
-        assert app.get_internal_hmm().start_vec == [0.5, 0.5]
+        assert repn.emission_mat == [[0.7, 0.3], [0.3, 0.7]]
+        assert repn.start_vec == [0.5, 0.5]
 
-    # CLM: This is a random test. Is that okay? -- I could also set a seed
     def test_run_simulations(self, app):
         seed = 1
         num_simulations = 5
         simulations = app.run_simulations(
             num=num_simulations, with_observations=True, seed=seed
         )
+
+        oracle_chmm = app.create_chmm("oracle")
         assert len(simulations) == num_simulations
         for i in range(num_simulations):
             assert len(simulations[i].hidden) == app.time
             assert len(simulations[i].observed) == app.time
-            assert app.oracle.is_feasible(simulations[i].hidden)
+            assert oracle_chmm.is_feasible(simulations[i].hidden)
 
-    def test_oracle_is_feasible(self, app):
+    def test_oracle_is_feasible(self, oracle_chmm):
         seq1 = [0] * 9
         seq2 = [0] * 10
         seq3 = [0] * 11
-        assert not app.oracle.is_feasible(seq1)
-        assert app.oracle.is_feasible(seq2)
-        assert not app.oracle.is_feasible(seq3)
+        assert not oracle_chmm.is_feasible(seq1)
+        assert oracle_chmm.is_feasible(seq2)
+        assert not oracle_chmm.is_feasible(seq3)
 
     # This assumes that the internal logic is correct, and is really just
     def test_initalize_hmm_from_simulations(self, app):
-        app.initialize_hmm_from_simulations(num=7)
-        assert app.hmm == app.oracle.hmm
-        assert app.hmm == app.algebraic.hmm
-        assert app.hmm.transition_mat != [
+        simulations = app.run_simulations(with_observations=True)
+        app.initialize_hmm_from_simulations(simulations=simulations)
+        repn = app.hidden_markov_model.repn
+        assert repn.transition_mat != [
             [0.6, 0.4],
             [0.4, 0.6],
         ]  # Just checks that it's updated