Hospital distributor

HospitalDistributor

Distributes people to work as health care workers in hospitals

TODO: sub sectors of doctors and nurses should be found

Healthcares sector 2211: Medical practitioners 2217: Medical radiographers 2231: Nurses 2232: Midwives

Source code in june/distributors/hospital_distributor.py

class HospitalDistributor:
    """Distributes people to work as health care workers in hospitals

        #TODO: sub sectors of doctors and nurses should be found
        Healthcares sector
        2211: Medical practitioners
        2217: Medical radiographers
        2231: Nurses
        2232: Midwives

    """

    def __init__(
        self,
        hospitals: Hospitals,
        medic_min_age: int,
        patients_per_medic: int,
        healthcare_sector_label: Optional[str] = None,
    ):
        """

        Parameters
        ----------
        hospitals:
            hospitals to populate with workers
        medic_min_age:
            minimum age to qualify as a worker
        patients_per_medic:
            ratio of patients per medic
        healthcare_sector_label:
            string that characterises the helathcare workers
        """
        # check if this msoarea has hospitals
        self.hospitals = hospitals
        self.medic_min_age = medic_min_age
        self.patients_per_medic = patients_per_medic
        self.healthcare_sector_label = healthcare_sector_label

    @classmethod
    def from_file(cls, hospitals, config_filename=default_config_filename):
        """

        Args:
            hospitals: 
            config_filename: (Default value = default_config_filename)

        """
        with open(config_filename) as f:
            config = yaml.load(f, Loader=yaml.FullLoader)
        return HospitalDistributor(
            hospitals=hospitals,
            medic_min_age=config["medic_min_age"],
            patients_per_medic=config["patients_per_medic"],
            healthcare_sector_label=config["healthcare_sector_label"],
        )

    def distribute_medics_from_world(self, people: List["Person"]):
        """Randomly distribute people from the world to work as medics for hospitals,
        useful if we don't have data on where do people work. It will still
        match the patients to medic ratio and the minimum age to be a medic.

        Args:
            people (List["Person"]): list of Persons in the world

        """
        medics = [person for person in people if person.age >= self.medic_min_age]
        shuffle(medics)
        total_medics_distributed = 0
        hospital_assignments = []

        for hospital in self.hospitals:
            max_capacity = hospital.n_beds + hospital.n_icu_beds
            if max_capacity == 0:
                continue
            n_medics = max(int(np.floor(max_capacity / self.patients_per_medic)), 1)
            hospital_medics = 0
            for _ in range(n_medics):
                if not medics:
                    logger.warning("Ran out of available medics before fully staffing all hospitals")
                    break
                medic = medics.pop()
                hospital.add(medic, hospital.SubgroupType.workers)
                medic.lockdown_status = "key_worker"
                hospital_medics += 1
                total_medics_distributed += 1

            hospital_assignments.append({
                "Hospital_ID": hospital.id,
                "Beds": hospital.n_beds,
                "ICU_Beds": hospital.n_icu_beds,
                "Total_Capacity": max_capacity,
                "Medics_Assigned": hospital_medics
            })

        # Log summary
        logger.info(f"=== MEDIC DISTRIBUTION SUMMARY ===")
        logger.info(f"Total eligible medics available: {len(medics) + total_medics_distributed}")
        logger.info(f"Total medics distributed: {total_medics_distributed}")
        logger.info(f"Medics remaining unassigned: {len(medics)}")
        logger.info(f"Hospitals staffed: {len([h for h in hospital_assignments if h['Medics_Assigned'] > 0])}")
        logger.info(f"Total hospital capacity: {sum(h['Total_Capacity'] for h in hospital_assignments)}")

        # Display detailed breakdown as DataFrame
        df_assignments = pd.DataFrame(hospital_assignments)
        print("\n=== HOSPITAL MEDIC ASSIGNMENTS ===")
        print(df_assignments.to_string(index=False))

    def distribute_medics_to_super_areas(self, super_areas: SuperAreas):
        """Distribute medics to super areas, flow data is necessary to find medics in the
        super area according to their sector.

        Args:
            super_areas (SuperAreas): object containing all the super areas to distribute medics

        """
        logger.info("Distributing medics to hospitals")
        total_medics_distributed = 0
        super_area_stats = []

        for super_area in super_areas:
            medics_before = sum(len(hospital.subgroups[hospital.SubgroupType.workers].people) 
                              for hospital in self.get_hospitals_in_super_area(super_area))
            self.distribute_medics_to_hospitals(super_area)
            medics_after = sum(len(hospital.subgroups[hospital.SubgroupType.workers].people) 
                             for hospital in self.get_hospitals_in_super_area(super_area))
            medics_assigned = medics_after - medics_before
            total_medics_distributed += medics_assigned

            super_area_stats.append({
                "Super_Area": super_area.name,
                "Hospitals": len(self.get_hospitals_in_super_area(super_area)),
                "Medics_Assigned": medics_assigned,
                "Total_Medics": medics_after
            })

        # Collect all hospitals after distribution for comprehensive summary
        all_hospitals = [
            hospital
            for super_area in super_areas
            for hospital in self.get_hospitals_in_super_area(super_area)
        ]

        # Calculate comprehensive statistics
        hospitals_with_medics = sum(1 for h in all_hospitals if len(h.subgroups[h.SubgroupType.workers].people) > 0)
        total_hospital_capacity = sum(h.n_beds + h.n_icu_beds for h in all_hospitals)

        # Log comprehensive summary
        logger.info(f"=== MEDIC DISTRIBUTION SUMMARY ===")
        logger.info(f"Total medics distributed across all super areas: {total_medics_distributed}")
        logger.info(f"Total hospitals: {len(all_hospitals)}")
        logger.info(f"Hospitals with assigned medics: {hospitals_with_medics}")
        logger.info(f"Hospitals without medics: {len(all_hospitals) - hospitals_with_medics}")
        logger.info(f"Total hospital capacity (beds + ICU): {total_hospital_capacity}")
        logger.info(f"Average medics per hospital: {total_medics_distributed / len(all_hospitals) if all_hospitals else 0:.2f}")

    def get_hospitals_in_super_area(self, super_area: SuperArea) -> List["Hospital"]:
        """From all hospitals, filter the ones placed in a given super_area

        Args:
            super_area (SuperArea): super area

        """
        hospitals_in_super_area = [
            hospital
            for hospital in self.hospitals.members
            if hospital.super_area.name == super_area.name
        ]

        return hospitals_in_super_area

    def distribute_medics_to_hospitals(self, super_area: SuperArea):
        """Distribute medics to hospitals within a super area

        Args:
            super_area (SuperArea): super area to distribute medics

        """
        hospitals_in_super_area = self.get_hospitals_in_super_area(super_area)
        if not hospitals_in_super_area:
            return
        medics = [
            person
            for idx, person in enumerate(super_area.workers)
            if person.sector == self.healthcare_sector_label
            and person.age > self.medic_min_age
            and person.primary_activity is None
        ]
        if not medics:
            logger.info(
                f"\n The SuperArea {super_area.name} has no people that work in it!"
            )
            return
        else:
            shuffle(medics)
            for hospital in hospitals_in_super_area:
                max_capacity = hospital.n_beds + hospital.n_icu_beds
                if max_capacity == 0:
                    continue
                n_medics = min(
                    max(int(np.floor(max_capacity / self.patients_per_medic)), 1),
                    len(medics),
                )
                for _ in range(n_medics):
                    medic = medics.pop()
                    hospital.add(medic, hospital.SubgroupType.workers)
                    medic.lockdown_status = "key_worker"

    def assign_closest_hospitals_to_super_areas(self, super_areas):
        """

        Args:
            super_areas: 

        """
        if not self.hospitals.members:
            return
        for super_area in super_areas:
            super_area.closest_hospitals = self.hospitals.get_closest_hospitals(
                super_area.coordinates, self.hospitals.neighbour_hospitals
            )

        # Prepare data for visualization of a random sample of super areas
        sample_size = 5  # Adjust as needed
        sampled_super_areas = random.sample(list(super_areas), min(sample_size, len(super_areas)))        
        # Gather visualization data
        closest_hospitals_data = [
            {
                "Super Area": super_area.name,
                "Super Area Coordinates": super_area.coordinates,
                "Closest Hospital IDs": [hospital.id for hospital in super_area.closest_hospitals],
                "Closest Hospital Coordinates": [hospital.coordinates for hospital in super_area.closest_hospitals]
            }
            for super_area in sampled_super_areas
        ]

        # Display data as a DataFrame
        df_closest_hospitals = pd.DataFrame(closest_hospitals_data)
        print("\n===== Sample of Closest Hospitals Assigned to Super Areas =====")
        print(df_closest_hospitals)

