School

InteractiveSchool

Bases: InteractiveGroup

Source code in june/groups/school.py

class InteractiveSchool(InteractiveGroup):
    """ """
    def __init__(self, group: "Group", people_from_abroad=None):
        super().__init__(group=group, people_from_abroad=people_from_abroad)
        self.school_years = group.years
        self.sector = group.sector

    @classmethod
    def get_raw_contact_matrix(
        cls, contact_matrix, alpha_physical, proportion_physical, characteristic_time
    ):
        """Creates a global contact matrix for school, which is by default 20x20, to take into account
        all possible school years combinations. Each school will then use a slice of this matrix.
        We assume that the number of contacts between two different school years goes as
        $ xi**abs(age_difference_between_school_years) * contacts_between_students$
        Teacher contacts are left as specified in the config file.

        Args:
            contact_matrix: 
            alpha_physical: 
            proportion_physical: 
            characteristic_time: 

        """
        xi = 0.3
        age_min = 0
        age_max = 30
        n_subgroups_max = (age_max - age_min) + 2  # adding teachers
        age_differences = np.subtract.outer(
            range(age_min, age_max + 1), range(age_min, age_max + 1)
        )
        processed_contact_matrix = np.zeros((n_subgroups_max, n_subgroups_max))
        processed_contact_matrix[0, 0] = contact_matrix[0][0]
        processed_contact_matrix[0, 1:] = contact_matrix[0][1]
        processed_contact_matrix[1:, 0] = contact_matrix[1][0]
        processed_contact_matrix[1:, 1:] = (
            xi ** abs(age_differences) * contact_matrix[1][1]
        )
        physical_ratios = np.zeros((n_subgroups_max, n_subgroups_max))
        physical_ratios[0, 0] = proportion_physical[0][0]
        physical_ratios[0, 1:] = proportion_physical[0][1]
        physical_ratios[1:, 0] = proportion_physical[1][0]
        physical_ratios[1:, 1:] = proportion_physical[1][1]
        # add physical contacts
        processed_contact_matrix = processed_contact_matrix * (
            1.0 + (alpha_physical - 1.0) * physical_ratios
        )
        processed_contact_matrix *= 24 / characteristic_time
        # If same age but different class room, reduce contacts
        return processed_contact_matrix

    def get_processed_contact_matrix(self, contact_matrix):
        """

        Args:
            contact_matrix: 

        """
        n_school_years = len(self.school_years)
        n_subgroups = n_school_years + 1
        ret = np.zeros((n_subgroups, n_subgroups))
        for i in range(0, n_subgroups):
            for j in range(0, n_subgroups):
                if i == j:
                    if i != 0:
                        ret[i, j] = contact_matrix[1, 1]
                    else:
                        ret[0, 0] = contact_matrix[0, 0]
                else:
                    if i == 0:
                        ret[0, j] = contact_matrix[0][1] / n_school_years
                    elif j == 0:
                        ret[i, 0] = contact_matrix[1][0] / n_school_years
                    else:
                        year_idx_i = _translate_school_subgroup(i, self.school_years)
                        year_idx_j = _translate_school_subgroup(j, self.school_years)
                        if year_idx_i == year_idx_j:
                            ret[i, j] = contact_matrix[year_idx_i, year_idx_j] / 4
                        else:
                            ret[i, j] = contact_matrix[year_idx_i, year_idx_j]
        return ret

    def get_processed_beta(self, betas, beta_reductions):
        """Returns the processed contact intensity, by taking into account the policies
        beta reductions and regional compliance. This is a group method as different interactive
        groups may choose to treat this differently.

        Args:
            betas: 
            beta_reductions: 

        """
        if self.sector is None:
            spec = "school"
        elif "secondary" in self.sector:
            spec = "secondary_school"
        else:
            spec = "primary_school"
        if spec in betas:
            beta = betas[spec]
        else:
            beta = betas["school"]
        if spec in beta_reductions:
            beta_reduction = beta_reductions[spec]
        else:
            beta_reduction = beta_reductions.get("school", 1.0)
        try:
            regional_compliance = self.super_area.region.regional_compliance
        except AttributeError:
            regional_compliance = 1
        try:
            lockdown_tier = self.super_area.region.policy["lockdown_tier"]
            if lockdown_tier is None:
                lockdown_tier = 1
        except Exception:
            lockdown_tier = 1
        if int(lockdown_tier) == 4:
            tier_reduction = 0.5
        else:
            tier_reduction = 1.0

        return beta * (1 + regional_compliance * tier_reduction * (beta_reduction - 1))

get_processed_beta(betas, beta_reductions)

Returns the processed contact intensity, by taking into account the policies beta reductions and regional compliance. This is a group method as different interactive groups may choose to treat this differently.

Parameters:

Name	Type	Description	Default
betas			required
beta_reductions			required

Source code in june/groups/school.py

def get_processed_beta(self, betas, beta_reductions):
    """Returns the processed contact intensity, by taking into account the policies
    beta reductions and regional compliance. This is a group method as different interactive
    groups may choose to treat this differently.

    Args:
        betas: 
        beta_reductions: 

    """
    if self.sector is None:
        spec = "school"
    elif "secondary" in self.sector:
        spec = "secondary_school"
    else:
        spec = "primary_school"
    if spec in betas:
        beta = betas[spec]
    else:
        beta = betas["school"]
    if spec in beta_reductions:
        beta_reduction = beta_reductions[spec]
    else:
        beta_reduction = beta_reductions.get("school", 1.0)
    try:
        regional_compliance = self.super_area.region.regional_compliance
    except AttributeError:
        regional_compliance = 1
    try:
        lockdown_tier = self.super_area.region.policy["lockdown_tier"]
        if lockdown_tier is None:
            lockdown_tier = 1
    except Exception:
        lockdown_tier = 1
    if int(lockdown_tier) == 4:
        tier_reduction = 0.5
    else:
        tier_reduction = 1.0

    return beta * (1 + regional_compliance * tier_reduction * (beta_reduction - 1))

