services.py

services_infrastructure

Provides infrastructure to inialize and store services.

Attributes:

Name	Type	Description
services_parameters	dict	Parameters dictionary used for services initialization.
service	dict	Dict where services can be accessed by name.
all_services	list	List with all services indexed by id.

assign_visitors(self, service_param, comm)

Compute closest service per individual in population.

Parameters:

Name	Type	Description	Default
service_param	dict	Parameters for current service	required

Returns:

Type	Description
np.array	chosen services id for all population

Source code in

def assign_visitors(self, service_param, comm):
    """Compute closest service per individual in population.

    Args:
        service_param (dict): Parameters for current service
    Returns:
        np.array: chosen services id for all population
    """
    service_param["visitors"] = -1 * np.ones(comm.pop.Nparticles, dtype=int)
    if service_param["number"] > 0:
        if "number_per_nbhd" in service_param.keys() or comm.networks:
            if (
                not "number_per_nbhd" in service_param.keys()
                or service_param["number_per_nbhd"] == []
            ):
                service_param["number_per_nbhd"] = [service_param["number"]]
            iids = np.arange(service_param["number"])
            for nbhd in range(len(service_param["number_per_nbhd"])):
                mask_in_nbhd = comm.pop.neighborhood == nbhd
                mask_srvc_in_nbhd = service_param["neighborhood"] == nbhd
                iids_in_nbhd = iids[mask_srvc_in_nbhd]
                service_param["visitors"][mask_in_nbhd] = comm.pop.rng.choice(
                    iids_in_nbhd, np.count_nonzero(mask_in_nbhd)
                )
        else:
            sid = service_param["id"]
            seq_services_ids = comm.geo.services_ids[sid].reshape(-1)
            mytree = cKDTree(comm.geo.centers[seq_services_ids])
            indexes = mytree.query(comm.geo.homes[:, :2])[1]
            home_to_nearest_service_id = np.floor(
                indexes / (comm.geo.dservices[sid] * comm.geo.dservices[sid])
            ).astype(int)
            service_param["visitors"] = home_to_nearest_service_id[comm.pop.home_id]
        maskNotInAgeGroup = ~np.isin(comm.pop.ages, service_param["age_groups"])
        service_param["visitors"][maskNotInAgeGroup] = -1
        for i in range(len(service_param["workers_ids"])):
            service_param["visitors"][
                service_param["workers_ids"][i]
            ] = service_param["workers_instance"][i]
    visitation_period = service_param["visitation_period"]
    service_param["visitation_period"] = S.inf * np.ones(comm.Nparticles)
    mask_visitors = service_param["visitors"] != -1
    service_param["visitation_period"][mask_visitors] = visitation_period
    service_param["rooms"] = np.zeros(service_param["visitors"].shape, dtype=int)
    service_param["rooms"][service_param["workers_ids"]] = service_param[
        "workers_room"
    ]

assign_work(self, service_param, comm)

Prepare workers to be assigned to services, This method ramdomly pools workers from population.workersAvailable and assign them workplaces and shifts.

Parameters:

Name	Type	Description	Default
service_param	dict	Parameters for the service	required

Source code in

