Skip to content

serialization

Malaria-specific utilities for working with EMOD .dtk serialized population files.

Builds on top of emod_api's SerializedPopulation with malaria-aware inspection, modification, and export capabilities.

Example::

from emodpy_malaria.serialization import (
    MalariaSerializedPopulation, zero_infections, count_humans
)

population = MalariaSerializedPopulation("state-00100.dtk")
print(population.summary())
print(count_humans(population.ser_pop))
zero_infections(population.ser_pop)
population.write("state-00100-zeroed.dtk")

Genome

Represents a single parasite genome with barcode-to-DTK-dict conversion.

Parameters:

Name Type Description Default
barcode str

Nucleotide string (characters A/C/G/T).

 required
allele_root_id int

Root ID for allele tracking. Typically the individual's SUID or -999 for vectors.

 required
Source code in emodpy_malaria/serialization/_genomes.py 
 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
124
125
126
127
128
129
130
131
132
133
class Genome:
    """Represents a single parasite genome with barcode-to-DTK-dict conversion.

    Args:
        barcode (str): Nucleotide string (characters A/C/G/T).
        allele_root_id (int): Root ID for allele tracking. Typically the
            individual's SUID or -999 for vectors.
    """

    def __init__(self, barcode: str, allele_root_id: int) -> None:
        self._barcode = barcode
        self._nucleotides: list[int] = []
        self._allele_roots: list[int] = []
        self._hash_code = np.int32(_HASH_SEED)
        self._barcode_hash_code = np.int32(_HASH_SEED)

        with warnings.catch_warnings():
            warnings.simplefilter("ignore", RuntimeWarning)
            for ch in barcode:
                val = self.nucleotide_to_int(ch)
                self._nucleotides.append(val)
                self._allele_roots.append(allele_root_id)
                self._barcode_hash_code = np.int32(
                    _HASH_MULTIPLIER * self._barcode_hash_code + val
                )
                self._hash_code = np.int32(
                    _HASH_MULTIPLIER * self._hash_code + val
                )
                self._hash_code = np.int32(
                    _HASH_MULTIPLIER * self._hash_code + allele_root_id
                )

    @property
    def barcode(self) -> str:
        return self._barcode

    @property
    def hashcode(self) -> int:
        return int(self._hash_code)

    @property
    def barcode_hashcode(self) -> int:
        return int(self._barcode_hash_code)

    @property
    def nucleotides(self) -> list[int]:
        return list(self._nucleotides)

    @property
    def allele_roots(self) -> list[int]:
        return list(self._allele_roots)

    def to_dtk_dict(self) -> dict:
        """Convert to DTK genome dict format (``m_pInner`` structure)."""
        return {
            "m_pInner": {
                "__class__": "ParasiteGenomeInner",
                "m_HashCode": self.hashcode,
                "m_BarcodeHashcode": self.barcode_hashcode,
                "m_NucleotideSequence": self._nucleotides,
                "m_AlleleRoots": self._allele_roots,
            }
        }

    def to_dtk_map_entry(self) -> dict:
        """Convert to DTK genome map entry format (key/value pair)."""
        return {
            "key": self.hashcode,
            "value": self.to_dtk_dict()["m_pInner"],
        }

    @staticmethod
    def from_dtk_dict(dtk_dict: dict) -> Genome:
        """Construct a Genome from a DTK genome dict.

        Args:
            dtk_dict (dict): A dict with ``m_pInner`` key containing genome data.

        Returns:
            Genome instance with matching barcode and hash codes.
        """
        inner = dtk_dict["m_pInner"]
        nucleotides = inner["m_NucleotideSequence"]
        allele_roots = inner["m_AlleleRoots"]

        barcode = "".join(Genome.int_to_nucleotide(n) for n in nucleotides)
        allele_root_id = allele_roots[0] if allele_roots else 0

        genome = Genome(barcode, allele_root_id)
        return genome

    @staticmethod
    def nucleotide_to_int(ch: str) -> int:
        """Convert a single nucleotide character to its integer encoding."""
        if ch not in _NUCLEOTIDE_MAP:
            raise ValueError(
                f"Unknown nucleotide character {ch!r}. "
                f"Valid: {list(_NUCLEOTIDE_MAP.keys())}"
            )
        return _NUCLEOTIDE_MAP[ch]

    @staticmethod
    def int_to_nucleotide(val: int) -> str:
        """Convert an integer encoding back to a nucleotide character."""
        if val not in _NUCLEOTIDE_REVERSE:
            raise ValueError(
                f"Unknown nucleotide value {val!r}. Valid: {list(_NUCLEOTIDE_REVERSE.keys())}"
            )
        return _NUCLEOTIDE_REVERSE[val]

    def __repr__(self) -> str:
        return f"Genome(barcode={self._barcode!r}, hashcode={self.hashcode})"

from_dtk_dict(dtk_dict) staticmethod

Construct a Genome from a DTK genome dict.

Parameters:

Name Type Description Default
dtk_dict dict

A dict with m_pInner key containing genome data.

 required

Returns:

Type Description
Genome

