Activity manager

The Activity Manager is the class which allocates what each person (or rather subgroup of people) will be doing each timestep.

Currently, built into the file activity_manager.py is the activity hierarchy, list of activities.

The list of activities are actually loaded via the Timer.from_file() method, and held by the Timer class. The Timer than passes a list of activities for the allowed time of day & day of the week to the ActivityManager. The ActivityManager then picks which subgroups each person will go join according to what activity they are assigned.

It has the important method self.do_timestep(). This method calls the other method move_people_to_active_subgroups, which in turn calls move_to_active_subgroup for each person to actually send people to their active subgroups.

Once it sends folk to their subgroup, that'll set up for applying the contact matrices in InteractiveGroup.

ActivityManager

Source code in june/activity/activity_manager.py

class ActivityManager:
    """ """
    def __init__(
        self,
        world,
        policies,
        timer,
        all_activities,
        activity_to_super_groups: dict,
        record: Optional[Record] = None,
        leisure: Optional[Leisure] = None,
        travel: Optional[Travel] = None,
        sexual_encounter: Optional[SexualEncounter] = None,
    ):
        self.policies = policies
        if self.policies is not None:
            self.policies.init_policies(world=world, date=timer.date, record=record)
        self.world = world
        self.timer = timer
        self.leisure = leisure
        self.travel = travel
        self.sexual_encounter = sexual_encounter
        self.all_activities = all_activities

        self.activity_to_super_group_dict = {
            "medical_facility": activity_to_super_groups.get("medical_facility", []),
            "international_travel": activity_to_super_groups.get("international_travel", []),
            "primary_activity": activity_to_super_groups.get("primary_activity", []),
            "leisure": activity_to_super_groups.get("leisure", []),
            "residence": activity_to_super_groups.get("residence", []),
            "commute": activity_to_super_groups.get("commute", []),
            "rail_travel": activity_to_super_groups.get("rail_travel", []),
            "sexual_encounter": activity_to_super_groups.get("sexual_encounter", []),
        }

    @classmethod
    def from_file(
        cls,
        config_filename,
        world,
        policies,
        timer,
        record: Optional[Record] = None,
        leisure: Optional[Leisure] = None,
        travel: Optional[Travel] = None,
        sexual_encounter: Optional[SexualEncounter] = None,
    ):
        """

        Args:
            config_filename: 
            world: 
            policies: 
            timer: 
            record (Optional[Record], optional): (Default value = None)
            leisure (Optional[Leisure], optional): (Default value = None)
            travel (Optional[Travel], optional): (Default value = None)
            sexual_encounter (Optional[SexualEncounter], optional): (Default value = None)

        """
        with open(config_filename) as f:
            config = yaml.load(f, Loader=yaml.FullLoader)
        try:
            activity_to_super_groups = config["activity_to_super_groups"]
        except KeyError:
            logger.warning(
                "Activity to groups in config is deprecated"
                "please change it to activity_to_super_groups"
            )
            activity_to_super_groups = config["activity_to_groups"]
        time_config = config["time"]

        cls.check_inputs(time_config)
        weekday_activities = [
            activity for activity in time_config["step_activities"]["weekday"].values()
        ]
        weekend_activities = [
            activity for activity in time_config["step_activities"]["weekend"].values()
        ]
        all_activities = set(
            chain.from_iterable(weekday_activities + weekend_activities)
        )
        return cls(
            world=world,
            policies=policies,
            timer=timer,
            all_activities=all_activities,
            activity_to_super_groups=activity_to_super_groups,
            leisure=leisure,
            travel=travel,
            sexual_encounter=sexual_encounter,
            record=record,
        )

    @staticmethod
    def check_inputs(time_config: dict):
        """Check that the iput time configuration is correct, i.e., activities are among allowed activities
        and days have 24 hours.

        Args:
            time_config (dict): dictionary with time steps configuration

        """

        try:
            assert sum(time_config["step_duration"]["weekday"].values()) == 24
            assert sum(time_config["step_duration"]["weekend"].values()) == 24
        except AssertionError:
            raise SimulatorError(
                "Daily activity durations in config do not add to 24 hours."
            )

        # Check that all groups given in time_config file are in the valid group hierarchy
        all_super_groups = activity_hierarchy
        try:
            for step, activities in time_config["step_activities"]["weekday"].items():
                assert all(group in all_super_groups for group in activities)

            for step, activities in time_config["step_activities"]["weekend"].items():
                assert all(group in all_super_groups for group in activities)
        except AssertionError:
            raise SimulatorError("Config file contains unsupported activity name.")

    @property
    def all_super_groups(self):
        """ """
        return self.activities_to_super_groups(self.all_activities)

    @property
    def active_super_groups(self):
        """ """
        return self.activities_to_super_groups(self.timer.activities)

    @staticmethod
    def apply_activity_hierarchy(activities: list[str]) -> list[str]:
        """Returns a list of activities with the right order, obeying the permanent activity hierarchy
        and shuffling the random one.

        Args:
            activities (list[str]): 

        Returns:
            Ordered list of activities according to hierarchy: 

        """
        # Sort activities according to the activity hierarchy
        activities.sort(key=lambda x: activity_hierarchy.index(x))
        return activities

    def activities_to_super_groups(self, activities: List[str]) -> List[str]:
        """Converts activities into Supergroups, the interaction will run over these Groups.

        Args:
            activities (List[str]): 

        Returns:
            List of groups that are active.: 

        """
        return list(
            chain.from_iterable(
                self.activity_to_super_group_dict[activity] for activity in activities
            )
        )

    def get_personal_subgroup(self, person: "Person", activity: str):
        """

        Args:
            person ("Person"): 
            activity (str): 

        """
        return getattr(person, activity)

    def do_timestep(self, record=None):
        """

        Args:
            record: (Default value = None)

        """
        # get time data
        tick_interaction_timestep = perf_counter()
        date = self.timer.date
        day_type = self.timer.day_type
        activities = self.apply_activity_hierarchy(self.timer.activities)
        delta_time = self.timer.duration
        # apply leisure policies


        if self.leisure is not None:
            if self.policies is not None:
                self.policies.leisure_policies.apply(date=date, leisure=self.leisure)
            self.leisure.generate_leisure_probabilities_for_timestep(
                delta_time=delta_time,
                date=date,
                working_hours="primary_activity" in activities,
            )
       # move people to subgroups and get going abroad people
        to_send_abroad, potential_leisure_inviters, potential_sexual_encounter_inviters = self.move_people_to_active_subgroups(
            activities=activities,
            date=date,
            days_from_start=self.timer.now,
            record=record,
        )

        simulator = GlobalContext.get_simulator()

        # process partner invitations for sexual encounters
        # IMPORTANT: All ranks must participate in partner invitation process for MPI synchronization
        if simulator.sexual_encounters_enabled:
            if self.sexual_encounter is not None and "sexual_encounter" in activities:
                logger.info("Starting partner invitation process...")
                # ALL ranks must participate in the partner invitation process for MPI synchronization
                rejected_people = self.sexual_encounter.process_partner_invitations(potential_sexual_encounter_inviters, self.world)

                # Process rejected people through fallback activities
                if rejected_people:
                    logger.info(f"Processing {len(rejected_people)} rejected people for fallback activities...")
                    fallback_transfers = self.process_rejected_people_fallback(rejected_people, activities)
                    # merge fallback transfers into to_send_abroad
                    self._merge_movable_people(to_send_abroad, fallback_transfers)

                # collect additional transfers from partner acceptances
                additional_transfers = self.collect_sexual_encounter_transfers()
                # manually merge additional transfers into to_send_abroad
                self._merge_movable_people(to_send_abroad, additional_transfers)

        # process friend invitations for leisure activities
        # IMPORTANT: All ranks must participate in friend invitation process for MPI synchronization
        if simulator.friend_hangouts_enabled:
            if self.leisure is not None and "leisure" in activities:
                logger.info("Starting friend invitation process...")
                # ALL ranks must participate in the friend invitation process for MPI synchronization
                self.leisure.process_friend_invitations(potential_leisure_inviters, self.world)

                # collect additional transfers from friend acceptances
                additional_transfers = self.collect_friend_transfers()
                # manually merge additional transfers into to_send_abroad
                self._merge_movable_people(to_send_abroad, additional_transfers)
                logger.info("Friend invitation process finished.")

        tock_interaction_timestep = perf_counter()
        rank_logger.info(
            f"Rank {mpi_rank} -- move_people -- {tock_interaction_timestep-tick_interaction_timestep}"
        )
        tick_waiting = perf_counter()
        mpi_comm.Barrier()
        tock_waiting = perf_counter()
        rank_logger.info(
            f"Rank {mpi_rank} -- move_people_waiting -- {tock_waiting-tick_waiting}"
        )
        (
            people_from_abroad,
            n_people_from_abroad,
            n_people_going_abroad,
        ) = self.send_and_receive_people_from_abroad(to_send_abroad)

        return (
            people_from_abroad,
            n_people_from_abroad,
            n_people_going_abroad,
            to_send_abroad,
        )

    def move_people_to_active_subgroups(
        self,
        activities: List[str],
        date: datetime = datetime(2020, 2, 2),
        days_from_start=0,
        record=None,
    ):
        """Sends every person to one subgroup. If a person has a mild illness,
        they stay at home

        Args:
            activities (List[str]): 
            date (datetime, optional): (Default value = datetime(2020, 2, 2))
            days_from_start: (Default value = 0)
            record: (Default value = None)

        """
        tick = perf_counter()
        active_individual_policies = self.policies.individual_policies.get_active(
            date=date
        )
        to_send_abroad = MovablePeople()
        potential_leisure_inviters = []  # Track people who got assigned to leisure
        potential_sexual_encounter_inviters = []  # Track people who got assigned to sexual encounter
        counter = 0
        for person in self.world.people:
            counter += 1
            if person.dead or person.busy:
                continue
            allowed_activities = self.policies.individual_policies.apply(
                active_policies=active_individual_policies,
                person=person,
                activities=activities,
                days_from_start=days_from_start,
            )

            external_subgroup = self.move_to_active_subgroup(
                allowed_activities, person, to_send_abroad, potential_leisure_inviters, potential_sexual_encounter_inviters
            )
            if external_subgroup is not None:
                to_send_abroad.add_person(person, external_subgroup)

        tock = perf_counter()
        mpi_logger.info(f"{self.timer.date},{mpi_rank},activity,{tock-tick}")
        return to_send_abroad, potential_leisure_inviters, potential_sexual_encounter_inviters

    def collect_friend_transfers(self):
        """Collect people who accepted friend invitations and need to be transferred to other ranks.


        Returns:
            MovablePeople: Container with people who need to be transferred

        """
        additional_transfers = MovablePeople()
        transfer_count = 0

        for person in self.world.people:
            if (person.subgroups.leisure and 
                person.subgroups.leisure.external and 
                not person.busy):  # Not already flagged in normal assignment

                person.busy = True
                additional_transfers.add_person(person, person.subgroups.leisure)
                transfer_count += 1

        return additional_transfers

    def collect_sexual_encounter_transfers(self):
        """Collect people who accepted partner invitations and need to be transferred to other ranks.


        Returns:
            MovablePeople: Container with people who need to be transferred

        """
        additional_transfers = MovablePeople()
        transfer_count = 0

        for person in self.world.people:
            if (person.subgroups.sexual_encounter and 
                person.subgroups.sexual_encounter.external and 
                not person.busy):  # Not already flagged in normal assignment

                person.busy = True
                additional_transfers.add_person(person, person.subgroups.sexual_encounter)
                transfer_count += 1

        return additional_transfers

    def _merge_movable_people(self, primary, additional):
        """Merge additional MovablePeople into the primary MovablePeople object.

        Args:
            primary (MovablePeople): The main container to merge into
            additional (MovablePeople): The additional container to merge from

        """
        merged_domains = 0
        merged_people = 0

        # Merge skinny_out dictionaries
        for domain_id, domain_data in additional.skinny_out.items():

            if domain_id not in primary.skinny_out:
                primary.skinny_out[domain_id] = {}

            for group_spec, spec_data in domain_data.items():
                if group_spec not in primary.skinny_out[domain_id]:
                    primary.skinny_out[domain_id][group_spec] = {}

                for group_id, group_data in spec_data.items():
                    if group_id not in primary.skinny_out[domain_id][group_spec]:
                        primary.skinny_out[domain_id][group_spec][group_id] = {}

                    for subgroup_type, subgroup_data in group_data.items():
                        if subgroup_type not in primary.skinny_out[domain_id][group_spec][group_id]:
                            primary.skinny_out[domain_id][group_spec][group_id][subgroup_type] = {}

                        # Merge person data
                        before_count = len(primary.skinny_out[domain_id][group_spec][group_id][subgroup_type])
                        primary.skinny_out[domain_id][group_spec][group_id][subgroup_type].update(subgroup_data)
                        after_count = len(primary.skinny_out[domain_id][group_spec][group_id][subgroup_type])
                        people_added = after_count - before_count
                        merged_people += people_added

            merged_domains += 1

    def process_rejected_people_fallback(self, rejected_people: List[Person], activities: List[str]) -> MovablePeople:
        """Process people who were rejected from sexual encounters by reassigning them
        through the normal activity hierarchy (excluding sexual_encounter).

        Args:
            rejected_people (List[Person]): People who were rejected and need fallback activities
            activities (List[str]): Current activities for this timestep

        Returns:
            MovablePeople: People who need to be transferred to other ranks for fallback activities

        """
        fallback_transfers = MovablePeople()

        # Create fallback activities excluding sexual_encounter
        fallback_activities = [act for act in activities if act != "sexual_encounter"]

        reassigned_count = 0
        transfer_count = 0

        for person in rejected_people:

            # Use the normal move_to_active_subgroup process for fallback
            external_subgroup = self.move_to_active_subgroup(
                activities=fallback_activities,
                person=person,
                to_send_abroad=fallback_transfers,
                potential_leisure_inviters=None,  # Don't track as inviters for fallback
                potential_sexual_encounter_inviters=None
            )

            if external_subgroup is not None:
                # Person needs to be sent abroad
                fallback_transfers.add_person(person, external_subgroup)
                transfer_count += 1

            reassigned_count += 1

        logger.info(f"Fallback processing complete: {reassigned_count}/{len(rejected_people)} reassigned, {transfer_count} transfers")
        return fallback_transfers

    def move_to_active_subgroup(
        self, activities: List[str], person: Person, to_send_abroad=None, potential_leisure_inviters=None, potential_sexual_encounter_inviters=None
    ) -> Optional["Subgroup"]:
        """Given the hierarchy of activities and a person, decide what subgroup
        should they go to

        Args:
            activities (List[str]): list of activities that take place at a given time step
            person (Person): person that is looking for a subgroup to go to
            to_send_abroad: (Default value = None)
            potential_leisure_inviters: list to track people who get assigned to leisure activities (Default value = None)
            potential_sexual_encounter_inviters: (Default value = None)

        Returns:
            Subgroup to which person has to go, given the hierarchy of activities: 

        """

        for activity in activities:

            if activity == "sexual_encounter":
                # Check if person should engage in sexual encounter using the filter
                if self.sexual_encounter and self.sexual_encounter.should_engage_in_sexual_encounter(person):
                    subgroup = self.get_personal_subgroup(person=person, activity=activity)
                    if subgroup is not None and potential_sexual_encounter_inviters is not None:
                        potential_sexual_encounter_inviters.append(person)
                else:
                    subgroup = None
            elif activity == "leisure" and person.leisure is None:
                subgroup = self.leisure.get_subgroup_for_person_and_housemates(
                    person=person, to_send_abroad=to_send_abroad
                )
                # Track this person as a potential inviter if they got leisure
                if subgroup is not None and potential_leisure_inviters is not None:
                    potential_leisure_inviters.append(person)
            elif activity == "leisure" and person.leisure is not None:
                subgroup = None
            elif activity == "commute":
                subgroup = self.travel.get_commute_subgroup(person=person)
            else:
                subgroup = self.get_personal_subgroup(person=person, activity=activity)
            if subgroup is not None:
                if subgroup.external:
                    person.busy = True
                    # this person goes to another MPI domain
                    return subgroup

                subgroup.append(person)
                return
        raise SimulatorError(
            "Attention! Some people do not have an activity in this timestep."
        )

    def send_and_receive_people_from_abroad(self, movable_people):
        """Deal with the MPI comms.

        Args:
            movable_people: 

        """
        n_people_going_abroad = 0
        n_people_from_abroad = 0
        tick, tickw = perf_counter(), wall_clock()
        reqs = []

        for rank in range(mpi_size):

            if mpi_rank == rank:
                continue
            keys, data, n_this_rank = movable_people.serialise(rank)
            if n_this_rank:
                reqs.append(mpi_comm.isend(keys, dest=rank, tag=100))
                reqs.append(mpi_comm.isend(data, dest=rank, tag=200))
                n_people_going_abroad += n_this_rank
            else:
                reqs.append(mpi_comm.isend(None, dest=rank, tag=100))
                reqs.append(mpi_comm.isend(None, dest=rank, tag=200))

        # now it has all been sent, we can start the receiving.

        for rank in range(mpi_size):

            if rank == mpi_rank:
                continue
            keys = mpi_comm.recv(source=rank, tag=100)
            data = mpi_comm.recv(source=rank, tag=200)

            if keys is not None:
                movable_people.update(rank, keys, data)
                n_people_from_abroad += data.shape[0]

        for r in reqs:
            r.wait()

        tock, tockw = perf_counter(), wall_clock()
        logger.info(
            f"CMS: People COMS for rank {mpi_rank}/{mpi_size} - {tock - tick},{tockw - tickw} - {self.timer.date}"
        )
        mpi_logger.info(f"{self.timer.date},{mpi_rank},people_comms,{tock-tick}")
        return movable_people.skinny_in, n_people_from_abroad, n_people_going_abroad

