Skip to content

malaria_node

MalariaNode: malaria-aware extension of emod-api's Node.

MalariaNode

Bases: Node

Node subclass that supports malaria-specific demographics distributions.

Adds risk and innate-immune distribution storage, serialization, and deserialization. These distributions are simple-only (flag + value1 + value2) and are written into the IndividualAttributes section of the demographics JSON.

Source code in emodpy_malaria/demographics/malaria_node.py
class MalariaNode(Node):
    """Node subclass that supports malaria-specific demographics distributions.

    Adds risk and innate-immune distribution storage, serialization, and
    deserialization.  These distributions are simple-only (flag + value1 +
    value2) and are written into the ``IndividualAttributes`` section of
    the demographics JSON.
    """

    def _set_risk_simple_distribution(self, flag: int, value1: float, value2: float):
        """Called dynamically by Demographics._set_distribution(use_case='risk') via getattr."""
        self.individual_attributes.risk_distribution_flag = flag
        self.individual_attributes.risk_distribution1 = value1
        self.individual_attributes.risk_distribution2 = value2

    def _set_innate_immune_simple_distribution(self, flag: int, value1: float, value2: float):
        """Called dynamically by Demographics._set_distribution(use_case='innate_immune') via getattr."""
        self.individual_attributes.innate_immune_distribution_flag = flag
        self.individual_attributes.innate_immune_distribution1 = value1
        self.individual_attributes.innate_immune_distribution2 = value2

    def to_dict(self) -> dict:
        d = super().to_dict()
        ia = d.get("IndividualAttributes", {})

        risk_flag = getattr(self.individual_attributes, "risk_distribution_flag", None)
        if risk_flag is not None:
            ia["RiskDistributionFlag"] = risk_flag
            ia["RiskDistribution1"] = self.individual_attributes.risk_distribution1
            ia["RiskDistribution2"] = self.individual_attributes.risk_distribution2 if self.individual_attributes.risk_distribution2 is not None else 0
            d["IndividualAttributes"] = ia

        innate_flag = getattr(self.individual_attributes, "innate_immune_distribution_flag", None)
        if innate_flag is not None:
            ia["InnateImmuneDistributionFlag"] = innate_flag
            ia["InnateImmuneDistribution1"] = self.individual_attributes.innate_immune_distribution1
            ia["InnateImmuneDistribution2"] = self.individual_attributes.innate_immune_distribution2 if self.individual_attributes.innate_immune_distribution2 is not None else 0
            d["IndividualAttributes"] = ia

        return d

    @classmethod
    def from_data(cls, data: dict) -> Tuple["MalariaNode", List[Callable]]:
        node, implicit_functions = super().from_data(data)
        ia_dict = data.get("IndividualAttributes") or {}

        risk_flag = ia_dict.get("RiskDistributionFlag")
        if risk_flag is not None:
            node.individual_attributes.risk_distribution_flag = risk_flag
            node.individual_attributes.risk_distribution1 = ia_dict.get("RiskDistribution1")
            node.individual_attributes.risk_distribution2 = ia_dict.get("RiskDistribution2")
            implicit_functions.append(_set_enable_demog_risk)

        innate_flag = ia_dict.get("InnateImmuneDistributionFlag")
        if innate_flag is not None:
            node.individual_attributes.innate_immune_distribution_flag = innate_flag
            node.individual_attributes.innate_immune_distribution1 = ia_dict.get("InnateImmuneDistribution1")
            node.individual_attributes.innate_immune_distribution2 = ia_dict.get("InnateImmuneDistribution2")
            warnings.warn(
                "InnateImmuneDistribution loaded from file. Pyrogenic vs. cytokine-killing vs. NONE "
                "is unknown. Config may need updating to ensure Innate_Immune_Variation_Type is set properly.",
                Warning,
                stacklevel=2,
            )

        return node, implicit_functions