Source code for orgmatt.metabolism.base_metabolic_model

# -*- coding: utf-8 -*-
# SPDX-FileCopyrightText: 2022-2023 Tanguy Fardet
# SPDX-License-Identifier: GPL-3.0-or-later
# orgmatt/metabolism/base_metabolic_model.py

from abc import ABC, abstractmethod
from typing import Literal

from .._utils.age import _group_from_age
from ..typing import NumericArrayLike, NumericOrArray
from ..units import Quantity, check_dim, ureg
from .corrections import correction_factor
from .nutrient_flows import (
    fraction_nutrient_excreta,
    fraction_nutrient_retention,
)


class BaseModel(ABC):
    def __init__(
        self,
        age_group: str,
        sex: str | None = None,
        diet: str | None = None,
        country: str | None = None,
        region: str | None = None,
    ):
        '''
        Initialize a metabolic model to estimate the intake and excretions of
        a person based on their age (group), sex, and diet.

        Parameters
        ----------
        age_group : str
            Precise age or age group (among "infant", "toddler", "kid",
            "teenager", "young adult", "adult", "senior").
        sex : str, optional (default: None)
            Sex of the modeled person.
        diet : str, optional (default: "omnivorous")
            Diet of the modeled person.
        country : str, optional (default: None)
            Country of the modeled person.
        region : str, optional (default: None)
            World region of the modeled person (among "Africa", "Asia",
            "Europe", "North America", "Oceania", "South America")
        '''
        self._kwargs: dict[str, str] = {}

        if age_group in _age_groups:
            self._group = _age_groups[age_group]
            self._age = _age_from_group[self._group]
        else:
            # age was provided, get associated group
            self._kwargs["age"] = age_group
            self._group = _group_from_age(age_group)
            self._age = age_group

        self._kwargs["group"] = self._group

        assert sex in _available_sexes, (
            f"Available options for `sex` are {_available_sexes}"
        )

        assert diet in _available_diets, (
            f"Available options for `diet` are {_available_diets}"
        )

        self._sex = sex
        self._diet = diet or "ommnivorous"
        self._country = country
        self._region = region
        self._last_correction_status: int = 0

        # store kwargs
        keys = ("sex", "diet", "country", "region")
        vals = (sex, diet, country, region)

        for k, v in zip(keys, vals):
            if v is not None:
                self._kwargs[k] = v

    @property
    def age(self) -> str:
        '''Return the age of the modeled person'''
        return self._age

    @property
    def group(self) -> str:
        '''Return the age group of the modeled person'''
        return self._group

    @property
    def diet(self) -> str | None:
        '''Return the diet of the modeled person'''
        return self._diet

    @property
    def country(self) -> str | None:
        '''Return the country of the modeled person'''
        return self._country

    @property
    def region(self) -> str | None:
        '''Return the region of the modeled person'''
        return self._region

    @check_dim(arglist=('duration', 1, '[time]'), result='[mass]*[X]')
    def nutrient_intake(
        self,
        duration: Quantity,
        nutrient: str,
        ci: int = 0,
        q: NumericArrayLike | None = None,
        ignore_missing: bool = True,
        silent: bool = False,
        **kwargs,
    ) -> Quantity:
        '''
        Return nutrient intake from food.

        Parameters
        ----------
        duration : float [time]
            Time interval over which the nitrogen is excreted.
        nutrient : str
            Nutrient of interest among ('N', 'K', 'P', 'Mg', 'Ca').
        ci : int (optional, default: 0)
            Confidence interval (CI). If non-zero, the function will also
            return the low and high expected values corresponding to that CI.
        q : array-like of floats (optional: default: None)
            Percentiles to compute (supersedes `ci`).
        ignore_missing : bool, optional (default: True)
            Whether to raise an error or not if the value was not available
            and could not be properly corrected.
        silent : bool, optional (default: False)
            Skip all errors and warnings (overrides `ignore_missing`).

        Returns
        -------
        intake: quantity
            Intake of `nutrient` over `duration`.
        '''
        from ..agrifood import nutrient_intake

        skipped: list[str] = []

        res = nutrient_intake(
            1,
            duration,
            nutrient,
            ci=ci,
            q=q,
            auto_correct=True,
            skipped_constraints=skipped,
            **self._kwargs,
            **kwargs,
        )

        if skipped:
            # get correction factor
            missing = tuple(
                (k, v) for k, v in self._kwargs.items() if k in skipped
            )

            filters = (("compound", nutrient),)

            correction, ncorr = correction_factor(
                "nutrients_intake",
                "intake",
                filters,
                missing,
                tuple(self._kwargs.items()),
                silent=silent,
            )

            if correction == 1.0 and not ignore_missing:
                raise RuntimeError(
                    f"Could not correct missing factors: {skipped}."
                )

            self._last_correction_status = ncorr

            return res * correction

        self._last_correction_status = 0

        return res

    @abstractmethod
    def nutrient_excretion(self, *args, **kwargs):
        pass

    @check_dim(arglist=('duration', 1, '[time]'), result='[length]**3')
    def urine_excretion(
        self, duration: Quantity, water_intake: Quantity | None = None
    ) -> Quantity:
        '''
        Amount of urine excreted over `duration`.

        Parameters
        ----------
        duration : float [time]
            Time interval over which urine is excreted.
        water_intake : float [length]**3, optional (default: inferred)
            Amount of water ingested over `duration` from all solids and
            liquids.
        '''
        dwi = {
            "senior": 2.5 * ureg.L,
            "adult": 2.5 * ureg.L,
            "young|adult": 2.5 * ureg.L,
            "teenager": 2 * ureg.L,
            "infant": 0.6 * ureg.L,
            "toddler": 1 * ureg.L,
            "kid": 1.5 * ureg.L,
        }

        if water_intake is None:
            water_intake = dwi[self._group]

        non_urine_loss = 0.1 * ureg.L + dwi[self._group] / 3

        self._last_correction_status = 0

        return water_intake - duration / ureg.day * non_urine_loss

    @check_dim(result='[mass]*[X]/[length]**3')
    def urine_concentration(
        self,
        nutrient: str,
        water_intake: Quantity | None = None,
        ci: int = 0,
        q: NumericArrayLike | None = None,
    ) -> Quantity:
        '''
        Average concentration of a nutrient in urine.

        Parameters
        ----------
        nutrient : str
            Nutrient of interest among ('N', 'P', 'K').
        water_intake : float [length]**3, optional (default: inferred)
            Daily amount of water ingested over `duration` from all solids and
            liquids.
        ci : int (optional, default: 0)
            Confidence interval (CI). If non-zero, the function will also
            return the low and high expected values corresponding to that CI.
        q : array-like of floats (optional: default: None)
            Percentiles to compute (supersedes `ci`).
        '''
        n_excr = self.nutrient_excretion(
            1 * ureg.day, nutrient, excreta="urine", ci=ci, q=q
        )

        self._last_correction_status = 0

        return n_excr / self.urine_excretion(1 * ureg.day, water_intake)