Genome instance with matching barcode and hash codes.
Source code in emodpy_malaria/serialization/_genomes.py 
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
@staticmethod
def from_dtk_dict(dtk_dict: dict) -> Genome:
    """Construct a Genome from a DTK genome dict.

    Args:
        dtk_dict (dict): A dict with ``m_pInner`` key containing genome data.

    Returns:
        Genome instance with matching barcode and hash codes.
    """
    inner = dtk_dict["m_pInner"]
    nucleotides = inner["m_NucleotideSequence"]
    allele_roots = inner["m_AlleleRoots"]

    barcode = "".join(Genome.int_to_nucleotide(n) for n in nucleotides)
    allele_root_id = allele_roots[0] if allele_roots else 0

    genome = Genome(barcode, allele_root_id)
    return genome

int_to_nucleotide(val) staticmethod

Convert an integer encoding back to a nucleotide character.

Source code in emodpy_malaria/serialization/_genomes.py 
123
124
125
126
127
128
129
130
@staticmethod
def int_to_nucleotide(val: int) -> str:
    """Convert an integer encoding back to a nucleotide character."""
    if val not in _NUCLEOTIDE_REVERSE:
        raise ValueError(
            f"Unknown nucleotide value {val!r}. Valid: {list(_NUCLEOTIDE_REVERSE.keys())}"
        )
    return _NUCLEOTIDE_REVERSE[val]

nucleotide_to_int(ch) staticmethod

Convert a single nucleotide character to its integer encoding.

Source code in emodpy_malaria/serialization/_genomes.py 
113
114
115
116
117
118
119
120
121
@staticmethod
def nucleotide_to_int(ch: str) -> int:
    """Convert a single nucleotide character to its integer encoding."""
    if ch not in _NUCLEOTIDE_MAP:
        raise ValueError(
            f"Unknown nucleotide character {ch!r}. "
            f"Valid: {list(_NUCLEOTIDE_MAP.keys())}"
        )
    return _NUCLEOTIDE_MAP[ch]

to_dtk_dict()

Convert to DTK genome dict format (m_pInner structure).

Source code in emodpy_malaria/serialization/_genomes.py 
74
75
76
77
78
79
80
81
82
83
84
def to_dtk_dict(self) -> dict:
    """Convert to DTK genome dict format (``m_pInner`` structure)."""
    return {
        "m_pInner": {
            "__class__": "ParasiteGenomeInner",
            "m_HashCode": self.hashcode,
            "m_BarcodeHashcode": self.barcode_hashcode,
            "m_NucleotideSequence": self._nucleotides,
            "m_AlleleRoots": self._allele_roots,
        }
    }

to_dtk_map_entry()

Convert to DTK genome map entry format (key/value pair).

Source code in emodpy_malaria/serialization/_genomes.py 
86
87
88
89
90
91
def to_dtk_map_entry(self) -> dict:
    """Convert to DTK genome map entry format (key/value pair)."""
    return {
        "key": self.hashcode,
        "value": self.to_dtk_dict()["m_pInner"],
    }

MalariaSerializedPopulation

Malaria-aware wrapper around emod_api's SerializedPopulation.

Provides malaria-specific convenience methods for inspection, modification, and export of .dtk serialized population files. Delegates to the underlying SerializedPopulation for all low-level file I/O.

Parameters:

Name Type Description Default
file_path str | Path

Path to a .dtk serialized population file.

 required
Source code in emodpy_malaria/serialization/_population.py 
 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
class MalariaSerializedPopulation:
    """Malaria-aware wrapper around emod_api's SerializedPopulation.

    Provides malaria-specific convenience methods for inspection,
    modification, and export of ``.dtk`` serialized population files.
    Delegates to the underlying SerializedPopulation for all low-level
    file I/O.

    Args:
        file_path (str | Path): Path to a ``.dtk`` serialized population file.
    """

    def __init__(self, file_path: str | Path) -> None:
        self._file_path = Path(file_path)
        self._ser_pop = SerializedPopulation(str(self._file_path))

    @property
    def ser_pop(self) -> SerializedPopulation:
        """The underlying emod_api SerializedPopulation object."""
        return self._ser_pop

    @property
    def file_path(self) -> Path:
        """Original file path."""
        return self._file_path

    @property
    def nodes(self):
        """All nodes (delegated to SerializedPopulation.nodes)."""
        return self._ser_pop.nodes

    @property
    def simulation(self):
        """Simulation-level data dict."""
        return self._ser_pop.dtk.simulation

    @property
    def header(self):
        """File header metadata."""
        return self._ser_pop.dtk.header

    @property
    def version(self) -> int:
        """DTK file format version."""
        return self._ser_pop.dtk.header.version

    @property
    def num_nodes(self) -> int:
        """Number of nodes in the file."""
        return len(self._ser_pop.nodes)

    def get_next_infection_suid(self) -> dict:
        """Get a unique SUID for a new infection."""
        return self._ser_pop.get_next_infection_suid()

    def get_next_individual_suid(self, node_id: int) -> dict:
        """Get a unique SUID for a new individual in the given node."""
        return self._ser_pop.get_next_individual_suid(node_id)

    def write(self, output_file: str | Path = "my_sp_file.dtk") -> None:
        """Write the (possibly modified) population to a file.

        Args:
            output_file (str | Path): Destination file path. Parent directories are
                created if they do not exist.
        """
        output_path = Path(output_file)
        output_path.parent.mkdir(parents=True, exist_ok=True)
        self._ser_pop.write(str(output_path))

    # -- Inspection --

    def summary(self) -> dict:
        """Return a summary dict with node counts, human counts, etc."""
        from emodpy_malaria.serialization._inspect import summarize
        return summarize(self._ser_pop)

    def find_parameter(self, name: str) -> list[str]:
        """Search for a parameter by name (fuzzy) and return matching paths."""
        from emodpy_malaria.serialization._inspect import find_parameter
        return find_parameter(self._ser_pop, name)

    def __repr__(self) -> str:
        return (
            f"MalariaSerializedPopulation("
            f"file='{self._file_path.name}', "
            f"version={self.version}, "
            f"nodes={self.num_nodes})"
        )

