Create a process

This part of the tutorial is an introduction to the implementation of a new process. Anything that can happen to an individual should be implemented as a process. Processes can therefore be used to model, for example, the natural history of a disease. But also the demographic processes (birth and death) and some risk factors (smoking behavior) are implemented by subclassing the Process class.

In this tutorial, we will construct a Disease class and use this as a running example. The age of this disease onset will be drawn from a normal distribution. A fixed amount of time after the onset of the disease, a transition to the clinical state will be made. The clinical state will last for a fixed amount of time, after which an individual will die from the disease. We will log the age of onset, transition to a clinical state and death. The time spent in the clinical state will also be included in the simulation results.

Note

The main aim of this tutorial is to explain the main features of the MISCore to develop a process. However, showing how to develop an efficient process is absolutely not one of the aims. Please see the Optimize performance page for some hints on developing processes that run fast.

Subclass the Process class

One should implement a process by subclassing the Process class. Let’s import this class from MISCore.

1from miscore import Process

Next, create a class with Process as superclass. We’ll call the process Disease. The process as such is valid and could already be included in a model. However, it would not do anything.

1from miscore import Process
2
3
4class Disease(Process):
5    pass

Note

In Python, pass simply tells the interpreter to do nothing.

Note

The __init__ method of a Python class is called when creating an instance of that class. Learn how Python classes work by going through this tutorial in the Python documentation: https://docs.python.org/3/tutorial/classes.html.

Add parameters

In most cases, we need some process specific parameters. The value of the parameters can be initialized per instance of the subclass and tell the process how to behave. We can do so by defining the __init__() method.

Let’s add the parameters onset_age_mean and onset_age_sd to describe the normal distribution that determines the onset age. Furthermore, the parameter onset_to_clinical will determine how many years after onset an individual will transition to the clinical state. Finally, clinical_to_death determines how long an individual will stay in the clinical state before dying.

 1from miscore import Process
 2
 3
 4class Disease(Process):
 5    def __init__(self, onset_age_mean, onset_age_sd, onset_to_clinical,
 6                 clinical_to_death):
 7        # Store the parameters
 8        self.onset_age_mean = onset_age_mean
 9        self.onset_age_sd = onset_age_sd
10        self.onset_to_clinical = onset_to_clinical
11        self.clinical_to_death = clinical_to_death

To actually use the process in a model, we can initialize it with parameters as follows.

disease = Disease(onset_age_mean=90, onset_age_sd=25, onset_to_clinical=10, clinical_to_death=5)

The normal distribution will have a mean of 90 and a standard deviation of 25. After onset, it will take 10 years before an individual reaches the clinical state. Then, the individual will die after another 5 years.

Name

Each Process has a name, which is None by default. We’ll give the user the opportunity to pass another name by including name in the parameters. Note that we set this argument to have ‘disease’ as the default value. Therefore, it is not required to pass a name when creating a Disease object.

 1from miscore import Process
 2
 3
 4class Disease(Process):
 5    def __init__(self, onset_age_mean, onset_age_sd, onset_to_clinical,
 6                 clinical_to_death, name="disease"):
 7        # Store the parameters
 8        self.onset_age_mean = onset_age_mean
 9        self.onset_age_sd = onset_age_sd
10        self.onset_to_clinical = onset_to_clinical
11        self.clinical_to_death = clinical_to_death
12        self.name = name

Properties

Now let’s actually use the parameters to draw onset ages from a normal distribution. This should be done in the properties() method. In our case, we call the properties() method with the following parameters: a NumPy random number generator (numpy.random.Generator), the number of individuals for which properties should be calculated and the properties already defined by other processes. MISCore expects this method to return a dictionary with the new properties. Other methods (also in other processes) can access each property using the keys in this dictionary (disease_onset_age in this example).

 1from miscore import Process
 2
 3
 4class Disease(Process):
 5    def __init__(self, onset_age_mean, onset_age_sd, onset_to_clinical,
 6                 clinical_to_death, name="disease"):
 7        # Store the parameters
 8        self.onset_age_mean = onset_age_mean
 9        self.onset_age_sd = onset_age_sd
10        self.onset_to_clinical = onset_to_clinical
11        self.clinical_to_death = clinical_to_death
12        self.name = name
13
14    def properties(self, rng, n, properties):
15        # Draw the age of onset for each individual
16        onset_age = rng.normal(self.onset_age_mean, self.onset_age_sd, size=n)
17        onset_age[onset_age < 0] = 0
18
19        # Return the calculated properties
20        return {
21            "disease_onset_age": onset_age,
22        }

