net_generators.py

airborne

gen(self, mask_alive)

Generates networks in the airborne aerosol transmission context.

Parameters:

Name	Type	Description	Default
mask_alive	np.ndarray	Mask of the alive particles.	required

Returns:

Type	Description
Graph	Generated network.

Source code in comorbuss/net_generators.py

def gen(self, mask_alive):
    """Generates networks in the airborne aerosol transmission context.

    Args:
        mask_alive (np.ndarray): Mask of the alive particles.

    Returns:
        Graph: Generated network.
    """
    nets = []
    infectious_mask = self.pop.states == S.STATE_I
    susceptible_mask = self.pop.states == S.STATE_S
    self.sr_all = (
        self.disease.inf_susceptibility
        / np.max(self.comm.given_parameters["susceptibility"])
        * self.sr_multi
    )
    self.prob_getting_infected[:] = 0
    for instance in range(self.srvc.number):
        for room in self.srvc.rooms:
            particles_mask = (
                (self.pop.placement == self.srvc.placement)
                & (self.srvc.instance == instance)
                & (self.srvc.room == room)
                & mask_alive
            )
            nparticles = np.count_nonzero(particles_mask)
            # Checking whether room has been opened consecutively
            if nparticles > 0:
                if self.room_step[room] != (self.clk.step - 1):
                    self.prob_not_getting_infected_accum[room][:] = 1.0
                    self.initial_room_quanta_concentration[room] = 0.0
                    self.exposure_time[room] = 0.0
                self.room_step[room] = self.clk.step
            # Checking if there are infectious particles
            particle_ids = self.pop.pid[particles_mask]
            inf_mask = infectious_mask & particles_mask
            ninfectious = np.count_nonzero(inf_mask)
            if ninfectious > 0:
                self.exposure_time[room] += self.dt
                # Checking if there are susceptible articles, in which case
                # conections must be made
                sus_mask = susceptible_mask & particles_mask
                nsusceptible = np.count_nonzero(sus_mask)
                if nsusceptible > 0:
                    # Generates connections between all particles
                    net, particle_ids = full_net(nparticles)
                    nets.append((nparticles, particles_mask, net))
                # PS: Assuming that particles leave and enter the
                # room uniformly. That allows for storing "room"
                # quantities scalarly rather than vectorially
                p, C = self.inf_probability_calc(
                    sus_mask, inf_mask, nparticles, room
                )
                self.prob_getting_infected[inf_mask] = 0.0
                self.prob_getting_infected[sus_mask] = p
                # If there were susceptible particles in the room,
                # the probability of not getting infected must be
                # stored for later use, and the room quanta concentration
                # must be kept as initially. If there are no susceptible
                # particles, then the probability of not getting infected
                # must be reset, and room quanta concentration must be
                # stored.
                # PS: Ideally, initial room quanta concentration and
                # exposure time should also be stored individually!
                if nsusceptible == 0:
                    self.prob_not_getting_infected_accum[room][:] = 1.0
                    self.initial_room_quanta_concentration[room] = C
                else:
                    self.prob_not_getting_infected_accum[room][sus_mask] *= 1.0 - p
                    self.prob_not_getting_infected_accum[room][~sus_mask] = 1.0

    return nets

default

gen(self, mask_alive)

Default generator.

Parameters:

Name	Type	Description	Default
srvc	service	Service object.	required
mask_alive	np.ndarray	Mask of the alive particles.	required
seed	np.random	Random number generator.	required

Returns:

Type	Description
Graph	Generated network.

Source code in comorbuss/net_generators.py

def gen(self, mask_alive):
    """Default generator.

    Args:
        srvc (service): Service object.
        mask_alive (np.ndarray): Mask of the alive particles.
        seed (np.random, optional): Random number generator.

    Returns:
        Graph: Generated network.
    """
    nets = []
    for instance in range(self.srvc.number):
        for room in self.srvc.rooms:
            particles_mask = (
                (self.pop.placement == self.srvc.placement)
                & (self.srvc.instance == instance)
                & (self.srvc.room == room)
                & mask_alive
            )
            nparticles = np.count_nonzero(particles_mask)
            if nparticles > 1:
                nets.append(self.gen_net(nparticles, particles_mask))
    return nets

