goodwe-addon/goodwe_controller/app/control.py

"""The grid-following control law.

Pure functions on purpose: this is the part that moves real power, so it must be
testable without Home Assistant, without MQTT and without an inverter. See
test_control.py, and run it before shipping any change to this file.

Sign convention, used everywhere in this add-on:

    grid   > 0  = IMPORTING from the grid
    target > 0  = inverter should DISCHARGE
    target < 0  = inverter should CHARGE

Every constant here was measured on real hardware, not chosen for elegance.
The reasoning lives in the field guide under "Why the tuning is what it is";
the short version is in the comments below. Do not "clean this up" - each rule
exists because its absence produced a specific, observed failure.
"""

from dataclasses import dataclass


@dataclass(frozen=True)
class Tuning:
    gain: float = 0.6
    max_w: float = 2000.0
    slew_w: float = 1000.0
    deadband_w: float = 15.0
    step_w: int = 10
    saturation_w: float = 500.0
    saturation_cycles: int = 3
    # What the meter should rest at, in W. Negative = a slight export.
    # ⚠️ The deadband is a one-way ratchet: any resting point inside it holds
    # forever, and the meter's IMPORT register counts every positive one with
    # no export to cancel it. Resting at 0 W therefore leaks ~deadband/2 W of
    # billed import all day (15 W deadband ≈ 0.2-0.35 kWh). Biasing the rest
    # point below zero moves that leak into the export register, which is not
    # billed. Cost is ~|bias| W of given-away export; keep it small.
    target_grid_w: float = 0.0


@dataclass(frozen=True)
class Decision:
    target_w: float
    sat_count: int
    frozen: bool
    reason: str


def compute(
    prev_w: float,
    grid_w: float,
    actual_w: float,
    tuning: Tuning,
    sat_count: int = 0,
) -> Decision:
    """One control cycle. A cycle is one meter update (~5 s on a HomeWizard P1).

    `prev_w`   what we last commanded
    `grid_w`   net grid power, + = importing
    `actual_w` what the inverter reports it is doing, + = discharging
    """
    reason = "tracking"

    # --- saturation, WITH the duration term -------------------------------
    # Command and actual diverging means the inverter cannot follow - it is at
    # a limit. Then the magnitude may fall but never rise, which is the
    # anti-windup that the vendor controller lacked: it once commanded
    # -14547 W against an inverter reporting -5250 W and kept climbing.
    #
    # ⚠️ The duration term is not optional. Tested instantaneously, this fires
    # on EVERY large correction, because the plant itself needs 3-6 s to settle
    # while a cycle is ~5 s. Requiring N consecutive saturated cycles is what
    # lets slew be larger than saturation_w.
    saturated_now = abs(prev_w - actual_w) > tuning.saturation_w
    sat_count = min(sat_count + 1, 10) if saturated_now else 0
    frozen = sat_count >= tuning.saturation_cycles

    # --- deadband ----------------------------------------------------------
    # Inside the meter's own noise, hold. Measured on the reference install
    # while regulating: mean |error| 15.4 W, max 27 W. A 10 W deadband makes
    # ~69 % of cycles act and the command never rests; 15 W leaves a
    # recognisable resting state, which is worth more than it looks - "flat for
    # 70 s" is how a healthy loop is recognised at a glance, and a command
    # frozen where it should not be is how two real bugs were caught.
    error = grid_w - tuning.target_grid_w
    if abs(error) < tuning.deadband_w:
        want = prev_w
        reason = "deadband"
    else:
        want = prev_w + tuning.gain * error

    # ⚠️ Maintenance shaping (charge-only, cheap-window floor) used to live
    # here. It now belongs to arbiter.py as limit claims, so that precedence
    # between strategies is decided in ONE place. This function is again what it
    # should be: a controller that knows only about tracking the meter.

    # --- ORDER MATTERS: clamp -> slew -> freeze ----------------------------
    # An early draft applied a floor after the clamp and let demand escape it.
    target = max(-tuning.max_w, min(tuning.max_w, want))
    if target != want:
        reason = "clamped"

    slewed = max(prev_w - tuning.slew_w, min(prev_w + tuning.slew_w, target))
    if slewed != target:
        reason = "slew-limited"
    target = slewed

    if frozen:
        # Magnitude may fall, never rise.
        # ⚠️ At prev == 0 this forbids charging while saturated, because
        # max(t, 0) wins. That is deliberate and matches the reference
        # implementation: commanding 0 while the inverter reports >500 W means
        # something else is driving the bus, and that is not the moment to
        # start pushing power the other way.
        target = min(target, prev_w) if prev_w > 0 else max(target, prev_w)
        reason = "saturated-freeze"

    # --- quantise ----------------------------------------------------------
    # 10 W. The register is 1 W and the inverter reports at 1 W, but its
    # response lands on a coarser ladder (~17.6 W measured at ~900 W, consistent
    # with a fixed DC-side current step). So sub-17 W precision is nominal;
    # 10 W simply avoids a visible staircase on dashboards.
    step = max(1, int(tuning.step_w))
    target = round(target / step) * step

    return Decision(float(target), sat_count, frozen, reason)


def maintenance_charge_floor(
    charge_w: float,
    peak_forecast_w: float | None,
    peak_cap_w: float,
) -> float:
    """How hard a maintenance charge may pull, in W (positive number).

    ⚠️ CAPACITY TARIFF. In Belgium (capaciteitstarief) and similar markets the
    bill carries the month's worst quarter-hour AVERAGE OFFTAKE. A maintenance
    charge is the only thing this system does that is big enough and long
    enough to set that peak, so it is capped by the headroom left under the
    site's cap. Discharge is never capped this way: export is not offtake.

    `peak_forecast_w is None` means the site has no capacity tariff (or no
    forecast sensor) - then there is nothing to protect and the configured
    rate is used as-is.
    """
    if peak_forecast_w is None:
        return max(0.0, charge_w)
    headroom = max(0.0, peak_cap_w - peak_forecast_w)
    return max(0.0, min(charge_w, headroom))


def peak_at_risk(peak_forecast_w: float | None, peak_cap_w: float) -> bool:
    """True when the quarter-hour projection is already over the site's cap.

    ⚠️ MONEY OUTRANKS THE MAINTENANCE SCHEDULE. While charging, the battery is
    clamped out of discharging and therefore cannot shave a peak - and one oven
    during a charge phase can cost more than the whole cycle saves. When this
    is True the charge-only clamp is dropped and normal grid-following resumes;
    the charge picks up again afterwards.
    """
    if peak_forecast_w is None:
        return False
    return peak_forecast_w > peak_cap_w