build-py-surrogatemodel:

if: github.event.pull_request.draft == false

runs-on: ubuntu-latest

container:

image: quay.io/pypa/manylinux2014_x86_64

steps:

- uses: actions/checkout@v2

- uses: ./.github/actions/build-py

with:

package: surrogatemodel

test-py-surrogatemodel:

if: github.event.pull_request.draft == false

needs: [build-py-surrogatemodel, build-py-simulation]

runs-on: ubuntu-latest

steps:

- uses: actions/checkout@v2

- uses: ./.github/actions/test-py

with:

package: surrogatemodel

build-py-surrogatemodel:

runs-on: ubuntu-latest

container:

image: quay.io/pypa/manylinux2014_x86_64

steps:

- uses: actions/checkout@v2

- uses: ./.github/actions/build-py

with:

package: surrogatemodel

test-py-surrogatemodel:

needs: [build-py-surrogatemodel, build-py-simulation]

runs-on: ubuntu-latest

steps:

- uses: actions/checkout@v2

- uses: ./.github/actions/test-py

with:

package: surrogatemodel

MEmilio Surrogate Model Package

This package contains machine learning based surrogate models that make predictions based on the MEmilio simulation models. Currently there are only surrogate models for ODE models. These simulations of these equation-based models are used for data generation. The goal is to create a powerful tool that predicts the dynamics faster than a simulation of an expert model, e.g., a metapopulation or agent-based model while still having acceptable errors with respect to the original simulations.

Use the provided `setup.py` script install the package.

To install the package, use the command (from the directory containing `setup.py`)

pip install .

For developement of code use

pip install -e .[dev]