def assign_work(self, service_param, comm):
    """Prepare workers to be assigned to services,  This method ramdomly pools
    workers from population.workersAvailable and assign them workplaces and shifts.

    Args:
        service_param (dict): Parameters for the service
    """
    service_param["workers_parameters"] = service_param["workers"]
    service_param["workers_ids"] = []
    service_param["workers_instance"] = []
    service_param["workers_room"] = []
    service_param["workers_shift_start"] = []
    service_param["workers_shift_end"] = []
    service_param["workers_shift"] = []
    service_param["workers_types"] = []
    for wrkr in service_param["workers_parameters"]:
        # Calculate number of shifts and the number of workers to be chosen
        nShifts = np.shape(wrkr["shifts"])[0]
        service_param["number_of_shifts"] = nShifts
        nWorkers = int(wrkr["number"] * service_param["number"])
        if nShifts > 0 and nWorkers > 0:
            # Counts the number of suitable and available workers
            maskAvailableInGroup = np.isin(
                comm.pop.pid, comm.pop.workersAvailable
            ) & np.isin(comm.pop.ages, wrkr["age_groups"])
            nAvailable = np.count_nonzero(maskAvailableInGroup)
            # Limit the number of workers to be chosen accordingly to the number of available workers
            if nWorkers > nAvailable:
                comm.event_log(
                    "There is not enough workers available for {} for service {}, {} is needed but only {} is available.".format(
                        wrkr["name"],
                        service_param["name"],
                        nWorkers,
                        nAvailable,
                    ),
                    S.MSG_WRNG,
                )
                nWorkers = nAvailable
                wrkr["number"] = int(
                    nAvailable / service_param["number"]
                )  # Get the workers
            workersIds = comm.randgencom.choice(
                comm.pop.pid[maskAvailableInGroup], nWorkers, replace=False
            )
            # Delete workers from pool of available workers
            maskDelete = ~np.isin(comm.pop.workersAvailable, workersIds)
            comm.pop.workersAvailable = comm.pop.workersAvailable[maskDelete]
            maskWorkers = np.isin(comm.pop.pid, workersIds)
            comm.pop.workplace_id[maskWorkers] = service_param["id"]

            for j, wid in enumerate(workersIds):
                service_param["workers_ids"].append(wid)
                # Assign workers to workplace sequentially with workplaces (assign N workers to first workplace, then N to second workplace...)
                instance = int(
                    np.min(
                        [
                            np.trunc(j / wrkr["number"]),
                            service_param["number"] - 1,
                        ]
                    )
                )
                # Temporarily assign all workers to first room
                room = 0
                # Assign workers to shift sequentially with workers (first worker to first shift, second worker to second shift...)
                shift_id = j % nShifts
                shift = wrkr["shifts"][j % nShifts]
                type = wrkr["id"]

                # Store selected values
                service_param["workers_instance"].append(instance)
                comm.pop.workplace_instance[wid] = instance
                service_param["workers_room"].append(room)
                comm.pop.workplace_room[wid] = room
                service_param["workers_shift_start"].append(shift[0])
                service_param["workers_shift_end"].append(shift[1])
                service_param["workers_shift"].append(j % nShifts)
                comm.pop.worker_type[wid] = type
                service_param["workers_types"].append(type)
    service_param["workers_ids"] = np.array(service_param["workers_ids"])
    service_param["workers_instance"] = np.array(service_param["workers_instance"])
    service_param["workers_room"] = np.array(service_param["workers_room"])
    service_param["workers_shift_start"] = np.array(
        service_param["workers_shift_start"]
    )
    service_param["workers_shift_end"] = np.array(
        service_param["workers_shift_end"]
    )
    service_param["workers_shift"] = np.array(service_param["workers_shift"])
    service_param["workers_types"] = np.array(service_param["workers_types"])

    if service_param["rooms"] != []:
        self.generate_rooms(service_param, comm)

close_or_open_services(self)

Close or opens services according to the decision and time series set.

Source code in

def close_or_open_services(self):
    """Close or opens services according to the decision and time series set."""
    for srvc in self.all_services:
        if not srvc.service_is_closed:
            if to_close_services(self.comm, srvc.closing_par, srvc):
                srvc.close_service()
                self.event_log("Closing " + srvc.name, S.MSG_EVENT)
        elif srvc.service_is_closed:
            if to_reopen_services(self.comm, srvc.closing_par, srvc):
                srvc.open_service()
                self.event_log("Opening " + srvc.name, S.MSG_EVENT)

services_visit(self)

Executes visitation dynamics for services. The order in which the services are processed is shuffled, so that the mean visitation period is respected for many time steps, independently on the order in which particles are assigned to visit a service.

Source code in

def services_visit(self):
    """Executes visitation dynamics for services. The order in which the
    services are processed is shuffled, so that the mean visitation period
    is respected for many time steps, independently on the order in which
    particles are assigned to visit a service.
    """
    order = np.arange(len(self.all_services), dtype=int)
    self.rng.shuffle(order)
    for index in order:
        srvc = self.all_services[index]
        srvc.update_visit_probab()
        srvc.return_home()
        srvc.visit()

service

Class for services implementation.

Attributes:

