Skip to content

Hospital

AbstractHospital

Hospital functionality common for all hospitals (internal to the domain and external).

Source code in june/groups/hospital.py
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
class AbstractHospital:
    """Hospital functionality common for all hospitals (internal to the domain and external)."""

    def __init__(self):
        """
        Initialise the hospital with a disease configuration.

        Parameters
        ----------
        disease_config : DiseaseConfig
            Configuration object for the disease.
        """
        self.ward_ids = set()  # IDs of patients in the ward
        self.icu_ids = set()   # IDs of patients in the ICU

    def add_to_ward(self, person):
        """

        Args:
            person: 

        """
        self.ward_ids.add(person.id)
        person.subgroups.medical_facility = self.ward

    def remove_from_ward(self, person):
        """

        Args:
            person: 

        """
        self.ward_ids.remove(person.id)
        person.subgroups.medical_facility = None

    def add_to_icu(self, person):
        """

        Args:
            person: 

        """
        self.icu_ids.add(person.id)
        person.subgroups.medical_facility = self.icu

    def remove_from_icu(self, person):
        """

        Args:
            person: 

        """
        self.icu_ids.remove(person.id)
        person.subgroups.medical_facility = None

    def allocate_patient(self, person):
        """Allocate a patient inside the hospital, in the ward, in the ICU, or transfer.

        To correctly log if the person has been just admitted, transferred, or released,
        we return a few flags:
        - "ward_admitted": This person has been admitted to the ward.
        - "icu_admitted": This person has been directly admitted to ICU.
        - "ward_transferred": This person has been transferred to ward (from ICU).
        - "icu_transferred": This person has been transferred to ICU (from ward).
        - "no_change": No change respect to last time step.

        Args:
            person: 

        """
        disease_config = GlobalContext.get_disease_config()
        # Get the person's current symptom tag
        person_tag = person.infection.tag

        # Ensure person_tag is resolved to a value (integer) if it's an object
        person_tag_value = person_tag.value if hasattr(person_tag, "value") else person_tag

        # Check if the person is entering a hospital
        if (
            person.medical_facility is None
            or person.medical_facility.spec != "hospital"
        ):
            if person_tag_value == disease_config.symptom_manager.get_tag_value("hospitalised"):
                self.add_to_ward(person)
                return "ward_admitted"
            elif person_tag_value == disease_config.symptom_manager.get_tag_value("intensive_care"):
                self.add_to_icu(person)
                return "icu_admitted"
            else:
                raise HospitalError(
                    f"Person with symptoms {person_tag} (value: {person_tag_value}) "
                )
        else:
            # The person is already in a hospital
            if person_tag_value == disease_config.symptom_manager.get_tag_value("hospitalised"):
                if person.id in self.ward_ids:
                    return "no_change"
                else:
                    self.remove_from_icu(person)
                    self.add_to_ward(person)
                    return "ward_transferred"
            elif person_tag_value == disease_config.symptom_manager.get_tag_value("intensive_care"):
                if person.id in self.icu_ids:
                    return "no_change"
                else:
                    self.remove_from_ward(person)
                    self.add_to_icu(person)
                    return "icu_transferred"

    def release_patient(self, person):
        """Releases patient from hospital.

        Args:
            person: 

        """
        if person.id in self.ward_ids:
            self.remove_from_ward(person)
        elif person.id in self.icu_ids:
            self.remove_from_icu(person)
        else:
            raise HospitalError("Trying to release patient not located in icu or ward.")

__init__()

Initialise the hospital with a disease configuration.

Parameters

disease_config : DiseaseConfig Configuration object for the disease.

Source code in june/groups/hospital.py
34
35
36
37
38
39
40
41
42
43
44
def __init__(self):
    """
    Initialise the hospital with a disease configuration.

    Parameters
    ----------
    disease_config : DiseaseConfig
        Configuration object for the disease.
    """
    self.ward_ids = set()  # IDs of patients in the ward
    self.icu_ids = set()   # IDs of patients in the ICU

add_to_icu(person)

Parameters:

