Allow option to overwrite parameter file

Author: marghe-molaro

resources/ResourceFile_GenerateEventChains/parameter_values.csv

@@ -1,3 +1,3 @@
 version https://git-lfs.github.com/spec/v1
-oid sha256:788332259649a19889355820b194fe0d16af44436f208e3a948e833f0ad5139a
-size 419
+oid sha256:172a0c24c859aaafbad29f6016433cac7a7324efc582e6c4b19c74b6b97436e7
+size 420

src/scripts/analysis_data_generation/scenario_generate_chains.py

@@ -48,7 +48,7 @@
     wasting,
 )
 
-class GenerateDataChains(BaseScenario):
+class GenerateEventChains(BaseScenario):
     def __init__(self):
         super().__init__()
         self.seed = 42
@@ -71,101 +71,30 @@ def log_configuration(self):
                 'tlo.methods.demography.detail': logging.WARNING,
                 'tlo.methods.healthburden': logging.INFO,
                 'tlo.methods.healthsystem.summary': logging.INFO,
+                'tlo.methods.collect_event_chains': logging.INFO
             }
         }
 
     def modules(self):
-        # MODIFY
-        # Here instead of running full module
-        """
-        return [demography.Demography(resourcefilepath=self.resources),
-                enhanced_lifestyle.Lifestyle(resourcefilepath=self.resources),
-                healthburden.HealthBurden(resourcefilepath=self.resources),
-                symptommanager.SymptomManager(resourcefilepath=self.resources, spurious_symptoms=False),#,
-                #rti.RTI(resourcefilepath=self.resources),
-                pregnancy_supervisor.PregnancySupervisor(resourcefilepath=self.resources),
-                labour.Labour(resourcefilepath=self.resources),
-                care_of_women_during_pregnancy.CareOfWomenDuringPregnancy(resourcefilepath=self.resources),
-                contraception.Contraception(resourcefilepath=self.resources),
-                newborn_outcomes.NewbornOutcomes(resourcefilepath=self.resources),
-                postnatal_supervisor.PostnatalSupervisor(resourcefilepath=self.resources),
-                hiv.Hiv(resourcefilepath=self.resources),
-                tb.Tb(resourcefilepath=self.resources),
-                epi.Epi(resourcefilepath=self.resources),
-                healthseekingbehaviour.HealthSeekingBehaviour(resourcefilepath=self.resources),
-            #simplified_births.SimplifiedBirths(resourcefilepath=resourcefilepath),
-                healthsystem.HealthSystem(resourcefilepath=self.resources,
-                                          mode_appt_constraints=1,
-                                          cons_availability='all')]
-        """
         return (
-            fullmodel(resourcefilepath=self.resources)
-            + [ImprovedHealthSystemAndCareSeekingScenarioSwitcher(resourcefilepath=self.resources)]
+            fullmodel()
         )
-    """
-    def draw_parameters(self, draw_number, rng):
-        return mix_scenarios(
-            get_parameters_for_status_quo(),
-            {
-                'HealthSystem': {
-                    'Service_Availability': list(self._scenarios.values())[draw_number],
-                },
-            }
-        )
-
-    def _get_scenarios(self) -> Dict[str, list[str]]:
-        Return the Dict with values for the parameter `Service_Availability` keyed by a name for the scenario.
-        The sequences of scenarios systematically omits one of the TREATMENT_ID's that is defined in the model.
 
-        # Generate list of TREATMENT_IDs and filter to the resolution needed
-        treatments = get_filtered_treatment_ids(depth=2)
-        treatments_RTI = [item for item in treatments if 'Rti' in item]
-        
-        # Return 'Service_Availability' values, with scenarios for everything, nothing, and ones for which each
-        # treatment is omitted
-        service_availability = dict({"Everything": ["*", "Nothing": []})
-        #service_availability.update(
-        #    {f"No {t.replace('_*', '*')}": [x for x in treatments if x != t] for t in treatments_RTI}
-        #)
-        
-        return service_availability
-
-    """
     def draw_parameters(self, draw_number, rng):
         if draw_number < self.number_of_draws:
             return list(self._scenarios.values())[draw_number]
         else:
             return
 