get_processed_contact_matrix(contact_matrix)

Parameters:

Name	Type	Description	Default
contact_matrix			required

Source code in june/groups/school.py

def get_processed_contact_matrix(self, contact_matrix):
    """

    Args:
        contact_matrix: 

    """
    n_school_years = len(self.school_years)
    n_subgroups = n_school_years + 1
    ret = np.zeros((n_subgroups, n_subgroups))
    for i in range(0, n_subgroups):
        for j in range(0, n_subgroups):
            if i == j:
                if i != 0:
                    ret[i, j] = contact_matrix[1, 1]
                else:
                    ret[0, 0] = contact_matrix[0, 0]
            else:
                if i == 0:
                    ret[0, j] = contact_matrix[0][1] / n_school_years
                elif j == 0:
                    ret[i, 0] = contact_matrix[1][0] / n_school_years
                else:
                    year_idx_i = _translate_school_subgroup(i, self.school_years)
                    year_idx_j = _translate_school_subgroup(j, self.school_years)
                    if year_idx_i == year_idx_j:
                        ret[i, j] = contact_matrix[year_idx_i, year_idx_j] / 4
                    else:
                        ret[i, j] = contact_matrix[year_idx_i, year_idx_j]
    return ret

get_raw_contact_matrix(contact_matrix, alpha_physical, proportion_physical, characteristic_time) classmethod

Creates a global contact matrix for school, which is by default 20x20, to take into account all possible school years combinations. Each school will then use a slice of this matrix. We assume that the number of contacts between two different school years goes as $ xi**abs(age_difference_between_school_years) * contacts_between_students$ Teacher contacts are left as specified in the config file.

Parameters:

Name	Type	Description	Default
contact_matrix			required
alpha_physical			required
proportion_physical			required
characteristic_time			required

Source code in june/groups/school.py

@classmethod
def get_raw_contact_matrix(
    cls, contact_matrix, alpha_physical, proportion_physical, characteristic_time
):
    """Creates a global contact matrix for school, which is by default 20x20, to take into account
    all possible school years combinations. Each school will then use a slice of this matrix.
    We assume that the number of contacts between two different school years goes as
    $ xi**abs(age_difference_between_school_years) * contacts_between_students$
    Teacher contacts are left as specified in the config file.

    Args:
        contact_matrix: 
        alpha_physical: 
        proportion_physical: 
        characteristic_time: 

    """
    xi = 0.3
    age_min = 0
    age_max = 30
    n_subgroups_max = (age_max - age_min) + 2  # adding teachers
    age_differences = np.subtract.outer(
        range(age_min, age_max + 1), range(age_min, age_max + 1)
    )
    processed_contact_matrix = np.zeros((n_subgroups_max, n_subgroups_max))
    processed_contact_matrix[0, 0] = contact_matrix[0][0]
    processed_contact_matrix[0, 1:] = contact_matrix[0][1]
    processed_contact_matrix[1:, 0] = contact_matrix[1][0]
    processed_contact_matrix[1:, 1:] = (
        xi ** abs(age_differences) * contact_matrix[1][1]
    )
    physical_ratios = np.zeros((n_subgroups_max, n_subgroups_max))
    physical_ratios[0, 0] = proportion_physical[0][0]
    physical_ratios[0, 1:] = proportion_physical[0][1]
    physical_ratios[1:, 0] = proportion_physical[1][0]
    physical_ratios[1:, 1:] = proportion_physical[1][1]
    # add physical contacts
    processed_contact_matrix = processed_contact_matrix * (
        1.0 + (alpha_physical - 1.0) * physical_ratios
    )
    processed_contact_matrix *= 24 / characteristic_time
    # If same age but different class room, reduce contacts
    return processed_contact_matrix

School

Bases: Group

Source code in june/groups/school.py

