Care home

CareHome

Bases: Group

Represents a care home with its residents, workers, and visitors.

Source code in june/groups/care_home.py

class CareHome(Group):
    """Represents a care home with its residents, workers, and visitors."""

    __slots__ = ("max_residents", "area", "n_workers", "quarantine_starting_date", "registered_members_ids", "age_proportions")

    def __init__(
        self,
        area: Area = None,
        max_residents: int = None,
        n_workers: int = None,
        registered_members_ids: dict = None,
        age_proportions: dict = None,
    ):
        super().__init__()
        self.max_residents = max_residents
        self.n_workers = n_workers
        self.area = area
        self.quarantine_starting_date = None
        self.registered_members_ids = registered_members_ids if registered_members_ids is not None else {}
        self.age_proportions = age_proportions if age_proportions is not None else {
            'prop_0_15': 0.0,
            'prop_16_24': 0.0,
            'prop_25_34': 0.0,
            'prop_35_49': 0.0,
            'prop_50_64': 0.0,
            'prop_65_99': 1.0  # Default to elderly-only care home
        }

    def add(self, person, subgroup_type, activity: str = "residence"):
        """

        Args:
            person: 
            subgroup_type: 
            activity (str, optional): (Default value = "residence")

        """
        if activity == "leisure":
            super().add(
                person, subgroup_type=self.SubgroupType.visitors, activity="leisure"
            )
        else:
            super().add(person, subgroup_type=subgroup_type, activity=activity)

    def add_to_registered_members(self, person_id, subgroup_type=0):
        """Add a person to the registered members list for a specific subgroup.

        Args:
            person_id (int): The ID of the person to add
            subgroup_type (int, optional, optional): The subgroup to add the person to (default: 0)

        """
        # Create the subgroup if it doesn't exist
        if subgroup_type not in self.registered_members_ids:
            self.registered_members_ids[subgroup_type] = []

        # Add the person if not already in the list
        if person_id not in self.registered_members_ids[subgroup_type]:
            self.registered_members_ids[subgroup_type].append(person_id)

    @property
    def workers(self):
        """ """
        return self.subgroups[self.SubgroupType.workers]

    @property
    def residents(self):
        """ """
        return self.subgroups[self.SubgroupType.residents]

    @property
    def visitors(self):
        """ """
        return self.subgroups[self.SubgroupType.visitors]

    def quarantine(self, time, quarantine_days, household_compliance):
        """

        Args:
            time: 
            quarantine_days: 
            household_compliance: 

        """
        return True

    @property
    def coordinates(self):
        """ """
        return self.area.coordinates

    @property
    def super_area(self):
        """ """
        if self.area is None:
            return None
        else:
            return self.area.super_area

    @property
    def households_to_visit(self):
        """ """
        return None

    @property
    def care_homes_to_visit(self):
        """ """
        return None

    def get_leisure_subgroup(self, person, subgroup_type, to_send_abroad):
        """

        Args:
            person: 
            subgroup_type: 
            to_send_abroad: 

        """
        return self[self.SubgroupType.visitors]

    @property
    def type(self):
        """ """
        return "care_home"

care_homes_to_visit property

coordinates property

households_to_visit property

residents property

super_area property

type property

visitors property

workers property

add(person, subgroup_type, activity='residence')

Parameters:

Name	Type	Description	Default
person			required
subgroup_type			required
activity	str	(Default value = "residence")	'residence'

Source code in june/groups/care_home.py

def add(self, person, subgroup_type, activity: str = "residence"):
    """

    Args:
        person: 
        subgroup_type: 
        activity (str, optional): (Default value = "residence")

    """
    if activity == "leisure":
        super().add(
            person, subgroup_type=self.SubgroupType.visitors, activity="leisure"
        )
    else:
        super().add(person, subgroup_type=subgroup_type, activity=activity)

add_to_registered_members(person_id, subgroup_type=0)

Add a person to the registered members list for a specific subgroup.

Parameters:

Name	Type	Description	Default
person_id	int	The ID of the person to add	required
subgroup_type	(int, optional)	The subgroup to add the person to (default: 0)	0

Source code in june/groups/care_home.py

def add_to_registered_members(self, person_id, subgroup_type=0):
    """Add a person to the registered members list for a specific subgroup.

    Args:
        person_id (int): The ID of the person to add
        subgroup_type (int, optional, optional): The subgroup to add the person to (default: 0)

    """
    # Create the subgroup if it doesn't exist
    if subgroup_type not in self.registered_members_ids:
        self.registered_members_ids[subgroup_type] = []

    # Add the person if not already in the list
    if person_id not in self.registered_members_ids[subgroup_type]:
        self.registered_members_ids[subgroup_type].append(person_id)

