Residence visits

ResidenceVisitsDistributor

Bases: SocialVenueDistributor

This is a social distributor specific to model visits between residences, ie, visits between households or to care homes. The meaning of the parameters is the same as for the SVD. Residence visits are not decied on neighbours or distances so we ignore some parameters.

Source code in june/groups/leisure/residence_visits.py

class ResidenceVisitsDistributor(SocialVenueDistributor):
    """This is a social distributor specific to model visits between residences,
    ie, visits between households or to care homes. The meaning of the parameters
    is the same as for the SVD. Residence visits are not decied on neighbours or distances
    so we ignore some parameters.

    """

    def __init__(
        self,
        residence_type_probabilities,
        times_per_week,
        hours_per_day,
        daytypes=default_daytypes,
        drags_household_probability=0,
        invites_friends_probability=0
    ):
        # it is necessary to make them arrays for performance
        self.residence_type_probabilities = residence_type_probabilities
        self.policy_reductions = {}
        super().__init__(
            social_venues=None,
            times_per_week=times_per_week,
            daytypes=daytypes,
            hours_per_day=hours_per_day,
            drags_household_probability=drags_household_probability,
            invites_friends_probability=invites_friends_probability,
            neighbours_to_consider=None,
            maximum_distance=None,
            leisure_subgroup_type=None,
        )

    @classmethod
    def from_config(cls, daytypes, config_filename: str = default_config_filename):
        """

        Args:
            daytypes: 
            config_filename (str, optional): (Default value = default_config_filename)

        """
        # Load the configuration file
        with open(config_filename) as f:
            config = yaml.load(f, Loader=yaml.FullLoader)

        # Display configuration data for visualization
        #print("\n===== Loaded Configuration Settings =====")
        config_df = pd.DataFrame([config])  # Convert config to DataFrame for readability
        #print(config_df)

        # Display daytypes information based on its structure
        #print("\n===== Daytypes Information =====")
        #try:
            # Convert daytypes to DataFrame appropriately
            #daytypes_df = pd.DataFrame(daytypes)
            #print(daytypes_df)
        #except ValueError:
            #print("Unable to display daytypes as a DataFrame directly.")
            #for key, value in daytypes.items():
                #print(f"{key}: {value}")

        # Create and return the instance
        return cls(daytypes=daytypes, **config)

    def _get_visited_residents(self, person):
        """Get the residents of the place a person is visiting (household or care home)

        Args:
            person (Person): The person who is visiting

        Returns:
            list: List of residents at the visited location

        """
        # Get the leisure group (residence being visited)
        visited_group = person.leisure.group if person.leisure and hasattr(person.leisure, 'group') else None

        if not visited_group:
            return []

        # Get the residents of the visited location
        if hasattr(visited_group, 'residents'):
            return [resident for resident in visited_group.residents if resident != person]
        else:
            return []

    def coordinate_invitations(self, person, activity, to_send_abroad=None):
        """Coordinate invitations for residence visits - simplifies the base class version
        since residences are always local

        Args:
            person: 
            activity: 
            to_send_abroad: (Default value = None)

        Returns:
            list: List of all people who were invited

        """
        invited_people = []  # List to track all invited people

        # Get invitation priority from social network
        priority = self.social_network.decide_invitation(person, activity)

        # Define priority actions
        actions = {
            "household": self.person_drags_household,
            "friends": self.person_invites_friends,
        }

        # Try primary priority
        if actions[priority]():
            if priority == "household":
                household_invitees = self._coordinate_household_invitations(person, to_send_abroad)
                invited_people.extend(household_invitees or [])
            else:
                friend_invitees = self._coordinate_friend_invitations(person, to_send_abroad)
                invited_people.extend(friend_invitees or [])
        else:
            # Try fallback priority
            other_priority = "friends" if priority == "household" else "household"
            if actions[other_priority]():
                if other_priority == "household":
                    household_invitees = self._coordinate_household_invitations(person, to_send_abroad)
                    invited_people.extend(household_invitees or [])
                else:
                    friend_invitees = self._coordinate_friend_invitations(person, to_send_abroad)
                    invited_people.extend(friend_invitees or [])

        # Get residents of the location we're visiting to add to contacts
        visit_residents = self._get_visited_residents(person)

        # Add them to the list of invited people for contact tracking
        invited_people.extend(visit_residents)

        return invited_people

    def _coordinate_household_invitations(self, person, to_send_abroad):
        """Handle household invitations for residence visits

        Args:
            person: 
            to_send_abroad: 

        Returns:
            list: List of household members who were invited

        """
        if person.residence.group.spec in ["care_home", "communal", "other", "student"]:
            return []

        people_to_bring = [
            mate for mate in person.residence.group.residents 
            if mate != person
        ]
        self.bring_people_with_person(person, people_to_bring, to_send_abroad)
        return people_to_bring

    def _coordinate_friend_invitations(self, person, to_send_abroad):
        """Handle friend invitations for residence visits

        Args:
            person: 
            to_send_abroad: 

        Returns:
            list: List of friends who were invited locally

        """
        if person.residence.group.spec == "care_home":
            return []

        friend_candidates = self.invite_friends_with_hobbies(person)
        invited_friends = []

        # Get the current rank - default to 0 in non-MPI mode
        current_rank = 0 if not mpi_available else mpi_rank

        # For residence visits, we only handle local friends
        local_friends = [
            friend for friend in friend_candidates 
            if friend.get_friend_rank(friend.id) == current_rank
        ]

        if local_friends:
            self.bring_people_with_person(person, local_friends, to_send_abroad)
            invited_friends.extend(local_friends)

        return invited_friends

    def _send_friend_invitation(self, person, friend, friend_rank):
        """For residence visits, we don't send remote invitations.
        Friends must be local to visit residences.

        Args:
            person: 
            friend: 
            friend_rank: 

        """
        pass  # No remote invitations for residence visits

    def link_households_to_households(self, super_areas, debug=False):
        """Optimized version: Links people between households using efficient vectorized operations.
        Strategy: We pair each household with 2-4 other households. The household of the former then
        has a probability of visiting the household of the later at every time step.

        Args:
            super_areas: list of super areas
            debug (bool, optional): If True, collect and display sample data for verification (Default value = False)

        """
        total_links_created = 0

        for super_area in super_areas:
            # Pre-collect all households in super area once
            all_households = []
            for area in super_area.areas:
                all_households.extend(area.households)

            # Pre-filter valid households (have residents)
            valid_households = [h for h in all_households if h.n_residents > 0]

            if len(valid_households) < 2:
                continue  # Need at least 2 households to link

            # Process each household
            for household in valid_households:
                # Pre-filter candidates (exclude self, ensure residents)
                candidates = [
                    h for h in valid_households
                    if h.id != household.id and h.residents
                ]

                if len(candidates) == 0:
                    continue

                # Determine number of households to link (2-4)
                households_to_link_n = np.random.randint(2, 5)  # 2-4 inclusive
                n_to_select = min(households_to_link_n, len(candidates))

                # Efficient selection without replacement
                if n_to_select >= len(candidates):
                    households_to_visit = candidates
                else:
                    # Use random.sample for object selection without replacement
                    households_to_visit = sample(candidates, n_to_select)

                # Set the links
                if households_to_visit:
                    household.residences_to_visit["household"] = tuple(households_to_visit)
                    total_links_created += len(households_to_visit)

        # Optional debug output
        if debug:
            self._debug_household_links(super_areas, total_links_created)
        else:
            logger.info(f"Linked households: {total_links_created} total connections created")

    def _debug_household_links(self, super_areas, total_links):
        """Optional debug method to show household linking sample.

        Args:
            super_areas: 
            total_links: 

        """
        try:
            sample_data = []
            sample_super_areas = sample(list(super_areas), min(5, len(super_areas)))

            for super_area in sample_super_areas:
                all_households = []
                for area in super_area.areas:
                    all_households.extend(area.households)

                valid_households = [h for h in all_households if h.n_residents > 0]
                if not valid_households:
                    continue

                # Sample some households to show their links
                sample_households = sample(valid_households, min(3, len(valid_households)))

                for household in sample_households:
                    if "household" in household.residences_to_visit:
                        linked_ids = [h.id for h in household.residences_to_visit["household"]]
                        sample_data.append({
                            "Household ID": household.id,
                            "Super Area": super_area.name,
                            "Household Type": household.type,
                            "Residents": household.n_residents,
                            "Linked To": linked_ids,
                            "Links Count": len(linked_ids)
                        })

            if sample_data:
                df = pd.DataFrame(sample_data)
                print(f"\n===== Household Links Sample (Total: {total_links}) =====")
                print(df)

        except Exception as e:
            print(f"Debug sampling failed: {e}")

    def link_households_to_care_homes(self, super_areas, debug=False):
        """Optimized version: Links households and care homes using efficient operations.
        For each care home, we find random houses in the super area and link them.
        The house needs to be occupied by a family, or a couple.

        Args:
            super_areas: list of super areas
            debug (bool, optional): If True, collect and display sample data for verification (Default value = False)

        """
        total_care_home_links = 0

        for super_area in super_areas:
            # Pre-collect eligible households once per super area
            eligible_households = []
            for area in super_area.areas:
                eligible_households.extend([
                    household for household in area.households
                    if household.type in ["family", "older_family", "couple"]
                ])

            if not eligible_households:
                continue

            # Keep as list for object operations

            # Collect all care homes in super area
            all_care_homes = []
            for area in super_area.areas:
                all_care_homes.extend(area.care_homes)

            if not all_care_homes:
                continue

            # For each care home, link it to households
            for care_home in all_care_homes:
                if not care_home.residents:
                    continue

                n_residents = len(care_home.residents)
                n_households_needed = min(n_residents, len(eligible_households))

                if n_households_needed == 0:
                    continue

                # Efficiently select households without replacement
                if n_households_needed >= len(eligible_households):
                    selected_households = eligible_households
                else:
                    selected_households = sample(eligible_households, n_households_needed)

                # Link each selected household to this care home
                for household in selected_households:
                    current_care_homes = household.residences_to_visit.get("care_home", ())
                    household.residences_to_visit["care_home"] = (*current_care_homes, care_home)
                    total_care_home_links += 1

        # Optional debug output
        if debug:
            self._debug_care_home_links(super_areas, total_care_home_links)
        else:
            logger.info(f"Care home links: {total_care_home_links} total connections created")

    def _debug_care_home_links(self, super_areas, total_links):
        """Optional debug method to show care home linking sample.

        Args:
            super_areas: 
            total_links: 

        """
        try:
            sample_data = []
            sample_super_areas = sample(list(super_areas), min(3, len(super_areas)))

            for super_area in sample_super_areas:
                # Find households with care home links
                for area in super_area.areas:
                    for household in area.households:
                        if "care_home" in household.residences_to_visit and household.residences_to_visit["care_home"]:
                            care_home_ids = [ch.id for ch in household.residences_to_visit["care_home"]]
                            sample_data.append({
                                "Household ID": household.id,
                                "Household Type": household.type,
                                "Care Home IDs": care_home_ids,
                                "Super Area": super_area.name,
                                "Links Count": len(care_home_ids)
                            })

                            if len(sample_data) >= 10:  # Limit sample size
                                break
                    if len(sample_data) >= 10:
                        break
                if len(sample_data) >= 10:
                    break

            if sample_data:
                df = pd.DataFrame(sample_data)
                print(f"\n===== Care Home Links Sample (Total: {total_links}) =====")
                print(df)

        except Exception as e:
            print(f"Debug care home sampling failed: {e}") 

    def get_leisure_group(self, person):
        """

        Args:
            person: 

        """
        # Skip student dorms and boarding schools for now
        if person.residence.group.spec in ["student_dorm", "boarding_school"]:
            return
        residence_types = list(person.residence.group.residences_to_visit.keys())
        if not residence_types:
            return
        if len(residence_types) == 0:
            which_type = residence_types[0]
        else:
            if self.policy_reductions:
                probabilities = self.policy_reductions
            else:
                probabilities = self.residence_type_probabilities
            residence_type_probabilities = np.array(
                [probabilities[residence_type] for residence_type in residence_types]
            )
            residence_type_probabilities = (
                residence_type_probabilities / residence_type_probabilities.sum()
            )
            random_number_for_jit = np.random.random()
            type_sample = random_choice_numba_numpy(
                np.arange(len(residence_type_probabilities)),  # ← Use np.arange directly
                residence_type_probabilities,
                random_number_for_jit
            )
            which_type = residence_types[type_sample]
        candidates = person.residence.group.residences_to_visit[which_type]
        n_candidates = len(candidates)
        if n_candidates == 0:
            return
        elif n_candidates == 1:
            group = candidates[0]
        else:
            group = candidates[randint(0, n_candidates - 1)]
        return group

    def get_poisson_parameter(
        self, sex, age, day_type, working_hours, region=None, policy_reduction=None
    ):
        """This differs from the super() implementation in that we do not allow
        visits during working hours as most people are away.

        Args:
            sex: 
            age: 
            day_type: 
            working_hours: 
            region: (Default value = None)
            policy_reduction: (Default value = None)

        """
        if working_hours:
            return 0
        return super().get_poisson_parameter(
            sex=sex,
            age=age,
            day_type=day_type,
            working_hours=working_hours,
            region=region,
            policy_reduction=policy_reduction,
        )