active_super_groups property

all_super_groups property

activities_to_super_groups(activities)

Converts activities into Supergroups, the interaction will run over these Groups.

Parameters:

Name	Type	Description	Default
activities	List[str]		required

Returns:

Type	Description
List[str]	List of groups that are active.:

Source code in june/activity/activity_manager.py

def activities_to_super_groups(self, activities: List[str]) -> List[str]:
    """Converts activities into Supergroups, the interaction will run over these Groups.

    Args:
        activities (List[str]): 

    Returns:
        List of groups that are active.: 

    """
    return list(
        chain.from_iterable(
            self.activity_to_super_group_dict[activity] for activity in activities
        )
    )

apply_activity_hierarchy(activities) staticmethod

Returns a list of activities with the right order, obeying the permanent activity hierarchy and shuffling the random one.

Parameters:

Name	Type	Description	Default
activities	list[str]		required

Returns:

Type	Description
list[str]	Ordered list of activities according to hierarchy:

Source code in june/activity/activity_manager.py

@staticmethod
def apply_activity_hierarchy(activities: list[str]) -> list[str]:
    """Returns a list of activities with the right order, obeying the permanent activity hierarchy
    and shuffling the random one.

    Args:
        activities (list[str]): 

    Returns:
        Ordered list of activities according to hierarchy: 

    """
    # Sort activities according to the activity hierarchy
    activities.sort(key=lambda x: activity_hierarchy.index(x))
    return activities

