Node-RED flowt — rakenne ja toteutus

Tämä ohje kattaa EnergyHubin Node-RED-flowjen rakenteen ja toteutuksen. Ohje olettaa että Node-RED on asennettuna ja MQTT-yhteys Mosquittoon toimii. Arkkitehtuuritausta löytyy sarjan osista Osa 8 — MQTT ja Osa 9 — Hintaohjauksen flow.

Kolme flow’ta, kolme vastuuta

EnergyHub käyttää Node-REDissä kolmea erillistä flow’ta joilla on tarkasti rajatut vastuut:

EH – State Collector — kerää kaiken datan ja ylläpitää tilannekuvaa. Ei tee päätöksiä.

EH – Decision Engine — ajaa päätöslogiikan minuutin välein. Lukee tilannekuvan, tekee päätökset, julkaisee komennot.

EH – Safety Guardian — valvoo vaihevirrat jokaisen saapuvan viestin yhteydessä, käytännössä noin 10 sekunnin syklissä. Reagoi nopeammin kuin optimointilogiikka. Sen tehtävä ei ole optimoida, vaan estää ohjattujen kuormien pahentamasta kuormitustilannetta.

Tämä jako on arkkitehtuurinen valinta: data, päätökset ja turvallisuus ovat toisistaan riippumattomia. Safety Guardian toimii vaikka Decision Engine olisi pysähdyksissä.

Yksi sääntö ennen koodia: käytä ??, älä ||

Tässä ohjeessa kaikki parametrien oletusarvot käyttävät ??-operaattoria (nullish coalescing), ei ||-operaattoria. Ero on kriittinen: || kohtelee arvoa 0 falsy-arvona ja korvaa sen oletuksella, vaikka 0 olisi täysin validi mittausarvo. Esimerkiksi global.get('m_phase_l1_a') || 0 peittäisi todellisen nollavirran, ja negatiivisilla hintakynnyksillä || käyttäytyy väärin. ?? korvaa vain null– ja undefined-arvot — juuri ne tapaukset joissa data oikeasti puuttuu.

EH – State Collector

Rakenne

[mqtt in: Kaikki EnergyHub] → [function: Päivitä global context]
[inject: Tarkista konteksti] → [function: Lue konteksti] → [debug]

MQTT-tilaaja

Node kuuntelee kaikkia EnergyHub-topiceja:

• Topic: energyhub/#
• QoS: 0
• Output: Parsed JSON object

Päivitä global context -funktio

State Collector päivittää global contextin jokaisesta saapuvasta MQTT-viestistä:

javascript

const topic = msg.topic;
const payload = msg.payload;

// Verkko
if (topic === 'energyhub/telemetry/grid') {
    global.set('m_grid_power_w', payload.power_w);
    global.set('m_grid_export_w', payload.export_w);
    global.set('m_phase_l1_a', payload.l1_a);
    global.set('m_phase_l2_a', payload.l2_a);
    global.set('m_phase_l3_a', payload.l3_a);
}
// Aurinko
if (topic === 'energyhub/telemetry/pv') {
    global.set('m_pv_power_w', payload.power_w);
    global.set('m_pv_daily_kwh', payload.daily_kwh);
}
// Hinnat
if (topic === 'energyhub/telemetry/prices') {
    global.set('m_spot_price_eur_kwh', payload.spot_eur_kwh);
    global.set('m_buy_price_eur_kwh', payload.buy_eur_kwh);
}
// Lämpöpumppu
if (topic === 'energyhub/telemetry/heatpump') {
    global.set('m_hp_outdoor_temp_c', payload.outdoor_temp_c);
    global.set('m_hp_tap_water_top_c', payload.tap_water_top_c);
    global.set('m_hp_compressor_speed_pct', payload.compressor_speed_pct);
    global.set('m_hp_currently_running', payload.currently_running);
}
// Järjestelmän tila
if (topic === 'energyhub/system/mode') {
    global.set('sys_operating_mode', payload);
}
// Safety trip — OMA topic, ainoa omistaja
if (topic === 'energyhub/system/safety_trip') {
    global.set('sys_safety_trip', payload);
}
// Capabilityt — EI sisällä safety_trippiä
if (topic === 'energyhub/system/capabilities') {
    global.set('c_ev_charge_allowed', payload.c_ev_charge_allowed);
    global.set('c_hp_evu_allowed', payload.c_hp_evu_allowed);
    global.set('c_hp_boost_allowed', payload.c_hp_boost_allowed);
    global.set('c_pv_curtail_allowed', payload.c_pv_curtail_allowed);
}

