Skip to content

Commit

Permalink
added return statements and footprints and as treatment_id for Xrays,…
Browse files Browse the repository at this point in the history 
… culture, clinical diagnosis not printed
  • Loading branch information
@nchagoma503
nchagoma503 committed Jan 29, 2025
1 parent b60b7a6 commit f48132c
Show file tree
Hide file tree
Showing 2 changed files with 119 additions and 83 deletions.
38 changes: 19 additions & 19 deletions src/scripts/hiv/DAH/analysis_tb_DAH10x.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -36,7 +36,7 @@

#outputfilepath = Path("./outputs")

results_folder = get_scenario_outputs('tb_DAH_scenarios10x-2025-01-29T082948Z', outputfilepath) [-1]
results_folder = get_scenario_outputs('tb_DAH_scenarios10x-2025-01-29T090853Z', outputfilepath) [-1]
log = load_pickled_dataframes(results_folder)
info = get_scenario_info(results_folder)
print(info)
Expand Down Expand Up @@ -158,24 +158,24 @@ def set_param_names_as_column_index_level_0(_df):


#Extracting DALYs
def get_tb_dalys(df_):
# Get DALYs of TB
years = df_['year'].unique() # Get unique years
dalys = pd.Series(dtype='float64', index=years)

for year in years:
# Group data by year and sum relevant columns
tot_dalys = df_.drop(columns='date').groupby('year').sum()

# Ensure the labels exist before summing
if any(label in tot_dalys.columns for label in ["AIDS_TB", "TB", "AIDS_non_TB"]):
# Sum the DALYs for the specified labels for the year
dalys[year] = tot_dalys.loc[year, ["AIDS_TB", "TB", "AIDS_non_TB"]].sum()
else:
dalys[year] = 0 # Set it to 0 if the labels are not found

dalys.sort_index(inplace=True) # Sort the index inplace
return dalys
# def get_tb_dalys(df_):
# # Get DALYs of TB
# years = df_['year'].unique() # Get unique years
# dalys = pd.Series(dtype='float64', index=years)
#
# for year in years:
# # Group data by year and sum relevant columns
# tot_dalys = df_.drop(columns='date').groupby('year').sum()
#
# # Ensure the labels exist before summing
# if any(label in tot_dalys.columns for label in ["AIDS_TB", "TB", "AIDS_non_TB"]):
# # Sum the DALYs for the specified labels for the year
# dalys[year] = tot_dalys.loc[year, ["AIDS_TB", "TB", "AIDS_non_TB"]].sum()
# else:
# dalys[year] = 0 # Set it to 0 if the labels are not found
#
# dalys.sort_index(inplace=True) # Sort the index inplace
# return dalys


def get_tb_dalys(df_):
Expand Down
164 changes: 100 additions & 64 deletions src/tlo/methods/tb.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -976,7 +976,8 @@ def initialise_simulation(self, sim):

# 2) log at the end of the year
# Optional: Schedule the scale-up of programs
if self.parameters["type_of_scaleup"] != 'none':
#changed scale-up to none
if self.parameters["type_of_scaleup"] == 'none':
scaleup_start_date = Date(self.parameters["scaleup_start_year"], 1, 1)
assert scaleup_start_date >= self.sim.start_date, f"Date {scaleup_start_date} is before simulation starts."
sim.schedule_event(TbScaleUpEvent(self), scaleup_start_date)
Expand Down Expand Up @@ -1006,7 +1007,7 @@ def update_parameters_for_program_scaleup(self):
parse_csv_values_for_columns_with_mixed_datatypes
)

if p['type_of_scaleup'] == 'target':
if p['type_of_scaleup'] == 'none':
scaled_params = scaled_params_workbook.set_index('parameter')['target_value'].to_dict()
else:
scaled_params = scaled_params_workbook.set_index('parameter')['max_value'].to_dict()
Expand Down Expand Up @@ -1502,40 +1503,57 @@ def apply(self, population):
self.sim.modules['HealthSystem'].override_availability_of_consumables({175: 0.51})
self.sim.modules['HealthSystem'].override_availability_of_consumables({187: 0.85})
self.sim.modules["Tb"].parameters["probability_community_chest_xray"] = 0.0
self.sim.modules["Tb"].parameters["first_line_test"] = 'sputum'
self.sim.modules["Tb"].parameters["second_line_test"] = 'xpert'
return