check_inputs(time_config) staticmethod

Check that the iput time configuration is correct, i.e., activities are among allowed activities and days have 24 hours.

Parameters:

Name	Type	Description	Default
time_config	dict	dictionary with time steps configuration	required

Source code in june/activity/activity_manager.py

@staticmethod
def check_inputs(time_config: dict):
    """Check that the iput time configuration is correct, i.e., activities are among allowed activities
    and days have 24 hours.

    Args:
        time_config (dict): dictionary with time steps configuration

    """

    try:
        assert sum(time_config["step_duration"]["weekday"].values()) == 24
        assert sum(time_config["step_duration"]["weekend"].values()) == 24
    except AssertionError:
        raise SimulatorError(
            "Daily activity durations in config do not add to 24 hours."
        )

    # Check that all groups given in time_config file are in the valid group hierarchy
    all_super_groups = activity_hierarchy
    try:
        for step, activities in time_config["step_activities"]["weekday"].items():
            assert all(group in all_super_groups for group in activities)

        for step, activities in time_config["step_activities"]["weekend"].items():
            assert all(group in all_super_groups for group in activities)
    except AssertionError:
        raise SimulatorError("Config file contains unsupported activity name.")

collect_friend_transfers()

Collect people who accepted friend invitations and need to be transferred to other ranks.