Arkkitehtuurihuomio — sys_safety_trip luetaan vain omasta topicistaan. Aiemmassa toteutuksessa sys_safety_trip luettiin capabilities-viestistä (payload.sys_safety_trip). Tämä osoittautui zombie-latch-virheeksi: capabilities-viesti kantoi turvalaukaisun tilaa, jolloin vanha retained-viesti saattoi herättää jo nollatun laukaisun henkiin. Turvalaukaisulla on yksi ainoa omistaja — retained-topic energyhub/system/safety_trip. Mikään muu viesti ei saa kantaa tätä kenttää. Sama periaate koskee kaikkia tilan omistajuuksia: yksi topic, yksi omistaja.

Miksi global context eikä flow context? Global context on kaikkien flow’jen yhteinen muisti. Decision Engine ja Safety Guardian voivat lukea State Collectorin tallentamat arvot suoraan ilman MQTT-tilausta.

EH – Decision Engine

Rakenne

[inject: 1 min trigger]
    → [function: Lue järjestelmän tila]
    → [function: Priority Resolver]
    → [function: Julkaise komennot] → [mqtt out]
    → [function: Kirjaa päätökset] → [mqtt out: observability]

1 min trigger

Inject-node laukaisee flow’n kerran minuutissa:

• Repeat: interval, every 1 minute
• Payload: timestamp

Lue järjestelmän tila

Rakentaa tilaobjektin s global contextista ja laskee johdetut arvot:

javascript

const s = {
    mode: global.get('sys_operating_mode') ?? 'normal',
    safetyTrip: global.get('sys_safety_trip') ?? false,

    cap: {
        evAllowed: global.get('c_ev_charge_allowed') !== false,
        hpEvuAllowed: global.get('c_hp_evu_allowed') !== false,
        hpBoostAllowed: global.get('c_hp_boost_allowed') !== false,
        pvCurtailAllowed: global.get('c_pv_curtail_allowed') !== false,
    },

    gridPowerW: global.get('m_grid_power_w') ?? 0,
    pvPowerW: global.get('m_pv_power_w') ?? 0,
    spotPrice: global.get('m_spot_price_eur_kwh') ?? 0,
    phaseL1: global.get('m_phase_l1_a') ?? 0,
    phaseL2: global.get('m_phase_l2_a') ?? 0,
    phaseL3: global.get('m_phase_l3_a') ?? 0,
    tapWaterC: global.get('m_hp_tap_water_top_c') ?? 0,
    compressorPct: global.get('m_hp_compressor_speed_pct') ?? 0,
};

// Johdetut arvot
s.maxPhaseA = Math.max(Math.abs(s.phaseL1), Math.abs(s.phaseL2), Math.abs(s.phaseL3));
// Aurinkoylijäämä — sama kaava kuin Osa 10:ssä:
// PV-tuotanto miinus se osa josta ostetaan verkosta
s.solarExcessW = Math.max(0, s.pvPowerW - Math.max(0, s.gridPowerW));
s.isSolarExcess = s.solarExcessW > (global.get('p_solar_excess_w') ?? 2000);
s.isCheapPrice = s.spotPrice < (global.get('p_cheap_price') ?? 0.04);
s.isNegativePrice = s.spotPrice < (global.get('p_negative_price') ?? -0.02);
s.isExpensivePrice = s.spotPrice > (global.get('p_expensive_price') ?? 0.15);
s.isPeakLoad = s.maxPhaseA > (global.get('p_peak_load_a') ?? 25);

msg.state = s;
return msg;

Huomaa että isPeakLoad käyttää nyt suoraa ampeerirajaa (p_peak_load_a, oletus 25 A) eikä sulakerajan kerrointa. Tämä pitää Decision Enginen ja Safety Guardianin kynnykset luettavina ja erikseen säädettävinä.

Priority Resolver

Käy tilanteen läpi kiinteässä järjestyksessä — turvallisuus ensin, talous viimeisenä:

javascript

const s = msg.state;
const decisions = {};
const reasons = {};

// 1. EMERGENCY — kaikki alas
if (s.safetyTrip || s.mode === 'emergency') {
    msg.decisions = { ev: false, hpMode: 'block',
                      pvCurtailPct: null, saunaAllowed: false };
    msg.reasons = { all: 'EMERGENCY tai safety_trip' };
    return msg;
}

// 2. MANUAL OVERRIDE — ei muutoksia
if (s.mode === 'manual_override') {
    msg.decisions = null;
    msg.reasons = { all: 'Manual override' };
    return msg;
}