enviroment_net

gen(self, mask_alive)

Generator for networks in the environmental layer.

Parameters:

Name	Type	Description	Default
nparticles	int	Number of nodes in the network.	required
mask_environment	np.ndarray	Mask of particles in the environment.	required
pop	population	Population object.	required

Returns:

Type	Description
Graph	Generated network.

Source code in comorbuss/net_generators.py

def gen(self, mask_alive):
    """Generator for networks in the environmental layer.

    Args:
        nparticles (int): Number of nodes in the network.
        mask_environment (np.ndarray): Mask of particles in the environment.
        pop (population): Population object.

    Returns:
        Graph: Generated network.
    """
    mask_environment = (self.pop.placement == S.PLC_ENV) & mask_alive
    n_particles = np.count_nonzero(mask_environment)
    # Number of contacts based on random walk and infection radius
    n_contacts = int(
        0.5
        * n_particles
        * (n_particles - 1)
        * np.pi
        * self.inf_radii
        * self.inf_radii
    )
    if n_contacts >= n_particles:
        n_contacts = n_particles - 1
    if n_contacts > 0:
        # Get a certain number of unique contact ids for being one of the particles in each contact
        unique_contact_ids = self.rng.choice(
            self.pop.pid[mask_environment], n_contacts, False
        )
        real_n_contacts = len(unique_contact_ids)
        mask_environment[unique_contact_ids] = False
        # Probabilities of contact are given according to data in Table 2 of https://doi.org/10.1016/j.socnet.2007.04.005
        waifw_normed = np.zeros((len(S.WAIFW), np.count_nonzero(mask_environment)))
        for i, waifw_row in enumerate(S.WAIFW):
            waifw_normed[i, :] = np.array(
                norm(waifw_row[self.pop.ages[mask_environment]])
            )
        # Get particles on the other side of contacts
        contact_ids = np.zeros(real_n_contacts, dtype=int)
        for i, unique_contact_id in enumerate(unique_contact_ids):
            contact_ids[i] = self.rng.choice(
                self.pop.pid[mask_environment],
                size=1,
                replace=True,
                p=waifw_normed[self.pop.ages[unique_contact_id]],
            )
        # Make encounters mask
        [unique_ids, inverse_map] = np.unique(
            np.concatenate((unique_contact_ids, contact_ids)), return_inverse=True
        )
        n_particles_in_encounters = len(unique_ids)
        mask_in_encounters = np.full(
            (n_particles_in_encounters, n_particles_in_encounters), False
        )
        mask_in_encounters[
            inverse_map[:real_n_contacts], inverse_map[real_n_contacts:]
        ] = True
        mask_in_encounters = mask_in_encounters | mask_in_encounters.transpose()
        return [
            (
                n_particles_in_encounters,
                np.isin(self.pop.pid, unique_ids),
                mask_in_encounters,
            )
        ]
    else:
        return []

home_net

gen(self, mask_alive)

Generator for networks at home. Generate a network with fixed number of contacts.

Parameters:

Name	Type	Description	Default
nparticles	int	Number of nodes in the network.	required
n_conections	float	Mean number of connections.	required
seed	np.random	Random number generator.	required

Returns:

Type	Description
Graph	Generated network.

Source code in comorbuss/net_generators.py

def gen(self, mask_alive):
    """Generator for networks at home. Generate a network with fixed number of contacts.

    Args:
        nparticles (int): Number of nodes in the network.
        n_conections (float): Mean number of connections.
        seed (np.random): Random number generator.

    Returns:
        Graph: Generated network.
    """
    nets = []
    for home in self.homes.keys():
        mask_particles = (
            (self.pop.placement == S.PLC_HOME)
            & mask_alive
            & self.homes[home]["mask"]
        )
        nparticles = np.count_nonzero(mask_particles)
        if nparticles > 1:
            nets.append(
                (
                    nparticles,
                    mask_particles,
                    adjacency_generator(nparticles, self.mean_contacts, self.rng),
                )
            )
    return nets