coordinate_invitations(person, activity, to_send_abroad=None)

Coordinate invitations for residence visits - simplifies the base class version since residences are always local

Parameters:

Name	Type	Description	Default
person			required
activity			required
to_send_abroad		(Default value = None)	None

Returns:

Name	Type	Description
list		List of all people who were invited

Source code in june/groups/leisure/residence_visits.py

def coordinate_invitations(self, person, activity, to_send_abroad=None):
    """Coordinate invitations for residence visits - simplifies the base class version
    since residences are always local

    Args:
        person: 
        activity: 
        to_send_abroad: (Default value = None)

    Returns:
        list: List of all people who were invited

    """
    invited_people = []  # List to track all invited people

    # Get invitation priority from social network
    priority = self.social_network.decide_invitation(person, activity)

    # Define priority actions
    actions = {
        "household": self.person_drags_household,
        "friends": self.person_invites_friends,
    }

    # Try primary priority
    if actions[priority]():
        if priority == "household":
            household_invitees = self._coordinate_household_invitations(person, to_send_abroad)
            invited_people.extend(household_invitees or [])
        else:
            friend_invitees = self._coordinate_friend_invitations(person, to_send_abroad)
            invited_people.extend(friend_invitees or [])
    else:
        # Try fallback priority
        other_priority = "friends" if priority == "household" else "household"
        if actions[other_priority]():
            if other_priority == "household":
                household_invitees = self._coordinate_household_invitations(person, to_send_abroad)
                invited_people.extend(household_invitees or [])
            else:
                friend_invitees = self._coordinate_friend_invitations(person, to_send_abroad)
                invited_people.extend(friend_invitees or [])

    # Get residents of the location we're visiting to add to contacts
    visit_residents = self._get_visited_residents(person)

    # Add them to the list of invited people for contact tracking
    invited_people.extend(visit_residents)

    return invited_people

