Merge pull request #170 from nasa/feature/const_noise

Const load

src/progpy/predictors/monte_carlo.py

@@ -30,7 +30,8 @@ class MonteCarlo(Predictor):
 
     default_parameters = { 
         'n_samples': None,
-        'event_strategy': 'all'
+        'event_strategy': 'all',
+        'constant_noise': False
     }
 
     def predict(self, state: UncertainData, future_loading_eqn: Callable=None, events=None, **kwargs) -> PredictionResults:
@@ -64,6 +65,8 @@ def predict(self, state: UncertainData, future_loading_eqn: Callable=None, event
             Frequency at which results are saved (s)
         save_pts : list[float], optional
             Any additional savepoints (s) e.g., [10.1, 22.5]
+        constant_noise : bool, optional
+            If the same noise should be applied every step. Default: False
 
         Return
         ----------
@@ -147,10 +150,23 @@ def predict(self, state: UncertainData, future_loading_eqn: Callable=None, event
         outputs_all = []
         event_states_all = []
 
+        if params['constant_noise']:
+            # Save loads
+            process_noise = self.model['process_noise']
+            process_noise_dist = self.model.parameters.get('process_noise_dist', 'normal')
+
         # Perform prediction
         t0 = params.get('t0', 0)
         HORIZON = params.get('horizon', float('inf'))  # Save the horizon to be used later
         for x in state:
+            if params['constant_noise']:
+                # Calculate process noise
+                x_noise = self.model.apply_process_noise(x.copy(), 1)
+                x_noise = self.model.StateContainer({key: x_noise[key] - x[key] for key in x.keys()})
+
+                self.model['process_noise'] = x_noise
+                self.model['process_noise_dist'] = 'constant'
+
             first_output = self.model.output(x)
 
             time_of_event = {}
@@ -238,6 +254,11 @@ def predict(self, state: UncertainData, future_loading_eqn: Callable=None, event
             event_states_all.append(event_states)
             time_of_event_all.append(time_of_event)
             last_states.append(last_state)
+
+            # Reset noise
+            if params['constant_noise']:
+                self.model['process_noise'] = process_noise
+                self.model['process_noise_dist'] = process_noise_dist
 
         inputs_all = UnweightedSamplesPrediction(times_all, inputs_all)
         states_all = UnweightedSamplesPrediction(times_all, states_all)

src/progpy/utils/noise_functions.py

@@ -43,6 +43,12 @@ def no_measurement_noise(self, z):
 # ---------------------------
 
 
+def constant_process_noise(self, x, dt: float = 1):
+    noise = self.parameters['process_noise'].matrix
+    x.matrix = x.matrix + dt*noise
+    return x
+
+
 def triangular_process_noise(self, x, dt: float = 1):
     noise_mat = self.parameters['process_noise'].matrix
     noise = np.random.triangular(-1*noise_mat, 0, noise_mat, size=x.matrix.shape)
@@ -69,6 +75,7 @@ def no_process_noise(self, x, dt: float = 1) -> dict:
 
 
 process_noise_functions = {
+    'constant': constant_process_noise,
     'uniform': uniform_process_noise,
     'triangular': triangular_process_noise,
     'normal': normal_process_noise,

tests/test_base_models.py

@@ -506,6 +506,14 @@ def test_process_noise(self):
         # That key should be 0 (default)
         self.assertEqual(m.parameters['process_noise'][list(m.states)[-1]], 0)
 
+        # Const noise
+        m.parameters['process_noise_dist'] = 'constant'
+        x = m.StateContainer({key: 0 for key in m.states})
+        x = m.apply_process_noise(x)
+        for key in list(m.states)[:-1]:
+            self.assertAlmostEqual(x[key], 1.0)
+        self.assertAlmostEqual(x[list(m.states)[-1]], 0.0)
+
     def test_measurement_noise(self):
         self.__noise_test('measurement_noise', 'measurement_noise_dist', MockProgModel.outputs)