__init__(hospitals, medic_min_age, patients_per_medic, healthcare_sector_label=None)

Parameters

hospitals: hospitals to populate with workers medic_min_age: minimum age to qualify as a worker patients_per_medic: ratio of patients per medic healthcare_sector_label: string that characterises the helathcare workers

Source code in june/distributors/hospital_distributor.py

def __init__(
    self,
    hospitals: Hospitals,
    medic_min_age: int,
    patients_per_medic: int,
    healthcare_sector_label: Optional[str] = None,
):
    """

    Parameters
    ----------
    hospitals:
        hospitals to populate with workers
    medic_min_age:
        minimum age to qualify as a worker
    patients_per_medic:
        ratio of patients per medic
    healthcare_sector_label:
        string that characterises the helathcare workers
    """
    # check if this msoarea has hospitals
    self.hospitals = hospitals
    self.medic_min_age = medic_min_age
    self.patients_per_medic = patients_per_medic
    self.healthcare_sector_label = healthcare_sector_label

assign_closest_hospitals_to_super_areas(super_areas)

Parameters:

Name	Type	Description	Default
super_areas			required

Source code in june/distributors/hospital_distributor.py

def assign_closest_hospitals_to_super_areas(self, super_areas):
    """

    Args:
        super_areas: 

    """
    if not self.hospitals.members:
        return
    for super_area in super_areas:
        super_area.closest_hospitals = self.hospitals.get_closest_hospitals(
            super_area.coordinates, self.hospitals.neighbour_hospitals
        )

    # Prepare data for visualization of a random sample of super areas
    sample_size = 5  # Adjust as needed
    sampled_super_areas = random.sample(list(super_areas), min(sample_size, len(super_areas)))        
    # Gather visualization data
    closest_hospitals_data = [
        {
            "Super Area": super_area.name,
            "Super Area Coordinates": super_area.coordinates,
            "Closest Hospital IDs": [hospital.id for hospital in super_area.closest_hospitals],
            "Closest Hospital Coordinates": [hospital.coordinates for hospital in super_area.closest_hospitals]
        }
        for super_area in sampled_super_areas
    ]

    # Display data as a DataFrame
    df_closest_hospitals = pd.DataFrame(closest_hospitals_data)
    print("\n===== Sample of Closest Hospitals Assigned to Super Areas =====")
    print(df_closest_hospitals)