from_config(daytypes, config_filename=default_config_filename) classmethod

Parameters:

Name	Type	Description	Default
daytypes			required
config_filename	str	(Default value = default_config_filename)	default_config_filename

Source code in june/groups/leisure/residence_visits.py

@classmethod
def from_config(cls, daytypes, config_filename: str = default_config_filename):
    """

    Args:
        daytypes: 
        config_filename (str, optional): (Default value = default_config_filename)

    """
    # Load the configuration file
    with open(config_filename) as f:
        config = yaml.load(f, Loader=yaml.FullLoader)

    # Display configuration data for visualization
    #print("\n===== Loaded Configuration Settings =====")
    config_df = pd.DataFrame([config])  # Convert config to DataFrame for readability
    #print(config_df)

    # Display daytypes information based on its structure
    #print("\n===== Daytypes Information =====")
    #try:
        # Convert daytypes to DataFrame appropriately
        #daytypes_df = pd.DataFrame(daytypes)
        #print(daytypes_df)
    #except ValueError:
        #print("Unable to display daytypes as a DataFrame directly.")
        #for key, value in daytypes.items():
            #print(f"{key}: {value}")

    # Create and return the instance
    return cls(daytypes=daytypes, **config)

get_leisure_group(person)

Parameters:

Name	Type	Description	Default
person			required