class School(Group):
    """ """
    __slots__ = (
        "id",
        "coordinates",
        "n_pupils_max",
        "n_teachers_max",
        "age_min",
        "age_max",
        "age_structure",
        "sector",
        "years",
        "registered_members_ids",
        "boy_ratio",
        "girl_ratio",
        "student_deaths",
        "student_icu_transfers",
        "households_avoiding_school",
        "households_last_decision_at_death_count",
        "households_last_decision_at_icu_count",
    )

    def __init__(
        self,
        coordinates: Tuple[float, float] = None,
        n_pupils_max: int = None,
        age_min: int = 0,
        age_max: int = 18,
        sector: str = None,
        area: Area = None,
        n_classrooms: Optional[int] = None,
        years: Optional[int] = None,
        registered_members_ids: dict = None,
        boy_ratio: float = 0.5,
        girl_ratio: float = 0.5
    ):
        """
        Create a School given its description.
        """

        super().__init__()
        self.subgroups = []
        if n_classrooms is None:
            n_classrooms = age_max - age_min
        self.subgroups = [SchoolClass(self, i) for i in range(n_classrooms + 2)]

        self.n_classrooms = n_classrooms
        self.coordinates = coordinates
        self.area = area
        self.n_pupils_max = n_pupils_max
        self.n_teachers_max = None
        self.age_min = age_min
        self.age_max = age_max
        self.sector = sector
        self.years = tuple(range(age_min, age_max + 1)) if years is None else tuple(years)
        self.registered_members_ids = registered_members_ids if registered_members_ids is not None else {}
        self.boy_ratio = boy_ratio
        self.girl_ratio = girl_ratio
        # Death tracking for school avoidance behavior
        self.student_deaths = 0  # Number of student deaths at this school
        self.student_icu_transfers = 0  # Number of student ICU transfers at this school
        self.households_avoiding_school = set()  # Set of household IDs that are avoiding this school
        self.households_last_decision_at_death_count = {}  # household_id -> death_count when they last made a decision
        self.households_last_decision_at_icu_count = {}  # household_id -> icu_count when they last made a decision

    def get_interactive_group(self, people_from_abroad=None):
        """

        Args:
            people_from_abroad: (Default value = None)

        """
        return InteractiveSchool(self, people_from_abroad=people_from_abroad)

    def add(self, person):
        """

        Args:
            person: 

        """
        if person.age <= self.age_max:
            subgroup = self.subgroups[1 + person.age - self.age_min]
            subgroup.append(person)
            person.subgroups.primary_activity = subgroup
        else:  # teacher
            subgroup = self.subgroups[0]
            subgroup.append(person)
            person.subgroups.primary_activity = subgroup

    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)

    def limit_classroom_sizes(self, max_classroom_size: int):
        """Make all subgroups smaller than ```max_classroom_size```

        Args:
            max_classroom_size (int): maximum number of students per classroom (subgroup)

        """
        age_subgroups = self.subgroups.copy()
        year_age_group = deepcopy(self.years)
        self.subgroups = [age_subgroups[0]]  # keep teachers
        self.years = []
        counter = 1
        for idx, subgroup in enumerate(age_subgroups[1:]):
            if len(subgroup.people) > max_classroom_size:
                n_classrooms = math.ceil(len(subgroup.people) / max_classroom_size)
                self.years += [year_age_group[idx]] * n_classrooms
                pupils_in_classroom = np.array_split(subgroup.people, n_classrooms)
                for i in range(n_classrooms):
                    classroom = SchoolClass(self, counter)
                    for pupil in pupils_in_classroom[i]:
                        classroom.append(pupil)
                        pupil.subgroups.primary_activity = classroom
                    self.subgroups.append(classroom)
                    counter += 1
            else:
                subgroup.subgroup_type = counter
                self.subgroups.append(subgroup)
                counter += 1
                self.years.append(year_age_group[idx])
        self.years = tuple(self.years)
        self.n_classrooms = len(self.subgroups) - 1

    @property
    def is_full(self):
        """ """
        if self.n_pupils >= self.n_pupils_max:
            return True
        return False

    @property
    def n_pupils(self):
        """ """
        return len(self.students)

    @property
    def n_teachers(self):
        """ """
        return len(self.teachers)

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

    @property
    def students(self):
        """ """
        ret = []
        for subgroup in self.subgroups[1:]:
            ret += subgroup.people
        return ret

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

is_full property

n_pupils property

n_teachers property

students property

super_area property

teachers property

__init__(coordinates=None, n_pupils_max=None, age_min=0, age_max=18, sector=None, area=None, n_classrooms=None, years=None, registered_members_ids=None, boy_ratio=0.5, girl_ratio=0.5)

Create a School given its description.

Source code in june/groups/school.py

def __init__(
    self,
    coordinates: Tuple[float, float] = None,
    n_pupils_max: int = None,
    age_min: int = 0,
    age_max: int = 18,
    sector: str = None,
    area: Area = None,
    n_classrooms: Optional[int] = None,
    years: Optional[int] = None,
    registered_members_ids: dict = None,
    boy_ratio: float = 0.5,
    girl_ratio: float = 0.5
):
    """
    Create a School given its description.
    """

    super().__init__()
    self.subgroups = []
    if n_classrooms is None:
        n_classrooms = age_max - age_min
    self.subgroups = [SchoolClass(self, i) for i in range(n_classrooms + 2)]

    self.n_classrooms = n_classrooms
    self.coordinates = coordinates
    self.area = area
    self.n_pupils_max = n_pupils_max
    self.n_teachers_max = None
    self.age_min = age_min
    self.age_max = age_max
    self.sector = sector
    self.years = tuple(range(age_min, age_max + 1)) if years is None else tuple(years)
    self.registered_members_ids = registered_members_ids if registered_members_ids is not None else {}
    self.boy_ratio = boy_ratio
    self.girl_ratio = girl_ratio
    # Death tracking for school avoidance behavior
    self.student_deaths = 0  # Number of student deaths at this school
    self.student_icu_transfers = 0  # Number of student ICU transfers at this school
    self.households_avoiding_school = set()  # Set of household IDs that are avoiding this school
    self.households_last_decision_at_death_count = {}  # household_id -> death_count when they last made a decision
    self.households_last_decision_at_icu_count = {}  # household_id -> icu_count when they last made a decision

add(person)

Parameters:

Name	Type	Description	Default
person			required

Source code in june/groups/school.py

def add(self, person):
    """

    Args:
        person: 

    """
    if person.age <= self.age_max:
        subgroup = self.subgroups[1 + person.age - self.age_min]
        subgroup.append(person)
        person.subgroups.primary_activity = subgroup
    else:  # teacher
        subgroup = self.subgroups[0]
        subgroup.append(person)
        person.subgroups.primary_activity = subgroup

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/school.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_interactive_group(people_from_abroad=None)

Parameters:

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

Source code in june/groups/school.py

def get_interactive_group(self, people_from_abroad=None):
    """

    Args:
        people_from_abroad: (Default value = None)

    """
    return InteractiveSchool(self, people_from_abroad=people_from_abroad)

limit_classroom_sizes(max_classroom_size)

Make all subgroups smaller than max_classroom_size

Parameters:

Name	Type	Description	Default
max_classroom_size	int	maximum number of students per classroom (subgroup)	required

Source code in june/groups/school.py