Returns:

Name	Type	Description
MovablePeople		Container with people who need to be transferred

Source code in june/activity/activity_manager.py

def collect_friend_transfers(self):
    """Collect people who accepted friend invitations and need to be transferred to other ranks.


    Returns:
        MovablePeople: Container with people who need to be transferred

    """
    additional_transfers = MovablePeople()
    transfer_count = 0

    for person in self.world.people:
        if (person.subgroups.leisure and 
            person.subgroups.leisure.external and 
            not person.busy):  # Not already flagged in normal assignment

            person.busy = True
            additional_transfers.add_person(person, person.subgroups.leisure)
            transfer_count += 1

    return additional_transfers

collect_sexual_encounter_transfers()

Collect people who accepted partner invitations and need to be transferred to other ranks.

Returns:

Name	Type	Description
MovablePeople		Container with people who need to be transferred

Source code in june/activity/activity_manager.py

def collect_sexual_encounter_transfers(self):
    """Collect people who accepted partner invitations and need to be transferred to other ranks.


    Returns:
        MovablePeople: Container with people who need to be transferred

    """
    additional_transfers = MovablePeople()
    transfer_count = 0

    for person in self.world.people:
        if (person.subgroups.sexual_encounter and 
            person.subgroups.sexual_encounter.external and 
            not person.busy):  # Not already flagged in normal assignment

            person.busy = True
            additional_transfers.add_person(person, person.subgroups.sexual_encounter)
            transfer_count += 1

    return additional_transfers