file_path property

Original file path.

header property

File header metadata.

nodes property

All nodes (delegated to SerializedPopulation.nodes).

num_nodes property

Number of nodes in the file.

ser_pop property

The underlying emod_api SerializedPopulation object.

simulation property

Simulation-level data dict.

version property

DTK file format version.

find_parameter(name)

Search for a parameter by name (fuzzy) and return matching paths.

Source code in emodpy_malaria/serialization/_population.py 
89
90
91
92
def find_parameter(self, name: str) -> list[str]:
    """Search for a parameter by name (fuzzy) and return matching paths."""
    from emodpy_malaria.serialization._inspect import find_parameter
    return find_parameter(self._ser_pop, name)

get_next_individual_suid(node_id)

Get a unique SUID for a new individual in the given node.

Source code in emodpy_malaria/serialization/_population.py 
67
68
69
def get_next_individual_suid(self, node_id: int) -> dict:
    """Get a unique SUID for a new individual in the given node."""
    return self._ser_pop.get_next_individual_suid(node_id)

get_next_infection_suid()

Get a unique SUID for a new infection.

Source code in emodpy_malaria/serialization/_population.py 
63
64
65
def get_next_infection_suid(self) -> dict:
    """Get a unique SUID for a new infection."""
    return self._ser_pop.get_next_infection_suid()

summary()

Return a summary dict with node counts, human counts, etc.

Source code in emodpy_malaria/serialization/_population.py 
84
85
86
87
def summary(self) -> dict:
    """Return a summary dict with node counts, human counts, etc."""
    from emodpy_malaria.serialization._inspect import summarize
    return summarize(self._ser_pop)

write(output_file='my_sp_file.dtk')

Write the (possibly modified) population to a file.

Parameters:

Name Type Description Default
output_file str | Path

Destination file path. Parent directories are created if they do not exist.

 'my_sp_file.dtk'
Source code in emodpy_malaria/serialization/_population.py 
71
72
73
74
75
76
77
78
79
80
def write(self, output_file: str | Path = "my_sp_file.dtk") -> None:
    """Write the (possibly modified) population to a file.

    Args:
        output_file (str | Path): Destination file path. Parent directories are
            created if they do not exist.
    """
    output_path = Path(output_file)
    output_path.parent.mkdir(parents=True, exist_ok=True)
    self._ser_pop.write(str(output_path))

count_humans(ser_pop, *, node_index=None)

Count individuals across all nodes or in a specific node.

Parameters:

Name Type Description Default
ser_pop SerializedPopulation

A loaded SerializedPopulation.

 required
node_index int

If provided, count only in this node (0-based index).

 None

Returns:

Type Description
int

Number of individuals.
Source code in emodpy_malaria/serialization/_inspect.py 
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
def count_humans(
    ser_pop: SerializedPopulation,
    *,
    node_index: int | None = None,
) -> int:
    """Count individuals across all nodes or in a specific node.

    Args:
        ser_pop (SerializedPopulation): A loaded SerializedPopulation.
        node_index (int ): If provided, count only in this node (0-based index).

    Returns:
        Number of individuals.
    """
    if node_index is not None:
        return len(ser_pop.nodes[node_index].individualHumans)

    return sum(
        len(ser_pop.nodes[i].individualHumans)
        for i in range(len(ser_pop.nodes))
    )

count_infections(ser_pop, *, node_index=None)

Count total infections across all nodes or in a specific node.

Parameters:

Name Type Description Default
ser_pop SerializedPopulation

A loaded SerializedPopulation.

 required
node_index int

If provided, count only in this node.

 None

Returns:

Type Description
int

Number of infections.
Source code in emodpy_malaria/serialization/_inspect.py 
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
def count_infections(
    ser_pop: SerializedPopulation,
    *,
    node_index: int | None = None,
) -> int:
    """Count total infections across all nodes or in a specific node.

    Args:
        ser_pop (SerializedPopulation): A loaded SerializedPopulation.
        node_index (int ): If provided, count only in this node.

    Returns:
        Number of infections.
    """
    total = 0
    indices = [node_index] if node_index is not None else range(len(ser_pop.nodes))

    for idx in indices:
        node = ser_pop.nodes[idx]
        for human in node.individualHumans:
            total += len(human["infections"])

    return total

count_vectors(ser_pop, *, node_index=None, queue=None)

Count vector cohorts across all nodes or in a specific node.

Parameters:

Name Type Description Default
ser_pop SerializedPopulation

A loaded SerializedPopulation.

 required
node_index int

If provided, count only in this node.

 None
queue str

If provided, count only in this queue (e.g., "AdultQueues"). If None, count across all queues.

 None

Returns:

Type Description
int

