quarantines.py

quarantines

Provides infrastructure to initialize and store quarantines policies.

__init__(self, qrnt_param, comm) special

Initialize quarantines.

Parameters:

Name	Type	Description	Default
parameters	dict	Parameters for the simulation.	required
comm	community	The community object.	required

Returns:

Type	Description
List	List of the quarantine objects.

Source code in

def __init__(self, qrnt_param, comm):
    """Initialize quarantines.

    Args:
        parameters (dict): Parameters for the simulation.
        comm (community): The community object.

    Returns:
        List: List of the quarantine objects.
    """
    super(quarantines, self).__init__(qrnt_param, comm, name=__name__)

    self.quarantines = []
    self.confining_quarantines = []
    for i, qrnt_param in enumerate(qrnt_param):
        qrnt_param["id"] = i
        qrnt_param["filter_in"], fi_ok = filter_parse(
            qrnt_param["filter_in"],
            comm,
            allowed_filters=["tracing", "service", "workers"],
        )
        qrnt_param["filter_out"], fo_ok = filter_parse(
            qrnt_param["filter_out"], comm, allowed_filters=["days"]
        )
        plc_ok = True
        service = None
        if type(qrnt_param["placement"]) is str:
            plc_ok = False
            if comm.srv.exists(qrnt_param["placement"]):
                service = comm.srv[qrnt_param["placement"]]
                qrnt_param["placement"] = comm.srv[
                    qrnt_param["placement"]
                ].placement

                plc_ok = True
        if plc_ok and fi_ok and fo_ok:
            qrnt = quarantine(qrnt_param, comm, service)
            self.quarantines.append(qrnt)
            if qrnt.confine_particles:
                self.confining_quarantines.append(i)
        elif not plc_ok:
            comm.event_log(
                "Can't activate {} quarantine protocol, service {} doesn't exists.".format(
                    qrnt_param["name"], qrnt_param["placement"]
                ),
                S.MSG_WRNG,
            )
        else:
            comm.event_log(
                "Can't activate {} quarantine protocol, there is an invalid filter.".format(
                    qrnt_param["name"]
                ),
                S.MSG_WRNG,
            )
    self.quarantines.reverse()

    if len(self.quarantines) > 0:
        self.comm.add_to_execution(self.quarantine, priority=50)

    self.list = self.quarantines

quarantine(self)

Runs quarantines policies set on self.quarantines and checks if there isn't any particle violating a quarantine.

Source code in

def quarantine(self):
    """Runs quarantines policies set on self.quarantines and checks if there isn't any particle
    violating a quarantine.
    """
    for qrnt in self.quarantines:
        qrnt()
        if self.comm.show_warn_msgs:
            mask_violating_this = (
                self.pop.quarantine_states == qrnt.qrnt_state
            ) & (self.pop.placement != qrnt.placement)
            try:
                mask_violating_qrnt = mask_violating_qrnt & mask_violating_this
            except:
                mask_violating_qrnt = mask_violating_this

quarantine

This class stores the attributes for a quarantine policy and the methods for its progression.

Attributes:

Name	Type	Description
name	str	Name of the quarantine.
id	int	Id of the quarantine (same as qrnt_state).
filter_in	str	Parsed filter to select particles to start quarantine.
filter_out	str	Parsed filter to select particles to end quarantine.
delay	float	Delay between a particle is selected to start quarantine and when it is effectively started.
qrnt_state	int	Marker to use in the population's quarantine_states for particles in this quarantine.
placement	int	Placement for particles in this quarantine.
service	service	If particles are put in a service in this quarantine this will point to the service.
to_be_quarantined	np.ndarray	A mask to mark particles during the delay to enter quarantine.
time_identified	np.arry	Time when a particle was identified to enter quarantine (used with delay).

__call__(self) special

Runs methods related to this quarantine.

Source code in

def __call__(self):
    """Runs methods related to this quarantine."""
    self.release()
    self.put_in()
    self.sum_quarantined()

__init__(self, qrnt_param, comm, srvc=None) special

Initialize a quarantine policy.

Parameters:

Name	Type	Description	Default
qrnt_param	dict	Parameters related to the quarantine.	required
comm	community	The community object.	required
srvc	service	The service to quarantine particles.	None

Source code in

def __init__(self, qrnt_param, comm, srvc=None):
    """Initialize a quarantine policy.

    Args:
        qrnt_param (dict): Parameters related to the quarantine.
        comm (community): The community object.
        srvc (service, optional): The service to quarantine particles.
    """
    self.comm = comm
    self.pop = comm.pop
    self.clk = comm.clk
    self.event_log = comm.event_log
    self.name = qrnt_param["name"]
    self.id = qrnt_param["id"]
    self.filter_in = qrnt_param["filter_in"]
    self.delay = qrnt_param["delay"]
    self.filter_out = qrnt_param["filter_out"]
    self.qrnt_state = qrnt_param["id"]
    self.placement = qrnt_param["placement"]
    self.confine_particles = qrnt_param["confine_particles"]
    self.service = srvc
    self.to_be_quarantined = np.zeros(self.pop.Nparticles, dtype=bool)
    self.time_identified = np.zeros(self.pop.Nparticles)
    self.quarantined = np.zeros(self.comm.Nsteps)

    self.no_requarantine = not qrnt_param["allow_requarantine"]
    self.already_quarantined = np.zeros(self.pop.Nparticles, dtype=bool)