// 3. FALLBACK — minimiparametrit
if (s.mode === 'fallback') {
    msg.decisions = { ev: true, hpMode: 'normal',
                      pvCurtailPct: null, saunaAllowed: true };
    msg.reasons = { all: 'Fallback' };
    return msg;
}

// 4. PEAK PROTECTION — pudota kuormat
if (s.mode === 'peak_protection' || s.isPeakLoad) {
    decisions.ev = false;
    reasons.ev = 'PEAK: vaihevirta ' + s.maxPhaseA.toFixed(1) + 'A';
    decisions.hpMode = s.compressorPct > 60 ? 'block' : 'normal';
    decisions.pvCurtailPct = null;
    msg.decisions = decisions;
    msg.reasons = reasons;
    return msg;
}

// 5. NORMAALI OPTIMOINTI
// EV-lataus
if (!s.cap.evAllowed) {
    decisions.ev = false;
    reasons.ev = 'Capability pois';
} else if (s.isSolarExcess) {
    decisions.ev = true;
    reasons.ev = 'Aurinkoylijäämä: ' + Math.round(s.solarExcessW) + 'W';
} else if (s.isCheapPrice) {
    decisions.ev = true;
    reasons.ev = 'Halpa hinta: ' + s.spotPrice.toFixed(3) + ' €/kWh';
} else if (s.isExpensivePrice) {
    decisions.ev = false;
    reasons.ev = 'Kallis hinta: ' + s.spotPrice.toFixed(3) + ' €/kWh';
} else {
    decisions.ev = false;
    reasons.ev = 'Normaali — ei lataustarvetta';
}

// HP-moodi
if (!s.cap.hpEvuAllowed) {
    decisions.hpMode = 'normal';
    reasons.hp = 'Capability pois';
} else if (s.isNegativePrice && s.cap.hpBoostAllowed) {
    decisions.hpMode = 'boost';
    reasons.hp = 'Negatiivinen hinta: ' + s.spotPrice.toFixed(3);
} else if (s.isExpensivePrice) {
    decisions.hpMode = 'block';
    reasons.hp = 'Kallis hinta: ' + s.spotPrice.toFixed(3);
} else {
    decisions.hpMode = 'normal';
    reasons.hp = 'Normaali';
}

// PV-rajoitus
if (s.isNegativePrice && s.pvPowerW > 500 && s.cap.pvCurtailAllowed) {
    const evLoad = decisions.ev ? 3600 : 0;
    // Arvioi oma hetkellinen kulutus: PV-tuotanto − vienti (gridPowerW negatiivinen = vienti)
    // + mahdollisesti käynnistettävä EV-kuorma
    const ownLoad = Math.max(300, s.pvPowerW + Math.min(0, s.gridPowerW) + evLoad);
    const pct = Math.max(10, Math.min(95, Math.round((ownLoad / s.pvPowerW) * 100)));
    decisions.pvCurtailPct = pct;
    reasons.pv = 'Negatiivinen hinta: rajoitus ' + pct + '%';
} else {
    decisions.pvCurtailPct = null;
    reasons.pv = 'Ei rajoitusta';
}

msg.decisions = decisions;
msg.reasons = reasons;
return msg;  // msg haarautuu sekä Julkaise komennot- että Kirjaa päätökset -nodeille

Julkaise komennot

Retain-huomio: Komennot julkaistaan retain: true — ne säilyvät brokerissa ja toistuvat uusille tilaajille. Fyysistä ohjausta tekevän vastaanottajan pitää tarkistaa komennon tuoreus, järjestelmätila ja safety-ehdot ennen toimintaa.

javascript

if (!msg.decisions) return null;
const d = msg.decisions;

// String(d.ev) tuottaa 'true'/'false' — HA-automaatiot odottavat tätä muotoa
node.send({ topic: 'energyhub/command/ev/allowed',
            payload: String(d.ev), retain: true });

if (d.hpMode) {
    // HP-moodi julkaistaan non-retained — komento on hetkellinen ohjaus, ei pysyvä tila
    node.send({ topic: 'energyhub/command/hp/mode',
                payload: d.hpMode, retain: false });
}

if (d.pvCurtailPct !== null) {
    // PV-rajoitus julkaistaan merkkijonona — muuta numeroksi jos vastaanottaja sitä odottaa
    node.send({ topic: 'energyhub/command/pv/curtail_pct',
                payload: String(d.pvCurtailPct), retain: true });
}
return null;

Kirjaa päätökset (observability)

javascript