def limit_classroom_sizes(self, max_classroom_size: int):
    """Make all subgroups smaller than ```max_classroom_size```

    Args:
        max_classroom_size (int): maximum number of students per classroom (subgroup)

    """
    age_subgroups = self.subgroups.copy()
    year_age_group = deepcopy(self.years)
    self.subgroups = [age_subgroups[0]]  # keep teachers
    self.years = []
    counter = 1
    for idx, subgroup in enumerate(age_subgroups[1:]):
        if len(subgroup.people) > max_classroom_size:
            n_classrooms = math.ceil(len(subgroup.people) / max_classroom_size)
            self.years += [year_age_group[idx]] * n_classrooms
            pupils_in_classroom = np.array_split(subgroup.people, n_classrooms)
            for i in range(n_classrooms):
                classroom = SchoolClass(self, counter)
                for pupil in pupils_in_classroom[i]:
                    classroom.append(pupil)
                    pupil.subgroups.primary_activity = classroom
                self.subgroups.append(classroom)
                counter += 1
        else:
            subgroup.subgroup_type = counter
            self.subgroups.append(subgroup)
            counter += 1
            self.years.append(year_age_group[idx])
    self.years = tuple(self.years)
    self.n_classrooms = len(self.subgroups) - 1

SchoolClass

Bases: Subgroup

Source code in june/groups/school.py

class SchoolClass(Subgroup):
    """ """
    def __init__(self, group, subgroup_type: int):
        super().__init__(group, subgroup_type)
        self.quarantine_starting_date = -np.inf

SchoolError

Bases: BaseException

Source code in june/groups/school.py

class SchoolError(BaseException):
    """ """
    pass

Schools

Bases: Supergroup

Source code in june/groups/school.py