-    # case 1: gfHE = -0.030, factor = 1.01074
-    # case 2: gfHE = -0.020, factor = 1.02116
-    # case 3: gfHE = -0.015, factor = 1.02637
-    # case 4: gfHE =  0.015, factor = 1.05763
-    # case 5: gfHE =  0.020, factor = 1.06284
-    # case 6: gfHE =  0.030, factor = 1.07326
-
     def _get_scenarios(self) -> Dict[str, Dict]:
-        #Return the Dict with values for the parameters that are changed, keyed by a name for the scenario.
-        
-        treatments = get_filtered_treatment_ids(depth=2)
-        treatments_RTI = [item for item in treatments if 'Rti' in item]
-        
-        # Return 'Service_Availability' values, with scenarios for everything, nothing, and ones for which each
-        # treatment is omitted
-        service_availability = dict({"Everything": ["*"], "Nothing": []})
-        service_availability.update(
-            {f"No {t.replace('_*', '*')}": [x for x in treatments if x != t] for t in treatments_RTI}
-        )
-        print(service_availability.keys())
 
         return {
-            # =========== STATUS QUO ============
             "Baseline":
                 mix_scenarios(
                     self._baseline(),
                     {
-                     "HealthSystem": {
-                            "Service_Availability": service_availability["No Rti_ShockTreatment*"],
+                     "CollectEventChains": {
+                            "generate_event_chains": True,
                       },
                     }
                 ),

src/tlo/events.py

@@ -60,11 +60,10 @@ def apply(self, target):
         :param target: the target of the event
         """
         raise NotImplementedError
-        
 
     def run(self):
         """Make the event happen."""
-        
+
         # Dispatch notification that event is about to run
         notifier.dispatch("event.about_to_run", data={"target": self.target, "module" : self.module.name, "link_info" : {"EventName": type(self).__name__}})
 

src/tlo/methods/collect_event_chains.py

@@ -58,17 +58,9 @@ def initialise_simulation(self, sim):
         notifier.add_listener("event.has_just_ran", self.on_notification_event_has_just_ran)
 
     def read_parameters(self, resourcefilepath: Optional[Path] = None):
-        print("resource file path", resourcefilepath)
         self.load_parameters_from_dataframe(pd.read_csv(resourcefilepath/"ResourceFile_GenerateEventChains/parameter_values.csv"))
-
-        # If modules of interest is '*', set by default to all modules included in the simulation
-        if self.parameters["modules_of_interest"] == ['*']:
-            self.parameters["modules_of_interest"] = list(self.sim.modules.keys())
 
     def initialise_population(self, population):
-        pass
-
-    def initialise_simulation(self, sim):
         # Use parameter file values by default, if not overwritten
         self.generate_event_chains = self.parameters['generate_event_chains'] \
             if self.generate_event_chains is None \
@@ -81,6 +73,10 @@ def initialise_simulation(self, sim):
         self.events_to_ignore = self.parameters['events_to_ignore'] \
             if self.events_to_ignore is None \
             else self.events_to_ignore
+            
+        # If modules of interest is '*', set by default to all modules included in the simulation
+        if self.modules_of_interest == ['*']:
+            self.modules_of_interest = list(self.sim.modules.keys())
 
     def get_generate_event_chains(self) -> bool:
         """Returns `generate_event_chains`. (Should be equal to what is specified by the parameter, but
@@ -134,7 +130,6 @@ def on_notification_event_about_to_run(self, data):
         if not self.generate_event_chains or (data['module'] not in self.modules_of_interest) or (data['link_info']['EventName'] in self.events_to_ignore):
             return
         else:
-                      
             # Initialise these variables
             self.print_chains = False
             self.df_before = []