hospitals

gen(self, mask_alive)

Generates networks in the hospitals context.

Parameters:

Name	Type	Description	Default
srvc	service	Service object.	required
mask_alive	np.ndarray	Mask of the alive particles.	required
seed	np.random	Random number generator.	required

Returns:

Type	Description
Graph	Generated network.

Source code in comorbuss/net_generators.py

def gen(self, mask_alive):
    """Generates networks in the hospitals context.

    Args:
        srvc (service): Service object.
        mask_alive (np.ndarray): Mask of the alive particles.
        seed (np.random, optional): Random number generator.

    Returns:
        Graph: Generated network.
    """
    nets = []
    if self.has_disease_ward_weight:
        dis_ward_coeff = np.ones(self.pop.Nparticles)
    for instance in range(self.srvc.number):
        expected_visitors_per_instance = (
            self.srvc.expected_visits / self.srvc.number
        )
        for room in self.srvc.rooms:
            particles_mask = (
                (self.pop.placement == self.srvc.placement)
                & (self.srvc.instance == instance)
                & (self.srvc.room == room)
                & mask_alive
            )
            nparticles = np.count_nonzero(particles_mask)
            if nparticles > 1:
                particle_ids = self.pop.pid[particles_mask]
                workers_mask = np.isin(particle_ids, self.srvc.workers_ids)
                # If has Disease workers type of workers on service use them as disease workers
                if self.has_disease_workers:
                    disease_workers_mask = np.isin(
                        particle_ids, self.disease_workers
                    )
                    normal_workers_mask = workers_mask & ~disease_workers_mask
                # Else, if disease_worker_prob is 0 or 1 all workers are disease workers and normal workers
                elif self.disease_worker_prob == 0 or self.disease_worker_prob == 1:
                    normal_workers_mask = workers_mask
                    disease_workers_mask = np.zeros(workers_mask.shape, dtype=bool)
                # Else, if has pool of workers to select disease workers select from pool with disease_worker_prob
                elif self.has_disease_workers_pool:
                    n_disease_workers = math.ceil(
                        self.disease_worker_prob * np.count_nonzero(workers_mask)
                    )
                    # Check if there is workers selected for this instance today
                    try:
                        disease_workers_ids = self.today_disease_workers_ids[
                            (instance, room, self.clk.today)
                        ]
                        disease_workers_mask = np.isin(
                            particle_ids, disease_workers_ids
                        )
                        if np.sum(disease_workers_mask) < n_disease_workers:
                            raise KeyError("")
                    # If not select them from pool
                    except KeyError:
                        workers_ids = particle_ids[workers_mask]
                        n_workers_not_n_pool = math.floor(
                            (1 - self.disease_workers_pool_frac)
                            * np.count_nonzero(workers_mask)
                        )
                        workers_n_pool_ids = workers_ids[n_workers_not_n_pool:]
                        disease_workers_ids = self.rng.choice(
                            workers_n_pool_ids, n_disease_workers, replace=False
                        )
                        self.today_disease_workers_ids[
                            (instance, room, self.clk.today)
                        ] = disease_workers_ids
                        disease_workers_mask = np.isin(
                            particle_ids, disease_workers_ids
                        )
                    normal_workers_mask = workers_mask & ~disease_workers_mask
                # If none of the above select disease workers from the end of the list of workers with disease_worker_prob
                else:
                    n_workers = math.floor(
                        (1 - self.disease_worker_prob)
                        * np.count_nonzero(workers_mask)
                    )
                    normal_workers_ids = particle_ids[workers_mask][:n_workers]
                    normal_workers_mask = np.isin(particle_ids, normal_workers_ids)
                    disease_workers_ids = particle_ids[workers_mask][n_workers:]
                    disease_workers_mask = np.isin(
                        particle_ids, disease_workers_ids
                    )
                guests_mask = np.isin(particle_ids, self.srvc.guests_ids)
                visitors_mask = ~(workers_mask | guests_mask)
                net, particle_ids = empty_net(nparticles)
                # connect workers to workers
                workers_ids = particle_ids[normal_workers_mask]
                net, _ = gen_contacs_single_group(
                    normal_workers_mask, net, self.worker_mean_contacts, self.rng
                )
                # Only generates encounters between disease workers and workers if there is disease workers
                if np.sum(disease_workers_mask) > 0:
                    disease_workers_ids = particle_ids[disease_workers_mask]
                    net, _ = gen_contacs_single_group(
                        disease_workers_mask,
                        net,
                        self.disease_worker_mean_contacts,
                        self.rng,
                    )
                    net = gen_contacs_two_groups(
                        disease_workers_ids,
                        workers_ids,
                        net,
                        self.disease_worker_to_worker_mean_contacts,
                        self.rng,
                    )
                    if self.expose_visitors_to_disease_workers:
                        workers_ids = np.unique(
                            np.concatenate((workers_ids, disease_workers_ids))
                        )
                    if self.has_disease_ward_weight:
                        dis_ward_coeff[
                            np.append(
                                disease_workers_ids, self.srvc.guests_ids
                            ).astype(int)
                        ] = self.disease_ward_weight
                else:
                    disease_workers_ids = workers_ids
                # connect visitors to visitors
                visitors_ids = particle_ids[visitors_mask]
                net, _ = gen_contacs_single_group(
                    visitors_mask,
                    net,
                    self.visitor_mean_contacts,
                    self.rng,
                    expected_visitors_per_instance,
                )
                # connect visitors to workers
                net = gen_contacs_two_groups(
                    visitors_ids,
                    workers_ids,
                    net,
                    self.visitor_to_worker_mean_contacts,
                    self.rng,
                )
                # connect guests to disease workers
                guests_id = particle_ids[guests_mask]
                net = gen_contacs_two_groups(
                    guests_id,
                    disease_workers_ids,
                    net,
                    self.guest_to_disease_worker_mean_contacts,
                    self.rng,
                )
                # Post process and store net
                nets.append((nparticles, particles_mask, net))
    if self.has_disease_ward_weight:
        self.pop.disease.set_coefficient(
            "inf_susceptibility", dis_ward_coeff, "hosp_net"
        )
    return nets

