community.py

community

Main class, initialize and evolves a community.

Attributes:

Name	Type	Description
given_parameters	dict	Parameters given at the initialization without triage and normalization.
parameters	dict	Parameters used on the simulation, triaged and normalized.
initOk	bool	True indicates no error at the initialization, False indicates an error.
simOk	bool	True indicates that the simulation ended without errors.
events	list	A list where all events generated during the simulation is stored, each event is a list of tree elements, the first is the event's message, the second is the time the event was generated and the third is the type of event. Events can be printed with print_events() or saved to a file with save_events().
random_seed	int	Seed for all random number generators on the simulation.
Nparticles	int	Number of particles used in the simulation.
Nsteps	int	Number of time steps used on the simulation.
Nhours	int	Number of hours in the simulation.
dt	float	Size of the time step, can be set with the parameter resolution_t.
time_of_the_day	float	Current time of the day in the simulation.
scenario_id	string	Generated scenario id for the current simulation.
free_hours	list	A two elements list indicating the window of time the particles are allowed to move trough the city.
srv	services_infrastructure	A [services_infrastructure][comorbuss.services.services_infrastructure] object where all [services][comorbuss.services.service] are stored.
pop	population	A population object where all information relative to the population is generated and stored.
to_execute	list	A list of all functions to be executed on every step of the simulation.

__init__(self, skip_triage=False, **parameters) special

Community class initialization.

Parameters:

Name	Type	Description	Default
**parameters	dict	Parameters dictionary for the simulation	{}
skip_triage	bool	Skips parameters triage (code might not work)	False

Source code in comorbuss/community.py