Note that we draw the onset ages for all n individuals in the simulation. An Individual instance is available during the simulation and can be used to retrieve the current individual’s onset age by calling its properties() method with the key disease_onset_age:

age = individual.properties("disease_onset_age")

Note

To prevent duplicate property names between processes, it is good practice to prefix each name with (an abbreviation of) your process name. This also holds for messages and logged events and durations, which will also be introduced in this tutorial.

Note

The __birth__ and __stratum__ property keys are special cases. They are used in the simulation as the birth date and stratum of each individual.

Note

Properties returned by this method can be included in the simulation results (i.e. in an instance of Result). Pass a collection of property names as return_properties to the run() method of Model to include those properties in the returned Result. Pass return_properties=True to include all generated properties.

Temporary Properties

The temporary properties method properties_tmp() is a more efficient way to store individual properties. All properties that are not required to be part of the output should be written in this method. You can use the properties_tmp() method in the same way as the regular properties() method to draw and retrieve an individual’s temporary properties.

Note

Within the disease process, temporary properties can be retrieved through the properties_tmp() method. Temporary properties cannot be included in the simulation results.

The event queue

MISCore simulates an individual’s life by passing through a list of events scheduled in the event queue. Processes can schedule events at certain ages and respond to them. How a process responds to an event is defined in ‘callback functions’, which are mapped to event messages in the callbacks dictionary.

To signal that a new life has just begun, we use the event with the message __start__. This allows processes to initialize.

At the start of an individual’s life, our process needs to schedule an age of onset (as drawn in the properties() method). We will define this response in a new method of our process: schedule_onset. To tell MISCore this method needs to be called at the beginning of each life, override the callbacks attribute of Process and add an entry with "__start__" as key and self.schedule_onset as value.

In our schedule_onset method, we use the add_to_queue_age() method to add the individual’s age of disease onset to the queue of events with the message disease_onset. As explained under properties, we can retrieve the onset age by calling the properties() method of the Individual object.

 1from miscore import Process
 2
 3
 4class Disease(Process):
 5    def __init__(self, onset_age_mean, onset_age_sd, onset_to_clinical,
 6                 clinical_to_death, name="disease"):
 7        # Store the parameters
 8        self.onset_age_mean = onset_age_mean
 9        self.onset_age_sd = onset_age_sd
10        self.onset_to_clinical = onset_to_clinical
11        self.clinical_to_death = clinical_to_death
12        self.name = name
13
14        # Define how the process responds to messages
15        self.callbacks = {
16            "__start__": self.schedule_onset,
17        }
18
19    def properties(self, rng, n, properties):
20        # Draw the age of onset for each individual
21        onset_age = rng.normal(self.onset_age_mean, self.onset_age_sd, size=n)
22        onset_age[onset_age < 0] = 0
23
24        # Return the calculated properties
25        return {
26            "disease_onset_age": onset_age,
27        }
28
29    def schedule_onset(self, individual):
30        # Schedule onset event
31        individual.add_to_queue_age(
32            age=individual.properties("disease_onset_age"),
33            message="disease_onset"
34        )

If you have two events that occur at the same age or year, you can further specify the order in which they occur with the priority argument of add_to_queue(), add_to_queue_age() or add_to_queue_year(). priority can be any int value, higher values are taken from the queue first. The default value of priority is 0.

Log events

We’ll now create a method called onset to enable the process to respond to the disease_onset message that we scheduled previously. Note that events in the event queue are not by default logged to the events DataFrame. We will therefore tell MISCore to log that the disease started for the current individual by using the log_event() method of the passed Individual instance. In this way, all onsets (per age and year group) will be included in the result. We’ll also schedule a new event at the age at which the individual should transition to the clinical state.

Like before, add an entry to the callbacks attribute, linking the message disease_onset to the method self.onset.

You should gather all event tags the process can log in the event_tags attribute. If an event is missing in this list, the model will automatically raise an exception. Although the exception can be circumvented by setting the verify_tags parameter to False in the run() function, it is highly recommended to gather all event tags in the event_tags attribute for code clarity.

 1from miscore import Process
 2
 3
 4class Disease(Process):
 5    # All event tags that can be logged by this process
 6    event_tags = [
 7        "disease_onset",
 8    ]
 9