Name Type Description Default
person
required
Source code in june/groups/hospital.py
66
67
68
69
70
71
72
73
74
def add_to_icu(self, person):
    """

    Args:
        person: 

    """
    self.icu_ids.add(person.id)
    person.subgroups.medical_facility = self.icu

add_to_ward(person)

Parameters:

Name Type Description Default
person
required
Source code in june/groups/hospital.py
46
47
48
49
50
51
52
53
54
def add_to_ward(self, person):
    """

    Args:
        person: 

    """
    self.ward_ids.add(person.id)
    person.subgroups.medical_facility = self.ward

allocate_patient(person)

Allocate a patient inside the hospital, in the ward, in the ICU, or transfer.

To correctly log if the person has been just admitted, transferred, or released, we return a few flags: - "ward_admitted": This person has been admitted to the ward. - "icu_admitted": This person has been directly admitted to ICU. - "ward_transferred": This person has been transferred to ward (from ICU). - "icu_transferred": This person has been transferred to ICU (from ward). - "no_change": No change respect to last time step.

Parameters:

Name Type Description Default
person
required
Source code in june/groups/hospital.py
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
def allocate_patient(self, person):
    """Allocate a patient inside the hospital, in the ward, in the ICU, or transfer.

    To correctly log if the person has been just admitted, transferred, or released,
    we return a few flags:
    - "ward_admitted": This person has been admitted to the ward.
    - "icu_admitted": This person has been directly admitted to ICU.
    - "ward_transferred": This person has been transferred to ward (from ICU).
    - "icu_transferred": This person has been transferred to ICU (from ward).
    - "no_change": No change respect to last time step.

    Args:
        person: 

    """
    disease_config = GlobalContext.get_disease_config()
    # Get the person's current symptom tag
    person_tag = person.infection.tag

    # Ensure person_tag is resolved to a value (integer) if it's an object
    person_tag_value = person_tag.value if hasattr(person_tag, "value") else person_tag

    # Check if the person is entering a hospital
    if (
        person.medical_facility is None
        or person.medical_facility.spec != "hospital"
    ):
        if person_tag_value == disease_config.symptom_manager.get_tag_value("hospitalised"):
            self.add_to_ward(person)
            return "ward_admitted"
        elif person_tag_value == disease_config.symptom_manager.get_tag_value("intensive_care"):
            self.add_to_icu(person)
            return "icu_admitted"
        else:
            raise HospitalError(
                f"Person with symptoms {person_tag} (value: {person_tag_value}) "
            )
    else:
        # The person is already in a hospital
        if person_tag_value == disease_config.symptom_manager.get_tag_value("hospitalised"):
            if person.id in self.ward_ids:
                return "no_change"
            else:
                self.remove_from_icu(person)
                self.add_to_ward(person)
                return "ward_transferred"
        elif person_tag_value == disease_config.symptom_manager.get_tag_value("intensive_care"):
            if person.id in self.icu_ids:
                return "no_change"
            else:
                self.remove_from_ward(person)
                self.add_to_icu(person)
                return "icu_transferred"

release_patient(person)

Releases patient from hospital.

Parameters:

Name Type Description Default
person
required
Source code in june/groups/hospital.py
140
141
142
143
144
145
146
147
148
149
150
151
152
def release_patient(self, person):
    """Releases patient from hospital.

    Args:
        person: 

    """
    if person.id in self.ward_ids:
        self.remove_from_ward(person)
    elif person.id in self.icu_ids:
        self.remove_from_icu(person)
    else:
        raise HospitalError("Trying to release patient not located in icu or ward.")

remove_from_icu(person)

Parameters:

Name Type Description Default
person
required
Source code in june/groups/hospital.py
76
77
78
79
80
81
82
83
84
def remove_from_icu(self, person):
    """

    Args:
        person: 

    """
    self.icu_ids.remove(person.id)
    person.subgroups.medical_facility = None

remove_from_ward(person)

Parameters:

