simulation.py

FixedParamsSimulation

__init__(self, seeds, parameters, hdf5_file, nproc=1, to_store=[], to_store_srvc=[], pre_simulate=None, post_simulate=None, append_data=False) special

Run multiple seeds with a fixed set of parameters.

Parameters:

Name	Type	Description	Default
seeds	list	Seeds to run.	required
parameters	dict	Parameters dictionary.	required
hdf5_file	str	File to store data.	required
nproc	int	Number of simultaneous threads to run. Defaults to 1.	1
to_store	list	Data to be stored. If not informed will store default TO_STORE.	[]
to_store_srvc	list	Service data to be stored. If not informed will store default TO_STORE_SRVC.	[]
pre_simulate	function	A function to be run between community initialization and simulation, it must accept a community object as parameter.	None
post_simulate	function	A function to be run after simulation of the community, it must accept a community object as parameter.	None
append_data	bool	Try to append seeds that are not in the hdf5 file, will not run seeds already stored on the hdf5 file. Defaults to False.	False

Source code in comorbuss/lab/simulation.py

def __init__(
    self,
    seeds,
    parameters,
    hdf5_file,
    nproc=1,
    to_store=[],
    to_store_srvc=[],
    pre_simulate=None,
    post_simulate=None,
    append_data=False,
):
    """Run multiple seeds with a fixed set of parameters.

    Args:
        seeds (list): Seeds to run.
        parameters (dict): Parameters dictionary.
        hdf5_file (str): File to store data.
        nproc (int, optional): Number of simultaneous threads to run. Defaults to 1.
        to_store (list, optional): Data to be stored. If not informed will store default TO_STORE.
        to_store_srvc (list, optional): Service data to be stored. If not informed will store default TO_STORE_SRVC.
        pre_simulate (function, optional): A function to be run between community initialization and simulation,
            it must accept a community object as parameter.
        post_simulate (function, optional): A function to be run after simulation of the community,
            it must accept a community object as parameter.
        append_data (bool, optional): Try to append seeds that are not in the hdf5 file, will not run seeds already
            stored on the hdf5 file. Defaults to False.
    """
    self.seeds = list(seeds)
    self.nproc = nproc
    self.parameters = tools.recursive_copy(parameters)
    self.triaged_parameters = triage_parameters(dict(parameters), gen_pop=False)[0]
    self.pre_simulate = pre_simulate
    self.post_simulate = post_simulate
    self.to_store = to_store
    self.hdf5_file = hdf5_file
    self.to_store_srvc = to_store_srvc
    if append_data:
        self.append_data = True
        try:
            self.check_seeds()
        except:
            pass
    if not append_data:
        if os.path.isfile(hdf5_file):
            os.remove(hdf5_file)

check_seeds(self)

Remove seeds already on hdf5 file.

Source code in comorbuss/lab/simulation.py

def check_seeds(self):
    """Remove seeds already on hdf5 file."""
    stored_seeds = []
    with h5dict.File(self.hdf5_file, "r") as hdf5:
        stored_seeds = hdf5["realizations"].keys()
    removed_seeds = []
    for seed in stored_seeds:
        if seed in self.seeds:
            self.seeds.remove(seed)
            removed_seeds.append(seed)
    if removed_seeds != []:
        removed_seeds.sort()
        print(
            "\tSeeds: {} found on hdf5 file: {} will not be simulated.".format(
                removed_seeds, self.hdf5_file
            )
        )

simulate(self)

Run simulations.

Source code in comorbuss/lab/simulation.py

def simulate(self):
    """Run simulations."""
    ps = []
    for seed in self.seeds:
        parameters = tools.recursive_copy(self.parameters)
        parameters["random_seed"] = seed
        ps.append({})
        ps[-1]["parameters"] = parameters
        ps[-1]["to_store"] = self.to_store
        ps[-1]["to_store_srvc"] = self.to_store_srvc
        ps[-1]["pre_simulate"] = self.pre_simulate
        ps[-1]["post_simulate"] = self.post_simulate
        ps[-1]["hdf5_file"] = self.hdf5_file
    pool = Pool(processes=self.nproc)
    pool.map(realization, ps)
    pool.close()

Simulation

The Simulation class handles multiple simulations with multiple seeds and/or multiple combinations of parameters.

Seealso

Usage examples can be found in the jupyter-examples folder in the repository:

• Simulation_example.ipynb
• meanfield.ipynb

__init__(self, experiment_name, seeds, fixed_parameters={}, iteration_parameters=[], recording_data=[], recording_data_srvc=[], nproc=1, pre_simulate=None, post_simulate=None, out=None, append_data=False, store_git_hash=False) special

Instances and initialize a Simulation object.

Parameters:

