Source code for tlo.methods.epilepsy

from pathlib import Path
from typing import Union

import numpy as np
import pandas as pd

from tlo import DateOffset, Module, Parameter, Property, Types, logging
from tlo.events import IndividualScopeEventMixin, PopulationScopeEventMixin, RegularEvent
from tlo.methods import Metadata
from tlo.methods.causes import Cause
from tlo.methods.demography import InstantaneousDeath
from tlo.methods.healthsystem import HSI_Event
from tlo.methods.symptommanager import Symptom

logger = logging.getLogger(__name__)
logger.setLevel(logging.INFO)


[docs] class Epilepsy(Module):
[docs] def __init__(self, name=None, resourcefilepath=None): super().__init__(name) self.resourcefilepath = resourcefilepath self.item_codes = dict()
INIT_DEPENDENCIES = {'Demography', 'HealthBurden', 'HealthSystem', 'SymptomManager'} # Declare Metadata METADATA = { Metadata.DISEASE_MODULE, Metadata.USES_SYMPTOMMANAGER, Metadata.USES_HEALTHSYSTEM, Metadata.USES_HEALTHBURDEN } # Declare Causes of Death CAUSES_OF_DEATH = { 'Epilepsy': Cause(gbd_causes='Idiopathic epilepsy', label='Epilepsy'), } # Declare Causes of Disability CAUSES_OF_DISABILITY = { 'Epilepsy': Cause(gbd_causes='Idiopathic epilepsy', label='Epilepsy'), } # Module parameters PARAMETERS = { 'init_epil_seiz_status': Parameter(Types.LIST, 'Proportions in each seizure status category at baseline'), 'init_prop_antiepileptic_seiz_stat_1': Parameter( Types.REAL, 'initial proportions on antiepileptic by if seizure status = 1' ), 'init_prop_antiepileptic_seiz_stat_2': Parameter( Types.REAL, 'initial proportions on antiepileptic by if seizure status = 2' ), 'init_prop_antiepileptic_seiz_stat_3': Parameter( Types.REAL, 'initial proportions on antiepileptic by if seizure status = 3' ), 'base_3m_prob_epilepsy': Parameter(Types.REAL, 'base probability of epilepsy per 3 month period if age < 20'), 'rr_epilepsy_age_ge20': Parameter(Types.REAL, 'relative rate of epilepsy if age over 20'), 'prop_inc_epilepsy_seiz_freq': Parameter( Types.REAL, 'proportion of incident epilepsy cases with frequent seizures' ), 'base_prob_3m_seiz_stat_freq_infreq': Parameter( Types.REAL, 'base probability per 3 months of seizure status frequent if current infrequent' ), 'rr_effectiveness_antiepileptics': Parameter( Types.REAL, 'relative rate of seizure status frequent if current infrequent if on antiepileptic' ), 'base_prob_3m_seiz_stat_infreq_freq': Parameter( Types.REAL, 'base probability per 3 months of seizure status infrequent if current frequent' ), 'base_prob_3m_seiz_stat_infreq_none': Parameter( Types.REAL, 'base probability per 3 months of seizure status infrequent if current nonenow' ), 'base_prob_3m_seiz_stat_none_freq': Parameter( Types.REAL, 'base probability per 3 months of seizure status nonenow if current frequent' ), 'base_prob_3m_seiz_stat_none_infreq': Parameter( Types.REAL, 'base probability per 3 months of seizure status nonenow if current infrequent' ), 'base_prob_3m_stop_antiepileptic': Parameter( Types.REAL, 'base probability per 3 months of stopping antiepileptic, if nonenow seizures' ), 'rr_stop_antiepileptic_seiz_infreq_or_freq': Parameter( Types.REAL, 'relative rate of stopping antiepileptic if infrequent or frequent seizures' ), 'base_prob_3m_epi_death': Parameter(Types.REAL, 'base probability per 3 months of epilepsy death'), # these definitions for disability weights are the ones in the global burden of disease list (Salomon) 'daly_wt_epilepsy_severe': Parameter( Types.REAL, 'disability weight for severe epilepsy' 'controlled phase - code 860' ), 'daly_wt_epilepsy_less_severe': Parameter( Types.REAL, 'disability weight for less severe epilepsy' 'controlled phase - code 861' ), 'daly_wt_epilepsy_seizure_free': Parameter( Types.REAL, 'disability weight for less severe epilepsy' 'controlled phase - code 862' ), } """ 860,Severe epilepsy,'Epilepsy, seizures >= once a month','has sudden seizures one or more times each month, with violent muscle contractions and stiffness, loss of consciousness, and loss of urine or bowel control. Between seizures the person has memory loss and difficulty concentrating. ',0.552,0.375,0.71 861,Less severe epilepsy,'Epilepsy, seizures 1-11 per year','has sudden seizures two to five times a year, with violent muscle contractions and stiffness, loss of consciousness, and loss of urine or bowel control.',0.263,0.173,0.367 862,'Seizure-free, treated epilepsy',Generic uncomplicated disease: worry and daily medication, has a chronic disease that requires medication every day and causes some worry but minimal interference with daily activities.,0.049,0.031,0.072 """ # Properties of individuals 'owned' by this module PROPERTIES = { 'ep_seiz_stat': Property( Types.CATEGORICAL, '(0 = never epilepsy, 1 = previous seizures none now, 2 = infrequent seizures, 3 = frequent seizures)', categories=['0', '1', '2', '3'], ), 'ep_antiep': Property(Types.BOOL, 'on antiepileptic'), 'ep_epi_death': Property(Types.BOOL, 'epilepsy death this 3 month period'), 'ep_unified_symptom_code': Property(Types.CATEGORICAL, '', categories=['0', '1', '2', '3']), 'ep_disability': Property(Types.REAL, 'disability weight for current 3 month period'), } # Declaration of how we will refer to any treatments that are related to this disease. TREATMENT_ID = 'antiepileptic'
[docs] def read_parameters(self, data_folder): """Read parameter values from file, if required. Here we just assign parameter values explicitly. :param data_folder: path of a folder supplied to the Simulation containing data files. Typically modules would read a particular file within here. """ # Update parameters from the resource dataframe dfd = pd.read_excel(Path(self.resourcefilepath) / 'epilepsy' / 'ResourceFile_Epilepsy.xlsx', sheet_name='parameter_values') self.load_parameters_from_dataframe(dfd) p = self.parameters if 'HealthBurden' in self.sim.modules.keys(): # get the DALY weight - 860-862 are the sequale codes for epilepsy p['daly_wt_epilepsy_severe'] = self.sim.modules['HealthBurden'].get_daly_weight(sequlae_code=860) p['daly_wt_epilepsy_less_severe'] = self.sim.modules['HealthBurden'].get_daly_weight(sequlae_code=861) p['daly_wt_epilepsy_seizure_free'] = self.sim.modules['HealthBurden'].get_daly_weight(sequlae_code=862) # Register Symptom that this module will use self.sim.modules['SymptomManager'].register_symptom( Symptom.emergency("seizures"))
[docs] def initialise_population(self, population): """Set our property values for the initial population. This method is called by the simulation when creating the initial population, and is responsible for assigning initial values, for every individual, of those properties 'owned' by this module, i.e. those declared in the PROPERTIES dictionary above. """ df = population.props # a shortcut to the data-frame storing data for individuals p = self.parameters rng = self.rng df.loc[df.is_alive, 'ep_seiz_stat'] = '0' df.loc[df.is_alive, 'ep_antiep'] = False df.loc[df.is_alive, 'ep_epi_death'] = False df.loc[df.is_alive, 'ep_disability'] = 0 # allocate initial ep_seiz_stat alive_idx = df.index[df.is_alive] df.loc[alive_idx, 'ep_seiz_stat'] = self.rng.choice(['0', '1', '2', '3'], size=len(alive_idx), p=self.parameters['init_epil_seiz_status']) # Assign those with epilepsy seizure status 2 and 3 the seizure symptom at the start of the simulation dfg = df.index[df.is_alive & ((df.ep_seiz_stat == '2') | (df.ep_seiz_stat == '3'))] self.sim.modules['SymptomManager'].change_symptom( person_id=dfg.to_list(), symptom_string='seizures', add_or_remove='+', disease_module=self ) def allocate_antiepileptic(status, probability): mask = (df.is_alive & (df.ep_seiz_stat == status)) random_draw = rng.random_sample(size=mask.sum()) df.loc[mask, 'ep_antiep'] = random_draw < probability # allocate initial on antiepileptic seiz status 1, 2 and 3 allocate_antiepileptic('1', p['init_prop_antiepileptic_seiz_stat_1']) allocate_antiepileptic('2', p['init_prop_antiepileptic_seiz_stat_2']) allocate_antiepileptic('3', p['init_prop_antiepileptic_seiz_stat_3'])
[docs] def initialise_simulation(self, sim): """Get ready for simulation start. This method is called just before the main simulation loop begins, and after all modules have read their parameters and the initial population has been created. It is a good place to add initial events to the event queue. Here we add our three-monthly event to poll the population for epilepsy starting or stopping. """ epilepsy_poll = EpilepsyEvent(self) sim.schedule_event(epilepsy_poll, sim.date + DateOffset(months=3)) event = EpilepsyLoggingEvent(self) sim.schedule_event(event, sim.date + DateOffset(months=0)) # Get item_codes for the consumables used in the HSI hs = self.sim.modules['HealthSystem'] self.item_codes = dict() self.item_codes['phenobarbitone'] = hs.get_item_code_from_item_name("Phenobarbital, 100 mg") self.item_codes['carbamazepine'] = hs.get_item_code_from_item_name('Carbamazepine 200mg_1000_CMST') self.item_codes['phenytoin'] = hs.get_item_code_from_item_name('Phenytoin sodium 100mg_1000_CMST')
[docs] def on_birth(self, mother_id, child_id): """Initialise our properties for a newborn individual. This is called by the simulation whenever a new person is born. :param mother: the mother for this child :param child: the new child """ df = self.sim.population.props df.at[child_id, 'ep_seiz_stat'] = '0' df.at[child_id, 'ep_antiep'] = False df.at[child_id, 'ep_epi_death'] = False df.at[child_id, 'ep_disability'] = 0
[docs] def report_daly_values(self): df = self.sim.population.props # shortcut to population properties dataframe return df.loc[df.is_alive, 'ep_disability']
[docs] def transition_seizure_stat(self): """ This function handles all transitions in epilepsy seizure status, for those on and off anti epileptics. The function determines the current seizure status of those with epilepsy and based on their original status, and whether they are on anti epileptics determines a new seizure status :return: """ # Get the parameters used to determine transitions between seizure states p = self.parameters prop_transition_1_2 = p['base_prob_3m_seiz_stat_infreq_none'] prop_transition_2_1 = p['base_prob_3m_seiz_stat_none_infreq'] prop_transition_2_3 = p['base_prob_3m_seiz_stat_freq_infreq'] prop_transition_3_1 = p['base_prob_3m_seiz_stat_none_freq'] prop_transition_3_2 = p['base_prob_3m_seiz_stat_infreq_freq'] rr_effectiveness_antiepileptics = p['rr_effectiveness_antiepileptics'] # get the population and the current seizure status of those with epilepsy and those who are on anti epileptics df = self.sim.population.props population_with_seizure_status_1 = df.index[df.is_alive & (df.ep_seiz_stat == '1')] seizure_status_1_on_anti_epileptics = df.index[df.is_alive & (df.ep_seiz_stat == '1') & df.ep_antiep] population_with_seizure_status_2 = df.index[df.is_alive & (df.ep_seiz_stat == '2')] seizure_status_2_on_anti_epileptics = df.index[df.is_alive & (df.ep_seiz_stat == '2') & df.ep_antiep] population_with_seizure_status_3 = df.index[df.is_alive & (df.ep_seiz_stat == '3')] seizure_status_3_on_anti_epileptics = df.index[df.is_alive & (df.ep_seiz_stat == '3') & df.ep_antiep] # Determine who will transition from seizure status 1 to 2, first create a random number to determine likelihood # of transition random_draw_1_2 = self.rng.random_sample(size=len(population_with_seizure_status_1)) # Get the base probability of transitioning from state 1 to state 2 probability_of_transition_1_2 = pd.DataFrame([prop_transition_1_2] * len(random_draw_1_2), columns=['prob'], index=population_with_seizure_status_1) # Reduce the risk of worsening seizures if on anti epileptics probability_of_transition_1_2.loc[seizure_status_1_on_anti_epileptics, 'prob'] /= \ rr_effectiveness_antiepileptics # determine who will transition between seizure state 1 to 2 changing_1_2 = population_with_seizure_status_1[probability_of_transition_1_2['prob'] > random_draw_1_2] # update seizure status df.loc[changing_1_2, 'ep_seiz_stat'] = '2' # Determine if those with seizure status 2 increase or decrease in severity, first get random draws for each # transition state random_draw_2_1 = self.rng.random_sample(size=len(population_with_seizure_status_2)) random_draw_2_3 = self.rng.random_sample(size=len(population_with_seizure_status_2)) # create a dataframe for the transition probabilities probability_of_transition_2 = pd.DataFrame({'prob_down': [prop_transition_2_1] * len(random_draw_2_1), 'prob_up': [prop_transition_2_3] * len(random_draw_2_1)}, index=population_with_seizure_status_2) # Increase the likelihood of reducing seizure status for those on anti epileptics probability_of_transition_2.loc[seizure_status_2_on_anti_epileptics, 'prob_down'] *= \ rr_effectiveness_antiepileptics # Decrease the likelihood of reducing seizure status for those on anti epileptics probability_of_transition_2.loc[seizure_status_2_on_anti_epileptics, 'prob_up'] /= \ rr_effectiveness_antiepileptics # Establish who has a changing seizure status changing_2_1 = population_with_seizure_status_2[probability_of_transition_2['prob_down'] > random_draw_2_1] changing_2_3 = population_with_seizure_status_2[probability_of_transition_2['prob_up'] > random_draw_2_3] # If someone with seizure status 2 has been selected to both increase and decrease in severity, choose a # transition direction based on the likelihood of transitioning states both_up_down_seiz_stat_2 = changing_2_1.intersection(changing_2_3) if len(both_up_down_seiz_stat_2) > 0: for person in both_up_down_seiz_stat_2: chosen_direction = self.rng.choice( ['1', '3'], p=np.divide([prop_transition_2_1, prop_transition_2_3], sum([prop_transition_2_1, prop_transition_2_3])) ) df.loc[person, 'ep_seiz_stat'] = chosen_direction # Drop those who have already had their seizure status changed from the indexs changing_2_1 and changing_2_3 changing_2_1 = changing_2_1.drop(both_up_down_seiz_stat_2) changing_2_3 = changing_2_3.drop(both_up_down_seiz_stat_2) df.loc[changing_2_1, 'ep_seiz_stat'] = '1' df.loc[changing_2_3, 'ep_seiz_stat'] = '3' # Determine if those with seizure status 3 decrease in severity and by how much, first get random draws for each # transition state random_draw_3_1 = self.rng.random_sample(size=len(population_with_seizure_status_3)) random_draw_3_2 = self.rng.random_sample(size=len(population_with_seizure_status_3)) # create a dataframe for the transition probabilities probability_of_transition_3 = pd.DataFrame({'prob_down_1': [prop_transition_3_2] * len(random_draw_3_2), 'prob_down_2': [prop_transition_3_1] * len(random_draw_3_1)}, index=population_with_seizure_status_3) # Increase the likelihood of reducing seizure status for those on anti epileptics probability_of_transition_3.loc[seizure_status_3_on_anti_epileptics, 'prob_down_1'] *= \ rr_effectiveness_antiepileptics # Decrease the likelihood of reducing seizure status for those on anti epileptics probability_of_transition_3.loc[seizure_status_3_on_anti_epileptics, 'prob_down_2'] /= \ rr_effectiveness_antiepileptics # Establish who has a changing seizure status changing_3_2 = population_with_seizure_status_3[probability_of_transition_3['prob_down_1'] > random_draw_3_2] changing_3_1 = population_with_seizure_status_3[probability_of_transition_3['prob_down_2'] > random_draw_3_1] # If someone with seizure status 2 has been selected to both increase and decrease in severity, choose a # transition direction based on the likelihood of transitioning states both_down_seiz_stat_3 = changing_3_1.intersection(changing_3_2) if len(both_down_seiz_stat_3) > 0: for person in both_down_seiz_stat_3: chosen_direction = self.rng.choice( ['1', '2'], p=np.divide([prop_transition_3_1, prop_transition_3_2], sum([prop_transition_3_1, prop_transition_3_2])) ) df.loc[person, 'ep_seiz_stat'] = chosen_direction # Drop those who have already had their seizure status changed from the indexs changing_2_1 and changing_2_3 changing_3_1 = changing_3_1.drop(both_down_seiz_stat_3) changing_3_2 = changing_3_2.drop(both_down_seiz_stat_3) df.loc[changing_3_1, 'ep_seiz_stat'] = '1' df.loc[changing_3_2, 'ep_seiz_stat'] = '2'
[docs] def stop_antiep(self, indices, probability): """stop individuals on antiep with given probability""" df = self.sim.population.props df.loc[indices, 'ep_antiep'] = probability < self.rng.random_sample(size=len(indices))
[docs] def get_best_available_medicine(self, hsi_event) -> Union[None, str]: """Returns the best available medicine (as string), or None if none are available""" # Check what drugs are available. (`to_log` is set to false in order to not actually request the consumables # from the health system before we know what is available) possible_treatments = { 'phenobarbitone': hsi_event.get_consumables( self.item_codes['phenobarbitone'], to_log=False ), 'carbamazepine': hsi_event.get_consumables( self.item_codes['carbamazepine'], to_log=False ), 'phenytoin': hsi_event.get_consumables( self.item_codes['phenytoin'], to_log=False ), } # Now we know which treatments are available, we will determine the most preferable treatment by finding the # index in available_treatments of the first True value: best_option_index = next((i for i, j in enumerate(possible_treatments.values()) if j), None) if best_option_index is not None: # At least one of the treatment is available: Return the name of the most preferable medicine return list(possible_treatments.keys())[best_option_index] else: # None of the treatment is available: return None return None
[docs] class EpilepsyEvent(RegularEvent, PopulationScopeEventMixin): """The regular event that actually changes individuals' epilepsy status Regular events automatically reschedule themselves at a fixed frequency, and thus implement discrete timestep type behaviour. The frequency is specified when calling the base class constructor in our __init__ method. """
[docs] def __init__(self, module): """Create a new depr event. We need to pass the frequency at which we want to occur to the base class constructor using super(). We also pass the module that created this event, so that random number generators can be scoped per-module. :param module: the module that created this event """ super().__init__(module, frequency=DateOffset(months=3)) p = module.parameters self.base_3m_prob_epilepsy = p['base_3m_prob_epilepsy'] self.rr_epilepsy_age_ge20 = p['rr_epilepsy_age_ge20'] self.prop_inc_epilepsy_seiz_freq = p['prop_inc_epilepsy_seiz_freq'] self.base_prob_3m_stop_antiepileptic = p['base_prob_3m_stop_antiepileptic'] self.rr_stop_antiepileptic_seiz_infreq_or_freq = p['rr_stop_antiepileptic_seiz_infreq_or_freq'] self.base_prob_3m_epi_death = p['base_prob_3m_epi_death'] self.daly_wt_epilepsy_severe = p['daly_wt_epilepsy_severe'] self.daly_wt_epilepsy_less_severe = p['daly_wt_epilepsy_less_severe'] self.daly_wt_epilepsy_seizure_free = p['daly_wt_epilepsy_seizure_free']
[docs] def apply(self, population): """Apply this event to the population. For efficiency, we use pandas operations to scan the entire population in bulk. :param population: the current population """ ep = self.module df = population.props # set ep_epi_death back to False after death df.loc[~df.is_alive & df.ep_epi_death, 'ep_epi_death'] = False df.loc[df.is_alive, 'ep_disability'] = 0 # update ep_seiz_stat for people ep_seiz_stat = 0 # Find who does not have epilepsy alive_seiz_stat_0_idx = df.index[df.is_alive & (df.ep_seiz_stat == '0')] # Find who does not have epilepsy and is 20 & over ge20_seiz_stat_0_idx = df.index[df.is_alive & (df.ep_seiz_stat == '0') & (df.age_years >= 20)] # Create a pandas series of the length of people who are alive, this is the basic probability of people # developing epilepsy eff_prob_epilepsy = pd.Series(self.base_3m_prob_epilepsy, index=alive_seiz_stat_0_idx) # Find the indexes of people who are aged 20 and above and increase their risk of developing epilepsy eff_prob_epilepsy.loc[ge20_seiz_stat_0_idx] *= self.rr_epilepsy_age_ge20 # Create a list of random numbers, one per person, to determine who will develop epilepsy random_draw_01 = self.module.rng.random_sample(size=len(alive_seiz_stat_0_idx)) # If a person's number is less than their likelihood of developing epilepsy, then they will now have epilepsy epi_now = eff_prob_epilepsy > random_draw_01 # For those who have developed epilepsy, we need to work out what their seizure status will be, create another # list of random numbers to determine their seizure status random_draw_02 = self.module.rng.random_sample(size=len(alive_seiz_stat_0_idx)) # if a person's random number is less than the probability of frequent seizures, then update they will have a # seizure status of 2 assigned to them seiz_stat_3_idx = alive_seiz_stat_0_idx[epi_now & (self.prop_inc_epilepsy_seiz_freq > random_draw_02)] # if a person's random number is greater than the probability of frequent seizures, then update they will have a # seizure status of 1 assigned to them seiz_stat_2_idx = alive_seiz_stat_0_idx[epi_now & (self.prop_inc_epilepsy_seiz_freq <= random_draw_02)] # based on the above predictions, update each person's seizure status df.loc[seiz_stat_3_idx, 'ep_seiz_stat'] = '3' df.loc[seiz_stat_2_idx, 'ep_seiz_stat'] = '2' # Calculate & log the incidence of epilepsy n_incident_epilepsy = epi_now.sum() n_alive = df.is_alive.sum() incidence_epilepsy = (n_incident_epilepsy * 4 * 100000) / n_alive if n_alive > 0 else 0 logger.info( key='inc_epilepsy', data={ 'incidence_epilepsy': incidence_epilepsy, 'n_incident_epilepsy': n_incident_epilepsy, 'n_alive': n_alive } ) # For those who are not on anti epileptics, determine whether their seizure status changes in severity ep.transition_seizure_stat() # save all individuals that are currently on anti-epileptics (seizure status: 2 & 3 or 1) alive_seiz_stat_1_antiep_idx = df.index[df.is_alive & (df.ep_seiz_stat == '1') & df.ep_antiep] alive_seiz_stat_2_or_3_antiep_idx = df.index[df.is_alive & (df.ep_seiz_stat.isin(['2', '3'])) & df.ep_antiep] # rate of stop ep_antiep if ep_seiz_stat = 1 ep.stop_antiep(alive_seiz_stat_1_antiep_idx, self.base_prob_3m_stop_antiepileptic) # rate of stop ep_antiep if ep_seiz_stat = 2 or 3 ep.stop_antiep(alive_seiz_stat_2_or_3_antiep_idx, self.base_prob_3m_stop_antiepileptic * self.rr_stop_antiepileptic_seiz_infreq_or_freq) # disability # note disability weights from gbd do not map fully onto epilepsy states in model - could re-visit # this proposed mapping below df.loc[df.is_alive & (df.ep_seiz_stat == '1'), 'ep_disability'] = self.daly_wt_epilepsy_seizure_free df.loc[df.is_alive & (df.ep_seiz_stat == '2'), 'ep_disability'] = self.daly_wt_epilepsy_less_severe df.loc[df.is_alive & (df.ep_seiz_stat == '3'), 'ep_disability'] = self.daly_wt_epilepsy_severe # Work out who is having seizures which may lead to death alive_seiz_stat_2_or_3_idx = df.index[df.is_alive & (df.ep_seiz_stat.isin(['2', '3']))] # determine if those having seizures will die as a result of epilepsy chosen = self.base_prob_3m_epi_death > self.module.rng.random_sample(size=len(alive_seiz_stat_2_or_3_idx)) # update the ep_epi_death property df.loc[alive_seiz_stat_2_or_3_idx, 'ep_epi_death'] = chosen # schedule any deaths for individual_id in alive_seiz_stat_2_or_3_idx[chosen]: self.sim.schedule_event( InstantaneousDeath(self.module, individual_id, 'Epilepsy'), self.sim.date ) # -------------------- UPDATING OF SYMPTOM OF seizures OVER TIME -------------------------------- dfg = df.index[df.is_alive & ((df.ep_seiz_stat == '2') | (df.ep_seiz_stat == '3'))] self.sim.modules['SymptomManager'].change_symptom( person_id=dfg.to_list(), symptom_string='seizures', add_or_remove='+', disease_module=self.module ) dfh = df.index[df.is_alive & ((df.ep_seiz_stat == '0') | (df.ep_seiz_stat == '1'))] self.sim.modules['SymptomManager'].change_symptom( person_id=dfh.to_list(), symptom_string='seizures', add_or_remove='-', disease_module=self.module )
[docs] class EpilepsyLoggingEvent(RegularEvent, PopulationScopeEventMixin):
[docs] def __init__(self, module): # run this event every 3 month self.repeat = 3 super().__init__(module, frequency=DateOffset(months=self.repeat))
[docs] def apply(self, population): """Get some summary statistics and log them""" df = population.props status_groups = df[ ["ep_seiz_stat", "is_alive", "ep_antiep"] ].groupby('ep_seiz_stat').sum() n_seiz_stat_1_3 = sum(status_groups.iloc[1:].is_alive) n_seiz_stat_2_3 = sum(status_groups.iloc[2:].is_alive) n_antiep = (df.is_alive & df.ep_antiep).sum() n_epi_death = df.ep_epi_death.sum() status_groups['prop_seiz_stats'] = status_groups.is_alive / sum(status_groups.is_alive) status_groups['prop_seiz_stat_on_anti_ep'] = status_groups['ep_antiep'] / status_groups.is_alive status_groups['prop_seiz_stat_on_anti_ep'] = status_groups['prop_seiz_stat_on_anti_ep'].fillna(0) epi_death_rate = \ (n_epi_death * 4 * 1000) / n_seiz_stat_2_3 if n_seiz_stat_2_3 > 0 else 0 cum_deaths = (~df.is_alive).sum() logger.info(key='epilepsy_logging', data={ 'prop_seiz_stat_0': status_groups['prop_seiz_stats'].iloc[0], 'prop_seiz_stat_1': status_groups['prop_seiz_stats'].iloc[1], 'prop_seiz_stat_2': status_groups['prop_seiz_stats'].iloc[2], 'prop_seiz_stat_3': status_groups['prop_seiz_stats'].iloc[3], 'prop_antiepilep_seiz_stat_0': status_groups['prop_seiz_stat_on_anti_ep'].iloc[0], 'prop_antiepilep_seiz_stat_1': status_groups['prop_seiz_stat_on_anti_ep'].iloc[1], 'prop_antiepilep_seiz_stat_2': status_groups['prop_seiz_stat_on_anti_ep'].iloc[2], 'prop_antiepilep_seiz_stat_3': status_groups['prop_seiz_stat_on_anti_ep'].iloc[3], 'n_epi_death': n_epi_death, 'cum_deaths': cum_deaths, 'epi_death_rate': epi_death_rate, 'n_seiz_stat_1_3': n_seiz_stat_1_3, 'n_seiz_stat_2_3': n_seiz_stat_2_3, 'n_antiep': n_antiep })
[docs] class HSI_Epilepsy_Start_Anti_Epileptic(HSI_Event, IndividualScopeEventMixin): """ This is a Health System Interaction Event. """
[docs] def __init__(self, module, person_id): super().__init__(module, person_id=person_id) # Define the necessary information for an HSI self.TREATMENT_ID = 'Epilepsy_Treatment_Start' self.EXPECTED_APPT_FOOTPRINT = self.make_appt_footprint({'Over5OPD': 1}) self.ACCEPTED_FACILITY_LEVEL = '1b'
[docs] def apply(self, person_id, squeeze_factor): df = self.sim.population.props hs = self.sim.modules["HealthSystem"] # Check what drugs are available best_available_medicine = self.module.get_best_available_medicine(self) if best_available_medicine is not None: # Request the medicine from the health system self.get_consumables(self.module.item_codes[best_available_medicine]) # Update this person's properties to show that they are currently on medication df.at[person_id, 'ep_antiep'] = True # Schedule a follow-up for 3 months: hs.schedule_hsi_event( hsi_event=HSI_Epilepsy_Follow_Up( module=self.module, person_id=person_id, ), topen=self.sim.date + DateOffset(months=3), tclose=None, priority=0 ) else: # If no medicine is available, run this HSI again next month self.module.sim.modules['HealthSystem'].schedule_hsi_event(hsi_event=self, topen=self.sim.date + pd.DateOffset(months=1), tclose=None, priority=0)
[docs] class HSI_Epilepsy_Follow_Up(HSI_Event, IndividualScopeEventMixin):
[docs] def __init__(self, module, person_id): super().__init__(module, person_id=person_id) self._MAX_NUMBER_OF_FAILED_ATTEMPTS_BEFORE_DEFAULTING = 2 self._DEFAULT_APPT_FOOTPRINT = self.make_appt_footprint({'Over5OPD': 1}) self._REPEATED_APPT_FOOTPRINT = self.make_appt_footprint({'PharmDispensing': 1}) self.TREATMENT_ID = "Epilepsy_Treatment_Followup" self.EXPECTED_APPT_FOOTPRINT = self._DEFAULT_APPT_FOOTPRINT self.ACCEPTED_FACILITY_LEVEL = '1b' self._counter_of_failed_attempts_due_to_unavailable_medicines = 0
[docs] def apply(self, person_id, squeeze_factor): df = self.sim.population.props hs = self.sim.modules["HealthSystem"] if not df.at[person_id, 'is_alive']: return hs.get_blank_appt_footprint() # If the person does not remain on anti-epileptics, do nothing: if not df.at[person_id, 'ep_antiep']: return hs.get_blank_appt_footprint() # Request the medicine best_available_medicine = self.module.get_best_available_medicine(self) if best_available_medicine is not None: # The medicine is available, so request it self.get_consumables(self.module.item_codes[best_available_medicine]) # Reset counter of "failed attempts" and put the appointment for the next occurrence to the usual self._counter_of_failed_attempts_due_to_unavailable_medicines = 0 self.EXPECTED_APPT_FOOTPRINT = self._DEFAULT_APPT_FOOTPRINT # Schedule a reoccurrence of this follow-up in 3 months if ep_seiz_stat == '3', # else, schedule this reoccurrence of it in 1 year (i.e., if ep_seiz_stat == '2') hs.schedule_hsi_event( hsi_event=self, topen=self.sim.date + DateOffset(months=3 if df.at[person_id, 'ep_seiz_stat'] == '3' else 12), tclose=None, priority=0 ) elif ( self._counter_of_failed_attempts_due_to_unavailable_medicines < self._MAX_NUMBER_OF_FAILED_ATTEMPTS_BEFORE_DEFAULTING ): # Nothing is available currently: schedule a recurrence of this appointment in one month, with a modified # footprint. (This will repeat a certain number of time before the person defaults to being off # anti-epileptics: see next clause.) self._counter_of_failed_attempts_due_to_unavailable_medicines += 1 self.EXPECTED_APPT_FOOTPRINT = self._REPEATED_APPT_FOOTPRINT hs.schedule_hsi_event( hsi_event=self, topen=self.sim.date + DateOffset(months=1), tclose=None, priority=0 ) else: # No medicine is available and the maximum number of repeats has been reached: The person will default # to being off the anti-epileptics and no further follow-ups are scheduled. df.at[person_id, 'ep_antiep'] = False