Skip to content

SerializedPopulation

Class to load and manipulate a saved population.

SerializedPopulation

Opens the passed file and reads in all the nodes.

Parameters:

Name Type Description Default
file str

serialized population file

 required

Examples:

Create an instance of SerializedPopulation::

1
2
import emod_api.serialization.SerializedPopulation as SerPop
ser_pop = SerPop.SerializedPopulation('state-00001.dtk')
Source code in emod_api/serialization/SerializedPopulation.py 
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
class SerializedPopulation:
    """Opens the passed file and reads in all the nodes.

    Args:
        file: serialized population file

    Examples:
        Create an instance of SerializedPopulation::

            import emod_api.serialization.SerializedPopulation as SerPop
            ser_pop = SerPop.SerializedPopulation('state-00001.dtk')

     """

    def __init__(self, file: str):
        self.next_infection_suid = None
        self.next_infection_suid_initialized = False
        self.dtk = dft.read(file)
        self._nodes = [n for n in self.dtk.nodes]

    @property
    def nodes(self):
        """All nodes.

        Examples:
            Delete number_of_ind individuals from node 0::

                node = ser_pop.nodes[0]
                del node.individualHumans[0:number_of_ind]

            Only keep individuals with a certain condition::

                node.individualHumans = [ind for ind in node.individualHumans if keep_fct(ind)]

            Change susceptibility of an individual::

                print(node.individualHumans[0].susceptibility)
                new_susceptibility = {"age": 101.01, "mod_acquire": 0}
                node.individualHumans[0].susceptibility.update(new_susceptibility)

            Copy individual[0] from node 0, change properties and add individual as new individual::

                import copy
                individual_properties={"m_age": 1234}
                individual = copy.deepcopy(node.individualHumans[0])
                individual["suid"] = ser_pop.get_next_individual_suid(0)
                individual.update(individual_properties)
                ser_pop.nodes[0].individualHumans.append(individual)

            Infect an individual with an infection copied from another individual::

                infection = node["individualHumans"][0]["infections"][0]
                infection["suid"] = self.get_next_infection_suid()
                node["individualHumans"][1]["infections"].append(infection)
                node["individualHumans"][1].m_is_infected = True

        """
        return self._nodes

    def flush(self):
        """Save all made changes to the node(s)."""
        for idx in range(len(self._nodes)):
            self.dtk.nodes[idx] = self._nodes[idx]

    def write(self, output_file: str = "my_sp_file.dtk"):
        """Write the population to a file.

        Args:     
            output_file: output file
        """
        self.flush()
        sim = self.dtk.simulation
        sim["infectionSuidGenerator"]["next_suid"] = self.get_next_infection_suid()
        self.dtk.simulation = sim

        print(f"Saving file {output_file}.")
        dft.write(self.dtk, output_file)

    def get_next_infection_suid(self):
        """Each infection needs a unique identifier, this function returns one."""
        sim = self.dtk.simulation
        if not self.next_infection_suid_initialized:
            self.next_infection_suid = sim["infectionSuidGenerator"]["next_suid"]
            self.next_infection_suid_initialized = True
        else:
            self.next_infection_suid["id"] = (
                self.next_infection_suid["id"]
                + sim["infectionSuidGenerator"]["numtasks"]
            )

        return dict(self.next_infection_suid)

    def get_next_individual_suid(self, node_id: int) -> dict:
        """Each individual needs a unique identifier, this function returns one.

        Args:
            node_id: The first parameter.

        Returns:
            The return value. True for success, False otherwise.

        Examples:
            To get a unique id for an individual::

                print(sp.get_next_individual_suid(0))
                {'id': 2}
        """
        suid = self._nodes[node_id]["m_IndividualHumanSuidGenerator"]["next_suid"]
        self._nodes[node_id]["m_IndividualHumanSuidGenerator"]["id"] = (
            suid["id"]
            + self._nodes[node_id]["m_IndividualHumanSuidGenerator"]["numtasks"]
        )
        return dict(suid)

nodes property

All nodes.

Examples:

Delete number_of_ind individuals from node 0::

1
2
node = ser_pop.nodes[0]
del node.individualHumans[0:number_of_ind]

