_vectors

Inspection and modification of vector populations in serialized files.

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

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

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

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

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