do_timestep(record=None)

Parameters:

Name	Type	Description	Default
record		(Default value = None)	None

Source code in june/activity/activity_manager.py

def do_timestep(self, record=None):
    """

    Args:
        record: (Default value = None)

    """
    # get time data
    tick_interaction_timestep = perf_counter()
    date = self.timer.date
    day_type = self.timer.day_type
    activities = self.apply_activity_hierarchy(self.timer.activities)
    delta_time = self.timer.duration
    # apply leisure policies


    if self.leisure is not None:
        if self.policies is not None:
            self.policies.leisure_policies.apply(date=date, leisure=self.leisure)
        self.leisure.generate_leisure_probabilities_for_timestep(
            delta_time=delta_time,
            date=date,
            working_hours="primary_activity" in activities,
        )
   # move people to subgroups and get going abroad people
    to_send_abroad, potential_leisure_inviters, potential_sexual_encounter_inviters = self.move_people_to_active_subgroups(
        activities=activities,
        date=date,
        days_from_start=self.timer.now,
        record=record,
    )

    simulator = GlobalContext.get_simulator()

    # process partner invitations for sexual encounters
    # IMPORTANT: All ranks must participate in partner invitation process for MPI synchronization
    if simulator.sexual_encounters_enabled:
        if self.sexual_encounter is not None and "sexual_encounter" in activities:
            logger.info("Starting partner invitation process...")
            # ALL ranks must participate in the partner invitation process for MPI synchronization
            rejected_people = self.sexual_encounter.process_partner_invitations(potential_sexual_encounter_inviters, self.world)

            # Process rejected people through fallback activities
            if rejected_people:
                logger.info(f"Processing {len(rejected_people)} rejected people for fallback activities...")
                fallback_transfers = self.process_rejected_people_fallback(rejected_people, activities)
                # merge fallback transfers into to_send_abroad
                self._merge_movable_people(to_send_abroad, fallback_transfers)

            # collect additional transfers from partner acceptances
            additional_transfers = self.collect_sexual_encounter_transfers()
            # manually merge additional transfers into to_send_abroad
            self._merge_movable_people(to_send_abroad, additional_transfers)

    # process friend invitations for leisure activities
    # IMPORTANT: All ranks must participate in friend invitation process for MPI synchronization
    if simulator.friend_hangouts_enabled:
        if self.leisure is not None and "leisure" in activities:
            logger.info("Starting friend invitation process...")
            # ALL ranks must participate in the friend invitation process for MPI synchronization
            self.leisure.process_friend_invitations(potential_leisure_inviters, self.world)

            # collect additional transfers from friend acceptances
            additional_transfers = self.collect_friend_transfers()
            # manually merge additional transfers into to_send_abroad
            self._merge_movable_people(to_send_abroad, additional_transfers)
            logger.info("Friend invitation process finished.")

    tock_interaction_timestep = perf_counter()
    rank_logger.info(
        f"Rank {mpi_rank} -- move_people -- {tock_interaction_timestep-tick_interaction_timestep}"
    )
    tick_waiting = perf_counter()
    mpi_comm.Barrier()
    tock_waiting = perf_counter()
    rank_logger.info(
        f"Rank {mpi_rank} -- move_people_waiting -- {tock_waiting-tick_waiting}"
    )
    (
        people_from_abroad,
        n_people_from_abroad,
        n_people_going_abroad,
    ) = self.send_and_receive_people_from_abroad(to_send_abroad)

    return (
        people_from_abroad,
        n_people_from_abroad,
        n_people_going_abroad,
        to_send_abroad,
    )