Only keep individuals with a certain condition::

1
node.individualHumans = [ind for ind in node.individualHumans if keep_fct(ind)]

Change susceptibility of an individual::

1
2
3
print(node.individualHumans[0].susceptibility)
new_susceptibility = {"age": 101.01, "mod_acquire": 0}
node.individualHumans[0].susceptibility.update(new_susceptibility)

Copy individual[0] from node 0, change properties and add individual as new individual::

1
2
3
4
5
6
import copy
individual_properties={"m_age": 1234}
individual = copy.deepcopy(node.individualHumans[0])
individual["suid"] = ser_pop.get_next_individual_suid(0)
individual.update(individual_properties)
ser_pop.nodes[0].individualHumans.append(individual)

Infect an individual with an infection copied from another individual::

1
2
3
4
infection = node["individualHumans"][0]["infections"][0]
infection["suid"] = self.get_next_infection_suid()
node["individualHumans"][1]["infections"].append(infection)
node["individualHumans"][1].m_is_infected = True

flush()

Save all made changes to the node(s).

Source code in emod_api/serialization/SerializedPopulation.py 
70
71
72
73
def flush(self):
    """Save all made changes to the node(s)."""
    for idx in range(len(self._nodes)):
        self.dtk.nodes[idx] = self._nodes[idx]

get_next_individual_suid(node_id)

Each individual needs a unique identifier, this function returns one.

Parameters:

Name Type Description Default
node_id int

The first parameter.

 required

Returns:

Type Description
dict

The return value. True for success, False otherwise.

Examples:

To get a unique id for an individual::

1
2
print(sp.get_next_individual_suid(0))
{'id': 2}
Source code in emod_api/serialization/SerializedPopulation.py 
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
def get_next_individual_suid(self, node_id: int) -> dict:
    """Each individual needs a unique identifier, this function returns one.

    Args:
        node_id: The first parameter.

    Returns:
        The return value. True for success, False otherwise.

    Examples:
        To get a unique id for an individual::

            print(sp.get_next_individual_suid(0))
            {'id': 2}
    """
    suid = self._nodes[node_id]["m_IndividualHumanSuidGenerator"]["next_suid"]
    self._nodes[node_id]["m_IndividualHumanSuidGenerator"]["id"] = (
        suid["id"]
        + self._nodes[node_id]["m_IndividualHumanSuidGenerator"]["numtasks"]
    )
    return dict(suid)

get_next_infection_suid()

Each infection needs a unique identifier, this function returns one.

Source code in emod_api/serialization/SerializedPopulation.py 
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
def get_next_infection_suid(self):
    """Each infection needs a unique identifier, this function returns one."""
    sim = self.dtk.simulation
    if not self.next_infection_suid_initialized:
        self.next_infection_suid = sim["infectionSuidGenerator"]["next_suid"]
        self.next_infection_suid_initialized = True
    else:
        self.next_infection_suid["id"] = (
            self.next_infection_suid["id"]
            + sim["infectionSuidGenerator"]["numtasks"]
        )

    return dict(self.next_infection_suid)

write(output_file='my_sp_file.dtk')

Write the population to a file.

Parameters:

Name Type Description Default
output_file str

output file

 'my_sp_file.dtk'
Source code in emod_api/serialization/SerializedPopulation.py 
75
76
77
78
79
80
81
82
83
84
85
86
87
def write(self, output_file: str = "my_sp_file.dtk"):
    """Write the population to a file.

    Args:     
        output_file: output file
    """
    self.flush()
    sim = self.dtk.simulation
    sim["infectionSuidGenerator"]["next_suid"] = self.get_next_infection_suid()
    self.dtk.simulation = sim

    print(f"Saving file {output_file}.")
    dft.write(self.dtk, output_file)

find(name, handle, currentlevel='dtk.nodes')

Recursively searches for a paramters that matches or is close to name and prints out where to find it in the file.

Parameters:

Name Type Description Default
name str

the paramter you are looking for e.g. "age", "gender".

 required
handle Union[str, Iterable]

some iterable data structure, can be a list of nodes, a node, list of individuals, etc

 required
currentlevel str