Name	Type	Description
name	sting	Name of the service.
id	int	Id of this service.
number	int	Number of instances of this service.
rooms	int	List of rooms in this service.
chosen_instance	np.ndarray	List of the preferred instance of this service for all population (-1 particle will not go to this service).
visit_period	float	Normal period for visitation in days.
base_visit_probab	float	Normal base visitation probability per step of simulation.
visit_probab	float	Current visitation probability per step of simulation.
isol_period	float	Period for visitation for particles in lockdowns.
isol_probab	float	Base visitation probability for particles in lockdown.
visit_duration	float	Duration of visitation in hours.
return_probab	float	Probability for the end of an visitation.
guest_ids	list	List to store temporary guests (not visitors, currently only used to hospitalized particles).
workers_ids	np.ndarray	Id of the workers for this service.
mask_workers	np.ndarray	Boolean mask of the population that identifies workers of this service.
workers_instance	np.ndarray	Instance each worker is assigned, follows workers_ids order.
workers_shift_start	np.ndarray	Time the shift of each worker start, follows workers_ids order.
workers_shift_end	np.ndarray	Time the shift of each worker end, follows workers_ids order.
workers_room	np.ndarray	Room each worker is assigned, follows workers_ids order.
workers_types	np.ndarray	Type of worker for each worker, follows workers_ids order.
workers_parameters	list	Original parameters used on the assignment of the workers.
working_hours	list	Working hours for this service: [opening, closing].
placement	int	Placement tag for use in this service.
close	bool	Stores if service closes with interventions.
service_is_closed	bool	Stores if service is closed.
centersAddress	list	Address of the center point for all instances of this service (used with distance mechanics).
limits	list	Limits for all instances of this service (used with distance mechanics).
service_address	np.ndarray	Preferred address for visitation/work for all particles (used with distance mechanics).
grids	list	List with the internal grids addresses for all instances of this service (used with distance mechanics).
grid_address	np.ndarray	Addresses inside internal grids for all particles used when fixed point in grid is active for this service (used with distance mechanics).
controlledAccess	bool	If false particles can walk inside service (only relevant with distance mechanics)
net_type	int	Type of net generator to be used in this service (used with networks mechanics).
net_par	dict	Parameters for the network generator (used with networks mechanics).

allocate_guests(self, new_guest_id)

Add particles to the list of guests.

Parameters:

Name	Type	Description	Default
new_guest_id	np.ndarray	List of the ids to be added.	required

Source code in

def allocate_guests(self, new_guest_id):
    """Add particles to the list of guests.

    Args:
        new_guest_id (np.ndarray): List of the ids to be added.
    """
    if (not self.service_is_closed) & (len(new_guest_id) > 0):
        self.guests_ids = np.concatenate((self.guests_ids, new_guest_id))

close_service(self)

Closes a service.

Source code in

def close_service(self):
    """Closes a service."""
    self.service_is_closed = True

deallocate_guests(self, remove_guests_ids)

Removes particles from the list of guests.

Parameters:

Name	Type	Description	Default
remove_guests_ids	np.ndarray	List of ids to be removed.	required

Source code in

def deallocate_guests(self, remove_guests_ids):
    """Removes particles from the list of guests.

    Args:
        remove_guests_ids (np.ndarray): List of ids to be removed.
    """
    if len(remove_guests_ids) > 0:
        self.guests_ids = self.guests_ids[
            ~np.isin(self.guests_ids, remove_guests_ids)
        ]

open_service(self)

Opens a service.

Source code in

def open_service(self):
    """Opens a service."""
    self.service_is_closed = False

return_home(self)

Get particles back home after visitation.

Source code in

def return_home(self):
    """Get particles back home after visitation."""
    # Return particles according to return_probab during working hours
    if self.number > 0 and self.in_working_hours():
        probab = self.randgenserv.uniform(0.0, 1.0, size=(self.pop.Nparticles))
        mask_in_service = self.pop.placement == self.placement
        mask_visiting = self.pop.activity == S.ACT_VISIT
        mask_return = (
            mask_in_service & mask_visiting & (probab < self.return_probab)
        )
    # Return all visitors home at the end of working hours
    elif self.number > 0 and not self.in_working_hours():
        mask_in_service = self.pop.placement == self.placement
        mask_visiting = self.pop.activity == S.ACT_VISIT
        mask_return = mask_in_service & mask_visiting
    else:
        return None
    self.return_particles(mask_return)

visit(self)

Randomly chooses free particles to go to their service during business hours.

Source code in

