Merge pull request #76 from TTI-modelling/fix_daily_testing

Fix daily testing

Author: AnnAnnFryingPan

examples/individual_level_daily_contact_testing.ipynb

@@ -94,8 +94,8 @@
    "outputs": [],
    "source": [
     "model = bpm.IndividualTracingDailyTesting(params)\n",
-    "model.prob_testing_positive_lfa_func = prob_testing_positive_function_lfa\n",
-    "model.prob_testing_positive_pcr_func = prob_testing_positive_function_pcr\n",
+    "model.prob_lfa_positive = prob_testing_positive_function_lfa\n",
+    "model.prob_pcr_positive = prob_testing_positive_function_pcr\n",
     "controller = BranchingProcessController(model)"
    ]
   },
@@ -108,6 +108,7 @@
    "outputs": [],
    "source": [
     "controller.graph_view.set_display(True)\n",
+    "controller.timeline_view.set_display(True)\n",
     "controller.run_simulation(16)"
    ]
   },
@@ -153,8 +154,8 @@
    "outputs": [],
    "source": [
     "model = bpm.IndividualTracingDailyTesting(params_policy_2)\n",
-    "model.prob_testing_positive_lfa_func = prob_testing_positive_function_lfa\n",
-    "model.prob_testing_positive_pcr_func = prob_testing_positive_function_pcr\n",
+    "model.intervention.prob_testing_positive_lfa_func = prob_testing_positive_function_lfa\n",
+    "model.intervention.prob_testing_positive_pcr_func = prob_testing_positive_function_pcr\n",
     "controller.model = model"
    ]
   },