Number of vector cohorts.
Source code in emodpy_malaria/serialization/_inspect.py 
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
179
def count_vectors(
    ser_pop: SerializedPopulation,
    *,
    node_index: int | None = None,
    queue: str | None = None,
) -> int:
    """Count vector cohorts across all nodes or in a specific node.

    Args:
        ser_pop (SerializedPopulation): A loaded SerializedPopulation.
        node_index (int ): If provided, count only in this node.
        queue (str ): If provided, count only in this queue (e.g.,
            ``"AdultQueues"``). If None, count across all queues.

    Returns:
        Number of vector cohorts.
    """
    from emodpy_malaria.serialization._infections import INFECTION_QUEUES

    queues_to_check = (queue,) if queue else INFECTION_QUEUES
    total = 0
    indices = [node_index] if node_index is not None else range(len(ser_pop.nodes))

    for idx in indices:
        node = ser_pop.nodes[idx]
        for vp in node.m_vectorpopulations:
            for q in queues_to_check:
                if q in vp:
                    total += len(vp[q]["collection"])

    return total

count_vectors_by_state(ser_pop, *, node_index=None)

Count vector cohorts grouped by species and state.

Parameters:

Name Type Description Default
ser_pop SerializedPopulation

A loaded SerializedPopulation.

 required
node_index int

If given, count only this node. Otherwise sums all nodes.

 None

Returns:

Type Description
dict[str, dict[str, int]]

Nested dict: {species_name: {state_name: count, ...}, ...}.
Source code in emodpy_malaria/serialization/_vectors.py 
 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
def count_vectors_by_state(
    ser_pop: SerializedPopulation,
    *,
    node_index: int | None = None,
) -> dict[str, dict[str, int]]:
    """Count vector cohorts grouped by species and state.

    Args:
        ser_pop (SerializedPopulation): A loaded SerializedPopulation.
        node_index (int ): If given, count only this node. Otherwise sums all nodes.

    Returns:
        Nested dict: ``{species_name: {state_name: count, ...}, ...}``.
    """
    result: dict[str, dict[str, int]] = {}

    for _, node in _iter_nodes(ser_pop, node_index):
        for vp in node.m_vectorpopulations:
            species = str(vp.get("m_species_id", vp.get("Species", "unknown")))
            if species not in result:
                result[species] = {name: 0 for name in VECTOR_STATE_NAMES.values()}

            for queue_name in _LIFECYCLE_QUEUES:
                if queue_name not in vp:
                    continue
                queue_data = vp[queue_name]
                collection = queue_data.get("collection", queue_data) if isinstance(queue_data, dict) else queue_data
                if not hasattr(collection, '__len__'):
                    continue
                for cohort in collection:
                    state = cohort.get("state", None) if isinstance(cohort, dict) else getattr(cohort, "state", None)
                    if state is not None and state in VECTOR_STATE_NAMES:
                        result[species][VECTOR_STATE_NAMES[state]] += 1

    return result

export_humans_to_json(ser_pop, output_file)

Export human data to a JSON file, grouped by node.

Each key in the output JSON is "Node <external_id>" and the value is the list of individual dicts from that node.

Parameters:

Name Type Description Default
ser_pop SerializedPopulation

A loaded SerializedPopulation.

 required
output_file str | Path

Destination JSON file path. Parent directories are created if they do not exist.

 required
Source code in emodpy_malaria/serialization/_export.py 
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
def export_humans_to_json(
    ser_pop: SerializedPopulation,
    output_file: str | Path,
) -> None:
    """Export human data to a JSON file, grouped by node.

    Each key in the output JSON is ``"Node <external_id>"`` and the value
    is the list of individual dicts from that node.

    Args:
        ser_pop (SerializedPopulation): A loaded SerializedPopulation.
        output_file (str | Path): Destination JSON file path. Parent directories are
            created if they do not exist.
    """
    output_path = Path(output_file)
    output_path.parent.mkdir(parents=True, exist_ok=True)

    human_data = {}
    for idx in range(len(ser_pop.nodes)):
        node = ser_pop.nodes[idx]
        human_data[f"Node {node.externalId}"] = node.individualHumans

    with open(output_path, "w") as f:
        json.dump(human_data, f)

    logger.info("Exported human data to %s", output_path)

find_parameter(ser_pop, name, *, cutoff=0.6)

Search for parameters matching the given name using fuzzy matching.

Improved version of emod_api's find() that returns results as a list of dot-path strings instead of printing them.

Parameters:

Name Type Description Default
ser_pop SerializedPopulation

A loaded SerializedPopulation.

 required
name str

Parameter name to search for (e.g., "age", "gender").

 required
cutoff float

Similarity threshold for fuzzy matching (0.0-1.0).

 0.6

Returns:

Type Description
list[str]

List of dot-notation paths where the parameter was found.
Source code in emodpy_malaria/serialization/_inspect.py 
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
def find_parameter(
    ser_pop: SerializedPopulation,
    name: str,
    *,
    cutoff: float = 0.6,
) -> list[str]:
    """Search for parameters matching the given name using fuzzy matching.

    Improved version of emod_api's ``find()`` that returns results as a list
    of dot-path strings instead of printing them.

    Args:
        ser_pop (SerializedPopulation): A loaded SerializedPopulation.
        name (str): Parameter name to search for (e.g., ``"age"``, ``"gender"``).
        cutoff (float): Similarity threshold for fuzzy matching (0.0-1.0).

    Returns:
        List of dot-notation paths where the parameter was found.
    """
    results: list[str] = []
    _find_recursive(name, ser_pop.nodes, "nodes", cutoff, results)
    return results

