MNT: implementing back the last version of MonteCarlo class prior to …

…conflict solving
RocketPy-Team · Gui-FernandesBR · Mar 14, 2025 · Feb 10, 2025 · Feb 10, 2025 · Feb 10, 2025
commit fc38e3a730c9d053d575f4f19ba0ade4021643a8
@@ -218,50 +218,19 @@
         running the simulation again with `append=False`.
         """
         self._export_config = kwargs
-        # Create data files for inputs, outputs and error logging
-        open_mode = "a" if append else "w"
-        input_file = open(self._input_file, open_mode, encoding="utf-8")
-        output_file = open(self._output_file, open_mode, encoding="utf-8")
-        error_file = open(self._error_file, open_mode, encoding="utf-8")
-
-        # initialize counters
         self.number_of_simulations = number_of_simulations
         self._initial_sim_idx = self.num_of_loaded_sims if append else 0
 
         _SimMonitor.reprint("Starting Monte Carlo analysis")
 
         self.__setup_files(append)
-        try:
-            while self.__iteration_count < self.number_of_simulations:
-                self.__run_single_simulation(input_file, output_file)
-        except KeyboardInterrupt:
-            print("Keyboard Interrupt, files saved.")
-            error_file.write(
-                json.dumps(
-                    self._inputs_dict, cls=RocketPyEncoder, **self._export_config
-                )
-                + "\n"
-            )
-            self.__close_files(input_file, output_file, error_file)
-        except Exception as error:
-            print(f"Error on iteration {self.__iteration_count}: {error}")
-            error_file.write(
-                json.dumps(
-                    self._inputs_dict, cls=RocketPyEncoder, **self._export_config
-                )
-                + "\n"
-            )
-            self.__close_files(input_file, output_file, error_file)
-            raise error
-        finally:
-            self.total_wall_time = time() - self.__start_time
 
         if parallel:
             self.__run_in_parallel(n_workers)
        else:
            self.__run_in_serial()

        self.__terminate_simulation()

    def __setup_files(self, append):
        """
@@ -278,23 +247,23 @@
        None
        """
        # Create data files for inputs, outputs and error logging
        open_mode = "r+" if append else "w+"

        try:
            with open(self._input_file, open_mode, encoding="utf-8") as input_file:
                idx_i = len(input_file.readlines())
            with open(self._output_file, open_mode, encoding="utf-8") as output_file:
                idx_o = len(output_file.readlines())
            with open(self._error_file, open_mode, encoding="utf-8"):
                pass

            if idx_i != idx_o and not append:
                warnings.warn(
                    "Input and output files are not synchronized", UserWarning
                )

        except OSError as error:
            raise OSError(f"Error creating files: {error}") from error

    def __run_in_serial(self):
        """
