"""
Road traffic injury module.
"""
from __future__ import annotations
from pathlib import Path
from typing import TYPE_CHECKING, List
import numpy as np
import pandas as pd
from tlo import 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.causes import Cause
from tlo.methods.hsi_event import HSI_Event
from tlo.methods.hsi_generic_first_appts import GenericFirstAppointmentsMixin
from tlo.methods.symptommanager import Symptom
if TYPE_CHECKING:
from tlo.methods.hsi_generic_first_appts import HSIEventScheduler
from tlo.population import IndividualProperties
# ---------------------------------------------------------------------------------------------------------
# MODULE DEFINITIONS
# ---------------------------------------------------------------------------------------------------------
logger = logging.getLogger(__name__)
logger.setLevel(logging.DEBUG)
[docs]
class RTI(Module, GenericFirstAppointmentsMixin):
"""
The road traffic injuries module for the TLO model, handling all injuries related to road traffic accidents.
"""
[docs]
def __init__(self, name=None, resourcefilepath=None):
# NB. Parameters passed to the module can be inserted in the __init__ definition.
super().__init__(name)
self.resourcefilepath = resourcefilepath
self.ASSIGN_INJURIES_AND_DALY_CHANGES = None
self.cons_item_codes = None # (Will store consumable item codes)
INIT_DEPENDENCIES = {"SymptomManager",
"HealthBurden"}
ADDITIONAL_DEPENDENCIES = {
'Demography',
'Lifestyle',
'HealthSystem',
}
INJURY_INDICES = range(1, 9)
INJURY_COLUMNS = [f'rt_injury_{i}' for i in INJURY_INDICES]
DATE_TO_REMOVE_DALY_COLUMNS = [f'rt_date_to_remove_daly_{i}' for i in INJURY_INDICES]
# Bi-directional map from/to injury columns to/from date to remove daly columns
INJURY_DATE_COLUMN_MAP = {
**{f'rt_injury_{i}': f'rt_date_to_remove_daly_{i}' for i in INJURY_INDICES},
**{f'rt_date_to_remove_daly_{i}': f'rt_injury_{i}' for i in INJURY_INDICES},
}
INJURY_CODES = ['none', '112', '113', '133a', '133b', '133c', '133d', '134a', '134b', '135', '1101', '1114', '211',
'212', '241', '2101', '2114', '291', '342', '343', '361', '363', '322', '323', '3101', '3113',
'412', '414', '461', '463', '453a', '453b', '441', '442', '443', '4101', '4113', '552', '553',
'554', '5101', '5113', '612', '673a', '673b', '674a', '674b', '675a', '675b', '676', '712a',
'712b', '712c', '722', '782a', '782b', '782c', '783', '7101', '7113', '811', '813do', '812',
'813eo', '813a', '813b', '813bo', '813c', '813co', '822a', '822b', '882', '883', '884', '8101',
'8113', 'P133a', 'P133b', 'P133c', 'P133d', 'P134a', 'P134b', 'P135', 'P673a', 'P673b', 'P674a',
'P674b', 'P675a', 'P675b', 'P676', 'P782a', 'P782b', 'P782c', 'P783', 'P882', 'P883', 'P884']
SWAPPING_CODES = ['712b', '812', '3113', '4113', '5113', '7113', '8113', '813a', '813b', 'P673a', 'P673b', 'P674a',
'P674b', 'P675a', 'P675b', 'P676', 'P782b', 'P783', 'P883', 'P884', '813bo', '813co', '813do',
'813eo']
INJURIES_REQ_IMAGING = ['112', '113', '211', '212', '412', '414', '612', '712a', '712b', '712c', '811', '812',
'813a', '813b', '813c', '822a', '822b', '813bo', '813co', '813do', '813eo', '673', '674',
'675', '676', '322', '323', '722', '342', '343', '441', '443', '453', '133', '134', '135',
'552', '553', '554', '342', '343', '441', '443', '453', '361', '363', '461', '463']
FRACTURE_CODES = ['112', '113', '211', '212', '412', '414', '612', '712', '811', '812', '813']
NO_TREATMENT_RECOVERY_TIMES_IN_DAYS = {
'112': 49,
'113': 49,
'1101': 7,
'211': 49,
'212': 49,
'241': 7,
'2101': 7,
'291': 7,
'342': 42,
'343': 42,
'361': 7,
'363': 14,
'322': 42,
'323': 42,
'3101': 7,
'3113': 56,
'412': 35,
'414': 365,
'461': 7,
'463': 14,
'453a': 84,
'453b': 84,
'441': 14,
'442': 14,
'4101': 7,
'552': 90,
'553': 90,
'554': 90,
'5101': 7,
'5113': 56,
'612': 63,
'712a': 70,
'712b': 70,
'722': 84,
'7101': 7,
'7113': 56,
'813do': 240,
'811': 70,
'812': 70,
'813eo': 240,
'813bo': 240,
'813co': 240,
'822a': 60,
'822b': 180,
'8101': 7,
'8113': 56
}
# Module parameters
PARAMETERS = {
'base_rate_injrti': Parameter(
Types.REAL,
'Base rate of RTI per year',
),
'rr_injrti_age04': Parameter(
Types.REAL,
'risk ratio of RTI in age 0-4 compared to base rate of RTI'
),
'rr_injrti_age59': Parameter(
Types.REAL,
'risk ratio of RTI in age 5-9 compared to base rate of RTI'
),
'rr_injrti_age1017': Parameter(
Types.REAL,
'risk ratio of RTI in age 10-17 compared to base rate of RTI'
),
'rr_injrti_age1829': Parameter(
Types.REAL,
'risk ratio of RTI in age 18-29 compared to base rate of RTI',
),
'rr_injrti_age3039': Parameter(
Types.REAL,
'risk ratio of RTI in age 30-39 compared to base rate of RTI',
),
'rr_injrti_age4049': Parameter(
Types.REAL,
'risk ratio of RTI in age 40-49 compared to base rate of RTI',
),
'rr_injrti_age5059': Parameter(
Types.REAL,
'risk ratio of RTI in age 50-59 compared to base rate of RTI',
),
'rr_injrti_age6069': Parameter(
Types.REAL,
'risk ratio of RTI in age 60-69 compared to base rate of RTI',
),
'rr_injrti_age7079': Parameter(
Types.REAL,
'risk ratio of RTI in age 70-79 compared to base rate of RTI',
),
'rr_injrti_male': Parameter(
Types.REAL,
'risk ratio of RTI when male compared to females',
),
'rr_injrti_excessalcohol': Parameter(
Types.REAL,
'risk ratio of RTI in those that consume excess alcohol compared to those who do not'
),
'imm_death_proportion_rti': Parameter(
Types.REAL,
'Proportion of those involved in an RTI that die at site of accident or die before seeking medical '
'intervention'
),
'prob_bleeding_leads_to_shock': Parameter(
Types.REAL,
'The proportion of those with heavily bleeding injuries who go into shock'
),
'prob_death_iss_less_than_9': Parameter(
Types.REAL,
'Proportion of people who pass away in the following month after medical treatment for injuries with an ISS'
'score less than or equal to 9'
),
'prob_death_iss_10_15': Parameter(
Types.REAL,
'Proportion of people who pass away in the following month after medical treatment for injuries with an ISS'
'score from 10 to 15'
),
'prob_death_iss_16_24': Parameter(
Types.REAL,
'Proportion of people who pass away in the following month after medical treatment for injuries with an ISS'
'score from 16 to 24'
),
'prob_death_iss_25_35': Parameter(
Types.REAL,
'Proportion of people who pass away in the following month after medical treatment for injuries with an ISS'
'score from 25 to 34'
),
'prob_death_iss_35_plus': Parameter(
Types.REAL,
'Proportion of people who pass away in the following month after medical treatment for injuries with an ISS'
'score 35 and above'
),
'prob_perm_disability_with_treatment_severe_TBI': Parameter(
Types.REAL,
'probability that someone with a treated severe TBI is permanently disabled'
),
'prob_death_TBI_SCI_no_treatment': Parameter(
Types.REAL,
'probability that someone with a spinal cord injury will die without treatment'
),
'prop_death_burns_no_treatment': Parameter(
Types.REAL,
'probability that someone with a burn injury will die without treatment'
),
'prob_death_fractures_no_treatment': Parameter(
Types.REAL,
'probability that someone with a fracture injury will die without treatment'
),
'prob_TBI_require_craniotomy': Parameter(
Types.REAL,
'probability that someone with a traumatic brain injury will require a craniotomy surgery'
),
'prob_exploratory_laparotomy': Parameter(
Types.REAL,
'probability that someone with an internal organ injury will require a exploratory_laparotomy'
),
'prob_depressed_skull_fracture': Parameter(
Types.REAL,
'Probability that a skull fracture will be depressed and therefore require surgery'
),
'prob_mild_burns': Parameter(
Types.REAL,
'Probability that a burn within a region will result in < 10% total body surface area'
),
'prob_dislocation_requires_surgery': Parameter(
Types.REAL,
'Probability that a dislocation will require surgery to relocate the joint.'
),
'number_of_injured_body_regions_distribution': Parameter(
Types.LIST,
'The distribution of number of injured AIS body regions, used to decide how many injuries a person has'
),
'injury_location_distribution': Parameter(
Types.LIST,
'The distribution of where injuries are located in the body, based on the AIS body region definition'
),
# Length of stay
'mean_los_ISS_less_than_4': Parameter(
Types.REAL,
'Mean length of stay for someone with an ISS score < 4'
),
'sd_los_ISS_less_than_4': Parameter(
Types.REAL,
'Standard deviation in length of stay for someone with an ISS score < 4'
),
'mean_los_ISS_4_to_8': Parameter(
Types.REAL,
'Mean length of stay for someone with an ISS score between 4 and 8'
),
'sd_los_ISS_4_to_8': Parameter(
Types.REAL,
'Standard deviation in length of stay for someone with an ISS score between 4 and 8'
),
'mean_los_ISS_9_to_15': Parameter(
Types.REAL,
'Mean length of stay for someone with an ISS score between 9 and 15'
),
'sd_los_ISS_9_to_15': Parameter(
Types.REAL,
'Standard deviation in length of stay for someone with an ISS score between 9 and 15'
),
'mean_los_ISS_16_to_24': Parameter(
Types.REAL,
'Mean length of stay for someone with an ISS score between 16 and 24'
),
'sd_los_ISS_16_to_24': Parameter(
Types.REAL,
'Standard deviation in length of stay for someone with an ISS score between 16 and 24'
),
'mean_los_ISS_more_than_25': Parameter(
Types.REAL,
'Mean length of stay for someone with an ISS score between 16 and 24'
),
'sd_los_ISS_more_that_25': Parameter(
Types.REAL,
'Standard deviation in length of stay for someone with an ISS score between 16 and 24'
),
# DALY weights
'daly_wt_unspecified_skull_fracture': Parameter(
Types.REAL,
'daly_wt_unspecified_skull_fracture - code 1674'
),
'daly_wt_basilar_skull_fracture': Parameter(
Types.REAL,
'daly_wt_basilar_skull_fracture - code 1675'
),
'daly_wt_epidural_hematoma': Parameter(
Types.REAL,
'daly_wt_epidural_hematoma - code 1676'
),
'daly_wt_subdural_hematoma': Parameter(
Types.REAL,
'daly_wt_subdural_hematoma - code 1677'
),
'daly_wt_subarachnoid_hematoma': Parameter(
Types.REAL,
'daly_wt_subarachnoid_hematoma - code 1678'
),
'daly_wt_brain_contusion': Parameter(
Types.REAL,
'daly_wt_brain_contusion - code 1679'
),
'daly_wt_intraventricular_haemorrhage': Parameter(
Types.REAL,
'daly_wt_intraventricular_haemorrhage - code 1680'
),
'daly_wt_diffuse_axonal_injury': Parameter(
Types.REAL,
'daly_wt_diffuse_axonal_injury - code 1681'
),
'daly_wt_subgaleal_hematoma': Parameter(
Types.REAL,
'daly_wt_subgaleal_hematoma - code 1682'
),
'daly_wt_midline_shift': Parameter(
Types.REAL,
'daly_wt_midline_shift - code 1683'
),
'daly_wt_facial_fracture': Parameter(
Types.REAL,
'daly_wt_facial_fracture - code 1684'
),
'daly_wt_facial_soft_tissue_injury': Parameter(
Types.REAL,
'daly_wt_facial_soft_tissue_injury - code 1685'
),
'daly_wt_eye_injury': Parameter(
Types.REAL,
'daly_wt_eye_injury - code 1686'
),
'daly_wt_neck_soft_tissue_injury': Parameter(
Types.REAL,
'daly_wt_neck_soft_tissue_injury - code 1687'
),
'daly_wt_neck_internal_bleeding': Parameter(
Types.REAL,
'daly_wt_neck_internal_bleeding - code 1688'
),
'daly_wt_neck_dislocation': Parameter(
Types.REAL,
'daly_wt_neck_dislocation - code 1689'
),
'daly_wt_chest_wall_bruises_hematoma': Parameter(
Types.REAL,
'daly_wt_chest_wall_bruises_hematoma - code 1690'
),
'daly_wt_hemothorax': Parameter(
Types.REAL,
'daly_wt_hemothorax - code 1691'
),
'daly_wt_lung_contusion': Parameter(
Types.REAL,
'daly_wt_lung_contusion - code 1692'
),
'daly_wt_diaphragm_rupture': Parameter(
Types.REAL,
'daly_wt_diaphragm_rupture - code 1693'
),
'daly_wt_rib_fracture': Parameter(
Types.REAL,
'daly_wt_rib_fracture - code 1694'
),
'daly_wt_flail_chest': Parameter(
Types.REAL,
'daly_wt_flail_chest - code 1695'
),
'daly_wt_chest_wall_laceration': Parameter(
Types.REAL,
'daly_wt_chest_wall_laceration - code 1696'
),
'daly_wt_closed_pneumothorax': Parameter(
Types.REAL,
'daly_wt_closed_pneumothorax - code 1697'
),
'daly_wt_open_pneumothorax': Parameter(
Types.REAL,
'daly_wt_open_pneumothorax - code 1698'
),
'daly_wt_surgical_emphysema': Parameter(
Types.REAL,
'daly_wt_surgical_emphysema aka subcuteal emphysema - code 1699'
),
'daly_wt_abd_internal_organ_injury': Parameter(
Types.REAL,
'daly_wt_abd_internal_organ_injury - code 1700'
),
'daly_wt_spinal_cord_lesion_neck_with_treatment': Parameter(
Types.REAL,
'daly_wt_spinal_cord_lesion_neck_with_treatment - code 1701'
),
'daly_wt_spinal_cord_lesion_neck_without_treatment': Parameter(
Types.REAL,
'daly_wt_spinal_cord_lesion_neck_without_treatment - code 1702'
),
'daly_wt_spinal_cord_lesion_below_neck_with_treatment': Parameter(
Types.REAL,
'daly_wt_spinal_cord_lesion_below_neck_with_treatment - code 1703'
),
'daly_wt_spinal_cord_lesion_below_neck_without_treatment': Parameter(
Types.REAL,
'daly_wt_spinal_cord_lesion_below_neck_without_treatment - code 1704'
),
'daly_wt_vertebrae_fracture': Parameter(
Types.REAL,
'daly_wt_vertebrae_fracture - code 1705'
),
'daly_wt_clavicle_scapula_humerus_fracture': Parameter(
Types.REAL,
'daly_wt_clavicle_scapula_humerus_fracture - code 1706'
),
'daly_wt_hand_wrist_fracture_with_treatment': Parameter(
Types.REAL,
'daly_wt_hand_wrist_fracture_with_treatment - code 1707'
),
'daly_wt_hand_wrist_fracture_without_treatment': Parameter(
Types.REAL,
'daly_wt_hand_wrist_fracture_without_treatment - code 1708'
),
'daly_wt_radius_ulna_fracture_short_term_with_without_treatment': Parameter(
Types.REAL,
'daly_wt_radius_ulna_fracture_short_term_with_without_treatment - code 1709'
),
'daly_wt_radius_ulna_fracture_long_term_without_treatment': Parameter(
Types.REAL,
'daly_wt_radius_ulna_fracture_long_term_without_treatment - code 1710'
),
'daly_wt_dislocated_shoulder': Parameter(
Types.REAL,
'daly_wt_dislocated_shoulder - code 1711'
),
'daly_wt_amputated_finger': Parameter(
Types.REAL,
'daly_wt_amputated_finger - code 1712'
),
'daly_wt_amputated_thumb': Parameter(
Types.REAL,
'daly_wt_amputated_thumb - code 1713'
),
'daly_wt_unilateral_arm_amputation_with_treatment': Parameter(
Types.REAL,
'daly_wt_unilateral_arm_amputation_with_treatment - code 1714'
),
'daly_wt_unilateral_arm_amputation_without_treatment': Parameter(
Types.REAL,
'daly_wt_unilateral_arm_amputation_without_treatment - code 1715'
),
'daly_wt_bilateral_arm_amputation_with_treatment': Parameter(
Types.REAL,
'daly_wt_bilateral_arm_amputation_with_treatment - code 1716'
),
'daly_wt_bilateral_arm_amputation_without_treatment': Parameter(
Types.REAL,
'daly_wt_bilateral_arm_amputation_without_treatment - code 1717'
),
'daly_wt_foot_fracture_short_term_with_without_treatment': Parameter(
Types.REAL,
'daly_wt_foot_fracture_short_term_with_without_treatment - code 1718'
),
'daly_wt_foot_fracture_long_term_without_treatment': Parameter(
Types.REAL,
'daly_wt_foot_fracture_long_term_without_treatment - code 1719'
),
'daly_wt_patella_tibia_fibula_fracture_with_treatment': Parameter(
Types.REAL,
'daly_wt_patella_tibia_fibula_fracture_with_treatment - code 1720'
),
'daly_wt_patella_tibia_fibula_fracture_without_treatment': Parameter(
Types.REAL,
'daly_wt_patella_tibia_fibula_fracture_without_treatment - code 1721'
),
'daly_wt_hip_fracture_short_term_with_without_treatment': Parameter(
Types.REAL,
'daly_wt_hip_fracture_short_term_with_without_treatment - code 1722'
),
'daly_wt_hip_fracture_long_term_with_treatment': Parameter(
Types.REAL,
'daly_wt_hip_fracture_long_term_with_treatment - code 1723'
),
'daly_wt_hip_fracture_long_term_without_treatment': Parameter(
Types.REAL,
'daly_wt_hip_fracture_long_term_without_treatment - code 1724'
),
'daly_wt_pelvis_fracture_short_term': Parameter(
Types.REAL,
'daly_wt_pelvis_fracture_short_term - code 1725'
),
'daly_wt_pelvis_fracture_long_term': Parameter(
Types.REAL,
'daly_wt_pelvis_fracture_long_term - code 1726'
),
'daly_wt_femur_fracture_short_term': Parameter(
Types.REAL,
'daly_wt_femur_fracture_short_term - code 1727'
),
'daly_wt_femur_fracture_long_term_without_treatment': Parameter(
Types.REAL,
'daly_wt_femur_fracture_long_term_without_treatment - code 1728'
),
'daly_wt_dislocated_hip': Parameter(
Types.REAL,
'daly_wt_dislocated_hip - code 1729'
),
'daly_wt_dislocated_knee': Parameter(
Types.REAL,
'daly_wt_dislocated_knee - code 1730'
),
'daly_wt_amputated_toes': Parameter(
Types.REAL,
'daly_wt_amputated_toes - code 1731'
),
'daly_wt_unilateral_lower_limb_amputation_with_treatment': Parameter(
Types.REAL,
'daly_wt_unilateral_lower_limb_amputation_with_treatment - code 1732'
),
'daly_wt_unilateral_lower_limb_amputation_without_treatment': Parameter(
Types.REAL,
'daly_wt_unilateral_lower_limb_amputation_without_treatment - code 1733'
),
'daly_wt_bilateral_lower_limb_amputation_with_treatment': Parameter(
Types.REAL,
'daly_wt_bilateral_lower_limb_amputation_with_treatment - code 1734'
),
'daly_wt_bilateral_lower_limb_amputation_without_treatment': Parameter(
Types.REAL,
'daly_wt_bilateral_lower_limb_amputation_without_treatment - code 1735'
),
'rt_emergency_care_ISS_score_cut_off': Parameter(
Types.INT,
'A parameter to determine which level of injury severity corresponds to the emergency health care seeking '
'symptom and which to the non-emergency generic injury symptom'
),
'prob_death_MAIS3': Parameter(
Types.REAL,
'A parameter to determine the probability of death without medical intervention with a military AIS'
'score of 3'
),
'prob_death_MAIS4': Parameter(
Types.REAL,
'A parameter to determine the probability of death without medical intervention with a military AIS'
'score of 4'
),
'prob_death_MAIS5': Parameter(
Types.REAL,
'A parameter to determine the probability of death without medical intervention with a military AIS'
'score of 5'
),
'prob_death_MAIS6': Parameter(
Types.REAL,
'A parameter to determine the probability of death without medical intervention with a military AIS'
'score of 6'
),
'femur_fracture_skeletal_traction_mean_los': Parameter(
Types.INT,
'The mean length of stay for a person with a femur fracture being treated with skeletal traction'
),
'other_skeletal_traction_los': Parameter(
Types.INT,
'The mean length of stay for a person with a non-femur fracture being treated with skeletal traction'
),
'prob_foot_frac_require_cast': Parameter(
Types.REAL,
'The probability that a person with a foot fracture will be treated with a plaster cast'
),
'prob_foot_frac_require_maj_surg': Parameter(
Types.REAL,
'The probability that a person with a foot fracture will be treated with a major surgery'
),
'prob_foot_frac_require_min_surg': Parameter(
Types.REAL,
'The probability that a person with a foot fracture will be treated with a major surgery'
),
'prob_foot_frac_require_amp': Parameter(
Types.REAL,
'The probability that a person with a foot fracture will be treated with amputation via a major surgery'
),
'prob_tib_fib_frac_require_cast': Parameter(
Types.REAL,
'The probability that a person with a tibia/fibula fracture will be treated with a plaster cast'
),
'prob_tib_fib_frac_require_maj_surg': Parameter(
Types.REAL,
'The probability that a person with a tibia/fibula fracture will be treated with a major surgery'
),
'prob_tib_fib_frac_require_min_surg': Parameter(
Types.REAL,
'The probability that a person with a tibia/fibula fracture will be treated with a minor surgery'
),
'prob_tib_fib_frac_require_amp': Parameter(
Types.REAL,
'The probability that a person with a tibia/fibula fracture will be treated with an amputation via major '
'surgery'
),
'prob_tib_fib_frac_require_traction': Parameter(
Types.REAL,
'The probability that a person with a tibia/fibula fracture will be treated with skeletal traction'
),
'prob_femural_fracture_require_major_surgery': Parameter(
Types.REAL,
'The probability that a person with a femur fracture will be treated with major surgery'
),
'prob_femural_fracture_require_minor_surgery': Parameter(
Types.REAL,
'The probability that a person with a femur fracture will be treated with minor surgery'
),
'prob_femural_fracture_require_cast': Parameter(
Types.REAL,
'The probability that a person with a femur fracture will be treated with a plaster cast'
),
'prob_femural_fracture_require_amputation': Parameter(
Types.REAL,
'The probability that a person with a femur fracture will be treated with amputation via major surgery'
),
'prob_femural_fracture_require_traction': Parameter(
Types.REAL,
'The probability that a person with a femur fracture will be treated with skeletal traction'
),
'prob_pelvis_fracture_traction': Parameter(
Types.REAL,
'The probability that a person with a pelvis fracture will be treated with skeletal traction'
),
'prob_pelvis_frac_major_surgery': Parameter(
Types.REAL,
'The probability that a person with a pelvis fracture will be treated with major surgery'
),
'prob_pelvis_frac_minor_surgery': Parameter(
Types.REAL,
'The probability that a person with a pelvis fracture will be treated with minor surgery'
),
'prob_pelvis_frac_cast': Parameter(
Types.REAL,
'The probability that a person with a pelvis fracture will be treated with a cast'
),
'prob_dis_hip_require_maj_surg': Parameter(
Types.REAL,
'The probability that a person with a dislocated hip will be treated with a major surgery'
),
'prob_dis_hip_require_cast': Parameter(
Types.REAL,
'The probability that a person with a dislocated hip will be treated with a plaster cast'
),
'prob_hip_dis_require_traction': Parameter(
Types.REAL,
'The probability that a person with a dislocated hip will be treated with skeletal traction'
),
'hdu_cut_off_iss_score': Parameter(
Types.INT,
'The ISS score used as a criteria to admit patients to the HDU/ICU units'
),
'mean_icu_days': Parameter(
Types.REAL,
'The mean length of stay in the ICUfor those without TBI'
),
'sd_icu_days': Parameter(
Types.REAL,
'The standard deviation in length of stay in the ICU for those without TBI'
),
'mean_tbi_icu_days': Parameter(
Types.REAL,
'The mean length of stay in the ICU for those with TBI'
),
'sd_tbi_icu_days': Parameter(
Types.REAL,
'The standard deviation in length of stay in the ICU for those with TBI'
),
'prob_open_fracture_contaminated': Parameter(
Types.REAL,
'The probability that an open fracture will be contaminated'
),
'allowed_interventions': Parameter(
Types.LIST,
'List of additional interventions that can be included when performing model analysis'
),
'head_prob_112': Parameter(
Types.REAL,
"The probability that this person's head injury is a skull fracture"
),
'head_prob_113': Parameter(
Types.REAL,
"The probability that this person's head injury is a basilar skull fracture"
),
'head_prob_133a': Parameter(
Types.REAL,
"The probability that this person's head injury is a Subarachnoid hematoma"
),
'head_prob_133b': Parameter(
Types.REAL,
"The probability that this person's head injury is a Brain contusion"
),
'head_prob_133c': Parameter(
Types.REAL,
"The probability that this person's head injury is an Intraventricular haemorrhage"
),
'head_prob_133d': Parameter(
Types.REAL,
"The probability that this person's head injury is a Subgaleal hematoma"
),
'head_prob_134a': Parameter(
Types.REAL,
"The probability that this person's head injury is an Epidural hematoma"
),
'head_prob_134b': Parameter(
Types.REAL,
"The probability that this person's head injury is a Subdural hematoma"
),
'head_prob_135': Parameter(
Types.REAL,
"The probability that this person's head injury is a Diffuse axonal injury/midline shift"
),
'head_prob_1101': Parameter(
Types.REAL,
"The probability that this person's head injury is a laceration"
),
'head_prob_1114': Parameter(
Types.REAL,
"The probability that this person's head injury is a burn"
),
'face_prob_211': Parameter(
Types.REAL,
"The probability that this person's face injury is a Facial fracture (nasal/unspecified)"
),
'face_prob_212': Parameter(
Types.REAL,
"The probability that this person's face injury is a Facial fracture (mandible/zygomatic)"
),
'face_prob_241': Parameter(
Types.REAL,
"The probability that this person's face injury is a soft tissue injury"
),
'face_prob_2101': Parameter(
Types.REAL,
"The probability that this person's face injury is a laceration"
),
'face_prob_2114': Parameter(
Types.REAL,
"The probability that this person's face injury is a burn"
),
'face_prob_291': Parameter(
Types.REAL,
"The probability that this person's face injury is an eye injury"
),
'neck_prob_3101': Parameter(
Types.REAL,
"The probability that this person's neck injury is a laceration"
),
'neck_prob_3113': Parameter(
Types.REAL,
"The probability that this person's neck injury is a burn"
),
'neck_prob_342': Parameter(
Types.REAL,
"The probability that this person's neck injury is a Soft tissue injury in neck (vertebral artery "
"laceration)"
),
'neck_prob_343': Parameter(
Types.REAL,
"The probability that this person's neck injury is a Soft tissue injury in neck (pharynx contusion)"
),
'neck_prob_361': Parameter(
Types.REAL,
"The probability that this person's neck injury is a Sternomastoid m. hemorrhage/ Hemorrhage, "
"supraclavicular triangle/Hemorrhage, posterior triangle/Anterior vertebral vessel hemorrhage/ Neck muscle "
"hemorrhage"
),
'neck_prob_363': Parameter(
Types.REAL,
"The probability that this person's neck injury is a Hematoma in carotid sheath/Carotid sheath hemorrhage"
),
'neck_prob_322': Parameter(
Types.REAL,
"The probability that this person's neck injury is an Atlanto-occipital subluxation"
),
'neck_prob_323': Parameter(
Types.REAL,
"The probability that this person's neck injury is an Atlanto-axial subluxation"
),
'thorax_prob_4101': Parameter(
Types.REAL,
"The probability that this person's thorax injury is a laceration"
),
'thorax_prob_4113': Parameter(
Types.REAL,
"The probability that this person's thorax injury is a burn"
),
'thorax_prob_461': Parameter(
Types.REAL,
"The probability that this person's thorax injury is Chest wall bruises/haematoma"
),
'thorax_prob_463': Parameter(
Types.REAL,
"The probability that this person's thorax injury is Haemothorax"
),
'thorax_prob_453a': Parameter(
Types.REAL,
"The probability that this person's thorax injury is a Lung contusion"
),
'thorax_prob_453b': Parameter(
Types.REAL,
"The probability that this person's thorax injury is a Diaphragm rupture"
),
'thorax_prob_412': Parameter(
Types.REAL,
"The probability that this person's thorax injury is a rib fracture"
),
'thorax_prob_414': Parameter(
Types.REAL,
"The probability that this person's thorax injury is flail chest"
),
'thorax_prob_441': Parameter(
Types.REAL,
"The probability that this person's thorax injury is a Chest wall lacerations/avulsions"
),
'thorax_prob_442': Parameter(
Types.REAL,
"The probability that this person's thorax injury is a Surgical emphysema"
),
'thorax_prob_443': Parameter(
Types.REAL,
"The probability that this person's thorax injury is a Closed pneumothorax/ open pneumothorax"
),
'abdomen_prob_5101': Parameter(
Types.REAL,
"The probability that this person's abdomen injury is a laceration"
),
'abdomen_prob_5113': Parameter(
Types.REAL,
"The probability that this person's thorax injury is a burn"
),
'abdomen_prob_552': Parameter(
Types.REAL,
"The probability that this person's thorax injury is a skull fracture"
),
'abdomen_prob_553': Parameter(
Types.REAL,
"The probability that this person's thorax injury is an Injury to stomach/intestines/colon"
),
'abdomen_prob_554': Parameter(
Types.REAL,
"The probability that this person's thorax injury is an Injury to spleen/Urinary bladder/Liver/Urethra/"
"Diaphragm"
),
'spine_prob_612': Parameter(
Types.REAL,
"The probability that this person's spine injury is a vertabrae fracture"
),
'spine_prob_673a': Parameter(
Types.REAL,
"The probability that this person's spine injury is a Spinal cord injury at neck level"
),
'spine_prob_673b': Parameter(
Types.REAL,
"The probability that this person's spine injury is a Spinal cord injury below neck level"
),
'spine_prob_674a': Parameter(
Types.REAL,
"The probability that this person's spine injury is a Spinal cord injury at neck level"
),
'spine_prob_674b': Parameter(
Types.REAL,
"The probability that this person's spine injury is a Spinal cord injury below neck level"
),
'spine_prob_675a': Parameter(
Types.REAL,
"The probability that this person's spine injury is a Spinal cord injury at neck level"
),
'spine_prob_675b': Parameter(
Types.REAL,
"The probability that this person's spine injury is a Spinal cord injury below neck level"
),
'spine_prob_676': Parameter(
Types.REAL,
"The probability that this person's spine injury is a Spinal cord injury at neck level"
),
'upper_ex_prob_7101': Parameter(
Types.REAL,
"The probability that this person's upper extremity injury is a laceration"
),
'upper_ex_prob_7113': Parameter(
Types.REAL,
"The probability that this person's upper extremity injury is a burn"
),
'upper_ex_prob_712a': Parameter(
Types.REAL,
"The probability that this person's upper extremity injury is a Fracture to Clavicle, scapula, humerus"
),
'upper_ex_prob_712b': Parameter(
Types.REAL,
"The probability that this person's upper extremity injury is a Fracture to Hand/wrist"
),
'upper_ex_prob_712c': Parameter(
Types.REAL,
"The probability that this person's upper extremity injury is a Fracture to Radius/ulna"
),
'upper_ex_prob_722': Parameter(
Types.REAL,
"The probability that this person's upper extremity injury is a dislocated shoulder"
),
'upper_ex_prob_782a': Parameter(
Types.REAL,
"The probability that this person's upper extremity injury is an Amputated finger"
),
'upper_ex_prob_782b': Parameter(
Types.REAL,
"The probability that this person's upper extremity injury is a Unilateral arm amputation"
),
'upper_ex_prob_782c': Parameter(
Types.REAL,
"The probability that this person's upper extremity injury is a Thumb amputation"
),
'upper_ex_prob_783': Parameter(
Types.REAL,
"The probability that this person's upper extremity injury is a bilateral arm amputation"
),
'lower_ex_prob_8101': Parameter(
Types.REAL,
"The probability that this person's lower extremity injury is a laceration"
),
'lower_ex_prob_8113': Parameter(
Types.REAL,
"The probability that this person's lower extremity injury is a burn"
),
'lower_ex_prob_811': Parameter(
Types.REAL,
"The probability that this person's lower extremity injury is a foot fracture"
),
'lower_ex_prob_813do': Parameter(
Types.REAL,
"The probability that this person's lower extremity injury is an open foot fracture"
),
'lower_ex_prob_812': Parameter(
Types.REAL,
"The probability that this person's lower extremity injury is a Fracture to patella, tibia, fibula, ankle"
),
'lower_ex_prob_813eo': Parameter(
Types.REAL,
"The probability that this person's lower extremity injury is an open Fracture to patella, tibia, fibula, "
"ankle"
),
'lower_ex_prob_813a': Parameter(
Types.REAL,
"The probability that this person's lower extremity injury is a Hip fracture"
),
'lower_ex_prob_813b': Parameter(
Types.REAL,
"The probability that this person's lower extremity injury is a Pelvis fracture"
),
'lower_ex_prob_813bo': Parameter(
Types.REAL,
"The probability that this person's lower extremity injury is an open Pelvis fracture"
),
'lower_ex_prob_813c': Parameter(
Types.REAL,
"The probability that this person's lower extremity injury is a Femur fracture"
),
'lower_ex_prob_813co': Parameter(
Types.REAL,
"The probability that this person's lower extremity injury is an open Femur fracture"
),
'lower_ex_prob_822a': Parameter(
Types.REAL,
"The probability that this person's lower extremity injury is a Dislocated hip"
),
'lower_ex_prob_822b': Parameter(
Types.REAL,
"The probability that this person's lower extremity injury is a Dislocated knee"
),
'lower_ex_prob_882': Parameter(
Types.REAL,
"The probability that this person's lower extremity injury is an Amputation of toes"
),
'lower_ex_prob_883': Parameter(
Types.REAL,
"The probability that this person's lower extremity injury is a Unilateral leg amputation"
),
'lower_ex_prob_884': Parameter(
Types.REAL,
"The probability that this person's lower extremity injury is a Bilateral leg amputation "
),
'blocked_interventions': Parameter(
Types.LIST,
"A list of interventions that are blocked in a simulation"
),
'unavailable_treatment_mortality_mais_cutoff': Parameter(
Types.INT,
"A cut-off score above which an injury will result in additional mortality if the person has "
"sought healthcare and not received it."
),
'consider_death_no_treatment_ISS_cut_off': Parameter(
Types.INT,
"A cut-off score above which an injuries will be considered severe enough to cause mortality in those who"
"have not sought care."
),
'maximum_number_of_times_HSI_events_should_run': Parameter(
Types.INT,
"limit on the number of times an HSI event can run"
)
}
# Define the module's parameters
PROPERTIES = {
'rt_road_traffic_inc': Property(Types.BOOL, 'involved in a road traffic injury'),
'rt_inj_severity': Property(Types.CATEGORICAL,
'Injury status relating to road traffic injury: none, mild, severe',
categories=['none', 'mild', 'severe'],
),
**{
f'rt_injury_{injury_index}': Property(
Types.CATEGORICAL,
f'Codes for injury {injury_index} from RTI',
categories=categories,
)
# hacky solution to avoid issue that names defined in class scope are not
# accessible in scope of comprehension expresions _except for_ outermost
# iterator - see https://stackoverflow.com/a/13913933/4798943
for injury_index, categories in zip(
INJURY_INDICES, [INJURY_CODES] * len(INJURY_INDICES)
)
},
**{
f'rt_date_to_remove_daly_{injury_index}': Property(
Types.DATE,
f'Date to remove the daly weight for injury {injury_index}',
)
for injury_index in INJURY_INDICES
},
'rt_in_shock': Property(Types.BOOL, 'A property determining if this person is in shock'),
'rt_death_from_shock': Property(Types.BOOL, 'whether this person died from shock'),
'rt_injuries_to_cast': Property(Types.LIST, 'A list of injuries that are to be treated with casts'),
'rt_injuries_for_minor_surgery': Property(Types.LIST, 'A list of injuries that are to be treated with a minor'
'surgery'),
'rt_injuries_for_major_surgery': Property(Types.LIST, 'A list of injuries that are to be treated with a minor'
'surgery'),
'rt_injuries_to_heal_with_time': Property(Types.LIST, 'A list of injuries that heal without further treatment'),
'rt_injuries_for_open_fracture_treatment': Property(Types.LIST, 'A list of injuries that with open fracture '
'treatment'),
'rt_ISS_score': Property(Types.INT, 'The ISS score associated with the injuries resulting from a road traffic'
'accident'),
'rt_perm_disability': Property(Types.BOOL, 'whether the injuries from an RTI result in permanent disability'),
'rt_polytrauma': Property(Types.BOOL, 'polytrauma from RTI'),
'rt_imm_death': Property(Types.BOOL, 'death at scene True/False'),
'rt_diagnosed': Property(Types.BOOL, 'Person has had their injuries diagnosed'),
'rt_post_med_death': Property(Types.BOOL, 'death in following month despite medical intervention True/False'),
'rt_no_med_death': Property(Types.BOOL, 'death in following month without medical intervention True/False'),
'rt_unavailable_med_death': Property(Types.BOOL, 'death in the following month without medical intervention '
'being able to be provided'),
'rt_recovery_no_med': Property(Types.BOOL, 'recovery without medical intervention True/False'),
'rt_disability': Property(Types.REAL, 'disability weight for current month'),
'rt_date_inj': Property(Types.DATE, 'date of latest injury'),
'rt_med_int': Property(Types.BOOL, 'whether this person is currently undergoing medical treatment'),
'rt_in_icu_or_hdu': Property(Types.BOOL, 'whether this person is currently in ICU for RTI'),
'rt_MAIS_military_score': Property(Types.INT, 'the maximum AIS-military score, used as a proxy to calculate the'
'probability of mortality without medical intervention'),
'rt_date_death_no_med': Property(Types.DATE, 'the date which the person has is scheduled to die without medical'
'intervention'),
'rt_debugging_DALY_wt': Property(Types.REAL, 'The true value of the DALY weight burden'),
'rt_injuries_left_untreated': Property(Types.LIST, 'A list of injuries that have been left untreated due to a '
'blocked intervention')
}
# Declare Metadata
METADATA = {
Metadata.DISEASE_MODULE, # Disease modules: Any disease module should carry this label.
Metadata.USES_SYMPTOMMANAGER, # The 'Symptom Manager' recognises modules with this label.
Metadata.USES_HEALTHSYSTEM, # The 'HealthSystem' recognises modules with this label.
Metadata.USES_HEALTHBURDEN # The 'HealthBurden' module recognises modules with this label.
}
# Declare Causes of Death
CAUSES_OF_DEATH = {
'RTI_death_without_med': Cause(gbd_causes='Road injuries', label='Transport Injuries'),
'RTI_death_with_med': Cause(gbd_causes='Road injuries', label='Transport Injuries'),
'RTI_unavailable_med': Cause(gbd_causes='Road injuries', label='Transport Injuries'),
'RTI_imm_death': Cause(gbd_causes='Road injuries', label='Transport Injuries'),
'RTI_death_shock': Cause(gbd_causes='Road injuries', label='Transport Injuries'),
}
# Declare Causes of Death and Disability
CAUSES_OF_DISABILITY = {
'RTI': Cause(gbd_causes='Road injuries', label='Transport Injuries')
}
[docs]
def read_parameters(self, data_folder):
""" Reads the parameters used in the RTI module"""
p = self.parameters
dfd = pd.read_excel(Path(self.resourcefilepath) / 'ResourceFile_RTI.xlsx', sheet_name='parameter_values')
self.load_parameters_from_dataframe(dfd)
if "HealthBurden" in self.sim.modules:
# get the DALY weights of the seq associated with road traffic injuries
daly_sequlae_codes = {
'daly_wt_unspecified_skull_fracture': 1674,
'daly_wt_basilar_skull_fracture': 1675,
'daly_wt_epidural_hematoma': 1676,
'daly_wt_subdural_hematoma': 1677,
'daly_wt_subarachnoid_hematoma': 1678,
'daly_wt_brain_contusion': 1679,
'daly_wt_intraventricular_haemorrhage': 1680,
'daly_wt_diffuse_axonal_injury': 1681,
'daly_wt_subgaleal_hematoma': 1682,
'daly_wt_midline_shift': 1683,
'daly_wt_facial_fracture': 1684,
'daly_wt_facial_soft_tissue_injury': 1685,
'daly_wt_eye_injury': 1686,
'daly_wt_neck_soft_tissue_injury': 1687,
'daly_wt_neck_internal_bleeding': 1688,
'daly_wt_neck_dislocation': 1689,
'daly_wt_chest_wall_bruises_hematoma': 1690,
'daly_wt_hemothorax': 1691,
'daly_wt_lung_contusion': 1692,
'daly_wt_diaphragm_rupture': 1693,
'daly_wt_rib_fracture': 1694,
'daly_wt_flail_chest': 1695,
'daly_wt_chest_wall_laceration': 1696,
'daly_wt_closed_pneumothorax': 1697,
'daly_wt_open_pneumothorax': 1698,
'daly_wt_surgical_emphysema': 1699,
'daly_wt_abd_internal_organ_injury': 1700,
'daly_wt_spinal_cord_lesion_neck_with_treatment': 1701,
'daly_wt_spinal_cord_lesion_neck_without_treatment': 1702,
'daly_wt_spinal_cord_lesion_below_neck_with_treatment': 1703,
'daly_wt_spinal_cord_lesion_below_neck_without_treatment': 1704,
'daly_wt_vertebrae_fracture': 1705,
'daly_wt_clavicle_scapula_humerus_fracture': 1706,
'daly_wt_hand_wrist_fracture_with_treatment': 1707,
'daly_wt_hand_wrist_fracture_without_treatment': 1708,
'daly_wt_radius_ulna_fracture_short_term_with_without_treatment': 1709,
'daly_wt_radius_ulna_fracture_long_term_without_treatment': 1710,
'daly_wt_dislocated_shoulder': 1711,
'daly_wt_amputated_finger': 1712,
'daly_wt_amputated_thumb': 1713,
'daly_wt_unilateral_arm_amputation_with_treatment': 1714,
'daly_wt_unilateral_arm_amputation_without_treatment': 1715,
'daly_wt_bilateral_arm_amputation_with_treatment': 1716,
'daly_wt_bilateral_arm_amputation_without_treatment': 1717,
'daly_wt_foot_fracture_short_term_with_without_treatment': 1718,
'daly_wt_foot_fracture_long_term_without_treatment': 1719,
'daly_wt_patella_tibia_fibula_fracture_with_treatment': 1720,
'daly_wt_patella_tibia_fibula_fracture_without_treatment': 1721,
'daly_wt_hip_fracture_short_term_with_without_treatment': 1722,
'daly_wt_hip_fracture_long_term_with_treatment': 1723,
'daly_wt_hip_fracture_long_term_without_treatment': 1724,
'daly_wt_pelvis_fracture_short_term': 1725,
'daly_wt_pelvis_fracture_long_term': 1726,
'daly_wt_femur_fracture_short_term': 1727,
'daly_wt_femur_fracture_long_term_without_treatment': 1728,
'daly_wt_dislocated_hip': 1729,
'daly_wt_dislocated_knee': 1730,
'daly_wt_amputated_toes': 1731,
'daly_wt_unilateral_lower_limb_amputation_with_treatment': 1732,
'daly_wt_unilateral_lower_limb_amputation_without_treatment': 1733,
'daly_wt_bilateral_lower_limb_amputation_with_treatment': 1734,
'daly_wt_bilateral_lower_limb_amputation_without_treatment': 1735,
'daly_wt_burns_greater_than_20_percent_body_area': 1736,
'daly_wt_burns_less_than_20_percent_body_area_with_treatment': 1737,
'daly_wt_burns_less_than_20_percent_body_area_without_treatment': 1738,
}
hb = self.sim.modules["HealthBurden"]
for key, value in daly_sequlae_codes.items():
p[key] = hb.get_daly_weight(sequlae_code=value)
# ================== Test the parameter distributions to see whether they sum to roughly one ===============
# test the distribution of the number of injured body regions
assert 0.9999 < sum(p['number_of_injured_body_regions_distribution'][1]) < 1.0001, \
"The number of injured body region distribution doesn't sum to one"
# test the injury location distribution
assert 0.9999 < sum(p['injury_location_distribution'][1]) < 1.0001, \
"The injured body region distribution doesn't sum to one"
# test the distributions to assign injuries to certain body regions
# get the first characters of the parameter names
body_part_strings = ['head_prob_', 'face_prob_', 'neck_prob_', 'thorax_prob_', 'abdomen_prob_',
'spine_prob_', 'upper_ex_prob_', 'lower_ex_prob_']
# iterate over each body part, check the probabilities add to one
for body_part in body_part_strings:
probabilities_to_assign_injuries = [val for key, val in p.items() if body_part in key]
sum_probabilities = sum(probabilities_to_assign_injuries)
assert (sum_probabilities % 1 < 0.0001) or (sum_probabilities % 1 > 0.9999), "The probabilities" \
"chosen for assigning" \
"injuries don't" \
"sum to one"
# Check all other probabilities are between 0 and 1
probabilities = [val for key, val in p.items() if 'prob_' in key]
for probability in probabilities:
assert 0 <= probability <= 1, "Probability is not a feasible value"
# create a generic severe trauma symptom, which forces people into the health system
self.sim.modules['SymptomManager'].register_symptom(Symptom.emergency('severe_trauma'))
# create an injury lookup table to handle all assigning injuries/daly weights and daly weight changes. The table
# is writted in the following format: [[1], 2, 3, 4]. [1] contains information used in assigning injuries e.g.
# probability of injury occuring followed by information used in logging, specifically injury location, injury
# category and injury severity. 2 contains the daly weight initially assigned to people who have this injury.
# 3 contains any potential changes to the persons health burden upon treatment. 4 contains the daly weight to
# remove once an injury is healed.
self.ASSIGN_INJURIES_AND_DALY_CHANGES = {
'none': [0, 0, 0, 0],
# injuries to the head
'112': [[p['head_prob_112'], 1, 1, 2, 3], p['daly_wt_unspecified_skull_fracture'], 0,
- p['daly_wt_unspecified_skull_fracture']],
'113': [[p['head_prob_113'], 1, 1, 3, 4], p['daly_wt_basilar_skull_fracture'], 0,
- p['daly_wt_basilar_skull_fracture']],
'133a': [[p['head_prob_133a'], 1, 3, 3, 4], p['daly_wt_subarachnoid_hematoma'], 0,
- p['daly_wt_subarachnoid_hematoma']],
'133b': [[p['head_prob_133b'], 1, 3, 3, 4], p['daly_wt_brain_contusion'], 0,
- p['daly_wt_brain_contusion']],
'133c': [[p['head_prob_133c'], 1, 3, 3, 4], p['daly_wt_intraventricular_haemorrhage'], 0,
- p['daly_wt_intraventricular_haemorrhage']],
'133d': [[p['head_prob_133d'], 1, 3, 3, 4], p['daly_wt_subgaleal_hematoma'], 0,
- p['daly_wt_subgaleal_hematoma']],
'134a': [[p['head_prob_134a'], 1, 3, 4, 5], p['daly_wt_epidural_hematoma'], 0,
- p['daly_wt_epidural_hematoma']],
'134b': [[p['head_prob_134b'], 1, 3, 4, 5], p['daly_wt_subdural_hematoma'], 0,
- p['daly_wt_subdural_hematoma']],
'135': [[p['head_prob_135'], 1, 3, 5, 6], p['daly_wt_diffuse_axonal_injury'], 0,
- p['daly_wt_diffuse_axonal_injury']],
'1101': [[p['head_prob_1101'], 1, 10, 1, 2], p['daly_wt_facial_soft_tissue_injury'], 0,
- p['daly_wt_facial_soft_tissue_injury']],
'1114': [[p['head_prob_1114'], 1, 11, 4, 5], p['daly_wt_burns_greater_than_20_percent_body_area'], 0,
- p['daly_wt_burns_greater_than_20_percent_body_area']],
# injuries to the face
'211': [[p['face_prob_211'], 2, 1, 1, 2], p['daly_wt_facial_fracture'], 0, - p['daly_wt_facial_fracture']],
'212': [[p['face_prob_212'], 2, 1, 2, 3], p['daly_wt_facial_fracture'], 0, - p['daly_wt_facial_fracture']],
'241': [[p['face_prob_241'], 2, 4, 1, 2], p['daly_wt_facial_soft_tissue_injury'], 0,
- p['daly_wt_facial_soft_tissue_injury']],
'2101': [[p['face_prob_2101'], 2, 10, 1, 2], p['daly_wt_facial_soft_tissue_injury'], 0,
- p['daly_wt_facial_soft_tissue_injury']],
'2114': [[p['face_prob_2114'], 2, 11, 4, 5], p['daly_wt_burns_greater_than_20_percent_body_area'], 0,
- p['daly_wt_burns_greater_than_20_percent_body_area']],
'291': [[p['face_prob_291'], 2, 9, 1, 2], p['daly_wt_eye_injury'], 0, - p['daly_wt_eye_injury']],
# injuries to the neck
'3101': [[p['neck_prob_3101'], 3, 10, 1, 2], p['daly_wt_facial_soft_tissue_injury'], 0,
- p['daly_wt_facial_soft_tissue_injury']],
'3113': [[p['neck_prob_3113'], 3, 11, 3, 4],
p['daly_wt_burns_less_than_20_percent_body_area_without_treatment'],
- p['daly_wt_burns_less_than_20_percent_body_area_without_treatment'] +
p['daly_wt_burns_less_than_20_percent_body_area_with_treatment'],
- p['daly_wt_burns_less_than_20_percent_body_area_with_treatment']],
'342': [[p['neck_prob_342'], 3, 4, 2, 3], p['daly_wt_neck_internal_bleeding'], 0,
- p['daly_wt_neck_internal_bleeding']],
'343': [[p['neck_prob_343'], 3, 4, 3, 4], p['daly_wt_neck_internal_bleeding'], 0,
- p['daly_wt_neck_internal_bleeding']],
'361': [[p['neck_prob_361'], 3, 6, 1, 2], p['daly_wt_neck_internal_bleeding'], 0,
- p['daly_wt_neck_internal_bleeding']],
'363': [[p['neck_prob_363'], 3, 6, 3, 4], p['daly_wt_neck_internal_bleeding'], 0,
- p['daly_wt_neck_internal_bleeding']],
'322': [[p['neck_prob_322'], 3, 2, 2, 3], p['daly_wt_neck_dislocation'], 0,
- p['daly_wt_neck_dislocation']],
'323': [[p['neck_prob_323'], 3, 2, 3, 4], p['daly_wt_neck_dislocation'], 0,
- p['daly_wt_neck_dislocation']],
# injuries to the chest
'4101': [[p['thorax_prob_4101'], 4, 10, 1, 2], p['daly_wt_facial_soft_tissue_injury'], 0,
- p['daly_wt_facial_soft_tissue_injury']],
'4113': [[p['thorax_prob_4113'], 4, 11, 3, 4],
p['daly_wt_burns_less_than_20_percent_body_area_without_treatment'],
- p['daly_wt_burns_less_than_20_percent_body_area_without_treatment'] +
p['daly_wt_burns_less_than_20_percent_body_area_with_treatment'],
- p['daly_wt_burns_less_than_20_percent_body_area_with_treatment']],
'461': [[p['thorax_prob_461'], 4, 6, 1, 2], p['daly_wt_chest_wall_bruises_hematoma'], 0,
- p['daly_wt_chest_wall_bruises_hematoma']],
'463': [[p['thorax_prob_463'], 4, 6, 3, 4], p['daly_wt_hemothorax'], 0, - p['daly_wt_hemothorax']],
'453a': [[p['thorax_prob_453a'], 4, 5, 3, 4], p['daly_wt_diaphragm_rupture'], 0,
- p['daly_wt_diaphragm_rupture']],
'453b': [[p['thorax_prob_453b'], 4, 5, 3, 4], p['daly_wt_lung_contusion'], 0,
- p['daly_wt_lung_contusion']],
'412': [[p['thorax_prob_412'], 4, 1, 2, 3], p['daly_wt_rib_fracture'], 0, - p['daly_wt_rib_fracture']],
'414': [[p['thorax_prob_414'], 4, 1, 4, 5], p['daly_wt_flail_chest'], 0, - p['daly_wt_flail_chest']],
'441': [[p['thorax_prob_441'], 4, 4, 1, 2], p['daly_wt_closed_pneumothorax'], 0,
- p['daly_wt_closed_pneumothorax']],
'442': [[p['thorax_prob_442'], 4, 4, 2, 3], p['daly_wt_surgical_emphysema'], 0,
- p['daly_wt_surgical_emphysema']],
'443': [[p['thorax_prob_443'], 4, 4, 3, 4], p['daly_wt_open_pneumothorax'], 0,
- p['daly_wt_open_pneumothorax']],
# injuries to the abdomen
'5101': [[p['abdomen_prob_5101'], 5, 10, 1, 2], p['daly_wt_facial_soft_tissue_injury'], 0,
- p['daly_wt_facial_soft_tissue_injury']],
'5113': [[p['abdomen_prob_5113'], 5, 11, 3, 4],
p['daly_wt_burns_less_than_20_percent_body_area_without_treatment'],
- p['daly_wt_burns_less_than_20_percent_body_area_without_treatment'] +
p['daly_wt_burns_less_than_20_percent_body_area_with_treatment'],
- p['daly_wt_burns_less_than_20_percent_body_area_with_treatment']],
'552': [[p['abdomen_prob_552'], 5, 5, 2, 3], p['daly_wt_abd_internal_organ_injury'], 0,
- p['daly_wt_abd_internal_organ_injury']],
'553': [[p['abdomen_prob_553'], 5, 5, 3, 4], p['daly_wt_abd_internal_organ_injury'], 0,
- p['daly_wt_abd_internal_organ_injury']],
'554': [[p['abdomen_prob_554'], 5, 5, 4, 5], p['daly_wt_abd_internal_organ_injury'], 0,
- p['daly_wt_abd_internal_organ_injury']],
# injuries to the spine
'612': [[p['spine_prob_612'], 6, 1, 2, 3], p['daly_wt_vertebrae_fracture'], 0,
- p['daly_wt_vertebrae_fracture']],
'673a': [[p['spine_prob_673a'], 6, 7, 3, 4], p['daly_wt_spinal_cord_lesion_neck_without_treatment'],
- p['daly_wt_spinal_cord_lesion_neck_without_treatment'] +
p['daly_wt_spinal_cord_lesion_neck_with_treatment'], 0],
'673b': [[p['spine_prob_673b'], 6, 7, 3, 4], p['daly_wt_spinal_cord_lesion_below_neck_without_treatment'],
- p['daly_wt_spinal_cord_lesion_below_neck_without_treatment'] +
p['daly_wt_spinal_cord_lesion_below_neck_with_treatment'], 0],
'674a': [[p['spine_prob_674a'], 6, 7, 4, 5], p['daly_wt_spinal_cord_lesion_neck_without_treatment'],
- p['daly_wt_spinal_cord_lesion_neck_without_treatment'] +
p['daly_wt_spinal_cord_lesion_neck_with_treatment'], 0],
'674b': [[p['spine_prob_674b'], 6, 7, 4, 5], p['daly_wt_spinal_cord_lesion_below_neck_without_treatment'],
- p['daly_wt_spinal_cord_lesion_below_neck_without_treatment'] +
p['daly_wt_spinal_cord_lesion_below_neck_with_treatment'], 0],
'675a': [[p['spine_prob_675a'], 6, 7, 5, 6], p['daly_wt_spinal_cord_lesion_neck_without_treatment'],
- p['daly_wt_spinal_cord_lesion_neck_without_treatment'] +
p['daly_wt_spinal_cord_lesion_neck_with_treatment'], 0],
'675b': [[p['spine_prob_675b'], 6, 7, 5, 6], p['daly_wt_spinal_cord_lesion_below_neck_without_treatment'],
- p['daly_wt_spinal_cord_lesion_below_neck_without_treatment'] +
p['daly_wt_spinal_cord_lesion_below_neck_with_treatment'], 0],
'676': [[p['spine_prob_676'], 6, 7, 6, 6], p['daly_wt_spinal_cord_lesion_neck_without_treatment'],
- p['daly_wt_spinal_cord_lesion_neck_without_treatment'] +
p['daly_wt_spinal_cord_lesion_neck_with_treatment'], 0],
# injuries to the upper extremities
'7101': [[p['upper_ex_prob_7101'], 7, 10, 1, 2], p['daly_wt_facial_soft_tissue_injury'], 0,
- p['daly_wt_facial_soft_tissue_injury']],
'7113': [[p['upper_ex_prob_7113'], 7, 11, 3, 4],
p['daly_wt_burns_less_than_20_percent_body_area_without_treatment'],
- p['daly_wt_burns_less_than_20_percent_body_area_without_treatment'] +
p['daly_wt_burns_less_than_20_percent_body_area_with_treatment'],
- p['daly_wt_burns_less_than_20_percent_body_area_with_treatment']],
'712a': [[p['upper_ex_prob_712a'], 7, 1, 2, 3], p['daly_wt_clavicle_scapula_humerus_fracture'], 0,
- p['daly_wt_clavicle_scapula_humerus_fracture']],
'712b': [[p['upper_ex_prob_712b'], 7, 1, 2, 3], p['daly_wt_hand_wrist_fracture_without_treatment'],
- p['daly_wt_hand_wrist_fracture_without_treatment'] +
p['daly_wt_hand_wrist_fracture_with_treatment'],
- p['daly_wt_hand_wrist_fracture_with_treatment']],
'712c': [[p['upper_ex_prob_712c'], 7, 1, 2, 3],
p['daly_wt_radius_ulna_fracture_short_term_with_without_treatment'], 0,
- p['daly_wt_radius_ulna_fracture_short_term_with_without_treatment']],
'722': [[p['upper_ex_prob_722'], 7, 2, 2, 3], p['daly_wt_dislocated_shoulder'], 0,
- p['daly_wt_dislocated_shoulder']],
'782a': [[p['upper_ex_prob_782a'], 7, 8, 2, 3], p['daly_wt_amputated_finger'], 0, 0],
'782b': [[p['upper_ex_prob_782b'], 7, 8, 2, 3], p['daly_wt_unilateral_arm_amputation_without_treatment'],
- p['daly_wt_unilateral_arm_amputation_without_treatment'] +
p['daly_wt_unilateral_arm_amputation_with_treatment'], 0],
'782c': [[p['upper_ex_prob_782c'], 7, 8, 2, 3], p['daly_wt_amputated_thumb'], 0, 0],
'783': [[p['upper_ex_prob_783'], 7, 8, 3, 4], p['daly_wt_bilateral_arm_amputation_without_treatment'],
- p['daly_wt_bilateral_arm_amputation_without_treatment'] +
p['daly_wt_bilateral_arm_amputation_with_treatment'], 0],
# injuries to the lower extremities
'8101': [[p['lower_ex_prob_8101'], 8, 10, 1, 2], p['daly_wt_facial_soft_tissue_injury'], 0,
- p['daly_wt_facial_soft_tissue_injury']],
'8113': [[p['lower_ex_prob_8113'], 8, 11, 3, 4],
p['daly_wt_burns_less_than_20_percent_body_area_without_treatment'],
- p['daly_wt_burns_less_than_20_percent_body_area_without_treatment'] +
p['daly_wt_burns_less_than_20_percent_body_area_with_treatment'],
- p['daly_wt_burns_less_than_20_percent_body_area_with_treatment']],
# foot fracture, can be open or not, open is more severe
'811': [[p['lower_ex_prob_811'], 8, 1, 1, 2], p['daly_wt_foot_fracture_short_term_with_without_treatment'],
0, - p['daly_wt_foot_fracture_short_term_with_without_treatment']],
'813do': [[p['lower_ex_prob_813do'], 8, 1, 3, 4],
p['daly_wt_foot_fracture_short_term_with_without_treatment']
+ p['daly_wt_facial_soft_tissue_injury'], 0,
- p['daly_wt_foot_fracture_short_term_with_without_treatment'] -
p['daly_wt_facial_soft_tissue_injury']],
# lower leg fracture can be open or not
'812': [[p['lower_ex_prob_812'], 8, 1, 2, 3], p['daly_wt_patella_tibia_fibula_fracture_without_treatment'],
- p['daly_wt_patella_tibia_fibula_fracture_without_treatment'] +
p['daly_wt_patella_tibia_fibula_fracture_with_treatment'],
- p['daly_wt_patella_tibia_fibula_fracture_with_treatment']],
'813eo': [[p['lower_ex_prob_813eo'], 8, 1, 3, 4],
p['daly_wt_patella_tibia_fibula_fracture_without_treatment']
+ p['daly_wt_facial_soft_tissue_injury'], 0,
- p['daly_wt_patella_tibia_fibula_fracture_without_treatment'] -
p['daly_wt_facial_soft_tissue_injury']],
'813a': [[p['lower_ex_prob_813a'], 8, 1, 3, 4], p['daly_wt_hip_fracture_short_term_with_without_treatment'],
- p['daly_wt_hip_fracture_short_term_with_without_treatment'] +
p['daly_wt_hip_fracture_long_term_with_treatment'],
- p['daly_wt_hip_fracture_long_term_with_treatment']],
# pelvis fracture can be open or closed
'813b': [[p['lower_ex_prob_813b'], 8, 1, 3, 4], p['daly_wt_pelvis_fracture_short_term'],
- p['daly_wt_pelvis_fracture_short_term'] + p['daly_wt_pelvis_fracture_long_term'],
- p['daly_wt_pelvis_fracture_long_term']],
'813bo': [[p['lower_ex_prob_813bo'], 8, 1, 3, 4], p['daly_wt_pelvis_fracture_short_term'] +
p['daly_wt_facial_soft_tissue_injury'], - p['daly_wt_pelvis_fracture_short_term'] +
p['daly_wt_pelvis_fracture_long_term'], - p['daly_wt_pelvis_fracture_long_term'] -
p['daly_wt_facial_soft_tissue_injury']],
# femur fracture can be open or closed
'813c': [[p['lower_ex_prob_813c'], 8, 1, 3, 4], p['daly_wt_femur_fracture_short_term'], 0,
- p['daly_wt_femur_fracture_short_term']],
'813co': [[p['lower_ex_prob_813co'], 8, 1, 3, 4], p['daly_wt_femur_fracture_short_term'] +
p['daly_wt_facial_soft_tissue_injury'], 0, - p['daly_wt_femur_fracture_short_term'] -
p['daly_wt_facial_soft_tissue_injury']],
'822a': [[p['lower_ex_prob_822a'], 8, 2, 2, 3], p['daly_wt_dislocated_hip'], 0,
- p['daly_wt_dislocated_hip']],
'822b': [[p['lower_ex_prob_822b'], 8, 2, 2, 3], p['daly_wt_dislocated_knee'], 0,
- p['daly_wt_dislocated_knee']],
'882': [[p['lower_ex_prob_882'], 8, 8, 2, 3], p['daly_wt_amputated_toes'], 0, 0],
'883': [[p['lower_ex_prob_883'], 8, 8, 3, 4],
p['daly_wt_unilateral_lower_limb_amputation_without_treatment'],
- p['daly_wt_unilateral_lower_limb_amputation_without_treatment'] +
p['daly_wt_unilateral_lower_limb_amputation_with_treatment'], 0],
'884': [[p['lower_ex_prob_884'], 8, 8, 4, 5],
p['daly_wt_bilateral_lower_limb_amputation_without_treatment'],
- p['daly_wt_bilateral_lower_limb_amputation_without_treatment'] +
p['daly_wt_bilateral_lower_limb_amputation_with_treatment'], 0]
}
# The vast majority of the injuries should have a total change of daly weights that sum to zero, meaning that
# a person recieves an injury and has the health burden which will eventually be removed once the injury has
# healed. However some injuries are permanent so the person will always have some level of health burden. The
# injury codes for permanent injuries are given below.
permanent_injuries = ['673a', '673b', '674a', '674b', '675a', '675b', '676', '782a', '782b', '782c', '783',
'882', '883', '884']
# We need to check that the changes to all other DALY weights over the course of treatment sum to zero, do so
# using pandas, convert dictionary into a dataframe
check_daly_change_df = pd.DataFrame(self.ASSIGN_INJURIES_AND_DALY_CHANGES)
# drop the row of the dataframe used to assign people injuries
check_daly_change_df = check_daly_change_df.drop([0], axis=0)
# calculate the sum of the dataframe
sum_check_daly_change_df = check_daly_change_df.sum()
# find the injuries where the change in daly weights does not sum to zero
non_zero_total_daly_change = sum_check_daly_change_df.where(sum_check_daly_change_df > 0).dropna().index
# ensure that these injuries are the permanent injuries
assert non_zero_total_daly_change.to_list() == permanent_injuries
[docs]
def rti_injury_diagnosis(self, person_id, the_appt_footprint):
"""
A function used to alter the appointment footprint of the generic first appointments, based on the needs of
the patient to be properly diagnosed. Specifically, this function will assign x-rays/ct-scans for injuries
that require those diagnosis tools.
:param person_id: the person in a generic appointment with an injury
:param the_appt_footprint: the current appointment footprint to be altered
:return: the altered appointment footprint
"""
df = self.sim.population.props
# Filter the dataframe by the columns the injuries are stored in
persons_injuries = df.loc[[person_id], RTI.INJURY_COLUMNS]
# Injuries that require x rays are: fractures, spinal cord injuries, dislocations, soft tissue injuries in neck
# and soft tissue injury in thorax/ lung injury
codes_requiring_xrays = ['112', '113', '211', '212', '412', '414', '612', '712a', '712b', '712c', '811', '812',
'813a', '813b', '813c', '822a', '822b', '813bo', '813co', '813do', '813eo', '673',
'674', '675', '676', '322', '323', '722', '342', '343', '441', '443', '453']
# Injuries that require a ct scan are TBIs, abdominal trauma, soft tissue injury in neck, soft tissue injury in
# thorax/ lung injury and abdominal trauma
codes_requiring_ct_scan = ['133', '134', '135', '552', '553', '554', '342', '343', '441', '443', '453', '361',
'363', '461', '463']
def adjust_appt_footprint(_codes, _requirement):
_, counts = self.rti_find_and_count_injuries(persons_injuries, _codes)
if counts > 0:
the_appt_footprint[_requirement] = 1
adjust_appt_footprint(codes_requiring_xrays, 'DiagRadio')
adjust_appt_footprint(codes_requiring_ct_scan, 'Tomography')
[docs]
def initialise_population(self, population):
"""Sets up the default properties used in the RTI module and applies them to the dataframe. The default state
for the RTI module is that people haven't been involved in a road traffic accident and are therefor alive and
healthy."""
df = population.props
df.loc[df.is_alive, 'rt_road_traffic_inc'] = False
df.loc[df.is_alive, 'rt_inj_severity'] = "none" # default: no one has been injured in a RTI
for injury_index in RTI.INJURY_INDICES:
df.loc[df.is_alive, f'rt_injury_{injury_index}'] = "none"
df.loc[df.is_alive, f'rt_date_to_remove_daly_{injury_index}'] = pd.NaT
df.loc[df.is_alive, 'rt_in_shock'] = False
df.loc[df.is_alive, 'rt_death_from_shock'] = False
df.loc[df.is_alive, 'rt_polytrauma'] = False
df.loc[df.is_alive, 'rt_ISS_score'] = 0
df.loc[df.is_alive, 'rt_perm_disability'] = False
df.loc[df.is_alive, 'rt_imm_death'] = False # default: no one is dead on scene of crash
df.loc[df.is_alive, 'rt_diagnosed'] = False
df.loc[df.is_alive, 'rt_recovery_no_med'] = False # default: no recovery without medical intervention
df.loc[df.is_alive, 'rt_post_med_death'] = False # default: no death after medical intervention
df.loc[df.is_alive, 'rt_no_med_death'] = False
df.loc[df.is_alive, 'rt_unavailable_med_death'] = False
df.loc[df.is_alive, 'rt_disability'] = 0 # default: no DALY
df.loc[df.is_alive, 'rt_date_inj'] = pd.NaT
df.loc[df.is_alive, 'rt_med_int'] = False
df.loc[df.is_alive, 'rt_in_icu_or_hdu'] = False
df.loc[df.is_alive, 'rt_MAIS_military_score'] = 0
df.loc[df.is_alive, 'rt_date_death_no_med'] = pd.NaT
df.loc[df.is_alive, 'rt_debugging_DALY_wt'] = 0
alive_count = sum(df.is_alive)
df.loc[df.is_alive, 'rt_injuries_to_cast'] = pd.Series([[] for _ in range(alive_count)])
df.loc[df.is_alive, 'rt_injuries_for_minor_surgery'] = pd.Series([[] for _ in range(alive_count)])
df.loc[df.is_alive, 'rt_injuries_for_major_surgery'] = pd.Series([[] for _ in range(alive_count)])
df.loc[df.is_alive, 'rt_injuries_to_heal_with_time'] = pd.Series([[] for _ in range(alive_count)])
df.loc[df.is_alive, 'rt_injuries_for_open_fracture_treatment'] = pd.Series([[] for _ in range(alive_count)])
df.loc[df.is_alive, 'rt_injuries_left_untreated'] = pd.Series([[] for _ in range(alive_count)])
[docs]
def initialise_simulation(self, sim):
"""At the start of the simulation we schedule a logging event, which records the relevant information
regarding road traffic injuries in the last month.
Afterwards, we schedule three RTI events, the first is the main RTI event which takes parts
of the population and assigns them to be involved in road traffic injuries and providing they survived will
begin the interaction with the healthcare system. This event runs monthly.
The second is the begin scheduling the RTI recovery event, which looks at those in the population who have been
injured in a road traffic accident, checking every day whether enough time has passed for their injuries to have
healed. When the injury has healed the associated daly weight is removed.
The final event is one which checks if this person has not sought sought care or been given care, if they
haven't then it asks whether they should die away from their injuries
"""
# Begin modelling road traffic injuries
sim.schedule_event(RTIPollingEvent(self), sim.date + DateOffset(months=0))
# Begin checking whether the persons injuries are healed
sim.schedule_event(RTI_Recovery_Event(self), sim.date + DateOffset(months=0))
# Begin checking whether those with untreated injuries die
sim.schedule_event(RTI_Check_Death_No_Med(self), sim.date + DateOffset(months=0))
# Begin logging the RTI events
sim.schedule_event(RTI_Logging_Event(self), sim.date + DateOffset(months=1))
# Look-up consumable item codes
self.look_up_consumable_item_codes()
[docs]
def rti_do_when_diagnosed(self, person_id):
"""
This function is called by the generic first appointments when an injured person has been diagnosed
in A&E and needs to progress further in the health system. The injured person will then be scheduled a generic
'medical intervention' appointment which serves three purposes. The first is to determine what treatments they
require for their injuries and shedule those, the second is to contain them in the health care system with
inpatient days and finally, the appointment treats injuries that heal over time without further need for
resources in the health system.
:param person_id: the person requesting medical care
:return: n/a
"""
df = self.sim.population.props
# Check to see whether they have been sent here from A and E
assert df.at[person_id, 'rt_diagnosed']
# Get the relevant information about their injuries
person_injuries = df.loc[[person_id], RTI.INJURY_COLUMNS]
# check this person is injured, search they have an injury code that isn't "none"
_, counts = RTI.rti_find_and_count_injuries(person_injuries, RTI.INJURY_CODES[1:])
# also test whether the regular injury symptom has been given to the person via spurious symptoms
assert (counts > 0) or self.sim.modules['SymptomManager'].spurious_symptoms, \
'This person has asked for medical treatment despite not being injured'
# If they meet the requirements, send them to HSI_RTI_MedicalIntervention for further treatment
# Using counts condition to stop spurious symptoms progressing people through the model
if counts > 0:
self.sim.modules['HealthSystem'].schedule_hsi_event(
hsi_event=HSI_RTI_Medical_Intervention(module=self, person_id=person_id),
priority=0,
topen=self.sim.date
)
[docs]
def rti_do_for_major_surgeries(self, person_id, count):
"""
Function called in HSI_RTI_MedicalIntervention to schedule a major surgery if required. In
HSI_RTI_MedicalIntervention, we determine that they need a surgery. In this function, further to scheduling the
surgery, we double check that they do meet the conditions for needing a surgery. The conditions for needing a
surgery is that they are alive, currently seeking medical intervention and have an injury that is treated by
surgery.
:param person_id: The person requesting major surgeries
:param count: The amount of major surgeries required, used when scheduling surgeries to ensure that two major
surgeries aren't scheduled on the same day
:return: n/a
"""
df = self.sim.population.props
p = self.parameters
if df.at[person_id, 'is_alive']:
person = df.loc[person_id]
# Check to see whether they have been sent here from RTI_MedicalIntervention and they haven't died due to
# rti
assert person.rt_med_int, 'person sent here not been through RTI_MedInt'
# Determine what injuries are able to be treated by surgery by checking the injury codes which are currently
# treated in this simulation, it seems there is a limited available to treat spinal cord injuries and chest
# trauma in Malawi, so these are initially left out, but we will test different scenarios to see what
# happens when we include those treatments
surgically_treated_codes = ['112', '811', '812', '813a', '813b', '813c', '133a', '133b', '133c', '133d',
'134a', '134b', '135', '552', '553', '554', '342', '343', '414', '361', '363',
'782', '782a', '782b', '782c', '783', '822a', '882', '883', '884', 'P133a',
'P133b', 'P133c', 'P133d', 'P134a', 'P134b', 'P135', 'P782a', 'P782b', 'P782c',
'P783', 'P882', 'P883', 'P884']
# If we allow surgical treatment of spinal cord injuries, extend the surgically treated codes to include
# spinal cord injury codes
if 'include_spine_surgery' in p['allowed_interventions']:
additional_codes = ['673a', '673b', '674a', '674b', '675a', '675b', '676', 'P673a', 'P673b', 'P674',
'P674a', 'P674b', 'P675', 'P675a', 'P675b', 'P676']
surgically_treated_codes.extend(additional_codes)
# If we allow surgical treatment of chest trauma, extend the surgically treated codes to include chest
# trauma codes.
if 'include_thoroscopy' in p['allowed_interventions']:
additional_codes = ['441', '443', '453', '453a', '453b', '463']
surgically_treated_codes.extend(additional_codes)
# check this person has an injury which should be treated here
if count == 0:
assert len(set(person.rt_injuries_for_major_surgery) & set(surgically_treated_codes)) > 0, \
'This person has asked for surgery but does not have an appropriate injury'
# isolate the relevant injury information
person_injuries = df.loc[[person_id], RTI.INJURY_COLUMNS]
# Check whether the person sent to surgery has an injury which actually requires surgery
_, counts = RTI.rti_find_and_count_injuries(person_injuries, surgically_treated_codes)
if counts == 0:
logger.debug(key='rti_general_message',
data=f"This is rti do for major surgery person {person_id} asked for treatment but "
f"doesn't need it.")
# for each injury which has been assigned to be treated by major surgery make sure that the injury hasn't
# already been treated
for code in person.rt_injuries_for_major_surgery:
injury_column, _ = self.rti_find_injury_column(person_id, [code])
date_to_remove_daly_column = RTI.INJURY_DATE_COLUMN_MAP[injury_column]
if not pd.isnull(df.at[person_id, date_to_remove_daly_column]):
logger.debug(key='rti_general_message',
data=f"person {person_id} was assigned for a minor surgery but had already received "
f"treatment")
return
# schedule major surgeries
if 'Major Surgery' not in p['blocked_interventions']:
self.sim.modules['HealthSystem'].schedule_hsi_event(
hsi_event=HSI_RTI_Major_Surgeries(module=self,
person_id=person_id),
priority=0,
topen=self.sim.date + DateOffset(days=count),
tclose=self.sim.date + DateOffset(days=15))
else:
if count == 0:
df.at[person_id, 'rt_injuries_left_untreated'] = df.at[person_id, 'rt_injuries_for_major_surgery']
# remove the injury code from this treatment option
df.at[person_id, 'rt_injuries_for_major_surgery'] = []
# reset the time to check whether the person has died from their injuries
df.loc[person_id, 'rt_date_death_no_med'] = self.sim.date + DateOffset(days=1)
[docs]
def rti_do_for_minor_surgeries(self, person_id, count):
"""
Function called in HSI_RTI_MedicalIntervention to schedule a minor surgery if required. In
HSI_RTI_MedicalIntervention, we determine that they need a surgery. In this function, further to scheduling the
surgery, we double check that they do meet the conditions for needing a surgery. The conditions for needing a
surgery is that they are alive, currently seeking medical intervention and have an injury that is treated by
surgery.
:param person_id: The person requesting major surgeries
:param count: The amount of major surgeries required, used when scheduling surgeries to ensure that two minor
surgeries aren't scheduled on the same day
:return:
"""
df = self.sim.population.props
# Check to see whether they have been sent here from RTI_MedicalIntervention and they haven't been killed by the
# RTI module
assert df.at[person_id, 'rt_med_int'], 'Person sent for treatment did not go through rti med int'
# Isolate the person
if df.at[person_id, 'is_alive']:
# state the codes treated by minor surgery
surgically_treated_codes = ['211', '212', '291', '241', '322', '323', '722', '811', '812', '813a',
'813b', '813c']
# check that the person requesting surgery has an injury in their minor surgery treatment plan
assert len(df.at[person_id, 'rt_injuries_for_minor_surgery']) > 0 or \
len(df.at[person_id, 'rt_injuries_left_untreated']) > 0, 'this person has asked for a minor ' \
'surgery but does not need it'
# check that for each injury due to be treated with a minor surgery, the injury hasn't previously been
# treated
for code in df.at[person_id, 'rt_injuries_for_minor_surgery']:
injury_column, _ = self.rti_find_injury_column(person_id, [code])
if not pd.isnull(df.at[person_id, RTI.INJURY_DATE_COLUMN_MAP[injury_column]]):
logger.debug(key='rti_general_message',
data=f"person {person_id} was assigned for a minor surgery but had already received "
f"treatment")
return
# check that this person's injuries that were decided to be treated with a minor surgery and the injuries
# actually treated by minor surgeries coincide
if count == 0:
assert len(
set(df.at[person_id, 'rt_injuries_for_minor_surgery']) & set(surgically_treated_codes)
) > 0, 'This person has asked for a minor surgery but does not need it'
# Isolate the relevant injury information
person_injuries = df.loc[[person_id], RTI.INJURY_COLUMNS]
# Check whether the person requesting minor surgeries has an injury that requires minor surgery
_, counts = RTI.rti_find_and_count_injuries(person_injuries, surgically_treated_codes)
if counts == 0:
logger.debug(key='rti_general_message',
data=f"person {person_id} was assigned for a minor surgery but has no injury")
return
# schedule the minor surgery
if 'Minor Surgery' not in self.parameters['blocked_interventions']:
self.sim.modules['HealthSystem'].schedule_hsi_event(
hsi_event=HSI_RTI_Minor_Surgeries(module=self,
person_id=person_id),
priority=0,
topen=self.sim.date + DateOffset(days=count),
tclose=self.sim.date + DateOffset(days=15))
else:
if count == 0:
df.at[person_id, 'rt_injuries_left_untreated'] = df.at[person_id, 'rt_injuries_for_minor_surgery']
# remove the injury code from this treatment option
df.at[person_id, 'rt_injuries_for_minor_surgery'] = []
# reset the time to check whether the person has died from their injuries
df.loc[person_id, 'rt_date_death_no_med'] = self.sim.date + DateOffset(days=1)
[docs]
def rti_acute_pain_management(self, person_id):
"""
Function called in HSI_RTI_MedicalIntervention to request pain management. This should be called for every alive
injured person, regardless of what their injuries are. In this function we test whether they meet the
requirements to recieve for pain relief, that is they are alive and currently receiving medical treatment.
:param person_id: The person requesting pain management
:return: n/a
"""
df = self.sim.population.props
if df.at[person_id, 'is_alive']:
# Check to see whether they have been sent here from RTI_MedicalIntervention and they haven't died due to
# rti
assert df.at[person_id, 'rt_med_int'], 'person sent here not been through rti med int'
# Isolate the relevant injury information
person_injuries = df.loc[[person_id], RTI.INJURY_COLUMNS]
# check this person is injured, search they have an injury code that isn't "none".
idx, counts = RTI.rti_find_and_count_injuries(person_injuries,
self.PROPERTIES.get('rt_injury_1').categories[1:])
if counts == 0:
logger.debug(key='rti_general_message',
data=f"person {person_id} requested pain relief but does not need it")
return
# schedule pain management
self.sim.modules['HealthSystem'].schedule_hsi_event(
hsi_event=HSI_RTI_Acute_Pain_Management(module=self,
person_id=person_id),
priority=0,
topen=self.sim.date + DateOffset(days=1),
tclose=self.sim.date + DateOffset(days=15))
[docs]
def rti_ask_for_suture_kit(self, person_id):
"""
Function called by HSI_RTI_MedicalIntervention to centralise all suture kit requests. This function checks
that the person asking for a suture kit meets the requirements to get one. That is they are alive, currently
being treated for their injuries and that they have a laceration which needs stitching.
:param person_id: The person asking for a suture kit
:return: n/a
"""
df = self.sim.population.props
if df.at[person_id, 'is_alive']:
# Check to see whether they have been sent here from RTI_MedicalIntervention and they haven't died due to
# rti
assert df.at[person_id, 'rt_med_int'], 'person sent here not been through rti med int'
# Isolate the relevant injury information
person_injuries = df.loc[[person_id], RTI.INJURY_COLUMNS]
laceration_codes = ['1101', '2101', '3101', '4101', '5101', '6101', '7101', '8101']
# Check they have a laceration which needs stitches
_, counts = RTI.rti_find_and_count_injuries(person_injuries, laceration_codes)
if counts == 0:
logger.debug(key='rti_general_message',
data=f"person {person_id} requested a suture but does not need it")
return
# request suture
self.sim.modules['HealthSystem'].schedule_hsi_event(
hsi_event=HSI_RTI_Suture(module=self,
person_id=person_id),
priority=0,
topen=self.sim.date + DateOffset(days=1),
tclose=self.sim.date + DateOffset(days=15)
)
[docs]
def rti_ask_for_shock_treatment(self, person_id):
"""
A function called by the generic emergency appointment to treat the onset of hypovolemic shock
:param person_id:
:return:
"""
df = self.sim.population.props
if df.at[person_id, 'is_alive']:
assert df.at[person_id, 'rt_in_shock'], 'person requesting shock treatment is not in shock'
self.sim.modules['HealthSystem'].schedule_hsi_event(
hsi_event=HSI_RTI_Shock_Treatment(module=self,
person_id=person_id),
priority=0,
topen=self.sim.date + DateOffset(days=1),
tclose=self.sim.date + DateOffset(days=15)
)
[docs]
def rti_ask_for_burn_treatment(self, person_id):
"""
Function called by HSI_RTI_MedicalIntervention to centralise all burn treatment requests. This function
schedules burn treatments for the person if they meet the requirements, that is they are alive, currently being
treated, and they have a burn which needs to be treated.
:param person_id: The person requesting burn treatment
:return: n/a
"""
df = self.sim.population.props
if df.at[person_id, 'is_alive']:
# Check to see whether they have been sent here from RTI_MedicalIntervention and they haven't died due to
# rti
assert df.at[person_id, 'rt_med_int'], 'person not been through rti med int'
# Isolate the relevant injury information
person_injuries = df.loc[[person_id], RTI.INJURY_COLUMNS]
burn_codes = ['1114', '2114', '3113', '4113', '5113', '7113', '8113']
# Check to see whether they have a burn which needs treatment
_, counts = RTI.rti_find_and_count_injuries(person_injuries, burn_codes)
if counts == 0:
logger.debug(key='rti_general_message',
data=f"person {person_id} requested burn treatment but does not need it")
return
# if this person is alive ask for the hsi event
self.sim.modules['HealthSystem'].schedule_hsi_event(
hsi_event=HSI_RTI_Burn_Management(module=self,
person_id=person_id),
priority=0,
topen=self.sim.date + DateOffset(days=1),
tclose=self.sim.date + DateOffset(days=15)
)
[docs]
def rti_ask_for_fracture_casts(self, person_id):
"""
Function called by HSI_RTI_MedicalIntervention to centralise all fracture casting. This function schedules the
fracture cast treatment if they meet the requirements to ask for it. That is they are alive, currently being
treated and they have a fracture that needs casting (Note that this also handles slings for upper arm/shoulder
fractures).
:param person_id: The person asking for fracture cast/sling
:return: n/a
"""
df = self.sim.population.props
if df.at[person_id, 'is_alive']:
# Check to see whether they have been sent here from RTI_MedicalIntervention and they haven't died due to
# rti
assert df.at[person_id, 'rt_med_int'], 'person sent here not been through rti med int'
# Isolate the relevant injury information
person_injuries = df.loc[[person_id], RTI.INJURY_COLUMNS]
fracture_codes = ['712a', '712b', '712c', '811', '812', '813a', '813b', '813c', '822a', '822b']
# check that the codes assigned for treatment by rt_injuries_to_cast and the codes treated by
# rti_fracture_cast coincide
assert len(set(df.loc[person_id, 'rt_injuries_to_cast']) & set(fracture_codes)) > 0, \
'This person has asked for a fracture cast'
# Check they have an injury treated by HSI_RTI_Fracture_Cast
_, counts = RTI.rti_find_and_count_injuries(person_injuries, fracture_codes)
if counts == 0:
logger.debug(key='rti_general_message',
data=f"person {person_id} requested a fracture cast but does not need it")
return
# if this person is alive request the hsi
if 'Fracture Casts' not in self.parameters['blocked_interventions']:
self.sim.modules['HealthSystem'].schedule_hsi_event(
hsi_event=HSI_RTI_Fracture_Cast(module=self,
person_id=person_id),
priority=0,
topen=self.sim.date + DateOffset(days=1),
tclose=self.sim.date + DateOffset(days=15)
)
else:
df.at[person_id, 'rt_injuries_left_untreated'] = df.at[person_id, 'rt_injuries_to_cast']
df.at[person_id, 'rt_injuries_to_cast'] = []
# reset the time to check whether the person has died from their injuries
df.loc[person_id, 'rt_date_death_no_med'] = self.sim.date + DateOffset(days=1)
[docs]
def rti_ask_for_open_fracture_treatment(self, person_id, counts):
"""Function called by HSI_RTI_MedicalIntervention to centralise open fracture treatment requests. This function
schedules an open fracture event, conditional on whether they are alive, being treated and have an appropriate
injury.
:param person_id: the person requesting a tetanus jab
:param counts: the number of open fractures that requires a treatment
:return: n/a
"""
df = self.sim.population.props
if df.at[person_id, 'is_alive']:
# Check to see whether they have been sent here from RTI_MedicalIntervention and are haven't died due to rti
assert df.at[person_id, 'rt_med_int'], 'person sent here not been through rti med int'
# Isolate the relevant injury information
open_fracture_codes = ['813bo', '813co', '813do', '813eo']
person_injuries = df.loc[[person_id], RTI.INJURY_COLUMNS]
# Check that they have an open fracture
_, counts = RTI.rti_find_and_count_injuries(person_injuries, open_fracture_codes)
if counts == 0:
logger.debug(key='rti_general_message',
data=f"This is rti_ask_for_open_frac person {person_id} asked for treatment but doesn't"
f"need it.")
return
# if the person is alive request the hsi
for i in range(0, counts):
# schedule the treatments, say the treatments occur a day apart for now
self.sim.modules['HealthSystem'].schedule_hsi_event(
hsi_event=HSI_RTI_Open_Fracture_Treatment(module=self, person_id=person_id),
priority=0,
topen=self.sim.date + DateOffset(days=0 + i),
tclose=self.sim.date + DateOffset(days=15 + i)
)
[docs]
def rti_ask_for_tetanus(self, person_id):
"""
Function called by HSI_RTI_MedicalIntervention to centralise all tetanus requests. This function schedules a
tetanus event, conditional on whether they are alive, being treated and have an injury that requires a tetanus
vaccine, i.e. a burn or a laceration.
:param person_id: the person requesting a tetanus jab
:return: n/a
"""
df = self.sim.population.props
if df.at[person_id, 'is_alive']:
# Check to see whether they have been sent here from RTI_MedicalIntervention and are haven't died due to rti
assert df.at[person_id, 'rt_med_int'], 'person sent here not been through rti med int'
# Isolate the relevant injury information
codes_for_tetanus = ['1101', '2101', '3101', '4101', '5101', '7101', '8101',
'1114', '2114', '3113', '4113', '5113', '7113', '8113']
person_injuries = df.loc[[person_id], RTI.INJURY_COLUMNS]
# Check that they have a burn/laceration
_, counts = RTI.rti_find_and_count_injuries(person_injuries, codes_for_tetanus)
if counts == 0:
logger.debug(key='rti_general_message',
data=f"This is rti_ask_for_tetanus person {person_id} asked for treatment but doesn't"
f"need it.")
return
# if this person is alive, ask for the hsi
self.sim.modules['HealthSystem'].schedule_hsi_event(
hsi_event=HSI_RTI_Tetanus_Vaccine(module=self,
person_id=person_id),
priority=0,
topen=self.sim.date + DateOffset(days=1),
tclose=self.sim.date + DateOffset(days=15)
)
[docs]
def schedule_hsi_event_for_tomorrow(self, hsi_event: HSI_Event = None):
"""
A function to reschedule requested events for the following day if they have failed to run
:return:
"""
self.sim.modules['HealthSystem'].schedule_hsi_event(hsi_event, topen=self.sim.date + DateOffset(days=1),
tclose=self.sim.date + DateOffset(days=15), priority=0)
[docs]
def rti_find_injury_column(self, person_id, codes):
"""
This function is a tool to find the injury column an injury code occurs in, when calling this funtion
you will need to guarentee that the person has at least one of the code you are searching for, else this
function will raise an assertion error.
To call this function you need to provide the person who you want to perform the search on and the injury
codes which you want to find the corresponding injury column for. The function/search will return the injury
code which the person has from the list of codes you supplied, and which injury column from rt_injury_1 through
to rt_injury_8, the code appears in.
:param person_id: The person the search is being performed for
:param codes: The injury codes being searched for
:return: which column out of rt_injury_1 to rt_injury_8 the injury code occurs in, and the injury code itself
"""
df = self.sim.population.props
person_injuries = df.loc[person_id, RTI.INJURY_COLUMNS]
injury_column = ''
injury_code = ''
for code in codes:
for col in RTI.INJURY_COLUMNS:
if person_injuries[col] == code:
injury_column = col
injury_code = code
break
# Check that the search found the injury column
assert injury_column != '', df
# Return the found column for the injury code
return injury_column, injury_code
[docs]
def rti_find_all_columns_of_treated_injuries(self, person_id, codes):
"""
This function searches for treated injuries (supplied by the parameter codes) for a specific person, finding and
returning all the columns with treated injuries and all the injury codes for the treated injuries.
:param person_id: The person the search is being performed on
:param codes: The treated injury codes
:return: All columns and codes of the successfully treated injuries
"""
df = self.sim.population.props
person_injuries = df.loc[person_id, RTI.INJURY_COLUMNS]
# create empty variables to return the columns and codes of the treated injuries
columns_to_return = []
codes_to_return = []
# iterate over the codes in the list codes and also the injury columns
for col, val in person_injuries.items():
# Search a sub-dataframe that is non-empty if the code is present is in that column and empty if not
if val in codes:
columns_to_return.append(col)
codes_to_return.append(val)
return columns_to_return, codes_to_return
[docs]
def rti_assign_daly_weights(self, injured_index):
"""
This function assigns DALY weights associated with each injury when they happen.
By default this function gives the DALY weight for each condition without treatment, this will then be swapped
for the DALY weight associated with the injury with treatment when treatment occurs.
The properties that this function alters are rt_disability, which is the property used to report the
disability burden that this person has and rt_debugging_DALY_wt, which stores the true value of the
the disability.
:param injured_index: The people who have been involved in a road traffic accident for the current month and did
not die on the scene of the crash
:return: n/a
"""
df = self.sim.population.props
# ==============================================================================================================
# Check that those sent here have been involved in a road traffic accident
assert df.loc[injured_index, 'rt_road_traffic_inc'].all()
# Check everyone here has at least one injury to be given a daly weight to
assert (df.loc[injured_index, 'rt_injury_1'] != "none").all()
# Check everyone here is alive and hasn't died due to rti
rti_deaths = ['RTI_death_without_med', 'RTI_death_with_med', 'RTI_unavailable_med', 'RTI_imm_death',
'RTI_death_shock']
assert (sum(~df.loc[injured_index, 'cause_of_death'].isin(rti_deaths)) == len(injured_index)) & \
(sum(df.loc[injured_index, 'rt_imm_death']) == 0)
selected_for_rti_inj = df.loc[injured_index, RTI.INJURY_COLUMNS]
daly_change = selected_for_rti_inj.applymap(
lambda code: self.ASSIGN_INJURIES_AND_DALY_CHANGES[code][1]
).sum(axis=1, numeric_only=True)
df.loc[injured_index, 'rt_disability'] += daly_change
# Store the true sum of DALY weights in the df
df.loc[injured_index, 'rt_debugging_DALY_wt'] = df.loc[injured_index, 'rt_disability']
# Find who's disability burden is greater than one
DALYweightoverlimit = df.index[df['rt_disability'] > 1]
# Set the total daly weights to one in this case
df.loc[DALYweightoverlimit, 'rt_disability'] = 1
# Find who's disability burden is less than one
DALYweightunderlimit = df.index[df.rt_road_traffic_inc & ~ df.rt_imm_death & (df['rt_disability'] <= 0)]
# Check that no one has a disability burden less than or equal to zero
assert len(DALYweightunderlimit) == 0, ('Someone has not been given an injury burden',
selected_for_rti_inj.loc[DALYweightunderlimit])
df.loc[DALYweightunderlimit, 'rt_disability'] = 0
assert (df.loc[injured_index, 'rt_disability'] > 0).all()
[docs]
def rti_alter_daly_post_treatment(self, person_id, codes):
"""
This function removes the DALY weight associated with each injury code after treatment is complete. This
function is called by RTI_Recovery_event which removes asks to remove the DALY weight when the injury has
healed
The properties that this function alters are rt_disability, which is the property used to report the
disability burden that this person has and rt_debugging_DALY_wt, which stores the true value of the
the disability.
:param person_id: The person who needs a daly weight removed as their injury has healed
:param codes: The injury codes for the healed injury/injuries
:return: n/a
"""
df = self.sim.population.props
# Check everyone here has at least one injury to be alter the daly weight to
person_injuries = df.loc[[person_id], RTI.INJURY_COLUMNS]
# check this person is injured, search they have an injury code that isn't "none"
idx, counts = RTI.rti_find_and_count_injuries(person_injuries,
self.PROPERTIES.get('rt_injury_1').categories[1:])
assert counts > 0, 'This person has asked for medical treatment despite not being injured'
# Check everyone here is alive and hasn't died on scene
assert not df.at[person_id, 'rt_imm_death']
# ------------------------------- Remove the daly weights for treated injuries ---------------------------------
# update the total values of the daly weights
df.at[person_id, 'rt_debugging_DALY_wt'] += \
sum([self.ASSIGN_INJURIES_AND_DALY_CHANGES[code][3] for code in codes])
# round off any potential floating point errors
df.at[person_id, 'rt_debugging_DALY_wt'] = np.round(df.at[person_id, 'rt_debugging_DALY_wt'], 4)
# if the person's true total for daly weights is greater than one, report rt_disability as one, if not
# report the true disability burden.
if df.at[person_id, 'rt_debugging_DALY_wt'] > 1:
df.at[person_id, 'rt_disability'] = 1
else:
df.at[person_id, 'rt_disability'] = df.at[person_id, 'rt_debugging_DALY_wt']
# if the reported daly weight is below zero add make the model report the true (and always positive) daly weight
if df.at[person_id, 'rt_disability'] < 0:
df.at[person_id, 'rt_disability'] = df.at[person_id, 'rt_debugging_DALY_wt']
# Make sure the true disability burden is greater or equal to zero
if df.at[person_id, 'rt_debugging_DALY_wt'] < 0:
logger.debug(key='rti_general_message',
data=f"person {person_id} has had too many daly weights removed")
df.at[person_id, 'rt_debugging_DALY_wt'] = 0
# the reported disability should satisfy 0<=disability<=1, check that they do
# Check the daly weights fall within the accepted bounds
if (df.at[person_id, 'rt_disability'] < 0):
logger.warning(key="warning", data="rt disability < 0 in rti_alter_daly_post_treatment")
df.at[person_id, 'rt_disability'] = 0
if (df.at[person_id, 'rt_disability'] > 1):
logger.warning(key="warning", data="rt disability > 1 in rti_alter_daly_post_treatment")
df.at[person_id, 'rt_disability'] = 1
# remover the treated injury code from the person using rti_treated_injuries
RTI.rti_treated_injuries(self, person_id, codes)
[docs]
def rti_swap_injury_daly_upon_treatment(self, person_id, codes):
"""
This function swaps certain DALY weight codes upon when a person receives treatment(s). Some injuries have a
different daly weight associated with them for the treated and untreated injuries. If an injury is 'swap-able'
then this function removes the old daly weight for the untreated injury and gives the daly weight for the
treated injury.
The properties that this function alters are rt_disability, which is the property used to report the
disability burden that this person has and rt_debugging_DALY_wt, which stores the true value of the
the disability.
:param person_id: The person who has received treatment
:param codes: the 'swap-able' injury code
:return: n/a
"""
df = self.sim.population.props
# Check the people that are sent here have had medical treatment
assert df.at[person_id, 'rt_med_int']
# Check they have an appropriate injury code to swap
swapping_codes = RTI.SWAPPING_CODES[:]
relevant_codes = np.intersect1d(codes, swapping_codes)
person_injuries = df.loc[[person_id], RTI.INJURY_COLUMNS]
# check this person is injured, search they have an injury code that is swappable
idx, counts = RTI.rti_find_and_count_injuries(person_injuries, list(relevant_codes))
assert counts > 0, 'This person has asked to swap an injury code, but it is not swap-able'
# If there are any permanent injuries which are due to be swapped, remove the "P" writted at the start of injury
# code in order to access the injury dictionary
relevant_codes = [code.replace('P', '') for code in relevant_codes]
# swap the relevant code's daly weight, from the daly weight associated with the injury without treatment
# and the daly weight for the disability with treatment.
# keep track of the changes to the daly weights
# update the disability burdens
df.at[person_id, 'rt_debugging_DALY_wt'] += \
sum([self.ASSIGN_INJURIES_AND_DALY_CHANGES[code][2] for code in relevant_codes])
df.at[person_id, 'rt_debugging_DALY_wt'] = np.round(df.at[person_id, 'rt_debugging_DALY_wt'], 4)
# TODO: the injury '5113' seems to being treated multiple times for certain people, causing a repeated DALY
# weight swap which ultimately results in a negative daly weight. I need to work out why this is happening, the
# if statement below is a temporary fix
# Check that the person's true disability burden is positive
if df.at[person_id, 'rt_debugging_DALY_wt'] < 0:
logger.debug(key='rti_general_message',
data=f"person {person_id} has had too many daly weights removed")
df.at[person_id, 'rt_debugging_DALY_wt'] = 0
# catch rounding point errors where the disability weights should be zero but aren't
if df.at[person_id, 'rt_disability'] < 0:
df.at[person_id, 'rt_disability'] = 0
# Catch cases where the disability burden is greater than one in reality but needs to be
# capped at one, if not report the true disability burden
if df.at[person_id, 'rt_debugging_DALY_wt'] > 1:
df.at[person_id, 'rt_disability'] = 1
else:
df.at[person_id, 'rt_disability'] = df.at[person_id, 'rt_debugging_DALY_wt']
# Check the daly weights fall within the accepted bounds
if (df.at[person_id, 'rt_disability'] < 0):
logger.warning(key="warning", data="rt disability < 0 in rti_swap_injury_daly_upon_treatment")
df.at[person_id, 'rt_disability'] = 0
if (df.at[person_id, 'rt_disability'] > 1):
logger.warning(key="warning", data="rt disability > 1 in rti_swap_injury_daly_upon_treatment")
df.at[person_id, 'rt_disability'] = 1
[docs]
def rti_determine_LOS(self, person_id):
"""
This function determines the length of stay a person sent to the health care system will require, based on how
severe their injuries are (determined by the person's ISS score). Currently I use data from China, but once a
more appropriate source of data is found I can swap this over.
:param person_id: The person who needs their LOS determined
:return: the inpatient days required to treat this person (Their LOS)
"""
p = self.parameters
df = self.sim.population.props
def draw_days(_mean, _sd):
return int(self.rng.normal(_mean, _sd, 1))
# Create the length of stays required for each ISS score boundaries and check that they are >=0
rt_iss_score = df.at[person_id, 'rt_ISS_score']
if rt_iss_score < 4:
days_until_treatment_end = draw_days(p["mean_los_ISS_less_than_4"], p["sd_los_ISS_less_than_4"])
elif 4 <= rt_iss_score < 9:
days_until_treatment_end = draw_days(p["mean_los_ISS_4_to_8"], p["sd_los_ISS_4_to_8"])
elif 9 <= rt_iss_score < 16:
days_until_treatment_end = draw_days(p["mean_los_ISS_9_to_15"], p["sd_los_ISS_9_to_15"])
elif 16 <= rt_iss_score < 25:
days_until_treatment_end = draw_days(p["mean_los_ISS_16_to_24"], p["sd_los_ISS_16_to_24"])
elif 25 <= rt_iss_score:
days_until_treatment_end = draw_days(p["mean_los_ISS_more_than_25"], p["sd_los_ISS_more_that_25"])
else:
days_until_treatment_end = 0
# Make sure inpatient days is less that max available
if days_until_treatment_end > 150:
days_until_treatment_end = 150
# Return the LOS
return max(days_until_treatment_end, 0)
[docs]
@staticmethod
def rti_find_and_count_injuries(persons_injury_properties: pd.DataFrame, injury_codes: list):
"""
A function that searches a user given dataframe for a list of injuries (injury_codes). If the injury code is
found in the dataframe, this function returns the index for who has the injury/injuries and the number of
injuries found. This function works much faster if the dataframe is smaller, hence why the searched dataframe
is a parameter in the function.
:param persons_injury_properties: The dataframe to search for the tlo injury codes in
:param injury_codes: The injury codes to search for in the data frame
:return: the df index of who has the injuries and how many injuries in the search were found.
"""
assert isinstance(persons_injury_properties, pd.DataFrame)
assert isinstance(injury_codes, list)
injury_counts = persons_injury_properties.isin(injury_codes).sum(axis=1)
people_with_given_injuries = injury_counts[injury_counts > 0]
return people_with_given_injuries.index, people_with_given_injuries.sum()
[docs]
def rti_treated_injuries(self, person_id, tlo_injury_codes):
"""
A function that takes a person with treated injuries and removes the injury code from the properties rt_injury_1
to rt_injury_8
The properties that this function alters are rt_injury_1 through rt_injury_8 and the symptoms properties
:param person_id: The person who needs an injury code removed
:param tlo_injury_codes: the injury code(s) to be removed
:return: n/a
"""
df = self.sim.population.props
# Isolate the relevant injury information
permanent_injuries = {'P133', 'P133a', 'P133b', 'P133c', 'P133d', 'P134', 'P134a', 'P134b', 'P135', 'P673',
'P673a', 'P673b', 'P674', 'P674a', 'P674b', 'P675', 'P675a', 'P675b', 'P676', 'P782a',
'P782b', 'P782c', 'P783', 'P882', 'P883', 'P884'}
person_injuries = df.loc[person_id, RTI.INJURY_COLUMNS]
# only remove non-permanent injuries
codes_to_remove = [c for c in tlo_injury_codes if c not in permanent_injuries]
# get injury columns for all codes to remove
injury_cols = person_injuries.index[person_injuries.isin(codes_to_remove)].tolist()
# if no injuries to reset, exit
if len(injury_cols) == 0:
return
# Reset the treated injury code to "none"
df.loc[person_id, injury_cols] = "none"
# Reset symptoms so that after being treated for an injury the person won't interact with the
# health system again.
if df.at[person_id, 'sy_injury'] != 0:
self.sim.modules['SymptomManager'].change_symptom(
person_id=person_id,
disease_module=self.sim.modules['RTI'],
add_or_remove='-',
symptom_string='injury')
if df.at[person_id, 'sy_severe_trauma'] != 0:
self.sim.modules['SymptomManager'].change_symptom(
person_id=person_id,
disease_module=self.sim.modules['RTI'],
add_or_remove='-',
symptom_string='severe_trauma')
[docs]
def on_birth(self, mother_id, child_id):
"""
When a person is born this function sets up the default properties for the road traffic injuries module
:param mother_id: The mother
:param child_id: The newborn
:return: n/a
"""
df = self.sim.population.props
df.at[child_id, 'rt_road_traffic_inc'] = False
df.at[child_id, 'rt_inj_severity'] = "none" # default: no one has been injured in a RTI
for injury_index in RTI.INJURY_INDICES:
df.at[child_id, f'rt_injury_{injury_index}'] = "none"
df.at[child_id, f'rt_date_to_remove_daly_{injury_index}'] = pd.NaT
df.at[child_id, 'rt_in_shock'] = False
df.at[child_id, 'rt_death_from_shock'] = False
df.at[child_id, 'rt_injuries_to_cast'] = []
df.at[child_id, 'rt_injuries_for_minor_surgery'] = []
df.at[child_id, 'rt_injuries_for_major_surgery'] = []
df.at[child_id, 'rt_injuries_to_heal_with_time'] = []
df.at[child_id, 'rt_injuries_for_open_fracture_treatment'] = []
df.at[child_id, 'rt_polytrauma'] = False
df.at[child_id, 'rt_ISS_score'] = 0
df.at[child_id, 'rt_imm_death'] = False
df.at[child_id, 'rt_perm_disability'] = False
df.at[child_id, 'rt_med_int'] = False # default: no one has a had medical intervention
df.at[child_id, 'rt_in_icu_or_hdu'] = False
df.at[child_id, 'rt_diagnosed'] = False
df.at[child_id, 'rt_recovery_no_med'] = False # default: no recovery without medical intervention
df.at[child_id, 'rt_post_med_death'] = False # default: no death after medical intervention
df.at[child_id, 'rt_no_med_death'] = False
df.at[child_id, 'rt_unavailable_med_death'] = False
df.at[child_id, 'rt_disability'] = 0 # default: no disability due to RTI
df.at[child_id, 'rt_date_inj'] = pd.NaT
df.at[child_id, 'rt_MAIS_military_score'] = 0
df.at[child_id, 'rt_date_death_no_med'] = pd.NaT
df.at[child_id, 'rt_debugging_DALY_wt'] = 0
df.at[child_id, 'rt_injuries_left_untreated'] = []
[docs]
def look_up_consumable_item_codes(self):
"""Look up the item codes that used in the HSI in the module"""
get_item_codes = self.sim.modules['HealthSystem'].get_item_code_from_item_name
self.cons_item_codes = dict()
self.cons_item_codes['shock_treatment_child'] = {
get_item_codes("ringer's lactate (Hartmann's solution), 1000 ml_12_IDA"): 500,
get_item_codes("Dextrose (glucose) 5%, 1000ml_each_CMST"): 500,
get_item_codes('Cannula iv (winged with injection pot) 18_each_CMST'): 1,
get_item_codes('Blood, one unit'): 2,
get_item_codes("Oxygen, 1000 liters, primarily with oxygen cylinders"): 23_040
}
self.cons_item_codes['shock_treatment_adult'] = {
get_item_codes("ringer's lactate (Hartmann's solution), 1000 ml_12_IDA"): 2000,
get_item_codes('Cannula iv (winged with injection pot) 18_each_CMST'): 1,
get_item_codes('Blood, one unit'): 2,
get_item_codes("Oxygen, 1000 liters, primarily with oxygen cylinders"): 23_040
}
self.cons_item_codes['fracture_treatment_plaster'] = {
get_item_codes('Plaster of Paris (POP) 10cm x 7.5cm slab_12_CMST'): 1
# This is for one fracture.
}
self.cons_item_codes['fracture_treatment_bandage'] = {
get_item_codes('Bandage, crepe 7.5cm x 1.4m long , when stretched'): 200,
# (The 200 is a standard assumption for the amount of bandage needed, irrespective of the number of
# fractures.)
}
self.cons_item_codes['open_fracture_treatment'] = {
get_item_codes('Ceftriaxone 1g, PFR_each_CMST'): 2000,
get_item_codes('Cetrimide 15% + chlorhexidine 1.5% solution.for dilution _5_CMST'): 500,
get_item_codes("Gauze, absorbent 90cm x 40m_each_CMST"): 100,
get_item_codes('Suture pack'): 1,
}
self.cons_item_codes["open_fracture_treatment_additional_if_contaminated"] = {
get_item_codes('Metronidazole, injection, 500 mg in 100 ml vial'): 1500
}
self.cons_item_codes['laceration_treatment_suture_pack'] = {
get_item_codes('Suture pack'): 0,
}
self.cons_item_codes['laceration_treatment_cetrimide_chlorhexidine'] = {
get_item_codes('Cetrimide 15% + chlorhexidine 1.5% solution.for dilution _5_CMST'): 500,
}
self.cons_item_codes['burn_treatment_per_burn'] = {
get_item_codes("Gauze, absorbent 90cm x 40m_each_CMST"): 0,
get_item_codes('Cetrimide 15% + chlorhexidine 1.5% solution.for dilution _5_CMST'): 0,
}
self.cons_item_codes['ringers lactate for multiple burns'] = {
get_item_codes("ringer's lactate (Hartmann's solution), 1000 ml_12_IDA"): 4000
}
self.cons_item_codes['tetanus_treatment'] = {get_item_codes('Tetanus toxoid, injection'): 1}
self.cons_item_codes['pain_management_mild_under_16'] = {get_item_codes("Paracetamol 500mg_1000_CMST"): 8000}
self.cons_item_codes['pain_management_mild_above_16'] = {
get_item_codes("diclofenac sodium 25 mg, enteric coated_1000_IDA"): 300
}
self.cons_item_codes['pain_management_moderate'] = {
get_item_codes("tramadol HCl 100 mg/2 ml, for injection_100_IDA"): 300
}
self.cons_item_codes['pain_management_severe'] = {
get_item_codes("morphine sulphate 10 mg/ml, 1 ml, injection (nt)_10_IDA"): 120
}
self.cons_item_codes['major_surgery'] = {
# request a general anaesthetic
get_item_codes("Halothane (fluothane)_250ml_CMST"): 100,
# clean the site of the surgery
get_item_codes("Chlorhexidine 1.5% solution_5_CMST"): 500,
# tools to begin surgery
get_item_codes("Scalpel blade size 22 (individually wrapped)_100_CMST"): 1,
# administer an IV
get_item_codes('Cannula iv (winged with injection pot) 18_each_CMST'): 1,
get_item_codes("Giving set iv administration + needle 15 drops/ml_each_CMST"): 1,
get_item_codes("ringer's lactate (Hartmann's solution), 1000 ml_12_IDA"): 2000,
# repair incision made
get_item_codes("Suture pack"): 1,
get_item_codes("Gauze, absorbent 90cm x 40m_each_CMST"): 100,
# administer pain killer
get_item_codes('Pethidine, 50 mg/ml, 2 ml ampoule'): 6,
# administer antibiotic
get_item_codes("Ampicillin injection 500mg, PFR_each_CMST"): 1000,
# equipment used by surgeon, gloves and facemask
get_item_codes('Disposables gloves, powder free, 100 pieces per box'): 1,
get_item_codes('surgical face mask, disp., with metal nose piece_50_IDA'): 1,
# request syringe
get_item_codes("Syringe, Autodisable SoloShot IX "): 1
}
self.cons_item_codes['minor_surgery'] = {
# request a local anaesthetic
get_item_codes("Halothane (fluothane)_250ml_CMST"): 100,
# clean the site of the surgery
get_item_codes("Chlorhexidine 1.5% solution_5_CMST"): 500,
# tools to begin surgery
get_item_codes("Scalpel blade size 22 (individually wrapped)_100_CMST"): 1,
# administer an IV
get_item_codes('Cannula iv (winged with injection pot) 18_each_CMST'): 1,
get_item_codes("Giving set iv administration + needle 15 drops/ml_each_CMST"): 1,
get_item_codes("ringer's lactate (Hartmann's solution), 1000 ml_12_IDA"): 2000,
# repair incision made
get_item_codes("Suture pack"): 1,
get_item_codes("Gauze, absorbent 90cm x 40m_each_CMST"): 100,
# administer pain killer
get_item_codes('Pethidine, 50 mg/ml, 2 ml ampoule'): 6,
# administer antibiotic
get_item_codes("Ampicillin injection 500mg, PFR_each_CMST"): 1000,
# equipment used by surgeon, gloves and facemask
get_item_codes('Disposables gloves, powder free, 100 pieces per box'): 1,
get_item_codes('surgical face mask, disp., with metal nose piece_50_IDA'): 1,
# request syringe
get_item_codes("Syringe, Autodisable SoloShot IX "): 1
}
# Function to get the consumables for fracture treatment, which depends on the number of fractures:
self.cons_item_codes['fracture_treatment'] = lambda num_fractures: {
**{item: num_fractures for item in self.cons_item_codes['fracture_treatment_plaster']},
**self.cons_item_codes['fracture_treatment_bandage']
}
# Function to get the consumables for laceration treatment, which depends on the number of lacerations:
self.cons_item_codes['laceration_treatment'] = lambda num_laceration: {
**{item: num_laceration for item in self.cons_item_codes['laceration_treatment_suture_pack']},
**self.cons_item_codes['laceration_treatment_cetrimide_chlorhexidine']
}
self.cons_item_codes['burn_treatment'] = lambda num_burns: {
item: num_burns for item in self.cons_item_codes['burn_treatment_per_burn']
}
[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.
"""
logger.debug(key='rti_general_message',
data=f"This is RTI, being alerted about a health system interaction person %d for: %s, {person_id}"
f", {treatment_id}"
)
[docs]
def report_daly_values(self):
# This must send back a pd.Series or pd.DataFrame that reports on the average daly-weights that have been
# experienced by persons in the previous month. Only rows for alive-persons must be returned.
# The names of the series of columns is taken to be the label of the cause of this disability.
# It will be recorded by the healthburden module as <ModuleName>_<Cause>.
logger.debug(key='rti_general_message', data='This is RTI reporting my daly values')
df = self.sim.population.props
disability_series_for_alive_persons = df.loc[df.is_alive, "rt_disability"]
return disability_series_for_alive_persons
[docs]
def rti_assign_injuries(self, number):
"""
A function that can be called specifying the number of people affected by RTI injuries
and provides outputs for the number of injuries each person experiences from a RTI event, the location of the
injury, the TLO injury categories and the severity of the injuries. The severity of the injuries will then be
used to calculate the injury severity score (ISS), which will then inform mortality and disability from road
traffic injuries with treatment and the military abreviated injury score (MAIS) which will be used to predict
mortality without medical intervention.
:param number: The number of people who need injuries assigned to them
:return: injurydescription - a dataframe for the injury/injuries summarised in the TLO injury code form along
with data on their ISS score, used for calculating mortality and whether they have
polytrauma or not.
Todo: see if we can include the following factors for injury severity (taken from a preprint sent to me after
MOH meeting):
- The setting of the patient (rural/urban) as rural location was a predictor for severe injury AOR 2.41
(1.49-3.90)
- Seatbelt use (none compared to using AOR 4.49 (1.47-13.76))
- Role of person in crash (Different risk for different type, see paper)
- Alcohol use (AOR 1.74 (1.11-2.74) compared to none)
"""
p = self.parameters
# Import the fitted distribution of injured body regions
number_of_injured_body_regions_distribution = p['number_of_injured_body_regions_distribution']
# Get the probability distribution for the likelihood of a certain body region being injured.
injlocdist = p['injury_location_distribution']
# create a dataframe to store the injury persons information in
inj_df = pd.DataFrame(columns=['Injury_codes', 'AIS_scores', 'ISS', 'Polytrauma', 'MAIS', 'Number_of_injuries'])
# Create empty lists to store information of each person's injuries to be used in logging:
# predicted injury location
predinjlocs = []
# predicted injury severity
predinjsev = []
# predicted injury category
predinjcat = []
# create empty lists to store the qualitative description of injury severity and the number of injuries
# each person has
severity_category = []
# ============================= Begin assigning injuries to people =====================================
# Iterate over the total number of injured people
for n in range(0, number):
# Generate a random number which will decide how many injuries the person will have,
ninj = self.rng.choice(number_of_injured_body_regions_distribution[0],
p=number_of_injured_body_regions_distribution[1])
# create an empty list which stores the injury chosen for this person
injuries_chosen = []
# Create an empty vector which will store the severity of the injuries
injais = []
# Create an empty vector to store injury MAIS scores in
injmais = []
# generate the locations of the injuries for this person (chosed without replacement, meaning that each
# injury corresponds to a single body region)
injurylocation = self.rng.choice(injlocdist[0], ninj, p=injlocdist[1], replace=False)
# iterate over the chosen injury locations to determine the exact injuries that this person will have
for injlocs in injurylocation:
# get a list of the injuries that occur at this location to filter the dictionary
# self.ASSIGN_INJURIES_AND_DALY_CHANGES to the relevant injury information
injuries_at_location = [injury for injury in self.ASSIGN_INJURIES_AND_DALY_CHANGES.keys() if
injury.startswith(str(injlocs))]
# find the probability of each injury based on the above filter
prob_of_each_injury_at_location = [self.ASSIGN_INJURIES_AND_DALY_CHANGES[injury][0][0] for injury in
injuries_at_location]
# make sure there are no rounding errors (meaning all probabilities sum to one)
prob_of_each_injury_at_location = np.divide(prob_of_each_injury_at_location,
sum(prob_of_each_injury_at_location))
# chose an injury to occur at this location
injury_chosen = self.rng.choice(injuries_at_location, p=prob_of_each_injury_at_location)
# store this persons chosen injury at this location
injuries_chosen.append(injury_chosen)
# Store this person's injury location (used in logging)
predinjlocs.append(self.ASSIGN_INJURIES_AND_DALY_CHANGES[injury_chosen][0][1])
# store the injury category chosen (used in logging)
predinjcat.append(self.ASSIGN_INJURIES_AND_DALY_CHANGES[injury_chosen][0][2])
# store the severity of the injury chosen
injais.append(self.ASSIGN_INJURIES_AND_DALY_CHANGES[injury_chosen][0][3])
# store the MAIS score
injmais.append(self.ASSIGN_INJURIES_AND_DALY_CHANGES[injury_chosen][0][4])
# create the data needed for an additional row, the injuries chosen for this person, their corresponding
# AIS scores, their ISS score (calculated as the summed square of their three most severly injured body
# regions, whether or not they have polytrauma (calculated if they have two or more body regions with an ISS
# score greater than 2), their MAIS score and the number of injuries they have
new_row = {'Injury_codes': injuries_chosen,
'AIS_scores': injais,
'ISS': sum(sorted(np.square(injais))[-3:]),
'Polytrauma': sum(i > 2 for i in injais) > 1,
'MAIS': max(injmais),
'Number_of_injuries': ninj}
inj_df.loc[len(inj_df)] = new_row
# If person has an ISS score less than 15 they have a mild injury, otherwise severe
if new_row['ISS'] < 15:
severity_category.append('mild')
else:
severity_category.append('severe')
# Store this person's injury information into the lists which house each individual person's injury
# information
predinjsev.append(injais)
# If there is at least one injured person, expand the returned dataframe so that each injury has it's own column
if len(inj_df) > 0:
# create a copy of the injury codes
listed_injuries = inj_df.copy()
# expand the injury codes into their own columns
listed_injuries = listed_injuries['Injury_codes'].apply(pd.Series)
# rename the columns
listed_injuries = listed_injuries.rename(columns=lambda x: 'rt_injury_' + str(x + 1))
# join the expanded injuries to the injury df
inj_df = inj_df.join(listed_injuries)
# Fill dataframe entries where a person has not had an injury assigned with 'none'
inj_df = inj_df.fillna("none")
# Begin logging injury information
# ============================ Injury category incidence ======================================================
# log the incidence of each injury category
# count the number of injuries that fall in each category
amputationcounts = sum(1 for i in predinjcat if i == '8')
burncounts = sum(1 for i in predinjcat if i == '11')
fraccounts = sum(1 for i in predinjcat if i == '1')
tbicounts = sum(1 for i in predinjcat if i == '3')
minorinjurycounts = sum(1 for i in predinjcat if i == '10')
spinalcordinjurycounts = sum(1 for i in predinjcat if i == '7')
other_counts = sum(1 for i in predinjcat if i in ['2', '4', '5', '6', '9'])
# calculate the incidence of this injury in the population
df = self.sim.population.props
n_alive = len(df.is_alive)
inc_amputations = amputationcounts / ((n_alive - amputationcounts) * 1 / 12) * 100000
inc_burns = burncounts / ((n_alive - burncounts) * 1 / 12) * 100000
inc_fractures = fraccounts / ((n_alive - fraccounts) * 1 / 12) * 100000
inc_tbi = tbicounts / ((n_alive - tbicounts) * 1 / 12) * 100000
inc_sci = spinalcordinjurycounts / ((n_alive - spinalcordinjurycounts) * 1 / 12) * 100000
inc_minor = minorinjurycounts / ((n_alive - minorinjurycounts) * 1 / 12) * 100000
inc_other = other_counts / ((n_alive - other_counts) * 1 / 12) * 100000
tot_inc_all_inj = inc_amputations + inc_burns + inc_fractures + inc_tbi + inc_sci + inc_minor + inc_other
if number > 0:
number_of_injuries = int(inj_df['Number_of_injuries'].iloc[0])
else:
number_of_injuries = 0
dict_to_output = {'inc_amputations': inc_amputations,
'inc_burns': inc_burns,
'inc_fractures': inc_fractures,
'inc_tbi': inc_tbi,
'inc_sci': inc_sci,
'inc_minor': inc_minor,
'inc_other': inc_other,
'tot_inc_injuries': tot_inc_all_inj,
'number_of_injuries': number_of_injuries}
logger.info(key='Inj_category_incidence',
data=dict_to_output,
description='Incidence of each injury grouped as per the GBD definition')
# Log injury information
# Get injury severity information in an easily interpreted form to be logged.
# create a list of the predicted injury severity scores
flattened_injury_ais = [str(item) for sublist in predinjsev for item in sublist]
injury_info = {'Number_of_injuries': number_of_injuries,
'Location_of_injuries': predinjlocs,
'Injury_category': predinjcat,
'Per_injury_severity': flattened_injury_ais,
'Per_person_injury_severity': inj_df['ISS'].to_list(),
'Per_person_MAIS_score': inj_df['MAIS'].to_list(),
'Per_person_severity_category': severity_category
}
logger.info(key='Injury_information',
data=injury_info,
description='Relevant information on the injuries from road traffic accidents when they are '
'assigned')
# log the fraction of lower extremity fractions that are open
flattened_injuries = [str(item) for sublist in inj_df['Injury_codes'].to_list() for item in sublist]
lx_frac_codes = ['811', '813do', '812', '813eo', '813', '813a', '813b', '813bo', '813c', '813co']
lx_open_frac_codes = ['813do', '813eo', '813bo', '813co']
n_lx_fracs = len([inj for inj in flattened_injuries if inj in lx_frac_codes])
n_open_lx_fracs = len([inj for inj in flattened_injuries if inj in lx_open_frac_codes])
if n_lx_fracs > 0:
proportion_lx_fracture_open = n_open_lx_fracs / n_lx_fracs
else:
proportion_lx_fracture_open = float("nan")
injury_info = {'Proportion_lx_fracture_open': proportion_lx_fracture_open}
logger.info(key='Open_fracture_information',
data=injury_info,
description='The proportion of fractures that are open in specific body regions')
# Finally return the injury description information
return inj_df
[docs]
def _common_first_appt_steps(
self,
person_id: int,
individual_properties: IndividualProperties,
schedule_hsi_event: HSIEventScheduler,
) -> None:
"""
Shared logic steps that are used by the RTI module when a generic HSI
event is to be scheduled.
"""
# Things to do upon a person presenting at a Non-Emergency Generic
# HSI if they have an injury.
persons_injuries = [
individual_properties[injury] for injury in RTI.INJURY_COLUMNS
]
if (
pd.isnull(individual_properties["cause_of_death"])
and not individual_properties["rt_diagnosed"]
):
if set(RTI.INJURIES_REQ_IMAGING).intersection(set(persons_injuries)):
if individual_properties["is_alive"]:
event = HSI_RTI_Imaging_Event(module=self, person_id=person_id)
schedule_hsi_event(
event,
priority=0,
topen=self.sim.date + DateOffset(days=1),
tclose=self.sim.date + DateOffset(days=15),
)
individual_properties["rt_diagnosed"] = True
# The injured person has been diagnosed in A&E and needs to progress further
# through the health system.
# They will then me scheduled a generic "medical intervention" appointment,
# serving three purposes:
# 1. Determine which treatments they require for their injuries to and shedule them,
# 2. Contain them in the health care system with inpatient days,
# 3. The appointment treats injuries that heal over time without further need for
# resources in the health system.
# Check this person is injured, search they have an injury code that isn't "none"
_, counts = RTI.rti_find_and_count_injuries(
pd.DataFrame(data=[persons_injuries], columns=RTI.INJURY_COLUMNS),
RTI.INJURY_CODES[1:],
)
# also test whether the regular injury symptom has been given to the person via spurious symptoms
assert (counts > 0) or self.sim.modules[
"SymptomManager"
].spurious_symptoms, (
"This person has asked for medical treatment despite not being injured"
)
# If they meet the requirements, send them to HSI_RTI_MedicalIntervention for further treatment
# Using counts condition to stop spurious symptoms progressing people through the model
if counts > 0:
event = HSI_RTI_Medical_Intervention(module=self, person_id=person_id)
schedule_hsi_event(
event,
priority=0,
topen=self.sim.date,
)
# We now check if they need shock treatment
if (
individual_properties["rt_in_shock"]
and individual_properties["is_alive"]
):
event = HSI_RTI_Shock_Treatment(module=self, person_id=person_id)
schedule_hsi_event(
event,
priority=0,
topen=self.sim.date + DateOffset(days=1),
tclose=self.sim.date + DateOffset(days=15),
)
[docs]
def do_at_generic_first_appt(
self,
person_id: int,
individual_properties: IndividualProperties,
symptoms: List[str],
schedule_hsi_event: HSIEventScheduler,
**kwargs
) -> None:
if "injury" in symptoms:
return self._common_first_appt_steps(
person_id=person_id,
individual_properties=individual_properties,
schedule_hsi_event=schedule_hsi_event,
)
[docs]
def do_at_generic_first_appt_emergency(
self,
person_id: int,
individual_properties: IndividualProperties,
symptoms: List[str],
schedule_hsi_event: HSIEventScheduler,
**kwargs
) -> None:
# Same process is followed for emergency and non emergency appointments, except the
# initial symptom check
if "severe_trauma" in symptoms:
return self._common_first_appt_steps(
person_id=person_id,
individual_properties=individual_properties,
schedule_hsi_event=schedule_hsi_event,
)
# ---------------------------------------------------------------------------------------------------------
# DISEASE MODULE EVENTS
#
# These are the events which drive the simulation of the disease. It may be a regular event that updates
# the status of all the population of subsections of it at one time. There may also be a set of events
# that represent disease events for particular persons.
# ---------------------------------------------------------------------------------------------------------
[docs]
class RTIPollingEvent(RegularEvent, PopulationScopeEventMixin):
"""The regular RTI event which handles all the initial RTI related changes to the dataframe. It can be thought of
as the actual road traffic accident occurring. Specifically the event decides who is involved in a road traffic
accident every month (via the linear model helper class), whether those involved in a road traffic accident die on
scene or are given injuries (via the assign_injuries function) which they will attempt to interact with the health
system with for treatment.
Those who don't die on scene and are injured then attempt to go to an emergency generic first appointment
This event will change the rt_ properties:
1) rt_road_traffic_inc - False when not involved in a collision, True when RTI_Event decides they are in a collision
2) rt_date_inj - Change to current date if the person has been involved in a road traffic accident
3) rt_imm_death - True if they die on the scene of the crash, false otherwise
4) rt_injury_1 through to rt_injury_8 - a series of 8 properties which stores the injuries that need treating as a
code
5) rt_ISS_score - The metric used to calculate the probability of mortality from the person's injuries
6) rt_MAIS_military_score - The metric used to calculate the probability of mortality without medical intervention
7) rt_disability - after injuries are assigned to a person, RTI_event calls rti_assign_daly_weights to match the
person's injury codes in rt_injury_1 through 8 to their corresponding DALY weights
8) rt_polytrauma - If the person's injuries fit the definition for polytrauma we keep track of this here and use it
to calculate the probability for mortality later on.
9) rt_date_death_no_med - the projected date to determine mortality for those who haven't sought medical care
10) rt_inj_severity - The qualitative description of the severity of this person's injuries
11) the symptoms this person has
"""
[docs]
def __init__(self, module):
"""Schedule to take place every month
"""
super().__init__(module, frequency=DateOffset(months=1))
p = module.parameters
# Parameters which transition the model between states
self.base_1m_prob_rti = (p['base_rate_injrti'] / 12)
if 'reduce_incidence' in p['allowed_interventions']:
self.base_1m_prob_rti = self.base_1m_prob_rti * 0.335
self.rr_injrti_age04 = p['rr_injrti_age04']
self.rr_injrti_age59 = p['rr_injrti_age59']
self.rr_injrti_age1017 = p['rr_injrti_age1017']
self.rr_injrti_age1829 = p['rr_injrti_age1829']
self.rr_injrti_age3039 = p['rr_injrti_age3039']
self.rr_injrti_age4049 = p['rr_injrti_age4049']
self.rr_injrti_age5059 = p['rr_injrti_age5059']
self.rr_injrti_age6069 = p['rr_injrti_age6069']
self.rr_injrti_age7079 = p['rr_injrti_age7079']
self.rr_injrti_male = p['rr_injrti_male']
self.rr_injrti_excessalcohol = p['rr_injrti_excessalcohol']
self.imm_death_proportion_rti = p['imm_death_proportion_rti']
self.prob_bleeding_leads_to_shock = p['prob_bleeding_leads_to_shock']
self.rt_emergency_care_ISS_score_cut_off = p['rt_emergency_care_ISS_score_cut_off']
[docs]
def apply(self, population):
"""Apply this event to the population.
:param population: the current population
"""
df = population.props
now = self.sim.date
# Reset injury properties after death, get an index of people who have died due to RTI, all causes
diedfromrtiidx = df.index[df.rt_imm_death | df.rt_post_med_death | df.rt_no_med_death | df.rt_death_from_shock |
df.rt_unavailable_med_death]
df.loc[diedfromrtiidx, "rt_imm_death"] = False
df.loc[diedfromrtiidx, "rt_post_med_death"] = False
df.loc[diedfromrtiidx, "rt_no_med_death"] = False
df.loc[diedfromrtiidx, "rt_unavailable_med_death"] = False
df.loc[diedfromrtiidx, "rt_disability"] = 0
df.loc[diedfromrtiidx, "rt_med_int"] = False
df.loc[diedfromrtiidx, 'rt_in_icu_or_hdu'] = False
for index, row in df.loc[diedfromrtiidx].iterrows():
df.at[index, 'rt_injuries_to_cast'] = []
df.at[index, 'rt_injuries_for_minor_surgery'] = []
df.at[index, 'rt_injuries_for_major_surgery'] = []
df.at[index, 'rt_injuries_to_heal_with_time'] = []
df.at[index, 'rt_injuries_for_open_fracture_treatment'] = []
df.at[index, 'rt_injuries_left_untreated'] = []
df.loc[diedfromrtiidx, "rt_diagnosed"] = False
df.loc[diedfromrtiidx, "rt_polytrauma"] = False
df.loc[diedfromrtiidx, "rt_inj_severity"] = "none"
df.loc[diedfromrtiidx, "rt_perm_disability"] = False
for injury_column in RTI.INJURY_COLUMNS:
df.loc[diedfromrtiidx, injury_column] = "none"
for date_to_remove_daly_column in RTI.DATE_TO_REMOVE_DALY_COLUMNS:
df.loc[diedfromrtiidx, date_to_remove_daly_column] = pd.NaT
df.loc[diedfromrtiidx, 'rt_date_death_no_med'] = pd.NaT
df.loc[diedfromrtiidx, 'rt_MAIS_military_score'] = 0
df.loc[diedfromrtiidx, 'rt_debugging_DALY_wt'] = 0
df.loc[diedfromrtiidx, 'rt_in_shock'] = False
# reset whether they have been selected for an injury this month
df['rt_road_traffic_inc'] = False
# --------------------------------- UPDATING OF RTI OVER TIME -------------------------------------------------
# Currently we have the following conditions for being able to be involved in a road traffic injury, they are
# alive, they aren't currently injured, they didn't die immediately in
# a road traffic injury in the last month and finally, they aren't currently being treated for a road traffic
# injury.
rt_current_non_ind = df.index[df.is_alive & ~df.rt_road_traffic_inc & ~df.rt_imm_death & ~df.rt_med_int &
(df.rt_inj_severity == "none")]
# ========= Update for people currently not involved in a RTI, make some involved in a RTI event ==============
# Use linear model helper class
eq = LinearModel(LinearModelType.MULTIPLICATIVE,
self.base_1m_prob_rti,
Predictor('sex').when('M', self.rr_injrti_male),
Predictor(
'age_years',
conditions_are_mutually_exclusive=True
)
.when('.between(0,4)', self.rr_injrti_age04)
.when('.between(5,9)', self.rr_injrti_age59)
.when('.between(10,17)', self.rr_injrti_age1017)
.when('.between(18,29)', self.rr_injrti_age1829)
.when('.between(30,39)', self.rr_injrti_age3039)
.when('.between(40,49)', self.rr_injrti_age4049)
.when('.between(50,59)', self.rr_injrti_age5059)
.when('.between(60,69)', self.rr_injrti_age6069)
.when('.between(70,79)', self.rr_injrti_age7079),
Predictor('li_ex_alc').when(True, self.rr_injrti_excessalcohol)
)
pred = eq.predict(df.loc[rt_current_non_ind])
random_draw_in_rti = self.module.rng.random_sample(size=len(rt_current_non_ind))
selected_for_rti = rt_current_non_ind[pred > random_draw_in_rti]
# Update to say they have been involved in a rti
df.loc[selected_for_rti, 'rt_road_traffic_inc'] = True
# Set the date that people were injured to now
df.loc[selected_for_rti, 'rt_date_inj'] = now
# ========================= Take those involved in a RTI and assign some to death ==============================
# This section accounts for pre-hospital mortality, where a person is so severy injured that they die before
# being able to seek medical care
selected_to_die = selected_for_rti[self.imm_death_proportion_rti >
self.module.rng.random_sample(size=len(selected_for_rti))]
# Keep track of who experience pre-hospital mortality with the property rt_imm_death
df.loc[selected_to_die, 'rt_imm_death'] = True
# For each person selected to experience pre-hospital mortality, schedule an InstantaneosDeath event
for individual_id in selected_to_die:
self.sim.modules['Demography'].do_death(individual_id=individual_id, cause="RTI_imm_death",
originating_module=self.module)
# ============= Take those remaining people involved in a RTI and assign injuries to them ==================
# Drop those who have died immediately
selected_for_rti_inj_idx = selected_for_rti.drop(selected_to_die)
# Check that none remain
assert len(selected_for_rti_inj_idx.intersection(selected_to_die)) == 0
# take a copy dataframe, used to get the index of the population affected by RTI
selected_for_rti_inj = df.loc[selected_for_rti_inj_idx]
# Again make sure that those who have injuries assigned to them are alive, involved in a crash and didn't die on
# scene
selected_for_rti_inj = selected_for_rti_inj.loc[df.is_alive & df.rt_road_traffic_inc & ~df.rt_imm_death]
# To stop people who have died from causes outside of the RTI module progressing through the model, remove
# any person with the condition 'cause_of_death' is not null
died_elsewhere_index = selected_for_rti_inj[~ selected_for_rti_inj['cause_of_death'].isnull()].index
# drop the died_elsewhere_index from selected_for_rti_inj
selected_for_rti_inj.drop(died_elsewhere_index, inplace=True)
# Create shorthand link to RTI module
road_traffic_injuries = self.sim.modules['RTI']
# if people have been chosen to be injured, assign the injuries using the assign injuries function
description = road_traffic_injuries.rti_assign_injuries(len(selected_for_rti_inj))
# replace the nan values with 'none', this is so that the injuries can be copied over from this temporarily used
# pandas dataframe will fit in with the categories in the columns rt_injury_1 through rt_injury_8
description = description.replace('nan', 'none')
# set the index of the description dataframe, so that we can join it to the selected_for_rti_inj dataframe
description = description.set_index(selected_for_rti_inj.index)
# copy over values from the assign injury dataframe to self.sim.population.props
df.loc[selected_for_rti_inj.index, 'rt_ISS_score'] = \
description.loc[selected_for_rti_inj.index, 'ISS'].astype(int)
df.loc[selected_for_rti_inj.index, 'rt_MAIS_military_score'] = \
description.loc[selected_for_rti_inj.index, 'MAIS'].astype(int)
# ======================== Apply the injuries to the population dataframe ======================================
# Find the corresponding column names
injury_columns = pd.Index(RTI.INJURY_COLUMNS)
matching_columns = description.columns.intersection(injury_columns)
for col in matching_columns:
df.loc[selected_for_rti_inj.index, col] = description.loc[selected_for_rti_inj.index, col]
# Run assert statements to make sure the model is behaving as it should
# All those who are injured in a road traffic accident have this noted in the property 'rt_road_traffic_inc'
assert df.loc[selected_for_rti, 'rt_road_traffic_inc'].all()
# All those who are involved in a road traffic accident have these noted in the property 'rt_date_inj'
assert (df.loc[selected_for_rti, 'rt_date_inj'] != pd.NaT).all()
# All those who are injures and do not die immediately have an ISS score > 0
assert len(df.loc[df.rt_road_traffic_inc & ~df.rt_imm_death, 'rt_ISS_score'] > 0) == \
len(df.loc[df.rt_road_traffic_inc & ~df.rt_imm_death])
# ========================== Determine who will experience shock from blood loss ==============================
internal_bleeding_codes = ['361', '363', '461', '463', '813bo', '813co', '813do', '813eo']
df = self.sim.population.props
potential_shock_index, _ = \
road_traffic_injuries.rti_find_and_count_injuries(df.loc[df.rt_road_traffic_inc, RTI.INJURY_COLUMNS],
internal_bleeding_codes)
rand_for_shock = self.module.rng.random_sample(len(potential_shock_index))
shock_index = potential_shock_index[self.prob_bleeding_leads_to_shock > rand_for_shock]
df.loc[shock_index, 'rt_in_shock'] = True
# log the percentage of those with RTIs in shock
percent_in_shock = \
len(shock_index) / len(selected_for_rti_inj) if len(selected_for_rti_inj) > 0 else float("nan")
logger.info(key='Percent_of_shock_in_rti',
data={'Percent_of_shock_in_rti': percent_in_shock},
description='The percentage of those assigned injuries who were also assign the shock property')
# ========================== Decide survival time without medical intervention ================================
# todo: find better time for survival data without med int for ISS scores
# Assign a date in the future for which when the simulation reaches that date, the person's mortality will be
# checked if they haven't sought care
df.loc[selected_for_rti_inj.index, 'rt_date_death_no_med'] = now + DateOffset(days=7)
# ============================ Injury severity classification =================================================
# Find those with mild injuries and update the rt_inj_severity property so they have a mild injury
injured_this_month = df.loc[selected_for_rti_inj.index]
mild_rti_idx = injured_this_month.index[injured_this_month.is_alive & injured_this_month['rt_ISS_score'] < 15]
df.loc[mild_rti_idx, 'rt_inj_severity'] = 'mild'
# Find those with severe injuries and update the rt_inj_severity property so they have a severe injury
severe_rti_idx = injured_this_month.index[injured_this_month['rt_ISS_score'] >= 15]
df.loc[severe_rti_idx, 'rt_inj_severity'] = 'severe'
# check that everyone who has been assigned an injury this month has an associated injury severity
assert sum(df.loc[df.rt_road_traffic_inc & ~df.rt_imm_death & (df.rt_date_inj == now), 'rt_inj_severity']
!= 'none') == len(selected_for_rti_inj.index)
# Find those with polytrauma and update the rt_polytrauma property so they have polytrauma
polytrauma_idx = description.loc[description.Polytrauma].index
df.loc[polytrauma_idx, 'rt_polytrauma'] = True
# Assign daly weights for each person's injuries with the function rti_assign_daly_weights
road_traffic_injuries.rti_assign_daly_weights(selected_for_rti_inj.index)
# =============================== Health seeking behaviour set up =======================================
# Set up health seeking behaviour. Two symptoms are used in the RTI module, the generic injury symptom and an
# emergency symptom 'severe_trauma'.
# The condition to be sent to the health care system: 1) They must be alive 2) They must have been involved in a
# road traffic accident 3) they must have not died immediately in the accident 4) they must not have been to an
# A and E department previously and been diagnosed
# The symptom they are assigned depends injury severity, those with mild injuries will be assigned the generic
# symptom, those with severe injuries will have the emergency injury symptom
# Create the logical conditions for each symptom
condition_to_be_sent_to_em = \
df.is_alive & df.rt_road_traffic_inc & ~df.rt_diagnosed & ~df.rt_imm_death & (df.rt_date_inj == now) & \
(df.rt_injury_1 != "none") & (df.rt_ISS_score >= self.rt_emergency_care_ISS_score_cut_off)
condition_to_be_sent_to_begin_non_emergency = \
df.is_alive & df.rt_road_traffic_inc & ~df.rt_diagnosed & ~df.rt_imm_death & (df.rt_date_inj == now) & \
(df.rt_injury_1 != "none") & (df.rt_ISS_score < self.rt_emergency_care_ISS_score_cut_off)
# check that all those who meet the conditions to try and seek healthcare have at least one injury
assert sum(df.loc[condition_to_be_sent_to_em, 'rt_injury_1'] != "none") == \
len(df.loc[condition_to_be_sent_to_em])
assert sum(df.loc[condition_to_be_sent_to_begin_non_emergency, 'rt_injury_1'] != "none") == \
len(df.loc[condition_to_be_sent_to_begin_non_emergency])
# create indexes of people to be assigned each rti symptom
em_idx = df.index[condition_to_be_sent_to_em]
non_em_idx = df.index[condition_to_be_sent_to_begin_non_emergency]
# Assign the symptoms
self.sim.modules['SymptomManager'].change_symptom(
person_id=em_idx.tolist(),
disease_module=self.module,
add_or_remove='+',
symptom_string='severe_trauma',
)
self.sim.modules['SymptomManager'].change_symptom(
person_id=non_em_idx.tolist(),
disease_module=self.module,
add_or_remove='+',
symptom_string='injury',
)
[docs]
class RTI_Check_Death_No_Med(RegularEvent, PopulationScopeEventMixin):
"""
A regular event which organises whether a person who has not received medical treatment should die as a result of
their injuries. This even makes use of the maximum AIS-military score, a trauma scoring system developed for
injuries in a military environment, assumed here to be an indicator of the probability of mortality without
access to a medical system.
The properties this function changes are:
1) rt_no_med_death - the boolean property tracking who dies from road traffic injuries without medical intervention
2) rt_date_death_no_med - resetting the date to check the person's mortality without medical intervention if
they survive
3) rt_disability - if the person survives a non-fatal injury then this injury may heal and therefore the disability
burden is changed
4) rt_debugging_DALY_wt - if the person survives a non-fatal injury then this injury may heal and therefore the
disability burden is changed, this property keeping track of the true disability burden
5) rt_date_to_remove_daly_{injury_index} - In the event of recovering from a non-fatal injury without medical
intervention a recovery date will scheduled for the relevant injury
index
If the person is sent here and they don't die, we need to correctly model the level of disability they experience
from their untreated injuries, some injuries that are left untreated will have an associated daly weight for long
term disability without treatment, others don't.
# todo: consult with a doctor about the likelihood of survival without medical treatment
Currently I am assuming that anyone with an injury severity score of 9 or higher will seek care and have an
emergency symptom, that means that I have to decide what to do with the following injuries:
Lacerations - [1101, 2101, 3101, 4101, 5101, 7101, 8101]
What would a laceration do without stitching? Take longer to heal most likely
Fractures - ['112', '211', '212, '412', '612', '712', '712a', '712b', '712c', '811', '812']
Some fractures have an associated level of disability to them, others do not. So things like fractured radius/ulna
have a code to swap, but others don't. Some are of the no treatment type, such as fractured skulls, fractured ribs
or fractured vertebrae, so we can just add the same recovery time for these injuries. So '112', '412' and '612' will
need to have recovery events checked and recovered.
Dislocations will presumably be popped back into place, the injury will feasably be able to heal but most likely
with more pain and probably with more time
Amputations - ['782','782a', '782b', '782c', '882']
Amputations will presumably trigger emergency health seeking behaviour so they shouldn't turn up here really
soft tissue injuries - ['241', '342', '441', '442']
Presumably soft tissue injuries that turn up here will heal over time but with more pain
Internal organ injury - ['552']
Injury to the gastrointestinal organs can result in complications later on, but
Internal bleedings - ['361', '461']
Surviving internal bleeding is concievably possible, these are comparitively minor bleeds
"""
[docs]
def __init__(self, module):
super().__init__(module, frequency=DateOffset(days=1))
assert isinstance(module, RTI)
p = module.parameters
# Load parameters used by this event
self.prob_death_MAIS3 = p['prob_death_MAIS3']
self.prob_death_MAIS4 = p['prob_death_MAIS4']
self.prob_death_MAIS5 = p['prob_death_MAIS5']
self.prob_death_MAIS6 = p['prob_death_MAIS6']
self.daly_wt_radius_ulna_fracture_short_term_with_without_treatment = \
p['daly_wt_radius_ulna_fracture_short_term_with_without_treatment']
self.daly_wt_radius_ulna_fracture_long_term_without_treatment = \
p['daly_wt_radius_ulna_fracture_long_term_without_treatment']
self.daly_wt_foot_fracture_short_term_with_without_treatment = \
p['daly_wt_foot_fracture_short_term_with_without_treatment']
self.daly_wt_foot_fracture_long_term_without_treatment = \
p['daly_wt_foot_fracture_long_term_without_treatment']
self.daly_wt_hip_fracture_short_term_with_without_treatment = \
p['daly_wt_hip_fracture_short_term_with_without_treatment']
self.daly_wt_hip_fracture_long_term_without_treatment = \
p['daly_wt_hip_fracture_long_term_without_treatment']
self.daly_wt_pelvis_fracture_short_term = p['daly_wt_pelvis_fracture_short_term']
self.daly_wt_pelvis_fracture_long_term = \
p['daly_wt_pelvis_fracture_long_term']
self.daly_wt_femur_fracture_short_term = p['daly_wt_femur_fracture_short_term']
self.daly_wt_femur_fracture_long_term_without_treatment = \
p['daly_wt_femur_fracture_long_term_without_treatment']
self.no_treatment_mortality_mais_cutoff = p['unavailable_treatment_mortality_mais_cutoff']
self.no_treatment_ISS_cut_off = p['consider_death_no_treatment_ISS_cut_off']
[docs]
def apply(self, population):
df = population.props
now = self.sim.date
probabilities_of_death = {
'1': 0,
'2': 0,
'3': self.prob_death_MAIS3,
'4': self.prob_death_MAIS4,
'5': self.prob_death_MAIS5,
'6': self.prob_death_MAIS6
}
# check if anyone is due to have their mortality without medical intervention determined today
if len(df.loc[df['rt_date_death_no_med'] == now]) > 0:
# Get an index of those scheduled to have their mortality checked
due_to_die_today_without_med_int = df.loc[df['rt_date_death_no_med'] == now].index
# iterate over those scheduled to die
for person in due_to_die_today_without_med_int:
# Create a random number to determine mortality
rand_for_death = self.module.rng.random_sample(1)
# create a variable to show if a person has died due to their untreated injuries
# find which injuries have been untreated
untreated_injuries = []
persons_injuries = df.loc[[person], RTI.INJURY_COLUMNS]
non_empty_injuries = persons_injuries[persons_injuries != "none"]
non_empty_injuries = non_empty_injuries.dropna(axis=1)
for col in non_empty_injuries:
if pd.isnull(df.loc[person, RTI.INJURY_DATE_COLUMN_MAP[col]]):
untreated_injuries.append(df.at[person, col])
mais_scores = [1]
for injury in untreated_injuries:
mais_scores.append(self.module.ASSIGN_INJURIES_AND_DALY_CHANGES[injury][0][-1])
max_untreated_injury = max(mais_scores)
prob_death = probabilities_of_death[str(max_untreated_injury)]
if df.loc[person, 'rt_med_int'] and (max_untreated_injury < self.no_treatment_mortality_mais_cutoff):
# filter out non serious injuries from the consideration of mortality
prob_death = 0
if (rand_for_death < prob_death) and (df.at[person, 'rt_ISS_score'] > self.no_treatment_ISS_cut_off):
# If determined to die, schedule a death without med
df.loc[person, 'rt_no_med_death'] = True
self.sim.modules['Demography'].do_death(individual_id=person, cause="RTI_death_without_med",
originating_module=self.module)
else:
# If the people do not die from their injuries despite not getting care, we have to decide when and
# to what degree their injuries will heal.
df.loc[[person], 'rt_recovery_no_med'] = True
# Reset the date to check if they die
df.loc[[person], 'rt_date_death_no_med'] = pd.NaT
swapping_codes = ['712c', '811', '813a', '813b', '813c']
# create a dictionary to reference changes to daly weights done here
swapping_daly_weights_lookup = {
'712c': (- self.daly_wt_radius_ulna_fracture_short_term_with_without_treatment +
self.daly_wt_radius_ulna_fracture_long_term_without_treatment),
'811': (- self.daly_wt_foot_fracture_short_term_with_without_treatment +
self.daly_wt_foot_fracture_long_term_without_treatment),
'813a': (- self.daly_wt_hip_fracture_short_term_with_without_treatment +
self.daly_wt_hip_fracture_long_term_without_treatment),
'813b': - self.daly_wt_pelvis_fracture_short_term + self.daly_wt_pelvis_fracture_long_term,
'813c': (- self.daly_wt_femur_fracture_short_term +
self.daly_wt_femur_fracture_long_term_without_treatment),
'none': 0
}
road_traffic_injuries = self.sim.modules['RTI']
# If those who haven't sought health care have an injury for which we have a daly code
# associated with that injury long term without treatment, swap it
# Iterate over the person's injuries
injuries = df.loc[[person], RTI.INJURY_COLUMNS].values.tolist()
# Cannot iterate correctly over list like [[1,2,3]], so need to flatten
flattened_injuries = [item for sublist in injuries for item in sublist if item != 'none']
if df.loc[person, 'rt_med_int']:
flattened_injuries = [injury for injury in flattened_injuries if injury in
df.loc[person, 'rt_injuries_left_untreated']]
persons_injuries = df.loc[[person], RTI.INJURY_COLUMNS]
for code in flattened_injuries:
swapable_code = np.intersect1d(code, swapping_codes)
if len(swapable_code) > 0:
swapable_code = swapable_code[0]
else:
swapable_code = 'none'
# check that the person has the injury code
_, counts = road_traffic_injuries.rti_find_and_count_injuries(persons_injuries, [code])
assert counts > 0
df.loc[person, 'rt_debugging_DALY_wt'] += swapping_daly_weights_lookup[swapable_code]
if df.loc[person, 'rt_debugging_DALY_wt'] > 1:
df.loc[person, 'rt_disability'] = 1
else:
df.loc[person, 'rt_disability'] = df.loc[person, 'rt_debugging_DALY_wt']
# if the code is swappable, swap it
if df.loc[person, 'rt_disability'] < 0:
df.loc[person, 'rt_disability'] = 0
if df.loc[person, 'rt_disability'] > 1:
df.loc[person, 'rt_disability'] = 1
# If they don't have a swappable code, schedule the healing of the injury
# get the persons injuries
persons_injuries = df.loc[[person], RTI.INJURY_COLUMNS]
non_empty_injuries = persons_injuries[persons_injuries != "none"]
non_empty_injuries = non_empty_injuries.dropna(axis=1)
injury_column, _ = road_traffic_injuries.rti_find_injury_column(
person, [code]
)
date_to_remove_daly_column = RTI.INJURY_DATE_COLUMN_MAP[injury_column]
# assign a recovery date
# not all injuries have an assigned duration of recovery. These are more serious injuries that
# would normally be sent directly to the health system. In the instance that a serious injury
# occurs and no treatment is recieved but the person survives assume they will be disabled for
# the duration of the simulation
# if they haven't sought care at all we don't need to specify which injuries need a recovery
# date assigned
if not df.loc[person, 'rt_med_int']:
if code in self.module.NO_TREATMENT_RECOVERY_TIMES_IN_DAYS.keys():
df.loc[person, date_to_remove_daly_column] = (
self.sim.date + DateOffset(
days=self.module.NO_TREATMENT_RECOVERY_TIMES_IN_DAYS[code]
)
)
else:
df.loc[person, date_to_remove_daly_column] = (
self.sim.end_date + DateOffset(days=1)
)
else:
# if they have sought medical care and it hasn't been provided, we need to make sure only
# the untreated injuries have a recovery date assigned here
code_has_recovery_time = code in self.module.NO_TREATMENT_RECOVERY_TIMES_IN_DAYS.keys()
code_is_left_untreated = code in df.loc[person, 'rt_injuries_left_untreated']
if code_has_recovery_time & code_is_left_untreated:
df.loc[person, date_to_remove_daly_column] = (
self.sim.date + DateOffset(
days=self.module.NO_TREATMENT_RECOVERY_TIMES_IN_DAYS[code]
)
)
else:
df.loc[person, date_to_remove_daly_column] = (
self.sim.end_date + DateOffset(days=1)
)
# remove the injury code from columns to be treated, as they have not sought care and have
# survived without treatment
if code in df.loc[person, 'rt_injuries_left_untreated']:
if code in df.loc[person, 'rt_injuries_to_cast']:
df.loc[person, 'rt_injuries_to_cast'].remove(code)
if code in df.loc[person, 'rt_injuries_for_minor_surgery']:
df.loc[person, 'rt_injuries_for_minor_surgery'].remove(code)
if code in df.loc[person, 'rt_injuries_for_major_surgery']:
df.loc[person, 'rt_injuries_for_major_surgery'].remove(code)
if code in df.loc[person, 'rt_injuries_to_heal_with_time']:
df.loc[person, 'rt_injuries_to_heal_with_time'].remove(code)
if code in df.loc[person, 'rt_injuries_to_heal_with_time']:
df.loc[person, 'rt_injuries_to_heal_with_time'].remove(code)
if code in df.loc[person, 'rt_injuries_for_open_fracture_treatment']:
df.loc[person, 'rt_injuries_for_open_fracture_treatment'].remove(code)
assert df.loc[person, date_to_remove_daly_column] > self.sim.date
[docs]
class RTI_Recovery_Event(RegularEvent, PopulationScopeEventMixin):
"""
A regular event which checks the recovery date determined by each injury in columns rt_injury_1 through
rt_injury_8, which is stored in columns rt_date_to_remove_daly_1, through rt_date_to_remove_daly_8
respectively. This event checks the dates stored in the rt_date_to_remove_daly_* columns, when the
date matches one of the entries, the daly weight is removed and the injury is fully healed.
The properties changed in this functions is:
1) rt_date_to_remove_daly - resetting the date to remove the daly weight for each injury once the date is
reached in the sim
2) rt_inj_severity - resetting the person's injury severity once and injury is healed
3) rt_injuries_to_heal_with_time - resetting the list of injuries that are due to heal over time once healed
4) rt_injuries_for_minor_surgery - resetting the list of injuries that are treated with minor surgery once
healed
5) rt_injuries_for_major_surgery - resetting the list of injuries that are treated with major surgery once
healed
6) rt_injuries_for_open_fracture_treatment - resetting the list of injuries that are treated with open fracture
treatment once healed
7) rt_injuries_to_cast - resetting the list of injuries that are treated with fracture cast treatment once healed
"""
[docs]
def __init__(self, module):
super().__init__(module, frequency=DateOffset(days=1))
assert isinstance(module, RTI)
[docs]
def apply(self, population):
road_traffic_injuries = self.module
df = population.props
now = self.sim.date
# # Isolate the relevant population
any_not_null = df.loc[
df.is_alive, RTI.DATE_TO_REMOVE_DALY_COLUMNS
].notnull().any(axis=1)
relevant_population = any_not_null.index[any_not_null]
# Iterate over all the injured people who are having medical treatment
for person in relevant_population:
# Iterate over all the 'rt_date_to_remove_daly_*' dates
# storing queried date values from dataframe for reuse
date_values = {}
for date_column in RTI.DATE_TO_REMOVE_DALY_COLUMNS:
date_values[date_column] = df.at[person, date_column]
# check that a recovery date hasn't been assigned to the past
if not pd.isnull(date_values[date_column]):
assert date_values[date_column] >= self.sim.date, (
"Recovery date assigned to past"
)
# check if the recovery date is today
if date_values[date_column] == now:
# find the injury code associated with the healed injury
injury_column = RTI.INJURY_DATE_COLUMN_MAP[date_column]
code_to_remove = [df.loc[person, injury_column]]
# Set the healed injury recovery data back to the default state
# Also record updated values in date_values dict
date_values[date_column] = df.loc[person, date_column] = pd.NaT
# Remove the daly weight associated with the healed injury code
person_injuries = df.loc[[person], RTI.INJURY_COLUMNS]
_, counts = RTI.rti_find_and_count_injuries(
person_injuries, self.module.INJURY_CODES[1:]
)
if counts == 0:
pass
else:
road_traffic_injuries.rti_alter_daly_post_treatment(person, code_to_remove)
# Check whether all their injuries are healed so the injury properties can be reset
if all(pd.isnull(date) for date in date_values.values()):
# remove the injury severity as person is uninjured
df.at[person, 'rt_inj_severity'] = "none"
# Check that the date to remove dalys is removed if the date to remove the daly is today
assert all(date != now for date in date_values.values())
# finally ensure the reported disability burden is an appropriate value
current_rt_disability = df.at[person, 'rt_disability']
if current_rt_disability < 0:
df.at[person, 'rt_disability'] = 0
if current_rt_disability > 1:
df.at[person, 'rt_disability'] = 1
# ---------------------------------------------------------------------------------------------------------
# RTI SPECIFIC HEALTH SYSTEM INTERACTION EVENTS
#
# Here are all the different Health System Interactions Events that this module will use.
# ---------------------------------------------------------------------------------------------------------
[docs]
class HSI_RTI_Imaging_Event(HSI_Event, IndividualScopeEventMixin):
"""This HSI event is triggered by the generic first appointments. After first arriving into the health system at
either level 0 or level 1, should the injured person require a imaging to diagnose their injuries this HSI
event is caused and x-ray or ct scans are provided as needed"""
[docs]
def __init__(self, module, person_id):
super().__init__(module, person_id=person_id)
assert isinstance(module, RTI)
self.TREATMENT_ID = 'Rti_Imaging'
self.EXPECTED_APPT_FOOTPRINT = self.make_appt_footprint({'DiagRadio': 1})
self.ACCEPTED_FACILITY_LEVEL = '1b'
[docs]
def apply(self, person_id, squeeze_factor):
self.sim.population.props.at[person_id, 'rt_diagnosed'] = True
road_traffic_injuries = self.sim.modules['RTI']
road_traffic_injuries.rti_injury_diagnosis(person_id, self.EXPECTED_APPT_FOOTPRINT)
if 'DiagRadio' in list(self.EXPECTED_APPT_FOOTPRINT.keys()):
self.add_equipment(self.healthcare_system.equipment.from_pkg_names('X-ray'))
elif 'Tomography' in list(self.EXPECTED_APPT_FOOTPRINT.keys()):
self.ACCEPTED_FACILITY_LEVEL = '3'
self.add_equipment({'Computed Tomography (CT machine)', 'CT scanner accessories'})
[docs]
def did_not_run(self, *args, **kwargs):
pass
def _get_untreated_injury_columns(person_id, population_dataframe):
return [
RTI.INJURY_DATE_COLUMN_MAP[date_to_remove_daly_column]
for date_to_remove_daly_column in RTI.DATE_TO_REMOVE_DALY_COLUMNS
if pd.isnull(population_dataframe.loc[person_id, date_to_remove_daly_column])
]
[docs]
class HSI_RTI_Medical_Intervention(HSI_Event, IndividualScopeEventMixin):
"""This is a Health System Interaction Event.
An appointment of a person who has experienced a road traffic injury, had their injuries diagnosed through A&E
and now needs treatment.
This appointment is designed to organise the treatments needed. In the __init__ section the appointment footprint
is altered to fit the requirements of the person's treatment need. In this section we count the number of
minor/major surgeries required and determine how long they will be in the health system for. For some injuries,
the treatment plan is not entirely set into stone and may vary, for example, some skull fractures will need surgery
whilst some will not. The treatment plan in its entirety is designed here.
In the apply section, we send those who need surgeries to either HSI_RTI_Major_Surgery or HSI_RTI_Minor_Surgery,
those who need stitches to HSI_RTI_Suture, those who need burn treatment to HSI_RTI_Burn_Management and those who
need fracture casts to HSI_RTI_Casting.
Pain medication is also requested here with HSI_RTI_Acute_Pain_Management.
The properties changed in this event are:
rt_injuries_for_major_surgery - the injuries that are determined to be treated by major surgery are stored in
this list property
rt_injuries_for_minor_surgery - the injuries that are determined to be treated by minor surgery are stored in
this list property
rt_injuries_to_cast - the injuries that are determined to be treated with fracture casts are stored in this list
property
rt_injuries_for_open_fracture_treatment - the injuries that are determined to be treated with open fractre treatment
are stored in this list property
rt_injuries_to_heal_with_time - the injuries that are determined to heal with time are stored in this list property
rt_date_to_remove_daly - recovery dates for the heal with time injuries are set here
rt_date_death_no_med - the date to check mortality without medical intervention is removed as this person has
sought medical care
rt_med_int - the bool property that shows whether a person has sought medical care or not
"""
# TODO: include treatment or at least transfer between facilities, e.g. at KCH "Most patients transferred from
# either a health center, 2463 (47.2%), or district hospital, 1996 (38.3%)"
[docs]
def __init__(self, module, person_id):
super().__init__(module, person_id=person_id)
self.TREATMENT_ID = 'Rti_MedicalIntervention'
self.EXPECTED_APPT_FOOTPRINT = self.make_appt_footprint({'AccidentsandEmerg': 1})
self.ACCEPTED_FACILITY_LEVEL = '1b'
self.BEDDAYS_FOOTPRINT = self.make_beddays_footprint({'general_bed': 8})
p = module.parameters
# Load the parameters used in this event
self.prob_depressed_skull_fracture = p['prob_depressed_skull_fracture'] # proportion of depressed skull
# fractures in https://doi.org/10.1016/j.wneu.2017.09.084
self.prob_mild_burns = p['prob_mild_burns'] # proportion of burns accross SSA with TBSA < 10
# https://doi.org/10.1016/j.burns.2015.04.006
self.prob_TBI_require_craniotomy = p['prob_TBI_require_craniotomy']
self.prob_exploratory_laparotomy = p['prob_exploratory_laparotomy']
self.prob_dislocation_requires_surgery = p['prob_dislocation_requires_surgery']
self.allowed_interventions = p['allowed_interventions']
self.prob_perm_disability_with_treatment_severe_TBI = p['prob_perm_disability_with_treatment_severe_TBI']
# Create an empty list for injuries that are potentially healed without further medical intervention
self.heal_with_time_injuries = []
[docs]
def apply(self, person_id, squeeze_factor):
road_traffic_injuries = self.sim.modules['RTI']
df = self.sim.population.props
p = self.sim.modules['RTI'].parameters
person = df.loc[person_id]
# ======================= Design treatment plan, appointment type =============================================
""" Here, RTI_MedInt designs the treatment plan of the person's injuries, the following determines what the
major and minor surgery requirements will be
"""
# Create variables to count how many major or minor surgeries will be required to treat this person
major_surgery_counts = 0
minor_surgery_counts = 0
# Isolate the relevant injury information
person_injuries = df.loc[[person_id], RTI.INJURY_COLUMNS]
# todo: work out if the amputations need to be included as a swap or if they already exist
# create a dictionary to store the probability of each possible treatment for applicable injuries, we are
# assuming that any amputation treatment plan will just be a major surgery for now
treatment_plans = {
# Treatment plan options for skull fracture
'112': [[self.prob_depressed_skull_fracture, 1 - self.prob_depressed_skull_fracture], ['major', 'HWT']],
'113': [[1], ['HWT']],
# Treatment plan for facial fractures
'211': [[1], ['minor']],
'212': [[1], ['minor']],
# Treatment plan for rib fractures
'412': [[1], ['HWT']],
# Treatment plan for flail chest
'414': [[1], ['major']],
# Treatment plan options for foot fractures
'811': [[p['prob_foot_frac_require_cast'], p['prob_foot_frac_require_maj_surg'],
p['prob_foot_frac_require_min_surg'], p['prob_foot_frac_require_amp']],
['cast', 'major', 'minor', 'major']],
# Treatment plan options for lower leg fractures
'812': [[p['prob_tib_fib_frac_require_cast'], p['prob_tib_fib_frac_require_maj_surg'],
p['prob_tib_fib_frac_require_min_surg'], p['prob_tib_fib_frac_require_traction'],
p['prob_tib_fib_frac_require_amp']],
['cast', 'major', 'minor', 'HWT', 'major']],
# Treatment plan options for femur/hip fractures
'813a': [[p['prob_femural_fracture_require_major_surgery'],
p['prob_femural_fracture_require_minor_surgery'], p['prob_femural_fracture_require_cast'],
p['prob_femural_fracture_require_traction'], p['prob_femural_fracture_require_amputation']],
['major', 'minor', 'cast', 'HWT', 'major']],
# Treatment plan options for femur/hip fractures
'813c': [[p['prob_femural_fracture_require_major_surgery'],
p['prob_femural_fracture_require_minor_surgery'], p['prob_femural_fracture_require_cast'],
p['prob_femural_fracture_require_traction'], p['prob_femural_fracture_require_amputation']],
['major', 'minor', 'cast', 'HWT', 'major']],
# Treatment plan options for pelvis fractures
'813b': [[p['prob_pelvis_fracture_traction'], p['prob_pelvis_frac_major_surgery'],
p['prob_pelvis_frac_minor_surgery'], p['prob_pelvis_frac_cast']],
['HWT', 'major', 'minor', 'cast']],
# Treatment plan options for open fractures
'813bo': [[1], ['open']],
'813co': [[1], ['open']],
'813do': [[1], ['open']],
'813eo': [[1], ['open']],
# Treatment plan options for traumatic brain injuries
'133a': [[self.prob_TBI_require_craniotomy, 1 - self.prob_TBI_require_craniotomy], ['major', 'HWT']],
'133b': [[self.prob_TBI_require_craniotomy, 1 - self.prob_TBI_require_craniotomy], ['major', 'HWT']],
'133c': [[self.prob_TBI_require_craniotomy, 1 - self.prob_TBI_require_craniotomy], ['major', 'HWT']],
'133d': [[self.prob_TBI_require_craniotomy, 1 - self.prob_TBI_require_craniotomy], ['major', 'HWT']],
'134a': [[self.prob_TBI_require_craniotomy, 1 - self.prob_TBI_require_craniotomy], ['major', 'HWT']],
'134b': [[self.prob_TBI_require_craniotomy, 1 - self.prob_TBI_require_craniotomy], ['major', 'HWT']],
'135': [[self.prob_TBI_require_craniotomy, 1 - self.prob_TBI_require_craniotomy], ['major', 'HWT']],
# Treatment plan options for abdominal injuries
'552': [[self.prob_exploratory_laparotomy, 1 - self.prob_exploratory_laparotomy], ['major', 'HWT']],
'553': [[self.prob_exploratory_laparotomy, 1 - self.prob_exploratory_laparotomy], ['major', 'HWT']],
'554': [[self.prob_exploratory_laparotomy, 1 - self.prob_exploratory_laparotomy], ['major', 'HWT']],
# Treatment plan for vertebrae fracture
'612': [[1], ['HWT']],
# Treatment plan for dislocations
'822a': [[p['prob_dis_hip_require_maj_surg'], p['prob_hip_dis_require_traction'],
p['prob_dis_hip_require_cast']], ['major', 'HWT', 'cast']],
'322': [[self.prob_dislocation_requires_surgery, 1 - self.prob_dislocation_requires_surgery],
['minor', 'HWT']],
'323': [[self.prob_dislocation_requires_surgery, 1 - self.prob_dislocation_requires_surgery],
['minor', 'HWT']],
'722': [[self.prob_dislocation_requires_surgery, 1 - self.prob_dislocation_requires_surgery],
['minor', 'HWT']],
# Soft tissue injury in neck treatment plan
'342': [[1], ['major']],
'343': [[1], ['major']],
# Treatment plan for surgical emphysema
'442': [[1], ['HWT']],
# Treatment plan for internal bleeding
'361': [[1], ['major']],
'363': [[1], ['major']],
'461': [[1], ['HWT']],
# Treatment plan for amputations
'782a': [[1], ['major']],
'782b': [[1], ['major']],
'782c': [[1], ['major']],
'783': [[1], ['major']],
'882': [[1], ['major']],
'883': [[1], ['major']],
'884': [[1], ['major']],
# Treatment plan for eye injury
'291': [[1], ['minor']],
# Treatment plan for soft tissue injury
'241': [[1], ['minor']],
# treatment plan for simple fractures and dislocations
'712a': [[1], ['cast']],
'712b': [[1], ['cast']],
'712c': [[1], ['cast']],
'822b': [[1], ['cast']]
}
# store number of open fractures for use later
open_fractures = 0
# check if they have an injury for which we need to find the treatment plan for
for code in treatment_plans.keys():
_, counts = road_traffic_injuries.rti_find_and_count_injuries(person_injuries, [code])
if counts > 0:
treatment_choice = self.module.rng.choice(treatment_plans[code][1], p=treatment_plans[code][0])
if treatment_choice == 'cast':
df.loc[person_id, 'rt_injuries_to_cast'].append(code)
if treatment_choice == 'major':
df.loc[person_id, 'rt_injuries_for_major_surgery'].append(code)
major_surgery_counts += 1
if treatment_choice == 'minor':
df.loc[person_id, 'rt_injuries_for_minor_surgery'].append(code)
minor_surgery_counts += 1
if treatment_choice == 'HWT':
df.loc[person_id, 'rt_injuries_to_heal_with_time'].append(code)
if treatment_choice == 'open':
open_fractures += 1
df.loc[person_id, 'rt_injuries_for_open_fracture_treatment'].append(code)
# -------------------------------- Spinal cord injury requirements --------------------------------------------
# Check whether they have a spinal cord injury, if we allow spinal cord surgery capacilities here, ask for a
# surgery, otherwise make the injury permanent
codes = ['673', '673a', '673b', '674', '674a', '674b', '675', '675a', '675b', '676']
# Ask if this person has a spinal cord injury
_, counts = road_traffic_injuries.rti_find_and_count_injuries(person_injuries, codes)
if (counts > 0) & ('include_spine_surgery' in self.allowed_interventions):
# if this person has a spinal cord injury and we allow surgeries, determine their exact injury
actual_injury = np.intersect1d(codes, person_injuries.values)
# update the number of major surgeries
major_surgery_counts += 1
# add the injury to the injuries to be treated by major surgery
df.loc[person_id, 'rt_injuries_for_major_surgery'].append(actual_injury[0])
elif counts > 0:
# if no surgery assume that the person will be permanently disabled
df.at[person_id, 'rt_perm_disability'] = True
# Find the column and code where the permanent injury is stored
column, code = road_traffic_injuries.rti_find_injury_column(person_id=person_id, codes=codes)
# make the injury permanent by adding a 'P' before the code
df.loc[person_id, column] = "P" + code
code = df.loc[person_id, column]
# find which property the injury is stored in
columns, codes = road_traffic_injuries.rti_find_all_columns_of_treated_injuries(person_id, [code])
for col in columns:
# schedule the recovery date for the permanent injury for beyond the end of the simulation (making
# it permanent)
df.loc[person_id, RTI.INJURY_DATE_COLUMN_MAP[col]] = (
self.sim.end_date + DateOffset(days=1)
)
assert df.loc[person_id, RTI.INJURY_DATE_COLUMN_MAP[col]] > self.sim.date
# --------------------------------- Soft tissue injury in thorax/ lung injury ----------------------------------
# Check whether they have any soft tissue injuries in the thorax, if so schedule surgery if required else make
# the injuries heal over time without further medical care
codes = ['441', '443', '453', '453a', '453b']
# check if they have chest traume
_, counts = road_traffic_injuries.rti_find_and_count_injuries(person_injuries, codes)
if (counts > 0) & ('include_thoroscopy' in self.allowed_interventions):
# work out the exact injury they have
actual_injury = np.intersect1d(codes, person_injuries.values)
# update the number of major surgeries required
major_surgery_counts += 1
# add the injury to the injuries to be treated with major surgery so they aren't treated elsewhere
df.loc[person_id, 'rt_injuries_for_major_surgery'].append(actual_injury[0])
# -------------------------------- Internal bleeding -----------------------------------------------------------
# check if they have internal bleeding in the thorax, and if the surgery is available, schedule a major surgery
codes = ['463']
_, counts = road_traffic_injuries.rti_find_and_count_injuries(person_injuries, codes)
if (counts > 0) & ('include_thoroscopy' in self.allowed_interventions):
# update the number of major surgeries needed
major_surgery_counts += 1
# add the injury to the injuries to be treated with major surgery.
df.loc[person_id, 'rt_injuries_for_major_surgery'].append('463')
# ================ Determine how long the person will be in hospital based on their ISS score ==================
inpatient_days = road_traffic_injuries.rti_determine_LOS(person_id)
# If the patient needs skeletal traction for their injuries they need to stay at minimum 6 weeks,
# average length of stay for those with femur skeletal traction found from Kramer et al. 2016:
# https://doi.org/10.1007/s00264-015-3081-3
# todo: put in complications from femur fractures
femur_fracture_skeletal_traction_mean_los = p['femur_fracture_skeletal_traction_mean_los']
other_skeletal_traction_los = p['other_skeletal_traction_los']
min_los_for_traction = {
'813c': femur_fracture_skeletal_traction_mean_los,
'813b': other_skeletal_traction_los,
'813a': other_skeletal_traction_los,
'812': other_skeletal_traction_los,
}
traction_injuries = [injury for injury in df.loc[person_id, 'rt_injuries_to_heal_with_time'] if injury in
min_los_for_traction.keys()]
if len(traction_injuries) > 0:
if inpatient_days < min_los_for_traction[traction_injuries[0]]:
inpatient_days = min_los_for_traction[traction_injuries[0]]
# Specify the type of bed days needed? not sure if necessary
self.BEDDAYS_FOOTPRINT.update({'general_bed': inpatient_days})
# update the expected appointment foortprint
if inpatient_days > 0:
self.EXPECTED_APPT_FOOTPRINT.update({'InpatientDays': inpatient_days})
# ================ Determine whether the person will require ICU days =========================================
# Percentage of RTIs that required ICU stay 2.7% at KCH : https://doi.org/10.1007/s00268-020-05853-z
# Percentage of RTIs that require HDU stay 3.3% at KCH
# Assume for now that ICU admission is entirely dependent on injury severity so that only the 2.7% of most
# severe injuries get admitted to ICU and the following 3.3% of most severe injuries get admitted to HDU
# NOTE: LEAVING INPATIENT DAYS IN PLACE TEMPORARILY
# Seems only one level of care above normal so adjust accordingly
# self.icu_cut_off_iss_score = 38
self.hdu_cut_off_iss_score = p['hdu_cut_off_iss_score']
# Malawi ICU data: doi: 10.1177/0003134820950282
# General length of stay from Malawi source, not specifically for injuries though
# mean = 4.8, s.d. = 6, TBI admission mean = 8.4, s.d. = 6.4
# mortality percentage = 51.2 overall, 50% for TBI admission and 49% for hemorrhage
# determine the number of ICU days used to treat patient
if df.loc[person_id, 'rt_ISS_score'] > self.hdu_cut_off_iss_score:
self.add_equipment(self.healthcare_system.equipment.from_pkg_names('ICU'))
mean_icu_days = p['mean_icu_days']
sd_icu_days = p['sd_icu_days']
mean_tbi_icu_days = p['mean_tbi_icu_days']
sd_tbi_icu_days = p['sd_tbi_icu_days']
codes = ['133', '133a', '133b', '133c', '133d' '134', '134a', '134b', '135']
_, counts = road_traffic_injuries.rti_find_and_count_injuries(person_injuries, codes)
if counts > 0:
self.icu_days = int(self.module.rng.normal(mean_tbi_icu_days, sd_tbi_icu_days, 1))
else:
self.icu_days = int(self.module.rng.normal(mean_icu_days, sd_icu_days, 1))
# if the number of ICU days is less than zero make it zero
if self.icu_days < 0:
self.icu_days = 0
# update the property showing if a person is in ICU
df.loc[person_id, 'rt_in_icu_or_hdu'] = True
# update the bed days footprint
self.BEDDAYS_FOOTPRINT.update({'general_bed': self.icu_days})
# store the injury information of patients in ICU
logger.info(key='ICU_patients',
data=person_injuries,
description='The injuries of ICU patients')
# Check that each injury has only one treatment plan assigned to it
treatment_plan = \
person['rt_injuries_for_minor_surgery'] + person['rt_injuries_for_major_surgery'] + \
person['rt_injuries_to_heal_with_time'] + person['rt_injuries_for_open_fracture_treatment'] + \
person['rt_injuries_to_cast']
assert len(treatment_plan) == len(set(treatment_plan))
# Other test admission protocol. Basing ICU admission of whether they have a TBI
# 17.3% of head injury patients in KCH were admitted to ICU/HDU (7.9 and 9.4% respectively)
# Injury characteristics of patients admitted to ICU in Tanzania:
# 97.8% had lacerations
# 32.4% had fractures
# 21.5% had TBI
# 13.1% had abdominal injuries
# 2.9% had burns
# 3.8% had 'other' injuries
# https://doi.org/10.1186/1757-7241-19-61
if not df.at[person_id, 'is_alive']:
return self.make_appt_footprint({})
# Remove the scheduled death without medical intervention
df.loc[person_id, 'rt_date_death_no_med'] = pd.NaT
# Isolate relevant injury information
person = df.loc[person_id]
person_injuries = df.loc[[person_id], RTI.INJURY_COLUMNS]
non_empty_injuries = person_injuries[person_injuries != "none"]
non_empty_injuries = non_empty_injuries.dropna(axis=1)
# Check that those who arrive here are alive and have been through the first generic appointment, and didn't
# die due to rti
assert person['rt_diagnosed'], 'person sent here has not been through A and E'
# Check that those who arrive here have at least one injury
_, counts = RTI.rti_find_and_count_injuries(person_injuries,
self.module.PROPERTIES.get('rt_injury_1').categories[1:-1])
if counts == 0:
logger.debug(key='rti_general_message',
data=f"This is RTIMedicalInterventionEvent person {person_id} asked for treatment but doesn't"
f"need it.")
return self.make_appt_footprint({})
# log the number of injuries this person has
logger.info(key='number_of_injuries_in_hospital',
data={'number_of_injuries': counts},
description='The number of injuries of people in the healthsystem')
# update the model's properties to reflect that this person has sought medical care
df.at[person_id, 'rt_med_int'] = True
# =============================== Make 'healed with time' injuries disappear ===================================
# these are the injuries actually treated in this HSI
heal_with_time_recovery_times_in_days = {
# using estimated 6 weeks PLACEHOLDER FOR neck dislocations
'322': 42,
'323': 42,
# using estimated 12 weeks placeholder for dislocated shoulders
'722': 84,
# using estimated 2 month placeholder for dislocated knees
'822a': 60,
# using estimated 7 weeks PLACEHOLDER FOR SKULL FRACTURE
'112': 49,
'113': 49,
# using estimated 5 weeks PLACEHOLDER FOR rib FRACTURE
'412': 35,
# using estimated 9 weeks PLACEHOLDER FOR Vertebrae FRACTURE
'612': 63,
# using estimated 9 weeks PLACEHOLDER FOR skeletal traction for tibia/fib
'812': 63,
# using estimated 9 weeks PLACEHOLDER FOR skeletal traction for hip
'813a': 63,
# using estimated 9 weeks PLACEHOLDER FOR skeletal traction for pelvis
'813b': 63,
# using estimated 9 weeks PLACEHOLDER FOR skeletal traction for femur
'813c': 63,
# using estimated 3 month PLACEHOLDER FOR abdominal trauma
'552': 90,
'553': 90,
'554': 90,
# using 1 week placeholder for surgical emphysema
'442': 7,
# 2 week placeholder for chest wall bruising
'461': 14
}
tbi = ['133', '133a', '133b', '133c', '133d', '134', '134a', '134b', '135']
if len(df.at[person_id, 'rt_injuries_to_heal_with_time']) > 0:
# check whether the heal with time injuries include dislocations, which may have been sent to surgery
for code in person['rt_injuries_to_heal_with_time']:
# temporarily dealing with TBI heal dates seporately
if code in tbi:
pass
else:
injury_column, _ = road_traffic_injuries.rti_find_injury_column(
person_id, [code]
)
date_to_remove_daly_column = RTI.INJURY_DATE_COLUMN_MAP[injury_column]
df.loc[person_id, date_to_remove_daly_column] = (
self.sim.date
+ DateOffset(days=heal_with_time_recovery_times_in_days[code])
)
assert df.loc[person_id, date_to_remove_daly_column] > self.sim.date
heal_with_time_codes = []
# Check whether the heal with time injury is a skull fracture, which may have been sent to surgery
tbi = ['133', '133a', '133b', '133c', '133d', '134', '134a', '134b', '135']
tbi_injury = [injury for injury in tbi if injury in person['rt_injuries_to_heal_with_time']]
if len(tbi_injury) > 0:
injury_column, code = road_traffic_injuries.rti_find_injury_column(
person_id=person_id, codes=tbi_injury
)
date_to_remove_daly_column = RTI.INJURY_DATE_COLUMN_MAP[injury_column]
# ask if this injury will be permanent
perm_injury = self.module.rng.random_sample(size=1)
if perm_injury < self.prob_perm_disability_with_treatment_severe_TBI:
# injury is permanent
# put a P in front of the code to show it will be a perm injury
df.loc[person_id, injury_column] = "P" + code
# store the heal with time injury in heal_with_time_codes
heal_with_time_codes.append("P" + code)
# update the property 'rt_injuries_to_heal_with_time' to contain the new code
df.loc[person_id, 'rt_injuries_to_heal_with_time'].remove(code)
df.loc[person_id, 'rt_injuries_to_heal_with_time'].append("P" + code)
# schedule a recover date beyond this simulation's end
df.loc[person_id, date_to_remove_daly_column] = self.sim.end_date + DateOffset(days=1)
assert df.loc[person_id, date_to_remove_daly_column] > self.sim.date
else:
heal_with_time_codes.append(tbi_injury[0])
# using estimated 6 months PLACEHOLDER FOR TRAUMATIC BRAIN INJURY
df.loc[person_id, date_to_remove_daly_column] = self.sim.date + DateOffset(months=6)
assert df.loc[person_id, date_to_remove_daly_column] > self.sim.date
# swap potentially swappable codes
swapping_codes = RTI.SWAPPING_CODES[:]
# remove codes that will be treated elsewhere
for code in person['rt_injuries_for_minor_surgery']:
if code in swapping_codes:
swapping_codes.remove(code)
for code in person['rt_injuries_for_major_surgery']:
if code in swapping_codes:
swapping_codes.remove(code)
for code in person['rt_injuries_to_cast']:
if code in swapping_codes:
swapping_codes.remove(code)
for code in person['rt_injuries_for_open_fracture_treatment']:
if code in swapping_codes:
swapping_codes.remove(code)
# drop injuries potentially treated elsewhere
codes_to_swap = [code for code in heal_with_time_codes if code in swapping_codes]
if len(codes_to_swap) > 0:
road_traffic_injuries.rti_swap_injury_daly_upon_treatment(person_id, codes_to_swap)
# check every heal with time injury has a recovery date associated with it
for code in person['rt_injuries_to_heal_with_time']:
injury_column, _ = road_traffic_injuries.rti_find_injury_column(
person_id, [code]
)
date_to_remove_daly_column = RTI.INJURY_DATE_COLUMN_MAP[injury_column]
assert not pd.isnull(df.loc[person_id, date_to_remove_daly_column]), (
'no recovery date given for this injury ' + code
)
# check injury heal time is in the future
assert df.loc[person_id, date_to_remove_daly_column] > self.sim.date
# remove code from heal with time injury list
df.loc[person_id, 'rt_injuries_to_heal_with_time'].clear()
# schedule treatments of all injuries here
# ======================================= Schedule surgeries ==================================================
# Schedule the surgeries by calling the functions rti_do_for_major/minor_surgeries which in turn schedules the
# surgeries, people can have multiple surgeries scheduled so schedule surgeries seperate to the rest of the
# treatment plans
# Check they haven't died from another source
if not pd.isnull(df.loc[person_id, 'cause_of_death']):
pass
else:
if major_surgery_counts > 0:
# schedule major surgeries
for count in range(0, major_surgery_counts):
road_traffic_injuries.rti_do_for_major_surgeries(person_id=person_id, count=count)
if minor_surgery_counts > 0:
# shedule minor surgeries
for count in range(0, minor_surgery_counts):
road_traffic_injuries.rti_do_for_minor_surgeries(person_id=person_id, count=count)
# Schedule all other treatments here
# Fractures are sometimes treated via major/minor surgeries. Need to establish which injuries are due to be
# treated via fracture cast
frac_codes = ['712', '712a', '712b', '712c', '811', '812', '813a', '813b', '813c', '822a', '822b']
p = df.loc[person_id]
codes_treated_elsewhere = \
p['rt_injuries_for_minor_surgery'] + p['rt_injuries_for_major_surgery'] + \
p['rt_injuries_to_heal_with_time'] + p['rt_injuries_for_open_fracture_treatment']
frac_codes = [code for code in frac_codes if code not in codes_treated_elsewhere]
# Create a lookup table for treatment methods and the injuries that they are due to treat
single_option_treatments = {
'suture': ['1101', '2101', '3101', '4101', '5101', '7101', '8101'],
'burn': ['1114', '2114', '3113', '4113', '5113', '7113', '8113'],
'fracture': frac_codes,
'tetanus': ['1101', '2101', '3101', '4101', '5101', '7101', '8101', '1114', '2114', '3113', '4113', '5113',
'7113', '8113'],
'pain': self.module.PROPERTIES.get('rt_injury_1').categories[1:],
'open': ['813bo', '813co', '813do', '813eo']
}
# find this person's untreated injuries
untreated_injury_cols = _get_untreated_injury_columns(person_id, df)
person_untreated_injuries = df.loc[[person_id], untreated_injury_cols]
for treatment in single_option_treatments:
# If a person has an injury that hasn't been deliberately left untreated then schedule a treatment, or if
# the treatment is pain management
untreated_injuries = list(non_empty_injuries.values[0])
deliberately_untreated_injuries = df.loc[person_id, 'rt_injuries_left_untreated']
injuries_left_to_treat = [injury for injury in untreated_injuries if injury not in
deliberately_untreated_injuries]
no_injuries_for_this_treatment = (len(set(injuries_left_to_treat) &
set(single_option_treatments[treatment])) == 0)
condition_to_skip = no_injuries_for_this_treatment & (treatment != 'pain')
if condition_to_skip:
pass
else:
_, inj_counts = road_traffic_injuries.rti_find_and_count_injuries(person_untreated_injuries,
single_option_treatments[treatment])
if inj_counts > 0 & df.loc[person_id, 'is_alive']:
if treatment == 'suture':
road_traffic_injuries.rti_ask_for_suture_kit(person_id=person_id)
if treatment == 'burn':
road_traffic_injuries.rti_ask_for_burn_treatment(person_id=person_id)
if treatment == 'fracture':
road_traffic_injuries.rti_ask_for_fracture_casts(person_id=person_id)
if treatment == 'tetanus':
road_traffic_injuries.rti_ask_for_tetanus(person_id=person_id)
if treatment == 'pain':
road_traffic_injuries.rti_acute_pain_management(person_id=person_id)
if treatment == 'open':
road_traffic_injuries.rti_ask_for_open_fracture_treatment(person_id=person_id,
counts=open_fractures)
treatment_plan = \
p['rt_injuries_for_minor_surgery'] + p['rt_injuries_for_major_surgery'] + \
p['rt_injuries_to_heal_with_time'] + p['rt_injuries_for_open_fracture_treatment'] + \
p['rt_injuries_to_cast']
# make sure injuries are treated in one place only
assert len(treatment_plan) == len(set(treatment_plan))
# ============================== Ask if they die even with treatment ===========================================
self.sim.schedule_event(RTI_Medical_Intervention_Death_Event(self.module, person_id), self.sim.date +
DateOffset(days=inpatient_days))
logger.debug(key='rti_general_message',
data=f"This is RTIMedicalInterventionEvent scheduling a potential death on date "
f"{self.sim.date + DateOffset(days=inpatient_days)} (end of treatment) for person "
f"{person_id}")
[docs]
def did_not_run(self):
person_id = self.target
df = self.sim.population.props
logger.debug(key='rti_general_message',
data=f"RTIMedicalInterventionEvent did not run on date {self.sim.date} (end of treatment) for "
f"person {person_id}")
injurycodes = {'First injury': df.at[person_id, 'rt_injury_1'],
'Second injury': df.at[person_id, 'rt_injury_2'],
'Third injury': df.at[person_id, 'rt_injury_3'],
'Fourth injury': df.at[person_id, 'rt_injury_4'],
'Fifth injury': df.at[person_id, 'rt_injury_5'],
'Sixth injury': df.at[person_id, 'rt_injury_6'],
'Seventh injury': df.at[person_id, 'rt_injury_7'],
'Eight injury': df.at[person_id, 'rt_injury_8']}
logger.debug(key='rti_injury_profile_of_untreated_person', data=injurycodes)
# reset the treatment plan
df.at[person_id, 'rt_injuries_for_major_surgery'] = []
df.at[person_id, 'rt_injuries_for_minor_surgery'] = []
df.at[person_id, 'rt_injuries_to_cast'] = []
df.at[person_id, 'rt_injuries_to_heal_with_time'] = []
df.at[person_id, 'rt_injuries_for_open_fracture_treatment'] = []
[docs]
class HSI_RTI_Shock_Treatment(HSI_Event, IndividualScopeEventMixin):
"""
This HSI event handles the process of treating hypovolemic shock, as recommended by the pediatric
handbook for Malawi and (TODO: FIND ADULT REFERENCE)
"""
[docs]
def __init__(self, module, person_id):
super().__init__(module, person_id=person_id)
assert isinstance(module, RTI)
self.TREATMENT_ID = 'Rti_ShockTreatment'
self.EXPECTED_APPT_FOOTPRINT = self.make_appt_footprint({'AccidentsandEmerg': 1})
self.ACCEPTED_FACILITY_LEVEL = '1b'
self._number_of_times_this_event_has_run = 0
self._maximum_number_times_event_should_run = self.module.parameters['maximum_number_of_times_HSI_events_should_run']
[docs]
def apply(self, person_id, squeeze_factor):
df = self.sim.population.props
self._number_of_times_this_event_has_run += 1
# determine if this is a child
if df.loc[person_id, 'age_years'] < 15:
is_child = True
else:
is_child = False
if not df.at[person_id, 'is_alive']:
return self.make_appt_footprint({})
# TODO: find a more complete list of required consumables for adults
if is_child:
is_cons_available = self.get_consumables(
self.module.cons_item_codes['shock_treatment_child']
)
else:
is_cons_available = self.get_consumables(
self.module.cons_item_codes['shock_treatment_adult']
)
if is_cons_available:
logger.debug(key='rti_general_message',
data=f"Hypovolemic shock treatment available for person {person_id}")
df.at[person_id, 'rt_in_shock'] = False
self.add_equipment({'Infusion pump', 'Drip stand', 'Oxygen cylinder, with regulator', 'Nasal Prongs'})
else:
if self._number_of_times_this_event_has_run < self._maximum_number_times_event_should_run:
self.sim.modules['RTI'].schedule_hsi_event_for_tomorrow(self)
return self.make_appt_footprint({})
[docs]
def did_not_run(self):
# Assume that untreated shock leads to death for now
# Schedule the death
df = self.sim.population.props
person_id = self.target
df.at[person_id, 'rt_death_from_shock'] = True
self.sim.modules['Demography'].do_death(individual_id=person_id, cause="RTI_death_shock",
originating_module=self.module)
# Log the death
logger.debug(key='rti_general_message',
data=f"This is RTI_Shock_Treatment scheduling a death for person {person_id} who did not recieve "
f"treatment for shock on {self.sim.date}"
)
[docs]
class HSI_RTI_Fracture_Cast(HSI_Event, IndividualScopeEventMixin):
"""
This HSI event handles fracture casts/giving slings for those who need it. The HSI event tests whether the injured
person has an appropriate injury code, determines how many fractures the person and then requests fracture
treatment as required.
The injury codes dealt with in this HSI event are:
'712a' - broken clavicle, scapula, humerus
'712b' - broken hand/wrist
'712c' - broken radius/ulna
'811' - Fractured foot
'812' - broken tibia/fibula
'813a' - Broken hip
'813b' - broken pelvis
'813c' - broken femur
'822a' - dislocated hip
'822b' - dislocated knee
The properties altered by this function are
rt_date_to_remove_daly - setting recovery dates for injuries treated with fracture casts
rt_injuries_to_cast - once treated the codes used to denote injuries to be treated by fracture casts are removed
from the list of injuries due to be treated with fracture casts
rt_med_int - the property used to denote whether a person getting treatment for road traffic injuries
"""
[docs]
def __init__(self, module, person_id):
super().__init__(module, person_id=person_id)
assert isinstance(module, RTI)
self.TREATMENT_ID = 'Rti_FractureCast'
self.EXPECTED_APPT_FOOTPRINT = self.make_appt_footprint({'AccidentsandEmerg': 1})
self.ACCEPTED_FACILITY_LEVEL = '1b'
self._number_of_times_this_event_has_run = 0
self._maximum_number_times_event_should_run = self.module.parameters[
'maximum_number_of_times_HSI_events_should_run']
[docs]
def apply(self, person_id, squeeze_factor):
# Get the population and health system
df = self.sim.population.props
p = df.loc[person_id]
self._number_of_times_this_event_has_run += 1
# if the person isn't alive return a blank footprint
if not df.at[person_id, 'is_alive']:
return self.make_appt_footprint({})
# get a shorthand reference to RTI and consumables modules
road_traffic_injuries = self.sim.modules['RTI']
# isolate the relevant injury information
# Find the untreated injuries
untreated_injury_cols = _get_untreated_injury_columns(person_id, df)
person_injuries = df.loc[[person_id], untreated_injury_cols]
# check if they have a fracture that requires a cast
codes = ['712b', '712c', '811', '812', '813a', '813b', '813c', '822a', '822b']
_, fracturecastcounts = road_traffic_injuries.rti_find_and_count_injuries(person_injuries, codes)
# check if they have a fracture that requires a sling
codes = ['712a']
_, slingcounts = road_traffic_injuries.rti_find_and_count_injuries(person_injuries, codes)
# Check the person sent here is alive, been through the generic first appointment,
# been through the RTI med intervention
assert p['rt_diagnosed'], 'person sent here has not been diagnosed'
assert p['rt_med_int'], 'person sent here has not been treated'
# Check that the person sent here has an injury treated by this module
assert fracturecastcounts + slingcounts > 0
# Check this person has an injury intended to be treated here
assert len(p['rt_injuries_to_cast']) > 0
# Check this injury assigned to be treated here is actually had by the person
assert all(injuries in person_injuries.values for injuries in p['rt_injuries_to_cast'])
# If they have a fracture that needs a cast, ask for consumables, updating to match the number of
# fractures).
is_cons_available = self.get_consumables(
self.module.cons_item_codes['fracture_treatment'](fracturecastcounts)
)
# if the consumables are available then the appointment can run
if is_cons_available:
logger.debug(key='rti_general_message',
data=f"Fracture casts available for person %d's {fracturecastcounts + slingcounts} fractures, "
f"{person_id}"
)
self.add_equipment({'Casting platform', 'Casting chairs', 'Bucket, 10L'})
# update the property rt_med_int to indicate they are recieving treatment
df.at[person_id, 'rt_med_int'] = True
# Find the persons injuries
non_empty_injuries = person_injuries[person_injuries != "none"]
non_empty_injuries = non_empty_injuries.dropna(axis=1)
# Find the injury codes treated by fracture casts/slings
codes = ['712a', '712b', '712c', '811', '812', '813a', '813b', '813c', '822a', '822b']
# Some TLO codes have daly weights associated with treated and non-treated injuries, copy the list of
# swapping codes
swapping_codes = RTI.SWAPPING_CODES[:]
# find the relevant swapping codes for this treatment
swapping_codes = [code for code in swapping_codes if code in codes]
# remove codes that will be treated elsewhere
injuries_treated_elsewhere = \
p['rt_injuries_for_minor_surgery'] + p['rt_injuries_for_major_surgery'] + \
p['rt_injuries_to_heal_with_time'] + p['rt_injuries_for_open_fracture_treatment']
# remove codes that are being treated elsewhere
swapping_codes = [code for code in swapping_codes if code not in injuries_treated_elsewhere]
# find any potential codes this person has that are due to be swapped and then swap with
# rti_swap_injury_daly_upon_treatment
relevant_codes = np.intersect1d(non_empty_injuries.values, swapping_codes)
if len(relevant_codes) > 0:
road_traffic_injuries.rti_swap_injury_daly_upon_treatment(person_id, relevant_codes)
# Find the injuries that have been treated and then schedule a recovery date
columns, codes = road_traffic_injuries.rti_find_all_columns_of_treated_injuries(
person_id, df.loc[person_id, 'rt_injuries_to_cast']
)
# check that for each injury to be treated by this event we have a corresponding column
assert len(columns) == len(df.loc[person_id, 'rt_injuries_to_cast'])
# iterate over the columns of injuries treated here and assign a recovery date
for injury_column in columns:
# todo: update this with recovery times for casted broken hips/pelvis/femurs
# todo: update this with recovery times for casted dislocated hip
date_to_remove_daly_column = RTI.INJURY_DATE_COLUMN_MAP[injury_column]
df.loc[person_id, date_to_remove_daly_column] = (
self.sim.date + DateOffset(weeks=7)
)
# make sure the assigned injury recovery date is in the future
assert df.loc[person_id, date_to_remove_daly_column] > self.sim.date
person_injuries = df.loc[person_id, RTI.INJURY_COLUMNS]
for code in df.loc[person_id, 'rt_injuries_to_cast']:
injury_column, _ = road_traffic_injuries.rti_find_injury_column(
person_id, [code]
)
assert not pd.isnull(
df.loc[person_id, RTI.INJURY_DATE_COLUMN_MAP[injury_column]]
), 'no recovery date given for this injury'
# remove codes from fracture cast list
df.loc[person_id, 'rt_injuries_to_cast'].clear()
df.loc[person_id, 'rt_date_death_no_med'] = pd.NaT
else:
if self._number_of_times_this_event_has_run < self._maximum_number_times_event_should_run:
self.sim.modules['RTI'].schedule_hsi_event_for_tomorrow(self)
if pd.isnull(df.loc[person_id, 'rt_date_death_no_med']):
df.loc[person_id, 'rt_date_death_no_med'] = self.sim.date + DateOffset(days=7)
logger.debug(key='rti_general_message',
data=f"Person {person_id} has {fracturecastcounts + slingcounts} fractures without treatment"
)
return self.make_appt_footprint({})
[docs]
def did_not_run(self):
person_id = self.target
logger.debug(key='rti_general_message',
data=f"Fracture casts unavailable for person {person_id}")
[docs]
class HSI_RTI_Open_Fracture_Treatment(HSI_Event, IndividualScopeEventMixin):
"""
This HSI event handles fracture casts/giving slings for those who need it. The HSI event tests whether the injured
person has an appropriate injury code, determines how many fractures the person and then requests fracture
treatment as required.
The injury codes dealt with in this HSI event are:
'813bo' - Open fracture of the pelvis
'813co' - Open fracture of the femur
'813do' - Open fracture of the foot
'813eo' - Open fracture of the tibia/fibula/ankle/patella
The properties altered by this function are:
rt_med_int - to denote that this person is recieving treatment
rt_injuries_for_open_fracture_treatment - removing codes that have been treated by open fracture treatment
rt_date_to_remove_daly - to schedule recovery dates for open fractures that have recieved treatment
"""
[docs]
def __init__(self, module, person_id):
super().__init__(module, person_id=person_id)
assert isinstance(module, RTI)
self.TREATMENT_ID = 'Rti_OpenFractureTreatment'
self.EXPECTED_APPT_FOOTPRINT = self.make_appt_footprint({'MinorSurg': 1})
self.ACCEPTED_FACILITY_LEVEL = '1b'
self._number_of_times_this_event_has_run = 0
self._maximum_number_times_event_should_run = self.module.parameters[
'maximum_number_of_times_HSI_events_should_run']
[docs]
def apply(self, person_id, squeeze_factor):
df = self.sim.population.props
self._number_of_times_this_event_has_run += 1
if not df.at[person_id, 'is_alive']:
return self.make_appt_footprint({})
road_traffic_injuries = self.sim.modules['RTI']
# isolate the relevant injury information
person_injuries = df.loc[[person_id], RTI.INJURY_COLUMNS]
# check if they have a fracture that requires a cast
codes = ['813bo', '813co', '813do', '813eo']
_, open_fracture_counts = road_traffic_injuries.rti_find_and_count_injuries(person_injuries, codes)
assert open_fracture_counts > 0
# Check the person sent here is alive, been through the generic first appointment,
# been through the RTI med intervention
assert df.loc[person_id, 'rt_diagnosed'], 'person sent here has not been diagnosed'
assert df.loc[person_id, 'rt_med_int'], 'person sent here has not been treated'
# If they have an open fracture, ask for consumables to treat fracture
wound_contaminated = (
(open_fracture_counts > 0)
and (self.module.parameters['prob_open_fracture_contaminated'] > self.module.rng.random_sample())
)
# Check that there are enough consumables to treat this person's fractures
is_cons_available = self.get_consumables(self.module.cons_item_codes["open_fracture_treatment"]) and (
# If wound is "grossly contaminated" administer Metronidazole, else ignore
self.get_consumables(self.module.cons_item_codes["open_fracture_treatment_additional_if_contaminated"])
if wound_contaminated else True)
if is_cons_available:
logger.debug(key='rti_general_message',
data=f"Fracture casts available for person {person_id} {open_fracture_counts} open fractures"
)
self.add_equipment(self.healthcare_system.equipment.from_pkg_names('Major Surgery'))
person = df.loc[person_id]
# update the dataframe to show this person is recieving treatment
df.loc[person_id, 'rt_med_int'] = True
# Find the persons injuries to be treated
non_empty_injuries = person['rt_injuries_for_open_fracture_treatment']
columns, code = road_traffic_injuries.rti_find_all_columns_of_treated_injuries(
person_id, non_empty_injuries
)
# Some TLO codes have daly weights associated with treated and non-treated injuries
if code[0] == '813bo':
road_traffic_injuries.rti_swap_injury_daly_upon_treatment(person_id, code[0])
# Schedule a recovery date for the injury
# estimated 6-9 months recovery times for open fractures
date_to_remove_daly_column = RTI.INJURY_DATE_COLUMN_MAP[columns[0]]
df.loc[person_id, date_to_remove_daly_column] = (
self.sim.date + DateOffset(months=7)
)
assert df.loc[person_id, date_to_remove_daly_column] > self.sim.date
assert not pd.isnull(
df.loc[person_id, date_to_remove_daly_column]
), 'no recovery date given for this injury'
df.loc[person_id, 'rt_date_death_no_med'] = pd.NaT
# remove code from open fracture list
if code[0] in df.loc[person_id, 'rt_injuries_for_open_fracture_treatment']:
df.loc[person_id, 'rt_injuries_for_open_fracture_treatment'].remove(code[0])
else:
if self._number_of_times_this_event_has_run < self._maximum_number_times_event_should_run:
self.sim.modules['RTI'].schedule_hsi_event_for_tomorrow(self)
if pd.isnull(df.loc[person_id, 'rt_date_death_no_med']):
df.loc[person_id, 'rt_date_death_no_med'] = self.sim.date + DateOffset(days=7)
logger.debug(key='rti_general_message',
data=f"Person {person_id}'s has {open_fracture_counts} open fractures without treatment",
)
[docs]
def did_not_run(self):
person_id = self.target
logger.debug(key='rti_general_message',
data=f"Open fracture treatment unavailable for person {person_id}")
[docs]
class HSI_RTI_Suture(HSI_Event, IndividualScopeEventMixin):
"""
This HSI event handles lacerations giving suture kits for those who need it. The HSI event tests whether the injured
person has an appropriate injury code, determines how many lacerations the person and then requests suture kits
as required.
The codes dealt with are:
'1101' - Laceration to the head
'2101' - Laceration to the face
'3101' - Laceration to the neck
'4101' - Laceration to the thorax
'5101' - Laceration to the abdomen
'7101' - Laceration to the upper extremity
'8101' - Laceration to the lower extremity
The properties altered by this function are:
rt_med_int - to denote that this person is recieving treatment
rt_date_to_remove_daly - to schedule recovery dates for lacerations treated in this hsi
"""
[docs]
def __init__(self, module, person_id):
super().__init__(module, person_id=person_id)
assert isinstance(module, RTI)
self.TREATMENT_ID = 'Rti_Suture'
self.EXPECTED_APPT_FOOTPRINT = self.make_appt_footprint({
('Under5OPD' if self.sim.population.props.at[person_id, "age_years"] < 5 else 'Over5OPD'): 1})
self.ACCEPTED_FACILITY_LEVEL = '1b'
self._number_of_times_this_event_has_run = 0
self._maximum_number_times_event_should_run = self.module.parameters[
'maximum_number_of_times_HSI_events_should_run']
[docs]
def apply(self, person_id, squeeze_factor):
df = self.sim.population.props
self._number_of_times_this_event_has_run += 1
if not df.at[person_id, 'is_alive']:
return self.make_appt_footprint({})
road_traffic_injuries = self.sim.modules['RTI']
person_injuries = df.loc[[person_id], RTI.INJURY_COLUMNS]
codes = ['1101', '2101', '3101', '4101', '5101', '7101', '8101']
_, lacerationcounts = road_traffic_injuries.rti_find_and_count_injuries(person_injuries, codes)
# Check the person sent here didn't die due to rti, has been through A&E, through Med int
assert df.loc[person_id, 'rt_diagnosed'], 'person sent here has not been through A and E'
assert df.loc[person_id, 'rt_med_int'], 'person sent here has not been treated'
# Check that the person sent here has an injury that is treated by this HSI event
assert lacerationcounts > 0
if lacerationcounts > 0:
# check the number of suture kits required and request them
is_cons_available = self.get_consumables(
self.module.cons_item_codes['laceration_treatment'](lacerationcounts))
# Availability of consumables determines if the intervention is delivered...
if is_cons_available:
logger.debug(key='rti_general_message',
data=f"This facility has open wound treatment available which has been used for person "
f"{person_id}."
)
logger.debug(key='rti_general_message',
data=f"This facility treated their {lacerationcounts} open wounds")
columns, codes = road_traffic_injuries.rti_find_all_columns_of_treated_injuries(person_id, codes)
for injury_column in columns:
# heal time for lacerations is roughly two weeks according to:
# https://www.facs.org/~/media/files/education/patient%20ed/wound_lacerations.ashx#:~:text=of%20
# wound%20and%20your%20general,have%20a%20weakened%20immune%20system.
date_to_remove_daly_column = RTI.INJURY_DATE_COLUMN_MAP[injury_column]
df.loc[person_id, date_to_remove_daly_column] = (
self.sim.date + DateOffset(days=14)
)
assert df.loc[person_id, date_to_remove_daly_column] > self.sim.date
df.loc[person_id, 'rt_date_death_no_med'] = pd.NaT
else:
if self._number_of_times_this_event_has_run < self._maximum_number_times_event_should_run:
self.sim.modules['RTI'].schedule_hsi_event_for_tomorrow(self)
if pd.isnull(df.loc[person_id, 'rt_date_death_no_med']):
df.loc[person_id, 'rt_date_death_no_med'] = self.sim.date + DateOffset(days=7)
logger.debug(key='rti_general_message',
data="This facility has no treatment for open wounds available.")
return self.make_appt_footprint({})
[docs]
def did_not_run(self):
person_id = self.target
logger.debug(key='rti_general_message',
data=f"Suture kits unavailable for person {person_id}")
[docs]
class HSI_RTI_Burn_Management(HSI_Event, IndividualScopeEventMixin):
"""
This HSI event handles burns giving treatment for those who need it. The HSI event tests whether the injured
person has an appropriate injury code, determines how many burns the person and then requests appropriate treatment
as required.
The codes dealt with in this HSI event are:
'1114' - Burns to the head
'2114' - Burns to the face
'3113' - Burns to the neck
'4113' - Burns to the thorax
'5113' - Burns to the abdomen
'7113' - Burns to the upper extremities
'8113' - Burns to the lower extremities
The properties treated by this module are:
rt_med_int - to denote that this person is recieving treatment for their injuries
rt_date_to_remove_daly - to schedule recovery dates for injuries treated here
"""
[docs]
def __init__(self, module, person_id):
super().__init__(module, person_id=person_id)
assert isinstance(module, RTI)
self.TREATMENT_ID = 'Rti_BurnManagement'
self.EXPECTED_APPT_FOOTPRINT = self.make_appt_footprint({'MinorSurg': 1})
self.ACCEPTED_FACILITY_LEVEL = '1b'
self.BEDDAYS_FOOTPRINT = self.make_beddays_footprint({'general_bed': 1})
p = self.module.parameters
self.prob_mild_burns = p['prob_mild_burns']
self._number_of_times_this_event_has_run = 0
self._maximum_number_times_event_should_run = p['maximum_number_of_times_HSI_events_should_run']
[docs]
def apply(self, person_id, squeeze_factor):
df = self.sim.population.props
self._number_of_times_this_event_has_run += 1
if not df.at[person_id, 'is_alive']:
return self.make_appt_footprint({})
road_traffic_injuries = self.sim.modules['RTI']
person_injuries = df.loc[[person_id], RTI.INJURY_COLUMNS]
codes = ['1114', '2114', '3113', '4113', '5113', '7113', '8113']
_, burncounts = road_traffic_injuries.rti_find_and_count_injuries(person_injuries, codes)
# check the person sent here has an injury treated by this module
assert burncounts > 0
# check the person sent here didn't die due to rti, has been through A and E and had RTI_med_int
assert df.loc[person_id, 'rt_diagnosed'], 'this person has not been through a and e'
assert df.loc[person_id, 'rt_med_int'], 'this person has not been treated'
if burncounts > 0:
# Request materials for burn treatment
cons_needed = self.module.cons_item_codes['burn_treatment'](burncounts)
possible_large_TBSA_burn_codes = ['7113', '8113', '4113', '5113']
idx2, bigburncounts = \
road_traffic_injuries.rti_find_and_count_injuries(person_injuries, possible_large_TBSA_burn_codes)
random_for_severe_burn = self.module.rng.random_sample(size=1)
# ======================== If burns severe enough then give IV fluid replacement ===========================
if (burncounts > 1) or ((len(idx2) > 0) & (random_for_severe_burn > self.prob_mild_burns)):
# check if they have multiple burns, which implies a higher burned total body surface area (TBSA) which
# will alter the treatment plan
cons_needed.update(
self.module.cons_item_codes['ringers lactate for multiple burns']
)
is_cons_available = self.get_consumables(cons_needed)
if is_cons_available:
logger.debug(key='rti_general_message',
data=f"This facility has burn treatment available which has been used for person "
f"{person_id}")
logger.debug(key='rti_general_message',
data=f"This facility treated their {burncounts} burns")
df.at[person_id, 'rt_med_int'] = True
person = df.loc[person_id]
injury_column, _ = road_traffic_injuries.rti_find_injury_column(
person_id, codes
)
date_to_remove_daly_column = RTI.INJURY_DATE_COLUMN_MAP[injury_column]
# estimate burns take 4 weeks to heal
df.loc[person_id, date_to_remove_daly_column] = (
self.sim.date + DateOffset(weeks=4)
)
assert df.loc[person_id, date_to_remove_daly_column] > self.sim.date
persons_injuries = df.loc[[person_id], RTI.INJURY_COLUMNS]
non_empty_injuries = persons_injuries[persons_injuries != "none"]
non_empty_injuries = non_empty_injuries.dropna(axis=1)
swapping_codes = RTI.SWAPPING_CODES[:]
swapping_codes = [code for code in swapping_codes if code in codes]
# remove codes that will be treated elsewhere
treatment_plan = (
person['rt_injuries_for_major_surgery'] + person['rt_injuries_for_minor_surgery'] +
person['rt_injuries_for_minor_surgery'] + person['rt_injuries_to_cast'] +
person['rt_injuries_to_heal_with_time'] + person['rt_injuries_for_open_fracture_treatment']
)
swapping_codes = [code for code in swapping_codes if code not in treatment_plan]
relevant_codes = np.intersect1d(non_empty_injuries.values, swapping_codes)
if len(relevant_codes) > 0:
road_traffic_injuries.rti_swap_injury_daly_upon_treatment(person_id, relevant_codes)
assert df.loc[person_id, date_to_remove_daly_column] > self.sim.date, (
'recovery date assigned to past'
)
df.loc[person_id, 'rt_date_death_no_med'] = pd.NaT
else:
if self._number_of_times_this_event_has_run < self._maximum_number_times_event_should_run:
self.sim.modules['RTI'].schedule_hsi_event_for_tomorrow(self)
if pd.isnull(df.loc[person_id, 'rt_date_death_no_med']):
df.loc[person_id, 'rt_date_death_no_med'] = self.sim.date + DateOffset(days=7)
logger.debug(key='rti_general_message',
data="This facility has no treatment for burns available.")
[docs]
def did_not_run(self):
person_id = self.target
logger.debug(key='rti_general_message',
data=f"Burn treatment unavailable for person {person_id}")
[docs]
class HSI_RTI_Tetanus_Vaccine(HSI_Event, IndividualScopeEventMixin):
"""
This HSI event handles tetanus vaccine requests, the idea being that by separating these from the burn and
laceration and burn treatments, those treatments can go ahead without the availability of tetanus stopping the event
"""
[docs]
def __init__(self, module, person_id):
super().__init__(module, person_id=person_id)
assert isinstance(module, RTI)
self.TREATMENT_ID = 'Rti_TetanusVaccine'
self.EXPECTED_APPT_FOOTPRINT = self.make_appt_footprint({'EPI': 1})
self.ACCEPTED_FACILITY_LEVEL = '1b'
self._number_of_times_this_event_has_run = 0
self._maximum_number_times_event_should_run = self.module.parameters[
'maximum_number_of_times_HSI_events_should_run']
[docs]
def apply(self, person_id, squeeze_factor):
df = self.sim.population.props
self._number_of_times_this_event_has_run += 1
if not df.at[person_id, 'is_alive']:
return self.make_appt_footprint({})
person_injuries = df.loc[[person_id], RTI.INJURY_COLUMNS]
# check the person sent here hasn't died due to rti, has been through A and E and had RTI_med_int
assert df.loc[person_id, 'rt_diagnosed'], 'This person has not been through a and e'
assert df.loc[person_id, 'rt_med_int'], 'This person has not been through rti med int'
# check the person sent here has an injury treated by this module
codes_for_tetanus = ['1101', '2101', '3101', '4101', '5101', '7101', '8101',
'1114', '2114', '3113', '4113', '5113', '7113', '8113']
_, counts = RTI.rti_find_and_count_injuries(person_injuries, codes_for_tetanus)
if counts == 0:
logger.debug(key='rti_general_message',
data=f"This is RTI tetanus vaccine person {person_id} asked for treatment but doesn't"
f"need it.")
return self.make_appt_footprint({})
# If they have a laceration/burn ask request the tetanus vaccine
if counts > 0:
is_tetanus_available = self.get_consumables(self.module.cons_item_codes['tetanus_treatment'])
if is_tetanus_available:
logger.debug(key='rti_general_message',
data=f"Tetanus vaccine requested for person {person_id} and given")
else:
if self._number_of_times_this_event_has_run < self._maximum_number_times_event_should_run:
self.sim.modules['RTI'].schedule_hsi_event_for_tomorrow(self)
logger.debug(key='rti_general_message',
data=f"Tetanus vaccine requested for person {person_id}, not given")
return self.make_appt_footprint({})
[docs]
def did_not_run(self):
person_id = self.target
logger.debug(key='rti_general_message',
data=f"Tetanus vaccine unavailable for person {person_id}")
[docs]
class HSI_RTI_Acute_Pain_Management(HSI_Event, IndividualScopeEventMixin):
""" This HSI event handles all requests for pain management here, all injuries will pass through here and the pain
medicine required will be set to manage the level of pain they are experiencing, with mild pain being managed with
paracetamol/NSAIDS, moderate pain being managed with tramadol and severe pain being managed with morphine.
"There is a mismatch between the burden of musculoskeletal pain conditions and appropriate health policy response
and planning internationally that can be addressed with an integrated research and policy agenda."
SEE doi: 10.2105/AJPH.2018.304747
"""
[docs]
def __init__(self, module, person_id):
super().__init__(module, person_id=person_id)
assert isinstance(module, RTI)
self.TREATMENT_ID = 'Rti_AcutePainManagement'
self.EXPECTED_APPT_FOOTPRINT = self.make_appt_footprint({
('Under5OPD' if self.sim.population.props.at[person_id, "age_years"] < 5 else 'Over5OPD'): 1})
self.ACCEPTED_FACILITY_LEVEL = '1b'
self._number_of_times_this_event_has_run = 0
self._maximum_number_times_event_should_run = self.module.parameters[
'maximum_number_of_times_HSI_events_should_run']
[docs]
def apply(self, person_id, squeeze_factor):
df = self.sim.population.props
self._number_of_times_this_event_has_run += 1
if not df.at[person_id, 'is_alive']:
return self.make_appt_footprint({})
# Check that the person sent here is alive, has been through A&E and RTI_Med_int
assert df.loc[person_id, 'rt_diagnosed'], 'This person has not been through a and e'
assert df.loc[person_id, 'rt_med_int'], 'This person has not been through rti med int'
person_injuries = df.loc[[person_id], RTI.INJURY_COLUMNS]
road_traffic_injuries = self.sim.modules['RTI']
pain_level = "none"
# create a dictionary to associate the level of pain to the codes
pain_dict = {
'severe': ['1114', '2114', '3113', '4113', '5113', '7113', '8113', # burns
'P782', 'P782a', 'P782b', 'P782c', 'P783', 'P882', 'P883', 'P884', # amputations
'673', '673a', '673b', '674', '674a', '674b', '675', '675a', '675b', '676',
'P673', 'P673a', 'P673b', 'P674', 'P674a', 'P674b', 'P675', 'P675a', 'P675b', 'P676', # SCI
'552', '553', '554', # abdominal trauma
'463', '453', '453a', '453b', '441', '443' # severe chest trauma
],
'moderate': ['112', '113', '211', '212', '412', '414', '612', '712', '712a', '712b', '712c',
'811', '812', '813', '813a', '813b', '813c', # fractures
'322', '323', '722', '822', '822a', '822b', # dislocations
'342', '343', '361', '363', # neck trauma
'461', # chest wall bruising
'813bo', '813co', '813do', '813eo' # open fractures
],
'mild': ['1101', '2101', '3101', '4101', '5101', '7101', '8101', # lacerations
'241', # Minor soft tissue injuries
'133', '133a', '133b', '133c', '133d', '134', '134a', '134b', '135', # TBI
'P133', 'P133a', 'P133b', 'P133c', 'P133d', 'P134', 'P134a', 'P134b', 'P135', # Perm TBI
'291', # Eye injury
'442'
]
}
# iterate over the dictionary to find the pain level, going from highest pain to lowest pain in a for loop,
# then find the highest level of pain this person has by breaking the for loop
for severity in pain_dict.keys():
_, counts = road_traffic_injuries.rti_find_and_count_injuries(person_injuries, pain_dict[severity])
if counts > 0:
pain_level = severity
break
if pain_level == "mild":
# Multiple options, some are conditional
# Give paracetamol
# Give NSAIDS such as aspirin (unless they are under 16) for soft tissue pain, but not if they are pregnant
dict_to_output = {'person': person_id,
'pain level': pain_level}
logger.info(key='Requested_Pain_Management',
data=dict_to_output,
description='Summary of the pain medicine requested by each person')
if df.loc[person_id, 'age_years'] < 16:
cond = self.get_consumables(
self.module.cons_item_codes['pain_management_mild_under_16']
)
else:
cond1 = self.get_consumables(self.module.cons_item_codes['pain_management_mild_above_16'])
cond2 = self.get_consumables(self.module.cons_item_codes['pain_management_mild_under_16'])
if (cond1 is True) & (cond2 is True):
which = self.module.rng.random_sample(size=1)
if which <= 0.5:
cond = cond1
logger.debug(key='rti_general_message',
data=f"Person {person_id} requested paracetamol for their pain relief")
else:
cond = cond2
logger.debug(key='rti_general_message',
data=f"Person {person_id} requested diclofenac for their pain relief")
elif (cond1 is True) & (cond2 is False):
cond = cond1
logger.debug(key='rti_general_message',
data=f"Person {person_id} requested paracetamol for their pain relief")
elif (cond1 is False) & (cond2 is True):
cond = cond2
logger.debug(key='rti_general_message',
data=f"Person {person_id} requested diclofenac for their pain relief")
else:
which = self.module.rng.random_sample(size=1)
if which <= 0.5:
cond = cond1
logger.debug(key='rti_general_message',
data=f"Person {person_id} requested paracetamol for their pain relief")
else:
cond = cond2
logger.debug(key='rti_general_message',
data=f"Person {person_id} requested diclofenac for their pain relief")
# Availability of consumables determines if the intervention is delivered...
if cond:
logger.debug(key='rti_general_message',
data=f"This facility has pain management available for mild pain which has been used for "
f"person {person_id}.")
dict_to_output = {'person': person_id,
'pain level': pain_level}
logger.info(key='Successful_Pain_Management',
data=dict_to_output,
description='Pain medicine successfully provided to the person')
else:
if self._number_of_times_this_event_has_run < self._maximum_number_times_event_should_run:
self.sim.modules['RTI'].schedule_hsi_event_for_tomorrow(self)
logger.debug(key='rti_general_message',
data=f"This facility has no pain management available for their mild pain, person "
f"{person_id}.")
return self.make_appt_footprint({})
if pain_level == "moderate":
dict_to_output = {'person': person_id,
'pain level': pain_level}
logger.info(key='Requested_Pain_Management',
data=dict_to_output,
description='Summary of the pain medicine requested by each person')
is_cons_available = self.get_consumables(self.module.cons_item_codes['pain_management_moderate'])
logger.debug(key='rti_general_message',
data=f"Person {person_id} has requested tramadol for moderate pain relief")
if is_cons_available:
logger.debug(key='rti_general_message',
data=f"This facility has pain management available for moderate pain which has been used "
f"for person {person_id}.")
dict_to_output = {'person': person_id,
'pain level': pain_level}
logger.info(key='Successful_Pain_Management',
data=dict_to_output,
description='Pain medicine successfully provided to the person')
else:
if self._number_of_times_this_event_has_run < self._maximum_number_times_event_should_run:
self.sim.modules['RTI'].schedule_hsi_event_for_tomorrow(self)
logger.debug(key='rti_general_message',
data=f"This facility has no pain management available for moderate pain for person "
f"{person_id}.")
return self.make_appt_footprint({})
if pain_level == "severe":
dict_to_output = {'person': person_id,
'pain level': pain_level}
logger.info(key='Requested_Pain_Management',
data=dict_to_output,
description='Summary of the pain medicine requested by each person')
# give morphine
is_cons_available = self.get_consumables(
self.module.cons_item_codes['pain_management_severe']
)
logger.debug(key='rti_general_message',
data=f"Person {person_id} has requested morphine for severe pain relief")
if is_cons_available:
logger.debug(key='rti_general_message',
data=f"This facility has pain management available for severe pain which has been used for"
f" person {person_id}")
dict_to_output = {'person': person_id,
'pain level': pain_level}
logger.info(key='Successful_Pain_Management',
data=dict_to_output,
description='Pain medicine successfully provided to the person')
else:
if self._number_of_times_this_event_has_run < self._maximum_number_times_event_should_run:
self.sim.modules['RTI'].schedule_hsi_event_for_tomorrow(self)
logger.debug(key='rti_general_message',
data=f"This facility has no pain management available for severe pain for person "
f"{person_id}.")
return self.make_appt_footprint({})
[docs]
def did_not_run(self):
person_id = self.target
df = self.sim.population.props
logger.debug(key='rti_general_message',
data=f"Pain relief unavailable for person {person_id}")
injurycodes = {'First injury': df.at[person_id, 'rt_injury_1'],
'Second injury': df.at[person_id, 'rt_injury_2'],
'Third injury': df.at[person_id, 'rt_injury_3'],
'Fourth injury': df.at[person_id, 'rt_injury_4'],
'Fifth injury': df.at[person_id, 'rt_injury_5'],
'Sixth injury': df.at[person_id, 'rt_injury_6'],
'Seventh injury': df.at[person_id, 'rt_injury_7'],
'Eight injury': df.at[person_id, 'rt_injury_8']}
logger.debug(key='rti_general_message',
data=f"Injury profile of person {person_id}, {injurycodes}")
[docs]
class HSI_RTI_Major_Surgeries(HSI_Event, IndividualScopeEventMixin):
"""This is a Health System Interaction Event.
An appointment of a person who has experienced a road traffic injury, had their injuries diagnosed through
A and E and requires major surgery.
Major surgeries are defined here as surgeries that include extensive work such as entering a body cavity,
removing an organ or altering the body’s anatomy
The injuries treated in this module are as follows:
FRACTURES:
While district hospitals can provide some
emergency trauma care and surgeries, only central hospitals
are equipped to provide advanced orthopaedic surgery. - Lavy et al. 2007
'112' - Depressed skull fracture - reported use of surgery in Eaton et al. 2017
'811' - fractured foot - reported use of surgery in Chagomerana et al. 2017
'812' - fracture tibia/fibula - reported use of surgery in Chagomerana et al. 2017
'813a' - Fractured hip - reported use of surgery and Lavy et al. 2007
'813b' - Fractured pelvis - reported use of surgery and Lavy et al. 2007
'813c' - Fractured femur - reported use of surgery and Lavy et al. 2007
'414' - Flail chest - https://www.sciencedirect.com/science/article/abs/pii/S0020138303002900
SOFT TISSUE INJURIES:
'342' - Soft tissue injury of the neck
'343' - Soft tissue injury of the neck
Thoroscopy treated injuries:
https://www.ncbi.nlm.nih.gov/nlmcatalog/101549743
Ref from pediatric handbook for Malawi
'441' - Closed pneumothorax
'443' - Open pneumothorax
'463' - Haemothorax
'453a' - Diaphragm rupture
'453b' - Lung contusion
INTERNAL BLEEDING:
'361' - Internal bleeding in neck
'363' - Internal bleeding in neck
TRAUMATIC BRAIN INJURIES THAT REQUIRE A CRANIOTOMOY - reported use of surgery in Eaton et al 2017 and Lavy et
al. 2007
'133a' - Subarachnoid hematoma
'133b' - Brain contusion
'133c' - Intraventricular haemorrhage
'133d' - Subgaleal hematoma
'134a' - Epidural hematoma
'134b' - Subdural hematoma
'135' - diffuse axonal injury
Laparotomy - Recorded in Lavy et al. 2007 and here: https://www.ajol.info/index.php/mmj/article/view/174378
'552' - Injury to Intestine, stomach and colon
'553' - Injury to Spleen, Urinary bladder, Liver, Urethra, Diaphragm
'554' - Injury to kidney
SPINAL CORD LESIONS, REQUIRING LAMINOTOMY/FORAMINOTOMY/INTERSPINOUS PROCESS SPACER
Quote from Eaton et al. 2019:
"No patients received thoracolumbar braces or underwent spinal surgery."
https://journals.sagepub.com/doi/pdf/10.1177/0049475518808969
So those with spinal cord injuries are not likely to be treated here in RTI_Major_Surgeries..
'673a' - Spinal cord lesion at neck level
'673b' - Spinal cord lesion below neck level
'674a' - Spinal cord lesion at neck level
'674b' - Spinal cord lesion below neck level
'675a' - Spinal cord lesion at neck level
'675b' - Spinal cord lesion below neck level
'676' - Spinal cord lesion at neck level
AMPUTATIONS - Reported in Crudziak et al. 2019
'782a' - Amputated finger
'782b' - Unilateral arm amputation
'782c' - Amputated thumb
'783' - Bilateral arm amputation
'882' - Amputated toe
'883' - Unilateral lower limb amputation
'884' - Bilateral lower limb amputation
Dislocations - Reported in Chagomerana et al. 2017
'822a' Hip dislocation
The properties altered in this function are:
rt_injury_1 through rt_injury_8 - in the incidence that despite treatment the person treated is left
permanently disabled we need to update the injury code to inform the
model that the disability burden associated with the permanently
disabling injury shouldn't be removed
rt_perm_disability - when a person is decided to be permanently disabled we update this property to reflect this
rt_date_to_remove_daly - assign recovery dates for the injuries treated with the surgery
rt_injuries_for_major_surgery - to remove codes due to be treated by major surgery when that injury recieves
a treatment.
"""
[docs]
def __init__(self, module, person_id):
super().__init__(module, person_id=person_id)
assert isinstance(module, RTI)
self.TREATMENT_ID = 'Rti_MajorSurgeries'
self.EXPECTED_APPT_FOOTPRINT = self.make_appt_footprint({'MajorSurg': 1})
self.ACCEPTED_FACILITY_LEVEL = '1b'
self.BEDDAYS_FOOTPRINT = self.make_beddays_footprint({})
self._number_of_times_this_event_has_run = 0
self._maximum_number_times_event_should_run = self.module.parameters['maximum_number_of_times_HSI_events_should_run']
p = self.module.parameters
self.prob_perm_disability_with_treatment_severe_TBI = p['prob_perm_disability_with_treatment_severe_TBI']
self.allowed_interventions = p['allowed_interventions']
self.treated_code = 'none'
[docs]
def apply(self, person_id, squeeze_factor):
self._number_of_times_this_event_has_run += 1
df = self.sim.population.props
rng = self.module.rng
road_traffic_injuries = self.sim.modules['RTI']
# Request first draft of consumables used in major surgery
request_outcome = self.get_consumables(
self.module.cons_item_codes['major_surgery']
)
if not df.at[person_id, 'is_alive']:
return self.make_appt_footprint({})
# todo: think about consequences of certain consumables not being available for major surgery and model health
# outcomes
# Isolate the relevant injury information
surgically_treated_codes = ['112', '811', '812', '813a', '813b', '813c', '133a', '133b', '133c', '133d', '134a',
'134b', '135', '552', '553', '554', '342', '343', '414', '361', '363', '782',
'782a', '782b', '782c', '783', '822a', '882', '883', '884', 'P133a', 'P133b',
'P133c', 'P133d', 'P134a', 'P134b', 'P135', 'P782a', 'P782b', 'P782c', 'P783',
'P882', 'P883', 'P884']
# If we have allowed spinal cord surgeries to be treated in this simulation, include the associated injury
# codes here
if 'include_spine_surgery' in self.allowed_interventions:
additional_codes = ['673a', '673b', '674a', '674b', '675a', '675b', '676', 'P673a', 'P673b', 'P674',
'P674a', 'P674b', 'P675', 'P675a', 'P675b', 'P676']
for code in additional_codes:
surgically_treated_codes.append(code)
# If we have allowed greater access to thoroscopy, include the codes treated by thoroscopy here
if 'include_thoroscopy' in self.allowed_interventions:
additional_codes = ['441', '443', '453', '453a', '453b', '463']
for code in additional_codes:
surgically_treated_codes.append(code)
persons_injuries = df.loc[[person_id], RTI.INJURY_COLUMNS]
injuries_to_be_treated = df.loc[person_id, 'rt_injuries_for_major_surgery']
assert len(set(injuries_to_be_treated) & set(surgically_treated_codes)) > 0, \
'This person has asked for surgery but does not have an appropriate injury'
# check the people sent here have at least one injury treated by this HSI event
_, counts = road_traffic_injuries.rti_find_and_count_injuries(persons_injuries, surgically_treated_codes)
if counts == 0:
logger.debug(key='rti_general_message',
data=f"This is RTI major surgery person {person_id} asked for treatment but doesn't"
f"need it.")
return self.make_appt_footprint({})
# People can be sent here for multiple surgeries, but only one injury can be treated at a time. Decide which
# injury is being treated in this surgery
# find untreated injury codes that are treated with major surgery
relevant_codes = np.intersect1d(injuries_to_be_treated, surgically_treated_codes)
# check that the person sent here has an appropriate code(s)
assert len(relevant_codes) > 0
# choose a code at random
self.treated_code = rng.choice(relevant_codes)
if request_outcome:
# check the people sent here hasn't died due to rti, have had their injuries diagnosed and been through
# RTI_Med
assert df.loc[person_id, 'rt_diagnosed'], 'This person has not been through a and e'
assert df.loc[person_id, 'rt_med_int'], 'This person has not been through rti med int'
self.add_equipment(self.healthcare_system.equipment.from_pkg_names('Major Surgery'))
# ------------------------ Track permanent disabilities with treatment -------------------------------------
# --------------------------------- Perm disability from TBI -----------------------------------------------
codes = ['133', '133a', '133b', '133c', '133d', '134', '134a', '134b', '135']
""" Of patients that survived, 80.1% (n 148) had a good recovery with no appreciable clinical neurologic
deficits, 13.1% (n 24) had a moderate disability with deficits that still allowed the patient to live
independently, 4.9% (n 9) had severe disability which will require assistance with activities of daily life,
and 1.1% (n 2) were in a vegetative state
"""
# Check whether the person having treatment for their tbi will be left permanently disabled
if self.treated_code in codes:
prob_perm_disability = self.module.rng.random_sample(size=1)
if prob_perm_disability < self.prob_perm_disability_with_treatment_severe_TBI:
# Track whether they are permanently disabled
df.at[person_id, 'rt_perm_disability'] = True
# Find the column and code where the permanent injury is stored
column, code = road_traffic_injuries.rti_find_injury_column(person_id=person_id, codes=codes)
logger.debug(key='rti_general_message',
data=f"@@@@@@@@@@ Person {person_id} had intervention for TBI on {self.sim.date} but "
f"still disabled!!!!!!")
# Update the code to make the injury permanent, so it will not have the associated daly weight
# removed later on
code_to_drop_index = injuries_to_be_treated.index(self.treated_code)
injuries_to_be_treated.pop(code_to_drop_index)
# remove the old code from rt_injuries_for_major_surgery
self.treated_code = "P" + self.treated_code
df.loc[person_id, column] = self.treated_code
# include the new code in rt_injuries_for_major_surgery
df.loc[person_id, 'rt_injuries_for_major_surgery'].append(self.treated_code)
assert len(injuries_to_be_treated) == len(df.loc[person_id, 'rt_injuries_for_major_surgery'])
columns, codes = road_traffic_injuries.rti_find_all_columns_of_treated_injuries(person_id,
[self.treated_code])
# schedule the recovery date for the permanent injury for beyond the end of the simulation (making
# it permanent)
date_to_remove_daly_column = RTI.INJURY_DATE_COLUMN_MAP[columns[0]]
df.loc[person_id, date_to_remove_daly_column] = (
self.sim.end_date + DateOffset(days=1)
)
assert df.loc[person_id, date_to_remove_daly_column] > self.sim.date
# ------------------------------------- Perm disability from SCI -------------------------------------------
if 'include_spine_surgery' in self.allowed_interventions:
codes = ['673', '673a', '673b', '674', '674a', '674b', '675', '675a', '675b', '676']
if self.treated_code in codes:
# Track whether they are permanently disabled
df.at[person_id, 'rt_perm_disability'] = True
# Find the column and code where the permanent injury is stored
column, code = road_traffic_injuries.rti_find_injury_column(person_id=person_id,
codes=[self.treated_code])
logger.debug(key='rti_general_message',
data=f"@@@@@@@@@@ Person {person_id} had intervention for SCI on {self.sim.date} but "
f"still disabled!!!!!!")
code_to_drop_index = injuries_to_be_treated.index(self.treated_code)
injuries_to_be_treated.pop(code_to_drop_index)
# remove the code from 'rt_injuries_for_major_surgery'
df.loc[person_id, 'rt_injuries_for_major_surgery'].remove(self.treated_code)
self.treated_code = "P" + self.treated_code
# update the code for 'rt_injuries_for_major_surgery'
df.loc[person_id, 'rt_injuries_for_major_surgery'].append(self.treated_code)
df.loc[person_id, column] = self.treated_code
for injury in injuries_to_be_treated:
if injury not in df.loc[person_id, 'rt_injuries_for_major_surgery']:
df.loc[person_id, 'rt_injuries_for_major_surgery'].append(injury)
assert len(injuries_to_be_treated) == len(df.loc[person_id, 'rt_injuries_for_major_surgery'])
columns, codes = road_traffic_injuries.rti_find_all_columns_of_treated_injuries(person_id,
[self.treated_code])
# schedule the recovery date for the permanent injury for beyond the end of the simulation (making
# it permanent)
date_to_remove_daly_column = RTI.INJURY_DATE_COLUMN_MAP[columns[0]]
df.loc[person_id, date_to_remove_daly_column] = (
self.sim.end_date + DateOffset(days=1)
)
assert df.loc[person_id, date_to_remove_daly_column] > self.sim.date
# ------------------------------------- Perm disability from amputation ------------------------------------
codes = ['782', '782a', '782b', '782c', '783', '882', '883', '884']
if self.treated_code in codes:
# Track whether they are permanently disabled
df.at[person_id, 'rt_perm_disability'] = True
# Find the column and code where the permanent injury is stored
column, code = road_traffic_injuries.rti_find_injury_column(person_id=person_id,
codes=[self.treated_code])
logger.debug(key='rti_general_message',
data=f"@@@@@@@@@@ Person {person_id} had intervention for an amputation on {self.sim.date}"
f" but still disabled!!!!!!")
# Update the code to make the injury permanent, so it will not have the associated daly weight removed
# later on
code_to_drop_index = injuries_to_be_treated.index(self.treated_code)
injuries_to_be_treated.pop(code_to_drop_index)
# remove the old code from rt_injuries_for_major_surgery
self.treated_code = "P" + self.treated_code
# add the new code to rt_injuries_for_major_surgery
df.loc[person_id, 'rt_injuries_for_major_surgery'].append(self.treated_code)
df.loc[person_id, column] = self.treated_code
for injury in injuries_to_be_treated:
if injury not in df.loc[person_id, 'rt_injuries_for_major_surgery']:
df.loc[person_id, 'rt_injuries_for_major_surgery'].append(injury)
assert len(injuries_to_be_treated) == len(df.loc[person_id, 'rt_injuries_for_major_surgery'])
columns, codes = road_traffic_injuries.rti_find_all_columns_of_treated_injuries(person_id,
[self.treated_code])
# Schedule recovery for the end of the simulation, thereby making the injury permanent
date_to_remove_daly_column = RTI.INJURY_DATE_COLUMN_MAP[columns[0]]
df.loc[person_id, date_to_remove_daly_column] = (
self.sim.end_date + DateOffset(days=1)
)
assert df.loc[person_id, date_to_remove_daly_column] > self.sim.date
# ============================== Schedule the recovery dates for the non-permanent injuries ================
maj_surg_recovery_time_in_days = {
'112': 42,
'552': 90,
'553': 90,
'554': 90,
'822a': 270,
'811': 63,
'812': 63,
'813a': 270,
'813b': 70,
'813c': 120,
'133a': 42,
'133b': 42,
'133c': 42,
'133d': 42,
'134a': 42,
'134b': 42,
'135': 42,
'342': 42,
'343': 42,
'414': 365,
'441': 14,
'443': 14,
'453a': 42,
'453b': 42,
'361': 7,
'363': 7,
'463': 7,
}
# find the column of the treated injury
if self.treated_code in maj_surg_recovery_time_in_days.keys():
injury_column, _ = road_traffic_injuries.rti_find_injury_column(
person_id, [self.treated_code]
)
date_to_remove_daly_column = RTI.INJURY_DATE_COLUMN_MAP[injury_column]
if pd.isnull(df.loc[person_id, date_to_remove_daly_column]):
df.loc[person_id, date_to_remove_daly_column] = (
self.sim.date
+ DateOffset(
days=maj_surg_recovery_time_in_days[self.treated_code]
)
)
assert df.loc[person_id, date_to_remove_daly_column] > self.sim.date
# some injuries have a daly weight that swaps upon treatment, get list of those codes
swapping_codes = RTI.SWAPPING_CODES[:]
# isolate that swapping codes that will be treated here
swapping_codes = [code for code in swapping_codes if code in surgically_treated_codes]
# find the injuries this person will have treated in other forms of treatment
person = df.loc[person_id]
treatment_plan = (
person['rt_injuries_for_minor_surgery'] + person['rt_injuries_to_cast'] +
person['rt_injuries_to_heal_with_time'] + person['rt_injuries_for_open_fracture_treatment']
)
# remove codes that will be treated elsewhere
swapping_codes = [code for code in swapping_codes if code not in treatment_plan]
# swap the daly weight for any applicable injuries
if self.treated_code in swapping_codes:
road_traffic_injuries.rti_swap_injury_daly_upon_treatment(person_id, [self.treated_code])
# Check that every injury treated has a recovery time
injury_column, _ = road_traffic_injuries.rti_find_injury_column(
person_id, [self.treated_code]
)
date_to_remove_daly_column = RTI.INJURY_DATE_COLUMN_MAP[injury_column]
assert not pd.isnull(
df.loc[person_id, date_to_remove_daly_column]
), 'no recovery date given for this injury'
assert df.loc[person_id, date_to_remove_daly_column] > self.sim.date
logger.debug(key='rti_general_message',
data=f"This is RTI_Major_Surgeries supplying surgery for person {person_id} on date "
f"{self.sim.date}!!!!!!, removing code")
# remove code from major surgeries list
if self.treated_code in df.loc[person_id, 'rt_injuries_for_major_surgery']:
df.loc[person_id, 'rt_injuries_for_major_surgery'].remove(self.treated_code)
assert self.treated_code not in df.loc[person_id, 'rt_injuries_for_major_surgery'], \
['Treated injury code not removed', self.treated_code]
df.loc[person_id, 'rt_date_death_no_med'] = pd.NaT
else:
if self._number_of_times_this_event_has_run < self._maximum_number_times_event_should_run:
self.sim.modules['RTI'].schedule_hsi_event_for_tomorrow(self)
if pd.isnull(df.loc[person_id, 'rt_date_death_no_med']):
df.loc[person_id, 'rt_date_death_no_med'] = self.sim.date + DateOffset(days=7)
return self.make_appt_footprint({})
[docs]
def did_not_run(self):
person_id = self.target
df = self.sim.population.props
logger.debug(key='rti_general_message',
data=f"Major surgery not scheduled for person {person_id}")
injurycodes = {'First injury': df.at[person_id, 'rt_injury_1'],
'Second injury': df.at[person_id, 'rt_injury_2'],
'Third injury': df.at[person_id, 'rt_injury_3'],
'Fourth injury': df.at[person_id, 'rt_injury_4'],
'Fifth injury': df.at[person_id, 'rt_injury_5'],
'Sixth injury': df.at[person_id, 'rt_injury_6'],
'Seventh injury': df.at[person_id, 'rt_injury_7'],
'Eight injury': df.at[person_id, 'rt_injury_8']}
logger.debug(key='rti_general_message',
data=f"Injury profile of person {person_id}, {injurycodes}")
[docs]
class HSI_RTI_Minor_Surgeries(HSI_Event, IndividualScopeEventMixin):
"""This is a Health System Interaction Event.
An appointment of a person who has experienced a road traffic injury, had their injuries diagnosed through
A and E, treatment plan organised by RTI_MedInt and requires minor surgery.
Minor surgeries are defined here as surgeries are generally superficial and do not require penetration of a
body cavity. They do not involve assisted breathing or anesthesia and are usually performed by a single doctor.
The injuries treated in this module are as follows:
Evidence for all from Mkandawire et al. 2008:
https://link.springer.com/article/10.1007%2Fs11999-008-0366-5
'211' - Facial fractures
'212' - Facial fractures
'291' - Injury to the eye
'241' - Soft tissue injury of the face
'322' - Dislocation in the neck
'323' - Dislocation in the neck
'722' - Dislocated shoulder
External fixation of fractures
'811' - fractured foot
'812' - fractures tibia/fibula
'813a' - Fractured hip
'813b' - Fractured pelvis
'813C' - Fractured femur
The properties altered in this function are:
rt_med_int - update to show this person is being treated for their injuries.
rt_date_to_remove_daly - assign recovery dates for the injuries treated with the surgery
rt_injuries_for_minor_surgery - to remove codes due to be treated by minor surgery when that injury recieves
a treatment.
"""
[docs]
def __init__(self, module, person_id):
super().__init__(module, person_id=person_id)
assert isinstance(module, RTI)
self.TREATMENT_ID = 'Rti_MinorSurgeries'
self.EXPECTED_APPT_FOOTPRINT = self.make_appt_footprint({'MinorSurg': 1})
self.ACCEPTED_FACILITY_LEVEL = '1b'
self._number_of_times_this_event_has_run = 0
self._maximum_number_times_event_should_run = self.module.parameters[
'maximum_number_of_times_HSI_events_should_run']
[docs]
def apply(self, person_id, squeeze_factor):
self._number_of_times_this_event_has_run += 1
df = self.sim.population.props
if not df.at[person_id, 'is_alive']:
return self.make_appt_footprint({})
rng = self.module.rng
road_traffic_injuries = self.sim.modules['RTI']
surgically_treated_codes = ['322', '211', '212', '323', '722', '291', '241', '811', '812', '813a', '813b',
'813c']
persons_injuries = df.loc[[person_id], RTI.INJURY_COLUMNS]
person = df.loc[person_id]
# =========================================== Tests ============================================================
# check the people sent here hasn't died due to rti, have had their injuries diagnosed and been through RTI_Med
assert person['rt_diagnosed'], 'This person has not been through a and e'
assert person['rt_med_int'], 'This person has not been through rti med int'
# check they have at least one injury treated by minor surgery
_, counts = road_traffic_injuries.rti_find_and_count_injuries(persons_injuries, surgically_treated_codes)
if counts == 0:
logger.debug(key='rti_general_message',
data=f"This is RTI minor surgery person {person_id} asked for treatment but doesn't"
f"need it.")
return self.make_appt_footprint({})
# find the injuries which will be treated here
relevant_codes = np.intersect1d(df.loc[person_id, 'rt_injuries_for_minor_surgery'], surgically_treated_codes)
# Check that a code has been selected to be treated
assert len(relevant_codes) > 0
# choose an injury to treat
treated_code = rng.choice(relevant_codes)
# need to determine whether this person has an injury which will treated with external fixation
# external_fixation_codes = ['811', '812', '813a', '813b', '813c']
request_outcome = self.get_consumables(self.module.cons_item_codes['minor_surgery'])
# todo: think about consequences of certain consumables not being available for minor surgery and model health
# outcomes
if request_outcome:
self.add_equipment(self.healthcare_system.equipment.from_pkg_names('Major Surgery'))
# create a dictionary to store the recovery times for each injury in days
minor_surg_recov_time_days = {
'322': 180,
'323': 180,
'722': 49,
'211': 49,
'212': 49,
'291': 7,
'241': 7,
'811': 63,
'812': 63,
'813a': 63,
'813b': 63,
'813c': 63,
}
# assign a recovery time for the treated person from the dictionary, get the column which the injury is
# stored in
injury_column, _ = road_traffic_injuries.rti_find_injury_column(
person_id, [treated_code]
)
# assign a recovery date
df.loc[person_id, RTI.INJURY_DATE_COLUMN_MAP[injury_column]] = (
self.sim.date
+ DateOffset(days=minor_surg_recov_time_days[treated_code])
)
# make sure the injury recovery date is in the future
assert df.loc[person_id, RTI.INJURY_DATE_COLUMN_MAP[injury_column]] > self.sim.date
# some injuries have a change in daly weight if they are treated, find all possible swappable codes
swapping_codes = RTI.SWAPPING_CODES[:]
# exclude any codes that could be swapped but are due to be treated elsewhere
treatment_plan = (
person['rt_injuries_for_minor_surgery'] + person['rt_injuries_to_cast'] +
person['rt_injuries_to_heal_with_time'] + person['rt_injuries_for_open_fracture_treatment']
)
swapping_codes = [code for code in swapping_codes if code not in treatment_plan]
if treated_code in swapping_codes:
road_traffic_injuries.rti_swap_injury_daly_upon_treatment(person_id, [treated_code])
logger.debug(key='rti_general_message',
data=f"This is RTI_Minor_Surgeries supplying minor surgeries for person {person_id} on date "
f"{self.sim.date}!!!!!!")
# update the dataframe to reflect that this person is recieving medical care
df.at[person_id, 'rt_med_int'] = True
# Check if the injury has been given a recovery date
injury_column, _ = road_traffic_injuries.rti_find_injury_column(
person_id, [treated_code]
)
assert not pd.isnull(
df.loc[person_id, RTI.INJURY_DATE_COLUMN_MAP[injury_column]]
), 'no recovery date given for this injury'
# remove code from minor surgeries list as it has now been treated
if treated_code in df.loc[person_id, 'rt_injuries_for_minor_surgery']:
df.loc[person_id, 'rt_injuries_for_minor_surgery'].remove(treated_code)
assert treated_code not in df.loc[person_id, 'rt_injuries_for_minor_surgery'], \
['Injury treated not removed', treated_code]
df.loc[person_id, 'rt_date_death_no_med'] = pd.NaT
else:
if self._number_of_times_this_event_has_run < self._maximum_number_times_event_should_run:
self.sim.modules['RTI'].schedule_hsi_event_for_tomorrow(self)
if pd.isnull(df.loc[person_id, 'rt_date_death_no_med']):
df.loc[person_id, 'rt_date_death_no_med'] = self.sim.date + DateOffset(days=7)
logger.debug(key='rti_general_message',
data=f"This is RTI_Minor_Surgeries failing to provide minor surgeries for person {person_id} "
f"on date {self.sim.date}!!!!!!")
return self.make_appt_footprint({})
[docs]
def did_not_run(self):
person_id = self.target
df = self.sim.population.props
logger.debug(key='rti_general_message',
data=f"Minor surgery not scheduled for person {person_id}")
injurycodes = {'First injury': df.at[person_id, 'rt_injury_1'],
'Second injury': df.at[person_id, 'rt_injury_2'],
'Third injury': df.at[person_id, 'rt_injury_3'],
'Fourth injury': df.at[person_id, 'rt_injury_4'],
'Fifth injury': df.at[person_id, 'rt_injury_5'],
'Sixth injury': df.at[person_id, 'rt_injury_6'],
'Seventh injury': df.at[person_id, 'rt_injury_7'],
'Eight injury': df.at[person_id, 'rt_injury_8']}
logger.debug(key='rti_injury_profile_of_untreated_person',
data=injurycodes)
[docs]
class RTI_Medical_Intervention_Death_Event(Event, IndividualScopeEventMixin):
"""This is the MedicalInterventionDeathEvent. It is scheduled by the MedicalInterventionEvent to occur at the end of
the person's determined length of stay. The risk of mortality for the person wil medical intervention is determined
by the persons ISS score and whether they have polytrauma.
The properties altered by this event are:
rt_post_med_death - updated to reflect when a person dies from their injuries
"""
[docs]
def __init__(self, module, person_id):
super().__init__(module, person_id=person_id)
p = self.module.parameters
self.prob_death_iss_less_than_9 = p['prob_death_iss_less_than_9']
self.prob_death_iss_10_15 = p['prob_death_iss_10_15']
self.prob_death_iss_16_24 = p['prob_death_iss_16_24']
self.prob_death_iss_25_35 = p['prob_death_iss_25_35']
self.prob_death_iss_35_plus = p['prob_death_iss_35_plus']
[docs]
def apply(self, person_id):
df = self.sim.population.props
randfordeath = self.module.rng.random_sample(size=1)
# ======================================== Tests ==============================================================
assert df.loc[person_id, 'rt_ISS_score'] > 0
mortality_checked = False
probabilities_of_death = {
'1-4': [range(1, 5), 0],
'5-9': [range(5, 10), self.prob_death_iss_less_than_9],
'10-15': [range(10, 16), self.prob_death_iss_10_15],
'16-24': [range(16, 25), self.prob_death_iss_16_24],
'25-35': [range(25, 36), self.prob_death_iss_25_35],
'35-75': [range(25, 76), self.prob_death_iss_35_plus]
}
# Schedule death for those who died from their injuries despite medical intervention
if df.loc[person_id, 'cause_of_death'] == 'Other':
pass
for range_boundaries in probabilities_of_death.keys():
if df.loc[person_id].rt_ISS_score in probabilities_of_death[range_boundaries][0]:
if randfordeath < probabilities_of_death[range_boundaries][1]:
mortality_checked = True
df.loc[person_id, 'rt_post_med_death'] = True
dict_to_output = {'person': person_id,
'First injury': df.loc[person_id, 'rt_injury_1'],
'Second injury': df.loc[person_id, 'rt_injury_2'],
'Third injury': df.loc[person_id, 'rt_injury_3'],
'Fourth injury': df.loc[person_id, 'rt_injury_4'],
'Fifth injury': df.loc[person_id, 'rt_injury_5'],
'Sixth injury': df.loc[person_id, 'rt_injury_6'],
'Seventh injury': df.loc[person_id, 'rt_injury_7'],
'Eight injury': df.loc[person_id, 'rt_injury_8']}
logger.info(key='RTI_Death_Injury_Profile',
data=dict_to_output,
description='The injury profile of those who have died due to rtis despite medical care'
)
# Schedule the death
self.sim.modules['Demography'].do_death(individual_id=person_id, cause="RTI_death_with_med",
originating_module=self.module)
# Log the death
logger.debug(key='rti_general_message',
data=f"This is RTIMedicalInterventionDeathEvent scheduling a death for person "
f"{person_id} who was treated for their injuries but still died on date "
f"{self.sim.date}")
else:
mortality_checked = True
assert mortality_checked, 'Something missing in criteria'
[docs]
class RTI_No_Lifesaving_Medical_Intervention_Death_Event(Event, IndividualScopeEventMixin):
"""This is the NoMedicalInterventionDeathEvent. It is scheduled by the MedicalInterventionEvent which determines the
resources required to treat that person and if they aren't present, the person is sent here. This function is also
called by the did not run function for rti_major_surgeries for certain injuries, implying that if life saving
surgery is not available for the person, then we have to ask the probability of them dying without having this life
saving surgery.
some information on time to craniotomy here:
https://thejns.org/focus/view/journals/neurosurg-focus/45/6/article-pE2.xml?body=pdf-10653
The properties altered by this event are:
rt_unavailable_med_death - to denote that this person has died due to medical interventions not being available
"""
[docs]
def __init__(self, module, person_id):
super().__init__(module, person_id=person_id)
p = self.module.parameters
# load the parameteres used for this event
self.prob_death_TBI_SCI_no_treatment = p['prob_death_TBI_SCI_no_treatment']
self.prob_death_fractures_no_treatment = p['prob_death_fractures_no_treatment']
self.prop_death_burns_no_treatment = p['prop_death_burns_no_treatment']
self.prob_death_MAIS3 = p['prob_death_MAIS3']
self.prob_death_MAIS4 = p['prob_death_MAIS4']
self.prob_death_MAIS5 = p['prob_death_MAIS5']
self.prob_death_MAIS6 = p['prob_death_MAIS6']
self.allowed_interventions = p['allowed_interventions']
[docs]
def apply(self, person_id):
probabilities_of_death = {
'1': 0,
'2': 0,
'3': self.prob_death_MAIS3,
'4': self.prob_death_MAIS4,
'5': self.prob_death_MAIS5,
'6': self.prob_death_MAIS6
}
life_threatening_injuries = ['133a', '133b', '133c', '133d', '134a', '134b', '135', # TBI
'112', # Depressed skull fracture
'P133a', 'P133b', 'P133c', 'P133d', 'P134a', 'P134b', 'P135', # Perm TBI
'342', '343', '361', '363', # Injuries to neck
'414', '441', '443', '463', '453a', '453b', # Severe chest trauma
'782b', # Unilateral arm amputation
'783', # Bilateral arm amputation
'883', # Unilateral lower limb amputation
'884', # Bilateral lower limb amputation
'552', '553', '554' # Internal organ injuries
]
df = self.sim.population.props
untreated_injuries = []
persons_injuries = df.loc[[person_id], RTI.INJURY_COLUMNS]
non_empty_injuries = persons_injuries[persons_injuries != "none"]
non_empty_injuries = non_empty_injuries.dropna(axis=1)
# drop injuries that have a treatment scheduled
person = df.loc[person_id]
treatment_plan = (
person['rt_injuries_for_minor_surgery'] + person['rt_injuries_to_cast'] +
person['rt_injuries_to_heal_with_time'] + person['rt_injuries_for_open_fracture_treatment']
)
maj_surg_codes = ['112', '811', '812', '813a', '813b', '813c', '133a', '133b', '133c', '133d', '134a', '134b',
'135', '552', '553', '554', '342', '343', '414', '361', '363', '782', '782a', '782b', '782c',
'783', '822a', '882', '883', '884', 'P133a', 'P133b', 'P133c', 'P133d', 'P134a', 'P134b',
'P135', 'P782a', 'P782b', 'P782c', 'P783', 'P882', 'P883', 'P884']
# If we have allowed spinal cord surgeries to be treated in this simulation, include the associated injury
# codes here
if 'include_spine_surgery' in self.allowed_interventions:
additional_codes = ['673a', '673b', '674a', '674b', '675a', '675b', '676', 'P673a', 'P673b', 'P674',
'P674a', 'P674b', 'P675', 'P675a', 'P675b', 'P676']
for code in additional_codes:
maj_surg_codes.append(code)
# If we have allowed greater access to thoroscopy, include the codes treated by thoroscopy here
if 'include_thoroscopy' in self.allowed_interventions:
additional_codes = ['441', '443', '453', '453a', '453b', '463']
for code in additional_codes:
maj_surg_codes.append(code)
for col in non_empty_injuries:
# create the conditions to ignore untreated injuries
injury_treated_elsewhere = non_empty_injuries[col].values.to_list()[0] in treatment_plan
injury_not_treated_by_major_surgery = non_empty_injuries[col].values.to_list()[0] not in maj_surg_codes
condition_to_remove_column = injury_treated_elsewhere or injury_not_treated_by_major_surgery
if condition_to_remove_column:
non_empty_injuries = non_empty_injuries.drop(col, axis=1)
for col in non_empty_injuries:
if pd.isnull(df.loc[person_id, RTI.INJURY_DATE_COLUMN_MAP[col]]):
untreated_injuries.append(df.at[person_id, col])
untreated_injuries = [code for code in untreated_injuries if code in life_threatening_injuries]
mais_scores = [1]
for injury in untreated_injuries:
mais_scores.append(self.module.ASSIGN_INJURIES_AND_DALY_CHANGES[injury][0][-1])
max_untreated_injury = max(mais_scores)
prob_death = probabilities_of_death[str(max_untreated_injury)]
randfordeath = self.module.rng.random_sample(size=1)
if randfordeath < prob_death:
df.loc[person_id, 'rt_unavailable_med_death'] = True
self.sim.modules['Demography'].do_death(individual_id=person_id, cause="RTI_unavailable_med",
originating_module=self.module)
# Log the death
logger.debug(key='rti_general_message',
data=f"This is RTINoMedicalInterventionDeathEvent scheduling a death for person {person_id} on"
f" date {self.sim.date}")
else:
# person has survived their injuries despite the lack of treatment. Assign a recovery date to their injuries
# If a spinal injury, amputation, TBI is untreated, assign this injury's recovery time to the end of the
# simulation
codes = ['673', '673a', '673b', '674', '674a', '674b', '675', '675a', '675b', '676',
'782a', '782b', '782c', '783', '882', '883', '884', '133', '133a', '133b', '133c', '133d', '134',
'134a', '134b', '135']
for injury in untreated_injuries:
if injury in codes:
# Track whether they are permanently disabled
df.at[person_id, 'rt_perm_disability'] = True
# Find the column and code where the permanent injury is stored
col = self.module.rti_find_injury_column(person_id, injury)[0]
df.loc[person_id, col] = "P" + injury
# schedule the recovery date for the permanent injury for beyond the end of the simulation (making
# it permanent)
df.loc[person_id, RTI.INJURY_DATE_COLUMN_MAP[col]] = (
self.sim.end_date + DateOffset(days=1)
)
assert df.loc[person_id, RTI.INJURY_DATE_COLUMN_MAP[col]] > self.sim.date
# all injuries are handled by major surgery here, remove the untreated injury code
df.loc[person_id, 'rt_injuries_for_major_surgery'].remove(injury)
else:
# check if the injury has a heal time associated with no treamtent
if injury in self.module.NO_TREATMENT_RECOVERY_TIMES_IN_DAYS.keys():
df.loc[person_id, RTI.INJURY_DATE_COLUMN_MAP[col]] = \
self.sim.end_date + DateOffset(days=self.module.NO_TREATMENT_RECOVERY_TIMES_IN_DAYS[injury])
if injury in df.loc[person_id, 'rt_injuries_for_major_surgery']:
df.loc[person_id, 'rt_injuries_for_major_surgery'].remove(injury)
# ---------------------------------------------------------------------------------------------------------
# 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 RTI_Logging_Event(RegularEvent, PopulationScopeEventMixin):
[docs]
def __init__(self, module):
"""Produce a summary of the numbers of people with respect to the action of this module.
This is a regular event that can output current states of people or cumulative events since last logging event.
"""
# run this event every month
self.repeat = 1
super().__init__(module, frequency=DateOffset(months=self.repeat))
assert isinstance(module, RTI)
# Create variables used to store simulation data in
# Number of injured body region data
self.tot1inj = 0
self.tot2inj = 0
self.tot3inj = 0
self.tot4inj = 0
self.tot5inj = 0
self.tot6inj = 0
self.tot7inj = 0
self.tot8inj = 0
# Injury category data
self.totfracnumber = 0
self.totdisnumber = 0
self.tottbi = 0
self.totsoft = 0
self.totintorg = 0
self.totintbled = 0
self.totsci = 0
self.totamp = 0
self.toteye = 0
self.totextlac = 0
self.totburns = 0
# Injury location on body data
self.totAIS1 = 0
self.totAIS2 = 0
self.totAIS3 = 0
self.totAIS4 = 0
self.totAIS5 = 0
self.totAIS6 = 0
self.totAIS7 = 0
self.totAIS8 = 0
# Injury severity data
self.totmild = 0
self.totsevere = 0
# More model progression data
self.totinjured = 0
self.deathonscene = 0
self.soughtmedcare = 0
self.deathaftermed = 0
self.deathwithoutmed = 0
self.permdis = 0
self.ISSscore = []
self.severe_pain = 0
self.moderate_pain = 0
self.mild_pain = 0
# Create variables for averages over time in the model
self.numerator = 0
self.denominator = 0
self.death_inc_numerator = 0
self.death_in_denominator = 0
self.fracdenominator = 0
# Create variables to measure where certain injuries are located on the body
self.fracdist = [0, 0, 0, 0, 0, 0, 0, 0]
self.openwounddist = [0, 0, 0, 0, 0, 0, 0, 0]
self.burndist = [0, 0, 0, 0, 0, 0</