Name Type Description Default
person
required
Source code in june/groups/hospital.py
56
57
58
59
60
61
62
63
64
def remove_from_ward(self, person):
    """

    Args:
        person: 

    """
    self.ward_ids.remove(person.id)
    person.subgroups.medical_facility = None

ExternalHospital

Bases: ExternalGroup, AbstractHospital, MedicalFacility

Source code in june/groups/hospital.py
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
class ExternalHospital(ExternalGroup, AbstractHospital, MedicalFacility):
    """ """
    external = True
    __slots__ = "spec", "id", "domain_id", "region_name", "ward_ids", "icu_ids"

    class SubgroupType(IntEnum):
        """ """
        workers = 0
        patients = 1
        icu_patients = 2

    def __init__(self, id, spec, domain_id, region_name):
        ExternalGroup.__init__(self, id=id, spec=spec, domain_id=domain_id)
        AbstractHospital.__init__(self)
        self.region_name = region_name

        self.ward = ExternalSubgroup(
            group=self, subgroup_type=self.SubgroupType.patients
        )
        self.icu = ExternalSubgroup(
            group=self, subgroup_type=self.SubgroupType.icu_patients
        )

SubgroupType

Bases: IntEnum

Source code in june/groups/hospital.py
489
490
491
492
493
class SubgroupType(IntEnum):
    """ """
    workers = 0
    patients = 1
    icu_patients = 2

Hospital

Bases: Group, AbstractHospital, MedicalFacility

The Hospital class represents a hospital and contains information about its patients and workers - the latter being the usual "people".

We currently use three subgroups: 0 - workers (i.e. nurses, doctors, etc.), 1 - patients 2 - ICU patients

Source code in june/groups/hospital.py
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
class Hospital(Group, AbstractHospital, MedicalFacility):
    """The Hospital class represents a hospital and contains information about
    its patients and workers - the latter being the usual "people".

    We currently use three subgroups:
    0 - workers (i.e. nurses, doctors, etc.),
    1 - patients
    2 - ICU patients

    """

    __slots__ = "id", "n_beds", "n_icu_beds", "coordinates", "area", "trust_code"

    def __init__(
        self,
        n_beds: int,
        n_icu_beds: int,
        area: str = None,
        coordinates: Optional[Tuple[float, float]] = None,
        trust_code: str = None,
    ):
        """
        Create a Hospital given its description.

        Parameters
        ----------
        n_beds: int
            Total number of regular beds in the hospital.
        n_icu_beds: int
            Total number of ICU beds in the hospital.
        disease_config: DiseaseConfig
            Configuration object for the disease.
        area: str, optional
            Name of the super area the hospital belongs to.
        coordinates: tuple of float, optional
            Latitude and longitude.
        trust_code: str, optional
            Trust code associated with the hospital.
        """
        # Initialise the base Group class with disease_config
        Group.__init__(self)

        # Initialise AbstractHospital with disease_config
        AbstractHospital.__init__(self)

        # Assign attributes specific to Hospital
        self.area = area
        self.coordinates = coordinates
        self.n_beds = n_beds
        self.n_icu_beds = n_icu_beds
        self.trust_code = trust_code

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

    @property
    def region(self):
        """ """
        return self.super_area.region

    @property
    def region_name(self):
        """ """
        return self.region.name

    @property
    def full(self):
        """Check whether all regular beds are being used"""
        return self[self.SubgroupType.patients].size >= self.n_beds

    @property
    def full_ICU(self):
        """Check whether all ICU beds are being used"""
        return self[self.SubgroupType.icu_patients].size >= self.n_icu_beds

    def add(self, person, subgroup_type):
        """

        Args:
            person: 
            subgroup_type: 

        """
        if subgroup_type in [
            self.SubgroupType.patients,
            self.SubgroupType.icu_patients,
        ]:
            super().add(
                person, activity="medical_facility", subgroup_type=subgroup_type
            )
        else:
            super().add(
                person,
                activity="primary_activity",
                subgroup_type=self.SubgroupType.workers,
            )

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

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

full property

Check whether all regular beds are being used

full_ICU property

Check whether all ICU beds are being used

icu property