def __init__(self, skip_triage=False, **parameters):
    """Community class initialization.

    Args:
        **parameters (dict): Parameters dictionary for the simulation
        skip_triage (bool, optional): Skips parameters triage (code might not work)
    """
    time_start = time.time()
    self.comm = self
    self.clk = clock()
    self.given_parameters = recursive_copy(parameters)
    self.events = []
    self.events_printed = 0
    self.initOk = True
    self.simOk = False
    self.show_warn_msgs = get_value(parameters, "show_warning_messages", True)
    if not skip_triage:
        parameters, clk, self.randgencom = triage_parameters(parameters, self)
        self.print_events()
    self.random_seed = parameters["random_seed"]
    self.Nparticles = parameters["Nparticles"]
    self.Nsteps = parameters["Nsteps"]
    self.Nhours = parameters["Nhours"]
    self.Ndays = parameters["Ndays"]
    self.clk = clk
    self.clk.set_start_date(parameters["start_date"])
    self.dt = self.clk.dt
    self.steps_per_day = int(np.ceil(24 / self.dt))
    self.scenario_id = parameters["scenario_id"]
    self.parameters = parameters
    self.MASK_FALSE = np.zeros(self.Nparticles, dtype=bool)
    self.time_series = parameters["store_time_series"]

    self.isol_pct_time_series = parameters["isol_pct_time_series"]
    self.isol_stay_prob = parameters["isol_stay_prob"]
    self.isol_mask_prev = np.full(self.Nparticles, False)
    if parameters["social_isolation"]:
        self.social_isolation_mean = np.mean(self.isol_pct_time_series)
    else:
        self.social_isolation_mean = 0.0

    self.free_hours = parameters["free_hours"]
    self.time_series = parameters["store_time_series"]
    self.TO_STORE = []

    # Setup progress bar or disable it
    self.log_progress = parameters["log_progress"]
    self.tqdm_par = {
        "unit": "day",
        "unit_scale": self.dt / 24,
        "ncols": 80,
        "bar_format": "[Progress] {percentage:2.0f}% |{bar}| {n:.1f}/{total:.1f} days [{rate_fmt}, {remaining_s:.1f}s rem]",
        "smoothing": 0,
        "file": sys.stdout,
    }
    self.print_evnts = parameters["print_events"]

    # Population and services initialization
    if self.initOk:
        homes = self.assign_homes(parameters)
    if self.initOk:
        neighborhoods = self.assign_neighborhoods(parameters)
    if self.initOk:
        self.pop = population(parameters, self, homes, neighborhoods)
    if self.initOk:
        self.srv = services_infrastructure(parameters, self)

    # Initialize environment and homes net generators.
    if self.initOk:
        self.pop.add_net_generator(
            S.PLC_ENV, enviroment_net(parameters["env_net_par"], self)
        )
        self.pop.add_net_generator(
            S.PLC_HOME, home_net(parameters["home_net_par"], self)
        )

    ageGroupsChildrenIsolated = [0, 1]
    self.childrenIDs = []
    for ageGroup in ageGroupsChildrenIsolated:
        for pid in self.pop.pid[self.pop.ages == ageGroup]:
            self.childrenIDs.append(pid)
    self.childrenIDs = np.array(self.childrenIDs)

    if self.initOk:
        # Initialize modules
        self.modules = []
        modules = {}
        for module_name, module_param in parameters["modules"]:
            module, module_name = self.init_module(module_name, module_param)
            if module != None:
                modules[module_name] = module_param
        parameters["modules"] = modules  # Prevent parameters saving error

    if self.initOk:
        self.quarantines = quarantines(parameters["quarantines"], self)

    if self.initOk:
        # Parse stop filter
        self.stop_filter, fok = filter_parse(
            parameters["stop_filter"],
            self,
            comm_str="self",
            allowed_filters=[
                "service",
                "vaccination",
                "vaccination_attr",
                "diagnostic",
                "module",
                "workers",
            ],
        )
        if not fok:
            self.event_log(
                "'{}' is not a valid stop_filter, will not use stop_filter"
                " on this simulation.".format(parameters["stop_filter"]),
                S.MSG_WRNG,
            )
            self.stop_filter = "False"
        self.stop_condition = False

        if eval(self.stop_filter):
            self.initOk = False
            self.event_log(
                "Parsed stop_filter '{}' evaluates True before the start of the"
                " simulation, this is an invalid configuration.".format(
                    self.stop_filter
                ),
                S.MSG_ERROR,
            )

    if self.initOk:
        # Lockdown configuration
        self.lckd_progress = S.LCKD_PRGSS_OFF
        self.lckd_offset_days = 0
        self.lckd_adhere_probab = parameters["lockdown_adhere_percent"]
        self.lckd_decision_offset = parameters["lockdown_decision_offset"]
        self.lockdown_to_start = DECISON_FUNCTIONS[parameters["decision"]]["start"]
        self.lockdown_to_stop = DECISON_FUNCTIONS[parameters["decision"]]["stop"]
        self.lckd_decision_par = parameters["decision_par_lockdown"]
        if parameters["use_day_series"] and parameters["lockdown"]:
            self.lockdown_to_start2 = self.lockdown_to_start
            self.lockdown_to_stop2 = self.lockdown_to_stop
            self.lockdown_to_start = to_start_lockdown_day_series
            self.lockdown_to_stop = to_finish_lockdown_day_series
            self.lockdown_ds = parameters["lockdown_day_series"]
            self.lockdown_ds_len = len(self.lockdown_ds)

    ############################################################################
    ########################### Execution setup ################################
    ############################################################################

    # Priority convention:
    #
    # 0-24: Functions that need to run before secondary states change (select to test, select to vaccinate).
    # 25-49: Functions that change particles secondary states (eg. diagnostics, vaccination, etc).
    # 50-74: Functions that interfere with particles positions.
    # 75-99: Functions that change particles position.
    # 100: update_states.
    # 100+: post update_states: data consolidation, output, etc.

    if self.initOk:
        if parameters["lockdown"] and parameters["social_isolation"]:
            self.add_to_execution(self.lockdown, priority=53)
        if parameters["social_isolation"]:
            self.add_to_execution(
                self.isolation, priority=54, time_of_the_day=self.free_hours[0]
            )
        self.add_to_execution(self.update_position, priority=98)
        self.add_to_execution(self.check_placements_and_activities, priority=98.1)
        self.add_to_execution(self.pop.update_states, priority=100)
        if self.time_series:
            self.add_to_execution(self.store_time_series, priority=101)
        if parameters["print_events"]:
            self.add_to_execution(self.print_events, priority=102)
        if self.stop_filter != "False":
            self.add_to_execution(self.eval_stop_condition, priority=999)

    if self.initOk:
        for module in self.modules:
            if hasattr(module, "run_after_init"):
                module.run_after_init()

        ############################################################################
        ############################ Setup Warnings ################################
        ############################################################################

        if parameters["lockdown"] and not parameters["social_isolation"]:
            self.event_log(
                "Can't use Lockdown without Social Isolation, Lockdown will"
                " be deactivated, to force Lockdown turn on Social Isolation "
                "and set isol_pct_time_series to a negative number",
                S.MSG_WRNG,
            )

    time_end = time.time()
    self.time_init = time_end - time_start
    self.event_log(
        "Initialization took {:.2f} seconds".format(self.time_init),
        S.MSG_PRGS,
    )
    self.print_events()

