Source code for tlo.methods.cardio_metabolic_disorders
"""
The joint Cardio-Metabolic Disorders model determines onset, outcome and treatment of:
* Diabetes
* Hypertension
* Chronic Kidney Disease
* Chronic Ischemic Heart Disease
* Stroke
* Heart Attack
And:
* Chronic Lower Back Pain
"""
import math
from itertools import combinations
from pathlib import Path
from typing import List
import numpy as np
import pandas as pd
from tlo import DAYS_IN_YEAR, DateOffset, Module, Parameter, Property, Types, logging
from tlo.events import Event, IndividualScopeEventMixin, PopulationScopeEventMixin, RegularEvent
from tlo.lm import LinearModel, LinearModelType, Predictor
from tlo.methods import Metadata
from tlo.methods import demography as de
from tlo.methods.causes import Cause
from tlo.methods.dxmanager import DxTest
from tlo.methods.healthsystem import HSI_Event
from tlo.methods.symptommanager import Symptom
from tlo.util import random_date
# ---------------------------------------------------------------------------------------------------------
# MODULE DEFINITIONS
# ---------------------------------------------------------------------------------------------------------
logger = logging.getLogger(__name__)
logger.setLevel(logging.INFO)
[docs]
class CardioMetabolicDisorders(Module):
"""
CardioMetabolicDisorders module covers a subset of cardio-metabolic conditions and events. Conditions are binary
and individuals experience a risk of acquiring or losing a condition based on annual probability and
demographic/lifestyle risk factors.
"""
# Save a master list of the events that are covered in this module
conditions = ['diabetes',
'hypertension',
'chronic_kidney_disease',
'chronic_lower_back_pain',
'chronic_ischemic_hd']
# Save a master list of the events that are covered in this module
events = ['ever_stroke',
'ever_heart_attack']
INIT_DEPENDENCIES = {'Demography', 'Lifestyle', 'HealthSystem', 'SymptomManager'}
OPTIONAL_INIT_DEPENDENCIES = {'HealthBurden'}
ADDITIONAL_DEPENDENCIES = {'Depression'}
# Declare Metadata
METADATA = {
Metadata.DISEASE_MODULE,
Metadata.USES_SYMPTOMMANAGER,
Metadata.USES_HEALTHSYSTEM,
Metadata.USES_HEALTHBURDEN
}
# Declare Causes of Death
CAUSES_OF_DEATH = {
'diabetes': Cause(
gbd_causes='Diabetes mellitus', label='Diabetes'),
'chronic_ischemic_hd': Cause(
gbd_causes={'Ischemic heart disease', 'Hypertensive heart disease'}, label='Heart Disease'),
'ever_heart_attack': Cause(
gbd_causes={'Ischemic heart disease', 'Hypertensive heart disease'}, label='Heart Disease'),
'ever_stroke': Cause(
gbd_causes='Stroke', label='Stroke'),
'chronic_kidney_disease': Cause(
gbd_causes='Chronic kidney disease', label='Kidney Disease')
}
# Declare Causes of Disability
CAUSES_OF_DISABILITY = {
'diabetes': Cause(
gbd_causes='Diabetes mellitus', label='Diabetes'),
'chronic_ischemic_hd': Cause(
gbd_causes={'Ischemic heart disease', 'Hypertensive heart disease'}, label='Heart Disease'),
'heart_attack': Cause(
gbd_causes={'Ischemic heart disease', 'Hypertensive heart disease'}, label='Heart Disease'),
'stroke': Cause(
gbd_causes='Stroke', label='Stroke'),
'chronic_kidney_disease': Cause(
gbd_causes='Chronic kidney disease', label='Kidney Disease'),
'lower_back_pain': Cause(
gbd_causes={'Low back pain'}, label='Lower Back Pain'
)
}
# Create separate dicts for params for conditions and events which are read in via excel documents in resources/cmd
onset_conditions_param_dicts = {
f"{p}_onset": Parameter(Types.DICT, f"all the parameters that specify the linear models for onset of {p}")
for p in conditions
}
removal_conditions_param_dicts = {
f"{p}_removal": Parameter(Types.DICT, f"all the parameters that specify the linear models for removal of {p}")
for p in conditions
}
hsi_conditions_param_dicts = {
f"{p}_hsi": Parameter(Types.DICT, f"all the parameters that specify diagnostic tests and treatments for {p}")
for p in conditions
}
onset_events_param_dicts = {
f"{p}_onset": Parameter(Types.DICT, f"all the parameters that specify the linear models for onset of {p}")
for p in events
}
hsi_events_param_dicts = {
f"{p}_hsi": Parameter(Types.DICT, f"all the parameters that specify diagnostic tests and treatments for {p}")
for p in events
}
death_conditions_param_dicts = {
f"{p}_death": Parameter(Types.DICT, f"all the parameters that specify the linear models for death from {p}")
for p in conditions
}
death_events_param_dicts = {
f"{p}_death": Parameter(Types.DICT, f"all the parameters that specify the linear models for death from {p}")
for p in events
}
initial_prev_param_dicts = {
f"{p}_initial_prev": Parameter(Types.DICT, 'initial prevalence of condition') for p in conditions
}
other_params_dict = {
'interval_between_polls': Parameter(Types.INT, 'months between the main polling event'),
'pr_bmi_reduction': Parameter(Types.INT, 'probability of an individual having a reduction in BMI following '
'weight loss treatment')
}
PARAMETERS = {
**onset_conditions_param_dicts, **removal_conditions_param_dicts, **hsi_conditions_param_dicts,
**onset_events_param_dicts, **death_conditions_param_dicts, **death_events_param_dicts,
**hsi_events_param_dicts, **initial_prev_param_dicts, **other_params_dict,
'prob_care_provided_given_seek_emergency_care': Parameter(
Types.REAL, "The probability that correct care is fully provided to persons that have sought emergency care"
" for a Cardio-metabolic disorder.")
}
# Convert conditions and events to dicts and merge together into PROPERTIES
condition_list = {
f"nc_{p}": Property(Types.BOOL, f"Whether or not someone has {p}") for p in conditions
}
condition_diagnosis_list = {
f"nc_{p}_ever_diagnosed": Property(Types.BOOL, f"Whether or not someone has ever been diagnosed with {p}") for p
in conditions
}
condition_date_diagnosis_list = {
f"nc_{p}_date_diagnosis": Property(Types.DATE, f"When someone has been diagnosed with {p}") for p
in conditions
}
condition_date_of_last_test_list = {
f"nc_{p}_date_last_test": Property(Types.DATE, f"When someone has last been tested for {p}") for p
in conditions
}
condition_medication_list = {
f"nc_{p}_on_medication": Property(Types.BOOL, f"Whether or not someone is on medication for {p}") for p
in conditions
}
condition_medication_death_list = {
f"nc_{p}_medication_prevents_death": Property(Types.BOOL, f"Whether or not medication (if provided) will "
f"prevent death from {p}") for p in conditions
}
event_list = {
f"nc_{p}": Property(Types.BOOL, f"Whether or not someone has had a {p}") for p in events}
event_date_last_list = {
f"nc_{p}_date_last_event": Property(Types.DATE, f"Date of last {p}") for p in events
}
event_diagnosis_list = {
f"nc_{p}_ever_diagnosed": Property(Types.BOOL, f"Whether or not someone has ever been diagnosed with {p}") for p
in events
}
event_date_diagnosis_list = {
f"nc_{p}_date_diagnosis": Property(Types.DATE, f"When someone has last been diagnosed with {p}") for p
in events}
event_medication_list = {
f"nc_{p}_on_medication": Property(Types.BOOL, f"Whether or not someone has ever been diagnosed with {p}") for p
in events
}
event_scheduled_date_death_list = {
f"nc_{p}_scheduled_date_death": Property(Types.DATE, f"Scheduled date of death from {p}") for p
in events
}
event_medication_death_list = {
f"nc_{p}_medication_prevents_death": Property(Types.BOOL, f"Whether or not medication will prevent death from "
f"{p}") for p in events
}
PROPERTIES = {**condition_list, **event_list, **condition_diagnosis_list, **condition_date_diagnosis_list,
**condition_date_of_last_test_list, **condition_medication_list, **condition_medication_death_list,
**event_date_last_list, **event_diagnosis_list, **event_date_diagnosis_list, **event_medication_list,
**event_medication_death_list, **event_scheduled_date_death_list,
'nc_ever_weight_loss_treatment': Property(Types.BOOL,
'whether or not the person has ever had weight loss '
'treatment'),
'nc_weight_loss_worked': Property(Types.BOOL,
'whether or not weight loss treatment worked'),
'nc_risk_score': Property(Types.INT, 'score to represent number of risk conditions the person has')
}
[docs]
def __init__(self, name=None, resourcefilepath=None, do_log_df: bool = False, do_condition_combos: bool = False):
super().__init__(name)
self.resourcefilepath = resourcefilepath
self.conditions = CardioMetabolicDisorders.conditions
self.events = CardioMetabolicDisorders.events
# Create list that includes the nc_ prefix for conditions in this module
self.condition_list = ['nc_' + cond for cond in CardioMetabolicDisorders.conditions]
# Store the symptoms that this module will use (for conditions only):
self.symptoms = {f"{s}_symptoms" for s in self.conditions if s != "hypertension"}
# Dict to hold the probability of onset of different types of symptom given a condition
self.prob_symptoms = dict()
# Retrieve age range categories from Demography module
self.age_cats = None
# Dict to hold the DALY weights
self.daly_wts = dict()
# Retrieve age range categories from Demography module
self.age_cats = None
# Store bools for whether or not to log the df or log combinations of co-morbidities
self.do_log_df = do_log_df
self.do_condition_combos = do_condition_combos
# Dict to hold trackers for counting events
self.trackers = None
# Dicts of linear models
self.lms_onset = dict()
self.lms_removal = dict()
self.lms_death = dict()
self.lms_symptoms = dict()
self.lms_testing = dict()
# Build the LinearModel for occurrence of events
self.lms_event_onset = dict()
self.lms_event_death = dict()
self.lms_event_symptoms = dict()
[docs]
def read_parameters(self, data_folder):
"""Read parameter values from files for condition onset, removal, deaths, and initial prevalence.
ResourceFile_cmd_condition_onset.xlsx = parameters for onset of conditions
ResourceFile_cmd_condition_removal.xlsx = parameters for removal of conditions
ResourceFile_cmd_condition_death.xlsx = parameters for death rate from conditions
ResourceFile_cmd_condition_prevalence.xlsx = initial and target prevalence for conditions
ResourceFile_cmd_condition_symptoms.xlsx = symptoms for conditions
ResourceFile_cmd_condition_hsi.xlsx = HSI parameters for conditions
ResourceFile_cmd_condition_testing.xlsx = community testing parameters for conditions (currently only
hypertension)
ResourceFile_cmd_events.xlsx = parameters for occurrence of events
ResourceFile_cmd_events_death.xlsx = parameters for death rate from events
ResourceFile_cmd_events_symptoms.xlsx = symptoms for events
ResourceFile_cmd_events_hsi.xlsx = HSI parameters for events
"""
cmd_path = Path(self.resourcefilepath) / "cmd"
cond_onset = pd.read_excel(cmd_path / "ResourceFile_cmd_condition_onset.xlsx", sheet_name=None)
cond_removal = pd.read_excel(cmd_path / "ResourceFile_cmd_condition_removal.xlsx", sheet_name=None)
cond_death = pd.read_excel(cmd_path / "ResourceFile_cmd_condition_death.xlsx", sheet_name=None)
cond_prevalence = pd.read_excel(cmd_path / "ResourceFile_cmd_condition_prevalence.xlsx", sheet_name=None)
cond_symptoms = pd.read_excel(cmd_path / "ResourceFile_cmd_condition_symptoms.xlsx", sheet_name=None)
cond_hsi = pd.read_excel(cmd_path / "ResourceFile_cmd_condition_hsi.xlsx", sheet_name=None)
cond_testing = pd.read_excel(cmd_path / "ResourceFile_cmd_condition_testing.xlsx", sheet_name=None)
events_onset = pd.read_excel(cmd_path / "ResourceFile_cmd_events.xlsx", sheet_name=None)
events_death = pd.read_excel(cmd_path / "ResourceFile_cmd_events_death.xlsx", sheet_name=None)
events_symptoms = pd.read_excel(cmd_path / "ResourceFile_cmd_events_symptoms.xlsx", sheet_name=None)
events_hsi = pd.read_excel(cmd_path / "ResourceFile_cmd_events_hsi.xlsx", sheet_name=None)
self.load_parameters_from_dataframe(pd.read_csv(cmd_path / "ResourceFile_cmd_parameters.csv"))
def get_values(params, value):
"""replaces nans in the 'value' key with specified value"""
params['value'] = params['value'].replace(np.nan, value)
params['value'] = params['value'].astype(float)
return params.set_index('parameter_name')['value']
p = self.parameters
for condition in self.conditions:
p[f'{condition}_onset'] = get_values(cond_onset[condition], 1)
p[f'{condition}_removal'] = get_values(cond_removal[condition], 1)
p[f'{condition}_death'] = get_values(cond_death[condition], 1)
p[f'{condition}_initial_prev'] = get_values(cond_prevalence[condition], 0)
p[f'{condition}_symptoms'] = get_values(cond_symptoms[condition], 1)
p[f'{condition}_hsi'] = get_values(cond_hsi[condition], 1)
p['hypertension_testing'] = get_values(cond_testing['hypertension'], 1)
for event in self.events:
p[f'{event}_onset'] = get_values(events_onset[event], 1)
p[f'{event}_death'] = get_values(events_death[event], 1)
p[f'{event}_symptoms'] = get_values(events_symptoms[event], 1)
p[f'{event}_hsi'] = get_values(events_hsi[event], 1)
# Set the interval (in months) between the polls
p['interval_between_polls'] = 3
# Set the probability of an individual losing weight in the CardioMetabolicDisordersWeightLossEvent (this value
# doesn't vary by condition)
p['pr_bmi_reduction'] = 0.1
# Check that every value has been read-in successfully
for param_name in self.PARAMETERS:
assert self.parameters[param_name] is not None, f'Parameter "{param_name}" has not been set.'
# -------------------------------------------- SYMPTOMS -------------------------------------------------------
# Retrieve symptom probabilities
for condition in self.conditions:
if not self.parameters[f'{condition}_symptoms'].empty:
self.prob_symptoms[condition] = self.parameters[f'{condition}_symptoms']
else:
self.prob_symptoms[condition] = {}
for event in self.events:
if not self.parameters[f'{event}_symptoms'].empty:
self.prob_symptoms[event] = self.parameters[f'{event}_symptoms']
else:
self.prob_symptoms[event] = {}
# Register symptoms for conditions and give non-generic symptom 'average' healthcare seeking, except for
# chronic lower back pain, which has no healthcare seeking in adults
for symptom_name in self.symptoms:
if symptom_name == "chronic_lower_back_pain_symptoms":
self.sim.modules['SymptomManager'].register_symptom(
Symptom(name=symptom_name,
no_healthcareseeking_in_adults=True)
)
else:
self.sim.modules['SymptomManager'].register_symptom(
Symptom(name=symptom_name,
odds_ratio_health_seeking_in_adults=1.0)
)
# Register symptoms from events and make them emergencies
for event in self.events:
self.sim.modules['SymptomManager'].register_symptom(
Symptom.emergency(
name=f'{event}_damage', which='adults'
),
)
[docs]
def initialise_population(self, population):
"""
Set property values for the initial population.
"""
self.age_cats = self.sim.modules['Demography'].AGE_RANGE_CATEGORIES
df = population.props
men = df.is_alive & (df.sex == 'M')
women = df.is_alive & (df.sex == 'F')
def sample_eligible(_filter, _p, _condition):
"""uses filter to get eligible population and samples individuals for condition using p"""
eligible = df.index[_filter]
init_prev = self.rng.choice([True, False], size=len(eligible), p=[_p, 1 - _p])
if sum(init_prev):
df.loc[eligible[init_prev], f'nc_{_condition}'] = True
def sample_eligible_diagnosis_medication(_filter, _p, _condition):
"""uses filter to get eligible population and samples individuals for prior diagnosis & medication use
using p"""
eligible = df.index[_filter]
init_diagnosis = self.rng.choice([True, False], size=len(eligible), p=[_p, 1 - _p])
if sum(init_diagnosis):
df.loc[eligible[init_diagnosis], f'nc_{_condition}_ever_diagnosed'] = True
df.loc[eligible[init_diagnosis], f'nc_{_condition}_date_diagnosis'] = self.sim.date
df.loc[eligible[init_diagnosis], f'nc_{_condition}_date_last_test'] = self.sim.date
df.loc[eligible[init_diagnosis], f'nc_{_condition}_on_medication'] = True
def sample_eligible_treatment_success(_filter, _p, _condition):
"""uses filter to get eligible population and samples individuals for prior diagnosis & medication use
using p"""
eligible = df.index[_filter]
init_treatment_works = self.rng.choice([True, False], size=len(eligible), p=[_p, 1 - _p])
if sum(init_treatment_works):
df.loc[eligible[init_treatment_works], f'nc_{_condition}_medication_prevents_death'] = True
for condition in self.conditions:
p = self.parameters[f'{condition}_initial_prev']
# Men & women without condition
men_wo_cond = men & ~df[f'nc_{condition}']
women_wo_cond = women & ~df[f'nc_{condition}']
for _age_range in self.age_cats:
# Select all eligible individuals (men & women w/o condition and in age range)
sample_eligible(men_wo_cond & (df.age_range == _age_range), p[f'm_{_age_range}'], condition)
sample_eligible(women_wo_cond & (df.age_range == _age_range), p[f'f_{_age_range}'], condition)
# ----- Set variables to false / NaT for everyone
df.loc[df.is_alive, f'nc_{condition}_date_last_test'] = pd.NaT
df.loc[df.is_alive, f'nc_{condition}_ever_diagnosed'] = False
df.loc[df.is_alive, f'nc_{condition}_date_diagnosis'] = pd.NaT
df.loc[df.is_alive, f'nc_{condition}_on_medication'] = False
df.loc[df.is_alive, f'nc_{condition}_medication_prevents_death'] = False
# ----- Sample among eligible population who have condition to set initial proportion diagnosed & on
# medication
p_initial_diagnosis = self.parameters[f'{condition}_hsi']['pr_diagnosed']
# Population with condition
w_cond = df[f'nc_{condition}']
sample_eligible_diagnosis_medication(w_cond, p_initial_diagnosis, condition)
# For those on medication, sample to set initial proportion for whom medication prevents death
p_treatment_works = self.parameters[f'{condition}_hsi']['pr_treatment_works']
# Population already on medication
on_med = df[f'nc_{condition}_on_medication']
sample_eligible_treatment_success(on_med, p_treatment_works, condition)
# ----- Impose the symptom on random sample of those with each condition to have:
# TODO: @britta make linear model data-specific and add in needed complexity
for symptom in self.prob_symptoms[condition].keys():
lm_init_symptoms = LinearModel(
LinearModelType.MULTIPLICATIVE,
self.prob_symptoms[condition].get(f'{symptom}'),
Predictor(
f'nc_{condition}',
conditions_are_mutually_exclusive=True,
conditions_are_exhaustive=True
)
.when(True, 1.0)
.when(False, 0.0))
has_symptom_at_init = lm_init_symptoms.predict(df.loc[df.is_alive], self.rng)
self.sim.modules['SymptomManager'].change_symptom(
person_id=has_symptom_at_init.index[has_symptom_at_init].tolist(),
symptom_string=f'{symptom}',
add_or_remove='+',
disease_module=self)
# ----- Set ever_diagnosed, date of diagnosis, and on_medication to false / NaT
# for everyone
for event in self.events:
df.loc[df.is_alive, f'nc_{event}'] = False
df.loc[df.is_alive, f'nc_{event}_date_last_event'] = pd.NaT
df.loc[df.is_alive, f'nc_{event}_ever_diagnosed'] = False
df.loc[df.is_alive, f'nc_{event}_date_diagnosis'] = pd.NaT
df.loc[df.is_alive, f'nc_{event}_on_medication'] = False
df.loc[df.is_alive, f'nc_{event}_scheduled_date_death'] = pd.NaT
df.loc[df.is_alive, f'nc_{event}_medication_prevents_death'] = False
# ----- Generate the initial "risk score" for the population based on exercise, diet, tobacco, alcohol, BMI
self.update_risk_score()
# ----- Set all other parameters to False / NaT
df.loc[df.is_alive, 'nc_ever_weight_loss_treatment'] = False
df.loc[df.is_alive, 'nc_weight_loss_worked'] = False
[docs]
def initialise_simulation(self, sim):
"""Schedule:
* Main Polling Event
* Main Logging Event
* Build the LinearModels for the onset/removal of each condition:
"""
sim.schedule_event(CardioMetabolicDisorders_MainPollingEvent(self, self.parameters['interval_between_polls']),
sim.date)
sim.schedule_event(CardioMetabolicDisorders_LoggingEvent(self), sim.date)
# Get DALY weights
if 'HealthBurden' in self.sim.modules.keys():
self.daly_wts['daly_diabetes_uncomplicated'] = \
self.sim.modules['HealthBurden'].get_daly_weight(sequlae_code=971)
self.daly_wts['daly_diabetes_complicated'] = \
self.sim.modules['HealthBurden'].get_daly_weight(sequlae_code=970)
self.daly_wts['daly_chronic_kidney_disease_moderate'] = \
self.sim.modules['HealthBurden'].get_daly_weight(sequlae_code=978)
self.daly_wts['daly_chronic_ischemic_hd'] = \
self.sim.modules['HealthBurden'].get_daly_weight(sequlae_code=697)
self.daly_wts['daly_chronic_lower_back_pain'] = \
self.sim.modules['HealthBurden'].get_daly_weight(sequlae_code=1180)
self.daly_wts['daly_stroke'] = self.sim.modules['HealthBurden'].get_daly_weight(sequlae_code=701)
self.daly_wts['daly_heart_attack_acute_days_1_2'] = \
self.sim.modules['HealthBurden'].get_daly_weight(sequlae_code=689)
self.daly_wts['daly_heart_attack_acute_days_3_28'] = \
self.sim.modules['HealthBurden'].get_daly_weight(sequlae_code=693)
# Days 1-2 of heart attack have a more severe weight than days 3028; the average of these is what is used
# for the DALY weight for any heart attack event
self.daly_wts['daly_heart_attack_avg'] = (
(2 / 28) * self.daly_wts['daly_heart_attack_acute_days_1_2']
+ (26 / 28) * self.daly_wts['daly_heart_attack_acute_days_3_28']
)
self.trackers = {
'onset_condition': Tracker(conditions=self.conditions, age_groups=self.age_cats),
'incident_event': Tracker(conditions=self.events, age_groups=self.age_cats),
'prevalent_event': Tracker(conditions=self.events, age_groups=self.age_cats),
}
# Build the LinearModel for onset/removal/deaths for each condition
# Baseline probability of condition onset, removal, and death are annual; in LinearModel, rates are adjusted to
# be consistent with the polling interval
for condition in self.conditions:
self.lms_onset[condition] = self.build_linear_model(condition, self.parameters['interval_between_polls'],
lm_type='onset')
self.lms_removal[condition] = self.build_linear_model(condition, self.parameters['interval_between_polls'],
lm_type='removal')
self.lms_death[condition] = self.build_linear_model(condition, self.parameters['interval_between_polls'],
lm_type='death')
self.lms_symptoms[condition] = self.build_linear_model_symptoms(condition, self.parameters[
'interval_between_polls'])
# Hypertension is the only condition for which we assume some community-based testing occurs; build LM based on
# age / sex
self.lms_testing['hypertension'] = self.build_linear_model('hypertension', self.parameters[
'interval_between_polls'], lm_type='testing')
for event in self.events:
self.lms_event_onset[event] = self.build_linear_model(event, self.parameters['interval_between_polls'],
lm_type='onset')
self.lms_event_death[event] = self.build_linear_model(event, self.parameters['interval_between_polls'],
lm_type='death')
self.lms_event_symptoms[event] = self.build_linear_model_symptoms(event, self.parameters[
'interval_between_polls'])
# ------------------------------------------ DEFINE THE TESTS -------------------------------------------------
# Create the diagnostic representing the assessment for whether a person is diagnosed with diabetes
# NB. Sensitivity/specificity is assumed to be 100%
self.sim.modules['HealthSystem'].dx_manager.register_dx_test(
assess_diabetes=DxTest(
property='nc_diabetes',
item_codes=self.parameters['diabetes_hsi']['test_item_code'].astype(int)
)
)
# Create the diagnostic representing the assessment for whether a person is diagnosed with hypertension:
# blood pressure measurement
self.sim.modules['HealthSystem'].dx_manager.register_dx_test(
assess_hypertension=DxTest(
property='nc_hypertension'
)
)
# Create the diagnostic representing the assessment for whether a person is diagnosed with back pain
self.sim.modules['HealthSystem'].dx_manager.register_dx_test(
assess_chronic_lower_back_pain=DxTest(
property='nc_chronic_lower_back_pain'
)
)
# Create the diagnostic representing the assessment for whether a person is diagnosed with CKD
self.sim.modules['HealthSystem'].dx_manager.register_dx_test(
assess_chronic_kidney_disease=DxTest(
property='nc_chronic_kidney_disease',
item_codes=self.parameters['chronic_kidney_disease_hsi']['test_item_code'].astype(int)
)
)
# Create the diagnostic representing the assessment for whether a person is diagnosed with CIHD
self.sim.modules['HealthSystem'].dx_manager.register_dx_test(
assess_chronic_ischemic_hd=DxTest(
property='nc_chronic_ischemic_hd'
)
)
# Create the diagnostic representing the assessment for whether a person is diagnosed with stroke
self.sim.modules['HealthSystem'].dx_manager.register_dx_test(
assess_ever_stroke=DxTest(
property='nc_ever_stroke'
)
)
# Create the diagnostic representing the assessment for whether a person is diagnosed with heart attack
self.sim.modules['HealthSystem'].dx_manager.register_dx_test(
assess_ever_heart_attack=DxTest(
property='nc_ever_heart_attack'
)
)
[docs]
def build_linear_model(self, condition, interval_between_polls, lm_type):
"""
Build a linear model for the risk of onset, removal, or death from a condition, occurrence or death from an
event, and community-based testing for hypertension.
:param condition: the condition or event to build the linear model for
:param interval_between_polls: the duration (in months) between the polls
:param lm_type: whether or not the lm is for onset, removal, death, or event in order to select the correct
parameter set below
:return: a linear model
"""
# Load parameters for correct condition/event
p = self.parameters[f'{condition}_{lm_type}']
# For events, probability of death does not need to be standardised to poll interval since event is a discrete
# occurrence
if condition.startswith('ever_') and lm_type == 'death':
baseline_annual_probability = p['baseline_annual_probability']
else:
baseline_annual_probability = 1 - math.exp(-interval_between_polls / 12 * p['baseline_annual_probability'])
# LinearModel expects native python types - if it's numpy type, convert it
baseline_annual_probability = float(baseline_annual_probability)
linearmodel = LinearModel(
LinearModelType.MULTIPLICATIVE,
baseline_annual_probability,
Predictor('sex').when('M', p['rr_male']),
Predictor(
'age_years',
conditions_are_mutually_exclusive=True,
conditions_are_exhaustive=True
)
.when('.between(0, 4)', p['rr_0_4'])
.when('.between(5, 9)', p['rr_5_9'])
.when('.between(10, 14)', p['rr_10_14'])
.when('.between(15, 19)', p['rr_15_19'])
.when('.between(20, 24)', p['rr_20_24'])
.when('.between(25, 29)', p['rr_25_29'])
.when('.between(30, 34)', p['rr_30_34'])
.when('.between(35, 39)', p['rr_35_39'])
.when('.between(40, 44)', p['rr_40_44'])
.when('.between(45, 49)', p['rr_45_49'])
.when('.between(50, 54)', p['rr_50_54'])
.when('.between(55, 59)', p['rr_55_59'])
.when('.between(60, 64)', p['rr_60_64'])
.when('.between(65, 69)', p['rr_65_69'])
.when('.between(70, 74)', p['rr_70_74'])
.when('.between(75, 79)', p['rr_75_79'])
.when('.between(80, 84)', p['rr_80_84'])
.when('.between(85, 89)', p['rr_85_89'])
.when('.between(90, 94)', p['rr_90_94'])
.when('.between(95, 99)', p['rr_95_99'])
.when('>= 100', p['rr_100']),
Predictor('li_urban').when(True, p['rr_urban']),
Predictor(
'li_wealth',
conditions_are_mutually_exclusive=True,
conditions_are_exhaustive=True,
)
.when('1', p['rr_wealth_1'])
.when('2', p['rr_wealth_2'])
.when('3', p['rr_wealth_3'])
.when('4', p['rr_wealth_4'])
.when('5', p['rr_wealth_5']),
Predictor(
'li_bmi',
conditions_are_mutually_exclusive=True,
conditions_are_exhaustive=True
)
.when('1', p['rr_bmi_1'])
.when('2', p['rr_bmi_2'])
.when('3', p['rr_bmi_3'])
.when('4', p['rr_bmi_4'])
.when('5', p['rr_bmi_5']),
Predictor('li_low_ex').when(True, p['rr_low_exercise']),
Predictor('li_high_salt').when(True, p['rr_high_salt']),
Predictor('li_high_sugar').when(True, p['rr_high_sugar']),
Predictor('li_tob').when(True, p['rr_tobacco']),
Predictor('li_ex_alc').when(True, p['rr_alcohol']),
Predictor(
'li_mar_stat',
conditions_are_mutually_exclusive=True,
conditions_are_exhaustive=True
)
.when('1', p['rr_marital_status_1'])
.when('2', p['rr_marital_status_2'])
.when('3', p['rr_marital_status_3']),
Predictor('li_in_ed').when(True, p['rr_in_education']),
Predictor(
'li_ed_lev',
conditions_are_mutually_exclusive=True,
conditions_are_exhaustive=True
)
.when('1', p['rr_current_education_level_1'])
.when('2', p['rr_current_education_level_2'])
.when('3', p['rr_current_education_level_3']),
Predictor('li_unimproved_sanitation').when(True, p['rr_unimproved_sanitation']),
Predictor('li_no_access_handwashing').when(True, p['rr_no_access_handwashing']),
Predictor('li_no_clean_drinking_water').when(True, p['rr_no_clean_drinking_water']),
Predictor('li_wood_burn_stove').when(True, p['rr_wood_burning_stove']),
Predictor('nc_diabetes').when(True, p['rr_diabetes']),
Predictor('nc_hypertension').when(True, p['rr_hypertension']),
Predictor('de_depr').when(True, p['rr_depression']),
Predictor('nc_chronic_kidney_disease').when(True, p['rr_chronic_kidney_disease']),
Predictor('nc_chronic_lower_back_pain').when(True, p['rr_chronic_lower_back_pain']),
Predictor('nc_chronic_ischemic_hd').when(True, p['rr_chronic_ischemic_heart_disease']),
Predictor('nc_ever_stroke').when(True, p['rr_ever_stroke']),
Predictor('nc_ever_heart_attack').when(True, p['rr_ever_heart_attack']),
Predictor('nc_diabetes_on_medication').when(True, p['rr_diabetes_on_medication']),
Predictor('nc_hypertension_on_medication').when(True, p['rr_hypertension_on_medication']),
Predictor('nc_chronic_lower_back_pain_on_medication').when(True, p[
'rr_chronic_lower_back_pain_on_medication']),
Predictor('nc_chronic_kidney_disease_on_medication').when(True, p[
'rr_chronic_kidney_disease_on_medication']),
Predictor('nc_chronic_ischemic_hd_on_medication').when(True, p[
'rr_chronic_ischemic_heart_disease_on_medication']),
Predictor('nc_ever_stroke_on_medication').when(True, p['rr_stroke_on_medication']),
Predictor('nc_ever_heart_attack_on_medication').when(True, p['rr_heart_attack_on_medication'])
)
return linearmodel
[docs]
def build_linear_model_symptoms(self, condition, interval_between_polls):
"""
Build a linear model for the risk of symptoms from a condition.
:param condition: the condition to build the linear model for
:param interval_between_polls: the duration (in months) between the polls
:return: a linear model
"""
# Use temporary empty dict to save results
lms_symptoms_dict = dict()
lms_symptoms_dict[condition] = {}
# Load parameters for correct condition
p = self.prob_symptoms[condition]
for symptom in p.keys():
p_symptom_onset = 1 - math.exp(-interval_between_polls / 12 * p.get(f'{symptom}'))
lms_symptoms_dict[condition][f'{symptom}'] = LinearModel(LinearModelType.MULTIPLICATIVE,
p_symptom_onset, Predictor(f'nc_{condition}')
.when(True, 1.0).otherwise(0.0))
return lms_symptoms_dict[condition]
[docs]
def on_birth(self, mother_id, child_id):
"""Initialise our properties for a newborn individual.
:param mother_id: the mother for this child
:param child_id: the new child
"""
df = self.sim.population.props
for condition in self.conditions:
df.at[child_id, f'nc_{condition}'] = False
df.at[child_id, f'nc_{condition}_ever_diagnosed'] = False
df.at[child_id, f'nc_{condition}_date_diagnosis'] = pd.NaT
df.at[child_id, f'nc_{condition}_date_last_test'] = pd.NaT
df.at[child_id, f'nc_{condition}_on_medication'] = False
df.at[child_id, f'nc_{condition}_medication_prevents_death'] = False
for event in self.events:
df.at[child_id, f'nc_{event}'] = False
df.at[child_id, f'nc_{event}_date_last_event'] = pd.NaT
df.at[child_id, f'nc_{event}_ever_diagnosed'] = False
df.at[child_id, f'nc_{event}_on_medication'] = False
df.at[child_id, f'nc_{event}_date_diagnosis'] = pd.NaT
df.at[child_id, f'nc_{event}_scheduled_date_death'] = pd.NaT
df.at[child_id, f'nc_{event}_medication_prevents_death'] = False
df.at[child_id, 'nc_risk_score'] = 0
[docs]
def update_risk_score(self):
"""
Generates or updates the risk score for individuals at initialisation of population or at each polling event
"""
df = self.sim.population.props
df.loc[df.is_alive, 'nc_risk_score'] = (df[[
'li_low_ex', 'li_high_salt', 'li_high_sugar', 'li_tob', 'li_ex_alc']] > 0).sum(1)
df.loc[df['li_bmi'] >= 3, ['nc_risk_score']] += 1
[docs]
def report_daly_values(self):
"""Report disability weight (average values for the last month) to the HealthBurden module"""
def left_censor(obs, window_open):
return obs.apply(lambda x: max(x, window_open) if pd.notnull(x) else pd.NaT)
df = self.sim.population.props
dw = pd.DataFrame(data=0.0, index=df.index[df.is_alive], columns=self.CAUSES_OF_DISABILITY.keys())
# Diabetes: give everyone who is diagnosed or on medication uncomplicated diabetes daly weight
dw['diabetes'].loc[df['nc_diabetes_ever_diagnosed'] | df['nc_diabetes_on_medication']] = self.daly_wts[
'daly_diabetes_uncomplicated']
# Diabetes: overwrite those with symptoms to have complicated diabetes daly weight
dw['diabetes'].loc[
self.sim.modules['SymptomManager'].who_has('diabetes_symptoms')] = self.daly_wts[
'daly_diabetes_complicated']
# Chronic Lower Back Pain: give those who have symptoms moderate weight
dw['lower_back_pain'].loc[
self.sim.modules['SymptomManager'].who_has('chronic_lower_back_pain_symptoms')] = self.daly_wts[
'daly_chronic_lower_back_pain']
# Chronic Kidney Disease: give those who have symptoms moderate weight
dw['chronic_kidney_disease'].loc[
self.sim.modules['SymptomManager'].who_has('chronic_kidney_disease_symptoms')] = self.daly_wts[
'daly_chronic_kidney_disease_moderate']
# Stroke: give everyone moderate long-term consequences daly weight
dw['stroke'].loc[df.nc_ever_stroke] = self.daly_wts['daly_stroke']
# Chronic Ischemic Heart Disease: give everyone with CIHD symptoms moderate CIHD daly weight
dw['chronic_ischemic_hd'].loc[
self.sim.modules['SymptomManager'].who_has('chronic_ischemic_hd_symptoms')] = self.daly_wts[
'daly_chronic_ischemic_hd']
# Heart Attack: first calculate proportion of month spent following heart attack, then attach weighted daly
# Calculate fraction of the last month that was spent after having a heart attack
days_in_last_month = (self.sim.date - (self.sim.date - DateOffset(months=1))).days
start_heart_attack = left_censor(
df.loc[df.is_alive, 'nc_ever_heart_attack_date_last_event'], self.sim.date - DateOffset(months=1)
)
dur_heart_attack_in_days = (self.sim.date - start_heart_attack).dt.days.clip(
lower=0, upper=days_in_last_month).fillna(0.0)
fraction_of_month_heart_attack = dur_heart_attack_in_days / days_in_last_month
dw['heart_attack'] = fraction_of_month_heart_attack * self.daly_wts['daly_heart_attack_avg']
return dw
[docs]
def on_hsi_alert(self, person_id, treatment_id):
"""
This is called whenever there is an HSI event commissioned by one of the other disease modules.
"""
pass
[docs]
def determine_if_will_be_investigated(self, person_id):
"""
This is called by the HSI generic first appts module whenever a person attends an appointment and determines
if the person will be tested for one or more conditions.
A maximum of one instance of `HSI_CardioMetabolicDisorders_Investigations` is created for the person, during
which multiple conditions can be investigated.
"""
def is_next_test_due(current_date, date_of_last_test):
return pd.isnull(date_of_last_test) or (current_date - date_of_last_test).days > DAYS_IN_YEAR / 2
df = self.sim.population.props
person = df.loc[person_id, df.columns[df.columns.str.startswith('nc_')]]
symptoms = self.sim.modules['SymptomManager'].has_what(person_id=person_id)
conditions_to_investigate = [] # The list of conditions that will be investigated in follow-up HSI
has_any_cmd_symptom = False # Marker for whether the person has any symptoms of interest
# Determine if there are any conditions that should be investigated:
for condition in self.conditions:
is_already_diagnosed = person[f'nc_{condition}_ever_diagnosed']
has_symptom = f'{condition}_symptoms' in symptoms
next_test_due = is_next_test_due(
current_date=self.sim.date, date_of_last_test=df.at[person_id, f'nc_{condition}_date_last_test']
)
p_assess_if_no_symptom = self.parameters[f'{condition}_hsi'].get('pr_assessed_other_symptoms')
if (not is_already_diagnosed) and (
has_symptom or (next_test_due and (self.rng.random_sample() < p_assess_if_no_symptom))
):
# If the person is not already diagnosed and either has the symptom or is due a routine check...
# ... add this condition to be investigated in the appointment.
conditions_to_investigate.append(condition)
if (not is_already_diagnosed) and has_symptom:
has_any_cmd_symptom = True
# Schedule follow-up HSI *if* there are any conditions to investigate:
if conditions_to_investigate:
self.sim.modules['HealthSystem'].schedule_hsi_event(
HSI_CardioMetabolicDisorders_Investigations(
module=self,
person_id=person_id,
conditions_to_investigate=conditions_to_investigate,
has_any_cmd_symptom=has_any_cmd_symptom,
),
priority=0,
topen=self.sim.date,
tclose=None
)
[docs]
def determine_if_will_be_investigated_events(self, person_id):
"""
This is called by the HSI generic first appts module whenever a person attends an emergency appointment and
determines if they will receive emergency care based on the duration of time since symptoms have appeared.
A maximum of one instance of `HSI_CardioMetabolicDisorders_SeeksEmergencyCareAndGetsTreatment` is created
for the person, during which multiple events can be investigated.
"""
health_system = self.sim.modules["HealthSystem"]
symptoms = self.sim.modules['SymptomManager'].has_what(person_id=person_id)
ev_to_investigate = []
for ev in self.events:
# If the person has symptoms of damage from within the last 3 days, schedule them for emergency care
if f'{ev}_damage' in symptoms and \
((self.sim.date - self.sim.population.props.at[person_id, f'nc_{ev}_date_last_event']).days <= 3):
ev_to_investigate.append(ev)
if ev_to_investigate:
health_system.schedule_hsi_event(
HSI_CardioMetabolicDisorders_SeeksEmergencyCareAndGetsTreatment(
module=self,
person_id=person_id,
events_to_investigate=ev_to_investigate,
),
priority=1,
topen=self.sim.date
)
class Tracker:
"""Helper class for keeping a count with respect to some conditions and some age-groups"""
def __init__(self, conditions: list, age_groups: list):
self._conditions = conditions
self._age_groups = age_groups
self._tracker = self._make_new_tracker()
def _make_new_tracker(self):
return {c: {a: 0.0 for a in self._age_groups} for c in self._conditions}
def reset(self):
self._tracker = self._make_new_tracker()
def add(self, condition: str, _to_add: dict):
for _a in _to_add:
if _a in self._tracker[condition]:
self._tracker[condition][_a] += _to_add[_a]
def report(self):
return self._tracker
# ---------------------------------------------------------------------------------------------------------
# DISEASE MODULE EVENTS
#
# The regular event that actually changes individuals' condition or event status, occurring every 3 months
# and synchronously for all persons.
# Individual level events (HSI, death or cardio-metabolic events) may occur at other times.
# ---------------------------------------------------------------------------------------------------------
[docs]
class CardioMetabolicDisorders_MainPollingEvent(RegularEvent, PopulationScopeEventMixin):
"""The Main Polling Event.
* Establishes onset of each condition
* Establishes removal of each condition
* Schedules events that arise, according the condition.
"""
[docs]
def __init__(self, module, interval_between_polls):
"""The Main Polling Event of the CardioMetabolicDisorders Module
:param module: the module that created this event
"""
super().__init__(module, frequency=DateOffset(months=interval_between_polls))
assert isinstance(module, CardioMetabolicDisorders)
[docs]
def apply(self, population):
"""Apply this event to the population.
:param population: the current population
"""
df = population.props
m = self.module
rng = m.rng
# Function to schedule deaths on random day throughout polling period
def schedule_death_to_occur_before_next_poll(p_id, cond):
self.sim.schedule_event(
CardioMetabolicDisordersDeathEvent(self.module, p_id, cond),
random_date(self.sim.date, self.sim.date + self.frequency - pd.DateOffset(days=1), m.rng)
)
# -------------------------------- COMMUNITY SCREENING FOR HYPERTENSION ---------------------------------------
# A subset of individuals aged >= 50 will receive a blood pressure measurement without directly presenting to
# the healthcare system
eligible_population = df.is_alive & ~df['nc_hypertension_ever_diagnosed']
will_test = self.module.lms_testing['hypertension'].predict(
df.loc[eligible_population], rng, squeeze_single_row_output=False)
idx_will_test = will_test[will_test].index
# Schedule persons for community testing before next polling event
for person_id in idx_will_test:
self.sim.modules['HealthSystem'].schedule_hsi_event(
hsi_event=HSI_CardioMetabolicDisorders_CommunityTestingForHypertension(person_id=person_id,
module=self.module),
priority=1,
topen=random_date(self.sim.date, self.sim.date + self.frequency - pd.DateOffset(days=2), m.rng),
tclose=self.sim.date + self.frequency - pd.DateOffset(days=1) # (to occur before next polling)
)
# ------------------------------ DETERMINE ONSET / REMOVAL OF CONDITIONS -------------------------------------
for condition in self.module.conditions:
# Onset:
eligible_population = df.is_alive & ~df[f'nc_{condition}']
acquires_condition = self.module.lms_onset[condition].predict(
df.loc[eligible_population], rng, squeeze_single_row_output=False)
idx_acquires_condition = acquires_condition[acquires_condition].index
df.loc[idx_acquires_condition, f'nc_{condition}'] = True
# Add incident cases to the tracker
self.module.trackers['onset_condition'].add(
condition, df.loc[idx_acquires_condition].groupby('age_range').size().to_dict()
)
# Schedule symptom onset for both those with new onset of condition and those who already have condition,
# among those who do not have the symptom already
if len(self.module.lms_symptoms[condition]) > 0:
symptom_eligible_population = df.is_alive & df[f'nc_{condition}'] & ~df.index.isin(
self.sim.modules['SymptomManager'].who_has(f'{condition}_symptoms'))
symptom_onset = self.module.lms_symptoms[condition][f'{condition}_symptoms'].predict(
df.loc[symptom_eligible_population], rng, squeeze_single_row_output=False
)
idx_symptom_onset = symptom_onset[symptom_onset].index
if idx_symptom_onset.any():
# Schedule symptom onset some time before next polling event
for symptom in self.module.prob_symptoms[condition].keys():
date_onset = random_date(
self.sim.date, self.sim.date + self.frequency - pd.DateOffset(days=1), m.rng)
self.sim.modules['SymptomManager'].change_symptom(
person_id=idx_symptom_onset.tolist(),
symptom_string=f'{symptom}',
add_or_remove='+',
date_of_onset=date_onset,
disease_module=self.module)
# Removal:
eligible_population = df.is_alive & df[f'nc_{condition}']
if len(self.module.lms_symptoms[condition]) > 0:
eligible_population &= ~symptom_onset
loses_condition = self.module.lms_removal[condition].predict(
df.loc[eligible_population], rng, squeeze_single_row_output=False)
idx_loses_condition = loses_condition[loses_condition].index
df.loc[idx_loses_condition, f'nc_{condition}'] = False
df.loc[idx_loses_condition, f'nc_{condition}_on_medication'] = False
if condition != 'hypertension':
# Remove all symptoms of condition instantly
self.sim.modules['SymptomManager'].change_symptom(
person_id=idx_loses_condition.tolist(),
symptom_string=f'{condition}_symptoms',
add_or_remove='-',
disease_module=self.module)
# --------------------------- DEATH FROM CARDIO-METABOLIC CONDITION ---------------------------------------
# There is a risk of death for those who have a cardio-metabolic condition.
# Death is assumed to happen in the time before the next polling event.
eligible_population = df.is_alive & df[f'nc_{condition}']
selected_to_die = self.module.lms_death[condition].predict(df.loc[eligible_population], rng,
squeeze_single_row_output=False)
idx_selected_to_die = selected_to_die[selected_to_die].index
for person_id in idx_selected_to_die:
schedule_death_to_occur_before_next_poll(person_id, condition)
# ---------------------------------- DETERMINE OCCURRENCE OF EVENTS ------------------------------------------
for event in self.module.events:
eligible_population_for_event = df.is_alive
has_event = self.module.lms_event_onset[event].predict(df.loc[eligible_population_for_event], rng)
if has_event.any(): # catch in case no one has an event
idx_has_event = has_event[has_event].index
for person_id in idx_has_event:
self.sim.schedule_event(CardioMetabolicDisordersEvent(self.module, person_id, event),
random_date(self.sim.date, self.sim.date + self.frequency -
pd.DateOffset(days=1), m.rng))
# Add cases to either incident or prevalent list of cases
if df.at[person_id, f'nc_{event}']:
self.module.trackers['prevalent_event'].add(event, {df.at[person_id, 'age_range']: 1})
else:
self.module.trackers['incident_event'].add(event, {df.at[person_id, 'age_range']: 1})
[docs]
class CardioMetabolicDisordersEvent(Event, IndividualScopeEventMixin):
"""
This is an Cardio Metabolic Disorders event (indicating an emergency occurrence of stroke or heart attack).
It has been scheduled to occur by the CardioMetabolicDisorders_MainPollingEvent.
:param event: the event occurring
"""
[docs]
def __init__(self, module, person_id, event):
super().__init__(module, person_id=person_id)
self.event = event
[docs]
def apply(self, person_id):
df = self.sim.population.props
if not df.at[person_id, 'is_alive']:
return
df.at[person_id, f'nc_{self.event}'] = True
df.at[person_id, f'nc_{self.event}_date_last_event'] = self.sim.date
# Add the outward symptom to the SymptomManager. This will result in emergency care being sought for any
# event that takes place
self.sim.modules['SymptomManager'].change_symptom(
person_id=person_id,
disease_module=self.module,
add_or_remove='+',
symptom_string=f'{self.event}_damage'
)
# ------------------------------------ DETERMINE OUTCOME OF THIS EVENT ----------------------------------------
prob_death = self.module.lms_event_death[self.event].predict(df.loc[[person_id]])
# Schedule a future death event for 7 days' time
date_of_outcome = self.sim.date + DateOffset(days=7)
if self.module.rng.random_sample() < prob_death[person_id]:
df.at[person_id, f'nc_{self.event}_scheduled_date_death'] = date_of_outcome
self.sim.schedule_event(CardioMetabolicDisordersDeathEvent(self.module, person_id,
originating_cause=self.event), date_of_outcome)
[docs]
class CardioMetabolicDisordersDeathEvent(Event, IndividualScopeEventMixin):
"""
Performs the Death operation on an individual and logs it.
:param originating_cause: the originating cause of the death, which can be a condition or an event
"""
[docs]
def __init__(self, module, person_id, originating_cause):
super().__init__(module, person_id=person_id)
self.originating_cause = originating_cause
[docs]
def apply(self, person_id):
df = self.sim.population.props
person = df.loc[person_id]
if not person.is_alive:
return
# Check still have condition (has not resolved)
if person[f'nc_{self.originating_cause}']:
# Reduction in risk of death if being treated with regular medication for condition
if person[f'nc_{self.originating_cause}_on_medication']:
if not df.at[person_id, f'nc_{self.originating_cause}_medication_prevents_death']:
self.check_if_event_and_do_death(person_id)
else:
self.check_if_event_and_do_death(person_id)
[docs]
def check_if_event_and_do_death(self, person_id):
"""
Helper function to perform do_death if person dies of a condition or event. If person dies of an event, this
will only perform do_death if the scheduled date of death matches the current date (to allow for the possibility
that treatment will intercede in an prevent death from the event).
"""
df = self.sim.population.props
person = df.loc[person_id]
# Check if it's a death event for an event (e.g. stroke) in order to execute death only if the date equals
# scheduled date of death
if f'{self.originating_cause}' in self.module.events:
if self.sim.date == person[f'nc_{self.originating_cause}_scheduled_date_death']:
self.sim.modules['Demography'].do_death(individual_id=person_id,
cause=f'{self.originating_cause}',
originating_module=self.module)
else:
# Conditions have no scheduled date of death, so proceed with death
self.sim.modules['Demography'].do_death(individual_id=person_id,
cause=f'{self.originating_cause}',
originating_module=self.module)
[docs]
class CardioMetabolicDisordersWeightLossEvent(Event, IndividualScopeEventMixin):
"""
Gives an individual a probability of losing weight (via a reduction in li_bmi) and records the change in
population.props
"""
[docs]
def apply(self, person_id):
df = self.sim.population.props
person = df.loc[person_id]
if not person.is_alive:
return
else:
if self.module.rng.random_sample() < self.module.parameters['pr_bmi_reduction']:
df.at[person_id, 'li_bmi'] -= 1
df.at[person_id, 'nc_weight_loss_worked'] = True
# ---------------------------------------------------------------------------------------------------------
# LOGGING EVENTS
#
# Put the logging events here. There should be a regular logger outputting current states of the
# population. There may also be a loggig event that is driven by particular events.
# ---------------------------------------------------------------------------------------------------------
[docs]
class CardioMetabolicDisorders_LoggingEvent(RegularEvent, PopulationScopeEventMixin):
[docs]
def __init__(self, module):
"""Produce a summary of the numbers of people with respect to the action of this module.
"""
self.repeat = 12
super().__init__(module, frequency=DateOffset(months=self.repeat))
self.date_last_run = self.sim.date
self.AGE_RANGE_LOOKUP = self.sim.modules['Demography'].AGE_RANGE_LOOKUP
assert isinstance(module, CardioMetabolicDisorders)
[docs]
def apply(self, population):
# Create shortcut to the Demography module
demog_module = self.sim.modules['Demography']
# Log counts in the trackers
logger.info(key='incidence_count_by_condition',
data=self.module.trackers['onset_condition'].report(),
description=f"count of events occurring between each successive poll of logging event every "
f"{self.repeat} months")
logger.info(key='incidence_count_by_incident_event',
data=self.module.trackers['incident_event'].report(),
description=f"count of incident events occurring between each successive poll of logging event "
f"every "f"{self.repeat} months")
logger.info(key='incidence_count_by_prevalent_event',
data=self.module.trackers['prevalent_event'].report(),
description=f"count of prevalent events occurring between each successive poll of logging event "
f"every "f"{self.repeat} months")
# Reset trackers
for _tracker in self.module.trackers:
self.module.trackers[_tracker].reset()
def age_cats(ages_in_years):
AGE_RANGE_CATEGORIES = self.sim.modules['Demography'].AGE_RANGE_CATEGORIES
AGE_RANGE_LOOKUP = self.sim.modules['Demography'].AGE_RANGE_LOOKUP
_age_cats = pd.Series(
pd.Categorical(ages_in_years.map(AGE_RANGE_LOOKUP),
categories=AGE_RANGE_CATEGORIES, ordered=True)
)
return _age_cats
# Function to prepare a groupby for logging
def proportion_of_something_in_a_groupby_ready_for_logging(_df, something, groupbylist):
dfx = _df.groupby(groupbylist).apply(lambda dft: pd.Series(
{'something': dft[something].sum(), 'not_something': (~dft[something]).sum()}))
pr = dfx['something'] / dfx.sum(axis=1)
# create into a dict with keys as strings
pr = pr.reset_index()
pr['flat_index'] = ''
for _i in range(len(pr)):
pr.at[_i, 'flat_index'] = '__'.join([f"{_col}={pr.at[_i, _col]}" for _col in groupbylist])
pr = pr.set_index('flat_index', drop=True)
pr = pr.drop(columns=groupbylist)
return pr[0].to_dict()
# Output the person-years lived by single year of age in the past year
df = population.props
delta = pd.DateOffset(years=1)
for cond in self.module.conditions:
# mask is a Series restricting dataframe to individuals who do not have the condition, which is passed to
# demography module to calculate person-years lived without the condition
mask = (df.is_alive & ~df[f'nc_{cond}'])
py = de.Demography.calc_py_lived_in_last_year(demog_module, delta, mask)
py['age_range'] = age_cats(py.index)
py = py.groupby('age_range').sum()
logger.info(key=f'person_years_{cond}', data=py.to_dict())
for event in self.module.events:
# mask is a Series restricting dataframe to individuals who do not have the condition, which is passed to
# demography module to calculate person-years lived without the condition
mask = (df.is_alive & ~df[f'nc_{event}'])
py = de.Demography.calc_py_lived_in_last_year(demog_module, delta, mask)
py['age_range'] = age_cats(py.index)
py = py.groupby('age_range').sum()
logger.info(key=f'person_years_{event}', data=py.to_dict())
# Make some summary statistics for prevalence by age/sex for each condition
df = population.props
# Prevalence of conditions broken down by sex and age
for condition in self.module.conditions:
prev_age_sex = proportion_of_something_in_a_groupby_ready_for_logging(df, f'nc_{condition}',
['sex', 'age_range'])
# Prevalence of conditions broken down by sex and age
logger.info(
key=f'{condition}_prevalence_by_age_and_sex',
description='current fraction of the population classified as having condition, by sex and age',
data={'data': prev_age_sex}
)
# Prevalence of conditions by adults aged 20 or older
adult_prevalence = {
'prevalence': len(df[df[f'nc_{condition}'] & df.is_alive & (df.age_years >= 20)]) / len(
df[df.is_alive & (df.age_years >= 20)])}
logger.info(
key=f'{condition}_prevalence',
description='current fraction of the adult population classified as having condition',
data=adult_prevalence
)
if len(df[df[f'nc_{condition}'] & df.is_alive & (df.age_years >= 20)]) > 0:
diagnosed = {
f'{condition}_diagnosis_prevalence': len(df[df[f'nc_{condition}_ever_diagnosed'] & df.is_alive & (
df.age_years >= 20)]) / len(df[df[f'nc_{condition}'] & df.is_alive & (df.age_years >= 20)])
}
else:
diagnosed = {0.0}
logger.info(
key=f'{condition}_diagnosis_prevalence',
description='current fraction of the adult population diagnosed for the condition',
data=diagnosed
)
if len(df[df[f'nc_{condition}'] & df.is_alive & (df.age_years >= 20)]) > 0:
on_medication = {
f'{condition}_medication_prevalence': len(df[df[f'nc_{condition}_on_medication'] & df.is_alive & (
df.age_years >= 20)]) / len(df[df[f'nc_{condition}'] & df.is_alive & (df.age_years >= 20)])
}
else:
on_medication = {0.0}
logger.info(
key=f'{condition}_medication_prevalence',
description='current fraction of the adult population being on medication for the condition',
data=on_medication
)
for event in self.module.events:
prev_age_sex = proportion_of_something_in_a_groupby_ready_for_logging(df, f'nc_{event}',
['sex', 'age_range'])
# Prevalence of events broken down by sex and age
logger.info(
key=f'{event}_prevalence_by_age_and_sex',
description='current fraction of the population classified as having condition, by sex and age',
data={'data': prev_age_sex}
)
# Prevalence of events by adults aged 20 or older
adult_prevalence = {
'prevalence': len(df[df[f'nc_{event}'] & df.is_alive & (df.age_years >= 20)]) / len(
df[df.is_alive & (df.age_years >= 20)])}
logger.info(
key=f'{event}_prevalence',
description='current fraction of the adult population classified as having event',
data=adult_prevalence
)
# If param do_condition_combos = True, produce counters for number of co-morbidities by age and the combinations
# of different conditions in the population
if self.module.do_condition_combos:
df.loc[df.is_alive, 'nc_n_conditions'] = df.loc[df.is_alive, self.module.condition_list].sum(axis=1)
n_comorbidities_all = pd.DataFrame(index=self.module.age_cats,
columns=list(range(0, len(self.module.condition_list) + 1)))
df = df[['age_range', 'nc_n_conditions']]
for num in range(0, len(self.module.condition_list) + 1):
col = df.loc[df['nc_n_conditions'] == num].groupby(['age_range']).apply(lambda x: pd.Series(
{'count': x['nc_n_conditions'].count()}))
n_comorbidities_all.loc[:, num] = col['count']
prop_comorbidities_all = n_comorbidities_all.div(n_comorbidities_all.sum(axis=1), axis=0)
logger.info(key='mm_prevalence_by_age_all',
description='annual summary of multi-morbidities by age for all',
data=prop_comorbidities_all.to_dict()
)
# output combinations of different conditions
df = population.props
combos = combinations(self.module.condition_list, 2)
condition_combos = list(combos)
n_combos = pd.DataFrame(index=df['age_range'].value_counts().sort_index().index)
for i in range(0, len(condition_combos)):
df.loc[df.is_alive, 'nc_condition_combos'] = np.where(
df.loc[df.is_alive, f'{condition_combos[i][0]}'] &
df.loc[df.is_alive, f'{condition_combos[i][1]}'],
True, False)
col = df.loc[df.is_alive].groupby(['age_range'])['nc_condition_combos'].count()
n_combos.reset_index()
n_combos.loc[:, (f'{condition_combos[i][0]}' + '_' + f'{condition_combos[i][1]}')] = col.values
# output proportions of different combinations of conditions
prop_combos = n_combos.div(df.groupby(['age_range'])['age_range'].count(), axis=0)
prop_combos.index = prop_combos.index.astype(str)
logger.info(key='prop_combos',
description='proportion of combinations of morbidities',
data=prop_combos.to_dict()
)
# If param do_log_df = True, output entire dataframe for use in a logistic regression
if self.module.do_log_df:
df = population.props
columns_of_interest = ['is_alive', 'sex', 'age_range', 'li_urban', 'li_wealth', 'li_bmi', 'li_low_ex',
'li_high_salt', 'li_high_sugar', 'li_ex_alc', 'li_tob', 'nc_diabetes',
'nc_hypertension']
logger.info(key='df_snapshot', data=df[columns_of_interest],
message='dataframe of CMD variables for logistic regression')
# Update the risk score for everyone
self.module.update_risk_score()
# ---------------------------------------------------------------------------------------------------------
# HEALTH SYSTEM INTERACTION EVENTS
# ---------------------------------------------------------------------------------------------------------
[docs]
class HSI_CardioMetabolicDisorders_CommunityTestingForHypertension(HSI_Event, IndividualScopeEventMixin):
"""
This event is scheduled by HSI_GenericFirstApptAtFacilityLevel1 following presentation for care with any symptoms.
This event results in a blood pressure measurement being taken that may result in diagnosis and the scheduling of
treatment for a condition.
"""
[docs]
def __init__(self, module, person_id):
super().__init__(module, person_id=person_id)
self.TREATMENT_ID = "CardioMetabolicDisorders_Prevention_CommunityTestingForHypertension"
self.EXPECTED_APPT_FOOTPRINT = self.make_appt_footprint({"Over5OPD": 1})
self.ACCEPTED_FACILITY_LEVEL = '1a'
[docs]
def apply(self, person_id, squeeze_factor):
df = self.sim.population.props
person = df.loc[person_id]
hs = self.sim.modules["HealthSystem"]
# Ignore this event if the person is no longer alive:
if not person.is_alive:
return hs.get_blank_appt_footprint()
# Run a test to diagnose whether the person has condition:
dx_result = hs.dx_manager.run_dx_test(
dx_tests_to_run='assess_hypertension',
hsi_event=self
)
df.at[person_id, 'nc_hypertension_date_last_test'] = self.sim.date
if dx_result:
# Record date of diagnosis:
df.at[person_id, 'nc_hypertension_date_diagnosis'] = self.sim.date
df.at[person_id, 'nc_hypertension_ever_diagnosed'] = True
# Schedule HSI_CardioMetabolicDisorders_StartWeightLossAndMedication event
hs.schedule_hsi_event(
hsi_event=HSI_CardioMetabolicDisorders_StartWeightLossAndMedication(
module=self.module,
person_id=person_id,
condition='hypertension'
),
priority=0,
topen=self.sim.date,
tclose=None
)
[docs]
class HSI_CardioMetabolicDisorders_Investigations(HSI_Event, IndividualScopeEventMixin):
"""
This event is scheduled following presentation for care and a determination having been made about which
conditions should be investigated for this person (either due to presentation with symptoms or a routine check).
NB. If a treatment is needed, this does lead to multiple HSI to be scheduled for the person (relevant to each
condition).
:param conditions_to_investigate: list of the condition to investigate.
:param has_any_cmd_symptom: bool to indicate whether the person has any symptoms that prompted the investigation
"""
[docs]
def __init__(self, module, person_id, conditions_to_investigate: List, has_any_cmd_symptom: bool = False):
super().__init__(module, person_id=person_id)
self.TREATMENT_ID = "CardioMetabolicDisorders_Investigation"
self.EXPECTED_APPT_FOOTPRINT = self.make_appt_footprint({"Over5OPD": 1})
self.ACCEPTED_FACILITY_LEVEL = '1b'
self.conditions_to_investigate = conditions_to_investigate
self.has_any_cmd_symptom = has_any_cmd_symptom
[docs]
def do_for_each_condition(self, _c) -> bool:
"""What to do for each condition that will be investigated. Returns `bool` signalling whether a follow-up HSI
has been scheduled."""
hs = self.sim.modules['HealthSystem']
df = self.sim.population.props
person_id = self.target
# Do nothing if the condition is already diagnosed
if df.at[person_id, f'nc_{_c}_ever_diagnosed']:
return
# Run a test to diagnose whether the person has condition:
dx_result = hs.dx_manager.run_dx_test(
dx_tests_to_run=f'assess_{_c}',
hsi_event=self
)
df.at[person_id, f'nc_{_c}_date_last_test'] = self.sim.date
if dx_result:
# Record date of diagnosis:
df.at[person_id, f'nc_{_c}_date_diagnosis'] = self.sim.date
df.at[person_id, f'nc_{_c}_ever_diagnosed'] = True
# Schedule HSI_CardioMetabolicDisorders_StartWeightLossAndMedication event
hs.schedule_hsi_event(
hsi_event=HSI_CardioMetabolicDisorders_StartWeightLossAndMedication(
module=self.module,
person_id=person_id,
condition=f'{_c}'
),
priority=0,
topen=self.sim.date,
tclose=None
)
return dx_result
[docs]
def apply(self, person_id, squeeze_factor):
df = self.sim.population.props
hs = self.sim.modules['HealthSystem']
# Ignore this event if the person is no longer alive:
if not df.at[person_id, 'is_alive']:
return hs.get_blank_appt_footprint()
# Do test and trigger treatment (if necessary) for each condition:
hsi_scheduled = [self.do_for_each_condition(_c) for _c in self.conditions_to_investigate]
# If no follow-up treatment scheduled but the person has at least 2 risk factors, start weight loss treatment
if (
(not any(hsi_scheduled))
and (df.at[person_id, 'nc_risk_score'] >= 2)
and (not df.at[person_id, 'nc_ever_weight_loss_treatment'])
):
df.at[person_id, 'nc_ever_weight_loss_treatment'] = True
# Schedule a post-weight loss event for 6-12 months for individual to potentially lose weight:
self.sim.schedule_event(CardioMetabolicDisordersWeightLossEvent(self.module, person_id),
random_date(self.sim.date + pd.DateOffset(months=6),
self.sim.date + pd.DateOffset(months=12),
self.module.rng))
# If the person did have any symptoms, and the main condition to investigate was not `hypertension`,
# also run a test for hypertension (according to some probability),
if (
self.has_any_cmd_symptom
and ('hypertension' not in self.conditions_to_investigate)
and (self.module.rng.rand() < self.module.parameters['hypertension_hsi']['pr_assessed_other_symptoms'])
):
# Run a test to diagnose whether the person has condition:
dx_result = hs.dx_manager.run_dx_test(
dx_tests_to_run='assess_hypertension',
hsi_event=self
)
df.at[person_id, 'nc_hypertension_date_last_test'] = self.sim.date
if dx_result:
# Record date of diagnosis:
df.at[person_id, 'nc_hypertension_date_diagnosis'] = self.sim.date
df.at[person_id, 'nc_hypertension_ever_diagnosed'] = True
# Start weight loss treatment (except for CKD) and medication for all conditions
hs.schedule_hsi_event(
hsi_event=HSI_CardioMetabolicDisorders_StartWeightLossAndMedication(
module=self.module,
person_id=person_id,
condition='hypertension'
),
priority=0,
topen=self.sim.date,
tclose=None
)
[docs]
class HSI_CardioMetabolicDisorders_StartWeightLossAndMedication(HSI_Event, IndividualScopeEventMixin):
"""
This is a Health System Interaction Event in which a person receives a recommendation of weight loss.
This results in an individual having a probability of reducing their BMI by one category in the next 6-12 months.
:param condition: the condition to start medication for
"""
[docs]
def __init__(self, module, person_id, condition):
super().__init__(module, person_id=person_id)
self.TREATMENT_ID = 'CardioMetabolicDisorders_Prevention_WeightLoss'
self.EXPECTED_APPT_FOOTPRINT = self.make_appt_footprint({'Over5OPD': 1})
self.ACCEPTED_FACILITY_LEVEL = '1b'
self.condition = condition
[docs]
def apply(self, person_id, squeeze_factor):
df = self.sim.population.props
person = df.loc[person_id]
m = self.sim.modules['CardioMetabolicDisorders']
# Don't advise those with CKD to lose weight, but do so for all other conditions if BMI is higher than normal
if self.condition != 'chronic_kidney_disease' and (df.at[person_id, 'li_bmi'] > 2):
self.sim.population.props.at[person_id, 'nc_ever_weight_loss_treatment'] = True
# Schedule a post-weight loss event for individual to potentially lose weight in next 6-12 months:
self.sim.schedule_event(CardioMetabolicDisordersWeightLossEvent(m, person_id),
random_date(self.sim.date + pd.DateOffset(months=6),
self.sim.date + pd.DateOffset(months=12), m.rng))
# If person is already on medication, do not do anything
if person[f'nc_{self.condition}_on_medication']:
return self.sim.modules['HealthSystem'].get_blank_appt_footprint()
assert person[f'nc_{self.condition}_ever_diagnosed'], "The person is not diagnosed and so should not be " \
"receiving an HSI."
# Check availability of medication for condition
if self.get_consumables(
item_codes=self.module.parameters[f'{self.condition}_hsi'].get('medication_item_code').astype(int)
):
# If medication is available, flag as being on medication
df.at[person_id, f'nc_{self.condition}_on_medication'] = True
# Determine if the medication will work to prevent death
df.at[person_id, f'nc_{self.condition}_medication_prevents_death'] = \
self.module.rng.rand() < self.module.parameters[f'{self.condition}_hsi'].pr_treatment_works
# Schedule their next HSI for a refill of medication in one month
self.sim.modules['HealthSystem'].schedule_hsi_event(
hsi_event=HSI_CardioMetabolicDisorders_Refill_Medication(person_id=person_id, module=self.module,
condition=self.condition),
priority=1,
topen=self.sim.date + DateOffset(months=1),
tclose=self.sim.date + DateOffset(months=1) + DateOffset(days=7)
)
else:
# If person 'decides to' seek another appointment, schedule a new HSI appointment for tomorrow.
# NB. With a probability of 1.0, this will keep occurring, and the person will never give up coming back to
# pick up medication.
if (m.rng.random_sample() <
m.parameters[f'{self.condition}_hsi'].get('pr_seeking_further_appt_if_drug_not_available')):
self.sim.modules['HealthSystem'].schedule_hsi_event(
hsi_event=self,
topen=self.sim.date + pd.DateOffset(days=1),
tclose=self.sim.date + pd.DateOffset(days=15),
priority=1
)
[docs]
class HSI_CardioMetabolicDisorders_Refill_Medication(HSI_Event, IndividualScopeEventMixin):
"""
This is a Health System Interaction Event in which a person seeks a refill prescription of medication.
The next refill of medication is also scheduled.
If the person is flagged as not being on medication, then the event does nothing and returns a blank footprint.
If it does not run, then person ceases to be on medication and no further refill HSI are scheduled.
:param condition: the condition to refill medication for
"""
[docs]
def __init__(self, module, person_id, condition):
super().__init__(module, person_id=person_id)
self.TREATMENT_ID = 'CardioMetabolicDisorders_Treatment'
self.EXPECTED_APPT_FOOTPRINT = self.make_appt_footprint({'Over5OPD': 1})
self.ACCEPTED_FACILITY_LEVEL = '1b'
self.condition = condition
[docs]
def apply(self, person_id, squeeze_factor):
df = self.sim.population.props
person = df.loc[person_id]
m = self.sim.modules['CardioMetabolicDisorders']
assert person[f'nc_{self.condition}_ever_diagnosed'], "The person is not diagnosed and so should not be " \
"receiving an HSI."
# Check that the person still has the condition
if not person[f'nc_{self.condition}']:
# This person no longer has the condition so will not have this HSI and will drop off of medication
# Return the blank_appt_footprint() so that this HSI does not occupy any time resources
df.at[person_id, f'nc_{self.condition}_on_medication'] = False
return self.sim.modules['HealthSystem'].get_blank_appt_footprint()
# Check that the person is on medication
if not person[f'nc_{self.condition}_on_medication']:
# This person is not on medication so will not have this HSI
# Return the blank_appt_footprint() so that this HSI does not occupy any time resources
return self.sim.modules['HealthSystem'].get_blank_appt_footprint()
# Check availability of medication for condition
if self.get_consumables(
item_codes=self.module.parameters[f'{self.condition}_hsi'].get('medication_item_code').astype(int)
):
# Schedule their next HSI for a refill of medication, one month from now
self.sim.modules['HealthSystem'].schedule_hsi_event(
hsi_event=self,
priority=1,
topen=self.sim.date + DateOffset(months=1),
tclose=self.sim.date + DateOffset(months=1) + DateOffset(days=7)
)
else:
# If medication was not available, the person ceases to be taking medication
df.at[person_id, f'nc_{self.condition}_on_medication'] = False
# If person 'decides to' seek another appointment, schedule a new HSI appointment for tomorrow.
# NB. With a probability of 1.0, this will keep occurring, and the person will never give-up coming back to
# pick-up medication.
if (m.rng.random_sample() <
m.parameters[f'{self.condition}_hsi'].get('pr_seeking_further_appt_if_drug_not_available')):
self.sim.modules['HealthSystem'].schedule_hsi_event(
hsi_event=self,
topen=self.sim.date + pd.DateOffset(days=1),
tclose=self.sim.date + pd.DateOffset(days=15),
priority=1
)
[docs]
def did_not_run(self):
# If this HSI event did not run, then the persons ceases to be taking medication
person_id = self.target
self.sim.population.props.at[person_id, f'nc_{self.condition}_on_medication'] = False
[docs]
class HSI_CardioMetabolicDisorders_SeeksEmergencyCareAndGetsTreatment(HSI_Event, IndividualScopeEventMixin):
"""
This is a Health System Interaction Event.
It is the event when a person is investigated for certain condition following and emergency presentation. Treatment
may be scheduled from this event, and this may take the form of multiple streams of HSI event if more than one
condition is to be treated.
:param events_to_investigate: list of the events for which emergency care is sought
"""
[docs]
def __init__(self, module, person_id, events_to_investigate: List):
super().__init__(module, person_id=person_id)
assert isinstance(module, CardioMetabolicDisorders)
self.TREATMENT_ID = 'CardioMetabolicDisorders_Treatment'
self.EXPECTED_APPT_FOOTPRINT = self.make_appt_footprint({'AccidentsandEmerg': 1})
self.ACCEPTED_FACILITY_LEVEL = '2'
self.events_to_investigate = events_to_investigate
[docs]
def do_for_each_event_to_be_investigated(self, _ev):
"""Do what is required to be done for each event that is being investigated."""
person_id = self.target
df = self.sim.population.props
# Run a test to diagnose whether the person has condition:
dx_result = self.sim.modules['HealthSystem'].dx_manager.run_dx_test(
dx_tests_to_run=f'assess_{_ev}',
hsi_event=self
)
if dx_result:
# Record date of diagnosis
df.at[person_id, f'nc_{_ev}_date_diagnosis'] = self.sim.date
df.at[person_id, f'nc_{_ev}_ever_diagnosed'] = True
if self.module.parameters['prob_care_provided_given_seek_emergency_care'] > self.module.rng.random_sample():
# If care is provided....
if self.get_consumables(
item_codes=self.module.parameters[f'{_ev}_hsi'].get(
'emergency_medication_item_code').astype(int)
):
logger.debug(key='debug', data='Treatment will be provided.')
df.at[person_id, f'nc_{_ev}_on_medication'] = True
df.at[person_id, f'nc_{_ev}_medication_prevents_death'] = \
self.module.rng.rand() < self.module.parameters[f'{_ev}_hsi'].pr_treatment_works
if df.at[person_id, f'nc_{_ev}_medication_prevents_death']:
# Cancel the scheduled death data
df.at[person_id, f'nc_{_ev}_scheduled_date_death'] = pd.NaT
# Remove all symptoms of event instantly
self.sim.modules['SymptomManager'].change_symptom(
person_id=person_id,
symptom_string=f'{_ev}_damage',
add_or_remove='-',
disease_module=self.module)
# Start the person on regular medication
self.sim.modules['HealthSystem'].schedule_hsi_event(
hsi_event=HSI_CardioMetabolicDisorders_StartWeightLossAndMedication(
module=self.module,
person_id=person_id,
condition=_ev
),
priority=0,
topen=self.sim.date,
tclose=None
)
else:
# Consumables not available
logger.debug(key='debug', data='Treatment will not be provided due to no available consumables')
[docs]
def apply(self, person_id, squeeze_factor):
hs = self.sim.modules["HealthSystem"]
df = self.sim.population.props
logger.debug(
key='debug',
data=('This is HSI_CardioMetabolicDisorders_SeeksEmergencyCareAndGetsTreatment: '
f'We are now ready to treat this person {person_id}.'),
)
logger.debug(
key='debug',
data=('This is HSI_CardioMetabolicDisorders_SeeksEmergencyCareAndGetsTreatment: '
f'The squeeze-factor is {squeeze_factor}.'),
)
for _ev in self.events_to_investigate:
self.do_for_each_event_to_be_investigated(_ev)
# Also run a test for hypertension according to some probability, since both events are related to hypertension:
if self.module.rng.rand() < self.module.parameters['hypertension_hsi']['pr_assessed_other_symptoms']:
# Run a test to diagnose whether the person has condition:
dx_result = hs.dx_manager.run_dx_test(
dx_tests_to_run='assess_hypertension',
hsi_event=self
)
df.at[person_id, 'nc_hypertension_date_last_test'] = self.sim.date
if dx_result:
# Record date of diagnosis:
df.at[person_id, 'nc_hypertension_date_diagnosis'] = self.sim.date
df.at[person_id, 'nc_hypertension_ever_diagnosed'] = True
# Start weight loss treatment medication for hypertension
hs.schedule_hsi_event(
hsi_event=HSI_CardioMetabolicDisorders_StartWeightLossAndMedication(
module=self.module,
person_id=person_id,
condition='hypertension'
),
priority=0,
topen=self.sim.date,
tclose=None
)
[docs]
def did_not_run(self):
logger.debug(key='debug', data='HSI_CardioMetabolicDisorders_SeeksEmergencyCareAndGetsTreatment: did not run')