# sets availability of xpert to nil
if scenario == 1:
self.sim.modules['HealthSystem'].override_availability_of_consumables({175: 0.51})
self.sim.modules['HealthSystem'].override_availability_of_consumables({187: 0.0})
self.sim.modules["Tb"].parameters["probability_community_chest_xray"] = 0.0
self.sim.modules["Tb"].parameters["first_line_test"] = 'sputum'
self.sim.modules["Tb"].parameters["second_line_test"] = 'sputum'

# sets availability of xray to nil
if scenario == 2:
self.sim.modules['HealthSystem'].override_availability_of_consumables({175: 0.0})
self.sim.modules['HealthSystem'].override_availability_of_consumables({187: 0.85})
self.sim.modules["Tb"].parameters["probability_community_chest_xray"] = 0.0
self.sim.modules["Tb"].parameters["first_line_test"] = 'sputum'
self.sim.modules["Tb"].parameters["second_line_test"] = 'xpert'

#increases probability of accessing chest xray by 100% always available
if scenario == 3:
self.sim.modules['HealthSystem'].override_availability_of_consumables({175: 1.0})
self.sim.modules['HealthSystem'].override_availability_of_consumables({187: 0.85})
self.sim.modules["Tb"].parameters["probability_community_chest_xray"] = 0.0
self.sim.modules["Tb"].parameters["first_line_test"] = 'sputum'
self.sim.modules["Tb"].parameters["second_line_test"] = 'xpert'


# Introduce community Xray
if scenario == 4:
self.sim.modules["Tb"].parameters["probability_community_chest_xray"]=0.15
self.sim.modules['HealthSystem'].override_availability_of_consumables({175: 0.51})
self.sim.modules['HealthSystem'].override_availability_of_consumables({187: 0.85})

class TbTreatmentAndRelapseEvents(RegularEvent, PopulationScopeEventMixin):
""" This event runs each month and calls three functions:
* scheduling TB screening for the general population
* ending treatment if the end of treatment regimen has been reached
* determining who will relapse after a primary infection
"""
def apply(self, population):
self.module.relapse_event(population)
self.sim.modules["Tb"].parameters["first_line_test"] = 'sputum'
self.sim.modules["Tb"].parameters["second_line_test"] = 'xpert'

# class TbTreatmentAndRelapseEvents(RegularEvent, PopulationScopeEventMixin):
# """ This event runs each month and calls three functions:
# * scheduling TB screening for the general population
# * ending treatment if the end of treatment regimen has been reached
# * determining who will relapse after a primary infection
# """
# def apply(self, population):
# self.module.relapse_event(population)
#
# def apply(self, population):
# # schedule some background rates of tb testing (non-symptom-driven)
# self.module.send_for_screening_general(population)
#
# self.module.end_treatment(population)
# self.module.relapse_event(population)


class TbActiveCasePoll(RegularEvent, PopulationScopeEventMixin):
Expand Down Expand Up @@ -2178,6 +2196,7 @@ def apply(self, person_id, squeeze_factor):
test_result = self.sim.modules["HealthSystem"].dx_manager.run_dx_test(
dx_tests_to_run="tb_clinical", hsi_event=self
)
ACTUAL_APPT_FOOTPRINT = self.make_appt_footprint({"Under5OPD": 0.5})

# if clinical diagnosis returns positive result, refer for appropriate treatment
if test_result:
Expand All @@ -2188,7 +2207,7 @@ def apply(self, person_id, squeeze_factor):
key="message",
data=f"schedule HSI_Tb_StartTreatment for person {person_id}",
)
# ACTUAL_APPT_FOOTPRINT = self.EXPECTED_APPT_FOOTPRINT