isolate

gen(self, mask_alive)

Isolate all particles.

Parameters:

Name	Type	Description	Default
srvc	service	Service object.	required
mask_alive	np.ndarray	Mask of the alive particles.	required
seed	np.random	Random number generator.	required

Returns:

Type	Description
Graph	Generated network.

Source code in comorbuss/net_generators.py

def gen(self, mask_alive):
    """Isolate all particles.

    Args:
        srvc (service): Service object.
        mask_alive (np.ndarray): Mask of the alive particles.
        seed (np.random, optional): Random number generator.

    Returns:
        Graph: Generated network.
    """
    return []

markets

gen(self, mask_alive)

Generates networks in the markets context.

Parameters:

Name	Type	Description	Default
srvc	service	Service object.	required
mask_alive	np.ndarray	Mask of the alive particles.	required
seed	np.random	Random number generator.	required

Returns:

Type	Description
Graph	Generated network.

Source code in comorbuss/net_generators.py

def gen(self, mask_alive):
    """Generates networks in the markets context.

    Args:
        srvc (service): Service object.
        mask_alive (np.ndarray): Mask of the alive particles.
        seed (np.random, optional): Random number generator.

    Returns:
        Graph: Generated network.
    """
    nets = []
    for instance in range(self.srvc.number):
        expected_visitors_per_instance = (
            self.srvc.expected_visits / self.srvc.number
        )
        for room in self.srvc.rooms:
            particles_mask = (
                (self.pop.placement == self.srvc.placement)
                & (self.srvc.instance == instance)
                & (self.srvc.room == room)
                & mask_alive
            )
            nparticles = np.count_nonzero(particles_mask)
            if nparticles > 1:
                particle_ids = self.pop.pid[particles_mask]
                workers_mask = np.isin(particle_ids, self.srvc.workers_ids)
                workers_ids = particle_ids[workers_mask]
                n_workers = np.sum(workers_mask)
                # If has Cashiers type of workers on service use them as cashiers
                if self.has_cashiers_type:
                    cashiers_mask = np.isin(particle_ids, self.cashiers)
                    cashiers_ids = particle_ids[cashiers_mask]
                    n_cashiers = len(cashiers_ids)
                # Else, if has pool of workers to select cashiers select from pool with cashier_prob
                elif self.has_cashier_pool:
                    n_cashiers = int(np.ceil(self.cashier_prob * n_workers))
                    # Check if there is workers selected for this instance today
                    try:
                        cashiers_ids = self.today_cashiers_ids[
                            (instance, room, self.clk.today)
                        ]
                        cashiers_mask = np.isin(particle_ids, cashiers_ids)
                        if np.sum(cashiers_mask) < n_cashiers:
                            raise KeyError("")
                    # If not select them from pool
                    except KeyError:
                        workers_ids = particle_ids[workers_mask]
                        n_cashiers_n_pool = math.ceil(
                            self.cashiers_pool_frac * np.count_nonzero(workers_mask)
                        )
                        workers_n_pool_ids = workers_ids[:n_cashiers_n_pool]
                        cashiers_ids = self.rng.choice(
                            workers_n_pool_ids, n_cashiers, replace=False
                        )
                        self.today_cashiers_ids[
                            (instance, room, self.clk.today)
                        ] = cashiers_ids
                        cashiers_mask = np.isin(particle_ids, cashiers_ids)
                # If none of the above select cashiers from the end of the list of workers with cashier_prob
                else:
                    n_cashiers = int(np.ceil(self.cashier_prob * n_workers))
                    cashiers_ids = workers_ids[0:n_cashiers]
                    cashiers_mask = np.isin(particle_ids, cashiers_ids)
                visitors_mask = ~workers_mask
                net, particle_ids = empty_net(nparticles)
                # generation of contacts between workers
                workers_ids = particle_ids[workers_mask]
                net, n_workers = gen_contacs_single_group(
                    workers_mask, net, self.worker_mean_contacts, self.rng
                )
                # connect visitors to visitors
                visitors_ids = particle_ids[visitors_mask]
                net, n_visitors = gen_contacs_single_group(
                    visitors_mask,
                    net,
                    self.visitor_mean_contacts,
                    self.rng,
                    expected_visitors_per_instance,
                )
                # connect visitors to workers
                net = gen_contacs_two_groups(
                    visitors_ids,
                    workers_ids,
                    net,
                    self.visitor_to_worker_mean_contacts,
                    self.rng,
                )
                # every visitor particle makes a contact with a cashier
                cashiers_ids = particle_ids[cashiers_mask]
                if n_visitors > 0 and n_cashiers > 0:
                    random_cashiers_ids = self.rng.choice(cashiers_ids, n_visitors)
                    for cashier_id, visitor_id in zip(
                        random_cashiers_ids, visitors_ids
                    ):
                        net = add_edge(net, cashier_id, visitor_id)
                nets.append((nparticles, particles_mask, net))
    return nets