region property

region_name property

super_area property

ward property

__init__(n_beds, n_icu_beds, area=None, coordinates=None, trust_code=None)

Create a Hospital given its description.

Parameters

n_beds: int Total number of regular beds in the hospital. n_icu_beds: int Total number of ICU beds in the hospital. disease_config: DiseaseConfig Configuration object for the disease. area: str, optional Name of the super area the hospital belongs to. coordinates: tuple of float, optional Latitude and longitude. trust_code: str, optional Trust code associated with the hospital.

Source code in june/groups/hospital.py
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
def __init__(
    self,
    n_beds: int,
    n_icu_beds: int,
    area: str = None,
    coordinates: Optional[Tuple[float, float]] = None,
    trust_code: str = None,
):
    """
    Create a Hospital given its description.

    Parameters
    ----------
    n_beds: int
        Total number of regular beds in the hospital.
    n_icu_beds: int
        Total number of ICU beds in the hospital.
    disease_config: DiseaseConfig
        Configuration object for the disease.
    area: str, optional
        Name of the super area the hospital belongs to.
    coordinates: tuple of float, optional
        Latitude and longitude.
    trust_code: str, optional
        Trust code associated with the hospital.
    """
    # Initialise the base Group class with disease_config
    Group.__init__(self)

    # Initialise AbstractHospital with disease_config
    AbstractHospital.__init__(self)

    # Assign attributes specific to Hospital
    self.area = area
    self.coordinates = coordinates
    self.n_beds = n_beds
    self.n_icu_beds = n_icu_beds
    self.trust_code = trust_code

add(person, subgroup_type)

Parameters:

Name Type Description Default
person
required
subgroup_type
required
Source code in june/groups/hospital.py
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
def add(self, person, subgroup_type):
    """

    Args:
        person: 
        subgroup_type: 

    """
    if subgroup_type in [
        self.SubgroupType.patients,
        self.SubgroupType.icu_patients,
    ]:
        super().add(
            person, activity="medical_facility", subgroup_type=subgroup_type
        )
    else:
        super().add(
            person,
            activity="primary_activity",
            subgroup_type=self.SubgroupType.workers,
        )

Hospitals

Bases: Supergroup, MedicalFacilities