if (!msg.decisions) return null;
const logEntry = {
    ts: new Date().toISOString(),
    mode: msg.state.mode,
    grid_w: msg.state.gridPowerW,
    pv_w: msg.state.pvPowerW,
    spot_eur: msg.state.spotPrice,
    max_phase_a: msg.state.maxPhaseA,
    decisions: msg.decisions,
    reasons: msg.reasons
};

node.send({ topic: 'energyhub/observability/decisions',
            payload: logEntry, retain: false });
return null;

EH – Safety Guardian

Rakenne

[mqtt in: energyhub/telemetry/phases]
    → [function: Sulakevalvonta]
    → [mqtt out: → peak_protection]
    → [mqtt out: → normal]
    → [mqtt out: → safety trip]

Safety Guardian ei käytä inject-nodea — se reagoi aina kun uusi vaihevirtaviesti saapuu. Käytännössä tämä vastaa noin 10 sekunnin valvontasykliä, jos vaihevirtadata julkaistaan 10 sekunnin välein.

Kynnysarvot

Vahvistetut tuotantokynnykset ovat kiinteät ampeeriarvot, eivät sulakerajan kertoimia:

• warn 30 A → peak_protection
• reduce 36 A → nopeampi kuormanpudotus
• trip 40 A → safety trip (ehdoton viimeinen suoja)
• hystereesi 25 A → paluu normaaliin

Kynnykset eivät olleet alusta asti tällaiset. Suunnitteluvaiheessa lähdettiin konservatiivisemmista kertoimista (esim. 90 % ja 98 % sulakerajasta), mutta ne reagoivat normaaleihin kulutuspiikkeihin liian herkästi. Kiinteät 30 / 36 / 40 A antavat pelivaraa normaalille kuormalle ja puuttuvat silti tilanteeseen selvästi ennen 25 A -sulakkeen todellista laukeamista.

Sulakevalvonta-funktio

javascript

const d = msg.payload;
const l1 = Math.abs(d.l1_a ?? 0);
const l2 = Math.abs(d.l2_a ?? 0);
const l3 = Math.abs(d.l3_a ?? 0);
const maxA = Math.max(l1, l2, l3);

const currentMode = global.get('sys_operating_mode') ?? 'normal';
const safetyTripActive = global.get('sys_safety_trip') ?? false;

const warnLimit  = global.get('p_guard_warn_a')  ?? 30;
const tripLimit  = global.get('p_guard_trip_a')  ?? 40;
const hysteresis = global.get('p_guard_reset_a') ?? 25;

flow.set('last_max_a', maxA);

// SAFETY TRIP — ehdoton, tarkistetaan AINA ennen mitään muuta
if (maxA >= tripLimit) {
    global.set('sys_safety_trip', true);
    return [
        null,
        null,
        { topic: 'energyhub/system/safety_trip',
          payload: true, retain: true }
    ];
}

// AUTO-RESET — kun trip on päällä mutta virta laskenut turvarajan alle
if (safetyTripActive) {
    if (maxA < hysteresis) {
        global.set('sys_safety_trip', false);
        return [
            null,
            { topic: 'energyhub/system/mode',
              payload: 'normal', retain: true },
            { topic: 'energyhub/system/safety_trip',
              payload: false, retain: true }
        ];
    }
    // Trip yhä voimassa, virta ei vielä laskenut riittävästi
    return [null, null, null];
}

// VAROITUS → peak_protection
if (maxA >= warnLimit && currentMode !== 'peak_protection') {
    return [
        { topic: 'energyhub/system/mode',
          payload: 'peak_protection', retain: true },
        null,
        null
    ];
}

// PALUU NORMAALIIN — hystereesi
if (currentMode === 'peak_protection' && maxA < hysteresis) {
    return [
        null,
        { topic: 'energyhub/system/mode',
          payload: 'normal', retain: true },
        null
    ];
}

return [null, null, null];

Miksi auto-reset on koodattu tällä tavalla — opetus tuotannosta. Aiemmassa versiossa Guardianin alussa oli rivi if (safetyTripActive) return [null, null, null];. Se vaikutti turvalliselta: ”jos trip on päällä, älä tee mitään.” Käytännössä se oli dead-code deadlock — auto-reset jäi tämän ehdon taakse saavuttamattomiin, eikä järjestelmä koskaan palautunut itsestään ylivirtatilanteen poistuttua. Sama incidentti paljasti myös kaksi muuta juurisyytä: capabilities-viestistä luettu zombie-latch (ks. State Collector yllä) ja auto-resetin liian aikainen ev_allowed: true. Korjattu rakenne tarkistaa trip-rajan aina ensin, ja vasta sitten arvioi onko trip voimassa ja onko virta laskenut hystereesin alle. Tämä on suoraa materiaalia sarjan Osa 15:lle, joka käsittelee käyttöönoton paljastamia piiloja ja niiden systemaattista korjausta.

