#!/usr/bin/env python3 """ cost_methods.py — compares four ways to calculate the EV chargers' share of the electric bill, pulling directly from the same source data as the dashboard. All dollar figures are computed from real observed kWh/kW over the actual sample period (no extrapolation) — demand rates are $/kW/day, so they're applied over the real number of days sampled. A monthly-equivalent run-rate is printed separately at the end, clearly labeled as an extrapolation. Run: python3 cost_methods.py Sources: - data/CSU utility bills/CSU 15-min data 06-08_to_07-02/combined_clean.csv → building kWh - data/ChargeLab Exports/report-*.csv → charger kWh - monthly bill totals below (hardcoded from PDF bills, same as build_data.py) Methods: 1. Bundled rate — whole-building blended $/kWh × charger kWh 2. On-peak/off-peak — CSU energy rates by window × charger kWh (no demand charge) 3. + Demand allocation — Method 2 + chargers' proportional share of the demand charge (building's demand charge in each window × charger's % of that window's kWh) 4. + Own-peak demand — Method 2 + chargers' own peak kW in each window × demand rate (treats chargers as if independently demand-metered; assumes they are NOT causing the building's actual billed peak) """ import csv import glob import os from collections import defaultdict from datetime import datetime BASE = os.path.dirname(os.path.abspath(__file__)) # ─── RATES & MONTHLY BILL TOTALS ────────────────────────────────────────────── # ETL rate (2026) — Frozen Industrial, 1,000 kWh/Day Minimum ETL = { 'energy_on': 0.0459, # $/kWh, on-peak (ECA + ECC) 'energy_off': 0.0254, # $/kWh, off-peak (ECA + ECC) 'demand_on': 0.8954, # $/kW/day, on-peak 'demand_off': 0.5820, # $/kW/day, off-peak } # electricCost ($) and billKwh from the last 12 months of PDF bills — used only # to compute the whole-building blended rate for Method 1 MONTHLY_BILLS = [ {"electricCost": 2938.47, "billKwh": 20165}, {"electricCost": 2910.62, "billKwh": 19700}, {"electricCost": 2790.90, "billKwh": 19070}, {"electricCost": 2796.74, "billKwh": 18251}, {"electricCost": 4363.00, "billKwh": 41158}, {"electricCost": 5718.97, "billKwh": 59046}, {"electricCost": 5425.37, "billKwh": 59672}, {"electricCost": 6953.40, "billKwh": 72276}, {"electricCost": 5543.68, "billKwh": 59875}, {"electricCost": 5176.90, "billKwh": 51909}, {"electricCost": 4482.44, "billKwh": 46692}, {"electricCost": 3703.34, "billKwh": 30964}, ] def dict_rows(path): """csv.DictReader that tolerates stray leading blank lines before the header. Reads via binary + explicit decode rather than text-mode readlines() — text-mode reads of large files have been observed to intermittently return 0 lines in this environment despite the file being intact on disk. Raises loudly if a non-empty file still parses to 0 rows, rather than silently proceeding with wrong (too-low) totals. """ with open(path, 'rb') as f: text = f.read().decode('utf-8') lines = text.splitlines(keepends=True) start = next((i for i, l in enumerate(lines) if l.strip()), 0) rows = list(csv.DictReader(lines[start:])) if not rows and os.path.getsize(path) > 0: raise RuntimeError( f"{path} parsed to 0 rows but is {os.path.getsize(path)} bytes on disk — " f"read likely failed, file is not actually empty") return rows def is_onpeak(dt): """CSU on-peak window: summer (Apr-Sep) 11am-6pm, winter 4pm-10pm, weekdays only.""" if dt.weekday() >= 5: return False summer = dt.month in {4, 5, 6, 7, 8, 9} return 11 <= dt.hour < 18 if summer else 16 <= dt.hour < 22 def load_csu(): path = os.path.join( BASE, "data/CSU utility bills/CSU 15-min data 06-08_to_07-02/combined_clean.csv") building_kwh = {} with open(path) as f: for row in csv.DictReader(f): building_kwh[row['Timestamp'][:16]] = float(row['Consumption (kWh)']) return building_kwh def load_charger_kwh(): """Charger kWh per 15-min bucket, keyed 'YYYY-MM-DDTHH:MM', summed across all chargers. Report exports are cumulative/overlapping (each new export re-covers the whole date range), and the "15-minute interval ID" is regenerated on each export rather than staying stable — so dedup on (Session ID, interval start timestamp) instead, which is stable across exports. """ report_csvs = sorted(glob.glob(os.path.join(BASE, "data/ChargeLab Exports/report-*.csv"))) charger_kwh = defaultdict(float) seen = set() for path in report_csvs: for row in dict_rows(path): ts_str = row['Interval start date/time (YYYY-MM-DD hh:mm:ss) (local)'] dedup_key = (row['Session ID'], ts_str) if dedup_key in seen: continue seen.add(dedup_key) ts = datetime.strptime(ts_str, '%Y-%m-%d %H:%M:%S') bucket_min = (ts.minute // 15) * 15 bucket = ts.replace(minute=bucket_min, second=0, microsecond=0) key = bucket.strftime('%Y-%m-%dT%H:%M') charger_kwh[key] += float(row['Interval energy provided (kWh)']) return charger_kwh def load_charger_kw(): """Total simultaneous charger kW per 15-min bucket, for peak-demand detection. Uses ChargeLab's own "Rolling 15-minute average power (kW)" field directly rather than deriving kW from interval energy — one row in this data has an energy value inconsistent with its own rolling-power reading (a real export glitch), which would otherwise produce an impossible phantom demand spike. Takes the max per (charger, bucket) across report files, which is also naturally robust to the cumulative/overlapping exports. """ report_csvs = sorted(glob.glob(os.path.join(BASE, "data/ChargeLab Exports/report-*.csv"))) per_charger_bucket_kw = defaultdict(lambda: defaultdict(float)) for path in report_csvs: for row in dict_rows(path): kw = float(row['Rolling 15-minute average power (kW)']) if kw == 0: continue charger = row['Charger display ID'] ts = datetime.strptime( row['Interval start date/time (YYYY-MM-DD hh:mm:ss) (local)'], '%Y-%m-%d %H:%M:%S') bucket_min = (ts.minute // 15) * 15 bucket = ts.replace(minute=bucket_min, second=0, microsecond=0) key = bucket.strftime('%Y-%m-%dT%H:%M') if kw > per_charger_bucket_kw[charger][key]: per_charger_bucket_kw[charger][key] = kw total_kw = defaultdict(float) for charger, buckets in per_charger_bucket_kw.items(): for key, kw in buckets.items(): total_kw[key] += kw return total_kw def main(): building_kwh = load_csu() charger_kwh = load_charger_kwh() charger_kw = load_charger_kw() # Bound everything to the CSU meter's date range (that's what's billed) timestamps = sorted(building_kwh) first_dt = datetime.strptime(timestamps[0], '%Y-%m-%dT%H:%M') last_dt = datetime.strptime(timestamps[-1], '%Y-%m-%dT%H:%M') days_sampled = (last_dt.date() - first_dt.date()).days + 1 # Split building & charger kWh into on-peak / off-peak, and track both the # building's peak kW (whole meter, incl. chargers) and the chargers' own # peak kW in each window building_onpeak_kwh = building_offpeak_kwh = 0.0 charger_onpeak_kwh = charger_offpeak_kwh = 0.0 building_peak_onpeak_kw = building_peak_offpeak_kw = 0.0 charger_peak_onpeak_kw = charger_peak_offpeak_kw = 0.0 for ts in timestamps: dt = datetime.strptime(ts, '%Y-%m-%dT%H:%M') b_kwh = building_kwh[ts] c_kwh = charger_kwh.get(ts, 0.0) c_kw = charger_kw.get(ts, 0.0) # measured rolling power, not derived from energy b_kw = b_kwh * 4 # 15-min kWh -> kW if is_onpeak(dt): building_onpeak_kwh += b_kwh charger_onpeak_kwh += c_kwh building_peak_onpeak_kw = max(building_peak_onpeak_kw, b_kw) charger_peak_onpeak_kw = max(charger_peak_onpeak_kw, c_kw) else: building_offpeak_kwh += b_kwh charger_offpeak_kwh += c_kwh building_peak_offpeak_kw = max(building_peak_offpeak_kw, b_kw) charger_peak_offpeak_kw = max(charger_peak_offpeak_kw, c_kw) # Real, observed totals — no extrapolation. Demand rates are $/kW/day, so # they're applied over the actual number of days sampled, not a fictional # 30-day month. charger_total_kwh = charger_onpeak_kwh + charger_offpeak_kwh onpeak_share = charger_onpeak_kwh / building_onpeak_kwh if building_onpeak_kwh else 0 offpeak_share = charger_offpeak_kwh / building_offpeak_kwh if building_offpeak_kwh else 0 # ── Method 1: bundled rate ──────────────────────────────────────────────── bundled_rate = (sum(b['electricCost'] for b in MONTHLY_BILLS) / sum(b['billKwh'] for b in MONTHLY_BILLS)) method1_cost = charger_total_kwh * bundled_rate # ── Method 2: on-peak/off-peak energy only ──────────────────────────────── method2_cost = (charger_onpeak_kwh * ETL['energy_on'] + charger_offpeak_kwh * ETL['energy_off']) method2_rate = method2_cost / charger_total_kwh if charger_total_kwh else 0 # ── Method 3: + proportional demand charge allocation ───────────────────── demand_on_period = building_peak_onpeak_kw * ETL['demand_on'] * days_sampled demand_off_period = building_peak_offpeak_kw * ETL['demand_off'] * days_sampled charger_demand_cost = demand_on_period * onpeak_share + demand_off_period * offpeak_share method3_cost = method2_cost + charger_demand_cost method3_rate = method3_cost / charger_total_kwh if charger_total_kwh else 0 # ── Method 4: + chargers' own peak kW, independent of the building ──────── # Assumes chargers are NOT causing the building's billed peak — so their # demand charge is based solely on their own highest simultaneous kW in # each window, as if they were on their own demand meter. charger_demand_onpeak_4 = charger_peak_onpeak_kw * ETL['demand_on'] * days_sampled charger_demand_offpeak_4 = charger_peak_offpeak_kw * ETL['demand_off'] * days_sampled method4_cost = method2_cost + charger_demand_onpeak_4 + charger_demand_offpeak_4 method4_rate = method4_cost / charger_total_kwh if charger_total_kwh else 0 # ── Report ───────────────────────────────────────────────────────────────── print(f"Sample period: {first_dt.date()} to {last_dt.date()} ({days_sampled} days, real data)") print(f"Charger kWh (observed): {charger_total_kwh:,.0f} kWh") print(f"Charger share of building kWh: on-peak {onpeak_share:.1%}, off-peak {offpeak_share:.1%}") print() print(f"Method 1 — Bundled rate ({bundled_rate*100:.3f}¢/kWh): " f"${method1_cost:,.0f} over {days_sampled} days (${bundled_rate:.3f}/kWh)") print(f"Method 2 — On-peak/off-peak energy only: " f"${method2_cost:,.0f} over {days_sampled} days (${method2_rate:.3f}/kWh)") print(f"Method 3 — + demand charge allocation (proportional by kWh share): " f"${method3_cost:,.0f} over {days_sampled} days (${method3_rate:.3f}/kWh)") print(f" (demand charge: ${charger_demand_cost:,.0f} = " f"${demand_on_period:,.0f} on-peak × {onpeak_share:.1%} + " f"${demand_off_period:,.0f} off-peak × {offpeak_share:.1%})") print(f"Method 4 — + demand charge allocation (chargers' own peak kW): " f"${method4_cost:,.0f} over {days_sampled} days (${method4_rate:.3f}/kWh)") print(f" (demand charge: ${charger_demand_onpeak_4 + charger_demand_offpeak_4:,.0f} = " f"{charger_peak_onpeak_kw:.1f} kW on-peak × ${ETL['demand_on']}/kW/day × {days_sampled} + " f"{charger_peak_offpeak_kw:.1f} kW off-peak × ${ETL['demand_off']}/kW/day × {days_sampled})") print() print(f"Monthly-equivalent run-rate (× 30/{days_sampled}, for context only — not a real bill):") scale = 30 / days_sampled for label, cost in [("Method 1", method1_cost), ("Method 2", method2_cost), ("Method 3", method3_cost), ("Method 4", method4_cost)]: print(f" {label}: ${cost * scale:,.0f}/month") if __name__ == "__main__": main()