Source code in june/groups/hospital.py
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
class Hospitals(Supergroup, MedicalFacilities):
    """ """
    venue_class = Hospital

    def __init__(
        self, hospitals: List["Hospital"], neighbour_hospitals: int = 5, ball_tree=True
    ):
        """
        Create a group of hospitals, and provide functionality to locate patients
        to a nearby hospital. It will check in order the first ```neighbour_hospitals```,
        when one has space available the patient is allocated to it. If none of the closest
        ones has beds available it will pick one of them at random and that hospital will
        overflow

        Args:
          hospitals (List["Hospital"]):
            list of hospitals to aggrupate
          neighbour_hospitals (int, optional):
            number of closest hospitals to look for. Default is 5. 
          ball_tree (bool, optional): If true initialise a tree. Default is True. 
        """
        super().__init__(members=hospitals)
        self.neighbour_hospitals = neighbour_hospitals
        if ball_tree and self.members:
            coordinates = np.array([hospital.coordinates for hospital in hospitals])
            self.init_trees(coordinates)

    @classmethod
    def from_file(
        cls,
        filename: str = default_data_filename,
        config_filename: str = default_config_filename,
    ) -> "Hospitals":
        """Initialise Hospitals from path to data frame, and path to config file.

        Args:
            filename (str, optional): path to hospital dataframe (Default value = default_data_filename)
            config_filename (str, optional): path to hospital config dictionary (Default value = default_config_filename)

        """

        hospital_df = pd.read_csv(filename)
        # Filter to include both Board and Trust entries
        hospital_df = hospital_df[hospital_df['type'].isin(['Board', 'Trust'])]
        with open(config_filename) as f:
            config = yaml.load(f, Loader=yaml.FullLoader)
        neighbour_hospitals = config["neighbour_hospitals"]
        logger.info(f"There are {len(hospital_df)} hospitals in the world.")
        hospitals = cls.init_hospitals(cls, hospital_df)
        return Hospitals(hospitals, neighbour_hospitals)

    @classmethod
    def for_geography(
        cls,
        geography,
        filename: str = default_data_filename,
        config_filename: str = default_config_filename,
    ):
        """Create hospitals for the given geography based on input files and disease configuration.

        Args:
            geography (Geography): The geography object containing areas.
            filename (str, optional): Path to the file containing hospital data. (Default value = default_data_filename)
            config_filename (str, optional): Path to the configuration file. (Default value = default_config_filename)

        Returns:
            cls: An instance of the Hospitals class.

        """
        with open(config_filename) as f:
            config = yaml.load(f, Loader=yaml.FullLoader)
        neighbour_hospitals = config["neighbour_hospitals"]
        hospital_df = pd.read_csv(filename, index_col="area")
        # Filter to include both Board and Trust entries
        hospital_df = hospital_df[hospital_df['type'].isin(['Board', 'Trust'])]
        area_names = [area.name for area in geography.areas]
        hospital_df = hospital_df.loc[hospital_df.index.isin(area_names)]
        logger.info(f"There are {len(hospital_df)} hospitals in this geography.")
        total_hospitals = len(hospital_df)
        hospitals = []

        for area in geography.areas:
            if area.name in hospital_df.index:
                hospitals_in_area = hospital_df.loc[area.name]
                if isinstance(hospitals_in_area, pd.Series):
                    hospital = cls.create_hospital_from_df_row(
                        area, hospitals_in_area
                    )
                    hospitals.append(hospital)
                else:
                    for _, row in hospitals_in_area.iterrows():
                        hospital = cls.create_hospital_from_df_row(
                            area, row
                        )
                        hospitals.append(hospital)
                if len(hospitals) == total_hospitals:
                    break

        # Debugging: View hospital data as a DataFrame
        hospitals_data = [{
            "| Hospital ID": hospital.id,
            "| Area": hospital.area.name,
            "| Coordinates": hospital.coordinates,
            "| Total Beds": hospital.n_beds,
            "| ICU Beds": hospital.n_icu_beds,
            "| Trust Code": hospital.trust_code
        } for hospital in hospitals]

        hospitals_df = pd.DataFrame(hospitals_data)
        print("\n===== Sample of Created Hospitals =====")
        print(hospitals_df.head())

        return cls(
            hospitals=hospitals, neighbour_hospitals=neighbour_hospitals, ball_tree=True
        )

    @classmethod
    def create_hospital_from_df_row(cls, area, row):
        """Create a hospital from a row in the hospital dataframe.

        Args:
            area (Area): The area object associated with the hospital.
            row (pd.Series): The row from the hospital dataframe.

        Returns:
            Hospital: A newly created Hospital instance.

        """
        coordinates = row[["latitude", "longitude"]].values.astype(np.float64)
        n_beds = int(row["n_beds"])
        n_icu_beds = int(row["n_ICU_beds"])
        trust_code = row["code"]

        # Pass DiseaseConfig to the hospital initialization
        hospital = cls.venue_class(
            area=area,
            coordinates=coordinates,
            n_beds=n_beds,
            n_icu_beds=n_icu_beds,
            trust_code=trust_code
        )
        return hospital

    def init_hospitals(self, hospital_df: pd.DataFrame) -> List["Hospital"]:
        """Create Hospital objects with the right characteristics,
        as given by dataframe.

        Args:
            hospital_df (pd.DataFrame): dataframe with hospital characteristics data

        Returns:
            List["Hospital"]: list of hospital objects

        """
        hospitals = []
        for index, row in hospital_df.iterrows():
            n_beds = int(row["n_beds"])
            n_icu_beds = int(row["n_ICU_beds"])
            trust_code = row["code"]
            coordinates = row[["latitude", "longitude"]].values.astype(np.float64)
            hospital = Hospital(
                coordinates=coordinates,
                n_beds=n_beds,
                n_icu_beds=n_icu_beds,
                trust_code=trust_code,
            )
            hospitals.append(hospital)
        return hospitals

    def init_trees(self, hospital_coordinates: np.array) -> BallTree:
        """Reads hospital location and sizes, it initialises a KD tree on a sphere,
        to query the closest hospital to a given location.

        Args:
            hospital_coordinates (np.array): 

        Returns:
            BallTree: Tree to query nearby schools

        """
        self.hospital_trees = BallTree(
            np.deg2rad(hospital_coordinates), metric="haversine"
        )

    def get_closest_hospitals_idx(
        self, coordinates: Tuple[float, float], k: int
    ) -> Tuple[float, float]:
        """OUTDATED DOCUMENTATION?
        Get the k-th closest hospital to a given coordinate

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

        """
        k = min(k, len(list(self.hospital_trees.data)))
        distances, neighbours = self.hospital_trees.query(
            np.deg2rad(coordinates.reshape(1, -1)), k=k, sort_results=True
        )
        return neighbours[0]

    def get_closest_hospitals(
        self, coordinates: Tuple[float, float], k: int
    ) -> Tuple[float, float]:
        """OUTDATED DOCUMENTATION?
        Get the k-th closest hospital to a given coordinate

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

        """
        k = min(k, len(list(self.hospital_trees.data)))
        distances, neighbours = self.hospital_trees.query(
            np.deg2rad(coordinates.reshape(1, -1)), k=k, sort_results=True
        )
        return [self.members[index] for index in neighbours[0]]