10    def __init__(self, onset_age_mean, onset_age_sd, onset_to_clinical,
11                 clinical_to_death, name="disease"):
12        # Store the parameters
13        self.onset_age_mean = onset_age_mean
14        self.onset_age_sd = onset_age_sd
15        self.onset_to_clinical = onset_to_clinical
16        self.clinical_to_death = clinical_to_death
17        self.name = name
18
19        # Define how the process responds to messages
20        self.callbacks = {
21            "__start__": self.schedule_onset,
22            "disease_onset": self.onset,
23        }
24
25    def properties(self, rng, n, properties):
26        # Draw the age of onset for each individual
27        onset_age = rng.normal(self.onset_age_mean, self.onset_age_sd, size=n)
28        onset_age[onset_age < 0] = 0
29
30        # Return the calculated properties
31        return {
32            "disease_onset_age": onset_age,
33        }
34
35    def schedule_onset(self, individual):
36        # Schedule onset event
37        individual.add_to_queue_age(
38            age=individual.properties("disease_onset_age"),
39            message="disease_onset"
40        )
41
42    def onset(self, individual):
43        # Log that the disease has started
44        individual.log_event("disease_onset")
45
46        # Schedule transition to clinical state
47        individual.add_to_queue(
48            delta=self.onset_to_clinical,
49            message="disease_clinical"
50        )

Simulating death

As soon as the clinical state is reached, we should schedule a death. If terminate=True is passed to add_to_queue() (or add_to_queue_age()), the simulation will terminate after the time specified (self.clinical_to_death in this case). Also, MISCore will log the provided tag. Thus, we should add this tag to the event_tags list.

Note that we also log the transition to the clinical state.

 1from miscore import Process
 2
 3
 4class Disease(Process):
 5    # All event tags that can be logged by this process
 6    event_tags = [
 7        "disease_onset",
 8        "disease_clinical",
 9        "disease_death",
10    ]
11
12    def __init__(self, onset_age_mean, onset_age_sd, onset_to_clinical,
13                 clinical_to_death, name="disease"):
14        # Store the parameters
15        self.onset_age_mean = onset_age_mean
16        self.onset_age_sd = onset_age_sd
17        self.onset_to_clinical = onset_to_clinical
18        self.clinical_to_death = clinical_to_death
19        self.name = name
20
21        # Define how the process responds to messages
22        self.callbacks = {
23            "__start__": self.schedule_onset,
24            "disease_onset": self.onset,
25            "disease_clinical": self.clinical,
26        }
27
28    def properties(self, rng, n, properties):
29        # Draw the age of onset for each individual
30        onset_age = rng.normal(self.onset_age_mean, self.onset_age_sd, size=n)
31        onset_age[onset_age < 0] = 0
32
33        # Return the calculated properties
34        return {
35            "disease_onset_age": onset_age,
36        }
37
38    def schedule_onset(self, individual):
39        # Schedule onset event
40        individual.add_to_queue_age(
41            age=individual.properties("disease_onset_age"),
42            message="disease_onset"
43        )
44
45    def onset(self, individual):
46        # Log that the disease has started
47        individual.log_event("disease_onset")
48
49        # Schedule transition to clinical state
50        individual.add_to_queue(
51            delta=self.onset_to_clinical,
52            message="disease_clinical"
53        )
54
55    def clinical(self, individual):
56        # Log the transition to the clinical state
57        individual.log_event("disease_clinical")
58
59        # Schedule death event
60        individual.add_to_queue(
61            delta=self.clinical_to_death,
62            message="disease_death",
63            terminate=True
64        )

Memory

You might want to remember certain values during the simulation, so you can use them at a later age. Each instance of Individual has a memory attribute: a dictionary that is emptied before each new life. In this example, we’ll save the current age in the clinical method as disease_clinical_age.