restaurants_1

gen(self, mask_alive)

Generates networks in the restaurants context.

Parameters:

Name	Type	Description	Default
srvc	service	Service object.	required
mask_alive	np.ndarray	Mask of the alive particles.	required
seed	np.random	Random number generator.	required

Returns:

Type	Description
Graph	Generated network.

Source code in comorbuss/net_generators.py

def gen(self, mask_alive):
    """Generates networks in the restaurants context.

    Args:
        srvc (service): Service object.
        mask_alive (np.ndarray): Mask of the alive particles.
        seed (np.random, optional): Random number generator.

    Returns:
        Graph: Generated network.
    """
    nets = []
    for instance in range(self.srvc.number):
        for room in self.srvc.rooms:
            particles_mask = (
                (self.pop.placement == self.srvc.placement)
                & (self.srvc.instance == instance)
                & (self.srvc.room == room)
                & mask_alive
            )
            nparticles = np.count_nonzero(particles_mask)
            if nparticles > 1:
                particle_ids = self.pop.pid[particles_mask]
                workers_mask = np.isin(particle_ids, self.srvc.workers_ids)
                visitors_mask = ~workers_mask
                net, particle_ids = empty_net(nparticles)
                waiter_pct = self.waiter_prob
                n_workers = np.count_nonzero(workers_mask)
                n_waiters = int(np.ceil((waiter_pct * n_workers)))
                workers_ids = particle_ids[workers_mask]
                waiters_ids = workers_ids[0:n_waiters]
                net, _ = gen_contacs_single_group(
                    workers_mask, net, self.workers_mean_contacts, self.rng
                )
                n_visitors = np.count_nonzero(visitors_mask)
                visitors_ids = particle_ids[visitors_mask]
                random_waiter_ids = self.rng.choice(waiters_ids, n_visitors)
                for waiter in waiters_ids:
                    visitor_by_waiter_ids = visitors_ids[
                        np.isin(random_waiter_ids, waiter)
                    ]
                    if len(visitor_by_waiter_ids) > 0:
                        table_holder_id = self.rng.choice(visitor_by_waiter_ids, 1)
                        net = add_edge(net, waiter, table_holder_id[0])
                        net_waiter = nx.cycle_graph(len(visitor_by_waiter_ids))
                        net_waiter = nx.to_numpy_array(net_waiter, dtype=bool)
                        mask_waiter = np.isin(particle_ids, visitor_by_waiter_ids)
                        net = compose_adjacency(net, net_waiter, mask_waiter)
                nets.append((nparticles, particles_mask, net))
    return nets

