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

Jump to bottom

Adjust test calls to use Python interface #234

Merged
merged 1 commit into from
Jan 10, 2023
Merged

Adjust test calls to use Python interface #234

Show file tree
Hide file tree
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
43 changes: 24 additions & 19 deletions tests/test_demographics.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -7,7 +7,6 @@
@author: anelnurtay
"""

import subprocess
import numpy as np, pandas as pd
import pytest
import random as rd
Expand Down Expand Up @@ -176,16 +175,17 @@ def test_demographic_proportions(
"""

error_tolerance = 0.01

params = ParameterSet(constant.TEST_DATA_FILE, line_number = 1)

params = utils.get_params_swig()

params.set_param("n_total", n_total)
params.set_param("end_time", 1)
params.set_param("population_0_9",population_0_9)
params.set_param("population_10_19",population_10_19)
params.set_param("population_20_29",population_20_29)
params.set_param("population_30_39",population_30_39)
params.set_param("population_40_49",population_40_49)
params.set_param("population_50_59",population_50_59)
params.set_param("population_40_49",population_40_49)
params.set_param("population_50_59",population_50_59)
params.set_param("population_60_69",population_60_69)
params.set_param("population_70_79",population_70_79)
params.set_param("population_80",population_80)
Expand All @@ -194,11 +194,15 @@ def test_demographic_proportions(
population_30_39, population_40_49, population_50_59,
population_60_69, population_70_79, population_80 ]

params.write_params(constant.TEST_DATA_FILE)
file_output = open(constant.TEST_OUTPUT_FILE, "w")
completed_run = subprocess.run([constant.command], stdout = file_output, shell = True)
df_indiv = pd.read_csv(constant.TEST_INDIVIDUAL_FILE,
comment = "#", sep = ",", skipinitialspace = True )
model = utils.get_model_swig( params )

# step through the model and write the relevant files the end
for _ in range( params.get_param( "end_time") ):
model.one_time_step()

model.write_individual_file()

df_indiv = pd.read_csv( constant.TEST_INDIVIDUAL_FILE )

