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))