Note that the current age can be accessed using the age attribute of the Individual instance.

 1from miscore import Process
 2
 3
 4class Disease(Process):
 5    # All event tags that can be logged by this process
 6    event_tags = [
 7        "disease_onset",
 8        "disease_clinical",
 9        "disease_death",
10    ]
11
12    def __init__(self, onset_age_mean, onset_age_sd, onset_to_clinical,
13                 clinical_to_death, name="disease"):
14        # Store the parameters
15        self.onset_age_mean = onset_age_mean
16        self.onset_age_sd = onset_age_sd
17        self.onset_to_clinical = onset_to_clinical
18        self.clinical_to_death = clinical_to_death
19        self.name = name
20
21        # Define how the process responds to messages
22        self.callbacks = {
23            "__start__": self.schedule_onset,
24            "disease_onset": self.onset,
25            "disease_clinical": self.clinical,
26        }
27
28    def properties(self, rng, n, properties):
29        # Draw the age of onset for each individual
30        onset_age = rng.normal(self.onset_age_mean, self.onset_age_sd, size=n)
31        onset_age[onset_age < 0] = 0
32
33        # Return the calculated properties
34        return {
35            "disease_onset_age": onset_age,
36        }
37
38    def schedule_onset(self, individual):
39        # Schedule onset event
40        individual.add_to_queue_age(
41            age=individual.properties("disease_onset_age"),
42            message="disease_onset"
43        )
44
45    def onset(self, individual):
46        # Log that the disease has started
47        individual.log_event("disease_onset")
48
49        # Schedule transition to clinical state
50        individual.add_to_queue(
51            delta=self.onset_to_clinical,
52            message="disease_clinical"
53        )
54
55    def clinical(self, individual):
56        # Log the transition to the clinical state
57        individual.log_event("disease_clinical")
58
59        # Save the current age in the memory
60        individual.memory["disease_clinical_age"] = individual.age
61
62        # Schedule death event
63        individual.add_to_queue(
64            delta=self.clinical_to_death,
65            message="disease_death",
66            terminate=True
67        )

Log durations

Besides events, MISCore also allows us to log durations. In this example, it might be relevant to know how much time people spent in the clinical state. We can only know for sure how long an individual was clinical after its death. The __end__ message signals the simulation of the current individual was terminated. Let’s respond to this message with a new method log_durations. To know at which age this individual transitioned to the clinical state, we need the disease_clinical_age value that we previously stored in the memory. The log_duration() method should be used to log any duration. Let’s call this duration disease_clinical, let it start at disease_clinical_age and let it end the current age individual.age (i.e. the age of death).

The tags of each duration that can be logged by the process should be gathered in the duration_tags attribute.

When defining a callback for the __end__ message, MISCore calls this method not only with the Individual object, but also with the message that terminated the simulation. Therefore, we should also add the argument message to our method.

 1from miscore import Process
 2
 3
 4class Disease(Process):
 5    # All event tags that can be logged by this process
 6    event_tags = [
 7        "disease_onset",
 8        "disease_clinical",
 9        "disease_death",
10    ]
11
12    # All duration tags that can be logged by this process
13    duration_tags = [
14        "disease_clinical",
15    ]
16
17    def __init__(self, onset_age_mean, onset_age_sd, onset_to_clinical,
18                 clinical_to_death, name="disease"):
19        # Store the parameters
20        self.onset_age_mean = onset_age_mean
21        self.onset_age_sd = onset_age_sd
22        self.onset_to_clinical = onset_to_clinical
23        self.clinical_to_death = clinical_to_death
24        self.name = name
25
26        # Define how the process responds to messages
27        self.callbacks = {
28            "__start__": self.schedule_onset,
29            "disease_onset": self.onset,
30            "disease_clinical": self.clinical,
31            "__end__": self.log_durations,
32        }
33
34    def properties(self, rng, n, properties):
35        # Draw the age of onset for each individual
36        onset_age = rng.normal(self.onset_age_mean, self.onset_age_sd, size=n)
37        onset_age[onset_age < 0] = 0
38
39        # Return the calculated properties
40        return {
41            "disease_onset_age": onset_age,
42        }
43
44    def schedule_onset(self, individual):
45        # Schedule onset event
46        individual.add_to_queue_age(
47            age=individual.properties("disease_onset_age"),
48            message="disease_onset"
49        )
50
51    def onset(self, individual):
52        # Log that the disease has started
53        individual.log_event("disease_onset")
54
55        # Schedule transition to clinical state
56        individual.add_to_queue(
57            delta=self.onset_to_clinical,
58            message="disease_clinical"
59        )
60
61    def clinical(self, individual):
62        # Log the transition to the clinical state
63        individual.log_event("disease_clinical")
64
65        # Save the current age in the memory
66        individual.memory["disease_clinical_age"] = individual.age
67
68        # Schedule death event
69        individual.add_to_queue(
70            delta=self.clinical_to_death,
71            message="disease_death",
72            terminate=True
73        )
74
75    def log_durations(self, individual, message):
76        # Log durations only if individual reached the clinical state
77        if "disease_clinical_age" in individual.memory:
78            # Retrieve age at which this person reached the clinical state
79            clinical_age = individual.memory["disease_clinical_age"]
80
81            # Log the time that this individual was in the clinical state
82            individual.log_duration(
83                tag="disease_clinical",
84                start_age=clinical_age,
85                end_age=individual.age
86            )