restaurants_2

gen(self, mask_alive)

Generates networks in the restaurants context.

Parameters:

Name	Type	Description	Default
srvc	service	Service object.	required
mask_alive	np.ndarray	Mask of the alive particles.	required
seed	np.random	Random number generator.	required

Returns:

Type	Description
Graph	Generated network.

Source code in comorbuss/net_generators.py

def gen(self, mask_alive):
    """Generates networks in the restaurants context.

    Args:
        srvc (service): Service object.
        mask_alive (np.ndarray): Mask of the alive particles.
        seed (np.random, optional): Random number generator.

    Returns:
        Graph: Generated network.
    """
    nets = []
    for instance in range(self.srvc.number):
        for room in self.srvc.rooms:
            particles_mask = (
                (self.pop.placement == self.srvc.placement)
                & (self.srvc.instance == instance)
                & (self.srvc.room == room)
                & mask_alive
            )
            nparticles = np.count_nonzero(particles_mask)
            if nparticles > 1:
                particle_ids = self.pop.pid[particles_mask]
                workers_mask = np.isin(particle_ids, self.srvc.workers_ids)
                visitors_mask = ~workers_mask
                net, particle_ids = empty_net(nparticles)
                waiter_pct = self.waiter_prob
                persons_per_table = self.persons_per_table
                n_workers = np.count_nonzero(workers_mask)
                n_waiters = int(np.ceil((waiter_pct * n_workers)))
                workers_ids = particle_ids[workers_mask]
                waiters_ids = workers_ids[0:n_waiters]
                net, _ = gen_contacs_single_group(
                    workers_mask, net, self.workers_mean_contacts, self.rng
                )
                n_visitors = np.count_nonzero(visitors_mask)
                if n_visitors > 0 and n_waiters > 0:
                    visitors_ids = particle_ids[visitors_mask]
                    random_waiter_ids = self.rng.choice(waiters_ids, n_visitors)
                    for waiter in waiters_ids:
                        visitor_by_waiter_ids = visitors_ids[
                            np.isin(random_waiter_ids, waiter)
                        ]
                        n_visitors_by_waiter = len(visitor_by_waiter_ids)
                        if n_visitors_by_waiter > 0:
                            for start in range(
                                0, n_visitors_by_waiter, persons_per_table
                            ):
                                end = np.minimum(
                                    start + persons_per_table, n_visitors_by_waiter
                                )
                                table_net, _ = full_net(end - start)
                                mask_table = np.isin(
                                    particle_ids, visitor_by_waiter_ids[start:end]
                                )
                                net = compose_adjacency(net, table_net, mask_table)
                            for visitor_id in visitor_by_waiter_ids:
                                net = add_edge(net, waiter, visitor_id)
                nets.append((nparticles, particles_mask, net))
    return nets