add_to_execution(self, function, priority=0, time_of_the_day=(0, 24))

Adds a function to the execution list.

Parameters:

Name	Type	Description	Default
function	callable	Function or method to be called.	required
priority	int	Priority to be executed each step, lower priority functions will be executed first each step. Defaults to 0.	0
time_of_the_day	tuple	An specific hour of the day or a tuple containing an window to execute function. Defaults to (0, 24).	(0, 24)

Source code in comorbuss/community.py

def add_to_execution(self, function, priority=0, time_of_the_day=(0, 24)):
    """Adds a function to the execution list.

    Args:
        function (callable): Function or method to be called.
        priority (int, optional): Priority to be executed each step, lower
            priority functions will be executed first each step. Defaults to 0.
        time_of_the_day (tuple, optional): An specific hour of the day or a
            tuple containing an window to execute function. Defaults to (0, 24).
    """
    if not hasattr(self, "to_execute"):
        self.to_execute = []
    try:
        time_of_the_day = (
            float(time_of_the_day),
            float(time_of_the_day) + self.dt,
        )
    except TypeError:
        pass
    try:
        time_of_the_day = tuple(time_of_the_day)
    except TypeError:
        self.log_and_raise(
            "Error parsing time of the day for function: {}".format(
                function.__name__
            ),
            TypeError,
        )
    self.to_execute.append((priority, function, time_of_the_day))
    self.to_execute.sort(key=lambda x: x[0])  # Sort by priority

add_to_store(self, name, eval_str, compress=False)

Adds a value to the to_store list.

Parameters:

Name	Type	Description	Default
name	str	Name to be save the data.	required
eval_str	str	String to be evaluate to load data	required
compress	bool	Compress data marker. Defaults to False.	False

Source code in comorbuss/community.py

def add_to_store(self, name, eval_str, compress=False):
    """Adds a value to the to_store list.

    Args:
        name (str): Name to be save the data.
        eval_str (str): String to be evaluate to load data
        compress (bool, optional): Compress data marker. Defaults to False.
    """
    to_store = [name, eval_str, compress]
    self.TO_STORE.append(to_store)

assign_homes(self, parameters)

Assing a home to each individual in a population.

Parameters:

Name	Type	Description	Default
parameters	dict	Simulation parameters.	required

Returns:

Type	Description
np.array	Array of size Nparticles with the home id for each particle.

Source code in comorbuss/community.py

def assign_homes(self, parameters):
    """Assing a home to each individual in a population.

    Args:
        parameters (dict): Simulation parameters.

    Returns:
        np.array: Array of size Nparticles with the home id for each particle.
    """
    if "net_n_homes" in parameters:
        self.homesNumber = parameters["net_n_homes"]
    else:
        self.homesNumber = parameters["homesNumber"]
    homes = np.zeros((self.homesNumber, 3))
    if "net_homes" in parameters:
        home_id = parameters["net_homes"]
    else:
        home_id = self.randgencom.choice(self.homesNumber, self.Nparticles)
    return home_id

assign_neighborhoods(self, parameters)

Randomly assign a neighborhood for each particle in the simulation.

Parameters:

Name	Type	Description	Default
parameters	dict	Simulation parameters.	required