just a string to print out where the found item is located e.g. "dtk.nodes" or "dtk.node.individuals"

 'dtk.nodes'

Examples:

What is the exact paramteter name used for the age of an individual?::

1
2
3
4
5
6
7
SerPop.find("age", node)
...
1998   Found in:  dtk.nodes.individualHumans[999].m_age
1999   Found in:  dtk.nodes.individualHumans[999].susceptibility.age
2000   Found in:  dtk.nodes.m_vectorpopulations[0].EggQueues[0].age
2001   Found in:  dtk.nodes.m_vectorpopulations[0].EggQueues[1].age
...
Source code in emod_api/serialization/SerializedPopulation.py 
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
def find(name: str,
         handle: Union[str, Iterable],
         currentlevel: str = "dtk.nodes"):
    """Recursively searches for a paramters that matches or is close to name and prints out where to find it in the file.

    Args:     
        name: the paramter you are looking for e.g. "age", "gender".
        handle: some iterable data structure, can be a list of
                nodes, a node, list of individuals, etc
        currentlevel: just a string to print out where the found item
                is located e.g. "dtk.nodes" or "dtk.node.individuals"

    Examples:
        What is the exact paramteter name used for the age of an individual?::

            SerPop.find("age", node)
            ...
            1998   Found in:  dtk.nodes.individualHumans[999].m_age
            1999   Found in:  dtk.nodes.individualHumans[999].susceptibility.age
            2000   Found in:  dtk.nodes.m_vectorpopulations[0].EggQueues[0].age
            2001   Found in:  dtk.nodes.m_vectorpopulations[0].EggQueues[1].age
            ...

    """
    global COUNTER
    if isinstance(handle, str) and difflib.get_close_matches(name, [handle], cutoff=0.6):
        print(COUNTER, "  Found in: ", currentlevel)
        COUNTER += 1
        return

    if isinstance(handle, str) or not isinstance(handle, collections.Iterable):
        return

    # key can be a string or on dict/list/..
    for idx, key in enumerate(handle):
        level = (
            currentlevel + "." + key
            if isinstance(key, str)
            else currentlevel + "[" + str(idx) + "]"
        )
        try:
            tmp = handle[key]
            if isinstance(tmp, collections.Iterable):
                find(name, key, level + "[]")
            else:
                find(name, key, level)
        except BaseException:
            # list or keys of a dict, works in all cases but misses objects in
            # dicts
            find(name, key, level)
        if isinstance(handle, dict):
            find(name, handle[key], level)  # check if string is key for a dict

get_parameters(handle, currentlevel='dtk.nodes')

Return a set of all parameters in the serialized population file. Helpful to get an overview about what is in the serialized population file.

Parameters:

Name Type Description Default
handle Union[str, Iterable]

some iterable data structure, can be a list of nodes, a node, list of individuals, etc

required
currentlevel str

just a string to print out where the found item is located e.g. "dtk.nodes" or "dtk.node.individuals

 'dtk.nodes'

Examples:

Print all parameters in serialized population file::

1
2
for n in sorted(SerPop.get_parameters(node)):
    print(n)
Source code in emod_api/serialization/SerializedPopulation.py 
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
def get_parameters(handle: Union[str, Iterable],
                   currentlevel: str = "dtk.nodes"):
    """Return a set of all parameters in the serialized population file. Helpful to get an overview about what is in the serialized population file.

    Args:
        handle: some iterable data structure, can be a list of
                nodes, a node, list of individuals, etc     
        currentlevel: just a string to print out where the found item
                is located e.g. "dtk.nodes" or "dtk.node.individuals

    Examples:
        Print all parameters in serialized population file::

            for n in sorted(SerPop.get_parameters(node)):
                print(n)
    """
    global COUNTER
    param = set()

    if isinstance(handle, str):
        param.add(currentlevel)
        return param

    if not isinstance(handle, collections.Iterable):
        return param

    for _, d in enumerate(handle):
        level = currentlevel + " " + d if isinstance(d, str) else currentlevel
        param.update(get_parameters(d, level))
        if isinstance(handle, dict):
            param.update(get_parameters(handle[d], level))

    return param