# population proportion by age
N_tot = len( df_indiv )
Expand All @@ -213,7 +217,7 @@ def test_household_size(self, n_total, household_size_1, household_size_2,
"""

# Set the parameters we want for the simulation.
params = ParameterSet(constant.TEST_DATA_FILE, line_number=1)
params = utils.get_params_swig()
params.set_param("end_time", 1)
params.set_param("n_total", n_total)
params.set_param("household_size_1", household_size_1)
Expand All @@ -222,7 +226,6 @@ def test_household_size(self, n_total, household_size_1, household_size_2,
params.set_param("household_size_4", household_size_4)
params.set_param("household_size_5", household_size_5)
params.set_param("household_size_6", household_size_6)
params.write_params(constant.TEST_DATA_FILE)

# Calculate the number of people expected to be living in households of
# each different size, based on the parameter definitions.
Expand All @@ -234,16 +237,18 @@ def test_household_size(self, n_total, household_size_1, household_size_2,
household_size_counts_weighted *= float(n_total) / \
sum(household_size_counts_weighted)

# Run the simulation.
file_output = open(constant.TEST_OUTPUT_FILE, "w")
completed_run = subprocess.run([constant.command], stdout=file_output,
stderr=file_output, shell=True)
np.testing.assert_equal(completed_run.returncode, 0)
# Run the simulation
model = utils.get_model_swig( params )

# step through the model and write the relevant files the end
for _ in range( params.get_param( "end_time") ):
model.one_time_step()

model.write_individual_file()

# Find the number of people living in households of each different size
# in the simulation output.
df_indiv = pd.read_csv(constant.TEST_INDIVIDUAL_FILE,
comment="#", sep=",", skipinitialspace = True)
df_indiv = pd.read_csv( constant.TEST_INDIVIDUAL_FILE )

df_house = df_indiv.groupby(["house_no"]).size().reset_index(name="size")
df_house = df_house.groupby(["size"]).size().reset_index(
Expand Down
116 changes: 58 additions & 58 deletions tests/test_disease_dynamics.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -11,7 +11,6 @@
"""

import pytest
import subprocess
import sys
import numpy as np, pandas as pd
from math import sqrt
Expand Down Expand Up @@ -702,75 +701,77 @@ class TestClass(object):
"""
def test_zero_recovery(self):
"""
Setting recover times to be very large should avoid seeing any in recovered compartment
Setting recover times to be very large should avoid seeing any in cumulative recovered compartment ('n_recovered')
"""
params = ParameterSet(constant.TEST_DATA_FILE, line_number = 1)
params = utils.get_params_swig()
params.set_param("n_total", 10000)
params.set_param("end_time", 100)

# Make recovery very long
params.set_param("mean_time_to_recover", 200.0)
params.set_param("mean_asymptomatic_to_recovery", 200.0)
params.set_param("mean_time_hospitalised_recovery", 200.0)

params.write_params(constant.TEST_DATA_FILE)


# Call the model
file_output = open(constant.TEST_OUTPUT_FILE, "w")
completed_run = subprocess.run([constant.command], stdout = file_output, shell = True)
df_output = pd.read_csv(constant.TEST_OUTPUT_FILE, comment = "#", sep = ",")

np.testing.assert_array_equal(
df_output[["n_recovered"]].sum(),
model = utils.get_model_swig( params )
for time in range(params.get_param("end_time")):
model.one_time_step()
np.testing.assert_equal(
model.one_time_step_results()["n_recovered"],
0)

def test_zero_deaths(self):
"""
Set fatality ratio to zero, should have no deaths if always places in the ICU
"""
params = ParameterSet(constant.TEST_DATA_FILE, line_number = 1)
params = utils.get_params_swig()
params.set_param("n_total", 10000)
params = utils.set_fatality_fraction_all(params, 0.0)
params = utils.set_location_death_icu_all(params, 1.0)
params.write_params(constant.TEST_DATA_FILE)

# Call the model, pipe output to file, read output file
file_output = open(constant.TEST_OUTPUT_FILE, "w")
completed_run = subprocess.run([constant.command], stdout = file_output, shell = True)
df_output = pd.read_csv(constant.TEST_OUTPUT_FILE, comment = "#", sep = ",")

# Call the model
model = utils.get_model_swig( params )
for time in range(params.get_param("end_time")):
model.one_time_step()

np.testing.assert_equal(df_output["n_death"].sum(), 0)
np.testing.assert_equal(
model.one_time_step_results()["n_death"],
0)

def test_total_infectious_rate_zero(self):
"""
Set infectious rate to zero results in only "n_seed_infection" as total_infected
"""
params = ParameterSet(constant.TEST_DATA_FILE, line_number = 1)
params = utils.get_params_swig()
params.set_param("n_total", 10000)
params.set_param("infectious_rate", 0.0)
params.write_params(constant.TEST_DATA_FILE)

# Call the model, pipe output to file, read output file
file_output = open(constant.TEST_OUTPUT_FILE, "w")
completed_run = subprocess.run([constant.command], stdout = file_output, shell = True)

df_output = pd.read_csv(constant.TEST_OUTPUT_FILE, comment = "#", sep = ",")

output = df_output["total_infected"].iloc[-1]
expected_output = int(params.get_param("n_seed_infection"))

np.testing.assert_equal(output, expected_output)
# Call the model
model = utils.get_model_swig( params )
for time in range(params.get_param("end_time")):
model.one_time_step()

np.testing.assert_equal(
model.one_time_step_results()["total_infected"],
int(params.get_param("n_seed_infection")))

def test_zero_infected(self):
"""
Set seed-cases to zero should result in zero sum of output column
"""
params = ParameterSet(constant.TEST_DATA_FILE, line_number = 1)
params = utils.get_params_swig()
params.set_param("n_seed_infection", 0)
params.write_params(constant.TEST_DATA_FILE)
params.set_param("n_total", 10000)
params.set_param("end_time", 100)

# Call the model, pipe output to file, read output file
file_output = open(constant.TEST_OUTPUT_FILE, "w")
completed_run = subprocess.run([constant.command], stdout = file_output, shell = True)
df_output = pd.read_csv(constant.TEST_OUTPUT_FILE, comment = "#", sep = ",")
# Call the model
model = utils.get_model_swig( params )

for _ in range(params.get_param("end_time")):
model.one_time_step()

np.testing.assert_equal(df_output["total_infected"].sum(), 0)
np.testing.assert_equal(model.one_time_step_results()["total_infected"], 0)

def test_disease_transition_times( self, test_params ):
"""
Expand All @@ -779,21 +780,22 @@ def test_disease_transition_times( self, test_params ):
"""
std_error_limit = 4

params = ParameterSet(constant.TEST_DATA_FILE, line_number=1)
params = utils.get_params_swig()
params = utils.turn_off_interventions(params, 50)
params.set_param("n_total", 50000)
params.set_param("n_seed_infection", 200)
params.set_param("end_time", 30)
params.set_param("infectious_rate", 6.0)
params.set_param("hospital_on", 0) #turning off hospital as this affects disease transitions
params.set_param( test_params )
for param, value in test_params.items():
params.set_param( param, value )

params.write_params(constant.TEST_DATA_FILE)
file_output = open(constant.TEST_OUTPUT_FILE, "w")
completed_run = subprocess.run([constant.command], stdout=file_output, shell=True)
df_indiv = pd.read_csv(
constant.TEST_INDIVIDUAL_FILE, comment="#", sep=",", skipinitialspace=True
)
model = utils.get_model_swig( params )
for _ in range(params.get_param("end_time")):
model.one_time_step()

model.write_transmissions()
model.write_individual_file()

df_trans = pd.read_csv(constant.TEST_TRANSMISSION_FILE)
df_indiv = pd.read_csv(constant.TEST_INDIVIDUAL_FILE)
Expand Down Expand Up @@ -953,7 +955,7 @@ def test_disease_outcome_proportions( self, test_params ):
"""
std_error_limit = 5

params = ParameterSet(constant.TEST_DATA_FILE, line_number=1)
params = utils.get_params_swig()
params = utils.turn_off_interventions(params, 50)

params.set_param("n_total", 20000)
Expand All @@ -962,7 +964,8 @@ def test_disease_outcome_proportions( self, test_params ):
params.set_param("infectious_rate", 4.0)
params.set_param("mild_infectious_factor", 1.0)
params.set_param("hospital_on", 0)
params.set_param( test_params )
for param, value in test_params.items():
params.set_param( param, value )

fraction_asymptomatic = [
test_params[ "fraction_asymptomatic_0_9" ],
Expand Down Expand Up @@ -1024,12 +1027,12 @@ def test_disease_outcome_proportions( self, test_params ):
test_params[ "fatality_fraction_80" ],
]

params.write_params(constant.TEST_DATA_FILE)
file_output = open(constant.TEST_OUTPUT_FILE, "w")
completed_run = subprocess.run([constant.command], stdout=file_output, shell=True)
df_indiv = pd.read_csv(
constant.TEST_INDIVIDUAL_FILE, comment="#", sep=",", skipinitialspace=True
)
model = utils.get_model_swig( params )
for _ in range(params.get_param("end_time")):
model.one_time_step()

model.write_transmissions()
model.write_individual_file()

df_trans = pd.read_csv(constant.TEST_TRANSMISSION_FILE)
df_indiv = pd.read_csv(constant.TEST_INDIVIDUAL_FILE)
Expand Down Expand Up @@ -1720,6 +1723,3 @@ def test_multi_strain_infectious_factor(self, test_params):
np.testing.assert_( sum( ( df_trans["strain_idx"] == 2 ) & ( df_trans["symptom_type"] == 0 ) ) > 1000, msg = "insufficient asymptomatic infectors with strain 2")
np.testing.assert_( sum( ( df_trans["strain_idx"] == 2 ) & ( df_trans["symptom_type"] == 1 ) ) == 0, msg = "mild infector with strain 2")
np.testing.assert_( sum( ( df_trans["strain_idx"] == 2 ) & ( df_trans["symptom_type"] == 2 ) ) == 0, msg = "severe infectorswith strain 2")



Loading