__init__(hospitals, neighbour_hospitals=5, ball_tree=True)

Create a group of hospitals, and provide functionality to locate patients to a nearby hospital. It will check in order the first neighbour_hospitals, when one has space available the patient is allocated to it. If none of the closest ones has beds available it will pick one of them at random and that hospital will overflow

Parameters:

Name Type Description Default
hospitals List[Hospital]

list of hospitals to aggrupate

required
neighbour_hospitals int

number of closest hospitals to look for. Default is 5.

5
ball_tree bool

If true initialise a tree. Default is True.

True
Source code in june/groups/hospital.py
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
def __init__(
    self, hospitals: List["Hospital"], neighbour_hospitals: int = 5, ball_tree=True
):
    """
    Create a group of hospitals, and provide functionality to locate patients
    to a nearby hospital. It will check in order the first ```neighbour_hospitals```,
    when one has space available the patient is allocated to it. If none of the closest
    ones has beds available it will pick one of them at random and that hospital will
    overflow

    Args:
      hospitals (List["Hospital"]):
        list of hospitals to aggrupate
      neighbour_hospitals (int, optional):
        number of closest hospitals to look for. Default is 5. 
      ball_tree (bool, optional): If true initialise a tree. Default is True. 
    """
    super().__init__(members=hospitals)
    self.neighbour_hospitals = neighbour_hospitals
    if ball_tree and self.members:
        coordinates = np.array([hospital.coordinates for hospital in hospitals])
        self.init_trees(coordinates)

create_hospital_from_df_row(area, row) classmethod

Create a hospital from a row in the hospital dataframe.

Parameters:

Name Type Description Default
area Area

The area object associated with the hospital.

required
row Series

The row from the hospital dataframe.

required

Returns:

Name Type Description
Hospital

A newly created Hospital instance.

Source code in june/groups/hospital.py
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
@classmethod
def create_hospital_from_df_row(cls, area, row):
    """Create a hospital from a row in the hospital dataframe.

    Args:
        area (Area): The area object associated with the hospital.
        row (pd.Series): The row from the hospital dataframe.

    Returns:
        Hospital: A newly created Hospital instance.

    """
    coordinates = row[["latitude", "longitude"]].values.astype(np.float64)
    n_beds = int(row["n_beds"])
    n_icu_beds = int(row["n_ICU_beds"])
    trust_code = row["code"]

    # Pass DiseaseConfig to the hospital initialization
    hospital = cls.venue_class(
        area=area,
        coordinates=coordinates,
        n_beds=n_beds,
        n_icu_beds=n_icu_beds,
        trust_code=trust_code
    )
    return hospital