Returns:

Type	Description
np.array	Array of size Nparticles with the neighborhood id for each particle.

Source code in comorbuss/community.py

def assign_neighborhoods(self, parameters):
    """Randomly assign a neighborhood for each particle in the simulation.

    Args:
        parameters (dict): Simulation parameters.

    Returns:
        np.array: Array of size Nparticles with the neighborhood id for each particle.
    """
    neigh_perc = parameters["neighborhoods_percentages"]
    available_pids = np.arange(self.Nparticles)
    neighborhoods = np.zeros(self.Nparticles)
    for i in range(1, len(neigh_perc)):
        n = int(neigh_perc[i] * self.Nparticles)
        pids = self.randgencom.choice(available_pids, size=n, replace=False)
        neighborhoods[pids] = i
        maskDelete = ~np.isin(available_pids, pids)
        available_pids = available_pids[maskDelete]
    return neighborhoods

compute_R0t(self)

Calculate the Basic Reproduction Number for each time

The Basic Reproduction Number is the mean number of particles infected by one particle

Source code in comorbuss/community.py

def compute_R0t(self):
    """Calculate the Basic Reproduction Number for each time

    The Basic Reproduction Number is the mean number of particles infected by one particle
    """
    if self.time_series:
        spreads = self.pop.spreads
        states = self.pop.states_ts
        [l, _] = np.shape(spreads)
        # Initialize the Basic Reproduction Number to be zero at each time
        R0t = np.zeros(l)
        mask_IR = np.logical_or(states == S.STATE_R, states == S.STATE_I)
        cumulative_spreads = np.cumsum(spreads, axis=0)
        Nvec = np.sum(mask_IR & (cumulative_spreads > 0), axis=1)
        # Set the Basic Reproduction Number at each time to be the mean of the previously calculated cumulative sum
        for i in range(l):
            recov_spreads = cumulative_spreads[i, mask_IR[i, :]]
            R0t[i] = np.sum(recov_spreads) / np.maximum(1, Nvec[i])
        self.pop.R0t = R0t
        return R0t
    else:
        print("Can't calculate R0t with store_time_series=False.")
        return np.zeros(self.Nsteps)

event_log(self, message, msgtype)

Stores a message in the simulation log.

Parameters:

Name	Type	Description	Default
message	str	Message.	required
msgtype	int	Message type marker, see settings.py.	required

Source code in comorbuss/community.py

def event_log(self, message, msgtype):
    """Stores a message in the simulation log.

    Args:
        message (str): Message.
        msgtype (int): Message type marker, see `settings.py`.
    """
    self.events.append([message, self.clk.time, msgtype])

init_module(self, module, module_parameters)

Load and initialize a module from the modules folder.

Parameters:

Name	Type	Description	Default
module_name	str	Module name.	required
module_parameters	Any	Module_parameters	required

Source code in comorbuss/community.py

def init_module(self, module, module_parameters):
    """Load and initialize a module from the modules folder.

    Args:
        module_name (str): Module name.
        module_parameters (Any): Module_parameters
    """
    if type(module) is str:
        module = importlib.import_module("comorbuss.modules.{}".format(module))
    module = module.module(module_parameters, self)
    if hasattr(module, "name"):
        module_name = module.name
    else:
        self.initOk = False
        self.event_log(
            "{} module has no name attribute, can't run.".format(module),
            S.MSG_ERROR,
        )
        return
    if hasattr(module, "scenario_id"):
        self.scenario_id += module.scenario_id
    if hasattr(module, "init_time_series") and self.time_series:
        module.init_time_series()
    self.modules.append(module)
    setattr(self, module_name, module)
    return module, module_name

isolation(self)

Run social isolation mechanics, to be isolated as a module.

Source code in comorbuss/community.py