from_file(config_filename, world, policies, timer, record=None, leisure=None, travel=None, sexual_encounter=None) classmethod

Parameters:

Name	Type	Description	Default
config_filename			required
world			required
policies			required
timer			required
record	Optional[Record]	(Default value = None)	None
leisure	Optional[Leisure]	(Default value = None)	None
travel	Optional[Travel]	(Default value = None)	None
sexual_encounter	Optional[SexualEncounter]	(Default value = None)	None

Source code in june/activity/activity_manager.py

@classmethod
def from_file(
    cls,
    config_filename,
    world,
    policies,
    timer,
    record: Optional[Record] = None,
    leisure: Optional[Leisure] = None,
    travel: Optional[Travel] = None,
    sexual_encounter: Optional[SexualEncounter] = None,
):
    """

    Args:
        config_filename: 
        world: 
        policies: 
        timer: 
        record (Optional[Record], optional): (Default value = None)
        leisure (Optional[Leisure], optional): (Default value = None)
        travel (Optional[Travel], optional): (Default value = None)
        sexual_encounter (Optional[SexualEncounter], optional): (Default value = None)

    """
    with open(config_filename) as f:
        config = yaml.load(f, Loader=yaml.FullLoader)
    try:
        activity_to_super_groups = config["activity_to_super_groups"]
    except KeyError:
        logger.warning(
            "Activity to groups in config is deprecated"
            "please change it to activity_to_super_groups"
        )
        activity_to_super_groups = config["activity_to_groups"]
    time_config = config["time"]

    cls.check_inputs(time_config)
    weekday_activities = [
        activity for activity in time_config["step_activities"]["weekday"].values()
    ]
    weekend_activities = [
        activity for activity in time_config["step_activities"]["weekend"].values()
    ]
    all_activities = set(
        chain.from_iterable(weekday_activities + weekend_activities)
    )
    return cls(
        world=world,
        policies=policies,
        timer=timer,
        all_activities=all_activities,
        activity_to_super_groups=activity_to_super_groups,
        leisure=leisure,
        travel=travel,
        sexual_encounter=sexual_encounter,
        record=record,
    )