class Schools(Supergroup):
    """ """
    venue_class = School

    def __init__(
        self,
        schools: List["venue_class"],
        school_trees: Optional[Dict[int, BallTree]] = None,
        agegroup_to_global_indices: dict = None,
    ):
        """
        Create a group of Schools, and provide functionality to access closest school

        Parameters
        ----------
        area_names
            list of areas for which to build schools
        schools:
            list of school instances
        school_tree:
            BallTree built on all schools coordinates
        agegroup_to_global_indices:
            dictionary to map the
        """
        super().__init__(members=schools)
        self.school_trees = school_trees
        self.school_agegroup_to_global_indices = agegroup_to_global_indices

    @classmethod
    def for_geography(
        cls,
        geography: Geography,
        config_file: str = default_config_filename,
    ) -> "Schools":
        """Creates Schools for the given geography with a disease configuration.

        Args:
            geography (Geography): 
            config_file (str, optional): (Default value = default_config_filename)

        """
        schools = cls.for_areas(
            geography.areas,
            config_file=config_file
        )

        # Sample 5 schools from each area for visualization
        sampled_schools = []
        for area in geography.areas:
            if hasattr(area, 'schools') and area.schools:
                # Sample 5 schools or fewer if there are less than 5
                sample_schools = random.sample(area.schools, min(5, len(area.schools)))
                for school in sample_schools:
                    sampled_schools.append({
                        "| School ID": school.id,
                        "| Area": area.name,
                        "| School Sector": school.sector,
                        "| Number of Pupils": school.n_pupils,
                        "| Coordinates": school.coordinates,
                        "| Max Pupils": school.n_pupils_max,
                        "| Min Age": school.age_min,
                        "| Max Age": school.age_max
                    })

        # Convert the sample data to a DataFrame
        df_schools = pd.DataFrame(sampled_schools)
        print("\n===== Sample of Created Schools =====")
        print(df_schools)

        logger.info(f"Created {len(schools)} schools in this geography.")
        return schools

    @classmethod
    def for_areas(
        cls,
        areas: Areas,
        data_file_ew: str = default_data_filename_ew,
        data_file_sct: str = default_data_filename_sct,
        data_file_ni: str = default_data_filename_ni,
        config_file: str = default_config_filename,
    ) -> "Schools":
        """Creates Schools for specified areas using a disease configuration.

        Args:
            areas (Areas): 
            data_file_ew (str, optional): (Default value = default_data_filename_ew)
            data_file_sct (str, optional): (Default value = default_data_filename_sct)
            data_file_ni (str, optional): (Default value = default_data_filename_ni)
            config_file (str, optional): (Default value = default_config_filename)

        """
        return cls.from_file(areas, data_file_ew, data_file_sct, data_file_ni)

    @classmethod
    def from_file(
        cls,
        areas: Areas,
        data_file_ew: str = default_data_filename_ew,
        data_file_sct: str = default_data_filename_sct,
        data_file_ni: str = default_data_filename_ni,
    ) -> "Schools":
        """Initialise Schools from England/Wales, Scotland, and Northern Ireland data frames.

        Args:
            areas (Areas): 
            data_file_ew (str, optional): (Default value = default_data_filename_ew)
            data_file_sct (str, optional): (Default value = default_data_filename_sct)
            data_file_ni (str, optional): (Default value = default_data_filename_ni)

        """
        # Load England/Wales schools
        school_df_ew = pd.read_csv(data_file_ew)
        school_df_ew = school_df_ew.rename(columns={
            'OA': 'oa',
            'NumberOfPupils': 'NOR',
            'PhaseOfEducation': 'sector',
            'StatutoryLowAge': 'age_min',
            'StatutoryHighAge': 'age_max',
            'Latitude': 'latitude',
            'Longitude': 'longitude',
            'NumberOfBoys': 'n_boys',
            'NumberOfGirls': 'n_girls'
        })

        # Load Scotland schools
        school_df_sct = pd.read_csv(data_file_sct)
        school_df_sct = school_df_sct.rename(columns={
            'OA': 'oa',
            'NumberOfPupils': 'NOR',
            'PhaseOfEducation': 'sector',
            'Latitude': 'latitude',
            'Longitude': 'longitude',
            'NumberOfBoys': 'n_boys',
            'NumberOfGirls': 'n_girls'
        })
        # Scotland doesn't have explicit age columns, we'll handle this in build_schools_for_areas
        school_df_sct['age_min'] = np.nan
        school_df_sct['age_max'] = np.nan

        # Load Northern Ireland schools
        school_df_ni = pd.read_csv(data_file_ni)
        school_df_ni = school_df_ni.rename(columns={
            'OA': 'oa',
            'NumberOfPupils': 'NOR',
            'PhaseOfEducation': 'sector',
            'Latitude': 'latitude',
            'Longitude': 'longitude',
            'NumberOfBoys': 'n_boys',
            'NumberOfGirls': 'n_girls',
            'StatutoryLowAge': 'age_min',
            'StatutoryHighAge': 'age_max'
        })

        # Combine all dataframes
        school_df = pd.concat([school_df_ew, school_df_sct, school_df_ni], ignore_index=True)

        # Apply age defaults to the dataframe BEFORE building trees
        def get_default_ages(sector):
            """

            Args:
                sector: 

            """
            if pd.isna(sector):
                return 4, 11
            sector_lower = str(sector).lower()
            if 'primary' in sector_lower:
                return 4, 11
            elif 'secondary' in sector_lower:
                return 11, 18
            elif 'nursery' in sector_lower:
                return 2, 5
            else:
                return 4, 11

        # Apply defaults where missing
        for idx, row in school_df.iterrows():
            if pd.isna(row['age_min']) or pd.isna(row['age_max']):
                default_min, default_max = get_default_ages(row['sector'])
                school_df.at[idx, 'age_min'] = default_min
                school_df.at[idx, 'age_max'] = default_max

        # Ensure age columns are numeric
        school_df['age_min'] = pd.to_numeric(school_df['age_min'], errors='coerce')
        school_df['age_max'] = pd.to_numeric(school_df['age_max'], errors='coerce')

        # Remove special schools from dataframe
        school_df = school_df[~school_df['sector'].str.lower().str.contains('special', na=False)]

        # Filter by areas
        area_names = [area.name for area in areas]
        if area_names is not None:
            school_df = school_df[school_df["oa"].isin(area_names)]
        school_df.reset_index(drop=True, inplace=True)
        logger.info(f"There are {len(school_df)} schools in this geography.")
        return cls.build_schools_for_areas(areas, school_df)

    @classmethod
    def build_schools_for_areas(
        cls,
        areas: Areas,
        school_df: pd.DataFrame,
        age_range: Tuple[int, int] = (0, 19),
        employee_per_clients: Dict[str, int] = None,
    ) -> "Schools":
        """Build schools for specified areas with disease configuration.

        Args:
            areas (Areas): 
            school_df (pd.DataFrame): 
            age_range (Tuple[int, int], optional): (Default value = (0, 19))
            employee_per_clients (Dict[str, int], optional): (Default value = None)

        """

        def get_default_ages(sector):
            """Get default age ranges based on school sector/phase

            Args:
                sector: 

            """
            if pd.isna(sector):
                return 4, 11  # default to primary
            sector_lower = str(sector).lower()
            if 'primary' in sector_lower:
                return 4, 11
            elif 'secondary' in sector_lower:
                return 11, 18
            elif 'nursery' in sector_lower:
                return 2, 5
            else:
                return 4, 11  # default to primary

        employee_per_clients = employee_per_clients or {"primary": 30, "secondary": 30}
        schools = []
        for school_name, row in school_df.iterrows():
            # Handle missing pupil numbers
            n_pupils_max = row["NOR"] if pd.notna(row["NOR"]) else 100  # default

            # Handle school type/sector
            school_type = row["sector"] if pd.notna(row["sector"]) else "primary"

            # Handle coordinates
            if pd.notna(row["latitude"]) and pd.notna(row["longitude"]):
                coordinates = np.array([row["latitude"], row["longitude"]], dtype=np.float64)
            else:
                continue  # skip schools without coordinates

            # Handle ages with defaults based on sector
            if pd.notna(row["age_min"]) and pd.notna(row["age_max"]):
                age_min = int(row["age_min"])
                age_max = min(int(row["age_max"]), 19)  # Cap to prevent contact matrix index errors
            else:
                age_min, age_max = get_default_ages(school_type)

            # Calculate gender ratios
            n_boys = row["n_boys"] if pd.notna(row["n_boys"]) else 0
            n_girls = row["n_girls"] if pd.notna(row["n_girls"]) else 0

            # Calculate ratios with defaults
            total_gender_pupils = n_boys + n_girls
            if total_gender_pupils > 0:
                boy_ratio = n_boys / total_gender_pupils
                girl_ratio = n_girls / total_gender_pupils
            else:
                # Default to 50/50 split when no gender data available
                boy_ratio = 0.5
                girl_ratio = 0.5

            # Use the area code specified in the CSV rather than closest geographic area
            try:
                area = areas.members_by_name[row["oa"]]
            except KeyError:
                # Fallback to closest area if the specified area isn't in our geography
                area = areas.get_closest_area(coordinates)
            school = cls.venue_class(
                coordinates=coordinates,
                n_pupils_max=int(n_pupils_max),
                age_min=age_min,
                age_max=age_max,
                sector=school_type,
                area=area,
                boy_ratio=boy_ratio,
                girl_ratio=girl_ratio
            )
            schools.append(school)
            area.schools.append(school)

        school_trees, agegroup_to_global_indices = Schools.init_trees(school_df, age_range)
        return Schools(
            schools,
            school_trees=school_trees,
            agegroup_to_global_indices=agegroup_to_global_indices,
        )

    @staticmethod
    def init_trees(school_df: pd.DataFrame, age_range: Tuple[int, int]) -> "Schools":
        """Create trees to easily find the closest school that
        accepts a pupil given their age

        Args:
            school_df (pd.DataFrame): dataframe with school characteristics data
            age_range (Tuple[int, int]): 

        """
        school_trees = {}
        school_agegroup_to_global_indices = {
            k: [] for k in range(int(age_range[0]), int(age_range[1]) + 1)
        }
        # have a tree per age
        for age in range(int(age_range[0]), int(age_range[1]) + 1):
            _school_df_agegroup = school_df[
                (school_df["age_min"] <= age) & (school_df["age_max"] >= age)
            ]
            schools_coords = _school_df_agegroup[["latitude", "longitude"]].values
            if not schools_coords.size:
                logger.info(f"No school for the age {age} in this world.")
                continue
            school_trees[age] = Schools._create_school_tree(schools_coords)
            school_agegroup_to_global_indices[age] = _school_df_agegroup.index.values
        return school_trees, school_agegroup_to_global_indices

    @staticmethod
    def _create_school_tree(schools_coordinates: np.ndarray) -> BallTree:
        """Reads school location and sizes, it initialises a KD tree on a sphere,
        to query the closest schools to a given location.

        Args:
            schools_coordinates (np.ndarray): 

        Returns:
            Tree to query nearby schools: 

        """
        school_tree = BallTree(np.deg2rad(schools_coordinates), metric="haversine")
        return school_tree

    def get_closest_schools(
        self, age: int, coordinates: Tuple[float, float], k: int
    ) -> int:
        """Get the k-th closest school to a given coordinate, that accepts pupils
        aged age

        Args:
            age (int): age of the pupil
            coordinates (Tuple[float, float]): latitude and longitude
            k (int): k-th neighbour

        Returns:
            ID of the k-th closest school, within school trees for: 
            a given age group: 

        """
        school_tree = self.school_trees[age]
        coordinates_rad = np.deg2rad(coordinates).reshape(1, -1)
        k = min(k, school_tree.data.shape[0])
        distances, neighbours = school_tree.query(
            coordinates_rad, k=k, sort_results=True
        )
        return neighbours[0]

    @property
    def n_teachers(self):
        """ """
        return sum([school.n_teachers for school in self.members])

    @property
    def n_pupils(self):
        """ """
        return sum([school.n_pupils for school in self.members])