Kolme ulostuloa vastaa kolmea mqtt out -nodea: peak_protection, normal, safety_trip. Vain yksi ulostuloista aktivoituu kerrallaan.

Asennus ja konfigurointi

MQTT-broker

Lisää Node-REDiin MQTT-broker-yhteys (Manage palette → ei tarvita, se tulee oletuksena):

• Server: localhost
• Port: 1883
• Client ID: nodered-energyhub

Global context -tallennustapa

Node-REDin settings.js:ssä varmista että contextStorage on konfiguroitu. EnergyHubissa käytetään file-moodia, jotta konteksti säilyy uudelleenkäynnistyksen yli:

javascript

contextStorage: {
    default: { module: "localfilesystem" },
    memory: { module: "memory" }
}

Vaikka konteksti tallentuisi levylle, turvalaukaisun ja moodin retained-MQTT-viestit ovat aina lopullinen totuus — ne palauttavat tilan brokerista käynnistyksen yhteydessä.

Flowjen tuonti

Flowt voidaan exportoida ja importoida JSON-muodossa:

• Export: Node-RED valikko → Export → Download
• Import: Node-RED valikko → Import → paste tai tiedosto

Node-RED-muutoksissa kannattaa suosia JSON-tietoista lähestymistä: parsitaan flows.json JSONina ja kohdistetaan muutos noden id:n perusteella, raa’an tekstihaun sijaan.

Tämä flow ei ohjaa fyysisiä laitteita

Decision Engine julkaisee päätökset MQTT-komentoina. Varsinainen fyysinen ohjaus — Shelly-releen kytkentä, Modbus-rekisterin kirjoitus, invertterin rajoituskomento — tapahtuu erillisessä integraatiokerroksessa (Home Assistant). Node-RED päättää, HA toteuttaa.

Topic-rakenne

Topic	Suunta	Sisältö
energyhub/telemetry/grid	sisään	verkko, vienti, vaihevirrat
energyhub/telemetry/pv	sisään	PV-teho, päivätuotanto
energyhub/telemetry/prices	sisään	spot-hinta, ostohinta
energyhub/telemetry/heatpump	sisään	lämpötilat, kompressori
energyhub/telemetry/phases	sisään	vaihevirrat (Safety Guardian)
energyhub/system/mode	sisään/ulos	käyttötila
energyhub/system/capabilities	sisään	capability-flagit (ei safety_trip)
energyhub/system/safety_trip	sisään/ulos	turvalaukaisu (oma omistaja)
energyhub/command/ev/allowed	ulos	EV-latauksen lupa
energyhub/command/hp/mode	ulos	HP-moodi (normal/block/boost)
energyhub/command/pv/curtail_pct	ulos	PV-rajoitusprosentti
energyhub/observability/decisions	ulos	päätösloki syineen

Fail-safe oletukset

• Jos data puuttuu, ei käynnistetä uusia kuormia — capability-oletukset johtavat turvalliseen tilaan
• Jos safety trip on päällä, koko optimointilogiikka pysähtyy (Priority Resolver palauttaa EMERGENCY-haaran)
• Jos Node-RED kaatuu, viimeiset retain-komennot jäävät voimaan — tämä on tiedostettu rajoite (watchdog kehitysvaiheessa)
• Fyysinen suojaus ei saa riippua Node-REDistä — sulakkeet ja laitteiden omat suojat ovat ensisijaisia

Debuggaus

Tarkista global context: Node-RED → Context Data → Global — näyttää kaikki tallennetut arvot reaaliajassa.

Seuraa MQTT-liikennettä:

bash

mosquitto_sub -h localhost -t "energyhub/#" -v

Tarkista päätökset:

bash

mosquitto_sub -h localhost -t "energyhub/observability/decisions" -C 1 | python3 -m json.tool

Safety trip -kuittaus:

bash

# Nollaa safety trip manuaalisesti
mosquitto_pub -h localhost -t "energyhub/system/safety_trip" -m "false" -r
mosquitto_pub -h localhost -t "energyhub/system/mode" -m "normal" -r

Sama onnistuu myös ilman komentoriviä Home Assistantin Developer Tools -näkymästä: input_boolean.sys_safety_trip ja input_select.sys_operating_mode.

Pakota Decision Engine manuaalisesti: Inject-noden ”Inject once” -painike ajaa flow’n heti ilman minuutin odotusta.