get_personal_subgroup(person, activity)

Parameters:

Name	Type	Description	Default
person	Person		required
activity	str		required

Source code in june/activity/activity_manager.py

def get_personal_subgroup(self, person: "Person", activity: str):
    """

    Args:
        person ("Person"): 
        activity (str): 

    """
    return getattr(person, activity)

move_people_to_active_subgroups(activities, date=datetime(2020, 2, 2), days_from_start=0, record=None)

Sends every person to one subgroup. If a person has a mild illness, they stay at home

Parameters:

Name	Type	Description	Default
activities	List[str]		required
date	datetime	(Default value = datetime(2020, 2, 2))	datetime(2020, 2, 2)
days_from_start		(Default value = 0)	0
record		(Default value = None)	None

Source code in june/activity/activity_manager.py

def move_people_to_active_subgroups(
    self,
    activities: List[str],
    date: datetime = datetime(2020, 2, 2),
    days_from_start=0,
    record=None,
):
    """Sends every person to one subgroup. If a person has a mild illness,
    they stay at home

    Args:
        activities (List[str]): 
        date (datetime, optional): (Default value = datetime(2020, 2, 2))
        days_from_start: (Default value = 0)
        record: (Default value = None)

    """
    tick = perf_counter()
    active_individual_policies = self.policies.individual_policies.get_active(
        date=date
    )
    to_send_abroad = MovablePeople()
    potential_leisure_inviters = []  # Track people who got assigned to leisure
    potential_sexual_encounter_inviters = []  # Track people who got assigned to sexual encounter
    counter = 0
    for person in self.world.people:
        counter += 1
        if person.dead or person.busy:
            continue
        allowed_activities = self.policies.individual_policies.apply(
            active_policies=active_individual_policies,
            person=person,
            activities=activities,
            days_from_start=days_from_start,
        )

        external_subgroup = self.move_to_active_subgroup(
            allowed_activities, person, to_send_abroad, potential_leisure_inviters, potential_sexual_encounter_inviters
        )
        if external_subgroup is not None:
            to_send_abroad.add_person(person, external_subgroup)

    tock = perf_counter()
    mpi_logger.info(f"{self.timer.date},{mpi_rank},activity,{tock-tick}")
    return to_send_abroad, potential_leisure_inviters, potential_sexual_encounter_inviters

move_to_active_subgroup(activities, person, to_send_abroad=None, potential_leisure_inviters=None, potential_sexual_encounter_inviters=None)

Given the hierarchy of activities and a person, decide what subgroup should they go to

Parameters:

Name	Type	Description	Default
activities	List[str]	list of activities that take place at a given time step	required
person	Person	person that is looking for a subgroup to go to	required
to_send_abroad		(Default value = None)	None
potential_leisure_inviters		list to track people who get assigned to leisure activities (Default value = None)	None
potential_sexual_encounter_inviters		(Default value = None)	None

Returns:

Type	Description
Optional[Subgroup]	Subgroup to which person has to go, given the hierarchy of activities:

Source code in june/activity/activity_manager.py

def move_to_active_subgroup(
    self, activities: List[str], person: Person, to_send_abroad=None, potential_leisure_inviters=None, potential_sexual_encounter_inviters=None
) -> Optional["Subgroup"]:
    """Given the hierarchy of activities and a person, decide what subgroup
    should they go to

    Args:
        activities (List[str]): list of activities that take place at a given time step
        person (Person): person that is looking for a subgroup to go to
        to_send_abroad: (Default value = None)
        potential_leisure_inviters: list to track people who get assigned to leisure activities (Default value = None)
        potential_sexual_encounter_inviters: (Default value = None)

    Returns:
        Subgroup to which person has to go, given the hierarchy of activities: 

    """

    for activity in activities:

        if activity == "sexual_encounter":
            # Check if person should engage in sexual encounter using the filter
            if self.sexual_encounter and self.sexual_encounter.should_engage_in_sexual_encounter(person):
                subgroup = self.get_personal_subgroup(person=person, activity=activity)
                if subgroup is not None and potential_sexual_encounter_inviters is not None:
                    potential_sexual_encounter_inviters.append(person)
            else:
                subgroup = None
        elif activity == "leisure" and person.leisure is None:
            subgroup = self.leisure.get_subgroup_for_person_and_housemates(
                person=person, to_send_abroad=to_send_abroad
            )
            # Track this person as a potential inviter if they got leisure
            if subgroup is not None and potential_leisure_inviters is not None:
                potential_leisure_inviters.append(person)
        elif activity == "leisure" and person.leisure is not None:
            subgroup = None
        elif activity == "commute":
            subgroup = self.travel.get_commute_subgroup(person=person)
        else:
            subgroup = self.get_personal_subgroup(person=person, activity=activity)
        if subgroup is not None:
            if subgroup.external:
                person.busy = True
                # this person goes to another MPI domain
                return subgroup

            subgroup.append(person)
            return
    raise SimulatorError(
        "Attention! Some people do not have an activity in this timestep."
    )

