simulate_readers.py

#!/usr/bin/python3.11
"""
simulate_readers.py

Simuleert de externe reader-/actuator-scripts die in een echte installatie
apart draaien (warmtepomp-bus, weerstation, laadpaal-meter, PV-omvormer).
Schrijft periodiek plausibele testwaarden naar reader_data.json, zodat je
E_manager.py (en optioneel p1_reader.py ernaast) kunt testen zonder de
echte hardware/scripts.

NIET bedoeld voor productiegebruik - alleen om E_manager.py te kunnen
draaien en observeren voordat je eigen WP/LP/PV-scripts er zijn.

Gebruik:
    nohup python3.11 simulate_readers.py > /dev/null 2>&1 &

Let op p1_reader: als je p1_reader.py ook laat draaien, schrijft dat
script zelf al de "p1_reader"-key naar reader_data.json - dit dummy-script
simuleert die key ALLEEN als p1_reader.py niet meedraait (zie
SIMULATE_P1_READER hieronder).
"""
import random
from datetime import datetime
from time import sleep

from shared_store import set_key
from logger_setup import get_logger
from config import READER_DATA_JSON, SETPOINTS_JSON, ENERGY_JSON

logger = get_logger("E_manager")  # geen eigen aparte log-service voor een dummy-script

SIM_INTERVAL = 15          # seconden tussen updates
SIMULATE_P1_READER = True  # zet op False als p1_reader.py zelf al draait

# De webapp/energy-dashboard verwacht ook energy.json-keys "pv"/"lp"/"wp"
# (met "levering"/"verbruik"), die normaal door je eigen PV-/LP-/WP-
# uitleesscripts worden geschreven. Zet op False zodra die scripts draaien.
SIMULATE_ENERGY_DASHBOARD = True


def timestamp() -> str:
    return datetime.now().strftime("%Y-%m-%d %H:%M:%S")


def simulate_once() -> None:
    ts = timestamp()

    if SIMULATE_P1_READER:
        levering = round(random.uniform(-0.5, 3.0), 1)  # kan ook 'export' zijn (negatief = verbruik)
        amps = [round(random.uniform(5, 20), 1) for _ in range(3)]
        set_key("p1_reader", {"levering": levering, "amps": amps, "timestamp": ts}, path=READER_DATA_JSON)

    set_key("WP_reader", {
        "WPsetpoint": round(random.uniform(18, 22), 1),
        "WPcoolpoint": round(random.uniform(22, 25), 1),
        "WPpower": round(random.uniform(0, 3.3), 1),
        "WPoutsideT": round(random.uniform(5, 30), 1),
        "timestamp": ts,
    }, path=READER_DATA_JSON)

    # KNMI levert alleen Tmin (minimumtemperatuur-voorspelling, via
    # energietarieven.py's getTemp()) - niet WPoutsideT, dat komt van
    # WP_reader zelf (live, van de warmtepomp).
    set_key("KNMI", {
        "Tmin": round(random.uniform(-2, 8), 1),
    }, path=READER_DATA_JSON)

    set_key("LP_reader", {
        "LPsetpoint": random.randint(6, 16),
        "laadpower": round(random.uniform(0, 11), 1),
    }, path=READER_DATA_JSON)

    set_key("PV_reader", {
        "production": random.randint(0, 4000),
    }, path=READER_DATA_JSON)

    if SIMULATE_ENERGY_DASHBOARD:
        # pv: alleen 'levering' wordt door de dashboard-JS gebruikt
        set_key("pv", {"levering": round(random.uniform(0, 3.0), 1)}, path=ENERGY_JSON)
        set_key("lp", {"levering": 0.0, "verbruik": round(random.uniform(0, 3.5), 1)}, path=ENERGY_JSON)
        set_key("wp", {"levering": 0.0, "verbruik": round(random.uniform(0, 1.5), 1)}, path=ENERGY_JSON)

    logger.info(f"simulate_readers: nieuwe testdata geschreven ({ts})")


def seed_setpoints_if_missing() -> None:
    """
    laadpaal wordt door LPmanager() zelf pas gezet zodra de pauze-status
    wisselt - bij een koude start bestaat de key nog niet. laadsessie hoort
    normaal bij een apart sessie-registratie-onderdeel (niet in dit project
    aanwezig) dat bijhoudt of er een auto is aangesloten en laadt. Voor
    testdoeleinden zetten we hier eenmalig startwaardes, zodat LPmanager()
    niet crasht op een ontbrekende key.
    """
    from shared_store import get_key
    if get_key("laadpaal", path=SETPOINTS_JSON) is None:
        set_key("laadpaal", {"state": "active"}, path=SETPOINTS_JSON)
        logger.info("simulate_readers: laadpaal-status geseed op 'active'")
    if get_key("laadsessie", path=READER_DATA_JSON) is None:
        set_key("laadsessie", {"laden": True}, path=READER_DATA_JSON)
        logger.info("simulate_readers: laadsessie geseed op laden=True")


def main() -> None:
    seed_setpoints_if_missing()
    while True:
        simulate_once()
        sleep(SIM_INTERVAL)


if __name__ == "__main__":
    main()