_inspect

Read-only inspection and query functions for serialized populations.

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

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

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

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

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

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_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

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")

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

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

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

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

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,
    }