for_geography(geography, filename=default_data_filename, config_filename=default_config_filename) classmethod

Create hospitals for the given geography based on input files and disease configuration.

Parameters:

Name Type Description Default
geography Geography

The geography object containing areas.

required
filename str

Path to the file containing hospital data. (Default value = default_data_filename)

default_data_filename
config_filename str

Path to the configuration file. (Default value = default_config_filename)

default_config_filename

Returns:

Name Type Description
cls

An instance of the Hospitals class.

Source code in june/groups/hospital.py
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
@classmethod
def for_geography(
    cls,
    geography,
    filename: str = default_data_filename,
    config_filename: str = default_config_filename,
):
    """Create hospitals for the given geography based on input files and disease configuration.

    Args:
        geography (Geography): The geography object containing areas.
        filename (str, optional): Path to the file containing hospital data. (Default value = default_data_filename)
        config_filename (str, optional): Path to the configuration file. (Default value = default_config_filename)

    Returns:
        cls: An instance of the Hospitals class.

    """
    with open(config_filename) as f:
        config = yaml.load(f, Loader=yaml.FullLoader)
    neighbour_hospitals = config["neighbour_hospitals"]
    hospital_df = pd.read_csv(filename, index_col="area")
    # Filter to include both Board and Trust entries
    hospital_df = hospital_df[hospital_df['type'].isin(['Board', 'Trust'])]
    area_names = [area.name for area in geography.areas]
    hospital_df = hospital_df.loc[hospital_df.index.isin(area_names)]
    logger.info(f"There are {len(hospital_df)} hospitals in this geography.")
    total_hospitals = len(hospital_df)
    hospitals = []

    for area in geography.areas:
        if area.name in hospital_df.index:
            hospitals_in_area = hospital_df.loc[area.name]
            if isinstance(hospitals_in_area, pd.Series):
                hospital = cls.create_hospital_from_df_row(
                    area, hospitals_in_area
                )
                hospitals.append(hospital)
            else:
                for _, row in hospitals_in_area.iterrows():
                    hospital = cls.create_hospital_from_df_row(
                        area, row
                    )
                    hospitals.append(hospital)
            if len(hospitals) == total_hospitals:
                break

    # Debugging: View hospital data as a DataFrame
    hospitals_data = [{
        "| Hospital ID": hospital.id,
        "| Area": hospital.area.name,
        "| Coordinates": hospital.coordinates,
        "| Total Beds": hospital.n_beds,
        "| ICU Beds": hospital.n_icu_beds,
        "| Trust Code": hospital.trust_code
    } for hospital in hospitals]

    hospitals_df = pd.DataFrame(hospitals_data)
    print("\n===== Sample of Created Hospitals =====")
    print(hospitals_df.head())

    return cls(
        hospitals=hospitals, neighbour_hospitals=neighbour_hospitals, ball_tree=True
    )

from_file(filename=default_data_filename, config_filename=default_config_filename) classmethod

Initialise Hospitals from path to data frame, and path to config file.

Parameters:

Name Type Description Default
filename str

path to hospital dataframe (Default value = default_data_filename)

default_data_filename
config_filename str

path to hospital config dictionary (Default value = default_config_filename)

default_config_filename
Source code in june/groups/hospital.py
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
@classmethod
def from_file(
    cls,
    filename: str = default_data_filename,
    config_filename: str = default_config_filename,
) -> "Hospitals":
    """Initialise Hospitals from path to data frame, and path to config file.

    Args:
        filename (str, optional): path to hospital dataframe (Default value = default_data_filename)
        config_filename (str, optional): path to hospital config dictionary (Default value = default_config_filename)

    """

    hospital_df = pd.read_csv(filename)
    # Filter to include both Board and Trust entries
    hospital_df = hospital_df[hospital_df['type'].isin(['Board', 'Trust'])]
    with open(config_filename) as f:
        config = yaml.load(f, Loader=yaml.FullLoader)
    neighbour_hospitals = config["neighbour_hospitals"]
    logger.info(f"There are {len(hospital_df)} hospitals in the world.")
    hospitals = cls.init_hospitals(cls, hospital_df)
    return Hospitals(hospitals, neighbour_hospitals)

