goodwe-addon/goodwe_controller/app/maintenance.py

"""The monthly battery maintenance cycle.

⚠️ THIS IS THE PART THE VENDOR CONTROLLER DID THAT NOTHING ELSE REPLACES.
A full low -> full cycle each month is a BMS-health behaviour on a 16S LiFePO4
pack: the full charge is when the BMS gets to BALANCE CELLS, the deep discharge
is what RECALIBRATES THE COULOMB COUNTER. Dropping it breaks nothing visibly -
it degrades the pack over months, and the first symptom is a state-of-charge
reading nobody can trust any more. Highest consequence, lowest visibility.

OWNERSHIP, not priority. Exactly one thing writes the setpoint at any moment:

    phase    setpoint owned by   behaviour
    idle     the control loop    normal grid-to-zero
    drain    THIS module         forced discharge, exports the surplus
    charge   the control loop    charge-only clamp + cheap-window floor
    hold     THIS module         0 W, parked full so the BMS can balance

The loop is never "disabled" - it yields. The heartbeat to the inverter keeps
running in every phase, so the hardware watchdog stays fed and a crash here
still ends with the inverter at 0 W rather than latched.
"""

import logging
from dataclasses import dataclass, field
from datetime import datetime, timedelta, timezone

_LOG = logging.getLogger("goodwe.maint")

IDLE, DRAIN, CHARGE, HOLD = "idle", "drain", "charge", "hold"

# A drain of a 13.8 kWh usable pack (300 Ah x 51.2 V, DoD 90) at the shipped
# 2000 W is ~6.9 h even with no house load helping - and less in practice, since
# house load drains alongside the export and soc_floor stops short of empty.
# 12 h means the battery is not following commands - a fault, not a slow day. The charge phase deliberately waits for sun and for cheap hours, so
# it is expected to span a night; 36 h still bounds it.
DRAIN_TIMEOUT_MIN = 720
CHARGE_TIMEOUT_MIN = 2160
# After an abort, do not immediately re-trigger in the same start hour.
RESTART_COOLDOWN_MIN = 90


@dataclass(frozen=True)
class MaintConfig:
    enabled: bool = False
    interval_days: int = 28
    start_hour: int = 10
    discharge_w: float = 2500.0
    charge_w: float = 2500.0
    soc_floor: float = 11.0
    soc_target: float = 99.0
    hold_min: int = 120


@dataclass
class MaintResult:
    phase: str
    owns_setpoint: bool
    setpoint_w: float | None = None
    charge_only: bool = False
    events: list = field(default_factory=list)


class Maintenance:
    def __init__(self, cfg: MaintConfig, store):
        self.cfg = cfg
        self.store = store

    # -- state ---------------------------------------------------------------
    @property
    def phase(self) -> str:
        return self.store.data.get("phase", IDLE)

    def _enter(self, phase: str, now: datetime, events: list, msg: str) -> None:
        self.store.set("phase", phase)
        self.store.set_time("phase_started", now)
        _LOG.info("%s", msg)
        events.append(msg)

    def _elapsed_min(self, now: datetime) -> float:
        started = self.store.get_time("phase_started")
        return 0.0 if started is None else (now - started).total_seconds() / 60.0

    def due(self, now: datetime) -> bool:
        last = self.store.get_time("last_completed")
        if last is None:
            return True
        return (now - last) >= timedelta(days=self.cfg.interval_days)

    def next_due(self, now: datetime):
        last = self.store.get_time("last_completed")
        return None if last is None else last + timedelta(days=self.cfg.interval_days)

    # -- operations ----------------------------------------------------------
    def force_start(self, now: datetime) -> list:
        """Start a cycle regardless of the calendar.

        Commissioning uses this: a cycle must be proven end to end before a site
        is signed off, and nobody waits a month to find out the schedule never
        fires.
        """
        events = []
        self.store.set_time("last_start_attempt", now)
        self._enter(DRAIN, now, events, "maintenance: forced start -> drain")
        return events

    def abort(self, now: datetime, why: str) -> list:
        events = []
        self._enter(IDLE, now, events, f"maintenance: ABORT ({why})")
        return events

    # -- the machine ---------------------------------------------------------
    def tick(self, now: datetime, soc: float | None) -> MaintResult:
        events: list = []
        phase = self.phase

        if phase == IDLE:
            if self.cfg.enabled and now.hour == self.cfg.start_hour and self.due(now):
                last_try = self.store.get_time("last_start_attempt")
                cooled = (
                    last_try is None
                    or (now - last_try) >= timedelta(minutes=RESTART_COOLDOWN_MIN)
                )
                if cooled:
                    self.store.set_time("last_start_attempt", now)
                    self._enter(DRAIN, now, events,
                                "maintenance: due -> drain (exports the surplus)")
                    return MaintResult(DRAIN, True, self.cfg.discharge_w, False, events)
            return MaintResult(IDLE, False, None, False, events)

        # ⚠️ Fail toward inaction. Without a state-of-charge reading none of the
        # transitions below mean anything, and a stale "100 %" would end the
        # charge phase having charged nothing. Hold the current phase's command
        # and wait; the phase timeout is the backstop.
        if soc is None:
            _LOG.warning("maintenance: no SoC reading, holding phase %s", phase)
            if phase == DRAIN:
                return MaintResult(DRAIN, True, self.cfg.discharge_w, False, events)
            if phase == HOLD:
                return MaintResult(HOLD, True, 0.0, False, events)
            return MaintResult(CHARGE, False, None, True, events)

        elapsed = self._elapsed_min(now)

        if phase == DRAIN:
            if soc <= self.cfg.soc_floor:
                self._enter(CHARGE, now, events,
                            f"maintenance: drain complete at {soc:.0f} % "
                            f"after {elapsed:.0f} min -> charge")
                return MaintResult(CHARGE, False, None, True, events)
            if elapsed > DRAIN_TIMEOUT_MIN:
                events += self.abort(now, f"drain stuck at {soc:.0f} % after {elapsed:.0f} min")
                return MaintResult(IDLE, True, 0.0, False, events)
            return MaintResult(DRAIN, True, self.cfg.discharge_w, False, events)

        if phase == CHARGE:
            if soc >= self.cfg.soc_target:
                self._enter(HOLD, now, events,
                            f"maintenance: charge complete at {soc:.0f} % after "
                            f"{elapsed:.0f} min -> holding {self.cfg.hold_min} min to balance")
                return MaintResult(HOLD, True, 0.0, False, events)
            if elapsed > CHARGE_TIMEOUT_MIN:
                events += self.abort(now, f"charge stuck at {soc:.0f} % after {elapsed:.0f} min")
                return MaintResult(IDLE, True, 0.0, False, events)
            return MaintResult(CHARGE, False, None, True, events)

        if phase == HOLD:
            if elapsed >= self.cfg.hold_min:
                self.store.set_time("last_completed", now)
                self._enter(IDLE, now, events,
                            f"maintenance: cycle COMPLETE ({elapsed:.0f} min hold) "
                            "- handing back to the loop")
                return MaintResult(IDLE, False, None, False, events)
            # 0 W, not "do nothing": commanding zero is what stops the loop
            # pulling the pack straight back down the moment it is full, which
            # is exactly when the BMS needs it parked at the top.
            return MaintResult(HOLD, True, 0.0, False, events)

        _LOG.error("maintenance: unknown phase %r, forcing idle", phase)
        events += self.abort(now, f"unknown phase {phase!r}")
        return MaintResult(IDLE, True, 0.0, False, events)