Source code in june/groups/leisure/residence_visits.py

def get_leisure_group(self, person):
    """

    Args:
        person: 

    """
    # Skip student dorms and boarding schools for now
    if person.residence.group.spec in ["student_dorm", "boarding_school"]:
        return
    residence_types = list(person.residence.group.residences_to_visit.keys())
    if not residence_types:
        return
    if len(residence_types) == 0:
        which_type = residence_types[0]
    else:
        if self.policy_reductions:
            probabilities = self.policy_reductions
        else:
            probabilities = self.residence_type_probabilities
        residence_type_probabilities = np.array(
            [probabilities[residence_type] for residence_type in residence_types]
        )
        residence_type_probabilities = (
            residence_type_probabilities / residence_type_probabilities.sum()
        )
        random_number_for_jit = np.random.random()
        type_sample = random_choice_numba_numpy(
            np.arange(len(residence_type_probabilities)),  # ← Use np.arange directly
            residence_type_probabilities,
            random_number_for_jit
        )
        which_type = residence_types[type_sample]
    candidates = person.residence.group.residences_to_visit[which_type]
    n_candidates = len(candidates)
    if n_candidates == 0:
        return
    elif n_candidates == 1:
        group = candidates[0]
    else:
        group = candidates[randint(0, n_candidates - 1)]
    return group

