"""E49-S2-12-v1: zelfstandige les-simulator, geen installatiebesturing. Getest met Python 3.13.1 / NumPy 2.4.0. SI in de kern; kWh en L in boekhouding. Klok UTC+1 vast. Geen hardware, MQTT-server of cursusimport. CLI: python habitat_brein.py --out resultaat De ontvangen telegrammen worden als gewone dictionaries beproefd. """ import argparse import csv import hashlib import io import json import math from datetime import datetime, timedelta from pathlib import Path import numpy as np SCHEMA = "e49-resultaten-v1" def finite(x): return isinstance(x, (int, float)) and not isinstance(x, bool) and math.isfinite(x) def valideer(c): for k, v in c.items(): if isinstance(v, (int, float)) and not finite(v): raise ValueError("niet-eindig: "+k) assert c["b48_serie"]*c["b48_cel_min_V"] == c["b48_rail_min_V"] assert abs(c["b48_serie"]*c["b48_cel_max_V"]-c["b48_rail_max_V"]) < 1e-9 assert 44 <= c["b48_rail_min_V"] < c["b48_rail_max_V"] <= 56 assert c["b12_kWh"] == 3.072 and c["b48_kWh"] > 0 for b in ("b12", "b48"): assert 0 <= c[b+"_soc_min"] < c[b+"_soc_max"] <= 1 assert c[b+"_soc_min"] <= c[b+"_soc_start"] <= c[b+"_soc_max"] assert c["Ca_J_K"] == 80000 and c["Cb_J_K"] == 1100000 assert c["Cwp_J_K"] == 100*4186 and c["convector_max_W"] == 800 assert c["Hwoon_W_K"] == 29.8 and c["vent_m3_h"] == 100 assert c["nft_vlotter_L"] == 24 and c["nft_max_L"] == 60 assert c["drink_max_L"] == 500 assert c["besturing_s"] == c["telemetrie_s"] == 60 assert c["besturing_s"] % c["plantstap_s"] == 0 return hashlib.sha256(json.dumps(c, sort_keys=True, separators=(",", ":"), allow_nan=False).encode()).hexdigest() def lees_csv(bron): return list(csv.DictReader(io.StringIO(bron))) class PID: """Afgeleide op meting; integrale toestand heeft eenheid watt.""" def __init__(self, kp, ki, kd, maximum, begin=0, tau_d=60, tau_aw=600): if not all(finite(x) for x in (kp, ki, kd, maximum, begin, tau_d, tau_aw)): raise ValueError("niet-eindige PID-instelling") if min(kp, ki, kd) < 0 or min(maximum, tau_d, tau_aw) <= 0: raise ValueError("ongeldige PID-instelling") self.kp, self.ki, self.kd = kp, ki, kd self.maximum, self.I = maximum, begin self.tau_d, self.tau_aw = tau_d, tau_aw self.vorig, self.d = None, 0.0 def vraag(self, setpoint, y, dt): if not all(finite(a) for a in (setpoint, y, dt)) or dt <= 0: raise ValueError("ongeldige PID-invoer") afgeleid = 0 if self.vorig is None else (y-self.vorig)/dt a = math.exp(-dt/self.tau_d) self.d = a*self.d+(1-a)*afgeleid self.vorig = y self.e = setpoint-y self.onbegrensd = self.kp*self.e+self.I-self.kd*self.d return min(self.maximum, max(0.0, self.onbegrensd)) def terug(self, werkelijk, dt, aw=True): correctie = (werkelijk-self.onbegrensd)/self.tau_aw if aw else 0 self.I += dt*(self.ki*self.e+correctie) class StopVrijgave: """Logisch model van een onafhankelijke, stopdominante vergrendeling.""" def __init__(self): self.vergrendeld, self.vorig_reset = True, False def stap(self, stop, voeding, gezond, reset, verzoek_uit, contact_uit): flank = reset and not self.vorig_reset self.vorig_reset = reset if stop or not voeding or not gezond: self.vergrendeld = True elif flank and verzoek_uit and contact_uit: self.vergrendeld = False return voeding and gezond and not stop and not self.vergrendeld class LokaleEigenaar: """Een eigenaar; replayregister overleeft het gesimuleerde herstarten. Het aangeleverde register stelt duurzame opslag voor. De JSON-roundtrip in de test bewijst logica, geen EEPROM-/filesystem-proof-test. """ def __init__(self, c, boot="woon-A", register=None): self.c, self.boot, self.mode = c, boot, "NORMAAL" self.source_boot = "pi-A" self.register = {} if register is None else register self.lease, self.setpoint = -1, 20.0 self.stop = StopVrijgave() def voorstel(self, m, nu): vereist = ("node", "owner_boot", "source", "source_boot", "seq", "tijd", "ttl", "mode", "quality", "set_C") reden = "OK" if not all(k in m for k in vereist): reden = "SCHEMA" elif not all(finite(m[k]) for k in ("seq", "tijd", "ttl", "set_C")): reden = "NIET_EINDIG" elif not isinstance(m["seq"], int) or m["seq"] < 0: reden = "SEQ" elif (m["node"], m["owner_boot"], m["source"], m["source_boot"]) != ( "woon", self.boot, "pi", self.source_boot): reden = "IDENTITEIT_BOOT" elif m["mode"] != self.mode or self.mode != "NORMAAL": reden = "MODE" elif m["quality"] != "goed": reden = "QUALITY" elif not 0 < m["ttl"] <= self.c["voorstel_ttl_s"]: reden = "TTL" elif not m["tijd"] <= nu < m["tijd"]+m["ttl"]: reden = "VERLOPEN_TOEKOMST" elif not self.c["home_set_min_C"] <= m["set_C"] <= self.c["home_set_max_C"]: reden = "LIMIET" elif m["seq"] <= self.register.get(m["source_boot"], -1): reden = "REPLAY" elif self.stop.vergrendeld: reden = "STOP" if reden == "OK": self.register[m["source_boot"]] = m["seq"] self.lease = m["tijd"]+m["ttl"] self.setpoint = float(m["set_C"]) return {"node": "woon", "owner_boot": self.boot, "source_boot": m.get("source_boot"), "seq": m.get("seq") if finite(m.get("seq")) else None, "tijd_s": nu, "ACK": reden} def doel(self, nu): return self.setpoint if nu < self.lease else 20.0 class MeetPoort: """Alleen modulelucht: geen menging van lucht en reservoir-NTC.""" def __init__(self, c): self.c, self.seq, self.historie = c, -1, [] def meet(self, m, nu, voorspelde_sensor_C): if not finite(nu) or not finite(voorspelde_sensor_C): return {"waarde": None, "status": "ONGELDIG_MODEL"} if m is None: return {"waarde": None, "status": "ONTBREEKT"} if not all(k in m for k in ("node", "boot", "seq", "tijd", "C", "quality")): return {"waarde": None, "status": "SCHEMA"} status = "goed" if m["node"] != "modulelucht" or m["boot"] != "mega-A": status = "BOOT" elif not all(finite(m[k]) for k in ("seq", "tijd", "C")): status = "NIET_EINDIG" elif not isinstance(m["seq"], int) or m["seq"] <= self.seq: status = "REPLAY" elif not 0 <= nu-m["tijd"] <= self.c["meting_max_leeftijd_s"]: status = "STALE_KLOK" elif m["quality"] != "goed" or not -20 <= m["C"] <= 50: status = "QUALITY_BEREIK" elif abs(m["C"]-voorspelde_sensor_C) > .8: status = "INNOVATIE" else: self.seq = m["seq"] self.historie.append((m["C"], voorspelde_sensor_C)) self.historie = self.historie[-12:] if len(self.historie) == 12: gem = [a for a, _ in self.historie] pred = [b for _, b in self.historie] if max(gem)-min(gem) < 1e-8 and max(pred)-min(pred) > .15: status = "FREEZE" return {"waarde": float(m["C"]) if status == "goed" else None, "status": status} def laad_toegestaan(c, rail, temp_C, pack_V, cellen): if not finite(temp_C) or not finite(pack_V): return False if rail == "b12": return temp_C >= c["b12_laad_min_C"] and 10 <= pack_V <= 14.2 return (temp_C >= c["b48_laad_min_C"] and c["b48_rail_min_V"] <= pack_V <= c["b48_rail_max_V"] and len(cellen) == c["b48_serie"] and all(finite(v) and c["b48_cel_min_V"] <= v <= c["b48_cel_max_V"] for v in cellen)) def pq(lasten): p = sum(w for w, pf in lasten) q = sum(w*math.tan(math.acos(pf)) for w, pf in lasten) return p, q, math.hypot(p, q) def kies_lasten(c, gevraagd, bus_W, spanning, piek=False): """Volgorde bepaalt prioriteit; elke afgewezen dienst blijft in het log.""" geleverd, geweigerd, lasten = {}, {}, [] limiet = c["omvormer_piek_VA"] if piek else c["omvormer_cont_VA"] stroom = c["b48_piek_A"] if piek else c["b48_ontlaad_A"] for naam, w, pf in gevraagd: p, q, s = pq(lasten+[(w, pf)]) dc = p/c["omvormer_eta"]+c["omvormer_nullast_W"] if s <= limiet and dc <= bus_W and dc/spanning <= stroom: geleverd[naam] = w lasten.append((w, pf)) else: geleverd[naam], geweigerd[naam] = 0.0, w p, q, s = pq(lasten) dc = p/c["omvormer_eta"]+c["omvormer_nullast_W"] if p else 0 return geleverd, geweigerd, (p, q, s, dc) def accu(c, E, rail, generatie_W, last_W, dt_h, temp_C): """Paden begrenzen vóór energie-integratie; geen achteraf-clipping van E.""" cap = c[rail+"_kWh"] emin, emax = cap*c[rail+"_soc_min"], cap*c[rail+"_soc_max"] U = 12.8 if rail == "b12" else c["b48_rail_min_V"] ec, ed = c["eta_laad"], c["eta_ontlaad"] max_dis = min(U*c[rail+"_ontlaad_A"]*ed, max(0, E-emin)*1000*ed/dt_h) bediend = min(last_W, generatie_W+max_dis) tekort = last_W-bediend surplus = max(0, generatie_W-bediend) cel = U/16 kan_laden = laad_toegestaan(c, rail, temp_C, U, [cel]*16) charge = min(surplus, U*c[rail+"_laad_A"], max(0, emax-E)*1000/(ec*dt_h)) if kan_laden else 0 discharge = max(0, bediend-generatie_W) dE = (ec*charge-discharge/ed)*dt_h/1000 verlies = ((1-ec)*charge+(1/ed-1)*discharge)*dt_h/1000 curtail = surplus-charge # Beschikbare bronenergie + opslagafname = last + verliezen + ongebruikt. residu = (generatie_W-bediend-curtail)*dt_h/1000-dE-verlies assert emin-1e-10 <= E+dE <= emax+1e-10 assert abs(residu) < 1e-10 return E+dE, bediend, tekort, curtail, charge, discharge, verlies, residu def warmtestap(c, z, tout, qm_req, qh_req, qwp, qnood, gains_m, gains_h, extra_H, hwoon, dt): """RK4 integreert toestanden EN dezelfde fluxen voor de balans.""" capaciteiten = np.array([c["Ca_J_K"], c["Cb_J_K"], c["Cwoon_J_K"], c["Cwp_J_K"]]) def f(t): a, b, h, w = t qm = min(qm_req, c["Gconvector_W_K"]*max(0, w-a)) qh = min(qh_req, c["Gvloer_W_K"]*max(0, w-h)) qbw = c["Hwp_naar_woon_W_K"]*(w-h) g = c["Gmodule_W_K"]*(a-b) lm = (c["Hmodule_W_K"]+extra_H)*(a-tout) lh = hwoon*(h-tout) vermogen = np.array([qm+gains_m-lm-g, g, qh+qbw+qnood+gains_h-lh, qwp-qm-qh-qbw]) flux = np.array([qm, qh, lm, lh, qbw]) return vermogen/capaciteiten, flux voor = z.copy() som = np.zeros(5) for _ in range(round(c["besturing_s"]/dt)): k1, f1 = f(z) k2, f2 = f(z+dt*k1/2) k3, f3 = f(z+dt*k2/2) k4, f4 = f(z+dt*k3) z = z+dt*(k1+2*k2+2*k3+k4)/6 som += dt*(f1+2*f2+2*f3+f4)/6 delta = float(capaciteiten@(z-voor)) netto = (qwp+qnood+gains_m+gains_h)*c["besturing_s"]-som[2]-som[3] assert abs(delta-netto) < 1e-5 return z, som, delta-netto def voorspellers(c, archief): data = archief[2:219] # 217 verleden-dagen, week geen trainingsdata train, val, test = data[:-70], data[-70:-35], data[-35:] assert train[-1]["doel_dag"] < "2021-03-07" def features(r): t, s = float(r["prognose_T_C"]), float(r["prognose_zonuren"]) return [max(12-t, 0), max(3-t, 0)**2, s] X = np.array([features(r) for r in train]) mean, scale = X.mean(axis=0), X.std(axis=0) scale[scale == 0] = 1 def ontwerp(rows): x = np.array([features(r) for r in rows]) return np.column_stack([np.ones(len(rows)), (x-mean)/scale]) Y = np.array([[float(r["vraag_ref_kWh"]), float(r["pv48_kWh"])] for r in train]) beta = np.linalg.lstsq(ontwerp(train), Y, rcond=None)[0] def voorspel(r): try: x = np.asarray(features(r), dtype=float) except (ValueError, TypeError, KeyError): return None if not np.isfinite(x).all(): return None # Extrapolatie buiten het verstrekte proefgebied: onthouding. if float(r["prognose_T_C"]) < -15 or float(r["prognose_T_C"]) > 30: return None pred = np.r_[1, (x-mean)/scale]@beta return np.maximum(pred, 0).tolist() scores = {} for naam, rows in (("validatie", val), ("test", test)): y = np.array([[float(r["vraag_ref_kWh"]), float(r["pv48_kWh"])] for r in rows]) yp = np.maximum(ontwerp(rows)@beta, 0) base = np.array([[float(r["laatste_complete_vraag_kWh"]), float(r["laatste_complete_pv48_kWh"])] for r in rows]) scores[naam] = {"dagen": len(rows), "model_RMSE_kWh": np.sqrt(np.mean((yp-y)**2, axis=0)).tolist(), "basis_RMSE_kWh": np.sqrt(np.mean((base-y)**2, axis=0)).tolist(), "prognose_T_RMSE_K": math.sqrt(np.mean([ (float(r["prognose_T_C"])-float(r["gerealiseerd_T_C"]))**2 for r in rows])), "prognose_zonuren_RMSE_h": math.sqrt(np.mean([ (float(r["prognose_zonuren"])-float(r["gerealiseerde_zonuren"]))**2 for r in rows]))} return {"train_dagen": len(train), "train_van": train[0]["doel_dag"], "train_tot": train[-1]["doel_dag"], "validatie_van": val[0]["doel_dag"], "validatie_tot": val[-1]["doel_dag"], "test_van": test[0]["doel_dag"], "test_tot": test[-1]["doel_dag"], "scores_status": "ACHTERAF_SYNTHETISCH", "beta": beta.tolist(), "mean": mean.tolist(), "scale": scale.tolist(), "scores": scores}, voorspel def simuleer(c, uren, forecast, naam="normaal", plantstap=None, beheer=True): dt = c["besturing_s"] h = dt/3600 plantstap = c["plantstap_s"] if plantstap is None else plantstap z = np.array([18., 18., 20., 35.]) E12, E48 = c["b12_kWh"]*c["b12_soc_start"], c["b48_kWh"]*c["b48_soc_start"] if naam == "vol": E12, E48 = c["b12_kWh"], c["b48_kWh"]*c["b48_soc_max"] if naam == "koud_leeg": E12, E48 = c["b12_kWh"]*.2, c["b48_kWh"]*.2 if naam == "h2_reserve": E48 = c["b48_kWh"]*.2 massah2 = 0 if naam == "koud_leeg" else c["h2_start_kg"] if naam == "vol": massah2 = c["h2_max_kg"] proceswater = 0.0 if naam == "h2_water_leeg" else c["h2_water_start_L"] begin_proceswater = proceswater nft, drink = c["nft_start_L"], c["drink_start_L"] if naam == "waterlek": nft = 31.0 if naam == "drink_leeg": drink = 0.0 eigenaar = LokaleEigenaar(c) eigenaar.stop.stap(False, True, True, True, True, True) eigenaar.stop.stap(False, True, True, False, True, True) pm = PID(220, .03, 1200, 800, 200) ph = PID(350, .025, 3000, 2500, 350) pw = PID(550, .12, 0, 5300, 700) therm, rows, events = 0.0, [], [] total = {k: 0.0 for k in ("pv12", "pv48", "wind48", "last12", "last48", "tekort12", "tekort48", "curtail12", "curtail48", "wind_dump_raw", "batverlies12", "batverlies48", "h2_in", "h2_uit", "h2_gemaakt", "h2_verbruikt", "h2_gelekt", "h2_dcverlies", "h2_comp", "h2_procverlies", "h2_celverlies", "qwp", "wp_elektrisch", "qmodule", "qwoon", "nood_tekort", "druk_tekort", "drink_tekort_L", "vent_tekort", "nft_opname_L", "nft_verdamping_L", "nft_lek_L", "nft_fill_L", "drink_fill_L", "drink_verbruik_L", "recirculatie_L", "comfort_woon_h", "comfort_module_h", "elektrisch_residu", "water_residu")} total.update({"laad12_kWh": 0.0, "laad48_kWh": 0.0, "warmte_geweigerd_kWh": 0.0, "nft_dienst_tekort_L": 0.0, "nood_geleverd_kWh": 0.0, "winst_module_kWh": 0.0, "winst_woon_kWh": 0.0, "verlies_module_kWh": 0.0, "verlies_woon_kWh": 0.0, "warmtevoorraad_delta_kWh": 0.0, "buiten_watermodel_h": 0.0}) total.update({"h2_water_verbruik_L": 0.0, "h2_cel_retourwater_L": 0.0}) beginE12, beginE48, beginH2 = E12, E48, massah2 begin_nft, begin_drink = nft, drink poort = MeetPoort(c) sensor = 18.0 dak_open = False niveau_alarm, flow_alarm, misflow = False, False, 0 fallback_count, max_qm = 0, 0.0 h2_ratio = c["h2_elektrisch_kWh_kg"]+c["h2_compressie_kWh_kg"] arch, predict = voorspellers(c, forecast) f_by_dag = {r["doel_dag"]: r for r in forecast} start = datetime.fromisoformat(c["weekstart"]) geldig_plan = 20.0 vorig_contact = False for j in range(len(uren)*3600//dt): sec = j*dt r = {k: float(v) for k, v in uren[sec//3600].items() if k != "tijd"} now = start+timedelta(seconds=sec) dag = now.hour >= 6 and now.hour < 22 if naam == "koud_leeg": r["buiten_C"] = -15 r["pv12_W"] *= .06 r["pv48_W"] *= .06 r["wind48_W"] *= .02 r["b12_C"], r["b48_C"] = -2, -2 if naam == "vol": r["pv12_W"] *= 1.8 r["pv48_W"] *= 1.8 r["wind48_W"] = 2500 if naam == "h2_reserve": r["pv48_W"] *= .06 r["wind48_W"] *= .02 # Het eerste plan is om 20:00 de vorige dag opgeslagen. Nieuwe # plannen lezen uitsluitend gearchiveerde prognosekolommen. if j == 0 or now.hour == 20 and now.minute == 0: doeldag = now.date() if j == 0 else (now+timedelta(days=1)).date() fr = f_by_dag.get(doeldag.isoformat()) if fr and datetime.fromisoformat(fr["plan_tijd"]) > now: fr = None pred = predict(fr) if fr else None reserve = (E48-c["b48_kWh"]*.2)*c["eta_ontlaad"] geldig_plan = 19.0 if beheer and pred and pred[0]-pred[1] > reserve else 20.0 events.append({"tijd": now.isoformat(), "soort": "PLAN", "prognose": pred, "doel_dag": doeldag.isoformat(), "set_C": geldig_plan, "bewijs": "AI-01"}) broker = not (naam == "broker" and 86400 <= sec < 172800) m = {"node": "woon", "owner_boot": eigenaar.boot, "source": "pi", "source_boot": "pi-A", "seq": j, "tijd": sec, "ttl": c["voorstel_ttl_s"], "mode": "NORMAAL", "quality": "goed", "set_C": geldig_plan} if broker: eigenaar.voorstel(m, sec) elif sec >= eigenaar.lease: fallback_count += 1 # Een aparte suite injecteert echte stop-/voedingsgebeurtenissen. # De logische stop grijpt vóór een volgende plantintegratie in; # echte contactorreactietijd vraagt een fysieke proef. stop = naam == "stop_herstel" and 90000 <= sec < 90060 voeding = not (naam == "stop_herstel" and 90120 <= sec < 90180) reset = naam == "stop_herstel" and sec == 90720 verzoek_uit = naam == "stop_herstel" and 90000 <= sec <= 90720 # voeding betreft hier de vrijgavekring; essentiële ventilatie # heeft haar eigen gevoede tak. Het vorige contact wordt teruggelezen. contact_uit = not vorig_contact vrij = eigenaar.stop.stap(stop, voeding, True, reset, verzoek_uit, contact_uit) actief = vrij and not verzoek_uit if naam == "stop_herstel" and sec in (90000, 90060, 90120, 90180, 90720): events.append({"tijd": now.isoformat(), "soort": "VRIJGAVE", "vrij": bool(vrij), "stop": bool(stop), "voeding": bool(voeding), "reset": bool(reset), "bewijs": "V-01"}) sensor += (1-math.exp(-dt/60))*(z[0]-sensor) tele = {"node": "modulelucht", "boot": "mega-A", "seq": j, "tijd": sec, "C": float(sensor), "quality": "goed"} if naam == "sensorbias" and 86400 <= sec < 172800: tele["C"] += 3 quality = poort.meet(tele, sec, float(sensor)) if quality["status"] != "goed": fallback_count += 1 # De lokale modulelus gebruikt dezelfde primaire luchtmeetketen. # Bij afwijzing wordt de convector uitgeschakeld; de simulator # vervangt de defecte meting niet stil door de ware temperatuur. # Deze modelwatcher is lokaal. Zijn ijking en foutgrens zijn nog # proefaannamen; een gedeelde model/sensorbias kan ontsnappen. home_set = eigenaar.doel(sec) lokaal_y = quality["waarde"] if lokaal_y is None: lokaal_y = pm.vorig if pm.vorig is not None else 18.0 qm_req = pm.vraag(18.0 if dag else 16.5, float(lokaal_y), dt) if quality["status"] != "goed": qm_req = 0.0 qh_req = ph.vraag(home_set, float(z[2]), dt) cop = float(np.interp(r["buiten_C"], [-15, -7, 2, 7], [2.9, 3.3, 3.7, 4.6])) qmax = float(np.interp(r["buiten_C"], [-15, 7], [2500, 5300])) qmax = min(qmax, (c["wp_max_elektrisch_W"]-50)*cop) pw.maximum = qmax qwp_req = pw.vraag(c["wp_buffer_set_C"], float(z[3]), dt) wp_req = qwp_req/cop+c["wp_hulp_W"] if max(qwp_req, qm_req, qh_req) > 1 else 0 wp_storing = naam == "wp_storing" and 23*3600 <= sec < 28*3600 # De hydraulische hulp heeft in deze proef een afzonderlijke # voeding: een compressorstoring stopt niet automatisch de pomp. if wp_storing: wp_req = c["wp_hulp_W"] if max(qm_req, qh_req) > 1 else 0 nood_req = 2000.0 if wp_storing and z[2] < 20 else 0.0 recirc_wens = now.minute < 10 # Het tankpeil, niet de totale kringvoorraad, bedient de vlotter. tank_bij_pomp = nft-c["nft_onderweg_L"] if tank_bij_pomp <= c["nft_vlotter_L"]: niveau_alarm = True pomp = recirc_wens and not niveau_alarm and not flow_alarm and actief lek_req = 2*dt/60 if naam == "waterlek" and sec < 7200 else 0 waterverlies = lek_req+(r["nft_opname_L_d"]+r["nft_verdamping_L_d"])*h/24 pompfractie = 1.0 if pomp else 0.0 # Een ideaal onafhankelijk vlottercontact wordt binnen de modelstap # op zijn fysieke drempel geopend. De analytische kruising is geen # claim over werkelijke contactvertraging of leidingterugloop. if pomp and waterverlies > 0 and tank_bij_pomp-waterverlies < 24: pompfractie = max(0, (tank_bij_pomp-24)/waterverlies) niveau_alarm = True events.append({"tijd": (now+timedelta(seconds=pompfractie*dt)).isoformat(), "soort": "NFT_VLOTTER_STOP", "tank_L": 24.0, "bewijs": "W-02"}) flow = 0 if naam == "pompfout" else 4*c["nft_goot_L_min"] if pomp and flow < 3.84: misflow += 1 if misflow >= 2: flow_alarm, pomp = True, False pompfractie = 0.0 events.append({"tijd": now.isoformat(), "soort": "NFT_FLOW_STOP", "bewijs": "W-02"}) lamp_wens = 190*max(0, 1-r["straling_W_m2"]*50/54/200) if dag else 0 if not actief: lamp_wens, qm_req, qh_req = 0.0, 0.0, 0.0 base12 = 6+1.5+(12 if qm_req > 1 else 0)+12*pompfractie soc12 = E12/c["b12_kWh"] if soc12 <= .2+1e-9: lamp_wens = 0 elif soc12 < .4: lamp_wens = min(lamp_wens, max(0, r["pv12_W"]-base12)) a12 = accu(c, E12, "b12", r["pv12_W"], base12+lamp_wens, h, r["b12_C"]) E12, l12, tekort12, cur12, ch12, dis12, loss12, res12 = a12 lamp = max(0, min(lamp_wens, l12-base12)) pomp = pomp and l12 >= base12-1e-6 if not pomp: flow = 0 pompfractie = 0.0 else: flow *= pompfractie # De convector vraagt eigen 12 V-ventilatorvermogen; zonder voeding # wordt ook de warmtestroom nul, geen gratis 800 W. if l12 < base12-1e-6: qm_req = 0 gen = r["pv48_W"]+r["wind48_W"] U48 = c["b48_rail_min_V"] # conservatieve stroomtoets max_bat = min(U48*c["b48_ontlaad_A"]*c["eta_ontlaad"], max(0, E48-c["b48_kWh"]*.2)*1000*c["eta_ontlaad"]/h) fc = 0 if actief and E48 < c["b48_kWh"]*.3 and massah2 > 0: fc = min(c["h2_cel_max_W"], massah2*c["h2_lhv_kWh_kg"]* c["h2_cel_eta_net"]*1000/h) budget = gen+max_bat+fc crit = min(48.0, budget) drink_wens = now.hour == 7 and now.minute < 10 druk_req = c["drink_pomp_W"] if drink_wens and drink > 0 else 0 fridge = 150.0 if now.minute < 20 else 0.0 rest = 150.0 if dag else 75.0 gevraagd = [("druk", druk_req, .81), ("koelkast", fridge, .7), ("warmtepomp", wp_req if actief else 0, .98), ("rest", rest, .95), ("nood", nood_req if actief else 0, 1.0)] u, afgewezen, ac = kies_lasten(c, gevraagd, max(0, budget-crit), U48) p, q, s, dc_ac = ac dc_last = crit+dc_ac # Een brandstofcel werkt alleen voor het werkelijke tekort; geen # gelijktijdige elektrolyse of laden uit de brandstofcel. fc = min(fc, max(0, dc_last-gen)) dm_fc = fc*h/1000/(c["h2_lhv_kWh_kg"]*c["h2_cel_eta_net"]) massah2 -= dm_fc a48 = accu(c, E48, "b48", gen+fc, dc_last, h, r["b48_C"]) E48, l48, tekort48, cur48, ch48, dis48, loss48, res48 = a48 if tekort48 > 1e-6: raise AssertionError("dispatch en accubudget spreken elkaar tegen") el = min(cur48, c["h2_elektrolyser_max_W"], max(0, c["h2_max_kg"]-massah2)*h2_ratio/c["h2_dc_eta"]*1000/h, proceswater/c["h2_water_L_kg"]*h2_ratio/c["h2_dc_eta"]*1000/h) if el < c["h2_elektrolyser_min_W"] or fc > 0 or not actief: el = 0 dm_el = el*h/1000*c["h2_dc_eta"]/h2_ratio proceswater -= dm_el*c["h2_water_L_kg"] massah2 += dm_el cur48 -= el leak_h2 = min(massah2, massah2*c["h2_opslag_lek_per_dag"]*h/24) massah2 -= leak_h2 pv_cur = min(r["pv48_W"], cur48) wind_dump = (cur48-pv_cur)/c["wind_buck_eta"] assert wind_dump <= c["wind_dump_max_W"]+1e-7 wp = u["warmtepomp"] qwp = max(0, wp-c["wp_hulp_W"])*cop if wp_storing: qwp = 0 qnood = u["nood"] if wp == 0: qwp = 0 qm_req, qh_req = 0.0, 0.0 vorig_contact = actief and (wp > 0 or lamp > 0 or pomp or el > 0 or qnood > 0) gains_m = .6*r["straling_W_m2"]+lamp+min(l12-lamp, 20) gains_h = 180+u["koelkast"]+u["rest"]+5 if quality["status"] == "goed": if lokaal_y > 27: dak_open = True elif lokaal_y < 25: dak_open = False extra_H = 100 if dak_open else 0 hwoon = c["Hwoon_W_K"] if crit >= 48 else c["Hwoon_W_K"]-c["Hvent_W_K"] z_voor = z.copy() z, flux, residu = warmtestap(c, z, r["buiten_C"], qm_req, qh_req, qwp, qnood, gains_m, gains_h, extra_H, hwoon, plantstap) pm.terug(flux[0]/dt, dt) ph.terug(flux[1]/dt, dt) pw.terug(qwp, dt) therm = max(therm, abs(residu)) max_qm = max(max_qm, flux[0]/dt) total["warmte_geweigerd_kWh"] += max(0, qwp_req-qwp)*h/1000 total["nft_dienst_tekort_L"] += max(0, (4*c["nft_goot_L_min"] if recirc_wens else 0)-flow)*dt/60 # Waterbalans: recirculatie is een aparte doorzetteller. voor_nft, voor_drink = nft, drink opname = min(nft, r["nft_opname_L_d"]*h/24) nft -= opname evap = min(nft, r["nft_verdamping_L_d"]*h/24) nft -= evap lek = min(nft, lek_req) nft -= lek fill = min(c["nft_max_L"]-nft, 20) if sec == 2*86400+10*3600 else 0 nft += fill if fill > 0 and nft > 28.8: niveau_alarm = False drink_req = c["drink_debiet_L_min"]*dt/60 if drink_wens else 0 verbruik = min(drink, drink_req) if u["druk"] > 0 else 0 drink -= verbruik dfill = min(c["drink_max_L"]-drink, 200) if sec == 5*86400+10*3600 else 0 drink += dfill wres = (nft-voor_nft)-(fill-opname-evap-lek) wres += (drink-voor_drink)-(dfill-verbruik) vals = {"pv12": r["pv12_W"]*h/1000, "pv48": r["pv48_W"]*h/1000, "wind48": r["wind48_W"]*h/1000, "last12": l12*h/1000, "last48": l48*h/1000, "tekort12": tekort12*h/1000, "tekort48": (sum(afgewezen.values())/c["omvormer_eta"]+48-crit)*h/1000, "curtail12": cur12*h/1000, "curtail48": pv_cur*h/1000, "wind_dump_raw": wind_dump*h/1000, "batverlies12": loss12, "laad12_kWh": ch12*h/1000, "laad48_kWh": ch48*h/1000, "batverlies48": loss48, "h2_in": el*h/1000, "h2_uit": fc*h/1000, "h2_gemaakt": dm_el, "h2_verbruikt": dm_fc, "h2_gelekt": leak_h2, "h2_water_verbruik_L": dm_el*c["h2_water_L_kg"], "h2_cel_retourwater_L": dm_fc*c["h2_water_L_kg"], "h2_dcverlies": el*h/1000*(1-c["h2_dc_eta"]), "h2_comp": dm_el*c["h2_compressie_kWh_kg"], "h2_procverlies": dm_el*(c["h2_elektrisch_kWh_kg"]-c["h2_lhv_kWh_kg"]), "h2_celverlies": dm_fc*c["h2_lhv_kWh_kg"]*(1-c["h2_cel_eta_net"]), "qwp": qwp*h/1000, "wp_elektrisch": wp*h/1000, "qmodule": flux[0]/3600000, "qwoon": flux[1]/3600000, "nood_geleverd_kWh": qnood*h/1000, "winst_module_kWh": gains_m*h/1000, "winst_woon_kWh": gains_h*h/1000, "verlies_module_kWh": flux[2]/3600000, "verlies_woon_kWh": flux[3]/3600000, "warmtevoorraad_delta_kWh": float(np.array([c["Ca_J_K"], c["Cb_J_K"], c["Cwoon_J_K"], c["Cwp_J_K"]])@(z-z_voor))/3600000, "buiten_watermodel_h": h if min(z[1], z[3]) <= 0 else 0, "nood_tekort": afgewezen.get("nood", 0)*h/1000, "druk_tekort": afgewezen.get("druk", 0)*h/1000, "drink_tekort_L": drink_req-verbruik, "vent_tekort": (48-crit)*h/1000, "nft_opname_L": opname, "nft_verdamping_L": evap, "nft_lek_L": lek, "nft_fill_L": fill, "drink_fill_L": dfill, "drink_verbruik_L": verbruik, "recirculatie_L": flow*dt/60, "comfort_woon_h": h if z[2] < c["comfort_woon_min_C"] else 0, "comfort_module_h": h if z[0] < (17.5 if 7 <= now.hour < 22 else 15.5) else 0, "elektrisch_residu": abs(res12)+abs(res48), "water_residu": abs(wres)} for k, v in vals.items(): total[k] += v rows.append({"tijd": now.isoformat(), "t_h": sec/3600, "module_C": float(z[0]), "buffer_module_C": float(z[1]), "woon_C": float(z[2]), "buffer_wp_C": float(z[3]), "soc12": E12/c["b12_kWh"], "soc48": E48/c["b48_kWh"], "h2_kg": massah2, "nft_L": nft, "drink_L": drink, "P_W": p, "Q_var": q, "S_VA": s, "I48_A": dc_last/U48, "I48_batterij_A": dis48/(c["eta_ontlaad"]*U48), "I48_laad_A": ch48/U48, "proceswater_L": proceswater, "wp_elektrisch_W": wp, "qwp_W": qwp, "qmodule_W": flux[0]/dt, "nood_W": qnood, "qwoon_W": flux[1]/dt, "lamp_W": lamp, "nft_pomp": int(pomp), "nft_pompfractie": pompfractie, "laad12_W": ch12, "laad48_W": ch48, "ontlaad12_W": dis12, "ontlaad48_W": dis48, "vrijgave": int(vrij), "contact_aan": int(vorig_contact), "kritiek_DC_W": crit, "watermodel_geldig": int(min(z[1], z[3]) > 0), "nft_flow_alarm": int(flow_alarm), "nft_niveau_alarm": int(niveau_alarm), "meet_status": quality["status"], "set_woon_C": home_set, "h2_in_W": el, "h2_uit_W": fc, "wind_dump_W": wind_dump}) E12end, E48end = E12, E48 res12 = total["pv12"]+beginE12-E12end-total["last12"]-total["curtail12"]-total["batverlies12"] res48 = total["pv48"]+total["wind48"]+total["h2_uit"]+beginE48-E48end-total["last48"] res48 -= total["curtail48"]+total["wind_dump_raw"]*c["wind_buck_eta"]+total["h2_in"]+total["batverlies48"] resH = beginH2+total["h2_gemaakt"]-total["h2_verbruikt"]-total["h2_gelekt"]-massah2 resW = begin_nft+total["nft_fill_L"]-total["nft_opname_L"]-total["nft_verdamping_L"]-total["nft_lek_L"]-nft resD = begin_drink+total["drink_fill_L"]-total["drink_verbruik_L"]-drink resP = begin_proceswater-total["h2_water_verbruik_L"]-proceswater resT = total["qwp"]+total["nood_geleverd_kWh"]+total["winst_module_kWh"]+total["winst_woon_kWh"] resT -= total["verlies_module_kWh"]+total["verlies_woon_kWh"]+total["warmtevoorraad_delta_kWh"] assert abs(resT) < 1e-7 assert max(abs(res12), abs(res48), abs(resH), abs(resW), abs(resD), abs(resP)) < 1e-7 summary = {"scenario": naam, "totalen": total, "eind": {"E12_kWh": E12end, "E48_kWh": E48end, "h2_kg": massah2, "nft_L": nft, "drink_L": drink, "proceswater_L": proceswater}, "eind_temperaturen_C": z.tolist(), "minimum": {k: min(a[k] for a in rows) for k in ("module_C", "woon_C", "soc12", "soc48", "nft_L", "drink_L")}, "maximum": {k: max(a[k] for a in rows) for k in ("module_C", "woon_C", "S_VA", "I48_A", "I48_batterij_A", "I48_laad_A", "wind_dump_W")}, "balansen": {"rail12_kWh": res12, "rail48_kWh": res48, "h2_kg": resH, "nft_L": resW, "drink_L": resD, "proceswater_L": resP, "warmte_week_kWh": resT, "warmte_max_residu_J": therm}, "fallback_minuten": fallback_count, "max_modulewarmte_W": max_qm, "eerste_watermodel_ongeldig": next((a["tijd"] for a in rows if not a["watermodel_geldig"]), None), "nft_flow_stop": bool(flow_alarm)} return summary, rows, events def contracttests(c): tests = [] def check(eis, naam, goed, resultaat): assert goed, naam tests.append({"eis": eis, "test": naam, "status": "PASS_LOGICA", "resultaat": resultaat}) o = LokaleEigenaar(c) base = {"node": "woon", "owner_boot": "woon-A", "source": "pi", "source_boot": "pi-A", "seq": 0, "tijd": 0, "ttl": 60, "mode": "NORMAAL", "quality": "goed", "set_C": 19.5} check("V-01", "opstart vergrendeld", o.voorstel(base, 0)["ACK"] == "STOP", "STOP") o.stop.stap(False, True, True, True, True, True) ok = o.voorstel(base, 0) check("D-02", "geldig voorstel", ok["ACK"] == "OK", ok) check("D-02", "dubbel voorstel", o.voorstel(base, 1)["ACK"] == "REPLAY", "REPLAY") for wijzig, nu, reden in (({"seq": 1, "tijd": -60}, 0, "VERLOPEN_TOEKOMST"), ({"seq": 1, "mode": "ONDERHOUD"}, 0, "MODE"), ({"seq": 1, "owner_boot": "woon-oud"}, 0, "IDENTITEIT_BOOT"), ({"seq": 1, "source_boot": "pi-B"}, 0, "IDENTITEIT_BOOT"), ({"seq": 1, "set_C": float("nan")}, 0, "NIET_EINDIG"), ({"seq": 1, "set_C": 24}, 0, "LIMIET")): r = o.voorstel(dict(base, **wijzig), nu)["ACK"] check("D-02", "weigering "+reden, r == reden, r) check("D-03", "lease verlopen/broker weg", o.doel(60) == 20, o.doel(60)) opgeslagen = json.loads(json.dumps(o.register)) nieuw = LokaleEigenaar(c, boot="woon-B", register=opgeslagen) nieuw.stop.stap(False, True, True, True, True, True) r = nieuw.voorstel(dict(base, owner_boot="woon-B"), 0)["ACK"] check("D-04", "replay na eigenaarherstart", r == "REPLAY", r) g = StopVrijgave() inputs = [(False, True, True, True, True, True), (True, True, True, False, False, False), (False, True, True, False, True, True), (False, False, True, False, True, True), (False, True, True, False, True, True), (False, True, True, True, False, True), (False, True, True, True, True, True), (False, True, True, False, True, True), (False, True, True, True, True, False), (False, True, True, False, True, True), (False, True, True, True, True, True)] trace = [g.stap(*a) for a in inputs] check("V-01", "stop/voeding/resetflank/contact", trace == [True, False, False, False, False, False, False, False, False, False, True], trace) for naam, m, now, pred, want in ( ("ontbreekt", None, 0, 18, "ONTBREEKT"), ("stale", {"tijd": 0, "C": 18}, 91, 18, "STALE_KLOK"), ("NaN", {"tijd": 0, "C": float("nan")}, 0, 18, "NIET_EINDIG"), ("ongeldig model", {"tijd": 0, "C": 18}, 0, float("nan"), "ONGELDIG_MODEL"), ("bias/innovatie", {"tijd": 0, "C": 21}, 0, 18, "INNOVATIE")): pp = MeetPoort(c) telegram = None if m is None else dict(node="modulelucht", boot="mega-A", seq=0, quality="goed", **m) r = pp.meet(telegram, now, pred) check("D-01", naam, r["status"] == want and r["waarde"] is None, r) pp = MeetPoort(c) for i in range(12): r = pp.meet(dict(node="modulelucht", boot="mega-A", seq=i, quality="goed", tijd=60*i, C=18), 60*i, 18+.02*i) check("D-01", "freeze met verwachte verandering", r["status"] == "FREEZE", r) pp = MeetPoort(c) r = pp.meet(dict(node="modulelucht", boot="mega-A", seq=0, quality="goed", tijd=0, C=19), 0, 19) check("D-01", "gedeelde bias niet observeerbaar", r["status"] == "goed", "aanvaard; absolute juistheid NIET bewezen") check("E-03", "module kou", not laad_toegestaan(c, "b12", -2, 13, []), "laadstroom nul") check("E-03", "station kou", not laad_toegestaan(c, "b48", -2, 51.2, [3.2]*16), "laadstroom nul") check("E-04", "16S 58.4V mismatch", not laad_toegestaan(c, "b48", 15, 58.4, [3.65]*16), "geweigerd") check("E-04", "celongelijkheid", not laad_toegestaan(c, "b48", 15, 51.2, [3.2]*15+[3.6]), "geweigerd") v = accu(c, c["b48_kWh"]*.9, "b48", 5000, 0, 1/60, 15) check("E-01", "volle accu", v[4] == 0 and v[3] == 5000, "geen lading; 5000W ongebruikt") v = accu(c, c["b48_kWh"]*.2, "b48", 0, 100, 1/60, 15) check("E-01", "lege bruikbare accu", v[5] == 0 and v[2] == 100, "100W diensttekort") v = accu(c, c["b48_kWh"]*.5, "b48", 7000, 0, 1/60, 15) check("E-03", "laadlimiet los van ontladen", abs(v[4]/44.8-60) < 1e-9, {"laad_A": v[4]/44.8, "afregelen_W": v[3]}) gevraagd = [("druk", 750, .81), ("koelkast", 150, .7), ("warmtepomp", c["wp_max_elektrisch_W"], .98), ("rest", 300, .95), ("nood", 2000, 1)] voor = pq([(w, pf) for _, w, pf in gevraagd]) u, no, ac = kies_lasten(c, gevraagd, 7000, 44) check("E-02", "4.48kVA verzoek", voor[2] > 4000 and u["nood"] == 0 and ac[2] <= 4000, {"gevraagd_PQS": voor, "geleverd_PQS_DC": ac, "afgewezen_W": no, "duur_s": 60, "nood_niet_geleverd_kWh": no["nood"]/60000}) p, q, s = 5232, math.sqrt(7280**2-5232**2), 7280 I = (p/c["omvormer_eta"]+c["omvormer_nullast_W"]+48)/(44.8*c["eta_ontlaad"]) check("E-02", "5s pompaanloop te hoge bankstroom", s < 8000 and I > 130, {"P_W": p, "Q_var": q, "S_VA": s, "I44_8_A": I, "duur_s": 5, "dienst_status": "NIET_GEHAALD"}) p2 = 5000 I2 = (p2/c["omvormer_eta"]+c["omvormer_nullast_W"]+48)/(44.8*c["eta_ontlaad"]) check("E-02", "lagere selectieproef binnen envelope", I2 < 130, {"P_W": p2, "S_VA": 7280, "I44_8_A": I2, "duur_s": 5, "status": "SELECTIEPROEF_GEEN_MOTORBEWIJS"}) # Dit toetst de fysieke leemte, geen veilige overtoerenproef. dump = 245**2/c["wind_dump_R_ohm"] tests.append({"eis": "E-05", "test": "245V direct op dump", "status": "NIET_FYSISCH_AANGETOOND", "resultaat_W": dump, "grens_W": c["wind_dump_max_W"]}) return tests def antiwindup_proef(): rows = [] for aw in (True, False): p = PID(100, .5, 0, 800, tau_aw=120) I = [] for k in range(180): t = k*60 p.vraag(25 if t < 7200 else 20, 20, 60) # De actuator is 2 h uitgevallen; de fout blijft bestaan. actual = 0 if t < 7200 else min(800, max(0, p.onbegrensd)) p.terug(actual, 60, aw=aw) I.append((t, p.I, actual)) rows.append({"aw": aw, "I_op_2h_W": I[119][1], "actuator_na_herstel_W": I[120][2], "trace": I}) return rows def lca(c, B, I, leven=10): H = c["lca_horizon_jaar"] posten = [("hout", 1200, .26, 30), ("staal", 300, 2.5, 30), ("beton", 2000, .12, 30), ("isolatie", 150, 3, 30), ("modulegebouw", 1, 650, 30), ("modulerek/water", 1, 110, 15), ("PV12", 1.4, 600, 25), ("PV48", 4.2, 600, 25), ("batterij12", 3.072, 100, 8), ("lamp12", 2, 50, 8), ("elektronica12", 1, 60, 10), ("wind", 1, 650, 20), ("WP", 1, 800, 15), ("H2", 1, 1000, 10), ("elektronica48", 1, 150, 10), ("batterij48", B, 100, leven)] inventaris = [{"post": n, "hoeveelheid": q, "factor": f, "leven_jaar": L, "aantal": math.ceil(H/L), "kg": q*f*math.ceil(H/L)} for n, q, f, L in posten] overige = {"transport": 300, "eind": 500, "verbruiksmateriaal": 30*H, "netimport": I*.15*H, "overig_gebruik": 15*H} totaal = sum(r["kg"] for r in inventaris)+sum(overige.values()) return {"B_kWh": B, "import_kWh_jaar": I, "leven_jaar": leven, "inventaris": inventaris, "overige": overige, "kg_20jaar": totaal, "kg_jaar": totaal/H} def csv_text(rows): b = io.StringIO(newline="") w = csv.DictWriter(b, list(rows[0]), lineterminator="\n") w.writeheader() w.writerows(rows) return b.getvalue() def run_all(c, uurbron, forecastbron, programma_sha256=None): checksum = valideer(c) uren, arch = lees_csv(uurbron), lees_csv(forecastbron) for i, r in enumerate(uren): assert datetime.fromisoformat(r["tijd"]) == datetime.fromisoformat(c["weekstart"])+timedelta(hours=i) assert all(finite(float(v)) for k, v in r.items() if k != "tijd") model, predict = voorspellers(c, arch) assert predict(dict(arch[-1], prognose_T_C="")) is None assert predict(dict(arch[-1], prognose_T_C="-25")) is None summaries, logs, events = {}, {}, {} for naam in ("normaal", "koud_leeg", "vol", "broker", "sensorbias", "stop_herstel", "wp_storing", "waterlek", "pompfout", "drink_leeg", "h2_reserve", "h2_water_leeg"): s, r, e = simuleer(c, uren, arch, naam) summaries[naam], logs[naam], events[naam] = s, r, e normaal = logs["normaal"] verfijning = {} for dt in (60, 15): s, r, e = simuleer(c, uren, arch, plantstap=dt) fout = max(abs(a[k]-b[k]) for a, b in zip(normaal, r) for k in ("module_C", "buffer_module_C", "woon_C", "buffer_wp_C")) verfijning[str(dt)] = {"max_temperatuurverschil_K": fout, "wp_kWh_verschil": s["totalen"]["wp_elektrisch"]-summaries["normaal"]["totalen"]["wp_elektrisch"]} assert verfijning["15"]["max_temperatuurverschil_K"] < .02 onafh = antiwindup_proef() result = {"schema": SCHEMA, "config_id": c["config_id"], "config_sha256": checksum, "klok": c["klok"], "seed": c["seed"], "programma_sha256": programma_sha256, "weekinvoer_sha256": hashlib.sha256(uurbron.encode()).hexdigest(), "forecastinvoer_sha256": hashlib.sha256(forecastbron.encode()).hexdigest(), "week": summaries, "forecast": model, "ai_onthouding": {"ontbrekend": predict(dict(arch[-1], prognose_T_C="")), "buiten_proefgebied": predict(dict(arch[-1], prognose_T_C="-25")), "status": "ONTHOUDEN_LOKAAL_20_C"}, "stapverfijning": verfijning, "contracttests": contracttests(c), "antiwindup": onafh, "lca": {"S1": lca(c, 10, 200), "S2": lca(c, c["b48_kWh"], c["lca_import_scenario_kWh_jaar"]), "gevoeligheid": [lca(c, B, I, L) for B in (8, 12, 16) for I in (200, 500, 1000) for L in (8, 10)]}} # JSON mag geen NaN/Infinity bevatten; onthouding is null + status. json.dumps(result, allow_nan=False) return result, logs, events def kort(result): regels = ["E49-S2-12-v1 | UTC+1 vast | seed 4902026", "config sha256 "+result["config_sha256"]] for naam, r in result["week"].items(): t = r["totalen"] regels.append(f'{naam}: last12={t["last12"]:.6f} kWh ' f'last48={t["last48"]:.6f} kWh ' f'comforttekort woon={t["comfort_woon_h"]:.3f} h ' f'module={t["comfort_module_h"]:.3f} h') regels.append("forecast test model/basis RMSE kWh "+ str(result["forecast"]["scores"]["test"])) regels.append("contracttests "+str(len(result["contracttests"]))) regels.append("LCA S1/S2 kg/jaar "+str([result["lca"][k]["kg_jaar"] for k in ("S1", "S2")])) return "\n".join(regels)+"\n" def main(): p = argparse.ArgumentParser() p.add_argument("--config", default="config.json") p.add_argument("--week", default="week_uur.csv") p.add_argument("--forecast", default="forecast_archief.csv") p.add_argument("--out", default="resultaat") a = p.parse_args() c = json.loads(Path(a.config).read_text(encoding="utf-8")) result, logs, events = run_all(c, Path(a.week).read_text(encoding="utf-8"), Path(a.forecast).read_text(encoding="utf-8"), hashlib.sha256(Path(__file__).read_text(encoding="utf-8").encode()).hexdigest()) out = Path(a.out) out.mkdir(parents=True, exist_ok=True) (out/"resultaten.json").write_text(json.dumps(result, indent=2, ensure_ascii=False, allow_nan=False), encoding="utf-8", newline="\n") for naam, r in logs.items(): (out/(naam+".csv")).write_text(csv_text(r), encoding="utf-8", newline="") (out/(naam+"_events.json")).write_text(json.dumps(events[naam], ensure_ascii=False, indent=2, allow_nan=False), encoding="utf-8", newline="\n") (out/"uitvoer.txt").write_text(kort(result), encoding="utf-8", newline="\n") print(kort(result), end="") if __name__ == "__main__": main()