get_all_barcodes(ser_pop)

Extract all unique barcode strings from the population's genome map.

Parameters:

Name Type Description Default
ser_pop SerializedPopulation

A loaded SerializedPopulation.

 required

Returns:

Type Description
list[str]

List of unique barcode strings.
Source code in emodpy_malaria/serialization/_genomes.py 
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
def get_all_barcodes(ser_pop: SerializedPopulation) -> list[str]:
    """Extract all unique barcode strings from the population's genome map.

    Args:
        ser_pop (SerializedPopulation): A loaded SerializedPopulation.

    Returns:
        List of unique barcode strings.
    """
    genome_map = ser_pop.dtk.simulation["ParasiteGenetics"]["m_ParasiteGenomeMap"]
    barcodes = []
    for entry in genome_map:
        nucleotides = entry["value"]["m_NucleotideSequence"]
        barcode = "".join(Genome.int_to_nucleotide(n) for n in nucleotides)
        barcodes.append(barcode)
    return barcodes

get_all_parameters(ser_pop)

Return the set of all parameter paths in the serialized population.

Parameters:

Name Type Description Default
ser_pop SerializedPopulation

A loaded SerializedPopulation.

 required

Returns:

Type Description
set[str]

Set of dot-notation parameter paths.
Source code in emodpy_malaria/serialization/_inspect.py 
253
254
255
256
257
258
259
260
261
262
def get_all_parameters(ser_pop: SerializedPopulation) -> set[str]:
    """Return the set of all parameter paths in the serialized population.

    Args:
        ser_pop (SerializedPopulation): A loaded SerializedPopulation.

    Returns:
        Set of dot-notation parameter paths.
    """
    return _get_params_recursive(ser_pop.nodes, "nodes")

get_infection_barcodes(ser_pop, *, node_index=None)

Extract barcode information for each infection in the population.

Parameters:

Name Type Description Default
ser_pop SerializedPopulation

A loaded SerializedPopulation.

 required
node_index int

If provided, inspect only this node.

 None

Returns:

Type Description
list[dict]

List of dicts with keys node_id, individual_id,
list[dict]

infection_index, barcode, hashcode.
Source code in emodpy_malaria/serialization/_genomes.py 
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
def get_infection_barcodes(
    ser_pop: SerializedPopulation,
    *,
    node_index: int | None = None,
) -> list[dict]:
    """Extract barcode information for each infection in the population.

    Args:
        ser_pop (SerializedPopulation): A loaded SerializedPopulation.
        node_index (int ): If provided, inspect only this node.

    Returns:
        List of dicts with keys ``node_id``, ``individual_id``,
        ``infection_index``, ``barcode``, ``hashcode``.
    """
    results = []
    nodes = ser_pop.nodes
    indices = [node_index] if node_index is not None else range(len(nodes))

    for idx in indices:
        node = nodes[idx]
        node_id = node.externalId
        for person in node["individualHumans"]:
            person_id = person["suid"]["id"]
            for inf_idx, infection in enumerate(person["infections"]):
                inner = infection["infection_strain"]["m_Genome"]["m_pInner"]
                nucleotides = inner["m_NucleotideSequence"]
                barcode = "".join(Genome.int_to_nucleotide(n) for n in nucleotides)
                results.append({
                    "node_id": node_id,
                    "individual_id": person_id,
                    "infection_index": inf_idx,
                    "barcode": barcode,
                    "hashcode": inner["m_HashCode"],
                })

    return results

get_vector_infection_summary(ser_pop, *, node_index=None)

Summarize vector infection state across all species and queues.

Parameters:

Name Type Description Default
ser_pop SerializedPopulation

A loaded SerializedPopulation.

 required
node_index int

If given, summarize only this node.

 None

Returns:

Type Description
dict[str, Any]

Dict with total_cohorts, infected_cohorts,
dict[str, Any]

infectious_cohorts, total_oocyst_cohorts,
dict[str, Any]