Name	Type	Description	Default
experiment_name	str	Name of the experiment, will me used to name files and folders.	required
fixed_parameters	dict	Fixed parameters.	{}
iteration_parameters	dict or list	Parameters to iterate during simulation. If list of dicts: Will run a simulation for each dict, replacing the parameters in each dict. If dict of list: Will run a simulation for each multiplicative combination of the parameters.	[]
recording_data	list	List with all data to be recorded after the simulation.	[]
recording_data_srvc	list	List with all service data to be recorded after the simulation.	[]
pre_simulate	function	A function to be run between community initialization and simulation, it must accept a community object as parameter.	None
post_simulate	function	A function to be run after simulation of the community, it must accept a community object as parameter.	None
seeds	list	List of seeds.	required
nproc	int	Number of concurrent threads. Defaults to 1.	1
append_data	bool	Append data to existing files, will not simulate existing seeds. Defaults to False.	False
out	str	Output directory.	None
store_git_hash	bool	Runs 'git rev-parse HEAD' on current folder and stores it to configurations files. Defaults to False.	False

Source code in comorbuss/lab/simulation.py

def __init__(
    self,
    experiment_name,
    seeds,
    fixed_parameters={},
    iteration_parameters=[],
    recording_data=[],
    recording_data_srvc=[],
    nproc=1,
    pre_simulate=None,
    post_simulate=None,
    out=None,
    append_data=False,
    store_git_hash=False,
):
    """Instances and initialize a Simulation object.

    Args:
        experiment_name (str): Name of the experiment, will me used to name files and folders.
        fixed_parameters (dict): Fixed parameters.
        iteration_parameters (dict or list): Parameters to iterate during simulation.

            * **If list of dicts**: Will run a simulation for each dict, replacing the parameters in
                each dict.
            * **If dict of list**: Will run a simulation for each multiplicative combination of the
                parameters.
        recording_data (list, optional): List with all data to be recorded after the simulation.
        recording_data_srvc (list, optional): List with all service data to be recorded after the simulation.
        pre_simulate (function, optional): A function to be run between community initialization and simulation,
            it must accept a community object as parameter.
        post_simulate (function, optional): A function to be run after simulation of the community,
            it must accept a community object as parameter.
        seeds (list): List of seeds.
        nproc (int, optional): Number of concurrent threads. Defaults to 1.
        append_data (bool, optional): Append data to existing files, will not simulate existing seeds.
            Defaults to False.
        out (str, optional): Output directory.
        store_git_hash (bool, optional): Runs 'git rev-parse HEAD' on current folder and stores it to
            configurations files. Defaults to False.
    """
    self.nproc = nproc
    self.experiment_name = experiment_name
    self.seeds = seeds
    self.count_seeds = len(seeds)
    self.append_data = append_data
    if store_git_hash:
        self.git_hash = subprocess.check_output(["git", "rev-parse", "HEAD"])

    # Select data to be recorded
    self.recording_data = recording_data
    self.recording_data_srvc = recording_data_srvc
    # initialize to_store array
    self.to_store = []
    to_store_default = S.TO_STORE
    # compute to_store
    for i in range(len(to_store_default)):
        if to_store_default[i][0] in self.recording_data:
            self.to_store.append(to_store_default[i])
    # initialize service to_store array
    self.to_store_srvc = []
    to_store_srvc_default = S.TO_STORE_SRVC
    # compute service to_store
    for i in range(len(to_store_srvc_default)):
        if to_store_srvc_default[i][0] in self.recording_data_srvc:
            self.to_store_srvc.append(to_store_srvc_default[i])

    # Stores pre and post simulate functions
    self.pre_simulate = pre_simulate
    self.post_simulate = post_simulate

    # Set output directory
    self.out = out or os.path.join("out", "simulations", self.experiment_name)
    self.out = os.path.abspath(self.out)
    tools.check_dir(self.out)

    # Set parameters
    self.fixed_parameters = fixed_parameters
    self.iteration_parameters = iteration_parameters

    if not "city_name" in self.fixed_parameters:
        self.fixed_parameters["city_name"] = self.experiment_name
    self.fixed_parameters["log_progress"] = False

    self.iterable = self.gen_configurations(
        self.fixed_parameters, self.iteration_parameters, self.experiment_name
    )

    # Calculate complexity
    self.complexity = self.gen_complexity(self.iterable)
    self.sum_complexity = np.sum(self.complexity)

get_hdf5_files(self)

Generates a dict of the hdf5 files for the set simulations with the sim_id/name as the keys.

Returns:

Type	Description
dict	Paths to the HDF5 files.

Source code in comorbuss/lab/simulation.py

def get_hdf5_files(self):
    """Generates a dict of the hdf5 files for the set simulations with the sim_id/name as the keys.

    Returns:
        dict: Paths to the HDF5 files.
    """
    files = {}
    for config in self.iterable:
        files[config["name"]] = config["hdf5_file"]
    return files

get_iteration_parameters(self)

Outputs the iteration parameters as a list of dicts.

Returns:

Type	Description
dict	Iteration parameters as list of dicts.

Source code in comorbuss/lab/simulation.py

def get_iteration_parameters(self):
    """Outputs the iteration parameters as a list of dicts.

    Returns:
        dict: Iteration parameters as list of dicts.
    """
    ips = []
    for config in self.iterable:
        ips.append(tools.recursive_copy(config["iteration_parameters"]))
        ips[-1]["name"] = config["name"]
    return ips

print_iteration_parameters(self)

Prints the iteration parameters as a list of dicts.

