Functional Weave
Code in Python

energy.water-bill@1.0.0

impl/python.py

2,948 bytes · the Python implementation · view raw

Imports name this capability’s declared dependencies, which fune builds next to it in your project; each one links to its page.

from .dates_days_between import days_between  ← from dates.days-between ^1.0.0 · built alongside by fune
from .energy_meter_advance import meter_advance  ← from energy.meter-advance ^1.0.0 · built alongside by fune
from .energy_water_bill_types import MeterRead, WaterBill, WaterTariff
from .math_round_div import RoundingMode, round_div  ← from math.round-div ^1.0.0 · built alongside by fune
from .money_amount import money  ← from money.amount ^1.0.0 · built alongside by fune

MAX_SAFE = 9007199254740991


def _check_rate(what: str, value: object) -> None:
    if not isinstance(value, int) or isinstance(value, bool) or value < 0:
        raise ValueError("%s must be a whole number of 0 or more, received %s" % (what, value))


def _exact(product: int) -> int:
    if product > MAX_SAFE:
        raise ValueError("water charge too large to calculate exactly")
    return product


def water_bill(
    opening_read: MeterRead,
    closing_read: MeterRead,
    meter_digits: int,
    tariff: WaterTariff,
    surface_water_drains: bool,
    currency: str,
    mode: RoundingMode,
) -> WaterBill:
    """A metered water and sewerage bill: volumetric water, sewerage on the
    share returned to the sewer, and standing and surface water charges
    apportioned by days / 365, each line rounded once.
    """
    cubic_metres = meter_advance(opening_read.value, closing_read.value, meter_digits)
    days = days_between(opening_read.date, closing_read.date)
    if days < 0:
        raise ValueError(
            "the closing read must not be dated before the opening read, received %s and %s"
            % (opening_read.date, closing_read.date)
        )
    _check_rate("waterVolumeRate", tariff.water_volume_rate)
    _check_rate("sewerageVolumeRate", tariff.sewerage_volume_rate)
    _check_rate("waterStandingCharge", tariff.water_standing_charge)
    _check_rate("sewerageStandingCharge", tariff.sewerage_standing_charge)
    _check_rate("surfaceWaterCharge", tariff.surface_water_charge)
    share = tariff.return_to_sewer
    if not isinstance(share, int) or isinstance(share, bool) or share < 0 or share > 10000:
        raise ValueError("returnToSewer must be basis points from 0 to 10000, received %s" % (share,))
    water = round_div(_exact(cubic_metres * tariff.water_volume_rate), 1000, mode)
    sewerage = round_div(_exact(cubic_metres * tariff.sewerage_volume_rate * share), 10_000_000, mode)
    water_standing = round_div(_exact(tariff.water_standing_charge * days), 365, mode)
    sewerage_standing = round_div(_exact(tariff.sewerage_standing_charge * days), 365, mode)
    surface = round_div(_exact(tariff.surface_water_charge * days), 365, mode) if surface_water_drains else 0
    return WaterBill(
        cubic_metres=cubic_metres,
        days=days,
        water_volume_charge=money(water, currency),
        sewerage_volume_charge=money(sewerage, currency),
        water_standing_charge=money(water_standing, currency),
        sewerage_standing_charge=money(sewerage_standing, currency),
        surface_water_charge=money(surface, currency),
        total=money(water + sewerage + water_standing + sewerage_standing + surface, currency),
    )