Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Minimal counterfactualism #19

Closed
wants to merge 1 commit into from
Closed

Minimal counterfactualism #19

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
122 changes: 40 additions & 82 deletions ringvax/__init__.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,5 +1,6 @@
from typing import Any, List, Optional

import numpy as np
import numpy.random


Expand All @@ -12,7 +13,6 @@ def __init__(self, params: dict[str, Any], seed: Optional[int] = None):

def run(self):
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

We should probably now just rename run_infections -> run

self.run_infections()
self.intervene()

def create_person(self) -> str:
"""Add a new person to the data"""
Expand Down Expand Up @@ -44,12 +44,15 @@ def generate_person_properties(
"""Generate properties of a single infected person"""
# disease state history in this individual, and when they infect others
infection_history = self.generate_infection_history(t_exposed=t_exposed)

# passive detection
passive_detected = self.bernoulli(self.params["p_passive_detect"])

if passive_detected:
t_passive_detected = t_exposed + self.generate_passive_detection_delay()
# Remove infections that never happened do to passive detection
infection_history["t_infections"] = infection_history["t_infections"][
infection_history["t_infections"] < t_passive_detected
]
else:
t_passive_detected = None

Expand All @@ -66,9 +69,8 @@ def generate_person_properties(
"t_passive_detected": t_passive_detected,
"active_detected": None,
"t_active_detected": None,
"detected": None,
"t_detected": None,
"actually_infected": None,
"detected": True if passive_detected else None,
"t_detected": t_passive_detected if passive_detected else None,
} | infection_history

def run_infections(self) -> None:
Expand All @@ -95,6 +97,7 @@ def run_infections(self) -> None:
infector=infector, t_exposed=t_exposed
),
)
self.propagate_intervention(infectee)
# keep track of the people infected in the next generation
next_generation.append(infectee)

Expand All @@ -119,85 +122,40 @@ def run_infections(self) -> None:
n_infectees = len(self.query_people({"generation": g + 1}))
assert n_infections == n_infectees

def intervene(self) -> None:
"""Draw intervention outcomes and update chains of infection"""
for generation in range(self.params["n_generations"]):
# get infections in this generation
for infectee in self.query_people({"generation": generation}):
# process each infectee
self._intervene1(infectee)

def _intervene1(self, infectee: str) -> None:
"""Process intervention for a single infectee"""
infector = self.get_person_property(infectee, "infector")

is_index = infector is None
if is_index:
assert self.get_person_property(infectee, "generation") == 0

# you are actually infected if:
# you are the index infection OR (
# your infector was actually infected AND
# NOT they were detected early enough to stop your getting infected
# )
actually_infected = is_index or (
self.get_person_property(infector, "actually_infected")
and not (
self.get_person_property(infector, "detected")
and (
self.get_person_property(infector, "t_detected")
< self.get_person_property(infectee, "t_exposed")
)
)
)
def propagate_intervention(self, person: str):
infector = self.get_person_property(person, "infector")

# if you were actually infected, see if you infector was detected,
# so that you have a chance for active detection
if (
actually_infected
and not is_index
and self.get_person_property(infector, "detected")
# Run active detection on this individual, if eligible
if self.get_person_property(infector, "detected") and self.bernoulli(
self.params["p_active_detect"]
):
active_detected = self.bernoulli(self.params["p_active_detect"])
else:
active_detected = False

# if you were actively detected, when?
if active_detected:
t_active_detected = (
t_active_detection = (
self.get_person_property(infector, "t_detected")
+ self.generate_active_detection_delay()
)
else:
t_active_detected = None

# now reconcile everything that's happened to you
passive_detected = self.get_person_property(infectee, "passive_detected")
t_passive_detected = self.get_person_property(infectee, "t_passive_detected")

if active_detected and passive_detected:
assert t_passive_detected is not None
assert t_active_detected is not None
t_detected = min(t_passive_detected, t_active_detected)
elif active_detected and not passive_detected:
assert t_active_detected is not None
t_detected = t_active_detected
elif not active_detected and passive_detected:
assert t_passive_detected is not None
t_detected = t_passive_detected
else:
t_detected = None
t_detected = t_active_detection

self.update_person(
infectee,
{
"actually_infected": actually_infected,
"active_detected": active_detected,
"t_active_detected": t_active_detected,
"detected": passive_detected or active_detected,
"t_detected": t_detected,
},
)
# Adjust true detection time, if needed
if self.get_person_property(person, "passive_detected"):
t_detected = min(
t_active_detection,
self.get_person_property(person, "t_passive_detected"),
)

# Remove any infections they can't cause due to active detection
t_infections = self.get_person_property(person, "t_infections")
t_infections = t_infections[t_infections < t_detected]

self.update_person(
person,
{
"t_infections": t_infections,
"active_detected": True,
"t_active_detected": t_active_detection,
"detected": True,
"t_detected": t_detected,
},
)

def generate_infection_history(self, t_exposed: float) -> dict[str, Any]:
"""Generate infection history for a single infected person"""
Expand All @@ -213,7 +171,7 @@ def generate_infection_history(self, t_exposed: float) -> dict[str, Any]:

t_infectious = t_exposed + latent_duration
t_recovered = t_infectious + infectious_duration
t_infections = [t_infectious + d for d in infection_delays]
t_infections = t_infectious + infection_delays

return {
"t_exposed": t_exposed,
Expand All @@ -240,18 +198,18 @@ def generate_active_detection_delay(self) -> float:
@staticmethod
def generate_infection_delays(
rng: numpy.random.Generator, rate: float, infectious_duration: float
) -> List[float]:
) -> np.ndarray:
"""Times from onset of infectiousness to each infection"""
assert rate >= 0.0
assert infectious_duration >= 0.0

if rate == 0.0:
return []
return np.array(())

n_events = rng.poisson(infectious_duration * rate)
times = rng.uniform(0.0, infectious_duration, n_events)
times.sort()
return list(times)
return times

def bernoulli(self, p: float) -> bool:
return self.rng.binomial(n=1, p=p) == 1
Loading