def isolation(self):
    """Run social isolation mechanics, to be isolated as a module."""
    # Release particles from isolation (some may be isolated again right after)
    mask_free_from_quarantine = ~np.isin(
        self.pop.quarantine_states, self.quarantines.confining_quarantines
    )
    mask_activity_free = self.pop.activity <= S.ACT_FREE
    mask_free_for_isolation = mask_free_from_quarantine & mask_activity_free
    mask_release_from_isolation = self.isol_mask_prev & mask_free_for_isolation
    self.pop.mark_for_new_position(
        mask_release_from_isolation,
        self.pop.placement[mask_release_from_isolation],
        S.ACT_FREE,
    )
    # Set particles to isolation
    curr_pct = self.isol_pct_time_series[self.clk.today]
    if curr_pct >= 0:
        mask_not_yet_isolated = mask_free_for_isolation
        # First isolate by age (babies or small children)
        mask_isolate_by_age = (
            np.isin(self.pop.pid, self.childrenIDs) & mask_not_yet_isolated
        )
        self.pop.mark_for_new_position(mask_isolate_by_age, S.PLC_HOME, S.ACT_ISOL)
        mask_not_yet_isolated = mask_not_yet_isolated & (~mask_isolate_by_age)
        # Next isolate some particles free to move around
        nIsolatedSoFar = np.count_nonzero(mask_isolate_by_age)
        nIsolatedNowOn = np.minimum(
            int(curr_pct * self.Nparticles) - nIsolatedSoFar,
            np.count_nonzero(mask_not_yet_isolated),
        )
        if nIsolatedNowOn > 0:
            probab = np.zeros(self.Nparticles)
            mask_probab_prev_isol = mask_not_yet_isolated & self.isol_mask_prev
            mask_probab_prev_not_isol = mask_not_yet_isolated & (
                ~self.isol_mask_prev
            )
            M1 = np.count_nonzero(mask_probab_prev_isol)
            M2 = np.count_nonzero(mask_probab_prev_not_isol)
            probab[mask_probab_prev_isol] = self.isol_stay_prob / np.maximum(M1, 1)
            probab[mask_probab_prev_not_isol] = (
                1.0 - self.isol_stay_prob
            ) / np.maximum(M2, 1)
            probab = normalize(probab)
            isolate_by_random_choice = self.randgencom.choice(
                self.Nparticles, nIsolatedNowOn, replace=False, p=probab
            )
            self.pop.mark_for_new_position(
                isolate_by_random_choice, S.PLC_HOME, S.ACT_ISOL
            )
            self.isol_mask_prev = mask_isolate_by_age
            self.isol_mask_prev[isolate_by_random_choice] = True
        else:
            self.isol_mask_prev = mask_isolate_by_age
    else:
        self.isol_mask_prev[:] = False

lockdown(self)

Run lockdown mechanics, to be isolated as a module.

Source code in comorbuss/community.py

def lockdown(self):
    """Run lockdown mechanics, to be isolated as a module."""
    self.pop.in_lckd[self.clk.step] = self.pop.in_lckd[self.clk.step - 1]
    first_hour_of_the_day = self.clk.at_time_of_day(self.free_hours[0])
    if self.clk.between_hours(self.free_hours[0], self.free_hours[1]):
        lckd_is_off = self.lckd_progress == S.LCKD_PRGSS_OFF
        lckd_is_to_start = self.lckd_progress == S.LCKD_PRGSS_START
        lckd_is_to_finish = self.lckd_progress == S.LCKD_PRGSS_FINISH
        lckd_is_on = self.lckd_progress == S.LCKD_PRGSS_ON
        # start lockdown at next day after having identified it is needed
        if lckd_is_to_start & first_hour_of_the_day:
            # if (self.lckd_decision_offset == self.lckd_offset_days):
            self.lockdown_start()
            self.lckd_progress = S.LCKD_PRGSS_ON
            self.lckd_offset_days = 0
            self.event_log("Lockdown started", S.MSG_EVENT)
        # else:
        #     self.lckd_offset_days = self.lckd_offset_days + 1
        # finish lockdown at next day after having identified it is needed, or maintain its status until then
        elif lckd_is_to_finish & first_hour_of_the_day:
            if self.lckd_decision_offset == self.lckd_offset_days:
                self.lockdown_finish()
                self.lckd_progress = S.LCKD_PRGSS_OFF
                self.lckd_offset_days = 0
                self.event_log("Lockdown finished", S.MSG_EVENT)
            else:
                self.lockdown_maintain()
                self.lckd_offset_days = self.lckd_offset_days + 1
        # if lockdown is off, check wether to start it
        elif lckd_is_off:
            if self.lockdown_to_start(self, self.lckd_decision_par):
                self.lckd_progress = S.LCKD_PRGSS_START
        # if lockdown is currently on, maintain its status
        elif lckd_is_on:
            self.lockdown_maintain()
            if self.lockdown_to_stop(self, self.lckd_decision_par):
                self.lckd_progress = S.LCKD_PRGSS_FINISH
    elif first_hour_of_the_day and (self.lckd_progress == S.LCKD_PRGSS_OFF):
        if self.lockdown_to_start(self, self.lckd_decision_par):
            self.lckd_progress = S.LCKD_PRGSS_START

