energy.tariff-time-of-use

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

import re
from typing import Dict, List, Sequence

from .energy_tariff_time_of_use_types import HalfHourUsage, TariffRate, TimeOfUseCost, TimeOfUseLine
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

SLOT = re.compile(r"[0-9]{4}-[0-9]{2}-[0-9]{2}T([01][0-9]|2[0-3]):(00|30)")
CLOCK = re.compile(r"([01][0-9]|2[0-3]):(00|30)")
# watt-hours x thousandths of a minor unit per kWh is millionths of a minor unit.
DIVISOR = 1_000_000
MAX_SAFE = 9007199254740991


def _is_int(value: object) -> bool:
    return isinstance(value, int) and not isinstance(value, bool)


def _covers(rate: TariffRate, start: str) -> bool:
    if len(rate.start) == 5:
        time = start[11:]
        if rate.start == rate.end:
            return True
        if rate.start < rate.end:
            return rate.start <= time < rate.end
        return time >= rate.start or time < rate.end
    return rate.start <= start < rate.end


def time_of_use_cost(
    usage: Sequence[HalfHourUsage], rates: Sequence[TariffRate], currency: str, mode: RoundingMode
) -> TimeOfUseCost:
    """Cost half-hourly usage against time-of-use rates. Each half hour must
    fall in exactly one rate; each rate name becomes a line costed exactly and
    rounded once.
    """
    names: List[str] = []
    for r in rates:
        daily = CLOCK.fullmatch(r.start) is not None and CLOCK.fullmatch(r.end) is not None
        dated = SLOT.fullmatch(r.start) is not None and SLOT.fullmatch(r.end) is not None
        if not daily and not dated:
            raise ValueError(
                "rate %s: start and end must both be HH:MM or both YYYY-MM-DDTHH:MM on the hour or half hour, "
                "received %s and %s" % (r.name, r.start, r.end)
            )
        if dated and r.start >= r.end:
            raise ValueError("rate %s: start must be before end, received %s and %s" % (r.name, r.start, r.end))
        if not _is_int(r.rate):
            raise ValueError(
                "rate %s: rate must be a whole number of thousandths of a minor unit, received %s" % (r.name, r.rate)
            )
        if r.name not in names:
            names.append(r.name)
    exact: Dict[str, int] = {n: 0 for n in names}
    energy: Dict[str, int] = {n: 0 for n in names}
    seen = set()
    watt_hours = 0
    for u in usage:
        if not isinstance(u.start, str) or not SLOT.fullmatch(u.start):
            raise ValueError("usage start must be YYYY-MM-DDTHH:MM on the hour or half hour, received %s" % (u.start,))
        if not _is_int(u.watt_hours) or u.watt_hours < 0:
            raise ValueError(
                "usage wattHours must be a whole number of 0 or more, received %s at %s" % (u.watt_hours, u.start)
            )
        if u.start in seen:
            raise ValueError("usage for the half hour starting %s appears twice" % (u.start,))
        seen.add(u.start)
        matches = [r for r in rates if _covers(r, u.start)]
        if not matches:
            raise ValueError("no rate covers the half hour starting %s" % (u.start,))
        if len(matches) > 1:
            raise ValueError(
                "rates %s and %s both cover the half hour starting %s" % (matches[0].name, matches[1].name, u.start)
            )
        r = matches[0]
        part = u.watt_hours * r.rate
        nxt = exact[r.name] + part
        if abs(nxt) > MAX_SAFE or abs(part) > MAX_SAFE:
            raise ValueError("usage cost too large to calculate exactly")
        exact[r.name] = nxt
        energy[r.name] += u.watt_hours
        watt_hours += u.watt_hours
    lines = []
    total = 0
    for name in names:
        minor = round_div(exact[name], DIVISOR, mode)
        total += minor
        lines.append(TimeOfUseLine(name=name, watt_hours=energy[name], cost=money(minor, currency)))
    return TimeOfUseCost(lines=lines, watt_hours=watt_hours, total=money(total, currency))