diff --git a/docs/user/stochastic.rst b/docs/user/stochastic.rst index 6e3376236..1fd8c2f56 100644 --- a/docs/user/stochastic.rst +++ b/docs/user/stochastic.rst @@ -289,3 +289,11 @@ better reflecting the inherent uncertainties in rocketry. .. note:: See the ``MonteCarlo`` class documentation for more information on how to run \ Monte Carlo simulations with stochastic objects. + +.. note:: + A whole run is fixed by ``MonteCarlo.simulate(random_seed=...)`` rather than + by seeding these models yourself. Each simulation takes its seed from its + own index, so simulation 7 draws the same inputs whether the run was serial + or split over any number of workers, and appending with the same seed + carries the same stream on. Without it a run draws fresh entropy and + reproduces nothing. diff --git a/rocketpy/simulation/monte_carlo.py b/rocketpy/simulation/monte_carlo.py index c2dcd4030..44f019119 100644 --- a/rocketpy/simulation/monte_carlo.py +++ b/rocketpy/simulation/monte_carlo.py @@ -18,6 +18,7 @@ import os import traceback import warnings +from copy import deepcopy from numbers import Real from pathlib import Path from time import time @@ -31,6 +32,7 @@ from rocketpy.prints.monte_carlo_prints import _MonteCarloPrints from rocketpy.simulation.flight import Flight from rocketpy.tools import ( + _seed_sequence_to_int, generate_monte_carlo_ellipses, generate_monte_carlo_ellipses_coordinates, import_optional_dependency, @@ -44,6 +46,57 @@ _SIMULATION_LOG_SUFFIX = ".txt" +def _root_seed_sequence(random_seed): + """The immutable root a run derives every simulation's seed from. + + A ``SeedSequence`` is rebuilt from its full state rather than used as + given, since ``spawn`` advances a counter the caller still holds. A + ``Generator`` is refused rather than read, because using a consume-on-use + object as an immutable seed cannot mean what it says. + """ + if isinstance(random_seed, np.random.SeedSequence): + return np.random.SeedSequence(**random_seed.state) + if isinstance(random_seed, (np.random.Generator, np.random.BitGenerator)): + raise TypeError( + f"random_seed must be an int, a sequence of non-negative integers, " + f"or a numpy.random.SeedSequence, not a " + f"{type(random_seed).__name__}. Pass the seed the generator was " + f"built from." + ) + return np.random.SeedSequence(random_seed) + + +def _root_state_of(root): + """A root as the four picklable values a worker can rebuild it from. + + Sent to each worker instead of the object, and instead of the list of + children, so a run of a million simulations costs four values. The entropy + is copied because a sequence one is kept by reference all the way from the + caller, who could otherwise still move every child by editing their list. + """ + return ( + deepcopy(root.entropy), + tuple(root.spawn_key), + root.pool_size, + root.n_children_spawned, + ) + + +def _seed_of_simulation(root_state, sim_idx): + """The seed for one simulation index, without spawning the ones before it. + + ``spawn`` derives child ``i`` by appending ``n_children_spawned + i`` to + the parent spawn key, so rebuilding that one child directly reproduces it + and any index can be reached from the four values above alone. + """ + entropy, spawn_key, pool_size, base = root_state + return np.random.SeedSequence( + entropy=entropy, + spawn_key=(*spawn_key, base + sim_idx), + pool_size=pool_size, + ) + + def _refuse_logs_this_run_cannot_write( input_file, output_file, error_file, export_config=None ): @@ -265,6 +318,8 @@ def simulate( append=False, parallel=False, n_workers=None, + *, + random_seed=None, **kwargs, ): """ @@ -284,6 +339,20 @@ def simulate( number of workers will be equal to the number of CPUs available. A minimum of 2 workers is required for parallel mode. Default is None. + random_seed : int, sequence of int or numpy.random.SeedSequence, optional + Fixes what every simulation draws. Simulation ``i`` takes the same + inputs whichever way the run was split up, so serial and parallel + results agree and the number of workers does not reach the + sampling. Keyword-only. Default is None, which draws fresh entropy + and reproduces nothing. + + Appending continues the same stream when the same seed is given + again, since an index maps to a seed and to nothing else. Nothing + here records the seed, so nothing here can tell you that a later + append was given the same one; that is #1075. + + A ``Generator`` or ``BitGenerator`` is refused rather than read. + Pass the seed it was built from. kwargs : dict Custom arguments for simulation export of the ``inputs`` file. Options are: @@ -317,6 +386,11 @@ def simulate( self._export_config = kwargs self.number_of_simulations = number_of_simulations self._initial_sim_idx = self.num_of_loaded_sims if append else 0 + # Validated here, before __setup_files truncates anything, so an + # unusable seed cannot cost a previous run its results. Kept as four + # picklable values rather than as the object, since a worker rebuilds + # any index from them. + self.__root_state = _root_state_of(_root_seed_sequence(random_seed)) # Before anything is opened: __setup_files truncates for append=False. _refuse_logs_this_run_cannot_write( @@ -413,14 +487,19 @@ def __run_in_serial(self): n_simulations=self.number_of_simulations, start_time=time(), ) + sim_idx = sim_monitor.count try: while sim_monitor.keep_simulating(): - sim_monitor.increment() + # Counted from zero, as the parallel path already does. The two + # named the same simulation differently: three of them wrote + # 1, 2, 3 here and 0, 1, 2 there. + sim_idx = sim_monitor.increment() - 1 inputs_json, outputs_json = "", "" + self.__seed_this_simulation(sim_idx) flight = self.__run_single_simulation() - inputs_json = self.__evaluate_flight_inputs(sim_monitor.count) - outputs_json = self.__evaluate_flight_outputs(flight, sim_monitor.count) + inputs_json = self.__evaluate_flight_inputs(sim_idx) + outputs_json = self.__evaluate_flight_outputs(flight, sim_idx) self._append_simulation_record(inputs_json, outputs_json) @@ -434,7 +513,7 @@ def __run_in_serial(self): f.write(inputs_json) except Exception as error: - print(f"Error on iteration {sim_monitor.count}: {error}") + print(f"Error on iteration {sim_idx}: {error}") with open(self._error_file, "a", encoding="utf-8") as f: f.write(inputs_json) raise error @@ -469,13 +548,14 @@ def __run_in_parallel(self, n_workers=None): ) processes = [] - seeds = np.random.SeedSequence().spawn(n_workers) - for seed in seeds: + # No seed per worker any more: every simulation takes its own from + # its index, so the workers are interchangeable and how many there + # are does not reach the sampling. + for _ in range(n_workers): sim_producer = multiprocess.Process( target=self.__sim_producer, args=( - seed, sim_monitor, mutex, simulation_error_event, @@ -517,13 +597,11 @@ def __validate_number_of_workers(self, n_workers): raise ValueError("Number of workers must be at least 2 for parallel mode.") return n_workers - def __sim_producer(self, seed, sim_monitor, mutex, error_event): # pylint: disable=too-many-statements + def __sim_producer(self, sim_monitor, mutex, error_event): """Simulation producer to be used in parallel by multiprocessing. Parameters ---------- - seed : int - The seed to set the random number generator. sim_monitor : _SimMonitor The simulation monitor object to keep track of the simulations. mutex : multiprocess.Lock @@ -532,15 +610,11 @@ def __sim_producer(self, seed, sim_monitor, mutex, error_event): # pylint: disa 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 inputs_json, outputs_json = "", "" + self.__seed_this_simulation(sim_idx) flight = self.__run_single_simulation() inputs_json = self.__evaluate_flight_inputs(sim_idx) outputs_json = self.__evaluate_flight_outputs(flight, sim_idx) @@ -580,6 +654,20 @@ def __sim_producer(self, seed, sim_monitor, mutex, error_event): # pylint: disa error_event.set() mutex.release() + def __seed_this_simulation(self, sim_idx): + """Reseed the three models from this index's own child of the root. + + Per index rather than per worker, which is what makes a simulation's + inputs the same however the run was split up. The child is split three + ways so the environment, rocket and flight draw independently instead + of sharing one stream. + """ + child = _seed_of_simulation(self.__root_state, sim_idx) + environment, rocket, flight = child.spawn(3) + self.environment._set_stochastic(_seed_sequence_to_int(environment)) + self.rocket._set_stochastic(_seed_sequence_to_int(rocket)) + self.flight._set_stochastic(_seed_sequence_to_int(flight)) + def __run_single_simulation(self): """Runs a single simulation and returns the inputs and outputs. diff --git a/rocketpy/stochastic/stochastic_model.py b/rocketpy/stochastic/stochastic_model.py index d42fb76c5..1dadb2f01 100644 --- a/rocketpy/stochastic/stochastic_model.py +++ b/rocketpy/stochastic/stochastic_model.py @@ -8,7 +8,7 @@ from rocketpy.mathutils.function import Function from rocketpy.stochastic.custom_sampler import CustomSampler -from ..tools import get_distribution +from ..tools import _seed_sequence_to_int, get_distribution def _names_as_spawn_key(input_names): @@ -41,6 +41,18 @@ def _format_number(value): return f"array of shape {np.shape(value)}" +def _seed_as_entropy(seed): + """A seed as something ``SeedSequence`` will take as entropy. + + A parallel run is handed a ``SeedSequence``, which it will not take. Any + other seed goes through untouched, so the stream an int reaches stays where + it was. + """ + if not isinstance(seed, np.random.SeedSequence): + return seed + return _seed_sequence_to_int(seed) + + def _sampler_seed(seed, input_names): """Derive a seed for one sampler, or for one group that shares a generator. @@ -54,10 +66,10 @@ def _sampler_seed(seed, input_names): # Sorted here rather than trusting the caller, so a future call site cannot # give one group two different seeds by listing its members another way. root = np.random.SeedSequence( - entropy=seed, spawn_key=_names_as_spawn_key(tuple(sorted(input_names))) + entropy=_seed_as_entropy(seed), + spawn_key=_names_as_spawn_key(tuple(sorted(input_names))), ) - words = root.generate_state(4, dtype=np.uint32) - return sum(int(word) << (32 * position) for position, word in enumerate(words)) + return _seed_sequence_to_int(root) # TODO: Stop using assert in production code. Use exceptions instead. diff --git a/rocketpy/tools.py b/rocketpy/tools.py index 0d7f1a74e..7f31f3e19 100644 --- a/rocketpy/tools.py +++ b/rocketpy/tools.py @@ -1377,6 +1377,17 @@ def euler313_to_quaternions(phi, theta, psi): return e0, e1, e2, e3 +def _seed_sequence_to_int(seed_sequence): + """Returns a ``SeedSequence`` as the 128-bit ``int`` it can be rebuilt from. + + Folded through ``generate_state`` rather than read off ``entropy``, since + the children of one root differ only by ``spawn_key``, and combined by + value so it does not depend on byte order. + """ + words = seed_sequence.generate_state(4, dtype=np.uint32) + return sum(int(word) << (32 * position) for position, word in enumerate(words)) + + def get_matplotlib_supported_file_endings(): """Gets the file endings supported by matplotlib. diff --git a/tests/unit/simulation/test_monte_carlo_parallel_runs.py b/tests/unit/simulation/test_monte_carlo_parallel_runs.py new file mode 100644 index 000000000..4ab0be440 --- /dev/null +++ b/tests/unit/simulation/test_monte_carlo_parallel_runs.py @@ -0,0 +1,32 @@ +import pytest + +from rocketpy.simulation.monte_carlo import MonteCarlo + + +@pytest.mark.parametrize("parallel", [False, True]) +def test_a_monte_carlo_run_finishes( + stochastic_environment, stochastic_calisto, stochastic_flight, tmp_path, parallel +): + # The parallel path hands each worker a SeedSequence rather than an int, and + # nothing else in the suite exercises that. A worker that dies on it is not + # reported, so this reads as a hang rather than as a failure. + # + # Built here rather than taken from the monte_carlo_calisto fixture, whose + # own filename is fixed, since `filename` is a plain attribute and the three + # working paths are settled when the object is constructed. + analysis = MonteCarlo( + filename=str(tmp_path / "study"), + environment=stochastic_environment, + rocket=stochastic_calisto, + flight=stochastic_flight, + ) + + analysis.simulate( + number_of_simulations=2, + append=False, + parallel=parallel, + n_workers=2 if parallel else None, + ) + + assert analysis.num_of_loaded_sims == 2 + assert str(tmp_path) in str(analysis.output_file) diff --git a/tests/unit/simulation/test_monte_carlo_seeding.py b/tests/unit/simulation/test_monte_carlo_seeding.py new file mode 100644 index 000000000..b820c5152 --- /dev/null +++ b/tests/unit/simulation/test_monte_carlo_seeding.py @@ -0,0 +1,157 @@ +import json +import os + +import numpy as np +import pytest + +from rocketpy.simulation.monte_carlo import ( + MonteCarlo, + _root_seed_sequence, + _root_state_of, + _seed_of_simulation, +) + + +def _sampled_inputs(analysis): + with open(analysis.input_file, "r", encoding="utf-8") as written: + rows = [json.loads(line) for line in written if line.strip()] + return {row["index"]: row for row in rows} + + +def _a_run(tmp_path, stem, models, *, parallel=False, workers=None, seed=None, count=4): + environment, rocket, flight = models + analysis = MonteCarlo( + filename=str(tmp_path / stem), + environment=environment, + rocket=rocket, + flight=flight, + ) + analysis.simulate( + number_of_simulations=count, + append=False, + parallel=parallel, + n_workers=workers, + random_seed=seed, + ) + return analysis + + +@pytest.fixture(name="models") +def _models(stochastic_environment, stochastic_calisto, stochastic_flight): + return stochastic_environment, stochastic_calisto, stochastic_flight + + +# --------------------------------------------------------------- the derivation + + +@pytest.mark.parametrize("seed", [42, None, [1, 2, 3], np.random.SeedSequence(7)]) +def test_a_simulation_gets_the_child_spawn_would_have_given_it(seed): + # The whole point of deriving one index directly: it has to be the same + # child, bit for bit, as spawning every index before it would produce. + root = _root_seed_sequence(seed) + state = _root_state_of(root) + spawned = np.random.SeedSequence(**root.state).spawn(6) + + for index, expected in enumerate(spawned): + assert np.array_equal( + expected.generate_state(4), + _seed_of_simulation(state, index).generate_state(4), + ) + + +def test_deriving_an_index_costs_nothing_for_the_ones_before_it(): + state = _root_state_of(_root_seed_sequence(42)) + + far = _seed_of_simulation(state, 1_000_000) + + assert far.spawn_key == (1_000_000,) + + +def test_two_indices_do_not_share_a_stream(): + state = _root_state_of(_root_seed_sequence(42)) + + first = _seed_of_simulation(state, 0).generate_state(4) + second = _seed_of_simulation(state, 1).generate_state(4) + + assert not np.array_equal(first, second) + + +# ------------------------------------------------------------------- the root + + +def test_a_caller_seed_sequence_is_not_consumed(): + given = np.random.SeedSequence(42) + + _root_seed_sequence(given).spawn(5) + + assert given.n_children_spawned == 0 + + +def test_a_caller_cannot_move_the_run_by_editing_the_list_it_passed(): + # SeedSequence keeps a sequence entropy by reference, so without a copy of + # its own the run would follow whatever the caller did to that list next. + given = [1, 2, 3] + state = _root_state_of(_root_seed_sequence(given)) + + before = _seed_of_simulation(state, 0).generate_state(4) + given[0] = 999 + + assert np.array_equal(_seed_of_simulation(state, 0).generate_state(4), before) + + +@pytest.mark.parametrize("given", [np.random.default_rng(42), np.random.PCG64(42)]) +def test_a_generator_is_refused_rather_than_read(given): + # Using a consume-on-use object as an immutable seed cannot mean what it + # says, so it is refused instead of quietly meaning something else. + with pytest.raises(TypeError, match="random_seed must be"): + _root_seed_sequence(given) + + +def test_an_unusable_seed_costs_the_previous_run_nothing(models, tmp_path): + analysis = _a_run(tmp_path, "study", models, seed=42, count=2) + kept = _sampled_inputs(analysis) + + with pytest.raises(TypeError): + analysis.simulate(2, append=False, random_seed=np.random.default_rng(1)) + + assert _sampled_inputs(analysis) == kept + + +# ------------------------------------------------------------- what a run gives + + +def test_one_seed_gives_one_set_of_inputs(models, tmp_path): + first = _a_run(tmp_path, "first", models, seed=42) + again = _a_run(tmp_path, "again", models, seed=42) + + assert _sampled_inputs(first) == _sampled_inputs(again) + + +def test_another_seed_gives_another_set(models, tmp_path): + # The control. Without this the test above passes on a run that ignores + # the seed entirely. + first = _a_run(tmp_path, "first", models, seed=42) + other = _a_run(tmp_path, "other", models, seed=7) + + assert _sampled_inputs(first) != _sampled_inputs(other) + + +@pytest.mark.skipif(os.cpu_count() < 4, reason="needs four workers to be four") +def test_an_index_gets_the_same_inputs_however_the_run_was_split(models, tmp_path): + # This is the guarantee. Splitting the work differently must not change + # what any one simulation drew. + serial = _a_run(tmp_path, "serial", models, seed=42) + two = _a_run(tmp_path, "two", models, parallel=True, workers=2, seed=42) + four = _a_run(tmp_path, "four", models, parallel=True, workers=4, seed=42) + + assert _sampled_inputs(serial) == _sampled_inputs(two) + assert _sampled_inputs(serial) == _sampled_inputs(four) + + +def test_an_appended_run_carries_the_same_stream_on(models, tmp_path): + whole = _a_run(tmp_path, "whole", models, seed=42, count=4) + + part = _a_run(tmp_path, "part", models, seed=42, count=2) + part.simulate(4, append=True, random_seed=42) + + assert _sampled_inputs(part) == _sampled_inputs(whole) diff --git a/tests/unit/simulation/test_monte_carlo_simulation_index.py b/tests/unit/simulation/test_monte_carlo_simulation_index.py new file mode 100644 index 000000000..7d2d6a52a --- /dev/null +++ b/tests/unit/simulation/test_monte_carlo_simulation_index.py @@ -0,0 +1,76 @@ +import json + +import pytest + +from rocketpy.simulation.monte_carlo import MonteCarlo + + +def _indices(analysis): + with open(analysis.output_file, "r", encoding="utf-8") as written: + return sorted(json.loads(line)["index"] for line in written if line.strip()) + + +def _a_study(tmp_path, stem, environment, rocket, flight): + return MonteCarlo( + filename=str(tmp_path / stem), + environment=environment, + rocket=rocket, + flight=flight, + ) + + +@pytest.mark.parametrize("parallel", [False, True]) +def test_a_run_numbers_its_simulations_from_zero( + stochastic_environment, stochastic_calisto, stochastic_flight, tmp_path, parallel +): + # Serial used to write 1, 2, 3 while parallel wrote 0, 1, 2, so the same + # simulation had two names depending on how the run was started. + analysis = _a_study( + tmp_path, + f"study-{parallel}", + stochastic_environment, + stochastic_calisto, + stochastic_flight, + ) + + analysis.simulate( + number_of_simulations=3, + append=False, + parallel=parallel, + n_workers=2 if parallel else None, + ) + + assert _indices(analysis) == [0, 1, 2] + + +def test_both_modes_agree_on_what_a_simulation_is_called( + stochastic_environment, stochastic_calisto, stochastic_flight, tmp_path +): + one = _a_study( + tmp_path, + "serial", + stochastic_environment, + stochastic_calisto, + stochastic_flight, + ) + other = _a_study( + tmp_path, "para", stochastic_environment, stochastic_calisto, stochastic_flight + ) + + one.simulate(number_of_simulations=3, append=False, parallel=False) + other.simulate(number_of_simulations=3, append=False, parallel=True, n_workers=2) + + assert _indices(one) == _indices(other) + + +def test_an_appended_run_carries_on_from_the_last_index( + stochastic_environment, stochastic_calisto, stochastic_flight, tmp_path +): + analysis = _a_study( + tmp_path, "study", stochastic_environment, stochastic_calisto, stochastic_flight + ) + + analysis.simulate(number_of_simulations=2, append=False) + analysis.simulate(number_of_simulations=4, append=True) + + assert _indices(analysis) == [0, 1, 2, 3] diff --git a/tests/unit/stochastic/test_seed_types.py b/tests/unit/stochastic/test_seed_types.py new file mode 100644 index 000000000..ba3d42583 --- /dev/null +++ b/tests/unit/stochastic/test_seed_types.py @@ -0,0 +1,81 @@ +import numpy as np +import pytest + +from rocketpy.stochastic.stochastic_model import ( + _names_as_spawn_key, + _sampler_seed, +) +from rocketpy.tools import _seed_sequence_to_int + + +def _a_worker_seed(index=0, workers=2): + # What MonteCarlo.__run_in_parallel spawns and hands to each worker, which + # passes it straight to environment/rocket/flight._set_stochastic. + return np.random.SeedSequence().spawn(workers)[index] + + +def test_the_seed_type_a_worker_is_handed_is_accepted(stochastic_calisto): + stochastic_calisto._set_stochastic(_a_worker_seed()) + + stochastic_calisto.create_object() + + +def test_a_parachute_derives_its_noise_seed_from_a_worker_seed( + stochastic_main_parachute, +): + stochastic_main_parachute._set_stochastic(_a_worker_seed()) + + assert stochastic_main_parachute.create_object().noise[2] is not None + + +def test_two_workers_do_not_share_a_sampler_stream(): + first, second = np.random.SeedSequence(7).spawn(2) + # They come off one root, so they carry the same entropy and differ only in + # spawn_key. Reading the entropy alone would put both on one stream. + assert first.entropy == second.entropy + + assert _sampler_seed(first, ("__list_choice__",)) != _sampler_seed( + second, ("__list_choice__",) + ) + + +def test_a_caller_seed_sequence_is_not_consumed(): + root = np.random.SeedSequence(42) + + _sampler_seed(root, ("__list_choice__",)) + + assert root.n_children_spawned == 0 + assert root.spawn(1)[0].spawn_key == (0,) + + +def test_the_same_seed_sequence_twice_gives_the_same_sampler_seed(): + root = np.random.SeedSequence(42) + + first = _sampler_seed(root, ("pressure_noise", "main")) + second = _sampler_seed(root, ("pressure_noise", "main")) + + assert first == second + + +@pytest.mark.parametrize("seed", [42, 7, [1, 2, 3]]) +@pytest.mark.parametrize("names", [("__list_choice__",), ("pressure_noise", "main")]) +def test_a_seed_that_is_not_a_sequence_reaches_numpy_untouched(seed, names): + # The control. Every fixed-seed baseline in the suite was recorded through + # this path, so anything but a SeedSequence has to arrive as it always did. + # Compared with the expression rather than with a recorded number, which + # would go red on a NumPy release instead of on a change of ours. + unchanged = np.random.SeedSequence( + entropy=seed, spawn_key=_names_as_spawn_key(tuple(sorted(names))) + ) + + assert _sampler_seed(seed, names) == _seed_sequence_to_int(unchanged) + + +def test_no_seed_still_means_no_seed(): + # None is left out above on purpose: it asks NumPy for fresh entropy, so + # two calls must not agree, and comparing one against another would be + # asserting the opposite of what an unseeded run promises. + first = _sampler_seed(None, ("__list_choice__",)) + second = _sampler_seed(None, ("__list_choice__",)) + + assert first != second