put_in(self)

Evaluates filters and put selected particles in quarantine.

Source code in

def put_in(self):
    """Evaluates filters and put selected particles in quarantine."""
    mask_not_in_higher_qrnt = ~(
        (self.pop.quarantine_states < self.id)
        & (self.pop.quarantine_states != S.QRNT_FREE)
    )
    mask_quarantanable = (
        mask_not_in_higher_qrnt
        & ~self.get_in_this_quarantine()
        & ~(self.already_quarantined & self.no_requarantine)
    )
    try:
        mask_to_quarantine = (
            mask_quarantanable & eval(self.filter_in) & ~self.to_be_quarantined
        )
    except:
        self.raise_filters(self.filter_in, "filter_in")
    if self.delay > 0:
        # Mark particles identified on this step on to_be_quarantined array
        self.to_be_quarantined[mask_to_quarantine] = True
        self.time_identified[mask_to_quarantine] = self.clk.time

        # Get particles that the delay already passed and unmark them on to_be_quarantined array
        mask_time = self.clk.time_since(self.time_identified) >= self.delay
        mask_to_quarantine = self.to_be_quarantined & mask_time
        self.to_be_quarantined[mask_to_quarantine] = False
        mask_to_quarantine = mask_to_quarantine & mask_not_in_higher_qrnt
        if np.sum(mask_to_quarantine) > 0 and qrnt_debug:
            self.event_log(
                "{} in: {} {}".format(
                    self.name,
                    self.clk.time_to_days(
                        self.clk.time_since(
                            self.time_identified[mask_to_quarantine]
                        )
                    ),
                    self.pop.pid[mask_to_quarantine],
                ),
                S.MSG_PRGS,
            )
    else:
        if np.sum(mask_to_quarantine) > 0 and qrnt_debug:
            self.event_log(
                "{} in: {} {}".format(
                    self.name,
                    np.zeros(np.count_nonzero(mask_to_quarantine)),
                    self.pop.pid[mask_to_quarantine],
                ),
                S.MSG_PRGS,
            )
    self.pop.quarantine_states[mask_to_quarantine] = self.qrnt_state
    self.pop.mark_for_new_position(mask_to_quarantine, self.placement, S.ACT_QRNT)
    self.pop.time_quarantined[mask_to_quarantine] = self.clk.time
    self.pop.update_inf_tree_attributes(
        mask_to_quarantine, "time_quarantined", self.clk.time
    )
    self.pop.update_inf_tree_attributes(
        mask_to_quarantine, "quarantine_id", self.id
    )
    if self.service != None:
        self.service.allocate_guests(self.pop.pid[mask_to_quarantine])
    self.already_quarantined[mask_to_quarantine] = True

release(self)

Evaluates filters and release selected particles from quarantine.

Source code in

def release(self):
    """Evaluates filters and release selected particles from quarantine."""
    mask_in_this_quarantine = self.get_in_this_quarantine()
    try:
        mask_to_release = mask_in_this_quarantine & eval(self.filter_out)
    except:
        self.raise_filters(self.filter_out, "filter_out")
    if np.sum(mask_to_release) > 0 and qrnt_debug:
        self.event_log(
            "{} out: {} {}".format(
                self.name,
                self.clk.time_to_days(
                    self.clk.time_since(self.pop.time_quarantined[mask_to_release])
                ),
                self.pop.pid[mask_to_release],
            ),
            S.MSG_PRGS,
        )
    self.pop.quarantine_states[mask_to_release] = S.QRNT_FREE
    self.pop.mark_for_new_position(mask_to_release, S.PLC_HOME, S.ACT_FREE)
    self.pop.update_inf_tree_attributes(
        mask_to_release, "time_released_quarantine", self.clk.time
    )
    if self.service != None:
        self.service.deallocate_guests(self.pop.pid[mask_to_release])

DIAGNOSTICS

Diagnosed particles will quarantine at home 1 day after the diagnostic and stay quarantined for 14 days.

HOSPITALIZATION

All symptomatic severe particles will imediatly be quarantined on Hospitals and stay quarantined until the particle recovers.

SYMPTOMS

Symptomatic particles will stay home until they recover.

TRACING

Particles with tracing capabilities (see tracing_percent) that had contact with diagnosed particles also with tracing capabilities in the last 5 days will be quarantined at home 1 day after they are identified and stay quarantined for 14 days.