schools

gen(self, mask_alive)

Generates networks in the schools context.

Parameters:

Name	Type	Description	Default
srvc	service	Service object.	required
mask_alive	np.ndarray	Mask of the alive particles.	required
seed	np.random	Random number generator.	required

Returns:

Type	Description
Graph	Generated network.

Source code in comorbuss/net_generators.py

def gen(self, mask_alive):
    """Generates networks in the schools context.

    Args:
        srvc (service): Service object.
        mask_alive (np.ndarray): Mask of the alive particles.
        seed (np.random, optional): Random number generator.

    Returns:
        Graph: Generated network.
    """
    nets = []
    for instance in range(self.srvc.number):
        break_period = self.break_period
        classrooms_per_break = self.break_classrooms
        if self.clk.between_hours(
            self.srvc.working_hours[0], self.srvc.working_hours[0] + break_period
        ):
            # random generation of contacts between all particles in same school (classroom?)
            # To do: improve break period
            jRange = [
                self.srvc.rooms[
                    i : np.minimum(i + classrooms_per_break, len(self.srvc.rooms))
                ]
                for i in range(0, len(self.srvc.rooms), classrooms_per_break)
            ]
            for j in jRange:
                particles_mask = (
                    (self.pop.placement == self.srvc.placement)
                    & (self.srvc.instance == instance)
                    & (np.isin(self.srvc.room, j))
                    & mask_alive
                )
                nparticles = np.count_nonzero(particles_mask)
                if nparticles > 1:
                    contacts_per_particle = self.break_mean_contacts
                    net = adjacency_generator(
                        nparticles, contacts_per_particle, self.rng
                    )
                    nets.append((nparticles, particles_mask, net))
        else:
            for room in self.srvc.rooms:
                particles_mask = (
                    (self.pop.placement == self.srvc.placement)
                    & (self.srvc.instance == instance)
                    & (self.srvc.room == room)
                    & mask_alive
                )
                nparticles = np.count_nonzero(particles_mask)
                if nparticles > 1:
                    net, _ = empty_net_nx(nparticles)
                    # students are positioned in a 2d grid, which then generates the
                    # contacts according to the neighbors
                    students_x = int(np.sqrt(nparticles))
                    students_y = students_x + 1
                    students_xy = students_x * students_y
                    net2d = nx.grid_2d_graph(students_x, students_y)
                    mapping = dict(zip(net2d, range(students_xy)))
                    net2d = nx.relabel_nodes(net2d, mapping)
                    for node in range(nparticles, students_xy):
                        net2d.remove_node(node)
                    net = nx.compose(net, net2d)
                    net.remove_edges_from(nx.selfloop_edges(net))
                    net = nx.to_numpy_array(net, dtype=bool)
                    nets.append((nparticles, particles_mask, net))
    return nets

street_markets

gen(self, mask_alive)