[docs] class DataBasedMetabolicModel(BaseModel): """ Data-only metabolic model to account for the consumption and excretion of an individual characterized by an age (group), a sex, a diet, and a localization (country or world region). The model returns values present in the databases, potentially applying a correction to try to account for all requested characteristics. These corrections are computed, within the whole database, as the average contribution from the fields that were missing from the requested data. E.g. a Spanish vegan male teenager is requested but only "vegan male" is available in the database, so correction for "Spanish teenager" is computed, if this is also missing, correction for "Spanish" and "teenager" are applied separately. If one of the items is also missing, no correction is applied for this field and a warning is raised. .. warning:: The consistency of all results is not guaranteed in this model, in particular, nutrient excretion in urine and feces may be greater than nutrient intake. """ def __init__( self, age_group: str, sex: str | None = None, diet: str | None = None, country: str | None = None, region: str | None = None, ): ''' Initialize a metabolic model to estimate the intake and excretions of a person based on their age (group), sex, and diet. Parameters ---------- age_group : str Precise age or age group (among "infant", "toddler", "kid", "teenager", "young adult", "adult", "senior"). sex : str, optional (default: None) Sex of the modeled person. diet : str, optional (default: "omnivorous") Diet of the modeled person. country : str, optional (default: None) Country of the modeled person. region : str, optional (default: None) World region of the modeled person (among "Africa", "Asia", "Europe", "North America", "Oceania", "South America") ''' super().__init__(age_group, sex, diet, country, region) @property def last_correction_status(self) -> int: ''' Number of corrections made to obtain the last returned value. This can take the following values: * 0 if the result was exact (directly present in the database) * -n < 0 if no correction could be provided for n separate fields * m > 0 if n individual corrections were combined to compute the final value ''' return self._last_correction_status
[docs] @check_dim(arglist=('duration', 1, '[time]'), result='[mass]*[X]') def nutrient_excretion( self, duration: Quantity, nutrient: str, excreta: str = "all", ci: int = 0, q: NumericArrayLike | None = None, ignore_missing: bool = True, silent: bool = False, ) -> Quantity: ''' Return the amount of nutrient excreted in urine and feces. .. warning:: Computing excretion from urine and feces separately may lead to individual or combined excretions that are larger than the nutrient intake from food. Only ``excreta="all"`` is guaranteed to provide consistent results with the intake. Parameters ---------- duration : float [time] Time interval over which the nitrogen is excreted. nutrient : str Nutrient of interest among ('N', 'K', 'P', 'Mg', 'Ca'). excreta : str (optional, default: "all") Type of excreta ("urine", "feces", or "all"). ci : int (optional, default: 0) Confidence interval (CI). If non-zero, the function will also return the low and high expected values corresponding to that CI. q : array-like of floats (optional: default: None) Percentiles to compute (supersedes `ci`). ignore_missing : bool, optional (default: True) Whether to raise an error or not if the value was not available and could not be properly corrected. silent : bool, optional (default: False) Skip all errors and warnings (overrides `ignore_missing`). Returns ------- excretion: Quantity [mass]*[nutrient] Amount of `nutrient` excreted over `duration`. ''' from ..nutrients import nutrient_from_population skipped: list[str] = [] if excreta == "all": factor = fraction_nutrient_excreta( nutrient, auto_correct=True, skipped_constraints=skipped, **self._kwargs, ) res = factor * self.nutrient_intake( duration, nutrient, ci=ci, q=q, ignore_missing=ignore_missing ) else: res = nutrient_from_population( 1, duration, nutrient, excreta=excreta, ci=ci, q=q, auto_correct=True, skipped_constraints=skipped, **self._kwargs, ) if skipped: # get correction factor missing = tuple( (k, v) for k, v in self._kwargs.items() if k in skipped ) filters = [("compound", nutrient)] correction = 1.0 if excreta == "all": correction, ncorr = correction_factor( "nutrients_intake", "intake", tuple(filters), missing, tuple(self._kwargs.items()), silent=silent, ) else: filters.append(("flow", excreta)) correction, ncorr = correction_factor( "nutrients_flows_body", "amount", tuple(filters), missing, tuple(self._kwargs.items()), silent=silent, ) if excreta == "urine": # use correction furthest from 1 between flows and intake intake_corr, ncorr2 = correction_factor( "nutrients_intake", "intake", tuple(filters[:-1]), missing, tuple(self._kwargs.items()), silent=silent, ) if abs(1 - correction) < abs(1 - intake_corr): correction = intake_corr ncorr = ncorr2 if correction == 1.0 and not (ignore_missing or silent): raise RuntimeError( f"Could not correct missing factors: {skipped}." ) self._last_correction_status = ncorr return res * correction self._last_correction_status = 0 return res
[docs] def flow_fraction( self, nutrient: str, flow: Literal["urine", "feces", "retention"], ci: int = 0, q: NumericArrayLike | None = None, ignore_missing: bool = True, silent: bool = False, ) -> NumericOrArray: ''' Fraction of a nutrient going in a type of flow ("urine", "feces", or "retention" in the body). Parameters ---------- nutrient : str The nutrient of interest. flow : str ("urine", "feces", or "retention") The flow considered. ci : int (optional, default: 0) Confidence interval (CI). If non-zero, the function will also return the low and high expected values corresponding to that CI. q : array-like of floats (optional: default: None) Percentiles to compute (supersedes `ci`). ignore_missing : bool, optional (default: True) Whether to raise an error or not if the value was not available and could not be properly corrected. silent : bool, optional (default: False) Skip all errors and warnings (overrides `ignore_missing`). ''' func = ( fraction_nutrient_retention if flow == "retention" else fraction_nutrient_excreta ) kwargs = self._kwargs.copy() if flow != "retention": kwargs["excreta"] = flow skipped: list[str] = [] res = func( nutrient, ci=ci, q=q, **kwargs, auto_correct=True, skipped_constraints=skipped, ) if skipped: # get correction factor missing = tuple( (k, v) for k, v in self._kwargs.items() if k in skipped ) filters = ( ("compound", nutrient), ("flow", flow), ) correction, ncorr = correction_factor( "nutrients_flows_body", "amount", tuple(filters), missing, tuple(self._kwargs.items()), silent=silent, ) if correction == 1.0 and not (ignore_missing or silent): raise RuntimeError( f"Could not correct missing factors: {skipped}." ) self._last_correction_status = ncorr return res * correction self._last_correction_status = 0 return res
_age_groups = { "infant": "infant", "toddler": "toddler", "kid": "kid", "teenager": "teenager", "young": "young|adult", "young adult": "young|adult", "young|adult": "young|adult", "adult": "adult", "senior": "senior", } _age_from_group = { "infant": "< 1", "toddler": "1-3", "kid": "4-9", "teenager": "10-17", "young|adult": "18-25", "adult": "26-59", "senior": ">= 60", } _available_diets = {None, "omnivorous", "vegetarian", "vegan"} _available_sexes = {None, "male", "female"}