total_sporozoite_cohorts, and a by_species breakdown.
Source code in emodpy_malaria/serialization/_vectors.py 
114
115
116
117
118
119
120
121
122
123
124
125
126
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
def get_vector_infection_summary(
    ser_pop: SerializedPopulation,
    *,
    node_index: int | None = None,
) -> dict[str, Any]:
    """Summarize vector infection state across all species and queues.

    Args:
        ser_pop (SerializedPopulation): A loaded SerializedPopulation.
        node_index (int ): If given, summarize only this node.

    Returns:
        Dict with ``total_cohorts``, ``infected_cohorts``,
        ``infectious_cohorts``, ``total_oocyst_cohorts``,
        ``total_sporozoite_cohorts``, and a ``by_species`` breakdown.
    """
    total_cohorts = 0
    infected_cohorts = 0
    infectious_cohorts = 0
    total_oocysts = 0
    total_sporozoites = 0
    by_species: dict[str, dict[str, int]] = {}

    for _, node in _iter_nodes(ser_pop, node_index):
        for vp in node.m_vectorpopulations:
            species = str(vp.get("m_species_id", vp.get("Species", "unknown")))
            if species not in by_species:
                by_species[species] = {
                    "adult_count": 0,
                    "infected_count": 0,
                    "infectious_count": 0,
                }

            for queue in INFECTION_QUEUES:
                if queue not in vp:
                    continue
                cohorts = vp[queue]["collection"]
                for cohort in cohorts:
                    total_cohorts += 1
                    state = cohort.state if hasattr(cohort, "state") else cohort.get("state")
                    if state == STATE_INFECTED:
                        infected_cohorts += 1
                        by_species[species]["infected_count"] += 1
                    elif state == STATE_INFECTIOUS:
                        infectious_cohorts += 1
                        by_species[species]["infectious_count"] += 1
                    elif state == STATE_ADULT:
                        by_species[species]["adult_count"] += 1

                    oocysts = cohort.get("m_OocystCohorts", []) if isinstance(cohort, dict) else getattr(cohort, "m_OocystCohorts", [])
                    sporozoites = cohort.get("m_SporozoiteCohorts", []) if isinstance(cohort, dict) else getattr(cohort, "m_SporozoiteCohorts", [])
                    total_oocysts += len(oocysts)
                    total_sporozoites += len(sporozoites)

    return {
        "total_cohorts": total_cohorts,
        "infected_cohorts": infected_cohorts,
        "infectious_cohorts": infectious_cohorts,
        "total_oocyst_cohorts": total_oocysts,
        "total_sporozoite_cohorts": total_sporozoites,
        "by_species": by_species,
    }

get_vector_species_names(ser_pop, *, node_index=None)

Return the names of vector species present in the population.

Parameters:

Name Type Description Default
ser_pop SerializedPopulation

A loaded SerializedPopulation.

 required
node_index int

If given, inspect only this node. Otherwise inspects the first node (species are typically identical across nodes).

 None

Returns:

Type Description
list[str]

List of species name strings.
Source code in emodpy_malaria/serialization/_vectors.py 
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
def get_vector_species_names(
    ser_pop: SerializedPopulation,
    *,
    node_index: int | None = None,
) -> list[str]:
    """Return the names of vector species present in the population.

    Args:
        ser_pop (SerializedPopulation): A loaded SerializedPopulation.
        node_index (int ): If given, inspect only this node. Otherwise inspects
            the first node (species are typically identical across nodes).

    Returns:
        List of species name strings.
    """
    idx = node_index if node_index is not None else 0
    node = ser_pop.nodes[idx]
    names = []
    for vp in node.m_vectorpopulations:
        name = vp.get("m_species_id", vp.get("Species", "unknown"))
        names.append(str(name))
    return names

list_node_ids(ser_pop)

Return the external IDs of all nodes.

Parameters:

Name Type Description Default
ser_pop SerializedPopulation

A loaded SerializedPopulation.

 required

Returns:

Type Description
list[int]

List of node external IDs.
Source code in emodpy_malaria/serialization/_inspect.py 
182
183
184
185
186
187
188
189
190
191
def list_node_ids(ser_pop: SerializedPopulation) -> list[int]:
    """Return the external IDs of all nodes.

    Args:
        ser_pop (SerializedPopulation): A loaded SerializedPopulation.

    Returns:
        List of node external IDs.
    """
    return [ser_pop.nodes[i].externalId for i in range(len(ser_pop.nodes))]

read_header(file_path)

Read only the header of a .dtk file without loading node data.

Useful for quickly checking file version, compression, node count, and EMOD build info without the cost of decompressing population data.

Parameters:

Name Type Description Default
file_path str | Path

Path to the .dtk file.

 required

Returns:

Type Description
dict

Header dict with keys like version, date, author,
dict

emod_info, compression info, and chunk metadata.
Source code in emodpy_malaria/serialization/_inspect.py 
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
def read_header(file_path: str | Path) -> dict:
    """Read only the header of a .dtk file without loading node data.

    Useful for quickly checking file version, compression, node count,
    and EMOD build info without the cost of decompressing population data.

    Args:
        file_path (str | Path): Path to the .dtk file.

    Returns:
        Header dict with keys like ``version``, ``date``, ``author``,
        ``emod_info``, compression info, and chunk metadata.
    """
    with open(file_path, "rb") as handle:
        magic = handle.read(4).decode()
        if magic != dft.IDTK:
            raise ValueError(f"File has incorrect magic number: {magic!r}")

        size_string = handle.read(12)
        header_size = int(size_string)
        header_text = handle.read(header_size)
        header_json = json.loads(header_text.decode())

        if "metadata" in header_json:
            header_json = header_json["metadata"]
        if "version" not in header_json:
            header_json["version"] = 1

    return header_json

replace_genomes(ser_pop, next_barcode_fn)

Replace all parasite genomes in humans and vectors (in-place).

Clears the simulation-level ParasiteGenomeMap and rebuilds it with new genomes generated by calling next_barcode_fn() for each infection.

Parameters:

Name Type Description Default
ser_pop SerializedPopulation

A loaded SerializedPopulation.

 required
next_barcode_fn Callable[[], str]

Callable returning a barcode string each time it is called.

 required

Returns:

Type Description
int

Total number of genomes replaced.

Raises:

Type Description
ValueError