Generates networks in the street markets context.

Parameters:

Name	Type	Description	Default
srvc	service	Service object.	required
mask_alive	np.ndarray	Mask of the alive particles.	required
seed	np.random	Random number generator.	required

Returns:

Type	Description
Graph	Generated network.

Source code in comorbuss/net_generators.py

def gen(self, mask_alive):
    """Generates networks in the street markets context.

    Args:
        srvc (service): Service object.
        mask_alive (np.ndarray): Mask of the alive particles.
        seed (np.random, optional): Random number generator.

    Returns:
        Graph: Generated network.
    """
    nets = []
    for instance in range(self.srvc.number):
        expected_visitors_per_instance = (
            self.srvc.expected_visits / self.srvc.number
        )
        for room in self.srvc.rooms:
            particles_mask = (
                (self.pop.placement == self.srvc.placement)
                & (self.srvc.instance == instance)
                & (self.srvc.room == room)
                & mask_alive
            )
            nparticles = np.count_nonzero(particles_mask)
            if nparticles > 1:
                particle_ids = self.pop.pid[particles_mask]
                workers_mask = np.isin(particle_ids, self.srvc.workers_ids)
                visitors_mask = ~workers_mask
                net, particle_ids = empty_net(nparticles)
                # random generation of contacts between sellers inside street market
                sellers_id = particle_ids[workers_mask]
                net, n_sellers = gen_contacs_single_group(
                    workers_mask, net, self.seller_mean_contacts, self.rng
                )
                # random generation of contacts between sellers inside street market
                visitors_id = particle_ids[visitors_mask]
                net, n_visitors = gen_contacs_single_group(
                    visitors_mask,
                    net,
                    self.visitor_mean_contacts,
                    self.rng,
                    expected_visitors_per_instance,
                )
                # every visitor particle makes a few contacts with a seller
                if (n_visitors > 0) and (n_sellers > 0):
                    n_choice_samples = round(
                        self.rng.normal(
                            self.visitor_to_seller_mean_contacts * n_visitors, 0.25
                        )
                    )
                    if n_choice_samples > 0:
                        random_seller_ids = self.rng.choice(
                            sellers_id, n_choice_samples
                        )
                        random_visitors_ids = self.rng.choice(
                            visitors_id, n_choice_samples
                        )
                        for seller, visitor_id in zip(
                            random_seller_ids, random_visitors_ids
                        ):
                            net = add_edge(net, seller, visitor_id)
                nets.append((nparticles, particles_mask, net))
    return nets

adjacency_generator(nodes, mean_conn, generator)

Generates an random adjacency matrix following the mean number of connections.

Parameters:

Name	Type	Description	Default
nodes	int	Number of nodes.	required
mean_conn	float	Mean number of connections per node.	required
generator	np.random.Generator	Random number generator	required

Returns:

Type	Description
np.array	Boolean adjacency matrix.

Source code in comorbuss/net_generators.py

def adjacency_generator(nodes, mean_conn, generator):
    """Generates an random adjacency matrix following the mean number of connections.

    Args:
        nodes (int): Number of nodes.
        mean_conn (float): Mean number of connections per node.
        generator (np.random.Generator): Random number generator

    Returns:
        np.array: Boolean adjacency matrix.
    """
    probab = generator.uniform(0, 1, size=(nodes, nodes))
    return _adjacency_generator(nodes, mean_conn, probab)

empty_net(n)

Generate an adjacency mask with no connections.

Parameters:

Name	Type	Description	Default
n	int	Number of nodes in the network.	required

Returns:

Type	Description
np.array	nxn adjacency mask.

Source code in comorbuss/net_generators.py

def empty_net(n):
    """Generate an adjacency mask with no connections.

    Args:
        n (int): Number of nodes in the network.

    Returns:
        np.array: nxn adjacency mask.
    """
    net = np.zeros((n, n), dtype=bool)
    nodes = np.arange(n)
    return net, nodes