n_pupils property

n_teachers property

__init__(schools, school_trees=None, agegroup_to_global_indices=None)

Create a group of Schools, and provide functionality to access closest school

Parameters

area_names list of areas for which to build schools schools: list of school instances school_tree: BallTree built on all schools coordinates agegroup_to_global_indices: dictionary to map the

Source code in june/groups/school.py

def __init__(
    self,
    schools: List["venue_class"],
    school_trees: Optional[Dict[int, BallTree]] = None,
    agegroup_to_global_indices: dict = None,
):
    """
    Create a group of Schools, and provide functionality to access closest school

    Parameters
    ----------
    area_names
        list of areas for which to build schools
    schools:
        list of school instances
    school_tree:
        BallTree built on all schools coordinates
    agegroup_to_global_indices:
        dictionary to map the
    """
    super().__init__(members=schools)
    self.school_trees = school_trees
    self.school_agegroup_to_global_indices = agegroup_to_global_indices

build_schools_for_areas(areas, school_df, age_range=(0, 19), employee_per_clients=None) classmethod

Build schools for specified areas with disease configuration.

Parameters:

Name	Type	Description	Default
areas	Areas		required
school_df	DataFrame		required
age_range	Tuple[int, int]	(Default value = (0, 19))	(0, 19)
employee_per_clients	Dict[str, int]	(Default value = None)	None

Source code in june/groups/school.py

@classmethod
def build_schools_for_areas(
    cls,
    areas: Areas,
    school_df: pd.DataFrame,
    age_range: Tuple[int, int] = (0, 19),
    employee_per_clients: Dict[str, int] = None,
) -> "Schools":
    """Build schools for specified areas with disease configuration.

    Args:
        areas (Areas): 
        school_df (pd.DataFrame): 
        age_range (Tuple[int, int], optional): (Default value = (0, 19))
        employee_per_clients (Dict[str, int], optional): (Default value = None)

    """

    def get_default_ages(sector):
        """Get default age ranges based on school sector/phase

        Args:
            sector: 

        """
        if pd.isna(sector):
            return 4, 11  # default to primary
        sector_lower = str(sector).lower()
        if 'primary' in sector_lower:
            return 4, 11
        elif 'secondary' in sector_lower:
            return 11, 18
        elif 'nursery' in sector_lower:
            return 2, 5
        else:
            return 4, 11  # default to primary

    employee_per_clients = employee_per_clients or {"primary": 30, "secondary": 30}
    schools = []
    for school_name, row in school_df.iterrows():
        # Handle missing pupil numbers
        n_pupils_max = row["NOR"] if pd.notna(row["NOR"]) else 100  # default

        # Handle school type/sector
        school_type = row["sector"] if pd.notna(row["sector"]) else "primary"

        # Handle coordinates
        if pd.notna(row["latitude"]) and pd.notna(row["longitude"]):
            coordinates = np.array([row["latitude"], row["longitude"]], dtype=np.float64)
        else:
            continue  # skip schools without coordinates

        # Handle ages with defaults based on sector
        if pd.notna(row["age_min"]) and pd.notna(row["age_max"]):
            age_min = int(row["age_min"])
            age_max = min(int(row["age_max"]), 19)  # Cap to prevent contact matrix index errors
        else:
            age_min, age_max = get_default_ages(school_type)

        # Calculate gender ratios
        n_boys = row["n_boys"] if pd.notna(row["n_boys"]) else 0
        n_girls = row["n_girls"] if pd.notna(row["n_girls"]) else 0

        # Calculate ratios with defaults
        total_gender_pupils = n_boys + n_girls
        if total_gender_pupils > 0:
            boy_ratio = n_boys / total_gender_pupils
            girl_ratio = n_girls / total_gender_pupils
        else:
            # Default to 50/50 split when no gender data available
            boy_ratio = 0.5
            girl_ratio = 0.5

        # Use the area code specified in the CSV rather than closest geographic area
        try:
            area = areas.members_by_name[row["oa"]]
        except KeyError:
            # Fallback to closest area if the specified area isn't in our geography
            area = areas.get_closest_area(coordinates)
        school = cls.venue_class(
            coordinates=coordinates,
            n_pupils_max=int(n_pupils_max),
            age_min=age_min,
            age_max=age_max,
            sector=school_type,
            area=area,
            boy_ratio=boy_ratio,
            girl_ratio=girl_ratio
        )
        schools.append(school)
        area.schools.append(school)

    school_trees, agegroup_to_global_indices = Schools.init_trees(school_df, age_range)
    return Schools(
        schools,
        school_trees=school_trees,
        agegroup_to_global_indices=agegroup_to_global_indices,
    )