get_leisure_subgroup(person, subgroup_type, to_send_abroad)

Parameters:

Name	Type	Description	Default
person			required
subgroup_type			required
to_send_abroad			required

Source code in june/groups/care_home.py

def get_leisure_subgroup(self, person, subgroup_type, to_send_abroad):
    """

    Args:
        person: 
        subgroup_type: 
        to_send_abroad: 

    """
    return self[self.SubgroupType.visitors]

quarantine(time, quarantine_days, household_compliance)

Parameters:

Name	Type	Description	Default
time			required
quarantine_days			required
household_compliance			required

Source code in june/groups/care_home.py

def quarantine(self, time, quarantine_days, household_compliance):
    """

    Args:
        time: 
        quarantine_days: 
        household_compliance: 

    """
    return True

CareHomeError

Bases: BaseException

Source code in june/groups/care_home.py

class CareHomeError(BaseException):
    """ """
    pass

CareHomes

Bases: Supergroup

Source code in june/groups/care_home.py

class CareHomes(Supergroup):
    """ """
    venue_class = CareHome

    def __init__(self, care_homes: List[venue_class]):
        super().__init__(members=care_homes)

    @classmethod
    def for_geography(
        cls,
        geography: Geography,
        data_file: str = None,
    ) -> "CareHomes":
        """Initialises care homes from geography using a disease configuration.

        Args:
            geography (Geography): The geography object with areas for initialising care homes.
            data_file (str, optional): Path to the care home data file. (Default value = None)

        Returns:
            CareHomes: An instance containing all created care homes.

        """
        areas = geography.areas
        if not areas:
            raise CareHomeError("Empty geography!")
        return cls.for_areas(areas, data_file)

    @classmethod
    def for_areas(
        cls,
        areas: List[Area],
        data_file: str = None,
        config_file: str = default_config_filename,
    ) -> "CareHomes":
        """

        Args:
            areas (List[Area]): list of areas for which to create populations
            data_file (str, optional): The path to the data file (CSV with flexible column support) (Default value = None)
            config_file (str, optional): The path to the config file (Default value = default_config_filename)

        """
        with open(config_file) as f:
            config = yaml.load(f, Loader=yaml.FullLoader)

        # Handle mixed geographies by detecting all unique regions and loading appropriate files
        if data_file is None:
            # Detect all unique geographies present in the areas
            geographies = set()
            for area in areas:
                if area.name.startswith(('E0', 'W0')):
                    geographies.add('ew')
                elif area.name.startswith('S0'):
                    geographies.add('sct')
                elif area.name.startswith('N'):
                    geographies.add('ni')

            # Load and combine data from all relevant geography files
            care_home_dfs = []
            for geo in geographies:
                if geo == 'ew':
                    geo_file = default_data_filename_ew
                elif geo == 'sct':
                    geo_file = default_data_filename_sct
                elif geo == 'ni':
                    geo_file = default_data_filename_ni

                try:
                    if geo_file.exists():
                        df = pd.read_csv(geo_file)
                        care_home_dfs.append(df)
                        logger.info(f"Loaded {len(df)} care homes from {geo_file}")
                    else:
                        logger.warning(f"Care homes file not found: {geo_file}")
                except Exception as e:
                    logger.warning(f"Failed to load care homes from {geo_file}: {e}")

            # Combine all dataframes
            if care_home_dfs:
                care_home_df = pd.concat(care_home_dfs, ignore_index=True)
            else:
                # Fallback to default file if no geography-specific files found
                logger.warning("No geography-specific files found, using default")
                care_home_df = pd.read_csv(default_data_filename)
        else:
            # Use the specified data file
            care_home_df = pd.read_csv(data_file)

        # Check which columns are available for flexible handling
        has_number_staff = 'number_staff' in care_home_df.columns

        if areas:
            area_names = [area.name for area in areas]
            # Filter care homes that are in the areas of interest
            care_home_df = care_home_df[care_home_df['area'].isin(area_names)]

        care_homes = []
        # Calculate total capacity safely, handling NaN values
        total_capacity = care_home_df['capacity'].fillna(0).sum()
        logger.info(
            f"There are {len(care_home_df)} individual care_homes in this geography with total capacity of {total_capacity}."
        )

        # Create a mapping of area names to area objects for quick lookup
        area_dict = {area.name: area for area in areas}

        # Create individual care homes for each row in the CSV
        for _, row in care_home_df.iterrows():
            try:
                area_name = row['area']

                # Safely convert capacity to int, handling NaN values
                capacity_val = row['capacity']
                if pd.isna(capacity_val) or np.isinf(capacity_val):
                    max_residents = 20  # Default capacity
                    logger.warning(f"Care home {row.get('id', 'unknown')} in area {area_name}: invalid capacity value {capacity_val}, using default 20")
                else:
                    max_residents = int(capacity_val)

                # Handle staff numbers flexibly based on available columns
                if has_number_staff:
                    staff_val = row['number_staff']
                    if pd.isna(staff_val) or np.isinf(staff_val):
                        n_workers = 1  # Default staff count
                        logger.warning(f"Care home {row.get('id', 'unknown')} in area {area_name}: invalid staff value {staff_val}, using default 1")
                    else:
                        n_workers = int(staff_val)
                else:
                    # Estimate staff based on capacity if number_staff column is missing
                    # Use a reasonable ratio: 1 staff per 2 residents as starting point
                    n_workers = max(1, max_residents // 2)
                    logger.info(f"Care home {row.get('id', 'unknown')} in area {area_name}: estimating {n_workers} staff based on capacity {max_residents}")
            except (ValueError, TypeError) as e:
                logger.error(f"Error processing care home {row.get('id', 'unknown')}: {e}")
                logger.error(f"Row data: {row.to_dict()}")
                continue  # Skip this care home

            # Apply minimum staffing ratio: at least 1 worker per 10 residents
            min_workers_needed = max(1, max_residents // 10)
            if n_workers == 0 or n_workers < min_workers_needed:
                logger.info(f"Care home {row.get('id', 'unknown')} in area {area_name}: adjusting staff from {n_workers} to {min_workers_needed} (1 per 10 residents)")
                n_workers = min_workers_needed

            if max_residents > 0 and area_name in area_dict:
                area = area_dict[area_name]
                care_home = cls.venue_class(area, max_residents, n_workers)
                care_homes.append(care_home)

                # Add care home to area's care_homes list
                area.care_homes.append(care_home)

        # Visualization - Sample 5 care homes for inspection
        sample_care_homes = [
            {
                "| Care Home ID": care_home.id,
                "| Area": care_home.area.name if care_home.area else "Unknown",
                "| Max Residents": care_home.max_residents,
                "| Workers": care_home.n_workers,
                "| Coordinates": care_home.coordinates if care_home.area else "Unknown",
            }
            for care_home in random.sample(care_homes, min(5, len(care_homes)))
        ]

        df_care_homes = pd.DataFrame(sample_care_homes)
        print("\n===== Sample of Created Care Homes =====")
        print(df_care_homes)
        return cls(care_homes)

for_areas(areas, data_file=None, config_file=default_config_filename) classmethod

Parameters:

Name	Type	Description	Default
areas	List[Area]	list of areas for which to create populations	required
data_file	str	The path to the data file (CSV with flexible column support) (Default value = None)	None
config_file	str	The path to the config file (Default value = default_config_filename)	default_config_filename

Source code in june/groups/care_home.py

@classmethod
def for_areas(
    cls,
    areas: List[Area],
    data_file: str = None,
    config_file: str = default_config_filename,
) -> "CareHomes":
    """

    Args:
        areas (List[Area]): list of areas for which to create populations
        data_file (str, optional): The path to the data file (CSV with flexible column support) (Default value = None)
        config_file (str, optional): The path to the config file (Default value = default_config_filename)

    """
    with open(config_file) as f:
        config = yaml.load(f, Loader=yaml.FullLoader)

    # Handle mixed geographies by detecting all unique regions and loading appropriate files
    if data_file is None:
        # Detect all unique geographies present in the areas
        geographies = set()
        for area in areas:
            if area.name.startswith(('E0', 'W0')):
                geographies.add('ew')
            elif area.name.startswith('S0'):
                geographies.add('sct')
            elif area.name.startswith('N'):
                geographies.add('ni')

        # Load and combine data from all relevant geography files
        care_home_dfs = []
        for geo in geographies:
            if geo == 'ew':
                geo_file = default_data_filename_ew
            elif geo == 'sct':
                geo_file = default_data_filename_sct
            elif geo == 'ni':
                geo_file = default_data_filename_ni

            try:
                if geo_file.exists():
                    df = pd.read_csv(geo_file)
                    care_home_dfs.append(df)
                    logger.info(f"Loaded {len(df)} care homes from {geo_file}")
                else:
                    logger.warning(f"Care homes file not found: {geo_file}")
            except Exception as e:
                logger.warning(f"Failed to load care homes from {geo_file}: {e}")

        # Combine all dataframes
        if care_home_dfs:
            care_home_df = pd.concat(care_home_dfs, ignore_index=True)
        else:
            # Fallback to default file if no geography-specific files found
            logger.warning("No geography-specific files found, using default")
            care_home_df = pd.read_csv(default_data_filename)
    else:
        # Use the specified data file
        care_home_df = pd.read_csv(data_file)

    # Check which columns are available for flexible handling
    has_number_staff = 'number_staff' in care_home_df.columns

    if areas:
        area_names = [area.name for area in areas]
        # Filter care homes that are in the areas of interest
        care_home_df = care_home_df[care_home_df['area'].isin(area_names)]

    care_homes = []
    # Calculate total capacity safely, handling NaN values
    total_capacity = care_home_df['capacity'].fillna(0).sum()
    logger.info(
        f"There are {len(care_home_df)} individual care_homes in this geography with total capacity of {total_capacity}."
    )

    # Create a mapping of area names to area objects for quick lookup
    area_dict = {area.name: area for area in areas}

    # Create individual care homes for each row in the CSV
    for _, row in care_home_df.iterrows():
        try:
            area_name = row['area']

            # Safely convert capacity to int, handling NaN values
            capacity_val = row['capacity']
            if pd.isna(capacity_val) or np.isinf(capacity_val):
                max_residents = 20  # Default capacity
                logger.warning(f"Care home {row.get('id', 'unknown')} in area {area_name}: invalid capacity value {capacity_val}, using default 20")
            else:
                max_residents = int(capacity_val)

            # Handle staff numbers flexibly based on available columns
            if has_number_staff:
                staff_val = row['number_staff']
                if pd.isna(staff_val) or np.isinf(staff_val):
                    n_workers = 1  # Default staff count
                    logger.warning(f"Care home {row.get('id', 'unknown')} in area {area_name}: invalid staff value {staff_val}, using default 1")
                else:
                    n_workers = int(staff_val)
            else:
                # Estimate staff based on capacity if number_staff column is missing
                # Use a reasonable ratio: 1 staff per 2 residents as starting point
                n_workers = max(1, max_residents // 2)
                logger.info(f"Care home {row.get('id', 'unknown')} in area {area_name}: estimating {n_workers} staff based on capacity {max_residents}")
        except (ValueError, TypeError) as e:
            logger.error(f"Error processing care home {row.get('id', 'unknown')}: {e}")
            logger.error(f"Row data: {row.to_dict()}")
            continue  # Skip this care home

        # Apply minimum staffing ratio: at least 1 worker per 10 residents
        min_workers_needed = max(1, max_residents // 10)
        if n_workers == 0 or n_workers < min_workers_needed:
            logger.info(f"Care home {row.get('id', 'unknown')} in area {area_name}: adjusting staff from {n_workers} to {min_workers_needed} (1 per 10 residents)")
            n_workers = min_workers_needed

        if max_residents > 0 and area_name in area_dict:
            area = area_dict[area_name]
            care_home = cls.venue_class(area, max_residents, n_workers)
            care_homes.append(care_home)

            # Add care home to area's care_homes list
            area.care_homes.append(care_home)

    # Visualization - Sample 5 care homes for inspection
    sample_care_homes = [
        {
            "| Care Home ID": care_home.id,
            "| Area": care_home.area.name if care_home.area else "Unknown",
            "| Max Residents": care_home.max_residents,
            "| Workers": care_home.n_workers,
            "| Coordinates": care_home.coordinates if care_home.area else "Unknown",
        }
        for care_home in random.sample(care_homes, min(5, len(care_homes)))
    ]

    df_care_homes = pd.DataFrame(sample_care_homes)
    print("\n===== Sample of Created Care Homes =====")
    print(df_care_homes)
    return cls(care_homes)

for_geography(geography, data_file=None) classmethod

Initialises care homes from geography using a disease configuration.

Parameters:

Name	Type	Description	Default
geography	Geography	The geography object with areas for initialising care homes.	required
data_file	str	Path to the care home data file. (Default value = None)	None

Returns:

Name	Type	Description
CareHomes	CareHomes	An instance containing all created care homes.

Source code in june/groups/care_home.py

@classmethod
def for_geography(
    cls,
    geography: Geography,
    data_file: str = None,
) -> "CareHomes":
    """Initialises care homes from geography using a disease configuration.

    Args:
        geography (Geography): The geography object with areas for initialising care homes.
        data_file (str, optional): Path to the care home data file. (Default value = None)

    Returns:
        CareHomes: An instance containing all created care homes.

    """
    areas = geography.areas
    if not areas:
        raise CareHomeError("Empty geography!")
    return cls.for_areas(areas, data_file)