get_poisson_parameter(sex, age, day_type, working_hours, region=None, policy_reduction=None)

This differs from the super() implementation in that we do not allow visits during working hours as most people are away.

Parameters:

Name	Type	Description	Default
sex			required
age			required
day_type			required
working_hours			required
region		(Default value = None)	None
policy_reduction		(Default value = None)	None

Source code in june/groups/leisure/residence_visits.py

def get_poisson_parameter(
    self, sex, age, day_type, working_hours, region=None, policy_reduction=None
):
    """This differs from the super() implementation in that we do not allow
    visits during working hours as most people are away.

    Args:
        sex: 
        age: 
        day_type: 
        working_hours: 
        region: (Default value = None)
        policy_reduction: (Default value = None)

    """
    if working_hours:
        return 0
    return super().get_poisson_parameter(
        sex=sex,
        age=age,
        day_type=day_type,
        working_hours=working_hours,
        region=region,
        policy_reduction=policy_reduction,
    )

link_households_to_care_homes(super_areas, debug=False)

Optimized version: Links households and care homes using efficient operations. For each care home, we find random houses in the super area and link them. The house needs to be occupied by a family, or a couple.

Parameters:

Name	Type	Description	Default
super_areas		list of super areas	required
debug	bool	If True, collect and display sample data for verification (Default value = False)	False