for_areas(areas, data_file_ew=default_data_filename_ew, data_file_sct=default_data_filename_sct, data_file_ni=default_data_filename_ni, config_file=default_config_filename) classmethod

Creates Schools for specified areas using a disease configuration.

Parameters:

Name	Type	Description	Default
areas	Areas		required
data_file_ew	str	(Default value = default_data_filename_ew)	default_data_filename_ew
data_file_sct	str	(Default value = default_data_filename_sct)	default_data_filename_sct
data_file_ni	str	(Default value = default_data_filename_ni)	default_data_filename_ni
config_file	str	(Default value = default_config_filename)	default_config_filename

Source code in june/groups/school.py

@classmethod
def for_areas(
    cls,
    areas: Areas,
    data_file_ew: str = default_data_filename_ew,
    data_file_sct: str = default_data_filename_sct,
    data_file_ni: str = default_data_filename_ni,
    config_file: str = default_config_filename,
) -> "Schools":
    """Creates Schools for specified areas using a disease configuration.

    Args:
        areas (Areas): 
        data_file_ew (str, optional): (Default value = default_data_filename_ew)
        data_file_sct (str, optional): (Default value = default_data_filename_sct)
        data_file_ni (str, optional): (Default value = default_data_filename_ni)
        config_file (str, optional): (Default value = default_config_filename)

    """
    return cls.from_file(areas, data_file_ew, data_file_sct, data_file_ni)

for_geography(geography, config_file=default_config_filename) classmethod

Creates Schools for the given geography with a disease configuration.

Parameters:

Name	Type	Description	Default
geography	Geography		required
config_file	str	(Default value = default_config_filename)	default_config_filename

Source code in june/groups/school.py

@classmethod
def for_geography(
    cls,
    geography: Geography,
    config_file: str = default_config_filename,
) -> "Schools":
    """Creates Schools for the given geography with a disease configuration.

    Args:
        geography (Geography): 
        config_file (str, optional): (Default value = default_config_filename)

    """
    schools = cls.for_areas(
        geography.areas,
        config_file=config_file
    )

    # Sample 5 schools from each area for visualization
    sampled_schools = []
    for area in geography.areas:
        if hasattr(area, 'schools') and area.schools:
            # Sample 5 schools or fewer if there are less than 5
            sample_schools = random.sample(area.schools, min(5, len(area.schools)))
            for school in sample_schools:
                sampled_schools.append({
                    "| School ID": school.id,
                    "| Area": area.name,
                    "| School Sector": school.sector,
                    "| Number of Pupils": school.n_pupils,
                    "| Coordinates": school.coordinates,
                    "| Max Pupils": school.n_pupils_max,
                    "| Min Age": school.age_min,
                    "| Max Age": school.age_max
                })

    # Convert the sample data to a DataFrame
    df_schools = pd.DataFrame(sampled_schools)
    print("\n===== Sample of Created Schools =====")
    print(df_schools)

    logger.info(f"Created {len(schools)} schools in this geography.")
    return schools

from_file(areas, data_file_ew=default_data_filename_ew, data_file_sct=default_data_filename_sct, data_file_ni=default_data_filename_ni) classmethod

Initialise Schools from England/Wales, Scotland, and Northern Ireland data frames.

Parameters:

Name	Type	Description	Default
areas	Areas		required
data_file_ew	str	(Default value = default_data_filename_ew)	default_data_filename_ew
data_file_sct	str	(Default value = default_data_filename_sct)	default_data_filename_sct
data_file_ni	str	(Default value = default_data_filename_ni)	default_data_filename_ni

Source code in june/groups/school.py