log_and_raise(self, message, error)

Log an error message, print all messages and raise an exception.

Parameters:

Name	Type	Description	Default
message	str	Message.	required
error	Exception	Exception to be raised.	required

Source code in comorbuss/community.py

def log_and_raise(self, message, error):
    """Log an error message, print all messages and raise an exception.

    Args:
        message (str): Message.
        error (Exception): Exception to be raised.
    """
    self.event_log(message, S.MSG_ERROR)
    self.print_events()
    raise error(message)

log_results(self)

Add results to the simulation log.

Source code in comorbuss/community.py

def log_results(self):
    """Add results to the simulation log."""
    ninfections = np.sum(self.pop.states != S.STATE_S)
    self.event_log("Number of infections: {:}".format(ninfections), S.MSG_RSLT)
    nasymptomatic = np.sum(
        (self.pop.symptoms == S.SYMPT_NO) | (self.pop.symptoms == S.SYMPT_NO * 10)
    )
    self.event_log(
        "Number of asymptomatic cases: {:}".format(nasymptomatic), S.MSG_RSLT
    )
    nsymptomatic = np.sum(
        (self.pop.symptoms == S.SYMPT_YES) | (self.pop.symptoms == S.SYMPT_YES * 10)
    )
    self.event_log(
        "Number of light symptomatic cases: {:}".format(nsymptomatic),
        S.MSG_RSLT,
    )
    nsevere = np.sum(
        (self.pop.symptoms == S.SYMPT_SEVERE)
        | (self.pop.symptoms == S.SYMPT_SEVERE * 10)
    )
    self.event_log("Number of severe symptomatic: {:}".format(nsevere), S.MSG_RSLT)
    ndeceaseds = np.sum(self.pop.states == S.STATE_D)
    self.event_log("Number of deaths: {:}".format(ndeceaseds), S.MSG_RSLT)
    ninfections = np.sum(self.pop.inf_placement == S.PLC_HOME)
    self.event_log("Infections at home: {:}".format(ninfections), S.MSG_RSLT)
    ninfections = np.sum(self.pop.inf_placement == S.PLC_ENV)
    self.event_log("Infections on environment: {:}".format(ninfections), S.MSG_RSLT)
    for srvc in self.srv.all_services:
        ninfections = np.sum(self.pop.inf_placement == srvc.placement)
        self.event_log(
            "Infections on {:}: {:}".format(srvc.name, ninfections),
            S.MSG_RSLT,
        )
    for module in self.modules:
        module.log_results()

print_events(self)

Prints events registered during execution.

Source code in comorbuss/community.py

def print_events(self):
    """Prints events registered during execution."""
    write_screen = print
    if hasattr(self, "log_progress"):
        if self.log_progress:
            write_screen = tqdm.write
    for i in range(self.events_printed, len(self.events)):
        self.events_printed = i + 1
        if self.events[i][2] == S.MSG_RSLT:
            msg = "[" + S.MSG_TEXTS[self.events[i][2]] + "] " + self.events[i][0]
            write_screen(msg, file=sys.stdout)
        else:
            if self.events[i][2] != S.MSG_WRNG or self.show_warn_msgs:
                msg = (
                    "["
                    + S.MSG_TEXTS[self.events[i][2]]
                    + "] "
                    + self.events[i][0]
                    + " [step: {:}, day: {:}, hour: {:}]".format(
                        int(self.events[i][1]),
                        int(self.events[i][1] / 24),
                        self.events[i][1] % 24,
                    )
                )
                if self.events[i][2] in [S.MSG_ERROR, S.MSG_WRNG]:
                    write_screen(msg, file=sys.stderr)
                else:
                    write_screen(msg, file=sys.stdout)
            sys.stdout.flush()
            sys.stderr.flush()