Source code in june/groups/leisure/residence_visits.py

def link_households_to_care_homes(self, super_areas, debug=False):
    """Optimized version: Links households and care homes using efficient operations.
    For each care home, we find random houses in the super area and link them.
    The house needs to be occupied by a family, or a couple.

    Args:
        super_areas: list of super areas
        debug (bool, optional): If True, collect and display sample data for verification (Default value = False)

    """
    total_care_home_links = 0

    for super_area in super_areas:
        # Pre-collect eligible households once per super area
        eligible_households = []
        for area in super_area.areas:
            eligible_households.extend([
                household for household in area.households
                if household.type in ["family", "older_family", "couple"]
            ])

        if not eligible_households:
            continue

        # Keep as list for object operations

        # Collect all care homes in super area
        all_care_homes = []
        for area in super_area.areas:
            all_care_homes.extend(area.care_homes)

        if not all_care_homes:
            continue

        # For each care home, link it to households
        for care_home in all_care_homes:
            if not care_home.residents:
                continue

            n_residents = len(care_home.residents)
            n_households_needed = min(n_residents, len(eligible_households))

            if n_households_needed == 0:
                continue

            # Efficiently select households without replacement
            if n_households_needed >= len(eligible_households):
                selected_households = eligible_households
            else:
                selected_households = sample(eligible_households, n_households_needed)

            # Link each selected household to this care home
            for household in selected_households:
                current_care_homes = household.residences_to_visit.get("care_home", ())
                household.residences_to_visit["care_home"] = (*current_care_homes, care_home)
                total_care_home_links += 1

    # Optional debug output
    if debug:
        self._debug_care_home_links(super_areas, total_care_home_links)
    else:
        logger.info(f"Care home links: {total_care_home_links} total connections created")

link_households_to_households(super_areas, debug=False)

Optimized version: Links people between households using efficient vectorized operations. Strategy: We pair each household with 2-4 other households. The household of the former then has a probability of visiting the household of the later at every time step.

Parameters:

Name	Type	Description	Default
super_areas		list of super areas	required
debug	bool	If True, collect and display sample data for verification (Default value = False)	False

Source code in june/groups/leisure/residence_visits.py

def link_households_to_households(self, super_areas, debug=False):
    """Optimized version: Links people between households using efficient vectorized operations.
    Strategy: We pair each household with 2-4 other households. The household of the former then
    has a probability of visiting the household of the later at every time step.

    Args:
        super_areas: list of super areas
        debug (bool, optional): If True, collect and display sample data for verification (Default value = False)

    """
    total_links_created = 0

    for super_area in super_areas:
        # Pre-collect all households in super area once
        all_households = []
        for area in super_area.areas:
            all_households.extend(area.households)

        # Pre-filter valid households (have residents)
        valid_households = [h for h in all_households if h.n_residents > 0]

        if len(valid_households) < 2:
            continue  # Need at least 2 households to link

        # Process each household
        for household in valid_households:
            # Pre-filter candidates (exclude self, ensure residents)
            candidates = [
                h for h in valid_households
                if h.id != household.id and h.residents
            ]

            if len(candidates) == 0:
                continue

            # Determine number of households to link (2-4)
            households_to_link_n = np.random.randint(2, 5)  # 2-4 inclusive
            n_to_select = min(households_to_link_n, len(candidates))

            # Efficient selection without replacement
            if n_to_select >= len(candidates):
                households_to_visit = candidates
            else:
                # Use random.sample for object selection without replacement
                households_to_visit = sample(candidates, n_to_select)

            # Set the links
            if households_to_visit:
                household.residences_to_visit["household"] = tuple(households_to_visit)
                total_links_created += len(households_to_visit)

    # Optional debug output
    if debug:
        self._debug_household_links(super_areas, total_links_created)
    else:
        logger.info(f"Linked households: {total_links_created} total connections created")