If a generated barcode has a different length than the existing barcode at that position.
Source code in emodpy_malaria/serialization/_genomes.py 
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
179
180
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
214
215
216
217
218
219
220
221
222
223
224
225
226
227
def replace_genomes(
    ser_pop: SerializedPopulation,
    next_barcode_fn: Callable[[], str],
) -> int:
    """Replace all parasite genomes in humans and vectors (in-place).

    Clears the simulation-level ParasiteGenomeMap and rebuilds it with new
    genomes generated by calling ``next_barcode_fn()`` for each infection.

    Args:
        ser_pop (SerializedPopulation): A loaded SerializedPopulation.
        next_barcode_fn (Callable[[], str]): Callable returning a barcode string each time
            it is called.

    Returns:
        Total number of genomes replaced.

    Raises:
        ValueError: If a generated barcode has a different length than the
            existing barcode at that position.
    """
    if next_barcode_fn is None:
        raise ValueError("next_barcode_fn must not be None")

    genome_map = ser_pop.dtk.simulation["ParasiteGenetics"]["m_ParasiteGenomeMap"]
    genome_map.clear()
    cache: dict[str, Genome] = {}
    count = 0

    for node in ser_pop.nodes:
        for person in node["individualHumans"]:
            person_id = person["suid"]["id"]
            for infection in person["infections"]:
                existing = infection["infection_strain"]["m_Genome"]["m_pInner"]["m_NucleotideSequence"]
                new_genome = _get_next_genome(next_barcode_fn, person_id, genome_map, cache)
                if len(new_genome["m_pInner"]["m_NucleotideSequence"]) != len(existing):
                    raise ValueError(
                        f"New barcode length {len(new_genome['m_pInner']['m_NucleotideSequence'])} "
                        f"does not match existing length {len(existing)}"
                    )
                infection["infection_strain"]["m_Genome"] = new_genome
                count += 1

        for vector_pop in node["m_vectorpopulations"]:
            for vector in vector_pop["AdultQueues"]["collection"]:
                for oocyst in vector["m_OocystCohorts"]:
                    existing = oocyst["m_MaleGametocyteGenome"]["m_pInner"]["m_NucleotideSequence"]
                    new_genome = _get_next_genome(next_barcode_fn, -999, genome_map, cache)
                    if len(new_genome["m_pInner"]["m_NucleotideSequence"]) != len(existing):
                        raise ValueError("New barcode length does not match existing oocyst male genome length")
                    oocyst["m_MaleGametocyteGenome"] = new_genome

                    existing = oocyst["m_pStrainIdentity"]["m_Genome"]["m_pInner"]["m_NucleotideSequence"]
                    new_genome = _get_next_genome(next_barcode_fn, -999, genome_map, cache)
                    if len(new_genome["m_pInner"]["m_NucleotideSequence"]) != len(existing):
                        raise ValueError("New barcode length does not match existing oocyst female genome length")
                    oocyst["m_pStrainIdentity"]["m_Genome"] = new_genome
                    count += 2

                for sporo in vector["m_SporozoiteCohorts"]:
                    existing = sporo["m_MaleGametocyteGenome"]["m_pInner"]["m_NucleotideSequence"]
                    new_genome = _get_next_genome(next_barcode_fn, -999, genome_map, cache)
                    if len(new_genome["m_pInner"]["m_NucleotideSequence"]) != len(existing):
                        raise ValueError("New barcode length does not match existing sporozoite male genome length")
                    sporo["m_MaleGametocyteGenome"] = new_genome

                    existing = sporo["m_pStrainIdentity"]["m_Genome"]["m_pInner"]["m_NucleotideSequence"]
                    new_genome = _get_next_genome(next_barcode_fn, -999, genome_map, cache)
                    if len(new_genome["m_pInner"]["m_NucleotideSequence"]) != len(existing):
                        raise ValueError("New barcode length does not match existing sporozoite female genome length")
                    sporo["m_pStrainIdentity"]["m_Genome"] = new_genome
                    count += 2

    return count

summarize(ser_pop)

Return a comprehensive summary of the serialized population.

Parameters:

Name Type Description Default
ser_pop SerializedPopulation

A loaded SerializedPopulation.

 required

Returns:

Type Description
dict

Dict with num_nodes, total_humans, total_infections,
dict

and a nodes list with per-node details.
Source code in emodpy_malaria/serialization/_inspect.py 
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
def summarize(ser_pop: SerializedPopulation) -> dict:
    """Return a comprehensive summary of the serialized population.

    Args:
        ser_pop (SerializedPopulation): A loaded SerializedPopulation.

    Returns:
        Dict with ``num_nodes``, ``total_humans``, ``total_infections``,
        and a ``nodes`` list with per-node details.
    """
    nodes_info = []
    total_humans = 0
    total_infections = 0

    for idx in range(len(ser_pop.nodes)):
        node = ser_pop.nodes[idx]
        humans = node.individualHumans
        num_humans = len(humans)
        num_infections = sum(len(h["infections"]) for h in humans)
        num_infected = sum(1 for h in humans if h.get("m_is_infected", False))

        ages = [h.get("m_age", 0) for h in humans]
        mean_age = sum(ages) / len(ages) if ages else 0.0

        vector_pops = node.m_vectorpopulations
        species_names = []
        for vp in vector_pops:
            name = vp.get("m_species_id", vp.get("Species", "unknown"))
            species_names.append(str(name))

        nodes_info.append({
            "index": idx,
            "external_id": node.externalId,
            "num_humans": num_humans,
            "num_infections": num_infections,
            "num_infected_humans": num_infected,
            "mean_age_days": round(mean_age, 1),
            "num_vector_populations": len(vector_pops),
            "vector_species": species_names,
        })

        total_humans += num_humans
        total_infections += num_infections

    return {
        "num_nodes": len(ser_pop.nodes),
        "total_humans": total_humans,
        "total_infections": total_infections,
        "nodes": nodes_info,
    }