get_closest_hospitals(coordinates, k)

OUTDATED DOCUMENTATION? Get the k-th closest hospital to a given coordinate

Parameters:

Name Type Description Default
coordinates Tuple[float, float]

latitude and longitude

required
k int

k-th neighbour

required
Source code in june/groups/hospital.py
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
def get_closest_hospitals(
    self, coordinates: Tuple[float, float], k: int
) -> Tuple[float, float]:
    """OUTDATED DOCUMENTATION?
    Get the k-th closest hospital to a given coordinate

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

    """
    k = min(k, len(list(self.hospital_trees.data)))
    distances, neighbours = self.hospital_trees.query(
        np.deg2rad(coordinates.reshape(1, -1)), k=k, sort_results=True
    )
    return [self.members[index] for index in neighbours[0]]

get_closest_hospitals_idx(coordinates, k)

OUTDATED DOCUMENTATION? Get the k-th closest hospital to a given coordinate

Parameters:

Name Type Description Default
coordinates Tuple[float, float]

latitude and longitude

required
k int

k-th neighbour

required
Source code in june/groups/hospital.py
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
def get_closest_hospitals_idx(
    self, coordinates: Tuple[float, float], k: int
) -> Tuple[float, float]:
    """OUTDATED DOCUMENTATION?
    Get the k-th closest hospital to a given coordinate

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

    """
    k = min(k, len(list(self.hospital_trees.data)))
    distances, neighbours = self.hospital_trees.query(
        np.deg2rad(coordinates.reshape(1, -1)), k=k, sort_results=True
    )
    return neighbours[0]

init_hospitals(hospital_df)

Create Hospital objects with the right characteristics, as given by dataframe.

Parameters:

Name Type Description Default
hospital_df DataFrame

dataframe with hospital characteristics data

required

Returns:

Type Description
List[Hospital]

List["Hospital"]: list of hospital objects

Source code in june/groups/hospital.py
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
def init_hospitals(self, hospital_df: pd.DataFrame) -> List["Hospital"]:
    """Create Hospital objects with the right characteristics,
    as given by dataframe.

    Args:
        hospital_df (pd.DataFrame): dataframe with hospital characteristics data

    Returns:
        List["Hospital"]: list of hospital objects

    """
    hospitals = []
    for index, row in hospital_df.iterrows():
        n_beds = int(row["n_beds"])
        n_icu_beds = int(row["n_ICU_beds"])
        trust_code = row["code"]
        coordinates = row[["latitude", "longitude"]].values.astype(np.float64)
        hospital = Hospital(
            coordinates=coordinates,
            n_beds=n_beds,
            n_icu_beds=n_icu_beds,
            trust_code=trust_code,
        )
        hospitals.append(hospital)
    return hospitals

init_trees(hospital_coordinates)

Reads hospital location and sizes, it initialises a KD tree on a sphere, to query the closest hospital to a given location.

Parameters:

Name Type Description Default
hospital_coordinates array
required

Returns:

Name Type Description
BallTree BallTree

Tree to query nearby schools

Source code in june/groups/hospital.py
434
435
436
437
438
439
440
441
442
443
444
445
446
447
def init_trees(self, hospital_coordinates: np.array) -> BallTree:
    """Reads hospital location and sizes, it initialises a KD tree on a sphere,
    to query the closest hospital to a given location.

    Args:
        hospital_coordinates (np.array): 

    Returns:
        BallTree: Tree to query nearby schools

    """
    self.hospital_trees = BallTree(
        np.deg2rad(hospital_coordinates), metric="haversine"
    )

MedicalFacilities

Source code in june/groups/hospital.py
26
27
28
class MedicalFacilities:
    """ """
    pass

MedicalFacility

Source code in june/groups/hospital.py
21
22
23
class MedicalFacility:
    """ """
    pass