"""Shared helpers for the PRSM electrical-review scripts.
Every script imports `load_site()` and prints its inputs + the formula/NEC section
it used, so results are traceable. No script hard-codes a design number that
belongs in site.yaml. Each calc script also dumps its load-bearing outputs to
artifacts/results/<name>.json so downstream scripts (06_schematics) and the
markdown deliverables pull *computed* numbers, never re-keyed estimates.
"""
from __future__ import annotations
import csv
import json
import pathlib
import sys
try:
import yaml
except ImportError:
sys.exit("pip install pyyaml (add to requirements if missing)")
ROOT = pathlib.Path(__file__).resolve().parent.parent
SITE = ROOT / "site.yaml"
RESULTS = ROOT / "artifacts" / "results"
DATA = ROOT / "data"
def load_usage(name: str = "hardwick_usage.csv") -> list[dict]:
"""Load a saved billing-usage series (newest first) for reuse in calcs.
Skips '#' provenance comment lines. Raw fields net_kwh (int) and bill_amount
(float, USD net cost) come from the ledger; avg_cost_per_kwh is COMPUTED here
(bill_amount / net_kwh) so it never drifts from the raw data.
Source data lives in data/<name>.csv (see its header for provenance)."""
path = DATA / name
with open(path) as f:
lines = [ln for ln in f if ln.strip() and not ln.lstrip().startswith("#")]
rows = list(csv.DictReader(lines))
for r in rows:
r["net_kwh"] = int(r["net_kwh"])
if r.get("bill_amount") not in (None, ""):
r["bill_amount"] = float(r["bill_amount"])
r["avg_cost_per_kwh"] = round(r["bill_amount"] / r["net_kwh"], 4)
return rows
def load_site() -> dict:
"""Load site.yaml (the single source of truth). The `phase2:` block holds the
Phase-2 (Path A) build config — run lengths and the multi-row bifacial gcr — read
directly via dotted keys (e.g. get(cfg, 'phase2.runs_ft'))."""
with open(SITE) as f:
return yaml.safe_load(f)
def require(cfg: dict, dotted: str):
"""Fetch a nested key like 'panel.voc_v'; fail loudly if it's still null/CONFIRM."""
node = cfg
for part in dotted.split("."):
node = node[part]
if node is None or (isinstance(node, str) and "CONFIRM" in node):
raise SystemExit(
f"[BLOCKED] site.yaml:{dotted} is unconfirmed ({node!r}). "
"Confirm it from the datasheet/owner before this calc is valid."
)
return node
def get(cfg: dict, dotted: str, default=None):
"""Soft fetch; returns default if missing/null rather than failing."""
node = cfg
try:
for part in dotted.split("."):
node = node[part]
except (KeyError, TypeError):
return default
return default if node is None else node
def banner(title: str):
print("=" * 72)
print(title)
print("=" * 72)
def result(label: str, value, unit: str = "", basis: str = ""):
line = f" {label:<34} {value} {unit}".rstrip()
if basis:
line += f" [{basis}]"
print(line)
def dump_result(name: str, data: dict):
"""Persist a script's computed outputs for downstream consumption."""
RESULTS.mkdir(parents=True, exist_ok=True)
path = RESULTS / f"{name}.json"
with open(path, "w") as f:
json.dump(data, f, indent=2, default=str)
print(f"\n -> wrote {path.relative_to(ROOT)}")
def load_result(name: str) -> dict:
path = RESULTS / f"{name}.json"
if not path.exists():
raise SystemExit(f"[BLOCKED] {path} missing — run the upstream script first.")
with open(path) as f:
return json.load(f)