distribute_medics_from_world(people)

Randomly distribute people from the world to work as medics for hospitals, useful if we don't have data on where do people work. It will still match the patients to medic ratio and the minimum age to be a medic.

Parameters:

Name	Type	Description	Default
people	List[Person]	list of Persons in the world	required

Source code in june/distributors/hospital_distributor.py

def distribute_medics_from_world(self, people: List["Person"]):
    """Randomly distribute people from the world to work as medics for hospitals,
    useful if we don't have data on where do people work. It will still
    match the patients to medic ratio and the minimum age to be a medic.

    Args:
        people (List["Person"]): list of Persons in the world

    """
    medics = [person for person in people if person.age >= self.medic_min_age]
    shuffle(medics)
    total_medics_distributed = 0
    hospital_assignments = []

    for hospital in self.hospitals:
        max_capacity = hospital.n_beds + hospital.n_icu_beds
        if max_capacity == 0:
            continue
        n_medics = max(int(np.floor(max_capacity / self.patients_per_medic)), 1)
        hospital_medics = 0
        for _ in range(n_medics):
            if not medics:
                logger.warning("Ran out of available medics before fully staffing all hospitals")
                break
            medic = medics.pop()
            hospital.add(medic, hospital.SubgroupType.workers)
            medic.lockdown_status = "key_worker"
            hospital_medics += 1
            total_medics_distributed += 1

        hospital_assignments.append({
            "Hospital_ID": hospital.id,
            "Beds": hospital.n_beds,
            "ICU_Beds": hospital.n_icu_beds,
            "Total_Capacity": max_capacity,
            "Medics_Assigned": hospital_medics
        })

    # Log summary
    logger.info(f"=== MEDIC DISTRIBUTION SUMMARY ===")
    logger.info(f"Total eligible medics available: {len(medics) + total_medics_distributed}")
    logger.info(f"Total medics distributed: {total_medics_distributed}")
    logger.info(f"Medics remaining unassigned: {len(medics)}")
    logger.info(f"Hospitals staffed: {len([h for h in hospital_assignments if h['Medics_Assigned'] > 0])}")
    logger.info(f"Total hospital capacity: {sum(h['Total_Capacity'] for h in hospital_assignments)}")

    # Display detailed breakdown as DataFrame
    df_assignments = pd.DataFrame(hospital_assignments)
    print("\n=== HOSPITAL MEDIC ASSIGNMENTS ===")
    print(df_assignments.to_string(index=False))

distribute_medics_to_hospitals(super_area)

Distribute medics to hospitals within a super area

Parameters:

Name	Type	Description	Default
super_area	SuperArea	super area to distribute medics	required

Source code in june/distributors/hospital_distributor.py