def visit(self):
    """Randomly chooses free particles to go to their service during business hours."""
    # During working hours chooses particles to visit according to visitor_probab and isol_probab
    if self.number > 0 and self.in_working_hours() and not self.service_is_closed:
        # Get particles that can visit service
        mask_not_in_quarantine = ~np.isin(
            self.pop.quarantine_states, self.comm.quarantines.confining_quarantines
        )
        mask_act_free = self.pop.activity <= S.ACT_FREE
        mask_act_work = self.pop.activity == S.ACT_WORK
        mask_could_visit = mask_not_in_quarantine & mask_act_free
        # Select particles to visit, using it's respective probabilities visitor_probab and isol_probab
        mask_chosen_instance = self.instance > -1
        mask_could_visit = mask_could_visit & mask_chosen_instance
        probab = self.randgenserv.uniform(0.0, 1.0, size=(self.pop.Nparticles))
        visit_probab = 1 - ((1 - self.visit_probab) ** self.cumm_visit_probab)
        mask_go = mask_could_visit & (probab < visit_probab)
        # Updates placement, activity and movement for selected particles
        self.get_particles(mask_go, S.ACT_VISIT)
        # Reset cumulative probability for particles that can visit and accumulates for working particles
        self.cumm_visit_probab[mask_act_free | self.mask_workers] = 1
        mask_visitors_working = mask_chosen_instance & mask_act_work
        self.cumm_visit_probab[mask_visitors_working] += 1

work(self)

Gets particles to work and return them home.

Source code in

def work(self):
    """Gets particles to work and return them home."""
    if len(self.workers_ids) > 0:
        maskInShift = self.clk.between_hours_array(
            self.workers_shift_start, self.workers_shift_end
        )
        maskInShift = maskInShift & self.working_days[self.clk.dow]
        try:
            maskShiftEnd = (maskInShift == False) & (
                self.last_mask_in_shift == True
            )
        except AttributeError:
            maskShiftEnd = np.zeros(len(maskInShift), dtype=bool)
        self.last_mask_in_shift = maskInShift

        # Check if particles working in service should return home
        workersAfterShift = self.workers_ids[
            maskShiftEnd
        ]  # Get pid of workers after the end of the shift
        workersActivity = self.pop.activity[
            workersAfterShift
        ]  # Get activity of all workers
        working = workersActivity == S.ACT_WORK  # Select only workers with ACT_WORK
        workersToReturn = workersAfterShift[
            working
        ]  # Select only worker with ACT_WORK and after the end of the shift
        mask_return = np.isin(self.pop.pid, workersToReturn)
        self.return_particles(mask_return)

        if self.no_workers_available:
            self.service_is_closed = False

        # Check if particles should go to work
        if not self.service_is_closed:
            maskInShiftActive = maskInShift & self.workers_active
            workersInShift = self.workers_ids[
                maskInShiftActive
            ]  # Get pid of workers in shift hours
            workersQuarantine = self.pop.quarantine_states[
                workersInShift
            ]  # Get quarantine state of workers
            workersState = self.pop.states[workersInShift]
            maskWorkersAvailable = ~np.isin(
                workersQuarantine, self.comm.quarantines.confining_quarantines
            ) & (workersState != S.STATE_D)
            workersWorking = workersInShift[
                maskWorkersAvailable
            ]  # Select only workers in shift hours and off quarantine
            mask_to_work = np.isin(self.pop.pid, workersWorking) & (
                self.pop.placement != self.placement
            )
            self.get_particles(mask_to_work, S.ACT_WORK)

            # check if there is enough workers in every instance
            if self.in_working_hours():
                instances_w_workers = np.unique(
                    self.base_chosen_instance[workersWorking]
                )
                if len(instances_w_workers) == 0:
                    self.service_is_closed = True
                    if not self.no_workers_available:
                        self.no_workers_available = True
                        self.event_log(
                            "Closing {} service due to all workers in quarantine.".format(
                                self.name
                            ),
                            S.MSG_EVENT,
                        )
                    return
                else:
                    if self.no_workers_available:
                        self.no_workers_available = False
                        self.event_log(
                            "Opening {} service, there is workers back.".format(
                                self.name
                            ),
                            S.MSG_EVENT,
                        )
                for instance in self.instances:
                    if self.instance_is_closed[instance]:
                        if instance in instances_w_workers:
                            self.restore_instances(instance)
                            self.event_log(
                                "Opening a instance of {}, there is workers back.".format(
                                    self.name
                                ),
                                S.MSG_EVENT,
                            )
                            self.instance_is_closed[instance] = False
                    else:
                        if not instance in instances_w_workers:
                            self.reassign_instances(instance, instances_w_workers)
                            self.event_log(
                                "Closing a instance of {} service due to all workers in quarantine".format(
                                    self.name
                                ),
                                S.MSG_EVENT,
                            )
                            self.instance_is_closed[instance] = True

DEFAULTS

Default service settings.

HOSPITALS

Default settings for hospitals

HOTELS

A place to isolate all particles.

MARKETS

Default settings for markets

RESTAURANTS

Default settings for restaurants

SCHOOLS

Default settings for schools

STREET_MARKETS

Default settings for street markets