zero_human_infections(humans, *, keep_ids=None)

Reset infection state of individuals to uninfected.

Parameters:

Name Type Description Default
humans Any

Iterable of individual dicts (e.g., node.individualHumans).

 required
keep_ids list[int]

SUID IDs of individuals to skip.

 None

Returns:

Type Description
int

Number of individuals whose infections were zeroed.

Raises:

Type Description
KeyError

If an individual is missing expected infection fields.
Source code in emodpy_malaria/serialization/_infections.py 
 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
def zero_human_infections(
    humans: Any,
    *,
    keep_ids: list[int] | None = None,
) -> int:
    """Reset infection state of individuals to uninfected.

    Args:
        humans (Any): Iterable of individual dicts (e.g., ``node.individualHumans``).
        keep_ids (list[int] ): SUID IDs of individuals to skip.

    Returns:
        Number of individuals whose infections were zeroed.

    Raises:
        KeyError: If an individual is missing expected infection fields.
    """
    if keep_ids is None:
        keep_ids = []

    count = 0
    for person in humans:
        if person.suid.id in keep_ids:
            continue

        missing_keys = set(UNINFECTED_HUMAN_TEMPLATE) - set(person)
        if missing_keys:
            raise KeyError(
                f"Individual is missing expected infection fields: {missing_keys}"
            )
        person.update(UNINFECTED_HUMAN_TEMPLATE)
        count += 1

    return count

zero_infections(ser_pop, *, ignore_node_ids=None, keep_individual_ids=None, remove_vectors=False)

Zero all infections in the loaded population (in-place).

Resets human infection fields to uninfected state and either resets or removes infected vectors.

Parameters:

Name Type Description Default
ser_pop SerializedPopulation

A loaded SerializedPopulation.

 required
ignore_node_ids list[int]

Node external IDs to skip entirely.

 None
keep_individual_ids list[int]

Individual SUID IDs whose infections are preserved.

 None
remove_vectors bool

If True, remove infected/infectious vector cohorts. If False (default), reset their state to STATE_ADULT.

 False
Source code in emodpy_malaria/serialization/_infections.py 
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
def zero_infections(
    ser_pop: SerializedPopulation,
    *,
    ignore_node_ids: list[int] | None = None,
    keep_individual_ids: list[int] | None = None,
    remove_vectors: bool = False,
) -> None:
    """Zero all infections in the loaded population (in-place).

    Resets human infection fields to uninfected state and either resets
    or removes infected vectors.

    Args:
        ser_pop (SerializedPopulation): A loaded SerializedPopulation.
        ignore_node_ids (list[int] ): Node external IDs to skip entirely.
        keep_individual_ids (list[int] ): Individual SUID IDs whose infections are preserved.
        remove_vectors (bool): If True, remove infected/infectious vector cohorts.
            If False (default), reset their state to STATE_ADULT.
    """
    if ignore_node_ids is None:
        ignore_node_ids = []
    if keep_individual_ids is None:
        keep_individual_ids = []

    for node in ser_pop.nodes:
        if node.externalId in ignore_node_ids:
            logger.info("Skipping node %s", node.externalId)
            continue

        logger.info("Zeroing infections in node %s", node.externalId)
        zero_vector_infections(node.m_vectorpopulations, remove=remove_vectors)
        zero_human_infections(node.individualHumans, keep_ids=keep_individual_ids)

zero_vector_infections(vector_pop_list, *, remove=False)

Reset or remove infections from vector populations.

Parameters:

Name Type Description Default
vector_pop_list Any

List of vector populations (node.m_vectorpopulations).

 required
remove bool

If True, remove infected/infectious cohorts entirely. If False (default), reset to STATE_ADULT.

 False

Returns:

Type Description
int

Number of vector cohorts affected.
Source code in emodpy_malaria/serialization/_infections.py 
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
def zero_vector_infections(
    vector_pop_list: Any,
    *,
    remove: bool = False,
) -> int:
    """Reset or remove infections from vector populations.

    Args:
        vector_pop_list (Any): List of vector populations
            (``node.m_vectorpopulations``).
        remove (bool): If True, remove infected/infectious cohorts entirely.
            If False (default), reset to STATE_ADULT.

    Returns:
        Number of vector cohorts affected.
    """
    count = 0
    for idx, vector_population in enumerate(vector_pop_list):
        for queue in INFECTION_QUEUES:
            cohorts = vector_population[queue]["collection"]

            if remove:
                kept = [
                    c for c in cohorts
                    if c.state != STATE_INFECTED and c.state != STATE_INFECTIOUS
                ]
                removed = len(cohorts) - len(kept)
                vector_pop_list[idx][queue]["collection"] = kept
                count += removed
            else:
                for cohort in cohorts:
                    if cohort.state in (STATE_INFECTED, STATE_INFECTIOUS):
                        cohort.state = STATE_ADULT
                        cohort.progress = 0.0
                        cohort.m_pStrain = NullPtr()
                        count += 1

    return count