save_events(self, filename)

Saves events registered during execution to file.

Parameters:

Name	Type	Description	Default
filename	string	Filename to save events.	required

Returns:

Type	Description
file	File object.

Source code in comorbuss/community.py

def save_events(self, filename):
    """Saves events registered during execution to file.

    Args:
        filename (string): Filename to save events.

    Returns:
        file: File object.
    """
    f = open(filename, "w")
    for event in self.events:
        f.write(
            "["
            + S.MSG_TEXTS[event[2]]
            + "] "
            + event[0]
            + " [step: {:}, day: {:}, hour: {:}]\n".format(
                int(event[1]), int(event[1] / 24.0), event[1] % 24
            )
        )
    return f

simulate(self)

Runs the simulation, in each step all functions on self.to_execute will run.

Source code in comorbuss/community.py

def simulate(self):
    """Runs the simulation, in each step all functions on self.to_execute will run."""
    if self.initOk:
        self.event_log("Initialization OK, starting simulation", S.MSG_PRGS)
        self.print_events()
        time.sleep(0.5)
        self.time_start = time.time()
        self.time_last_p = self.time_start
        if self.log_progress:
            steps = tqdm(range(self.Nsteps - 1), **self.tqdm_par)
        else:
            steps = range(self.Nsteps - 1)
        # Perform a loop through every step of the simulation
        for _ in steps:
            self.clk.tick()
            for _, function, tod in self.to_execute:
                if self.clk.between_hours(*tod):
                    function()
                    self.last_function = function
            if self.stop_condition:
                self.event_log(
                    "Reached an stop condition, stopping simulation.",
                    S.MSG_PRGS,
                )
                self.parameters["last_step"] = self.clk.step
                break
        self.simOk = True
        time_end = time.time()
        self.time_run = time_end - self.time_start
        self.event_log(
            "Simulation took {:.2f} seconds".format(self.time_run), S.MSG_PRGS
        )
        self.log_results()
    else:
        self.event_log(
            "Initialization failed, please check parameters", S.MSG_ERROR
        )
    if self.print_evnts:
        self.print_events()

store_time_series(self)

Stores current step data on time series for post processing.

Source code in comorbuss/community.py

def store_time_series(self):
    """Stores current step data on time series for post processing."""
    self.pop.states_ts[self.clk.step, :] = self.pop.states
    self.pop.symptoms_ts[self.clk.step, :] = self.pop.symptoms
    self.pop.quarantine_states_ts[self.clk.step, :] = self.pop.quarantine_states
    self.pop.placement_ts[self.clk.step, :] = self.pop.placement
    for module in self.modules:
        if hasattr(module, "store_time_series"):
            module.store_time_series()
    for srvc in self.srv.all_services:
        srvc.guests_ids_ts.append(srvc.guests_ids)

update_position(self)

Updates the position of the particles according to a day cycle

Source code in comorbuss/community.py

def update_position(self):
    """Updates the position of the particles according to a day cycle"""
    if self.clk.between_hours(self.free_hours[0], self.free_hours[1]):
        maskAtHomeNotInQuarantine = (self.pop.placement == S.PLC_HOME) & (
            self.pop.activity == S.ACT_FREE
        )
        self.pop.mark_for_new_position(
            maskAtHomeNotInQuarantine, S.PLC_ENV, S.ACT_FREE
        )
    else:
        maskAtHome = self.pop.placement == S.PLC_ENV
        self.pop.mark_for_new_position(maskAtHome, S.PLC_HOME, S.ACT_FREE)