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 .money_amount import money 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))