@@ -304,38 +273,38 @@
        -------
        None
        """
        sim_monitor = _SimMonitor(
            initial_count=self._initial_sim_idx,
            n_simulations=self.number_of_simulations,
            start_time=time(),
        )
        try:
            while sim_monitor.keep_simulating():
                sim_monitor.increment()

                flight = self.__run_single_simulation()
                inputs_json = self.__evaluate_flight_inputs(sim_monitor.count)
                outputs_json = self.__evaluate_flight_outputs(flight, sim_monitor.count)

                with open(self.input_file, "a", encoding="utf-8") as f:
                    f.write(inputs_json)
                with open(self.output_file, "a", encoding="utf-8") as f:
                    f.write(outputs_json)

                sim_monitor.print_update_status(sim_monitor.count)

            sim_monitor.print_final_status()

        except KeyboardInterrupt:
            _SimMonitor.reprint("Keyboard Interrupt, files saved.")
            with open(self._error_file, "a", encoding="utf-8") as f:
                f.write(inputs_json)

        except Exception as error:
            _SimMonitor.reprint(f"Error on iteration {sim_monitor.count}: {error}")
            with open(self._error_file, "a", encoding="utf-8") as f:
                f.write(inputs_json)
            raise error

    # pylint: disable=too-many-statements
    def __run_in_parallel(self, n_workers=None):
@@ -352,30 +321,30 @@
        -------
        None
        """
        if n_workers is None or n_workers > os.cpu_count():
            n_workers = os.cpu_count()

        if n_workers < 2:
            raise ValueError("Number of workers must be at least 2 for parallel mode.")

        _SimMonitor.reprint(f"Running Monte Carlo simulation with {n_workers} workers.")

        multiprocess, managers = _import_multiprocess()

        with _create_multiprocess_manager(multiprocess, managers) as manager:
            mutex = manager.Lock()
            simulation_error_event = manager.Event()
            sim_monitor = manager._SimMonitor(
                initial_count=self._initial_sim_idx,
                n_simulations=self.number_of_simulations,
                start_time=time(),
            )

            processes = []
            seeds = np.random.SeedSequence().spawn(n_workers)

            for seed in seeds:
                sim_producer = multiprocess.Process(
                    target=self.__sim_producer,
                    args=(
                        seed,
@@ -384,33 +353,33 @@
                        simulation_error_event,
                    ),
                )
                processes.append(sim_producer)
                sim_producer.start()

            try:
                for sim_producer in processes:
                    sim_producer.join()

                # Handle error from the child processes
                if simulation_error_event.is_set():
                    raise RuntimeError(
                        "An error occurred during the simulation. \n"
                        f"Check the logs and error file {self.error_file} "
                        "for more information."
                    )

                sim_monitor.print_final_status()

            # Handle error from the main process
            # pylint: disable=broad-except
            except (Exception, KeyboardInterrupt) as error:
                simulation_error_event.set()

                for sim_producer in processes:
                    sim_producer.join()

                if not isinstance(error, KeyboardInterrupt):
                    raise error

    def __sim_producer(self, seed, sim_monitor, mutex, error_event):
        """Simulation producer to be used in parallel by multiprocessing.
@@ -426,49 +395,49 @@
        error_event : multiprocess.Event
            Event signaling an error occurred during the simulation.
        """
        try:
            # Ensure Processes generate different random numbers
            self.environment._set_stochastic(seed)
            self.rocket._set_stochastic(seed)
            self.flight._set_stochastic(seed)

            while sim_monitor.keep_simulating():
                sim_idx = sim_monitor.increment() - 1

                flight = self.__run_single_simulation()
                inputs_json = self.__evaluate_flight_inputs(sim_idx)
                outputs_json = self.__evaluate_flight_outputs(flight, sim_idx)

                try:
                    mutex.acquire()
                    if error_event.is_set():
                        sim_monitor.reprint(
                            "Simulation Interrupt, files from simulation "
                            f"{sim_idx} saved."
                        )
                        with open(self.error_file, "a", encoding="utf-8") as f:
                            f.write(inputs_json)

                        break

                    with open(self.input_file, "a", encoding="utf-8") as f:
                        f.write(inputs_json)
                    with open(self.output_file, "a", encoding="utf-8") as f:
                        f.write(outputs_json)

                    sim_monitor.print_update_status(sim_idx)
                finally:
                    mutex.release()

        except Exception:  # pylint: disable=broad-except
            mutex.acquire()
            with open(self.error_file, "a", encoding="utf-8") as f:
                f.write(inputs_json)

            sim_monitor.reprint(f"Error on iteration {sim_idx}:")
            sim_monitor.reprint(traceback.format_exc())
            error_event.set()
            mutex.release()

    def __run_single_simulation(self):
        """Runs a single simulation and returns the inputs and outputs.
@@ -478,7 +447,7 @@
        Flight
            The flight object of the simulation.
        """
        return Flight(
            rocket=self.rocket.create_object(),
            environment=self.environment.create_object(),
            rail_length=self.flight._randomize_rail_length(),
@@ -501,7 +470,7 @@
        str
            A JSON compatible dictionary with the inputs of the simulation.
        """
        inputs_dict = dict(
            item
            for d in [
                self.environment.last_rnd_dict,
@@ -510,8 +479,8 @@
            ]
            for item in d.items()
        )
        inputs_dict["index"] = sim_idx
        return json.dumps(inputs_dict, cls=RocketPyEncoder) + "\n"

    def __evaluate_flight_outputs(self, flight, sim_idx):
        """Evaluates the outputs of a single flight simulation.
@@ -528,11 +497,11 @@
        str
            A JSON compatible dictionary with the outputs of the simulation.
        """
        outputs_dict = {
            export_item: getattr(flight, export_item)
            for export_item in self.export_list
        }
        outputs_dict["index"] = sim_idx

        if self.data_collector is not None:
            additional_exports = {}
@@ -543,9 +512,9 @@
                    raise ValueError(
                        f"An error was encountered running 'data_collector' callback {key}. "
                    ) from e
            outputs_dict = outputs_dict | additional_exports

        return json.dumps(outputs_dict, cls=RocketPyEncoder) + "\n"

    def __terminate_simulation(self):
        """
@@ -556,18 +525,11 @@
        None
        """
        # resave the files on self and calculate post simulation attributes
        self.input_file = self._input_file
        self.output_file = self._output_file
         self.error_file = self._error_file
 
         _SimMonitor.reprint(f"Results saved to {self._output_file}")
-        self.input_file.write(
-            json.dumps(self.inputs_dict, cls=RocketPyEncoder, **self._export_config)
-            + "\n"
-        )
-        self.output_file.write(
-            json.dumps(self.results, cls=RocketPyEncoder, **self._export_config) + "\n"
-        )
 
     def __check_export_list(self, export_list):
         """
@@ -1149,10 +1111,10 @@
    tuple
        Tuple containing the imported modules.
    """
    multiprocess = import_optional_dependency("multiprocess")
    managers = import_optional_dependency("multiprocess.managers")

    return multiprocess, managers


 def _create_multiprocess_manager(multiprocess, managers):
@@ -1172,17 +1134,17 @@
        Subclass of BaseManager with the necessary classes registered.
    """

    class MonteCarloManager(managers.BaseManager):
        """Custom manager for shared objects in the Monte Carlo simulation."""

        def __init__(self):
            super().__init__()
            self.register('Lock', multiprocess.Lock)
            self.register('Queue', multiprocess.Queue)
            self.register('Event', multiprocess.Event)
            self.register('_SimMonitor', _SimMonitor)

    return MonteCarloManager()


 class _SimMonitor:
@@ -1191,17 +1153,17 @@
    _last_print_len = 0

    def __init__(self, initial_count, n_simulations, start_time):
        self.initial_count = initial_count
        self.count = initial_count
        self.n_simulations = n_simulations
        self.start_time = start_time

    def keep_simulating(self):
        return self.count < self.n_simulations

    def increment(self):
        self.count += 1
        return self.count

    def print_update_status(self, sim_idx):
        """Prints a message on the same line as the previous one and replaces
@@ -1217,23 +1179,23 @@
        -------
        None
        """
        average_time = (time() - self.start_time) / (self.count - self.initial_count)
        estimated_time = int((self.n_simulations - self.count) * average_time)

        msg = f"Current iteration: {sim_idx:06d}"
        msg += f" | Average Time per Iteration: {average_time:.3f} s"
        msg += f" | Estimated time left: {estimated_time} s"

        _SimMonitor.reprint(msg, end="\r", flush=True)

    def print_final_status(self):
        """Prints the final status of the simulation."""
        print()
        msg = f"Completed {self.count - self.initial_count} iterations."
        msg += f" In total, {self.count} simulations are exported.\n"
        msg += f"Total wall time: {time() - self.start_time:.1f} s"

        _SimMonitor.reprint(msg, end="\n", flush=True)

    @staticmethod
    def reprint(msg, end="\n", flush=True):