aegis_sim

 1import logging
 2import pathlib
 3
 4from aegis_sim.dataclasses.population import Population
 5from aegis_sim.bioreactor import Bioreactor
 6from aegis_sim import variables, submodels, parameterization
 7from aegis_sim.parameterization import parametermanager
 8from aegis_sim.recording import recordingmanager
 9
10
11def run(custom_config_path, pickle_path, overwrite, custom_input_params):
12    init(custom_config_path, overwrite, pickle_path, custom_input_params)
13
14    population = (
15        Population.initialize(
16            n=parametermanager.parameters.INITIAL_POPULATION_SIZE,
17            AGE_LIMIT=parametermanager.parameters.AGE_LIMIT,
18        )
19        if pickle_path is None
20        else Population.load_pickle_from(pickle_path)
21    )
22
23    bioreactor = Bioreactor(population)
24
25    sim(bioreactor=bioreactor)
26
27
28def init(custom_config_path, overwrite=False, pickle_path=None, custom_input_params={}):
29    """
30
31    When testing aegis, initialize all modules using this function, e.g.
32
33    import aegis_sim
34    aegis_sim.init("_.yml")
35
36    And then you can safely import any module.
37    """
38
39    custom_config_path = pathlib.Path(custom_config_path)
40
41    parametermanager.init(
42        custom_config_path=custom_config_path,
43        custom_input_params=custom_input_params,
44    )
45    variables.init(
46        variables,
47        custom_config_path=custom_config_path,
48        pickle_path=pickle_path,
49        RANDOM_SEED=parametermanager.parameters.RANDOM_SEED,
50    )
51    parameterization.init_traits(parameterization)
52    submodels.init(submodels, parametermanager=parametermanager)
53
54    recordingmanager.init(custom_config_path, overwrite)
55    recordingmanager.initialize_recorders(TICKER_RATE=parametermanager.parameters.TICKER_RATE)
56
57
58def sim(bioreactor):
59    # presim
60    recordingmanager.configrecorder.write_final_config_file(parametermanager.final_config)
61    recordingmanager.ticker.start_process()
62    ticker_pid = recordingmanager.ticker.pid
63    assert ticker_pid is not None
64    recordingmanager.summaryrecorder.write_input_summary(ticker_pid=recordingmanager.ticker.pid)
65    # TODO hacky solution of decrementing and incrementing steps
66    variables.steps -= 1
67    recordingmanager.featherrecorder.write(bioreactor.population)
68    variables.steps += 1
69
70    # sim
71    recordingmanager.phenomaprecorder.write()
72
73    while (variables.steps <= parametermanager.parameters.STEPS_PER_SIMULATION) and not recordingmanager.is_extinct():
74        recordingmanager.progressrecorder.write(len(bioreactor.population))
75        recordingmanager.simpleprogressrecorder.write()
76        bioreactor.run_step()
77        variables.steps += 1
78
79    # postsim
80    recordingmanager.summaryrecorder.write_output_summary()
81    logging.info("Simulation finished.")
82    recordingmanager.ticker.stop_process()
def run(custom_config_path, pickle_path, overwrite, custom_input_params):
12def run(custom_config_path, pickle_path, overwrite, custom_input_params):
13    init(custom_config_path, overwrite, pickle_path, custom_input_params)
14
15    population = (
16        Population.initialize(
17            n=parametermanager.parameters.INITIAL_POPULATION_SIZE,
18            AGE_LIMIT=parametermanager.parameters.AGE_LIMIT,
19        )
20        if pickle_path is None
21        else Population.load_pickle_from(pickle_path)
22    )
23
24    bioreactor = Bioreactor(population)
25
26    sim(bioreactor=bioreactor)
def init( custom_config_path, overwrite=False, pickle_path=None, custom_input_params={}):
29def init(custom_config_path, overwrite=False, pickle_path=None, custom_input_params={}):
30    """
31
32    When testing aegis, initialize all modules using this function, e.g.
33
34    import aegis_sim
35    aegis_sim.init("_.yml")
36
37    And then you can safely import any module.
38    """
39
40    custom_config_path = pathlib.Path(custom_config_path)
41
42    parametermanager.init(
43        custom_config_path=custom_config_path,
44        custom_input_params=custom_input_params,
45    )
46    variables.init(
47        variables,
48        custom_config_path=custom_config_path,
49        pickle_path=pickle_path,
50        RANDOM_SEED=parametermanager.parameters.RANDOM_SEED,
51    )
52    parameterization.init_traits(parameterization)
53    submodels.init(submodels, parametermanager=parametermanager)
54
55    recordingmanager.init(custom_config_path, overwrite)
56    recordingmanager.initialize_recorders(TICKER_RATE=parametermanager.parameters.TICKER_RATE)

When testing aegis, initialize all modules using this function, e.g.

import aegis_sim aegis_sim.init("_.yml")

And then you can safely import any module.

def sim(bioreactor):
59def sim(bioreactor):
60    # presim
61    recordingmanager.configrecorder.write_final_config_file(parametermanager.final_config)
62    recordingmanager.ticker.start_process()
63    ticker_pid = recordingmanager.ticker.pid
64    assert ticker_pid is not None
65    recordingmanager.summaryrecorder.write_input_summary(ticker_pid=recordingmanager.ticker.pid)
66    # TODO hacky solution of decrementing and incrementing steps
67    variables.steps -= 1
68    recordingmanager.featherrecorder.write(bioreactor.population)
69    variables.steps += 1
70
71    # sim
72    recordingmanager.phenomaprecorder.write()
73
74    while (variables.steps <= parametermanager.parameters.STEPS_PER_SIMULATION) and not recordingmanager.is_extinct():
75        recordingmanager.progressrecorder.write(len(bioreactor.population))
76        recordingmanager.simpleprogressrecorder.write()
77        bioreactor.run_step()
78        variables.steps += 1
79
80    # postsim
81    recordingmanager.summaryrecorder.write_output_summary()
82    logging.info("Simulation finished.")
83    recordingmanager.ticker.stop_process()