ENH: Adaptive Monte Carlo via Convergence Criteria

CHANGELOG.md

@@ -31,7 +31,9 @@ and this project adheres to [Semantic Versioning](http://semver.org/).
 Attention: The newest changes should be on top -->
 
 ### Added
-- TST: Add acceptance tests for 3DOF flight simulation based on Bella Lui rocket [#914] (https://github.com/RocketPy-Team/RocketPy/pull/914_
+
+- ENH: Adaptive Monte Carlo via Convergence Criteria [#922] (https://github.com/RocketPy-Team/RocketPy/pull/922)
+- TST: Add acceptance tests for 3DOF flight simulation based on Bella Lui rocket [#914] (https://github.com/RocketPy-Team/RocketPy/pull/914)
 - ENH: Add background map auto download functionality to Monte Carlo plots [#896](https://github.com/RocketPy-Team/RocketPy/pull/896)
 - MNT: net thrust addition to 3 dof in flight class [#907] (https://github.com/RocketPy-Team/RocketPy/pull/907)
 - ENH: 3-dof lateral motion improvement [#883](https://github.com/RocketPy-Team/RocketPy/pull/883)

docs/notebooks/monte_carlo_analysis/monte_carlo_class_usage.ipynb

@@ -800,6 +800,28 @@
     ")"
    ]
   },
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "Alternatively, we can target an attribute using the method `MonteCarlo.simulate_convergence()` such that when the tolerance is met, the flight simulations would terminate early."
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "test_dispersion.simulate_convergence(\n",
+    "        target_attribute=\"apogee_time\",\n",
+    "        target_confidence=0.95,\n",
+    "        tolerance=0.5,  # in seconds\n",
+    "        max_simulations=1000,\n",
+    "        batch_size=50\n",
+    "    )"
+   ]
+  },
   {
    "attachments": {},
    "cell_type": "markdown",

rocketpy/simulation/monte_carlo.py

@@ -525,6 +525,69 @@ def estimate_confidence_interval(
 
         return res.confidence_interval
 
+    def simulate_convergence(
+        self,
+        target_attribute="apogee_time",
+        target_confidence=0.95,
+        tolerance=0.5,
+        max_simulations=1000,
+        batch_size=50,
+        parallel=True,
+        n_workers=4,
+    ):
+        """Run simulations cumulatively in batches until the confidence interval meets tolerance.
+
+        Parameters
+        ----------
+        target_attribute : str
+            The target attribute to track its convergence (e.g., "apogee", "apogee_time", etc.).
+        target_confidence : float, optional
+            The confidence level for the interval (between 0 and 1). Default is 0.95.
+        tolerance : float, optional
+            The desired width of the confidence interval in seconds, meters, or other units. Default is 0.5.
+        max_simulations : int, optional
+            The maximum number of simulations to run to avoid infinite loops. Default is 1000.
+        batch_size : int, optional
+            The number of simulations to run in each batch. Default is 50.
+        parallel : bool, optional
+            Whether to run simulations in parallel. Default is True.
+        n_workers : int, optional
+            The number of worker processes to use if running in parallel. Default is 8.
+
+        Returns
+        -------
+        confidence_interval_width : float
+            The confidence interval width when the simulation stopped for either meeting the tolerance or maximum number of simulations.
+        """
+
+        self.import_outputs(self.filename.with_suffix(".outputs.txt"))
+        confidence_interval = []
+
+        while (self.num_of_loaded_sims < max_simulations):
+            total_sims = min(self.num_of_loaded_sims + batch_size, max_simulations)
+
+            self.simulate(
+                number_of_simulations=total_sims,
+                append=True,
+                include_function_data=False,
+                parallel=parallel,
+                n_workers=n_workers,
+            )
+
+            self.import_outputs(self.filename.with_suffix(".outputs.txt"))
+
+            ci = self.estimate_confidence_interval(
+                attribute=target_attribute,
+                confidence_level=target_confidence,
+            )
+
+            confidence_interval.append(float(ci.high - ci.low))
+
+            if float(ci.high - ci.low) <= tolerance:
+                break
+
+        return confidence_interval
+
     def __evaluate_flight_inputs(self, sim_idx):
         """Evaluates the inputs of a single flight simulation.
 

tests/integration/simulation/test_monte_carlo.py

@@ -236,3 +236,31 @@ def invalid_data_collector(flight):
             monte_carlo_calisto.simulate(number_of_simulations=10, append=False)
     finally:
         _post_test_file_cleanup()
+
+
+@pytest.mark.slow
+def test_monte_carlo_simulate_convergence(monte_carlo_calisto):
+    """Tests the simulate_convergence method of the MonteCarlo class.
+
+    Parameters
+    ----------
+    monte_carlo_calisto_pre_loaded : MonteCarlo
+        The MonteCarlo object, this is a pytest fixture.
+    """
+    try:
+        ci_history = monte_carlo_calisto.simulate_convergence(
+            target_attribute="apogee",
+            target_confidence=0.95,
+            tolerance=5.0,
+            max_simulations=20,
+            batch_size=5,
+            parallel=False,
+        )
+
+        assert isinstance(ci_history, list)
+        print(ci_history)
+        assert all(isinstance(width, float) for width in ci_history)
+        assert len(ci_history) >= 1
+        assert monte_carlo_calisto.num_of_loaded_sims <= 20
+    finally:
+        _post_test_file_cleanup()