```

Since we are running simulations to generate the data, the MEmilio `memilio-simulation` package (https://github.com/DLR-SC/memilio/tree/main/pycode/memilio-simulation) also needs to be installed.

The package currently provides the following modules:

- `models`: models for different specific tasks

Currently we have the following models:

- `ode_secir_simple`: A simple model allowing for asymptomatic as well as symptomatic infection not stratified by age groups.

Each model folder contains the following files:

- `data_generation`: data generated from expert model simulation.

- `model`: training and evaluation of the model.

- `network_architectures`: multiple network architectures are saved in this file.

- `tests`: this file contains all tests

The package provides a test suite in `memilio/surrogatemodel_test`. To run the tests, simply run the following command.

python -m unittest

__path__ = __import__('pkgutil').extend_path(__path__, __name__)

This model is an very simplified application of the SECIR model implemented in https://github.com/DLR-SC/memilio/tree/main/cpp/models/ode_secir/ not stratified by age groups.

The example is based on https://github.com/DLR-SC/memilio/tree/main/pycode/examples/simulation/secir_simple.py which uses python bindings to run the underlying C++ code.

import copy

import os

import pickle

import random

from datetime import date

import numpy as np

import pandas as pd

import tensorflow as tf

from progress.bar import Bar

from sklearn.preprocessing import FunctionTransformer

from memilio.simulation import (ContactMatrix, Damping, LogLevel,

UncertainContactMatrix, set_log_level)

from memilio.simulation.secir import (AgeGroup, Index_InfectionState,

InfectionState, Model, Simulation,

interpolate_simulation_result, simulate)

def run_secir_simulation(days):

"""! Uses an ODE SECIR model allowing for asymptomatic infection. The model is not stratified by region or demographic properties such as age.

set_log_level(LogLevel.Off)

populations = [50_000]

start_day = 1

start_month = 1

start_year = 2019

dt = 0.1

num_groups = 1

# Initialize Parameters

model = Model(1)

A0 = AgeGroup(0)

# Set parameters

# Compartment transition duration

model.parameters.IncubationTime[A0] = 5.2

model.parameters.TimeInfectedSymptoms[A0] = 6.

model.parameters.SerialInterval[A0] = 4.2

model.parameters.TimeInfectedSevere[A0] = 12.

model.parameters.TimeInfectedCritical[A0] = 8.

# Initial number of people in each compartment with random numbers

model.populations[A0, Index_InfectionState(

InfectionState.Exposed)] = 60 * random.uniform(0.2, 1)

model.populations[A0, Index_InfectionState(

InfectionState.InfectedNoSymptoms)] = 55 * random.uniform(0.2, 1)

model.populations[A0, Index_InfectionState(

InfectionState.InfectedSymptoms)] = 50 * random.uniform(0.2, 1)

model.populations[A0, Index_InfectionState(

InfectionState.InfectedSevere)] = 12 * random.uniform(0.2, 1)

model.populations[A0, Index_InfectionState(

InfectionState.InfectedCritical)] = 3 * random.uniform(0.2, 1)

model.populations[A0, Index_InfectionState(

InfectionState.Recovered)] = 50 * random.random()

model.populations[A0, Index_InfectionState(InfectionState.Dead)] = 0

model.populations.set_difference_from_total(

(A0, Index_InfectionState(InfectionState.Susceptible)), populations[0])

# Compartment transition propabilities

model.parameters.RelativeTransmissionNoSymptoms[A0] = 0.5

model.parameters.TransmissionProbabilityOnContact[A0] = 0.1

model.parameters.RecoveredPerInfectedNoSymptoms[A0] = 0.09

model.parameters.RiskOfInfectionFromSymptomatic[A0] = 0.25

model.parameters.SeverePerInfectedSymptoms[A0] = 0.2

model.parameters.CriticalPerSevere[A0] = 0.25

model.parameters.DeathsPerCritical[A0] = 0.3

# twice the value of RiskOfInfectionFromSymptomatic

model.parameters.MaxRiskOfInfectionFromSymptomatic[A0] = 0.5

model.parameters.StartDay = (

date(start_year, start_month, start_day) - date(start_year, 1, 1)).days

model.parameters.ContactPatterns.cont_freq_mat[0].baseline = np.ones(

(num_groups, num_groups)) * 10

model.parameters.ContactPatterns.cont_freq_mat[0].minimum = np.ones(

(num_groups, num_groups)) * 0

# Apply mathematical constraints to parameters

model.apply_constraints()

# Run Simulation

result = simulate(0, days, dt, model)

# Interpolate simulation result on days time scale

result = interpolate_simulation_result(result)

# Using an array instead of a list to avoid problems with references

result_array = result.as_ndarray()

dataset = []

# Omit first column, as the time points are not of interest here.

dataset_entries = copy.deepcopy(result_array[1:, :].transpose())

return dataset_entries.tolist()

def generate_data(

num_runs, path, input_width, label_width, normalize=True,

save_data=True):

"""! Generate data sets of num_runs many equation-based model simulations and transforms the computed results by a log(1+x) transformation.

data = {

"inputs": [],

"labels": []

}

# The number of days is the same as the sum of input and label width.

# Since the first day of the input is day 0, we still need to subtract 1.

days = input_width + label_width - 1

# show progess in terminal for longer runs

# Due to the random structure, theres currently no need to shuffle the data

bar = Bar('Number of Runs done', max=num_runs)

for _ in range(0, num_runs):

data_run = run_secir_simulation(days)

data['inputs'].append(data_run[:input_width])

data['labels'].append(data_run[input_width:])

bar.next()

bar.finish()

if normalize:

# logarithmic normalization

transformer = FunctionTransformer(np.log1p, validate=True)

inputs = np.asarray(data['inputs']).transpose(2, 0, 1).reshape(8, -1)

scaled_inputs = transformer.transform(inputs)

scaled_inputs = scaled_inputs.transpose().reshape(num_runs, input_width, 8)

scaled_inputs_list = scaled_inputs.tolist()

labels = np.asarray(data['labels']).transpose(2, 0, 1).reshape(8, -1)

scaled_labels = transformer.transform(labels)

scaled_labels = scaled_labels.transpose().reshape(num_runs, label_width, 8)

scaled_labels_list = scaled_labels.tolist()

# cast dfs to tensors

data['inputs'] = tf.stack(scaled_inputs_list)

data['labels'] = tf.stack(scaled_labels_list)

if save_data:

# check if data directory exists. If necessary, create it.

if not os.path.isdir(path):

os.mkdir(path)

# save dict to json file

with open(os.path.join(path, 'data_secir_simple.pickle'), 'wb') as f:

pickle.dump(data, f)

return data

if __name__ == "__main__":

# Store data relative to current file two levels higher.

path = os.path.dirname(os.path.realpath(__file__))

path_data = os.path.join(os.path.dirname(os.path.realpath(

os.path.dirname(os.path.realpath(path)))), 'data')

input_width = 5

label_width = 30

num_runs = 1000

data = generate_data(num_runs, path_data, input_width,

label_width)