# -*- coding: utf-8 -*-
# SPDX-FileCopyrightText: 2022-2023 Tanguy Fardet
# SPDX-License-Identifier: GPL-3.0-or-later
# orgmatt/impacts/water.py
from typing import Optional, Union
from pint import Quantity
from .._utils import _filter_dataframe, return_values, auto_format
from ..data import excr_freq
from ..typing import NumericArrayLike, NumericOrArray
from ..units import check_dim, day, L, ureg
[docs]
@auto_format
@check_dim(
arglist=(
('duration', 1, '[time]'), ('avg_flush_urine', 2, '[length]**3'),
('avg_flush_feces', 3, '[length]**3')
), result='[length]**3')
def water_consumption_flush(
pop: NumericOrArray,
duration: Quantity,
avg_flush_urine: Quantity = 3*L,
avg_flush_feces: Quantity = 7*L,
frac_urinations: float = 1,
frac_defecations: float = 1,
ci: int = 0,
q: Optional[NumericArrayLike] = None,
**kwargs
) -> Quantity:
'''
Compute water use from toilet flushes based on the urination and defecation
frequencies from the ``excretion_frequency.csv`` database.
Parameters
----------
pop : float or list-like object
Number of inhabitants.
duration : float [time]
Time interval over water is consumed.
avg_flush_urine : float [length]**3 (optional, default: 3 liters)
Average water consumption for a small flush.
avg_flush_feces : float [length]**3 (optional, default: 7 liters)
Average water consumption for a large flush.
frac_urinations : float (default: 1)
Fraction of urinations for which the flushes should be counted.
frac_defecations : float (default: 1)
Fraction of defecations for which the flushes should be counted.
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, otherwise it
will only return the average value.
kwargs : dict, optional
Additional arguments can be use to restrict the results to a subset of
the full database. E.g. one can add `group="senior"` to return only
data from senior individuals (otherwise the default is "adult"), or
`region="Europe"` to use only values obtained from studies in European
countries.
'''
df = excr_freq.copy()
# remove potential duplicate entry for type argument
kwargs = kwargs.copy()
if "type" in kwargs:
del kwargs["type"]
df = _filter_dataframe(df, kwargs)
defecation = df[df.type == "defecation"]
urination = df[df.type == "urination"]
freq_feces = return_values(defecation, "frequency", pop, ci, q) * (1 / day)
freq_urine = return_values(urination, "frequency", pop, ci, q) * (1 / day)
num_voids = (freq_urine - freq_feces) * duration * frac_urinations
num_defec = freq_feces * duration * frac_defecations
return num_defec*avg_flush_feces + num_voids*avg_flush_urine