Source code in comorbuss/lab/simulation.py

def print_iteration_parameters(self):
    """Prints the iteration parameters as a list of dicts."""
    pprint(self.get_iteration_parameters())

simulate(self, node=0, number_of_nodes=1, node_type='seeds')

Runs the simulations

Parameters:

Name	Type	Description	Default
node	int	[description]. Defaults to 0.	0
number_of_nodes	int	Divides the simulation in multiples nodes. Defaults to 1.	1
node_type	str	Specify where to divide nodes, seeds or configurations. Defaults to 'seed'.	'seeds'

Source code in comorbuss/lab/simulation.py

def simulate(self, node=0, number_of_nodes=1, node_type="seeds"):
    """Runs the simulations

    Args:
        node (int, optional): [description]. Defaults to 0.
        number_of_nodes (int, optional): Divides the simulation in multiples nodes.
            Defaults to 1.
        node_type (str, optional): Specify where to divide nodes, `seeds` or `configurations`.
            Defaults to 'seed'.
    """
    # initialize time and percentage
    ts = time.time()
    p_done = 0

    # run simulation for all configurations
    print("Simulating experiment {}...".format(self.experiment_name))
    confs = self.iterable
    sum_complexity = self.sum_complexity
    if number_of_nodes > 1 and node_type == "configurations":
        confs = [c for i, c in enumerate(confs) if i % number_of_nodes == node]
        complexity = self.gen_complexity(confs)
        sum_complexity = np.sum(complexity)
    count_conf = len(confs)
    for this_conf, params in enumerate(confs):
        # load simulation data
        sim_id = params["name"]
        sim_params = params["parameters"]
        sim_hdf5 = params["hdf5_file"]

        # Saves simulation data to disk
        with open(params["params_file"], "w") as f:
            pprint(params, stream=f)

        # Runs simulation
        print(
            "({}/{}) Computing configuration {}/{}".format(
                this_conf + 1, count_conf, self.experiment_name, sim_id
            )
        )

        this_conf = number_of_nodes * this_conf + node
        p_now = self.complexity[this_conf] / sum_complexity
        t_spent = time.time() - ts
        t_exp = -1 if p_done == 0 else t_spent / p_done
        print(
            "Time projection at {:.2f}s/{:.2f}s, {:.2f}% done, {:.2f}% processing now.".format(
                t_spent, t_exp, p_done * 100, p_now * 100
            )
        )
        seeds = self.seeds
        if number_of_nodes > 1 and node_type == "seeds":
            seeds = [s for i, s in enumerate(seeds) if i % number_of_nodes == node]
        sim = FixedParamsSimulation(
            seeds,
            sim_params,
            sim_hdf5,
            to_store=self.to_store,
            nproc=self.nproc,
            to_store_srvc=self.to_store_srvc,
            pre_simulate=self.pre_simulate,
            post_simulate=self.post_simulate,
            append_data=self.append_data,
        )
        print(
            "Running for {} particles with {} steps".format(
                sim.triaged_parameters["Nparticles"],
                sim.triaged_parameters["Nsteps"],
            )
        )  # ,   , memory around {:.2f}gb
        # 4.9e-8*sim.triaged_parameters['Nparticles']*sim.triaged_parameters['Nsteps']*self.count_seeds))
        sim.simulate()
        p_done += self.complexity[this_conf] / sum_complexity

    def run(self):
        """Alias to Simulation.simulate(), will be removed in future releases."""
        self.simulate()

realization(args)

Executes a single seed with a given set of parameters and stores the hdf5 file.

Parameters:

Name	Type	Description	Default
parameters	dict	Parameters dictionary.	required
hdf5_file	str	File to store data.	required
to_store	list	Data to be stored.	required
to_store_srvc	list	Service data to be stored.	required

Source code in comorbuss/lab/simulation.py

def realization(args):
    """Executes a single seed with a given set of parameters and stores the hdf5 file.

    Args:
        parameters (dict): Parameters dictionary.
        hdf5_file (str): File to store data.
        to_store (list): Data to be stored.
        to_store_srvc (list): Service data to be stored.
    """
    hdf5_file = args["hdf5_file"]
    to_store = args["to_store"]
    to_store_srvc = args["to_store_srvc"]
    pre_simulate = args["pre_simulate"] or (lambda *_: None)
    post_simulate = args["post_simulate"] or (lambda *_: None)
    parameters = args["parameters"]

    old_print = replace_print(str="\t\t[seed_{}]".format(parameters["random_seed"]))
    ti = time.time()
    comm = community(**parameters)
    pre_simulate(comm)
    comm.simulate()
    post_simulate(comm)
    if "R0t" in [item for item, _, _ in to_store]:
        comm.compute_R0t()
    replace_print(old_print)
    print(
        "\t Executed seed {} with {}s".format(
            parameters["random_seed"], time.time() - ti
        )
    )
    tools.save_hdf5(
        [comm],
        hdf5_file,
        to_store=to_store,
        to_store_srvc=to_store_srvc,
        skip_defaults=True,
        try_append=True,
    )
    print("\t Stored seed {} data.".format(parameters["random_seed"]))
    del comm