def distribute_medics_to_hospitals(self, super_area: SuperArea):
    """Distribute medics to hospitals within a super area

    Args:
        super_area (SuperArea): super area to distribute medics

    """
    hospitals_in_super_area = self.get_hospitals_in_super_area(super_area)
    if not hospitals_in_super_area:
        return
    medics = [
        person
        for idx, person in enumerate(super_area.workers)
        if person.sector == self.healthcare_sector_label
        and person.age > self.medic_min_age
        and person.primary_activity is None
    ]
    if not medics:
        logger.info(
            f"\n The SuperArea {super_area.name} has no people that work in it!"
        )
        return
    else:
        shuffle(medics)
        for hospital in hospitals_in_super_area:
            max_capacity = hospital.n_beds + hospital.n_icu_beds
            if max_capacity == 0:
                continue
            n_medics = min(
                max(int(np.floor(max_capacity / self.patients_per_medic)), 1),
                len(medics),
            )
            for _ in range(n_medics):
                medic = medics.pop()
                hospital.add(medic, hospital.SubgroupType.workers)
                medic.lockdown_status = "key_worker"

distribute_medics_to_super_areas(super_areas)

Distribute medics to super areas, flow data is necessary to find medics in the super area according to their sector.

Parameters:

Name	Type	Description	Default
super_areas	SuperAreas	object containing all the super areas to distribute medics	required

Source code in june/distributors/hospital_distributor.py

def distribute_medics_to_super_areas(self, super_areas: SuperAreas):
    """Distribute medics to super areas, flow data is necessary to find medics in the
    super area according to their sector.

    Args:
        super_areas (SuperAreas): object containing all the super areas to distribute medics

    """
    logger.info("Distributing medics to hospitals")
    total_medics_distributed = 0
    super_area_stats = []

    for super_area in super_areas:
        medics_before = sum(len(hospital.subgroups[hospital.SubgroupType.workers].people) 
                          for hospital in self.get_hospitals_in_super_area(super_area))
        self.distribute_medics_to_hospitals(super_area)
        medics_after = sum(len(hospital.subgroups[hospital.SubgroupType.workers].people) 
                         for hospital in self.get_hospitals_in_super_area(super_area))
        medics_assigned = medics_after - medics_before
        total_medics_distributed += medics_assigned

        super_area_stats.append({
            "Super_Area": super_area.name,
            "Hospitals": len(self.get_hospitals_in_super_area(super_area)),
            "Medics_Assigned": medics_assigned,
            "Total_Medics": medics_after
        })

    # Collect all hospitals after distribution for comprehensive summary
    all_hospitals = [
        hospital
        for super_area in super_areas
        for hospital in self.get_hospitals_in_super_area(super_area)
    ]

    # Calculate comprehensive statistics
    hospitals_with_medics = sum(1 for h in all_hospitals if len(h.subgroups[h.SubgroupType.workers].people) > 0)
    total_hospital_capacity = sum(h.n_beds + h.n_icu_beds for h in all_hospitals)

    # Log comprehensive summary
    logger.info(f"=== MEDIC DISTRIBUTION SUMMARY ===")
    logger.info(f"Total medics distributed across all super areas: {total_medics_distributed}")
    logger.info(f"Total hospitals: {len(all_hospitals)}")
    logger.info(f"Hospitals with assigned medics: {hospitals_with_medics}")
    logger.info(f"Hospitals without medics: {len(all_hospitals) - hospitals_with_medics}")
    logger.info(f"Total hospital capacity (beds + ICU): {total_hospital_capacity}")
    logger.info(f"Average medics per hospital: {total_medics_distributed / len(all_hospitals) if all_hospitals else 0:.2f}")

from_file(hospitals, config_filename=default_config_filename) classmethod

Parameters:

Name	Type	Description	Default
hospitals			required
config_filename		(Default value = default_config_filename)	default_config_filename

Source code in june/distributors/hospital_distributor.py

@classmethod
def from_file(cls, hospitals, config_filename=default_config_filename):
    """

    Args:
        hospitals: 
        config_filename: (Default value = default_config_filename)

    """
    with open(config_filename) as f:
        config = yaml.load(f, Loader=yaml.FullLoader)
    return HospitalDistributor(
        hospitals=hospitals,
        medic_min_age=config["medic_min_age"],
        patients_per_medic=config["patients_per_medic"],
        healthcare_sector_label=config["healthcare_sector_label"],
    )

get_hospitals_in_super_area(super_area)

From all hospitals, filter the ones placed in a given super_area

Parameters:

Name	Type	Description	Default
super_area	SuperArea	super area	required

Source code in june/distributors/hospital_distributor.py

def get_hospitals_in_super_area(self, super_area: SuperArea) -> List["Hospital"]:
    """From all hospitals, filter the ones placed in a given super_area

    Args:
        super_area (SuperArea): super area

    """
    hospitals_in_super_area = [
        hospital
        for hospital in self.hospitals.members
        if hospital.super_area.name == super_area.name
    ]

    return hospitals_in_super_area