@classmethod
def from_file(
    cls,
    areas: Areas,
    data_file_ew: str = default_data_filename_ew,
    data_file_sct: str = default_data_filename_sct,
    data_file_ni: str = default_data_filename_ni,
) -> "Schools":
    """Initialise Schools from England/Wales, Scotland, and Northern Ireland data frames.

    Args:
        areas (Areas): 
        data_file_ew (str, optional): (Default value = default_data_filename_ew)
        data_file_sct (str, optional): (Default value = default_data_filename_sct)
        data_file_ni (str, optional): (Default value = default_data_filename_ni)

    """
    # Load England/Wales schools
    school_df_ew = pd.read_csv(data_file_ew)
    school_df_ew = school_df_ew.rename(columns={
        'OA': 'oa',
        'NumberOfPupils': 'NOR',
        'PhaseOfEducation': 'sector',
        'StatutoryLowAge': 'age_min',
        'StatutoryHighAge': 'age_max',
        'Latitude': 'latitude',
        'Longitude': 'longitude',
        'NumberOfBoys': 'n_boys',
        'NumberOfGirls': 'n_girls'
    })

    # Load Scotland schools
    school_df_sct = pd.read_csv(data_file_sct)
    school_df_sct = school_df_sct.rename(columns={
        'OA': 'oa',
        'NumberOfPupils': 'NOR',
        'PhaseOfEducation': 'sector',
        'Latitude': 'latitude',
        'Longitude': 'longitude',
        'NumberOfBoys': 'n_boys',
        'NumberOfGirls': 'n_girls'
    })
    # Scotland doesn't have explicit age columns, we'll handle this in build_schools_for_areas
    school_df_sct['age_min'] = np.nan
    school_df_sct['age_max'] = np.nan

    # Load Northern Ireland schools
    school_df_ni = pd.read_csv(data_file_ni)
    school_df_ni = school_df_ni.rename(columns={
        'OA': 'oa',
        'NumberOfPupils': 'NOR',
        'PhaseOfEducation': 'sector',
        'Latitude': 'latitude',
        'Longitude': 'longitude',
        'NumberOfBoys': 'n_boys',
        'NumberOfGirls': 'n_girls',
        'StatutoryLowAge': 'age_min',
        'StatutoryHighAge': 'age_max'
    })

    # Combine all dataframes
    school_df = pd.concat([school_df_ew, school_df_sct, school_df_ni], ignore_index=True)

    # Apply age defaults to the dataframe BEFORE building trees
    def get_default_ages(sector):
        """

        Args:
            sector: 

        """
        if pd.isna(sector):
            return 4, 11
        sector_lower = str(sector).lower()
        if 'primary' in sector_lower:
            return 4, 11
        elif 'secondary' in sector_lower:
            return 11, 18
        elif 'nursery' in sector_lower:
            return 2, 5
        else:
            return 4, 11

    # Apply defaults where missing
    for idx, row in school_df.iterrows():
        if pd.isna(row['age_min']) or pd.isna(row['age_max']):
            default_min, default_max = get_default_ages(row['sector'])
            school_df.at[idx, 'age_min'] = default_min
            school_df.at[idx, 'age_max'] = default_max

    # Ensure age columns are numeric
    school_df['age_min'] = pd.to_numeric(school_df['age_min'], errors='coerce')
    school_df['age_max'] = pd.to_numeric(school_df['age_max'], errors='coerce')

    # Remove special schools from dataframe
    school_df = school_df[~school_df['sector'].str.lower().str.contains('special', na=False)]

    # Filter by areas
    area_names = [area.name for area in areas]
    if area_names is not None:
        school_df = school_df[school_df["oa"].isin(area_names)]
    school_df.reset_index(drop=True, inplace=True)
    logger.info(f"There are {len(school_df)} schools in this geography.")
    return cls.build_schools_for_areas(areas, school_df)

get_closest_schools(age, coordinates, k)

Get the k-th closest school to a given coordinate, that accepts pupils aged age

Parameters:

Name	Type	Description	Default
age	int	age of the pupil	required
coordinates	Tuple[float, float]	latitude and longitude	required
k	int	k-th neighbour	required

Returns:

Type	Description
int	ID of the k-th closest school, within school trees for:
int	a given age group:

Source code in june/groups/school.py

def get_closest_schools(
    self, age: int, coordinates: Tuple[float, float], k: int
) -> int:
    """Get the k-th closest school to a given coordinate, that accepts pupils
    aged age

    Args:
        age (int): age of the pupil
        coordinates (Tuple[float, float]): latitude and longitude
        k (int): k-th neighbour

    Returns:
        ID of the k-th closest school, within school trees for: 
        a given age group: 

    """
    school_tree = self.school_trees[age]
    coordinates_rad = np.deg2rad(coordinates).reshape(1, -1)
    k = min(k, school_tree.data.shape[0])
    distances, neighbours = school_tree.query(
        coordinates_rad, k=k, sort_results=True
    )
    return neighbours[0]

init_trees(school_df, age_range) staticmethod

Create trees to easily find the closest school that accepts a pupil given their age

Parameters:

Name	Type	Description	Default
school_df	DataFrame	dataframe with school characteristics data	required
age_range	Tuple[int, int]		required

Source code in june/groups/school.py

@staticmethod
def init_trees(school_df: pd.DataFrame, age_range: Tuple[int, int]) -> "Schools":
    """Create trees to easily find the closest school that
    accepts a pupil given their age

    Args:
        school_df (pd.DataFrame): dataframe with school characteristics data
        age_range (Tuple[int, int]): 

    """
    school_trees = {}
    school_agegroup_to_global_indices = {
        k: [] for k in range(int(age_range[0]), int(age_range[1]) + 1)
    }
    # have a tree per age
    for age in range(int(age_range[0]), int(age_range[1]) + 1):
        _school_df_agegroup = school_df[
            (school_df["age_min"] <= age) & (school_df["age_max"] >= age)
        ]
        schools_coords = _school_df_agegroup[["latitude", "longitude"]].values
        if not schools_coords.size:
            logger.info(f"No school for the age {age} in this world.")
            continue
        school_trees[age] = Schools._create_school_tree(schools_coords)
        school_agegroup_to_global_indices[age] = _school_df_agegroup.index.values
    return school_trees, school_agegroup_to_global_indices