process_rejected_people_fallback(rejected_people, activities)

Process people who were rejected from sexual encounters by reassigning them through the normal activity hierarchy (excluding sexual_encounter).

Parameters:

Name	Type	Description	Default
rejected_people	List[Person]	People who were rejected and need fallback activities	required
activities	List[str]	Current activities for this timestep	required

Returns:

Name	Type	Description
MovablePeople	MovablePeople	People who need to be transferred to other ranks for fallback activities

Source code in june/activity/activity_manager.py

def process_rejected_people_fallback(self, rejected_people: List[Person], activities: List[str]) -> MovablePeople:
    """Process people who were rejected from sexual encounters by reassigning them
    through the normal activity hierarchy (excluding sexual_encounter).

    Args:
        rejected_people (List[Person]): People who were rejected and need fallback activities
        activities (List[str]): Current activities for this timestep

    Returns:
        MovablePeople: People who need to be transferred to other ranks for fallback activities

    """
    fallback_transfers = MovablePeople()

    # Create fallback activities excluding sexual_encounter
    fallback_activities = [act for act in activities if act != "sexual_encounter"]

    reassigned_count = 0
    transfer_count = 0

    for person in rejected_people:

        # Use the normal move_to_active_subgroup process for fallback
        external_subgroup = self.move_to_active_subgroup(
            activities=fallback_activities,
            person=person,
            to_send_abroad=fallback_transfers,
            potential_leisure_inviters=None,  # Don't track as inviters for fallback
            potential_sexual_encounter_inviters=None
        )

        if external_subgroup is not None:
            # Person needs to be sent abroad
            fallback_transfers.add_person(person, external_subgroup)
            transfer_count += 1

        reassigned_count += 1

    logger.info(f"Fallback processing complete: {reassigned_count}/{len(rejected_people)} reassigned, {transfer_count} transfers")
    return fallback_transfers

send_and_receive_people_from_abroad(movable_people)

Deal with the MPI comms.

Parameters:

Name	Type	Description	Default
movable_people			required

Source code in june/activity/activity_manager.py

def send_and_receive_people_from_abroad(self, movable_people):
    """Deal with the MPI comms.

    Args:
        movable_people: 

    """
    n_people_going_abroad = 0
    n_people_from_abroad = 0
    tick, tickw = perf_counter(), wall_clock()
    reqs = []

    for rank in range(mpi_size):

        if mpi_rank == rank:
            continue
        keys, data, n_this_rank = movable_people.serialise(rank)
        if n_this_rank:
            reqs.append(mpi_comm.isend(keys, dest=rank, tag=100))
            reqs.append(mpi_comm.isend(data, dest=rank, tag=200))
            n_people_going_abroad += n_this_rank
        else:
            reqs.append(mpi_comm.isend(None, dest=rank, tag=100))
            reqs.append(mpi_comm.isend(None, dest=rank, tag=200))

    # now it has all been sent, we can start the receiving.

    for rank in range(mpi_size):

        if rank == mpi_rank:
            continue
        keys = mpi_comm.recv(source=rank, tag=100)
        data = mpi_comm.recv(source=rank, tag=200)

        if keys is not None:
            movable_people.update(rank, keys, data)
            n_people_from_abroad += data.shape[0]

    for r in reqs:
        r.wait()

    tock, tockw = perf_counter(), wall_clock()
    logger.info(
        f"CMS: People COMS for rank {mpi_rank}/{mpi_size} - {tock - tick},{tockw - tickw} - {self.timer.date}"
    )
    mpi_logger.info(f"{self.timer.date},{mpi_rank},people_comms,{tock-tick}")
    return movable_people.skinny_in, n_people_from_abroad, n_people_going_abroad