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()

SimulationAnalysis

Interface to make plots of multiple hdf5 files generated with Simulation.

All plot methods of the Analysis class are available in the SimulationAnalysis class.

Important

When calling an Analysis method from a SimulationAnalysis object every instance of "$id" in a str parameter will be replaced with the identifier of that hdf5, this can be useful to use with parameters like filename or title.

Example:

from comorbuss import SimulationAnalysis

analysis = SimulationAnalysis.from_folder("/data/folder")
analysis.plot_SEIR(filename="$id.pdf")
analysis.close()

empty_from_files(filenames) classmethod

Instances an empty SimulationAnalysis object from a list/array/tuples of hdf5 filenames

Parameters:

Name	Type	Description	Default
filenames		A list/numpy.array/tuple array of filenames	required

Returns:

Type	Description
SimulationAnalysis	SimulationAnalysis object.

Source code in comorbuss/lab/simulation_analysis.py

@classmethod
def empty_from_files(cls, filenames):
    """Instances an empty SimulationAnalysis object from a list/array/tuples of hdf5 filenames

    Args:
        filenames: A list/numpy.array/tuple array of filenames

    Returns:
        SimulationAnalysis: SimulationAnalysis object.
    """
    As = {}
    for simid, hdf5 in enumerate(filenames):
        As[simid] = Analysis({}, {}, False)
        As[simid].filename = hdf5
    return cls(As)

from_exp_parameters(*args, **kwargs) classmethod

Instance an SimulationAnalysis object from experiment parameters. Accepts same parameters as the Simulation class.

Returns:

Type	Description
SimulationAnalysis	SimulationAnalysis object.

Source code in comorbuss/lab/simulation_analysis.py

@classmethod
def from_exp_parameters(cls, *args, **kwargs):
    """Instance an SimulationAnalysis object from experiment parameters. Accepts same
        parameters as the [Simulation class](#comorbuss.lab.Simulation.__init__).

    Returns:
        SimulationAnalysis: SimulationAnalysis object.
    """
    sim = Simulation(*args, **kwargs)
    return SimulationAnalysis.from_sim(sim)

from_files(filenames) classmethod

Instances a SimulationAnalysis object from a list/array/tuples of hdf5 filenames

Parameters:

Name	Type	Description	Default
filenames		A list/numpy.array/tuple array of filenames	required

Returns:

Type	Description
SimulationAnalysis	SimulationAnalysis object.

Source code in comorbuss/lab/simulation_analysis.py

@classmethod
def from_files(cls, filenames):
    """Instances a SimulationAnalysis object from a list/array/tuples of hdf5 filenames

    Args:
        filenames: A list/numpy.array/tuple array of filenames

    Returns:
        SimulationAnalysis: SimulationAnalysis object.
    """
    As = {}
    for simid, hdf5 in enumerate(filenames):
        As[simid] = Analysis.from_hdf5(hdf5)
    return cls(As)

from_folder(folder, strings_to_ignore=[]) classmethod

Instance an SimulationAnalysis object from all hdf5 files in a folder.

Parameters:

Name	Type	Description	Default
folder	str	Path to the folder where the hdf5 files are stored.	required

Returns:

Type	Description
SimulationAnalysis	SimulationAnalysis object.

Source code in comorbuss/lab/simulation_analysis.py

@classmethod
def from_folder(cls, folder, strings_to_ignore=[]):
    """Instance an SimulationAnalysis object from all hdf5 files in a folder.

    Args:
        folder (str): Path to the folder where the hdf5 files are stored.

    Returns:
        SimulationAnalysis: SimulationAnalysis object.
    """
    files = os.listdir(path=folder)
    files.sort(key=lambda f: os.path.getctime(os.path.join(folder, f)))
    As = {}
    for f in files:
        ignore = any([s in f for s in strings_to_ignore])
        if f.lower()[-5:] == ".hdf5" and not ignore:
            hdf5 = os.path.join(folder, f)
            simid = os.path.split(f)[-1][:-5]
            As[simid] = Analysis.from_hdf5(hdf5)
    return cls(As)

from_sim(sim) classmethod

Instance an SimulationAnalysis object from a Simulation object.

Parameters:

Name	Type	Description	Default
sim	Simulation	Simulation object to load data.	required

Returns:

Type	Description
SimulationAnalysis	SimulationAnalysis object.

Source code in comorbuss/lab/simulation_analysis.py

@classmethod
def from_sim(cls, sim):
    """Instance an SimulationAnalysis object from a Simulation object.

    Args:
        sim (Simulation): Simulation object to load data.

    Returns:
        SimulationAnalysis: SimulationAnalysis object.
    """
    As = {}
    for simid, hdf5 in sim.get_hdf5_files().items():
        As[simid] = Analysis.from_hdf5(hdf5)
    return cls(As)