@@ -242,10 +243,12 @@
   {
    "cell_type": "code",
    "execution_count": null,
-   "metadata": {},
+   "metadata": {
+    "scrolled": false
+   },
    "outputs": [],
    "source": [
-    "controller.graph_view.set_display(False)\n",
+    "controller.graph_view.set_display(True)\n",
     "controller.timeline_view.set_display(True)\n",
     "controller.run_simulation(20)"
    ]

examples/run_testing_contact_model.py

@@ -1,12 +1,14 @@
 import os
+from copy import copy
 
 from household_contact_tracing.branching_process_controller import BranchingProcessController
 from household_contact_tracing.branching_process_models import IndividualTracingDailyTesting
 
 
 def main():
-    example_1()
-    example_2()
+    #example_1()
+    #example_2()
+    example_3()
 
 def example_1():
 
@@ -103,5 +105,54 @@ def prob_pcr_positive(infectious_age):
         controller.run_simulation(20)
 
 
+# define some easy to look at test sensitivity functions
+def prob_testing_positive_function_pcr(time_relative_to_symptom_onset):
+    # Prevents people testing positive as soon as they get it
+    if time_relative_to_symptom_onset in [4, 5, 6]:
+        return 0.75
+    else:
+        return 0
+
+def prob_testing_positive_function_lfa(time_relative_to_symptom_onset):
+    # Prevents people testing positive as soon as they get it
+    if time_relative_to_symptom_onset in [4, 5, 6]:
+        return 0 # this is unrealistic, but it makes it easier to see nodes being lfa tested since they wont
+                 # move to the isolation status due to lfa testing
+    else:
+        return 0
+
+
+def example_3():
+    params = {"outside_household_infectivity_scaling": 0.3,
+              "contact_tracing_success_prob": 0.7,
+              "overdispersion": 0.32,
+              "asymptomatic_prob": 0.2,
+              "asymptomatic_relative_infectivity": 0.35,
+              "infection_reporting_prob": 0.5,
+              "LFA_testing_requires_confirmatory_PCR": False,
+              "reduce_contacts_by": 0.5,
+              "test_delay": 1,
+              "contact_trace_delay": 1,
+              "incubation_period_delay": 5,
+              "symptom_reporting_delay": 1,
+              "starting_infections": 5,
+              "household_pairwise_survival_prob": 0.2,
+              "self_isolation_duration": 10,
+              "lateral_flow_testing_duration": 14,
+              "propensity_risky_behaviour_lfa_testing": 0,
+              "global_contact_reduction_risky_behaviour": 0,
+              "household_positive_policy": "lfa_testing_no_quarantine"
+              }
+
+    model = IndividualTracingDailyTesting(params)
+    model.prob_lfa_positive = prob_testing_positive_function_lfa
+    model.prob_pcr_positive = prob_testing_positive_function_pcr
+    controller = BranchingProcessController(model)
+
+    controller.graph_view.set_display(True)
+    controller.timeline_view.set_display(True)
+    controller.run_simulation(16)
+
+
 if __name__ == "__main__":
     main()

household_contact_tracing/behaviours/intervention/increment_tracing.py

@@ -184,9 +184,8 @@ def update_contact_tracing_index(self, time: int):
 class IncrementTracingIndividualLevel(IncrementTracingHouseholdLevel):
 
     def __init__(self, network: Network, params: dict):
-
         super().__init__(network, params)
-        self.prob_pcr_positive = self.default_prob_pcr_positive
+        self._prob_pcr_positive = None
 
     @property
     def prob_pcr_positive(self) -> Callable[[int], float]:
@@ -196,18 +195,10 @@ def prob_pcr_positive(self) -> Callable[[int], float]:
     def prob_pcr_positive(self, fn: Callable[[int], float]):
         self._prob_pcr_positive = fn
 
-    @staticmethod
-    def default_prob_pcr_positive(infectious_age):
-        """Default PCR test result probability."""
-        if infectious_age in [4, 5, 6]:
-            return 0
-        else:
-            return 0
-
-    def pcr_test_node(self, node: Node, time: int, prob_pcr_positive: Callable):
+    def pcr_test_node(self, node: Node, time: int):
         node.received_result = True
         infectious_age_when_tested = time - node.testing_delay - node.time_infected
-        prob_positive_result = prob_pcr_positive(infectious_age_when_tested)
+        prob_positive_result = self.prob_pcr_positive(infectious_age_when_tested)
         node.avenue_of_testing = TestType.pcr
 
         if np.random.binomial(1, prob_positive_result) == 1:
@@ -216,20 +207,20 @@ def pcr_test_node(self, node: Node, time: int, prob_pcr_positive: Callable):
         else:
             node.received_positive_test_result = False
 
-    def receive_pcr_test_results(self, time: int, prob_pcr_positive: Callable):
+    def receive_pcr_test_results(self, time: int):
         # self reporting infections
         for node in self.network.all_nodes():
             if node.time_of_reporting + node.testing_delay == time:
                 if not node.contact_traced:
                     if not node.received_result:
-                        self.pcr_test_node(node, time, prob_pcr_positive)
+                        self.pcr_test_node(node, time)
 
         # contact traced nodes
         for node in self.network.all_nodes():
             if node.symptom_onset_time + node.testing_delay == time:
                 if node.contact_traced:
                     if not node.received_result:
-                        self.pcr_test_node(node, time, prob_pcr_positive)
+                        self.pcr_test_node(node, time)
 
     def increment_contact_tracing(self, time: int):
 
@@ -249,7 +240,7 @@ def increment_contact_tracing(self, time: int):
 
         # Isolate all households under observation that now display symptoms (excludes those who will not take up
         # intervention if prob <1)
-        self.receive_pcr_test_results(time, self.prob_pcr_positive)
+        self.receive_pcr_test_results(time)
 
         for node in self.network.all_nodes():
             if node.symptom_onset_time <= time:
@@ -360,18 +351,18 @@ def attempt_contact_trace_of_household(self,
 
 class IncrementTracingIndividualDailyTesting(IncrementTracingIndividualLevel):
 
-    def receive_pcr_test_results(self, time: int, prob_pcr_positive: Callable):
+    def receive_pcr_test_results(self, time: int):
         """For nodes who would receive a PCR test result today, update"""
 
         if self.lfa_tested_nodes_book_pcr_on_symptom_onset:
-            super().receive_pcr_test_results(time, prob_pcr_positive)
+            super().receive_pcr_test_results(time)
         else:
             for node in self.network.all_nodes():
                 if node.time_of_reporting + node.testing_delay == time:
                     if not node.contact_traced:
                         if not node.received_result:
                             if not node.being_lateral_flow_tested:
-                                self.pcr_test_node(node, time, prob_pcr_positive)
+                                self.pcr_test_node(node, time)
 
     def increment_contact_tracing(self, time: int):
         for node in self.network.all_nodes():

household_contact_tracing/behaviours/intervention/isolation.py

@@ -131,6 +131,19 @@ def update_isolation(self, time: int):
 
 
 class DailyTestingIsolation(HouseholdIsolation):
+
+    def __init__(self, network: Network, params: dict):
+        super().__init__(network, params)
+        self._prob_pcr_positive = None
+
+    @property
+    def prob_pcr_positive(self) -> Callable[[int], float]:
+        return self._prob_pcr_positive
+
+    @prob_pcr_positive.setter
+    def prob_pcr_positive(self, fn: Callable[[int], float]):
+        self._prob_pcr_positive = fn
+
     def update_households_contact_traced(self, time: int):
         """Update the contact traced status for all households that have had the
          contact tracing process get there."""
@@ -151,8 +164,7 @@ def update_isolation(self, time: int):
                 if node.avenue_of_testing == TestType.pcr:
                     if node.received_positive_test_result:
                         if not node.household.applied_household_positive_policy:
-                            node.household.apply_positive_policy(time,
-                                                                 self.household_positive_policy)
+                            node.household.apply_positive_policy(time, self.household_positive_policy)
 
     def act_on_confirmatory_pcr_results(self, time: int):
         """Once on a individual receives a positive pcr result we need to act on it.
@@ -186,20 +198,17 @@ def isolate_positive_lateral_flow_tests(self, time: int, positive_nodes: List[No
                     not self.LFA_testing_requires_confirmatory_PCR:
                 node.household.apply_positive_policy(time, self.household_positive_policy)
 
-    def act_on_positive_LFA_tests(self, time: int, prob_pcr_positive: Callable,
-                                  positive_nodes: List[Node]):
+    def act_on_positive_LFA_tests(self, time: int, positive_nodes: List[Node]):
         """For nodes who test positive on their LFA test, take the appropriate action depending
         on the policy
         """
         self.isolate_positive_lateral_flow_tests(time, positive_nodes)
 
         if self.LFA_testing_requires_confirmatory_PCR:
-            self.confirmatory_pcr_test_LFA_nodes(time, positive_nodes, prob_pcr_positive)
+            self.confirmatory_pcr_test_LFA_nodes(time, positive_nodes)
 
-    @staticmethod
-    def confirmatory_pcr_test_LFA_nodes(time: int, positive_nodes: List[Node],
-                                        prob_pcr_positive: Callable):
+    def confirmatory_pcr_test_LFA_nodes(self, time: int, positive_nodes: List[Node]):
         """Nodes who receive a positive LFA result will be tested using a PCR test."""
         for node in positive_nodes:
             if not node.taken_confirmatory_PCR_test:
-                node.take_confirmatory_pcr_test(time, prob_pcr_positive)
+                node.take_confirmatory_pcr_test(time, self.prob_pcr_positive)

household_contact_tracing/branching_process_models.py

@@ -161,11 +161,13 @@ class IndividualLevelTracing(HouseholdLevelTracing):
         contacting every individual and their contacts, whether they have tested positive or not.
     """
     def __init__(self, params: dict):
-        super().__init__(params)
         # Set the test probabilities to default values - may be overridden by user later
         self._prob_lfa_positive = self.default_prob_lfa_positive
         self._prob_pcr_positive = self.default_prob_pcr_positive
 
+        # Run super constructor. Note: This must be run after above, as above attributes required.
+        super().__init__(params)
+
     @property
     def prob_lfa_positive(self) -> Callable[[int], float]:
         return self._prob_lfa_positive
@@ -181,20 +183,22 @@ def prob_pcr_positive(self) -> Callable[[int], float]:
     @prob_pcr_positive.setter
     def prob_pcr_positive(self, fn: Callable[[int], float]):
         self._prob_pcr_positive = fn
+        self.intervention.increment_tracing.prob_pcr_positive = fn
+        self.intervention.isolation.prob_pcr_positive = fn
 
     @staticmethod
-    def default_prob_pcr_positive(infectious_age):
-        """Default PCR test result probability."""
+    def default_prob_lfa_positive(infectious_age):
+        """Default LFA test result probability."""
         if infectious_age in [4, 5, 6]:
-            return 0
+            return 1
         else:
             return 0
 
     @staticmethod
-    def default_prob_lfa_positive(infectious_age):
-        """Default LFA test result probability."""
+    def default_prob_pcr_positive(infectious_age):
+        """Default PCR test result probability."""
         if infectious_age in [4, 5, 6]:
-            return 1
+            return 0
         else:
             return 0
 
@@ -210,10 +214,17 @@ def _initialise_infection(self):
 
     def _initialise_intervention(self):
         """ Initialise an Intervention class, passing in the required behaviours into its constructor """
-        return Intervention(self.network,
-                            isolation.IndividualIsolation,
-                            increment.IncrementTracingIndividualLevel,
-                            self.params)
+        new_intervention = Intervention(self.network,
+                                        isolation.IndividualIsolation,
+                                        increment.IncrementTracingIndividualLevel,
+                                        self.params)
+
+        # Set a new positive pcr probability function
+        new_intervention.increment_tracing.prob_pcr_positive = self.prob_pcr_positive
+        new_intervention.isolation.prob_pcr_positive = self.prob_pcr_positive
+
+        return new_intervention
+
 
     def simulate_one_step(self):
         """Simulates one day of the infection and contact tracing."""
@@ -225,8 +236,6 @@ def simulate_one_step(self):
         # isolate self-reporting-nodes while they wait for tests
         self.intervention.isolation.update_households_contact_traced(self.time)
         self.intervention.isolation.update_isolation(self.time)
-        # Set a new positive pcr probability function and propagate contact tracing
-        self.intervention.increment_tracing.prob_pcr_positive = self.prob_pcr_positive
         for step in range(5):
             self.intervention.increment_tracing.increment_contact_tracing(self.time)
         # node recoveries
@@ -259,24 +268,32 @@ def _initialise_infection(self):
                          new_infection.NewInfectionIndividualTracingDailyTesting,
                          ContactRateReductionIndividualTracingDaily,
                          self.params)
+
     def _initialise_intervention(self):
         """ Initialise an Intervention class, passing in the required behaviours into its constructor """
-        return Intervention(self.network,
-                            isolation.DailyTestingIsolation,
-                            increment.IncrementTracingIndividualDailyTesting,
-                            self.params)
+        new_intervention =  Intervention(self.network,
+                                         isolation.DailyTestingIsolation,
+                                         increment.IncrementTracingIndividualDailyTesting,
+                                         self.params)
+
+        # Set a new positive pcr probability function
+        new_intervention.increment_tracing.prob_pcr_positive = self.prob_pcr_positive
+        new_intervention.isolation.prob_pcr_positive = self.prob_pcr_positive
+
+        return new_intervention
 
     def simulate_one_step(self):
         """ Simulates one day of the infection and contact tracing.
         """
+        self.intervention.increment_tracing.receive_pcr_test_results(self.time)
         # isolate nodes reached by tracing, isolate nodes due to self-reporting
         self.intervention.isolation.isolate_self_reporting_cases(self.time)
         # isolate self-reporting-nodes while they wait for tests
         self.intervention.isolation.update_households_contact_traced(self.time)
         self.intervention.isolation.update_isolation(self.time)
         # isolate self reporting nodes
         positive_nodes = self.intervention.lft_nodes(self.time, self.prob_lfa_positive)
-        self.intervention.isolation.act_on_positive_LFA_tests(self.time, self.prob_pcr_positive, positive_nodes)
+        self.intervention.isolation.act_on_positive_LFA_tests(self.time, positive_nodes)
         # if we require PCR tests, to confirm infection we act on those
         if self.intervention.increment_tracing.LFA_testing_requires_confirmatory_PCR:
             self.intervention.increment_tracing.act_on_confirmatory_pcr_results(self.time)