"""13 — S4b panel-alternative auto-revalidation (script-grounded).
For each cheaper, NON-exact-SKU panel candidate with a complete datasheet, swap the
panel block in site.yaml and re-run the Path-A (EG4 6000XP) string + Voc + conductor
checks. Only candidates that PASS the binding cold-Voc (<=480 V at 7S), the MPPT
window, the per-MPPT input limits, AND the PV voltage-drop target are recommendable
(still flagged "auto-validated — human-confirm datasheet + tariff"). FAIL candidates
are listed with the failing check.
This is a HARNESS around the existing grounded scripts (no new electrical math):
01_string_sizing.py -> 7S cold Voc, MPPT window, per-input Imp/Isc (offline)
01b_vocmax.py -> vocmax Sandia-model corroboration (pinned venv)
03_conductors.py -> PV-string conductor gauge + voltage drop (offline)
site.yaml is backed up byte-for-byte and restored in a finally block; the candidate
override only changes the `panel:` block (and pins array.module_count=14 -> 7S/MPPT,
EG4 6000XP unchanged). The EXACT Waaree SKU must reproduce 430.1 V / PASS — that is the
harness self-test (run with --selftest).
Run: ~/.venvs/prsm-electrical/bin/python scripts/13_panel_revalidate.py candidates.json
~/.venvs/prsm-electrical/bin/python scripts/13_panel_revalidate.py --selftest
"""
import argparse
import copy
import json
import subprocess
import sys
from pathlib import Path
import yaml
ROOT = Path(__file__).resolve().parent.parent
SITE = ROOT / "site.yaml"
RESULTS = ROOT / "artifacts" / "results"
OUT = ROOT / "artifacts" / "procurement" / "panel_revalidation.json"
VENV_MAIN = Path.home() / ".venvs" / "prsm-electrical" / "bin" / "python"
VENV_VOCMAX = Path.home() / ".venvs" / "prsm-vocmax" / "bin" / "python"
VOC_CEILING = 480 # EG4 6000XP design ceiling (cold-Voc binding)
N_SERIES_PATHA = 7 # Path-A string length (4x7S)
VDROP_TARGET_PCT = 2.0 # DC PV voltage-drop target
# panel keys we override from a candidate datasheet (others keep site.yaml defaults)
DIRECT = ["pmp_w", "voc_v", "vmp_v", "isc_a", "imp_a",
"voc_tempco_pct_per_C", "isc_tempco_pct_per_C", "pmax_tempco_pct_per_C",
"max_system_voltage_v", "series_fuse_a", "cells", "cells_electrical_series",
"bifaciality", "bifacial_rear_fraction"]
REQUIRED = ["pmp_w", "voc_v", "vmp_v", "isc_a", "imp_a", "voc_tempco_pct_per_C"]
def run(py, script, *args):
r = subprocess.run([str(py), str(ROOT / "scripts" / script), *args],
capture_output=True, text=True, cwd=str(ROOT / "scripts"))
return r.returncode, r.stdout, r.stderr
def apply_candidate(base_cfg, cand):
cfg = copy.deepcopy(base_cfg)
p = cfg["panel"]
p["model"] = cand.get("name", cand.get("model", "candidate panel"))
# non-bifacial candidate -> kill the bifacial Voc adder unless it supplies its own
if cand.get("bifacial") is False and "bifacial_rear_fraction" not in cand:
p["bifacial_rear_fraction"] = 0.0
p["bifaciality"] = 0.0
for k in DIRECT:
if cand.get(k) is not None:
p[k] = cand[k]
cfg["array"]["module_count"] = 14 # 14 / 2 MPPT = 7S (Path-A string)
return cfg
def revalidate_one(base_cfg, cand):
name = cand.get("name", "?")
missing = [k for k in REQUIRED if cand.get(k) is None]
if missing:
return {"name": name, "result": "INCOMPLETE",
"reason": f"missing datasheet fields: {missing}",
"recommendable": False}
cfg = apply_candidate(base_cfg, cand)
SITE.write_text(yaml.safe_dump(cfg, sort_keys=False))
rc1, out1, err1 = run(VENV_MAIN, "01_string_sizing.py")
s01 = json.loads((RESULTS / "01_string.json").read_text()) if rc1 == 0 else None
rc1b, out1b, err1b = run(VENV_VOCMAX, "01b_vocmax.py")
s01b = json.loads((RESULTS / "01b_vocmax.json").read_text()) if rc1b == 0 else None
rc3, out3, err3 = run(VENV_MAIN, "03_conductors.py")
s03 = json.loads((RESULTS / "03_conductors.json").read_text()) if rc3 == 0 else None
rec = {"name": name, "vendor": cand.get("vendor"),
"country_of_origin": cand.get("country_of_origin"),
"ul_listing": cand.get("ul_listing")}
checks, reasons = {}, []
if s01 is None:
return {**rec, "result": "ERROR", "reason": f"01 failed rc={rc1}: {err1[-400:]}",
"recommendable": False}
cold_voc = s01.get("string_cold_voc_V")
max_series = s01.get("max_modules_series_le_480")
rec["string_cold_voc_V_7S"] = cold_voc
rec["max_modules_series_le_480"] = max_series
rec["string_verdict"] = s01.get("verdict")
checks["cold_voc_le_480"] = (cold_voc is not None and cold_voc <= VOC_CEILING)
checks["7S_within_max_series"] = (max_series is not None and N_SERIES_PATHA <= max_series)
checks["string_overall_pass"] = (s01.get("verdict") == "PASS")
if not checks["cold_voc_le_480"]:
reasons.append(f"cold Voc {cold_voc}V > {VOC_CEILING}V at 7S")
if not checks["7S_within_max_series"]:
reasons.append(f"7S exceeds max series {max_series}")
if not checks["string_overall_pass"]:
reasons.append("01 string verdict REVIEW (MPPT window or per-input Imp/Isc)")
if s01b is not None:
rec["vocmax_verdict"] = s01b.get("verdict")
checks["vocmax_pass"] = ("PASS" in str(s01b.get("verdict", "")))
if not checks["vocmax_pass"]:
reasons.append(f"vocmax verdict {s01b.get('verdict')}")
else:
rec["vocmax_verdict"] = f"unavailable (rc={rc1b})"
checks["vocmax_pass"] = None
if s03 is not None:
pvs = s03.get("pv_string", {})
rec["pv_gauge_AWG"] = pvs.get("final_gauge_AWG")
rec["pv_vdrop_pct_200ft"] = pvs.get("vdrop_pct")
checks["pv_vdrop_ok"] = (pvs.get("vdrop_pct") is not None
and pvs.get("vdrop_pct") <= VDROP_TARGET_PCT)
if not checks["pv_vdrop_ok"]:
reasons.append(f"PV Vdrop {pvs.get('vdrop_pct')}% > {VDROP_TARGET_PCT}%")
else:
rec["pv_vdrop_pct_200ft"] = f"unavailable (rc={rc3})"
checks["pv_vdrop_ok"] = None
hard = [checks["cold_voc_le_480"], checks["7S_within_max_series"],
checks["string_overall_pass"]]
soft = [v for v in (checks.get("vocmax_pass"), checks.get("pv_vdrop_ok")) if v is not None]
passed = all(hard) and all(soft)
rec["checks"] = checks
rec["result"] = "PASS" if passed else "FAIL"
rec["reason"] = "all binding checks pass" if passed else "; ".join(reasons)
rec["recommendable"] = passed
rec["flag"] = "auto-validated — human-confirm datasheet + tariff/origin before order"
return rec
def main():
ap = argparse.ArgumentParser()
ap.add_argument("candidates", nargs="?", help="JSON file: list of candidate panel datasheets")
ap.add_argument("--selftest", action="store_true",
help="revalidate the EXACT Waaree SKU; must reproduce 430.1V / PASS")
args = ap.parse_args()
base_cfg = yaml.safe_load(SITE.read_text())
if args.selftest:
p = base_cfg["panel"]
cands = [{"name": "SELFTEST — exact Waaree BiN-08-575",
**{k: p.get(k) for k in DIRECT if p.get(k) is not None},
"vendor": "(self-test)", "country_of_origin": "India",
"bifacial": True}]
else:
if not args.candidates:
ap.error("provide a candidates JSON file or --selftest")
cands = json.loads(Path(args.candidates).read_text())
if isinstance(cands, dict):
cands = cands.get("candidates", [cands])
original = SITE.read_text() # byte-for-byte backup
out = []
try:
for c in cands:
out.append(revalidate_one(base_cfg, c))
finally:
SITE.write_text(original) # ALWAYS restore the source of truth
# regenerate the canonical Path-A results so nothing is left in a candidate state
run(VENV_MAIN, "01_string_sizing.py")
run(VENV_MAIN, "03_conductors.py")
OUT.parent.mkdir(parents=True, exist_ok=True)
OUT.write_text(json.dumps({"voc_ceiling_V": VOC_CEILING, "n_series": N_SERIES_PATHA,
"candidates": out}, indent=2))
print(f"\n S4b PANEL REVALIDATION ({len(out)} candidate(s)) -> {OUT.relative_to(ROOT)}")
for r in out:
line = f" [{r['result']:>10}] {r['name']}"
if r.get("string_cold_voc_V_7S") is not None:
line += f" | 7S coldVoc {r['string_cold_voc_V_7S']}V"
if r.get("pv_vdrop_pct_200ft") not in (None,):
line += f" | Vdrop {r.get('pv_vdrop_pct_200ft')}%"
print(line)
if r["result"] != "PASS":
print(f" reason: {r['reason']}")
# self-test assertion
if args.selftest:
r = out[0]
ok = (r["result"] == "PASS" and abs((r.get("string_cold_voc_V_7S") or 0) - 430.1) < 0.5)
print(f"\n SELFTEST: {'OK — harness reproduces 430.1V PASS' if ok else 'FAILED — harness drift!'}")
sys.exit(0 if ok else 1)
if __name__ == "__main__":
main()