diff --git a/resources/healthsystem/ResourceFile_HealthSystem_parameters.csv b/resources/healthsystem/ResourceFile_HealthSystem_parameters.csv index 885fbf27de..6ca37170a7 100644 --- a/resources/healthsystem/ResourceFile_HealthSystem_parameters.csv +++ b/resources/healthsystem/ResourceFile_HealthSystem_parameters.csv @@ -1,3 +1,3 @@ version https://git-lfs.github.com/spec/v1 -oid sha256:e7e56da62d26488f05ec1a5a10f3790ec9416c029b040f1fed629f314a14816b -size 622 +oid sha256:5df67165565fc88987f848e43363616e2ea4135de7c74d131e785ddc0178f123 +size 706 diff --git a/resources/healthsystem/infrastructure_and_equipment/ResourceFile_EquipmentCatalogue.csv b/resources/healthsystem/infrastructure_and_equipment/ResourceFile_EquipmentCatalogue.csv new file mode 100644 index 0000000000..33ba052c64 --- /dev/null +++ b/resources/healthsystem/infrastructure_and_equipment/ResourceFile_EquipmentCatalogue.csv @@ -0,0 +1,3 @@ +version https://git-lfs.github.com/spec/v1 +oid sha256:3e151e16f7eea2ae61d2fa637c26449aa533ddc6a7f0d83aff495f5f6c9d1f8d +size 33201 diff --git a/resources/healthsystem/infrastructure_and_equipment/ResourceFile_Equipment_Availability_Estimates.csv b/resources/healthsystem/infrastructure_and_equipment/ResourceFile_Equipment_Availability_Estimates.csv new file mode 100644 index 0000000000..829b95d1f9 --- /dev/null +++ b/resources/healthsystem/infrastructure_and_equipment/ResourceFile_Equipment_Availability_Estimates.csv @@ -0,0 +1,3 @@ +version https://git-lfs.github.com/spec/v1 +oid sha256:b7c0364af516509b47f161278a061817d22bcd06685f469c09be089454c02a22 +size 582777 diff --git a/src/tlo/analysis/utils.py b/src/tlo/analysis/utils.py index 3aeff2bb11..36cee558d4 100644 --- a/src/tlo/analysis/utils.py +++ b/src/tlo/analysis/utils.py @@ -1125,6 +1125,7 @@ def get_parameters_for_status_quo() -> Dict: "mode_appt_constraints": 1, "cons_availability": "default", "beds_availability": "default", + "equip_availability": "all", # <--- NB. Existing calibration is assuming all equipment is available }, } diff --git a/src/tlo/methods/equipment.py b/src/tlo/methods/equipment.py new file mode 100644 index 0000000000..4de29107f9 --- /dev/null +++ b/src/tlo/methods/equipment.py @@ -0,0 +1,270 @@ +import warnings +from collections import defaultdict +from typing import Counter, Iterable, Literal, Set, Union + +import numpy as np +import pandas as pd + +from tlo import logging + +logger_summary = logging.getLogger("tlo.methods.healthsystem.summary") + + +class Equipment: + """This is the equipment class. It maintains a current record of the availability of equipment in the + health system. It is expected that this is instantiated by the :py:class:`~.HealthSystem` module. + + The basic paradigm is that an :py:class:`~.HSI_Event` can declare equipment that is required for delivering the healthcare + service that the ``HSI_Event`` represents. The ``HSI_Event`` uses :py:meth:`HSI_event.add_equipment` to make these declarations, + with reference to the items of equipment that are defined in ``ResourceFile_EquipmentCatalogue.csv``. (These + declaration can be in the form of the descriptor or the equipment item code). These declarations can be used when + the ``HSI_Event`` is created but before it is run (in ``__init__``), or during execution of the ``HSI_Event`` (in :py:meth:`.HSI_Event.apply`). + + As the ``HSI_Event`` can declare equipment that is required before it is run, the HealthSystem *can* use this to + prevent an ``HSI_Event`` running if the equipment declared is not available. Note that for equipment that is declared + whilst the ``HSI_Event`` is running, there are no checks on availability, and the ``HSI_Event`` is allowed to continue + running even if equipment is declared is not available. For this reason, the ``HSI_Event`` should declare equipment + that is *essential* for the healthcare service in its ``__init__`` method. If the logic inside the ``apply`` method + of the ``HSI_Event`` depends on the availability of equipment, then it can find the probability with which + item(s) will be available using :py:meth:`.HSI_Event.probability_equipment_available`. + + The data on the availability of equipment data refers to the proportion of facilities in a district of a + particular level (i.e., the ``Facility_ID``) that do have that piece of equipment. In the model, we do not know + which actual facility the person is attending (there are many actual facilities grouped together into one + ``Facility_ID`` in the model). Therefore, the determination of whether equipment is available is made + probabilistically for the ``HSI_Event`` (i.e., the probability that the actual facility being attended by the + person has the equipment is represented by the proportion of such facilities that do have that equipment). It is + assumed that the probabilities of each item being available are independent of one other (so that the + probability of all items being available is the product of the probabilities for each item). This probabilistic + determination of availability is only done _once_ for the ``HSI_Event``: i.e., if the equipment is determined to + not be available for the instance of the ``HSI_Event``, then it will remain not available if the same event is + re-scheduled / re-entered into the ``HealthSystem`` queue. This represents that if the facility that a particular + person attends for the ``HSI_Event`` does not have the equipment available, then it will still not be available on + another day. + + Where data on availability is not provided for an item, the probability of availability is inferred from the + average availability of other items in that facility ID. Likewise, the probability of an item being available + at a facility ID is inferred from the average availability of that item at other facilities. If an item code is + referred in ``add_equipment`` that is not recognised (not included in :py:attr:`catalogue`), a :py:exc:`UserWarning` is issued, but + that item is then silently ignored. If a facility ID is ever referred that is not recognised (not included in + :py:attr:`master_facilities_list`), an :py:exc:`AssertionError` is raised. + + :param catalogue: The database of all recognised item_codes. + :param data_availability: Specifies the probability with which each equipment (identified by an ``item_code``) is + available at a facility level. Note that information is not necessarily provided for every item in the :py:attr`catalogue` + or every facility ID in the :py:attr`master_facilities_list`. + :param: rng: The random number generator object to use for random numbers. + :param availability: Determines the mode availability of the equipment. If 'default' then use the availability + specified in :py:attr:`data_availability`; if 'none', then let no equipment be ever be available; if 'all', then all + equipment is always available. + :param master_facilities_list: The :py:class:`~pandas.DataFrame` with the line-list of all the facilities in the health system. + """ + + def __init__( + self, + catalogue: pd.DataFrame, + data_availability: pd.DataFrame, + rng: np.random.RandomState, + master_facilities_list: pd.DataFrame, + availability: Literal["all", "default", "none"] = "default", + ) -> None: + # - Store arguments + self.catalogue = catalogue + self.rng = rng + self.data_availability = data_availability + self.availability = availability + self.master_facilities_list = master_facilities_list + + # - Data structures for quick look-ups for items and descriptors + self._item_code_lookup = self.catalogue.set_index('Item_Description')['Item_Code'].to_dict() + self._all_item_descriptors = set(self._item_code_lookup.keys()) + self._all_item_codes = set(self._item_code_lookup.values()) + self._all_fac_ids = self.master_facilities_list['Facility_ID'].unique() + + # - Probabilities of items being available at each facility_id + self._probabilities_of_items_available = self._calculate_equipment_availability_probabilities() + + # - Internal store of which items have been used at each facility_id This is of the form + # {facility_id: {item_code: count}}. + self._record_of_equipment_used_by_facility_id = defaultdict(Counter) + + + def on_simulation_end(self): + """Things to do when the simulation ends: + * Log (to the summary logger) the equipment that has been used. + """ + self.write_to_log() + + @property + def availability(self): + return self._availability + + @availability.setter + def availability(self, value: Literal["all", "default", "none"]): + assert value in {"all", "none", "default"}, f"New availability value {value} not recognised." + self._availability = value + + def _calculate_equipment_availability_probabilities(self) -> pd.Series: + """ + Compute the probabilities that each equipment item is available (at a given + facility), for use when the equipment availability is set to "default". + + The probabilities computed in this method are constant throughout the simulation, + however they will not be used when the equipment availability is "all" or "none". + Computing them once and storing the result allows us to avoid repeating this + calculation if the equipment availability change event occurs during the simulation. + """ + # Create "full" dataset, where we force that there is probability of availability for every item_code at every + # observed facility + dat = pd.Series( + index=pd.MultiIndex.from_product( + [self._all_fac_ids, self._all_item_codes], names=["Facility_ID", "Item_Code"] + ), + data=float("nan"), + ).combine_first( + self.data_availability.set_index(["Facility_ID", "Item_Code"])[ + "Pr_Available" + ] + ) + + # Merge in original dataset and use the mean in that facility_id to impute availability of missing item_codes + dat = dat.groupby("Facility_ID").transform(lambda x: x.fillna(x.mean())) + # ... and also impute availability for any facility_ids for which no data, based on all other facilities + dat = dat.groupby("Item_Code").transform(lambda x: x.fillna(x.mean())) + + # Check no missing values + assert not dat.isnull().any() + + return dat + + def parse_items(self, items: Union[int, str, Iterable[int], Iterable[str]]) -> Set[int]: + """Parse equipment items specified as an item_code (integer), an item descriptor (string), or an iterable of + item_codes or descriptors (but not a mix of the two), and return as a set of item_code (integers). For any + item_code/descriptor not recognised, a ``UserWarning`` is issued.""" + + def check_item_codes_recognised(item_codes: set[int]): + if not item_codes.issubset(self._all_item_codes): + warnings.warn(f'Item code(s) "{item_codes}" not recognised.') + + def check_item_descriptors_recognised(item_descriptors: set[str]): + if not item_descriptors.issubset(self._all_item_descriptors): + warnings.warn(f'Item descriptor(s) "{item_descriptors}" not recognised.') + + # Make into a set if it is not one already + if isinstance(items, (str, int)): + items = set([items]) + else: + items = set(items) + + items_are_ints = all(isinstance(element, int) for element in items) + + if items_are_ints: + check_item_codes_recognised(items) + # In the return, any unrecognised item_codes are silently ignored. + return items.intersection(self._all_item_codes) + else: + check_item_descriptors_recognised(items) # Warn for any unrecognised descriptors + # In the return, any unrecognised descriptors are silently ignored. + return set(self._item_code_lookup[i] for i in items if i in self._item_code_lookup) + + def probability_all_equipment_available( + self, facility_id: int, item_codes: Set[int] + ) -> float: + """ + Returns the probability that all the equipment item_codes are available + at the given facility. + + It does so by looking at the probabilities of each equipment item being + available and multiplying these together to find the probability that *all* + are available. + + NOTE: This will error if the facility ID or any of the item codes is not recognised. + + :param facility_id: Facility at which to check for the equipment. + :param item_codes: Integer item codes corresponding to the equipment to check. + """ + + assert facility_id in self._all_fac_ids, f"Unrecognised facility ID: {facility_id=}" + assert item_codes.issubset(self._all_item_codes), f"At least one item code was unrecognised: {item_codes=}" + + if self.availability == "all": + return 1.0 + elif self.availability == "none": + return 0.0 + return self._probabilities_of_items_available.loc[ + (facility_id, list(item_codes)) + ].prod() + + def is_all_items_available( + self, item_codes: Set[int], facility_id: int + ) -> bool: + """ + Determine if all equipment items are available at the given facility_id. + Returns True only if all items are available at the facility_id, + otherwise returns False. + """ + if item_codes: + return self.rng.random_sample() < self.probability_all_equipment_available( + facility_id=facility_id, + item_codes=item_codes, + ) + else: + # In the case of an empty set, default to True without doing anything else ('no equipment' is always + # "available"). This is the most common case, so optimising for speed. + return True + + def record_use_of_equipment( + self, item_codes: Set[int], facility_id: int + ) -> None: + """Update internal record of the usage of items at equipment at the specified facility_id.""" + self._record_of_equipment_used_by_facility_id[facility_id].update(item_codes) + + def write_to_log(self) -> None: + """Write to the log: + * Summary of the equipment that was _ever_ used at each district/facility level. + Note that the info-level health system logger (key: `hsi_event_counts`) contains logging of the equipment used + in each HSI event (if finer splits are needed). Alternatively, different aggregations could be created here for + the summary logger, using the same pattern as used here. + """ + + mfl = self.master_facilities_list + + def set_of_keys_or_empty_set(x: Union[set, dict]): + if isinstance(x, set): + return x + elif isinstance(x, dict): + return set(x.keys()) + else: + return set() + + set_of_equipment_ever_used_at_each_facility_id = pd.Series({ + fac_id: set_of_keys_or_empty_set(self._record_of_equipment_used_by_facility_id.get(fac_id, set())) + for fac_id in mfl['Facility_ID'] + }, name='EquipmentEverUsed').astype(str) + + output = mfl.merge( + set_of_equipment_ever_used_at_each_facility_id, + left_on='Facility_ID', + right_index=True, + how='left', + ).drop(columns=['Facility_ID', 'Facility_Name']) + + # Log multi-row data-frame + for _, row in output.iterrows(): + logger_summary.info( + key='EquipmentEverUsed_ByFacilityID', + description='For each facility_id (the set of facilities of the same level in a district), the set of' + 'equipment items that are ever used.', + data=row.to_dict(), + ) + + def lookup_item_codes_from_pkg_name(self, pkg_name: str) -> Set[int]: + """Convenience function to find the set of item_codes that are grouped under a package name in the catalogue. + It is expected that this is used by the disease module once and then the resulting equipment item_codes are + saved on the module.""" + df = self.catalogue + + if pkg_name not in df['Pkg_Name'].unique(): + raise ValueError(f'That Pkg_Name is not in the catalogue: {pkg_name=}') + + return set(df.loc[df['Pkg_Name'] == pkg_name, 'Item_Code'].values) diff --git a/src/tlo/methods/healthsystem.py b/src/tlo/methods/healthsystem.py index e9a35b3669..1e71c04f46 100644 --- a/src/tlo/methods/healthsystem.py +++ b/src/tlo/methods/healthsystem.py @@ -27,6 +27,7 @@ get_item_codes_from_package_name, ) from tlo.methods.dxmanager import DxManager +from tlo.methods.equipment import Equipment from tlo.methods.hsi_event import ( LABEL_FOR_MERGED_FACILITY_LEVELS_1B_AND_2, FacilityInfo, @@ -193,6 +194,27 @@ class HealthSystem(Module): "Availability of beds. If 'default' then use the availability specified in the ResourceFile; if " "'none', then let no beds be ever be available; if 'all', then all beds are always available. NB. This " "parameter is over-ridden if an argument is provided to the module initialiser."), + 'EquipmentCatalogue': Parameter( + Types.DATA_FRAME, "Data on equipment items and packages."), + 'equipment_availability_estimates': Parameter( + Types.DATA_FRAME, "Data on the availability of equipment items and packages." + ), + 'equip_availability': Parameter( + Types.STRING, + "What to assume about the availability of equipment. If 'default' then use the availability specified in " + "the ResourceFile; if 'none', then let no equipment ever be available; if 'all', then all equipment is " + "always available. NB. This parameter is over-ridden if an argument is provided to the module initialiser." + ), + 'equip_availability_postSwitch': Parameter( + Types.STRING, + "What to assume about the availability of equipment after the switch (see `year_equip_availability_switch`" + "). The options for this are the same as `equip_availability`." + ), + 'year_equip_availability_switch': Parameter( + Types.INT, + "Year in which the assumption for `equip_availability` changes (The change happens on 1st January of that " + "year.)" + ), # Service Availability 'Service_Availability': Parameter( @@ -314,6 +336,7 @@ def __init__( mode_appt_constraints: Optional[int] = None, cons_availability: Optional[str] = None, beds_availability: Optional[str] = None, + equip_availability: Optional[str] = None, randomise_queue: bool = True, ignore_priority: bool = False, policy_name: Optional[str] = None, @@ -338,6 +361,8 @@ def __init__( or 'none', requests for consumables are not logged. :param beds_availability: If 'default' then use the availability specified in the ResourceFile; if 'none', then let no beds be ever be available; if 'all', then all beds are always available. + :param equip_availability: If 'default' then use the availability specified in the ResourceFile; if 'none', then + let no equipment ever be available; if 'all', then all equipment is always available. :param randomise_queue ensure that the queue is not model-dependent, i.e. properly randomised for equal topen and priority :param ignore_priority: If ``True`` do not use the priority information in HSI @@ -402,7 +427,6 @@ def __init__( 'LCOA_EHP'] self.arg_policy_name = policy_name - self.tclose_overwrite = None self.tclose_days_offset_overwrite = None @@ -431,13 +455,16 @@ def __init__( self.HSI_EVENT_QUEUE = [] self.hsi_event_queue_counter = 0 # Counter to help with the sorting in the heapq - # Store the argument provided for cons_availability + # Store the arguments provided for cons/beds/equip_availability assert cons_availability in (None, 'default', 'all', 'none') self.arg_cons_availability = cons_availability assert beds_availability in (None, 'default', 'all', 'none') self.arg_beds_availability = beds_availability + assert equip_availability in (None, 'default', 'all', 'none') + self.arg_equip_availability = equip_availability + # `compute_squeeze_factor_to_district_level` is a Boolean indicating whether the computation of squeeze_factors # should be specific to each district (when `True`), or if the computation of squeeze_factors should be on the # basis that resources from all districts can be effectively "pooled" (when `False). @@ -529,6 +556,16 @@ def read_parameters(self, data_folder): self.parameters['BedCapacity'] = pd.read_csv( path_to_resourcefiles_for_healthsystem / 'infrastructure_and_equipment' / 'ResourceFile_Bed_Capacity.csv') + # Read in ResourceFile_Equipment + self.parameters['EquipmentCatalogue'] = pd.read_csv( + path_to_resourcefiles_for_healthsystem + / 'infrastructure_and_equipment' + / 'ResourceFile_EquipmentCatalogue.csv') + self.parameters['equipment_availability_estimates'] = pd.read_csv( + path_to_resourcefiles_for_healthsystem + / 'infrastructure_and_equipment' + / 'ResourceFile_Equipment_Availability_Estimates.csv') + # Data on the priority of each Treatment_ID that should be adopted in the queueing system according to different # priority policies. Load all policies at this stage, and decide later which one to adopt. self.parameters['priority_rank'] = pd.read_excel(path_to_resourcefiles_for_healthsystem / 'priority_policies' / @@ -577,7 +614,6 @@ def read_parameters(self, data_folder): # Ensure that a value for the year at the start of the simulation is provided. assert all(2010 in sheet['year'].values for sheet in self.parameters['yearly_HR_scaling'].values()) - def pre_initialise_population(self): """Generate the accessory classes used by the HealthSystem and pass to them the data that has been read.""" @@ -585,6 +621,7 @@ def pre_initialise_population(self): self.rng_for_hsi_queue = np.random.RandomState(self.rng.randint(2 ** 31 - 1)) self.rng_for_dx = np.random.RandomState(self.rng.randint(2 ** 31 - 1)) rng_for_consumables = np.random.RandomState(self.rng.randint(2 ** 31 - 1)) + rng_for_equipment = np.random.RandomState(self.rng.randint(2 ** 31 - 1)) # Determine mode_appt_constraints self.mode_appt_constraints = self.get_mode_appt_constraints() @@ -609,6 +646,15 @@ def pre_initialise_population(self): availability=self.get_cons_availability() ) + # Determine equip_availability + self.equipment = Equipment( + catalogue=self.parameters['EquipmentCatalogue'], + data_availability=self.parameters['equipment_availability_estimates'], + rng=rng_for_equipment, + master_facilities_list=self.parameters['Master_Facilities_List'], + availability=self.get_equip_availability(), + ) + self.tclose_overwrite = self.parameters['tclose_overwrite'] self.tclose_days_offset_overwrite = self.parameters['tclose_days_offset_overwrite'] @@ -619,7 +665,6 @@ def pre_initialise_population(self): # Set up framework for considering a priority policy self.setup_priority_policy() - def initialise_population(self, population): self.bed_days.initialise_population(population.props) @@ -673,6 +718,18 @@ def initialise_simulation(self, sim): Date(self.parameters["year_cons_availability_switch"], 1, 1) ) + # Schedule an equipment availability switch + sim.schedule_event( + HealthSystemChangeParameters( + self, + parameters={ + 'equip_availability': self.parameters['equip_availability_postSwitch'] + } + ), + Date(self.parameters["year_equip_availability_switch"], 1, 1) + ) + + # Schedule a one-off rescaling of _daily_capabilities broken down by officer type and level. # This occurs on 1st January of the year specified in the parameters. sim.schedule_event(ConstantRescalingHRCapabilities(self), @@ -695,6 +752,8 @@ def on_simulation_end(self): """Put out to the log the information from the tracker of the last day of the simulation""" self.bed_days.on_simulation_end() self.consumables.on_simulation_end() + self.equipment.on_simulation_end() + if self._hsi_event_count_log_period == "simulation": self._write_hsi_event_counts_to_log_and_reset() self._write_never_ran_hsi_event_counts_to_log_and_reset() @@ -744,8 +803,6 @@ def setup_priority_policy(self): def process_human_resources_files(self, use_funded_or_actual_staffing: str): """Create the data-structures needed from the information read into the parameters.""" - - # * Define Facility Levels self._facility_levels = set(self.parameters['Master_Facilities_List']['Facility_Level']) - {'5'} assert self._facility_levels == {'0', '1a', '1b', '2', '3', '4'} # todo soft code this? @@ -1047,6 +1104,24 @@ def get_beds_availability(self) -> str: return _beds_availability + def get_equip_availability(self) -> str: + """Returns equipment availability. (Should be equal to what is specified by the parameter, but can be + overwritten with what was provided in argument if an argument was specified -- provided for backward + compatibility/debugging.)""" + + if self.arg_equip_availability is None: + _equip_availability = self.parameters['equip_availability'] + else: + _equip_availability = self.arg_equip_availability + + # Log the equip_availability + logger.info(key="message", + data=f"Running Health System With the Following Equipment Availability: " + f"{_equip_availability}" + ) + + return _equip_availability + def schedule_to_call_never_ran_on_date(self, hsi_event: 'HSI_Event', tdate: datetime.datetime): """Function to schedule never_ran being called on a given date""" self.sim.schedule_event(HSIEventWrapper(hsi_event=hsi_event, run_hsi=False), tdate) @@ -1588,6 +1663,7 @@ def write_to_hsi_log( priority: int, ): """Write the log `HSI_Event` and add to the summary counter.""" + # Debug logger gives simple line-list for every HSI event logger.debug( key="HSI_Event", data={ @@ -1600,15 +1676,19 @@ def write_to_hsi_log( 'did_run': did_run, 'Facility_Level': event_details.facility_level if event_details.facility_level is not None else -99, 'Facility_ID': facility_id if facility_id is not None else -99, + 'Equipment': sorted(event_details.equipment), }, description="record of each HSI event" ) if did_run: if self._hsi_event_count_log_period is not None: + # Do logging for HSI Event using counts of each 'unique type' of HSI event (as defined by + # `HSIEventDetails`). event_details_key = self._hsi_event_details.setdefault( event_details, len(self._hsi_event_details) ) self._hsi_event_counts_log_period[event_details_key] += 1 + # Do logging for 'summary logger' self._summary_counter.record_hsi_event( treatment_id=event_details.treatment_id, hsi_event_name=event_details.event_name, @@ -1815,7 +1895,10 @@ def on_end_of_year(self) -> None: # If we are at the end of the year preceeding the mode switch, and if wanted # to rescale capabilities to capture effective availability as was recorded, on # average, in the past year, do so here. - if (self.sim.date.year == self.parameters['year_mode_switch'] - 1) and self.parameters['scale_to_effective_capabilities']: + if ( + (self.sim.date.year == self.parameters['year_mode_switch'] - 1) + and self.parameters['scale_to_effective_capabilities'] + ): self._rescale_capabilities_to_capture_effective_capability() self._summary_counter.write_to_log_and_reset_counters() self.consumables.on_end_of_year() @@ -2133,9 +2216,19 @@ def process_events_mode_0_and_1(self, hold_over: List[HSIEventQueueItem]) -> Non # Run the list of population-level HSI events self.module.run_population_level_events(list_of_population_hsi_event_tuples_due_today) - # Run the list of individual-level events + # For each individual level event, check whether the equipment it has already declared is available. If it + # is not, then call the HSI's never_run function, and do not take it forward for running; if it is then + # add it to the list of events to run. + list_of_individual_hsi_event_tuples_due_today_that_have_essential_equipment = list() + for item in list_of_individual_hsi_event_tuples_due_today: + if not item.hsi_event.is_all_declared_equipment_available: + self.module.call_and_record_never_ran_hsi_event(hsi_event=item.hsi_event, priority=item.priority) + else: + list_of_individual_hsi_event_tuples_due_today_that_have_essential_equipment.append(item) + + # Try to run the list of individual-level events that have their essential equipment _to_be_held_over = self.module.run_individual_level_events_in_mode_0_or_1( - list_of_individual_hsi_event_tuples_due_today, + list_of_individual_hsi_event_tuples_due_today_that_have_essential_equipment, ) hold_over.extend(_to_be_held_over) @@ -2271,6 +2364,15 @@ def process_events_mode_2(self, hold_over: List[HSIEventQueueItem]) -> None: assert event.facility_info is not None, \ f"Cannot run HSI {event.TREATMENT_ID} without facility_info being defined." + # Check if equipment declared is available. If not, call `never_ran` and do not run the + # event. (`continue` returns flow to beginning of the `while` loop) + if not event.is_all_declared_equipment_available: + self.module.call_and_record_never_ran_hsi_event( + hsi_event=event, + priority=next_event_tuple.priority + ) + continue + # Expected appt footprint before running event _appt_footprint_before_running = event.EXPECTED_APPT_FOOTPRINT # Run event & get actual footprint @@ -2437,7 +2539,8 @@ def apply(self, population): treatment_id='Inpatient_Care', facility_level=self.module._facility_by_facility_id[_fac_id].level, appt_footprint=tuple(sorted(_inpatient_appts.items())), - beddays_footprint=() + beddays_footprint=(), + equipment=tuple(), # Equipment is normally a set, but this has to be hashable. ), person_id=-1, facility_id=_fac_id, @@ -2648,6 +2751,7 @@ class HealthSystemChangeParameters(Event, PopulationScopeEventMixin): * `capabilities_coefficient` * `cons_availability` * `beds_availability` + * `equip_availability` Note that no checking is done here on the suitability of values of each parameter.""" def __init__(self, module: HealthSystem, parameters: Dict): @@ -2674,6 +2778,9 @@ def apply(self, population): if 'beds_availability' in self._parameters: self.module.bed_days.availability = self._parameters['beds_availability'] + if 'equip_availability' in self._parameters: + self.module.equipment.availability = self._parameters['equip_availability'] + class DynamicRescalingHRCapabilities(RegularEvent, PopulationScopeEventMixin): """ This event exists to scale the daily capabilities assumed at fixed time intervals""" diff --git a/src/tlo/methods/hsi_event.py b/src/tlo/methods/hsi_event.py index 3d6b4c60cf..07b5d21978 100644 --- a/src/tlo/methods/hsi_event.py +++ b/src/tlo/methods/hsi_event.py @@ -1,7 +1,7 @@ from __future__ import annotations from collections import Counter -from typing import TYPE_CHECKING, Dict, Literal, NamedTuple, Optional, Tuple, Union +from typing import TYPE_CHECKING, Dict, Iterable, Literal, NamedTuple, Optional, Set, Tuple, Union import numpy as np @@ -41,6 +41,7 @@ class HSIEventDetails(NamedTuple): facility_level: Optional[str] appt_footprint: Tuple[Tuple[str, int]] beddays_footprint: Tuple[Tuple[str, int]] + equipment: Tuple[str] class HSIEventQueueItem(NamedTuple): @@ -72,7 +73,7 @@ class HSI_Event: """Base HSI event class, from which all others inherit. Concrete subclasses should also inherit from one of the EventMixin classes - defined in `src/tlo/events.py`, and implement at least an `apply` and + defined in `src/tlo/events.py`, and implement at least an `apply` and `did_not_run` method. """ @@ -102,15 +103,21 @@ def __init__(self, module, *args, **kwargs): self.module = module super().__init__(*args, **kwargs) - # Information that will later be received about this HSI + # Information that will later be received/computed about this HSI self._received_info_about_bed_days = None self.expected_time_requests = {} self.facility_info = None + self._is_all_declared_equipment_available = None self.TREATMENT_ID = "" self.ACCEPTED_FACILITY_LEVEL = None # Set "dynamic" default value self.BEDDAYS_FOOTPRINT = self.make_beddays_footprint({}) + self._EQUIPMENT: Set[int] = set() # The set of equipment that is used in the HSI. If any items in this set are + # not available at the point when the HSI will be run, then the HSI is not + # run, and the `never_ran` method is called instead. This is a declaration + # of resource needs, but is private because users are expected to use + # `add_equipment` to declare equipment needs. @property def bed_days_allocated_to_this_event(self): @@ -165,12 +172,23 @@ def never_ran(self) -> None: logger.debug(key="message", data=f"{self.__class__.__name__}: was never run.") def post_apply_hook(self) -> None: - """Impose the bed-days footprint (if target of the HSI is a person_id)""" + """ + Do things following the event's `apply` function running. + * Impose the bed-days footprint (if target of the HSI is a person_id) + * Record the equipment that has been added before and during the course of the HSI Event. + """ if isinstance(self.target, int): self.healthcare_system.bed_days.impose_beddays_footprint( person_id=self.target, footprint=self.bed_days_allocated_to_this_event ) + if self.facility_info is not None: + # If there is a facility_info (e.g., healthsystem not running in disabled mode), then record equipment used + self.healthcare_system.equipment.record_use_of_equipment( + item_codes=self._EQUIPMENT, + facility_id=self.facility_info.id + ) + def run(self, squeeze_factor): """Make the event happen.""" updated_appt_footprint = self.apply(self.target, squeeze_factor) @@ -262,6 +280,41 @@ def make_appt_footprint(self, dict_of_appts) -> Counter: "values" ) + def add_equipment(self, item_codes: Union[int, str, Iterable[int], Iterable[str]]) -> None: + """Declare that piece(s) of equipment are used in this HSI_Event. Equipment items can be identified by their + item_codes (int) or descriptors (str); a singular item or an iterable of items (either codes or descriptors but + not a mix of both) can be defined at once. Checks are done on the validity of the item_codes/item + descriptions and a warning issued if any are not recognised.""" + self._EQUIPMENT.update(self.healthcare_system.equipment.parse_items(item_codes)) + + @property + def is_all_declared_equipment_available(self) -> bool: + """Returns ``True`` if all the (currently) declared items of equipment are available. This is called by the + ``HealthSystem`` module before the HSI is run and so is looking only at those items that are declared when this + instance was created. The evaluation of whether equipment is available is only done *once* for this instance of + the event: i.e., if the equipment is not available for the instance of this ``HSI_Event``, then it will remain not + available if the same event is re-scheduled/re-entered into the HealthSystem queue. This is representing that + if the facility that a particular person attends for the ``HSI_Event`` does not have the equipment available, then + it will also not be available on another day.""" + + if self._is_all_declared_equipment_available is None: + # Availability has not already been evaluated: determine availability + self._is_all_declared_equipment_available = self.healthcare_system.equipment.is_all_items_available( + item_codes=self._EQUIPMENT, + facility_id=self.facility_info.id, + ) + return self._is_all_declared_equipment_available + + def probability_all_equipment_available(self, item_codes: Union[int, str, Iterable[int], Iterable[str]]) -> float: + """Returns the probability that all the equipment item_codes are available. This does not imply that the + equipment is being used and no logging happens. It is provided as a convenience to disease module authors in + case the logic during an ``HSI_Event`` depends on the availability of a piece of equipment. This function + accepts the item codes/descriptions in a variety of formats, so the argument needs to be parsed.""" + return self.healthcare_system.equipment.probability_all_equipment_available( + item_codes=self.healthcare_system.equipment.parse_items(item_codes), + facility_id=self.facility_info.id, + ) + def initialise(self) -> None: """Initialise the HSI: * Set the facility_info @@ -370,6 +423,7 @@ def as_namedtuple( beddays_footprint=tuple( sorted((k, v) for k, v in self.BEDDAYS_FOOTPRINT.items() if v > 0) ), + equipment=tuple(sorted(self._EQUIPMENT)), ) diff --git a/tests/test_alri.py b/tests/test_alri.py index a98d2f277c..0fba5fea8d 100644 --- a/tests/test_alri.py +++ b/tests/test_alri.py @@ -54,7 +54,7 @@ def _get_person_id(df, age_bounds: tuple = (0.0, np.inf)) -> int: ].index[0] -def get_sim(tmpdir, seed, cons_available): +def get_sim(tmpdir, seed, cons_available, equip_available='all'): """Return simulation objection with Alri and other necessary modules registered.""" sim = Simulation( start_date=start_date, @@ -77,7 +77,8 @@ def get_sim(tmpdir, seed, cons_available): healthseekingbehaviour.HealthSeekingBehaviour(resourcefilepath=resourcefilepath), healthburden.HealthBurden(resourcefilepath=resourcefilepath), healthsystem.HealthSystem(resourcefilepath=resourcefilepath, - cons_availability=cons_available), + cons_availability=cons_available, + equip_availability=equip_available), alri.Alri(resourcefilepath=resourcefilepath, log_indivdual=0, do_checks=True), AlriPropertiesOfOtherModules(), ) @@ -85,10 +86,10 @@ def get_sim(tmpdir, seed, cons_available): @pytest.fixture -def sim_hs_all_consumables(tmpdir, seed): +def sim_hs_all_consumables_and_equipment(tmpdir, seed): """Return simulation objection with Alri and other necessary modules registered. All consumables available""" - return get_sim(tmpdir=tmpdir, seed=seed, cons_available='all') + return get_sim(tmpdir=tmpdir, seed=seed, cons_available='all', equip_available='all') @pytest.fixture @@ -127,17 +128,17 @@ def sim_hs_default_consumables(tmpdir, seed): return sim -def check_dtypes(sim_hs_all_consumables): - sim = sim_hs_all_consumables +def check_dtypes(sim_hs_all_consumables_and_equipment): + sim = sim_hs_all_consumables_and_equipment # Check types of columns df = sim.population.props orig = sim.population.new_row assert (df.dtypes == orig.dtypes).all() -def test_integrity_of_linear_models(sim_hs_all_consumables): +def test_integrity_of_linear_models(sim_hs_all_consumables_and_equipment): """Run the models to make sure that is specified correctly and can run.""" - sim = sim_hs_all_consumables + sim = sim_hs_all_consumables_and_equipment sim.make_initial_population(n=5000) alri_module = sim.modules['Alri'] df = sim.population.props @@ -322,21 +323,21 @@ def test_integrity_of_linear_models(sim_hs_all_consumables): assert isinstance(res, float) and (res is not None) and (0.0 <= res <= 1.0), f"Problem with: {kwargs=}" -def test_basic_run(sim_hs_all_consumables): +def test_basic_run(sim_hs_all_consumables_and_equipment): """Short run of the module using default parameters with check on dtypes""" - sim = sim_hs_all_consumables + sim = sim_hs_all_consumables_and_equipment dur = pd.DateOffset(months=1) popsize = 100 sim.make_initial_population(n=popsize) sim.simulate(end_date=start_date + dur) - check_dtypes(sim_hs_all_consumables) + check_dtypes(sim_hs_all_consumables_and_equipment) @pytest.mark.slow -def test_basic_run_lasting_two_years(sim_hs_all_consumables): +def test_basic_run_lasting_two_years(sim_hs_all_consumables_and_equipment): """Check logging results in a run of the model for two years, including HSI, with daily property config checking""" - sim = sim_hs_all_consumables + sim = sim_hs_all_consumables_and_equipment dur = pd.DateOffset(years=2) popsize = 5000 @@ -362,9 +363,9 @@ def test_basic_run_lasting_two_years(sim_hs_all_consumables): assert set(log_one_person.columns) == set(sim.modules['Alri'].PROPERTIES.keys()) -def test_alri_polling(sim_hs_all_consumables): +def test_alri_polling(sim_hs_all_consumables_and_equipment): """Check polling events leads to incident cases""" - sim = sim_hs_all_consumables + sim = sim_hs_all_consumables_and_equipment # get simulation object: popsize = 100 @@ -386,10 +387,10 @@ def test_alri_polling(sim_hs_all_consumables): assert len([q for q in sim.event_queue.queue if isinstance(q[3], AlriIncidentCase)]) > 0 -def test_nat_hist_recovery(sim_hs_all_consumables): +def test_nat_hist_recovery(sim_hs_all_consumables_and_equipment): """Check: Infection onset --> recovery""" - sim = sim_hs_all_consumables + sim = sim_hs_all_consumables_and_equipment popsize = 100 sim.make_initial_population(n=popsize) @@ -466,9 +467,9 @@ def __will_die_of_alri(**kwargs): assert 0 == sim.modules['Alri'].logging_event.trackers['cured_cases'].report_current_total() -def test_nat_hist_death(sim_hs_all_consumables): +def test_nat_hist_death(sim_hs_all_consumables_and_equipment): """Check: Infection onset --> death""" - sim = sim_hs_all_consumables + sim = sim_hs_all_consumables_and_equipment popsize = 100 sim.make_initial_population(n=popsize) @@ -523,10 +524,10 @@ def __will_die_of_alri(**kwargs): assert 0 == sim.modules['Alri'].logging_event.trackers['cured_cases'].report_current_total() -def test_nat_hist_cure_if_recovery_scheduled(sim_hs_all_consumables): +def test_nat_hist_cure_if_recovery_scheduled(sim_hs_all_consumables_and_equipment): """Show that if a cure event is run before when a person was going to recover naturally, it cause the episode to end earlier.""" - sim = sim_hs_all_consumables + sim = sim_hs_all_consumables_and_equipment popsize = 100 @@ -598,10 +599,10 @@ def death(**kwargs): assert 1 == sim.modules['Alri'].logging_event.trackers['cured_cases'].report_current_total() -def test_nat_hist_cure_if_death_scheduled(sim_hs_all_consumables): +def test_nat_hist_cure_if_death_scheduled(sim_hs_all_consumables_and_equipment): """Show that if a cure event is run before when a person was going to die, it cause the episode to end without the person dying.""" - sim = sim_hs_all_consumables + sim = sim_hs_all_consumables_and_equipment popsize = 100 sim.make_initial_population(n=popsize) @@ -667,10 +668,10 @@ def death(**kwargs): assert 1 == sim.modules['Alri'].logging_event.trackers['cured_cases'].report_current_total() -def test_immediate_onset_complications(sim_hs_all_consumables): +def test_immediate_onset_complications(sim_hs_all_consumables_and_equipment): """Check that if probability of immediately onsetting complications is 100%, then a person has all those complications immediately onset""" - sim = sim_hs_all_consumables + sim = sim_hs_all_consumables_and_equipment popsize = 100 sim.make_initial_population(n=popsize) @@ -712,11 +713,11 @@ def test_immediate_onset_complications(sim_hs_all_consumables): assert df.at[person_id, 'ri_SpO2_level'] != '>=93%' -def test_no_immediate_onset_complications(sim_hs_all_consumables): +def test_no_immediate_onset_complications(sim_hs_all_consumables_and_equipment): """Check that if probability of immediately onsetting complications is 0%, then a person has none of those complications immediately onset """ - sim = sim_hs_all_consumables + sim = sim_hs_all_consumables_and_equipment popsize = 100 @@ -748,7 +749,7 @@ def test_no_immediate_onset_complications(sim_hs_all_consumables): assert not df.loc[person_id, complications_cols].any() -def test_classification_based_on_symptoms_and_imci(sim_hs_all_consumables): +def test_classification_based_on_symptoms_and_imci(sim_hs_all_consumables_and_equipment): """Check that `_get_disease_classification` gives the expected classification.""" def make_hw_assesement_perfect(sim): @@ -760,7 +761,7 @@ def make_hw_assesement_perfect(sim): p['sensitivity_of_classification_of_non_severe_pneumonia_facility_level2'] = 1.0 p['sensitivity_of_classification_of_severe_pneumonia_facility_level2'] = 1.0 - sim = sim_hs_all_consumables + sim = sim_hs_all_consumables_and_equipment make_hw_assesement_perfect(sim) sim.make_initial_population(n=1000) hsi_alri_treatment = HSI_Alri_Treatment(sim.modules['Alri'], 0) @@ -846,9 +847,9 @@ def make_hw_assesement_perfect(sim): ), f"{_correct_imci_classification_on_symptoms=}" -def test_do_effects_of_alri_treatment(sim_hs_all_consumables): +def test_do_effects_of_alri_treatment(sim_hs_all_consumables_and_equipment): """Check that running `do_alri_treatment` can prevent a death from occurring.""" - sim = sim_hs_all_consumables + sim = sim_hs_all_consumables_and_equipment popsize = 100 sim.make_initial_population(n=popsize) @@ -921,10 +922,10 @@ def test_do_effects_of_alri_treatment(sim_hs_all_consumables): assert 1 == sim.modules['Alri'].logging_event.trackers['cured_cases'].report_current_total() -def test_severe_pneumonia_referral_from_hsi_first_appts(sim_hs_all_consumables): +def test_severe_pneumonia_referral_from_hsi_first_appts(sim_hs_all_consumables_and_equipment): """Check that a person is scheduled a treatment HSI following a presentation at HSI_GenericFirstApptAtFacilityLevel0 with severe pneumonia.""" - sim = sim_hs_all_consumables + sim = sim_hs_all_consumables_and_equipment popsize = 100 sim.make_initial_population(n=popsize) @@ -1234,6 +1235,7 @@ def initialise_simulation(self, sim): healthsystem.HealthSystem(resourcefilepath=resourcefilepath, disable_and_reject_all=disable_and_reject_all, cons_availability='all', + equip_availability='all', ), alri.Alri(resourcefilepath=resourcefilepath), AlriPropertiesOfOtherModules(), @@ -1338,6 +1340,7 @@ def initialise_simulation(self, sim): healthburden.HealthBurden(resourcefilepath=resourcefilepath), healthsystem.HealthSystem(resourcefilepath=resourcefilepath, cons_availability='all', + equip_availability='all', ), alri.Alri(resourcefilepath=resourcefilepath), AlriPropertiesOfOtherModules(), diff --git a/tests/test_equipment.py b/tests/test_equipment.py new file mode 100644 index 0000000000..3e4daecc44 --- /dev/null +++ b/tests/test_equipment.py @@ -0,0 +1,478 @@ +"""This file contains all the tests to do with Equipment.""" +import os +from pathlib import Path +from typing import Dict + +import numpy as np +import pandas as pd +import pytest + +from tlo import Date, Module, Simulation +from tlo.analysis.utils import parse_log_file +from tlo.events import IndividualScopeEventMixin +from tlo.methods import Metadata, demography, healthsystem +from tlo.methods.equipment import Equipment +from tlo.methods.hsi_event import HSI_Event + +resourcefilepath = Path(os.path.dirname(__file__)) / "../resources" + + +def test_core_functionality_of_equipment_class(seed): + """Test that the core functionality of the equipment class works on toy data.""" + + # Create toy data + catalogue = pd.DataFrame( + [ + {"Item_Description": "ItemZero", "Item_Code": 0, "Pkg_Name": float('nan')}, + {"Item_Description": "ItemOne", "Item_Code": 1, "Pkg_Name": float('nan')}, + {"Item_Description": "ItemTwo", "Item_Code": 2, "Pkg_Name": 'PkgWith2+3'}, + {"Item_Description": "ItemThree", "Item_Code": 3, "Pkg_Name": 'PkgWith2+3'}, + ] + ) + data_availability = pd.DataFrame( + # item 0 is not available anywhere; item 1 is available everywhere; item 2 is available only at facility_id=1 + # No data for fac_id=2 + [ + {"Item_Code": 0, "Facility_ID": 0, "Pr_Available": 0.0}, + {"Item_Code": 0, "Facility_ID": 1, "Pr_Available": 0.0}, + {"Item_Code": 1, "Facility_ID": 0, "Pr_Available": 1.0}, + {"Item_Code": 1, "Facility_ID": 1, "Pr_Available": 1.0}, + {"Item_Code": 2, "Facility_ID": 0, "Pr_Available": 0.0}, + {"Item_Code": 2, "Facility_ID": 1, "Pr_Available": 1.0}, + ] + ) + + mfl = pd.DataFrame( + [ + {'District': 'D0', 'Facility_Level': '1a', 'Region': 'R0', 'Facility_ID': 0, 'Facility_Name': 'Fac0'}, + {'District': 'D0', 'Facility_Level': '1b', 'Region': 'R0', 'Facility_ID': 1, 'Facility_Name': 'Fac1'}, + {'District': 'D0', 'Facility_Level': '2', 'Region': 'R0', 'Facility_ID': 2, 'Facility_Name': 'Fac2'}, + ] + ) + + # Create instance of the Equipment class with these toy data and check availability of equipment... + # -- when using `default` behaviour: + eq_default = Equipment( + catalogue=catalogue, + data_availability=data_availability, + rng=np.random.RandomState(np.random.MT19937(np.random.SeedSequence(seed))), + master_facilities_list=mfl, + availability="default", + ) + + # Checks on parsing equipment items + # - using single integer for one item_code + assert {1} == eq_default.parse_items(1) + # - using list of integers for item_codes + assert {1, 2} == eq_default.parse_items([1, 2]) + # - using single string for one item descriptor + assert {1} == eq_default.parse_items('ItemOne') + # - using list of strings for item descriptors + assert {1, 2} == eq_default.parse_items(['ItemOne', 'ItemTwo']) + # - an empty iterable of equipment should always be work whether expressed as list/tuple/set + assert set() == eq_default.parse_items(list()) + assert set() == eq_default.parse_items(tuple()) + assert set() == eq_default.parse_items(set()) + + # - Calling for unrecognised item_codes (should raise warning) + with pytest.warns(): + eq_default.parse_items(10001) + with pytest.warns(): + eq_default.parse_items('ItemThatIsNotDefined') + + # Testing checking on available of items + # - calling when all items available (should be true) + assert eq_default.is_all_items_available(item_codes={1, 2}, facility_id=1) + # - calling when no items available (should be false) + assert not eq_default.is_all_items_available(item_codes={0, 2}, facility_id=0) + # - calling when some items available (should be false) + assert not eq_default.is_all_items_available(item_codes={1, 2}, facility_id=0) + # - calling for empty set of equipment (should always be available) + assert eq_default.is_all_items_available(item_codes=set(), facility_id=0) + + # - calling an item for which data on availability is not provided (should not raise error) + eq_default.is_all_items_available(item_codes={3}, facility_id=1) + # - calling an item at a facility that for which data is not provided (should give average behaviour for other + # facilities) + assert not eq_default.is_all_items_available(item_codes={0}, facility_id=2) + assert eq_default.is_all_items_available(item_codes={1}, facility_id=2) + # - calling a recognised item for which no data at a facility with no data (should not error) + eq_default.is_all_items_available(item_codes={3}, facility_id=2) + # -- calling for an unrecognised facility_id (should error) + with pytest.raises(AssertionError): + eq_default.is_all_items_available(item_codes={1}, facility_id=1001) + + # -- when using `none` availability behaviour: everything should not be available! + eq_none = Equipment( + catalogue=catalogue, + data_availability=data_availability, + rng=np.random.RandomState(np.random.MT19937(np.random.SeedSequence(seed))), + availability="none", + master_facilities_list=mfl, + ) + # - calling when all items available (should be false because using 'none' behaviour) + assert not eq_none.is_all_items_available(item_codes={1, 2}, facility_id=1) + # - calling when no items available (should be false) + assert not eq_none.is_all_items_available(item_codes={0, 2}, facility_id=0) + # - calling when some items available (should be false) + assert not eq_none.is_all_items_available(item_codes={1, 2}, facility_id=0) + # - calling for empty set of equipment (should always be available) + assert eq_none.is_all_items_available(item_codes=set(), facility_id=0) + + # -- when using `all` availability behaviour: everything should not be available! + eq_all = Equipment( + catalogue=catalogue, + data_availability=data_availability, + rng=np.random.RandomState(np.random.MT19937(np.random.SeedSequence(seed))), + availability="all", + master_facilities_list=mfl, + ) + # - calling when all items available (should be true) + assert eq_all.is_all_items_available(item_codes={1, 2}, facility_id=1) + # - calling when no items available (should be true because using 'all' behaviour) + assert eq_all.is_all_items_available(item_codes={0, 2}, facility_id=0) + # - calling when some items available (should be true because using 'all' behaviour) + assert eq_all.is_all_items_available(item_codes={1, 2}, facility_id=0) + # - calling for empty set of equipment (should always be available) + assert eq_all.is_all_items_available(item_codes=set(), facility_id=0) + + # Check recording use of equipment + # - Add records, using calls with integers and list to different facility_id + eq_default.record_use_of_equipment(item_codes={1}, facility_id=0) + eq_default.record_use_of_equipment(item_codes={0, 1}, facility_id=0) + eq_default.record_use_of_equipment(item_codes={0, 1}, facility_id=1) + # - Check that internal record is as expected + assert {0: {0: 1, 1: 2}, 1: {0: 1, 1: 1}} == dict(eq_default._record_of_equipment_used_by_facility_id) + + # Lookup the item_codes that belong in a particular package. + # - When package is recognised + assert {2, 3} == eq_default.lookup_item_codes_from_pkg_name(pkg_name='PkgWith2+3') # these items are in the same + # package + # - Error thrown when package is not recognised + with pytest.raises(ValueError): + eq_default.lookup_item_codes_from_pkg_name(pkg_name='') + + + +equipment_item_code_that_is_available = [0, 1, ] +equipment_item_code_that_is_not_available = [2, 3,] + +def run_simulation_and_return_log( + seed, tmpdir, equipment_in_init, equipment_in_apply +) -> Dict: + """Returns the parsed logs from `tlo.methods.healthsystem.summary` from a + simulation object in which a single event has been run with the specified equipment usage. The + availability of equipment has been manipulated so that the item_codes given in + `equipment_item_code_that_is_available` and `equipment_item_code_that_is_not_available` are as expected. """ + + class DummyHSIEvent(HSI_Event, IndividualScopeEventMixin): + def __init__( + self, + module, + person_id, + level, + essential_equipment, + other_equipment, + ): + super().__init__(module, person_id=person_id) + self.TREATMENT_ID = "DummyHSIEvent" + self.EXPECTED_APPT_FOOTPRINT = self.make_appt_footprint({}) + self.ACCEPTED_FACILITY_LEVEL = level + self.add_equipment(essential_equipment) # Declaration at init will mean that these items are considered + # essential. + self._other_equipment = other_equipment + + def apply(self, person_id, squeeze_factor): + if self._other_equipment is not None: + self.add_equipment(self._other_equipment) + + class DummyModule(Module): + METADATA = {Metadata.DISEASE_MODULE, Metadata.USES_HEALTHSYSTEM} + + def __init__(self, essential_equipment, other_equipment, name=None): + super().__init__(name) + self.essential_equipment = essential_equipment + self.other_equipment = other_equipment + + def read_parameters(self, data_folder): + pass + + def initialise_population(self, population): + pass + + def initialise_simulation(self, sim): + # Schedule the HSI_Event to occur on the first day of the simulation + sim.modules["HealthSystem"].schedule_hsi_event( + hsi_event=DummyHSIEvent( + person_id=0, + level="2", + module=sim.modules["DummyModule"], + essential_equipment=self.essential_equipment, + other_equipment=self.other_equipment, + ), + do_hsi_event_checks=False, + topen=sim.date, + tclose=None, + priority=0, + ) + + log_config = {"filename": "log", "directory": tmpdir} + sim = Simulation(start_date=Date(2010, 1, 1), seed=seed, log_config=log_config) + sim.register( + demography.Demography(resourcefilepath=resourcefilepath), + healthsystem.HealthSystem(resourcefilepath=resourcefilepath), + DummyModule( + essential_equipment=equipment_in_init, other_equipment=equipment_in_apply + ), + ) + + # Manipulate availability of equipment + df = sim.modules["HealthSystem"].parameters["equipment_availability_estimates"] + df.loc[df['Item_Code'].isin(equipment_item_code_that_is_available), 'Pr_Available'] = 1.0 + df.loc[df['Item_Code'].isin(equipment_item_code_that_is_not_available), 'Pr_Available'] = 0.0 + + # Run the simulation + sim.make_initial_population(n=100) + sim.simulate(end_date=sim.start_date + pd.DateOffset(months=1)) + + # Return the parsed log of `tlo.methods.healthsystem.summary` + return parse_log_file(sim.log_filepath)["tlo.methods.healthsystem.summary"] + + + +def test_equipment_use_is_logged(seed, tmpdir): + """Check that an HSI that after an HSI is run, the logs reflect the use of the equipment (and correctly record the + name of the HSI and the facility_level at which ran). + This is repeated for: + * An HSI that declares use of equipment during its `apply` method (but no essential equipment); + * An HSI that declare use of essential equipment but nothing in its `apply` method`; + * An HSI that declare use of essential equipment and equipment during its `apply` method; + * An HSI that declares not use of any equipment. + """ + the_item_code = equipment_item_code_that_is_available[0] + another_item_code = equipment_item_code_that_is_available[1] + + def all_equipment_ever_used(log: Dict) -> set: + """With the log of equipment used in the simulation, return a set of the equipment item that have been used + (at any facility).""" + s = set() + for i in log["EquipmentEverUsed_ByFacilityID"]['EquipmentEverUsed']: + s.update(eval(i)) + return s + + # * An HSI that declares no use of any equipment (logs should be empty). + assert set() == all_equipment_ever_used( + run_simulation_and_return_log( + seed=seed, + tmpdir=tmpdir, + equipment_in_init=set(), + equipment_in_apply=set(), + ) + ) + + # * An HSI that declares use of equipment only during its `apply` method. + assert {the_item_code} == all_equipment_ever_used( + run_simulation_and_return_log( + seed=seed, + tmpdir=tmpdir, + equipment_in_init=set(), + equipment_in_apply=the_item_code, + ) + ) + + # * An HSI that declare use of equipment only in its `__init__` method + assert {the_item_code} == all_equipment_ever_used( + run_simulation_and_return_log( + seed=seed, + tmpdir=tmpdir, + equipment_in_init=the_item_code, + equipment_in_apply=set(), + ) + ) + + # * An HSI that declare use of equipment in `__init__` _and_ `apply`. + assert {the_item_code, another_item_code} == all_equipment_ever_used( + run_simulation_and_return_log( + seed=seed, + tmpdir=tmpdir, + equipment_in_init=the_item_code, + equipment_in_apply=another_item_code, + ) + ) + + +def test_hsi_does_not_run_if_equipment_declared_in_init_is_not_available(seed, tmpdir): + """Check that an HSI which declares an item of equipment that is declared in the HSI_Event's __init__ does run if + that item is available and does not run if that item is not available.""" + + def did_the_hsi_run(log: Dict) -> bool: + """Read the log to work out if the `DummyHSIEvent` ran or not.""" + it_did_run = len(log['hsi_event_counts'].iloc[0]['hsi_event_key_to_counts']) > 0 + it_did_not_run = len(log['never_ran_hsi_event_counts'].iloc[0]['never_ran_hsi_event_key_to_counts']) > 0 + + # Check that there if it did not run, it has had its "never_ran" function called + assert it_did_run != it_did_not_run + + # Return indication of whether it did run + return it_did_run + + + # HSI_Event that requires equipment that is available --> will run + assert did_the_hsi_run( + run_simulation_and_return_log( + seed=seed, + tmpdir=tmpdir, + equipment_in_init=equipment_item_code_that_is_available, + equipment_in_apply=set(), + ) + ) + + # HSI_Event that requires equipment that is not available --> will not run + assert not did_the_hsi_run( + run_simulation_and_return_log( + seed=seed, + tmpdir=tmpdir, + equipment_in_init=equipment_item_code_that_is_not_available, + equipment_in_apply=set(), + ) + ) + + +def test_change_equipment_availability(seed): + """Test that we can change the probability of the availability of equipment midway through the simulation.""" + # Set-up simulation that starts with `all` availability and then changes to `none` after one year. In the + # simulation a DummyModule schedules a DummyHSI that runs every month and tries to get a piece of equipment; + # then, check that the probability that this piece of equipment is available each month during the simulation. + + class DummyHSIEvent(HSI_Event, IndividualScopeEventMixin): + def __init__( + self, + module, + person_id, + ): + super().__init__(module, person_id=person_id) + self.TREATMENT_ID = "DummyHSIEvent" + self.EXPECTED_APPT_FOOTPRINT = self.make_appt_footprint({}) + self.ACCEPTED_FACILITY_LEVEL = '1a' + self.store_of_equipment_checks = dict() + + def apply(self, person_id, squeeze_factor): + # Check availability of a piece of equipment, with item_code = 0 + self.store_of_equipment_checks.update( + { + self.sim.date: self.probability_all_equipment_available(item_codes={0}) + } + ) + + # Schedule recurrence of this event in one month's time + self.sim.modules["HealthSystem"].schedule_hsi_event( + hsi_event=self, + do_hsi_event_checks=False, + topen=self.sim.date + pd.DateOffset(months=1), + tclose=None, + priority=0, + ) + + class DummyModule(Module): + METADATA = {Metadata.DISEASE_MODULE, Metadata.USES_HEALTHSYSTEM} + + def read_parameters(self, data_folder): + pass + + def initialise_population(self, population): + pass + + def initialise_simulation(self, sim): + # Schedule the HSI_Event to occur on the first day of the simulation (it will schedule its own repeats) + self.the_hsi_event = DummyHSIEvent(person_id=0, module=self) + + sim.modules["HealthSystem"].schedule_hsi_event( + hsi_event=self.the_hsi_event, + do_hsi_event_checks=False, + topen=sim.date, + tclose=None, + priority=0, + ) + + sim = Simulation(start_date=Date(2010, 1, 1), seed=seed) + sim.register( + demography.Demography(resourcefilepath=resourcefilepath), + healthsystem.HealthSystem(resourcefilepath=resourcefilepath), + DummyModule(), + ) + # Modify the parameters of the healthsystem to effect a change in the availability of equipment + sim.modules['HealthSystem'].parameters['equip_availability'] = 'all' + sim.modules['HealthSystem'].parameters['equip_availability_postSwitch'] = 'none' + sim.modules['HealthSystem'].parameters['year_equip_availability_switch'] = 2011 + + sim.make_initial_population(n=100) + sim.simulate(end_date=sim.start_date + pd.DateOffset(years=2)) + + # Get store & check for availabilities of the equipment + log = pd.Series(sim.modules['DummyModule'].the_hsi_event.store_of_equipment_checks) + assert (1.0 == log[log.index < Date(2011, 1, 1)]).all() + assert (0.0 == log[log.index >= Date(2011, 1, 1)]).all() + + +def test_logging_of_equipment_from_multiple_hsi(seed, tmpdir): + """Test that we correctly capture in the log the equipment declared by different HSI_Events that run at different + levels.""" + + item_code_needed_at_each_level = { + '0': set({0}), '1a': set({10}), '2': set({30}), '3': set({44}), '4': set() + } + + class DummyHSIEvent(HSI_Event, IndividualScopeEventMixin): + def __init__( + self, + module, + person_id, + level, + equipment_item_code + ): + super().__init__(module, person_id=person_id) + self.TREATMENT_ID = f"DummyHSIEvent_Level:{level}" + self.EXPECTED_APPT_FOOTPRINT = self.make_appt_footprint({}) + self.ACCEPTED_FACILITY_LEVEL = level + self.add_equipment(equipment_item_code) + + def apply(self, person_id, squeeze_factor): + pass + + class DummyModule(Module): + METADATA = {Metadata.DISEASE_MODULE, Metadata.USES_HEALTHSYSTEM} + + def read_parameters(self, data_folder): + pass + + def initialise_population(self, population): + pass + + def initialise_simulation(self, sim): + # Schedule the HSI_Events to occur, with the level determining the item_code used + for level, item_code in item_code_needed_at_each_level.items(): + sim.modules["HealthSystem"].schedule_hsi_event( + hsi_event=DummyHSIEvent(person_id=0, module=self, level=level, equipment_item_code=item_code), + do_hsi_event_checks=False, + topen=sim.date, + tclose=None, + priority=0, + ) + + log_config = {"filename": "log", "directory": tmpdir} + sim = Simulation(start_date=Date(2010, 1, 1), seed=seed, log_config=log_config) + sim.register( + demography.Demography(resourcefilepath=resourcefilepath), + healthsystem.HealthSystem(resourcefilepath=resourcefilepath), + DummyModule(), + ) + sim.make_initial_population(n=100) + sim.simulate(end_date=sim.start_date + pd.DateOffset(days=1)) + + # Read log to find what equipment used + df = parse_log_file(sim.log_filepath)["tlo.methods.healthsystem.summary"]['EquipmentEverUsed_ByFacilityID'] + df = df.drop(index=df.index[~df['Facility_Level'].isin(item_code_needed_at_each_level.keys())]) + df['EquipmentEverUsed'] = df['EquipmentEverUsed'].apply(eval).apply(list) + + # Check that equipment used at each level matches expectations + assert item_code_needed_at_each_level == df.groupby('Facility_Level')['EquipmentEverUsed'].sum().apply(set).to_dict() diff --git a/tests/test_healthsystem.py b/tests/test_healthsystem.py index 5c52a907e4..91adb7bea1 100644 --- a/tests/test_healthsystem.py +++ b/tests/test_healthsystem.py @@ -948,7 +948,7 @@ def apply(self, person_id, squeeze_factor): detailed_consumables = log["tlo.methods.healthsystem"]['Consumables'] assert {'date', 'TREATMENT_ID', 'did_run', 'Squeeze_Factor', 'priority', 'Number_By_Appt_Type_Code', 'Person_ID', - 'Facility_Level', 'Facility_ID', 'Event_Name', + 'Facility_Level', 'Facility_ID', 'Event_Name', 'Equipment' } == set(detailed_hsi_event.columns) assert {'date', 'Frac_Time_Used_Overall', 'Frac_Time_Used_By_Facility_ID', 'Frac_Time_Used_By_OfficerType', } == set(detailed_capacity.columns) @@ -1346,6 +1346,7 @@ def test_HealthSystemChangeParameters(seed, tmpdir): 'capabilities_coefficient': 0.5, 'cons_availability': 'all', 'beds_availability': 'default', + 'equip_availability': 'default', } new_parameters = { 'mode_appt_constraints': 2, @@ -1353,6 +1354,7 @@ def test_HealthSystemChangeParameters(seed, tmpdir): 'capabilities_coefficient': 1.0, 'cons_availability': 'none', 'beds_availability': 'none', + 'equip_availability': 'all', } class CheckHealthSystemParameters(RegularEvent, PopulationScopeEventMixin): @@ -1368,6 +1370,7 @@ def apply(self, population): _params['capabilities_coefficient'] = hs.capabilities_coefficient _params['cons_availability'] = hs.consumables.cons_availability _params['beds_availability'] = hs.bed_days.availability + _params['equip_availability'] = hs.equipment.availability logger = logging.getLogger('tlo.methods.healthsystem') logger.info(key='CheckHealthSystemParameters', data=_params)