self.sim.modules["HealthSystem"].schedule_hsi_event(
HSI_Tb_StartTreatment(
Expand All @@ -2199,36 +2218,36 @@ def apply(self, person_id, squeeze_factor):
priority=0,
)


# ------------------------- Culture testing if program scale-up ------------------------- #
# under program scale-up, if a person tests negative but still has all symptoms
# indicative of TB, they are referred for culture test which has perfect sensitivity
# this has the effect to reduce false negatives
person_has_tb_symptoms = all(symptom in persons_symptoms for symptom in self.module.symptom_list)

if not test_result and person_has_tb_symptoms:
if p['type_of_scaleup'] != 'none' and self.sim.date.year >= p['scaleup_start_year']:

logger.debug(
key="message",
data=f"HSI_Tb_ClinicalDiagnosis: scheduling culture for person {person_id}",
)

culture_event = HSI_Tb_Culture(
self.module, person_id=person_id
)
self.sim.modules["HealthSystem"].schedule_hsi_event(
culture_event,
priority=0,
topen=now,
tclose=None,
)

# # Return the footprint. If it should be suppressed, return a blank footprint.
# if self.suppress_footprint:
# return self.make_appt_footprint({})
# else:
# return ACTUAL_APPT_FOOTPRINT
#
# # ------------------------- Culture testing if program scale-up ------------------------- #
# # under program scale-up, if a person tests negative but still has all symptoms
# # indicative of TB, they are referred for culture test which has perfect sensitivity
# # this has the effect to reduce false negatives
# person_has_tb_symptoms = all(symptom in persons_symptoms for symptom in self.module.symptom_list)
#
# if not test_result and person_has_tb_symptoms:
# if p['type_of_scaleup'] != 'none' and self.sim.date.year >= p['scaleup_start_year']:
#
# logger.debug(
# key="message",
# data=f"HSI_Tb_ClinicalDiagnosis: scheduling culture for person {person_id}",
# )
#
# culture_event = HSI_Tb_Culture(
# self.module, person_id=person_id
# )
# self.sim.modules["HealthSystem"].schedule_hsi_event(
# culture_event,
# priority=0,
# topen=now,
# tclose=None,
# )

# Return the footprint. If it should be suppressed, return a blank footprint.
if self.suppress_footprint:
return self.make_appt_footprint({})
else:
return ACTUAL_APPT_FOOTPRINT

class HSI_Tb_Culture(HSI_Event, IndividualScopeEventMixin):
"""
Expand Down Expand Up @@ -2270,7 +2289,7 @@ def apply(self, person_id, squeeze_factor):
df.at[person_id, "tb_diagnosed"] = True
df.at[person_id, "tb_date_diagnosed"] = self.sim.date

#ACTUAL_APPT_FOOTPRINT = self.EXPECTED_APPT_FOOTPRINT
ACTUAL_APPT_FOOTPRINT = self.make_appt_footprint({"LabTBMicro": 1})

self.sim.modules["HealthSystem"].schedule_hsi_event(
HSI_Tb_StartTreatment(
Expand All @@ -2281,11 +2300,11 @@ def apply(self, person_id, squeeze_factor):
priority=0,
)

# # Return the footprint. If it should be suppressed, return a blank footprint.
# if self.suppress_footprint:
# return self.make_appt_footprint({})
# else:
# return ACTUAL_APPT_FOOTPRINT
# Return the footprint. If it should be suppressed, return a blank footprint.
if self.suppress_footprint:
return self.make_appt_footprint({})
else:
return ACTUAL_APPT_FOOTPRINT

class HSI_Tb_Xray_level1b(HSI_Event, IndividualScopeEventMixin):
"""
Expand Down Expand Up @@ -2313,6 +2332,8 @@ def apply(self, person_id, squeeze_factor):
if not df.at[person_id, "is_alive"] or df.at[person_id, "tb_diagnosed"]:
return self.sim.modules["HealthSystem"].get_blank_appt_footprint()

ACTUAL_APPT_FOOTPRINT = self.EXPECTED_APPT_FOOTPRINT

smear_status = df.at[person_id, "tb_smear"]

# select sensitivity/specificity of test based on smear status
Expand All @@ -2324,14 +2345,17 @@ def apply(self, person_id, squeeze_factor):
test_result = self.sim.modules["HealthSystem"].dx_manager.run_dx_test(
dx_tests_to_run="tb_xray_smear_negative", hsi_event=self
)

if test_result is not None:
self.add_equipment(self.healthcare_system.equipment.from_pkg_names('X-ray'))

ACTUAL_APPT_FOOTPRINT = self.make_appt_footprint({"DiagRadio": 1})

# if consumables not available, refer to level 2
# return blank footprint as xray did not occur
if test_result is None:

# ACTUAL_APPT_FOOTPRINT = self.make_appt_footprint({})
ACTUAL_APPT_FOOTPRINT = self.make_appt_footprint({})

self.sim.modules["HealthSystem"].schedule_hsi_event(
HSI_Tb_Xray_level2(person_id=person_id, module=self.module),
Expand All @@ -2345,7 +2369,11 @@ def apply(self, person_id, squeeze_factor):
df.at[person_id, "tb_diagnosed"] = True
df.at[person_id, "tb_date_diagnosed"] = self.sim.date

ACTUAL_APPT_FOOTPRINT = self.EXPECTED_APPT_FOOTPRINT
# Log the diagnosis event
logger.debug(
key="message",
data=f"Person {person_id} diagnosed with TB via {self.TREATMENT_ID}. Scheduling treatment."
)

self.sim.modules["HealthSystem"].schedule_hsi_event(
HSI_Tb_StartTreatment(
Expand All @@ -2356,11 +2384,11 @@ def apply(self, person_id, squeeze_factor):
priority=0,
)

# # Return the footprint. If it should be suppressed, return a blank footprint.
# if self.suppress_footprint:
# return self.make_appt_footprint({})
# else:
# return ACTUAL_APPT_FOOTPRINT
# Return the footprint. If it should be suppressed, return a blank footprint.
if self.suppress_footprint:
return self.make_appt_footprint({})
else:
return ACTUAL_APPT_FOOTPRINT


class HSI_Tb_Xray_level2(HSI_Event, IndividualScopeEventMixin):
Expand Down Expand Up @@ -2388,7 +2416,7 @@ def apply(self, person_id, squeeze_factor):
if not df.at[person_id, "is_alive"] or df.at[person_id, "tb_diagnosed"]:
return self.sim.modules["HealthSystem"].get_blank_appt_footprint()

# ACTUAL_APPT_FOOTPRINT = self.EXPECTED_APPT_FOOTPRINT
ACTUAL_APPT_FOOTPRINT = self.EXPECTED_APPT_FOOTPRINT

smear_status = df.at[person_id, "tb_smear"]

Expand All @@ -2404,10 +2432,14 @@ def apply(self, person_id, squeeze_factor):
if test_result is not None:
self.add_equipment(self.healthcare_system.equipment.from_pkg_names('X-ray'))

ACTUAL_APPT_FOOTPRINT = self.make_appt_footprint({"DiagRadio": 1})

# if consumables not available, rely on clinical diagnosis
# return blank footprint as xray was not available
if test_result is None:

ACTUAL_APPT_FOOTPRINT = self.make_appt_footprint({})

self.sim.modules["HealthSystem"].schedule_hsi_event(
HSI_Tb_ClinicalDiagnosis(person_id=person_id, module=self.module),
topen=self.sim.date,
Expand All @@ -2420,7 +2452,7 @@ def apply(self, person_id, squeeze_factor):
df.at[person_id, "tb_diagnosed"] = True
df.at[person_id, "tb_date_diagnosed"] = self.sim.date

ACTUAL_APPT_FOOTPRINT = self.make_appt_footprint({})


self.sim.modules["HealthSystem"].schedule_hsi_event(
HSI_Tb_StartTreatment(
Expand All @@ -2431,11 +2463,11 @@ def apply(self, person_id, squeeze_factor):
priority=0,
)

# # Return the footprint. If it should be suppressed, return a blank footprint.
# if self.suppress_footprint:
# return self.make_appt_footprint({})
# else:
# return ACTUAL_APPT_FOOTPRINT
# Return the footprint. If it should be suppressed, return a blank footprint.
if self.suppress_footprint:
return self.make_appt_footprint({})
else:
return ACTUAL_APPT_FOOTPRINT


# # ---------------------------------------------------------------------------
Expand Down Expand Up @@ -2929,12 +2961,14 @@ def __init__(self, module, person_id, suppress_footprint=False):

def apply(self, person_id, squeeze_factor):

#print(f'"STARTING COMMUNITY CHEST XRAY SCREENING"')
print(f'"STARTING COMMUNITY CHEST XRAY SCREENING"')

logger.debug(key="message", data=f"Performing community chest X-ray screening for {person_id}")
df = self.sim.population.props # Shortcut to the dataframe
person = df.loc[person_id]

ACTUAL_APPT_FOOTPRINT = self.EXPECTED_APPT_FOOTPRINT

smear_status = person['tb_smear']

# Perform the X-ray and decide the result
Expand All @@ -2950,6 +2984,8 @@ def apply(self, person_id, squeeze_factor):
dx_tests_to_run="tb_xray_smear_negative", hsi_event=self
)

ACTUAL_APPOINTMENT=self.make_appt_footprint({"ConWithDCSA": 1})

# If the test returns a positive result, refer for appropriate treatment
if test_result:
df.at[person_id, "tb_diagnosed"] = True
Expand Down

0 comments on commit f48132c

Please sign in to comment.