diff --git a/apps/predbat/gateway.py b/apps/predbat/gateway.py index 1a9dbd561..d561501d4 100644 --- a/apps/predbat/gateway.py +++ b/apps/predbat/gateway.py @@ -104,6 +104,12 @@ def _serial_suffix(serial): # rounded down - 1300W on a 13.41kWh battery reads back 1206W (#5324) GIVENERGY_RATE_STEP_PERCENT_OF_CAPACITY = 1 GIVENERGY_INVERTER_TYPES = (pb.INVERTER_TYPE_GIVENERGY, pb.INVERTER_TYPE_GIVENERGY_EMS, pb.INVERTER_TYPE_GIVENERGY_GATEWAY) +# Coordinators: a Plant EMS or a GivEnergy Gateway manages battery inverters and has no battery of its own +GATEWAY_COORDINATOR_TYPES = (pb.INVERTER_TYPE_GIVENERGY_EMS, pb.INVERTER_TYPE_GIVENERGY_GATEWAY) +# Where the discovery record of each inverter in gateway_inverter_serial is kept (see _update_discovery) +GATEWAY_STORAGE_MODULE = "gateway" +# How long a failed save of the discovery records waits before it is tried again +GATEWAY_DISCOVERY_SAVE_RETRY_SECONDS = 5 * 60 # Entity attribute table — keyed by the semantic suffix used in dashboard_item calls GATEWAY_ATTRIBUTE_TABLE = { @@ -224,7 +230,19 @@ class GatewayMQTT(ComponentBase): # Defaults for an instance built without initialize() gateway_shared_ct = True gateway_integrate_power = False + gateway_inverter_serial = () _integrated_energy = None + _unavailable_inverters = ((), ()) + _discovery = {} # replaced, never changed in place, so this default is not shared state + _discovery_dirty = False + _discovery_save_retry_at = 0 + _pv_only_serials = () + _inverter_slot_serials = () + _slots_without_capacity = frozenset() + _last_ratings = {} # replaced, never changed in place + _assumed_ratings = frozenset() + _reserve_before_zero = {} # replaced, never changed in place + gateway_device_id = None def initialize(self, gateway_device_id=None, mqtt_host=None, mqtt_port=8883, mqtt_token=None, gateway_inverter_serial=None, gateway_evc_automatic=False, gateway_evc_control=False, gateway_shared_ct=True, gateway_integrate_power=False, **kwargs): """Initialize gateway configuration and build MQTT topic strings. @@ -328,6 +346,15 @@ def initialize(self, gateway_device_id=None, mqtt_host=None, mqtt_port=8883, mqt self._auto_configured = False self._configured_inverter_serials = frozenset() # serials discovered at the last auto-config self._inverter_slot_serials = [] # serial bound to each PredBat inverter slot at the last auto-config + self._pv_only_serials = () # listed inverters with no battery: counted for PV, never written to + self._slots_without_capacity = frozenset() # slots bound with no battery size sensor, re-bound once the hub reports a size + self._last_ratings = {} # (entity suffix, rating name) -> the last real value the hub reported + self._assumed_ratings = frozenset() # entity suffixes of inverters whose maximum rate has never been reported + self._reserve_before_zero = {} # entity suffix -> the last real reserve, set aside when Predbat wrote a reserve of 0 + self._discovery = {} # upper-case serial -> what the hub last reported it to be (see _update_discovery) + self._discovery_dirty = False + self._discovery_save_retry_at = 0 + self._unavailable_inverters = ((), ()) self._configured_ev_chargers = frozenset() # EV charge point ids registered at the last auto-config self._last_published_plan = None self._pending_plan = None @@ -593,6 +620,7 @@ async def run(self, seconds, first): # Capture the loop that will own the MQTT client/listener task, so # cross-loop callers (ha.py::run_async()) can dispatch onto it later. self._loop = asyncio.get_running_loop() + await self._load_discovery() # Start MQTT listener as a background task self._mqtt_task = asyncio.ensure_future(self._mqtt_loop()) self.log("Info: GatewayMQTT: MQTT listener task started") @@ -767,6 +795,7 @@ async def _handle_message(self, message): try: if topic == self.topic_status: self._process_telemetry(message.payload) + await self._save_discovery() elif topic.startswith(getattr(self, "_ack_topic_prefix", "\0")): self._process_ack(message.payload) elif topic == self.topic_online: @@ -831,17 +860,34 @@ def _process_telemetry(self, data): self._debug_dump("RX telemetry", status, raw=data) if len(status.inverters) == 0: - return - + if not (self.gateway_inverter_serial and self._discovery): + return + # With a serial list the discovered inverters are the site whatever the hub reports, + # so a status naming none of them is every one of them unread, not nothing to do + # (#5471). It is not fresh data though, so it does not count towards being alive. + else: + self._last_telemetry_time = time.time() + self.update_success_timestamp() self._last_status = status - self._last_telemetry_time = time.time() - self.update_success_timestamp() + self._update_discovery(status) self._update_integrated_energy(status) self._inject_entities(status) if self._needs_reconfigure(status): self.automatic_config() + if self._auto_configured: + # The injection above ran before this status's inverters were bound, so one that + # is only published once bound has no entities yet. Publish those before the API + # is declared started below. + for inv in status.inverters: + if not inv.primary and self._is_published(inv): + self._publish_inverter(inv) + # The integration above ran on the slots as they were before this status changed them + self._update_integrated_energy(status, rebase=True) + self._inject_unreported_inverters(status) + self._inject_shared_site_metering(status) + self._check_configured_inverters(status) # Declare the API started only once auto-config has wired up the inverter args. # ComponentManager.start() polls api_started from the main thread while this @@ -916,15 +962,11 @@ def _inject_entities(self, status): ) for inv in status.inverters: - # Inject primary (battery-bearing) units, plus whichever unit automatic_config - # bound PredBat's args to. On a multi-AIO site the control target is the - # Gateway/EMS, which firmware never flags primary — skipping it left every - # bound arg unwritten and frozen at its last value. Non-primary units that are - # NOT the control target stay skipped: they report overlapping power readings - # that would double up on the dashboard. - if not inv.primary and not self._is_bound_target(inv): - continue - self._inject_inverter_entities(inv, _serial_suffix(inv.serial), site_energy=self._integrated_site_energy(inv)) + if self._is_published(inv): + self._publish_inverter(inv) + + self._inject_unreported_inverters(status) + self._inject_shared_site_metering(status) # EV charger entities (device-level, present only when a charge point is connected) self._inject_ev_entities(status) @@ -951,7 +993,95 @@ def _inject_entities(self, status): self.dashboard_item(f"{sp}_grid_power", sub.grid_w, attributes=GATEWAY_ATTRIBUTE_TABLE.get("grid_power", {}), app="gateway") self.dashboard_item(f"{sp}_temp", sub.temp_c, attributes=GATEWAY_ATTRIBUTE_TABLE.get("temp", {}), app="gateway") - def _update_integrated_energy(self, status): + def _is_published(self, inv): + """Whether this unit's entities are published. + + Primary (battery-bearing) units are, plus whichever unit automatic_config bound + PredBat's args to and any PV-only inverter it configured. On a multi-AIO site the + control target is the Gateway/EMS, which firmware never flags primary — skipping it + left every bound arg unwritten and frozen at its last value. Other non-primary units + stay skipped: they report overlapping power readings that would double up on the + dashboard. + + Args: + inv: A ``predbat_InverterEntry`` from the gateway status. + + Returns: + bool: True if the unit's entities are published. + """ + return inv.primary or self._is_bound_target(inv) or inv.serial in self._pv_only_serials + + def _publish_inverter(self, inv): + """Publish one unit's entities from its status entry. + + An inverter the hub is not reading keeps the values it last reported, like one behind + a cloud API that has gone offline: publishing the empty message would read as a flat + battery. Only its power readings go to zero (see ``_inject_unread_inverter``). One with + no entities yet still gets its defaults (#5471). + + Args: + inv: A ``predbat_InverterEntry`` from the gateway status. + """ + suffix = _serial_suffix(inv.serial) + if not self._has_telemetry(inv): + if self.get_state_wrapper(f"sensor.{self.prefix}_gateway_{suffix}_soc") is not None: + self._inject_unread_inverter(suffix) + return + record = self._discovery.get(inv.serial.upper()) + if record is not None: + # Its battery size is known from when it was last read + inv = self._stored_unit(record) + self._inject_inverter_entities(inv, suffix, site_energy=self._integrated_site_energy(inv)) + + def _inject_unread_inverter(self, suffix): + """Zero the power readings of an inverter the hub is not reading. + + Its state of charge, battery size and settings are left as last reported, but a + power is a rate: left alone, an inverter that dropped out while discharging at 3kW + would be read as discharging at 3kW for as long as it is away. + + Args: + suffix: The inverter's entity suffix. + """ + pfx = f"{self.prefix}_gateway_{suffix}" + for name in ("battery_power", "pv_power", "grid_power", "load_power", "inverter_power"): + self.dashboard_item(f"sensor.{pfx}_{name}", 0, attributes=GATEWAY_ATTRIBUTE_TABLE.get(name, {}), app="gateway") + + def _inject_shared_site_metering(self, status): + """Keep the site's grid and load readings going when the inverter that carries them is not being read. + + On one shared CT clamp (gateway_shared_ct) Predbat reads grid and load power, and + today's import, export and load, from the first inverter's entities only. Every + inverter on the clamp reports the same readings, so while the first is not being read + its entities are filled from the next one that is. With gateway_integrate_power the + day's figures are the integrated ones, which are published as they stand. + + Args: + status: A decoded GatewayStatus protobuf message. + """ + slots = self._inverter_slot_serials + if not self.gateway_shared_ct or len(slots) < 2: + return + by_serial = {inv.serial: inv for inv in status.inverters} + first = by_serial.get(slots[0]) + if first is not None and self._has_telemetry(first): + return + pfx = f"{self.prefix}_gateway_{_serial_suffix(slots[0])}" + donor = next((by_serial[serial] for serial in slots[1:] if serial in by_serial and self._has_telemetry(by_serial[serial])), None) + if donor is not None: + self.dashboard_item(f"sensor.{pfx}_grid_power", donor.grid.power_w, attributes=GATEWAY_ATTRIBUTE_TABLE.get("grid_power", {}), app="gateway") + self.dashboard_item(f"sensor.{pfx}_load_power", donor.load.power_w, attributes=GATEWAY_ATTRIBUTE_TABLE.get("load_power", {}), app="gateway") + state = self._integrated_energy + if state is not None: + today = {"import": state["import_wh"], "export": state["export_wh"], "load": state["load_published_wh"]} + elif donor is not None and donor.energy.ByteSize() > 0: + today = {"import": donor.energy.grid_import_today_wh, "export": donor.energy.grid_export_today_wh, "load": donor.energy.consumption_today_wh} + else: + return + for name, wh in today.items(): + self.dashboard_item(f"sensor.{pfx}_{name}_today", round(wh / 1000.0, 2), attributes=GATEWAY_ATTRIBUTE_TABLE.get(f"{name}_today", {}), app="gateway") + + def _update_integrated_energy(self, status, rebase=False): """Integrate grid and battery power into today's import, export and load (gateway_integrate_power). Some inverters do not keep usable grid energy counters. GivEnergy AC inverters sharing one @@ -971,6 +1101,12 @@ def _update_integrated_energy(self, status): Each interval uses the mean of the readings at its two ends. The figures return to zero at local midnight. They are kept in memory only, so they also restart from zero when Predbat restarts, which Predbat reads the same way as the midnight reset. + + Args: + status: A decoded GatewayStatus protobuf message. + rebase: True when auto-config has just run on a status that was already integrated. + Nothing is added: its readings become the baseline for the inverters now in the + slots, so one that has just joined is counted from this status and not the next. """ if not self.gateway_integrate_power: return @@ -980,16 +1116,23 @@ def _update_integrated_energy(self, status): return by_serial = {inv.serial: inv for inv in status.inverters} inverters = [by_serial[serial] for serial in serials if serial in by_serial] + now = status.timestamp if status.timestamp > 0 else int(time.time()) + local_now = datetime.datetime.fromtimestamp(now, tz=self.local_tz) + day = local_now.date() + zeroed = {"import_wh": 0.0, "export_wh": 0.0, "charge_wh": 0.0, "discharge_wh": 0.0, "load_wh": 0.0, "load_published_wh": 0.0} if len(inverters) != len(serials) or not all(inv.connected for inv in inverters): # A missing or disconnected unit reports no power or a stale one. Drop the baseline too, # so the next good status starts a new interval rather than filling in the time away - if self._integrated_energy is not None: - self._integrated_energy["time"] = None + state = self._integrated_energy + if state is not None: + state["time"] = None + if day != state["day"]: + # Midnight still ends the day: an inverter kept configured while the hub cannot + # read it can be away for days, and yesterday's figures are not today's + state.update(zeroed) + state["day"] = day return - now = status.timestamp if status.timestamp > 0 else int(time.time()) - local_now = datetime.datetime.fromtimestamp(now, tz=self.local_tz) - day = local_now.date() grid_w = inverters[0].grid.power_w if self.gateway_shared_ct else sum(inv.grid.power_w for inv in inverters) power = { "import": max(-grid_w, 0), @@ -998,15 +1141,19 @@ def _update_integrated_energy(self, status): "charge": sum(max(inv.battery.power_w, 0) for inv in inverters), "discharge": sum(max(-inv.battery.power_w, 0) for inv in inverters), } - pv_wh = [inv.energy.pv_today_wh for inv in inverters] - zeroed = {"import_wh": 0.0, "export_wh": 0.0, "charge_wh": 0.0, "discharge_wh": 0.0, "load_wh": 0.0, "load_published_wh": 0.0} + # By serial and not by slot: auto-config can re-run and change the slots, and a counter + # must go on being compared with its own last value + pv_wh = {inv.serial: inv.energy.pv_today_wh for inv in inverters} + # A PV-only inverter's generation is part of the load too. It has no slot, so it is not + # needed for the site to be complete: one that is away simply adds nothing meanwhile. + pv_wh.update({serial: by_serial[serial].energy.pv_today_wh for serial in self._pv_only_serials if serial in by_serial and by_serial[serial].energy.ByteSize() > 0}) state = self._integrated_energy if state is None: state = {"day": day, "time": None, **zeroed} self._integrated_energy = state previous = state["time"] - if previous is not None and now <= previous: + if previous is not None and now <= previous and not rebase: # The same status again (a re-inject), or the hub's clock stepping back return @@ -1020,13 +1167,22 @@ def _update_integrated_energy(self, status): today_share = min(max((now - midnight) / (now - previous), 0.0), 1.0) if previous is None: - state["pv"] = [{"base": wh, "high": None, "high_time": 0} for wh in pv_wh] + state["pv"] = {serial: {"base": wh, "high": None, "high_time": 0} for serial, wh in pv_wh.items()} + elif rebase: + for serial, wh in pv_wh.items(): + state["pv"].setdefault(serial, {"base": wh, "high": None, "high_time": 0}) else: elapsed = now - previous hours = elapsed / 3600.0 # Per inverter, so one counter dropping a sample does not hide the others' generation pv_rise_limit_wh = GATEWAY_INTEGRATE_MAX_PV_W * hours + GATEWAY_INTEGRATE_PV_STEP_WH - pv_rise_wh = sum(self._pv_counter_rise(track, wh, now, pv_rise_limit_wh) for track, wh in zip(state["pv"], pv_wh)) + pv_rise_wh = 0.0 + for serial, wh in pv_wh.items(): + if serial in state["pv"]: + pv_rise_wh += self._pv_counter_rise(state["pv"][serial], wh, now, pv_rise_limit_wh) + else: + # First seen since the integration began: counted from here on + state["pv"][serial] = {"base": wh, "high": None, "high_time": 0} if elapsed <= GATEWAY_INTEGRATE_MAX_GAP_SECONDS: energy = {name: (state["power"][name] + power[name]) / 2.0 * hours * today_share for name in power} for name in energy: @@ -1074,6 +1230,24 @@ def _integrated_site_energy(self, inv): return None return {"import_wh": state["import_wh"], "export_wh": state["export_wh"], "load_wh": state["load_published_wh"]} + def _last_rating(self, suffix, name, value): + """A rating as the hub reports it, or the last real value it reported when it leaves it out. + + Args: + suffix: The inverter's entity suffix. + name: Which rating. + value: The value in this status; zero is the hub's "not reported". + + Returns: + The value, the last non-zero one for this inverter, or None if there has never been one. + """ + key = (suffix, name) + if value: + if self._last_ratings.get(key) != value: + self._last_ratings = {**self._last_ratings, key: value} + return value + return self._last_ratings.get(key) + def _inject_inverter_entities(self, inv, suffix, site_energy=None): """Inject entities for a single inverter using HA-style naming. @@ -1091,11 +1265,23 @@ def _inject_inverter_entities(self, inv, suffix, site_energy=None): self.dashboard_item(f"sensor.{pfx}_battery_voltage", bat.voltage_v, attributes=GATEWAY_ATTRIBUTE_TABLE.get("battery_voltage", {}), app="gateway") self.dashboard_item(f"sensor.{pfx}_battery_current", bat.current_a, attributes=GATEWAY_ATTRIBUTE_TABLE.get("battery_current", {}), app="gateway") self.dashboard_item(f"sensor.{pfx}_battery_temperature", bat.temperature_c, attributes=GATEWAY_ATTRIBUTE_TABLE.get("battery_temperature", {}), app="gateway") - if bat.capacity_wh: - self.dashboard_item(f"sensor.{pfx}_battery_capacity", round(bat.capacity_wh / 1000.0, 2), attributes=GATEWAY_ATTRIBUTE_TABLE.get("battery_capacity", {}), app="gateway") + # A size the hub leaves out keeps the last one known, as the ratings below do + capacity_wh = bat.capacity_wh or self._discovery.get(inv.serial.upper(), {}).get("capacity_wh") + if capacity_wh: + self.dashboard_item(f"sensor.{pfx}_battery_capacity", round(capacity_wh / 1000.0, 2), attributes=GATEWAY_ATTRIBUTE_TABLE.get("battery_capacity", {}), app="gateway") if bat.soh_percent: self.dashboard_item(f"sensor.{pfx}_battery_soh", bat.soh_percent, attributes=GATEWAY_ATTRIBUTE_TABLE.get("battery_soh", {}), app="gateway") - battery_rate_max = bat.rate_max_w or 6000 + # A rating the hub leaves out keeps the last real value it reported. A status with the + # ratings unpopulated (seen for a few minutes after a hub restart) otherwise turned a 3kW + # inverter into a 6kW one with no export limit, and those rates were written to it (#5478). + record = self._discovery.get(inv.serial.upper(), {}) + battery_rate_max = self._last_rating(suffix, "battery_rate_max", bat.rate_max_w) or record.get("rate_max_w") + # With no rating ever reported the plan still needs a figure, but it is only an assumption: + # rate writes for the inverter are held until the real one arrives (see _rate_write_held) + assumed = not battery_rate_max and inv.type not in GATEWAY_COORDINATOR_TYPES + if assumed != (suffix in self._assumed_ratings): + self._assumed_ratings = self._assumed_ratings | {suffix} if assumed else self._assumed_ratings - {suffix} + battery_rate_max = battery_rate_max or 6000 self.dashboard_item(f"sensor.{pfx}_battery_rate_max", battery_rate_max, attributes=GATEWAY_ATTRIBUTE_TABLE.get("battery_rate_max", {}), app="gateway") self.dashboard_item(f"sensor.{pfx}_pv_power", inv.pv.power_w, attributes=GATEWAY_ATTRIBUTE_TABLE.get("pv_power", {}), app="gateway") @@ -1112,15 +1298,16 @@ def _inject_inverter_entities(self, inv, suffix, site_energy=None): self.dashboard_item(f"sensor.{pfx}_inverter_power", inv.inverter.active_power_w, attributes=GATEWAY_ATTRIBUTE_TABLE.get("inverter_power", {}), app="gateway") if inv.inverter.temperature_c: self.dashboard_item(f"sensor.{pfx}_inverter_temperature", inv.inverter.temperature_c, attributes=GATEWAY_ATTRIBUTE_TABLE.get("inverter_temperature", {}), app="gateway") - inverter_rate_max = inv.inverter.rate_max_w or battery_rate_max + inverter_rate_max = self._last_rating(suffix, "inverter_rate_max", inv.inverter.rate_max_w) or record.get("inverter_rate_max_w") or battery_rate_max if inverter_rate_max: self.dashboard_item(f"sensor.{pfx}_inverter_rate_max", inverter_rate_max, attributes=GATEWAY_ATTRIBUTE_TABLE.get("inverter_rate_max", {}), app="gateway") control = inv.control self.dashboard_item(f"switch.{pfx}_charge_enabled", "on" if control.charge_enabled else "off", attributes=GATEWAY_ATTRIBUTE_TABLE.get("charge_enabled", {}), app="gateway") self.dashboard_item(f"switch.{pfx}_discharge_enabled", "on" if control.discharge_enabled else "off", attributes=GATEWAY_ATTRIBUTE_TABLE.get("discharge_enabled", {}), app="gateway") - # Proto sentinel: 0=undefined→publish 99999 (unlimited), 1=zero limit→publish 0, otherwise use value as-is - _raw_export_limit = control.export_limit_w + # Proto sentinel: 0=undefined→publish 99999 (unlimited), 1=zero limit→publish 0, otherwise use value as-is. + # Undefined only means unlimited for an inverter that has never reported a limit. + _raw_export_limit = self._last_rating(suffix, "export_limit", control.export_limit_w) or record.get("export_limit_w") or 0 export_limit_publish = 99999 if _raw_export_limit == 0 else (0 if _raw_export_limit == 1 else _raw_export_limit) self.dashboard_item(f"sensor.{pfx}_export_limit_w", export_limit_publish, attributes=GATEWAY_ATTRIBUTE_TABLE.get("export_limit_w", {}), app="gateway") # The hub reports the rate the inverter holds, so a GivEnergy one reads back short of the rate written @@ -1141,7 +1328,10 @@ def _inject_inverter_entities(self, inv, suffix, site_energy=None): reserve_attributes = dict(GATEWAY_ATTRIBUTE_TABLE.get("reserve_soc", {})) reserve_soc_max = getattr(control, "reserve_soc_max", 0) reserve_attributes["max"] = reserve_soc_max if 1 <= reserve_soc_max <= 100 else 100 - self.dashboard_item(f"number.{pfx}_reserve_soc", control.reserve_soc, attributes=reserve_attributes, app="gateway") + # A reserve the hub leaves out keeps the last real one, like the ratings above: read as 0% it + # is rewritten by Predbat for nothing (#5478). A 0% that Predbat wrote is real (see number_event). + reserve_soc = self._last_rating(suffix, "reserve", control.reserve_soc) or 0 + self.dashboard_item(f"number.{pfx}_reserve_soc", reserve_soc, attributes=reserve_attributes, app="gateway") self.dashboard_item(f"number.{pfx}_target_soc", control.target_soc, attributes=GATEWAY_ATTRIBUTE_TABLE.get("target_soc", {}), app="gateway") # Schedule times (convert HHMM uint32 → HH:MM:SS string) # Always set with defaults so PredBat doesn't crash on missing charge_start_time @@ -1320,9 +1510,11 @@ def _inject_ev_entities(self, status): def _needs_reconfigure(self, status): """Whether automatic_config should (re-)run for this status. - Runs on first telemetry, and again when a *new* inverter serial appears — e.g. - a second AIO is discovered later, which (per GivTCP) moves the control point - from the AIO to the Gateway. Removals are deliberately ignored so a transient + Runs on first telemetry, and again when an inverter auto-config could act on + appears for the first time (see ``_usable_serials``) — e.g. a second AIO is + discovered later, which (per GivTCP) moves the control point from the AIO to + the Gateway, or an inverter the hub could not read at start-up starts reporting + battery data (#5471). Removals are deliberately ignored so a transient drop-out does not thrash the config; a permanent removal is handled by the onboarding/restart path. (NOTE: re-running re-selects the control target and rewrites the inverter args; whether PredBat core re-reads ``num_inverters`` at @@ -1330,7 +1522,11 @@ def _needs_reconfigure(self, status): """ if not self._auto_configured: return True - new_serials = frozenset(inv.serial for inv in status.inverters) - self._configured_inverter_serials + sized = sorted(inv.serial for inv in status.inverters if inv.serial in self._slots_without_capacity and inv.battery.capacity_wh > 0) + if sized: + self.log(f"Info: GatewayMQTT: inverter(s) {sized} now report a battery size — re-running auto-config") + return True + new_serials = self._usable_serials(status) - self._configured_inverter_serials if new_serials: self.log(f"Info: GatewayMQTT: new inverter(s) discovered {sorted(new_serials)} — re-running auto-config") return True @@ -1342,6 +1538,268 @@ def _needs_reconfigure(self, status): return True return False + def _usable_serials(self, status): + """What auto-config could act on in this status, as a set of keys. + + A coordinator (Plant EMS / Gateway) and a battery inverter that is reporting battery + data, by serial. An inverter the hub lists but is not yet reading is not usable, so it + is still new - and re-runs auto-config - on the first status in which it is (#5471). + + With ``gateway_inverter_serial`` set, the listed inverters count by their discovery + record instead (see ``_update_discovery``): the serial for a battery inverter and + ``:pv`` for one with no battery, so one that turns out to have a battery + after all is new again. + + Args: + status: A decoded GatewayStatus protobuf message. + + Returns: + frozenset: The keys. + """ + any_primary = any(inv.primary for inv in status.inverters if inv.type not in GATEWAY_COORDINATOR_TYPES) + usable = set() + for inv in status.inverters: + if inv.type in GATEWAY_COORDINATOR_TYPES: + usable.add(inv.serial) + elif inv.battery.ByteSize() > 0 and (inv.primary or not any_primary): + usable.add(inv.serial) + if self.gateway_inverter_serial: + for record in self._discovery.values(): + usable.add(record["serial"] if record.get("battery") else record["serial"] + ":pv") + return frozenset(usable) + + def _discovery_filename(self): + """Storage name of this hub's discovery records.""" + return f"discovery_{self.gateway_device_id}" + + def _listed_serials(self): + """The serials in ``gateway_inverter_serial``, upper-cased.""" + return set(serial.upper() for serial in self.gateway_inverter_serial) + + def _update_discovery(self, status): + """Record what the hub reports each inverter to be. + + The record - type, whether it has a battery, its size, ratings and model - is what lets a + listed inverter keep its place when the hub cannot read it, including across a + restart (#5471). It is taken only from an inverter the hub is reading, so an offline + one never overwrites what is known about it. Having a battery only ever turns on: an + early status can carry an inverter's other readings before its battery ones. + + With no serial list every inverter is recorded, for its last known size, ratings and + export limit only: there auto-config uses the inverters it finds, and one that is in + the records but not in the status is simply not there. + + Args: + status: A decoded GatewayStatus protobuf message. + """ + listed = self._listed_serials() + for inv in status.inverters: + key = inv.serial.upper() + if not key or (listed and key not in listed) or not self._has_telemetry(inv): + continue + known = self._discovery.get(key, {}) + record = { + "serial": inv.serial, + "type": int(inv.type), + "battery": bool(known.get("battery")) or inv.battery.ByteSize() > 0, + "capacity_wh": int(inv.battery.capacity_wh) or int(known.get("capacity_wh", 0)), + "rate_max_w": int(inv.battery.rate_max_w) or int(known.get("rate_max_w", 0)), + "inverter_rate_max_w": int(inv.inverter.rate_max_w) or int(known.get("inverter_rate_max_w", 0)), + # The hub's own value: 1 means a limit of zero, 0 means it has never reported one + "export_limit_w": int(inv.control.export_limit_w) or int(known.get("export_limit_w", 0)), + "model": inv.model or known.get("model", ""), + } + if record != known: + self._discovery = {**self._discovery, key: record} + self._discovery_dirty = True + + async def _load_discovery(self): + """Load the stored discovery records. + + With ``gateway_inverter_serial`` set, a record for a serial that is not listed is + dropped: the list is the setting, and it wins over what was stored for an earlier one. + """ + listed = self._listed_serials() + storage = self.storage + if storage is None: + return + try: + saved = await storage.load(GATEWAY_STORAGE_MODULE, self._discovery_filename()) + except Exception as e: + self.log(f"Warn: GatewayMQTT: failed to load stored inverter discovery: {e}") + return + records = saved.get("inverters") if isinstance(saved, dict) else None + if not isinstance(records, dict): + return + loaded = {} + for key, stored in records.items(): + record = self._valid_record(stored) + if record is None: + self.log(f"Warn: GatewayMQTT: ignoring unreadable stored discovery for inverter {key}") + elif not listed or record["serial"].upper() in listed: + loaded[record["serial"].upper()] = record + # Anything already learned from live telemetry is newer than the stored copy + self._discovery = {**loaded, **self._discovery} + if loaded: + self.log(f"Info: GatewayMQTT: loaded stored discovery for inverter(s) {sorted(loaded)}") + + @staticmethod + def _valid_record(stored): + """A stored discovery record with every field checked, or None if it cannot be used. + + The file can be edited or cut short, and a record auto-config cannot read would stop + it on every status message. + + Args: + stored: One record as loaded from storage. + + Returns: + dict: The record (see ``_update_discovery``), or None. + """ + try: + serial = str(stored["serial"]).strip() + record = { + "serial": serial, + "type": int(stored.get("type", pb.INVERTER_TYPE_GIVENERGY)), + "battery": stored.get("battery") is True, + "capacity_wh": max(int(stored.get("capacity_wh") or 0), 0), + "rate_max_w": max(int(stored.get("rate_max_w") or 0), 0), + "inverter_rate_max_w": max(int(stored.get("inverter_rate_max_w") or 0), 0), + "export_limit_w": max(int(stored.get("export_limit_w") or 0), 0), + "model": str(stored.get("model") or ""), + } + except (AttributeError, KeyError, TypeError, ValueError): + return None + return record if serial else None + + async def _save_discovery(self): + """Save the discovery records when they have changed. + + A save that fails stays pending and is tried again after + GATEWAY_DISCOVERY_SAVE_RETRY_SECONDS: without the stored record a configured + inverter the hub cannot read would be left out after the next restart. + """ + storage = self.storage + if not self._discovery_dirty or storage is None or _monotonic() < self._discovery_save_retry_at: + return + # Cleared before the save so a record that changes while it is being written is saved again + self._discovery_dirty = False + try: + saved = await storage.save(GATEWAY_STORAGE_MODULE, self._discovery_filename(), {"inverters": self._discovery}, format="json") + except Exception as e: + self.log(f"Warn: GatewayMQTT: failed to save inverter discovery: {e}") + saved = False + else: + if not saved: + self.log("Warn: GatewayMQTT: failed to save inverter discovery, will retry") + if saved: + self._discovery_save_retry_at = 0 + else: + self._discovery_dirty = True + self._discovery_save_retry_at = _monotonic() + GATEWAY_DISCOVERY_SAVE_RETRY_SECONDS + + @staticmethod + def _stored_unit(record): + """An inverter entry built from a discovery record, standing in for one the hub is not reporting. + + Args: + record: A discovery record (see ``_update_discovery``). + + Returns: + predbat_InverterEntry: The entry, with no readings other than the battery size. + """ + inv = pb.InverterEntry() + inv.serial = record["serial"] + inv.type = int(record.get("type", pb.INVERTER_TYPE_GIVENERGY)) + inv.model = str(record.get("model", "")) + if record.get("capacity_wh"): + inv.battery.capacity_wh = int(record["capacity_wh"]) + return inv + + def _inject_unreported_inverters(self, status): + """Give a configured inverter the hub is not reporting its entities, if it has none. + + After a restart a configured inverter, with a slot or PV only, can be known only from + its discovery record. The core reads its entities all the same, so they are published + once with defaults and the stored battery size. Ones that already exist are left as + they are, stale, apart from the power readings (see ``_inject_unread_inverter``). + + Args: + status: A decoded GatewayStatus protobuf message. + """ + reported = set(inv.serial.upper() for inv in status.inverters) + for serial in list(self._inverter_slot_serials) + list(self._pv_only_serials): + if serial.upper() in reported: + continue + record = self._discovery.get(serial.upper()) + suffix = _serial_suffix(serial) + if self.get_state_wrapper(f"sensor.{self.prefix}_gateway_{suffix}_soc") is not None: + self._inject_unread_inverter(suffix) + elif record is not None: + self._inject_inverter_entities(self._stored_unit(record), suffix) + + @staticmethod + def _has_telemetry(inv): + """Whether the hub is reporting any readings for this unit. + + A coordinator is always taken to be reporting: it has no battery of its own, and + what it sends differs by type. + + Args: + inv: A ``predbat_InverterEntry`` from the gateway status. + + Returns: + bool: False for a battery or PV inverter whose battery, PV, grid, load and inverter readings are all absent. + """ + if inv.type in GATEWAY_COORDINATOR_TYPES: + return True + return any(block.ByteSize() > 0 for block in (inv.battery, inv.pv, inv.grid, inv.load, inv.inverter)) + + def _check_configured_inverters(self, status): + """Report the configured inverters that the hub is not reading, when ``gateway_inverter_serial`` is set. + + The serial list is the user's (or the cloud's) statement of what the site has. A + listed inverter the hub has not discovered yet is left out until it is, with a + warning. Once it has a slot it keeps it: if the hub then stops reporting it, its + data goes stale and its writes fail, and that is an error for as long as it lasts + (#5471). Discovered means the hub has reported readings for it at least once, now or + in a stored record. Both are logged when the set changes, and every status with a configured + inverter unread marks the run as having had errors. + + Args: + status: A decoded GatewayStatus protobuf message. + """ + if not self.gateway_inverter_serial or not self._auto_configured: + return + + by_serial = {inv.serial.upper(): inv for inv in status.inverters} + bound = set(serial.upper() for serial in self._inverter_slot_serials) + undiscovered = sorted(serial for serial in self._listed_serials() if serial not in self._discovery and serial not in bound) + # Every battery inverter the hub has discovered counts, not only the ones with a slot: + # behind a Gateway or Plant EMS the coordinator holds the only slot, and one of its + # inverters going unread would otherwise pass unnoticed + watched = list(self._inverter_slot_serials) + watched += [record["serial"] for key, record in sorted(self._discovery.items()) if key not in bound and record.get("battery") and record.get("type") not in GATEWAY_COORDINATOR_TYPES] + unread = [] + for serial in watched: + inv = by_serial.get(serial.upper()) + # The connected flag is only relied on for an inverter with a slot, as it was before + disconnected = inv is not None and serial.upper() in bound and inv.type not in GATEWAY_COORDINATOR_TYPES and not inv.connected + if inv is None or not self._has_telemetry(inv) or disconnected: + unread.append(serial) + + unavailable = (tuple(undiscovered), tuple(unread)) + if unavailable != self._unavailable_inverters: + if undiscovered: + self.log(f"Warn: GatewayMQTT: inverter(s) {undiscovered} in gateway_inverter_serial have not been read by the hub yet; they are configured as soon as it reports their readings") + if unread: + self.log(f"Error: GatewayMQTT: configured inverter(s) {unread} are not being read by the hub; they stay configured with their last reported data, check the hub's connection to them") + if self._unavailable_inverters[1] and not unread: + self.log("Info: GatewayMQTT: every configured inverter is being read by the hub again") + self._unavailable_inverters = unavailable + if unread: + self.non_fatal_error_occurred() + def automatic_config(self): """Register gateway entities with PredBat's inverter model. @@ -1356,6 +1814,15 @@ def automatic_config(self): status = self._last_status all_inverters = list(status.inverters) + # With gateway_inverter_serial set, a listed inverter the hub has reported before is + # part of the site even when this status leaves it out (#5471). The records are brought + # up to date here as well as in _process_telemetry(), for a caller that comes straight here. + self._update_discovery(status) + serial_set = self._listed_serials() + reported = set(inv.serial.upper() for inv in all_inverters) + if serial_set: + all_inverters += [self._stored_unit(record) for key, record in sorted(self._discovery.items()) if key not in reported] + if not all_inverters: self.log("Error: GatewayMQTT: no inverters in gateway status") return @@ -1375,8 +1842,7 @@ def automatic_config(self): # charge window — and broke control after an NVS wipe re-ordered discovery.) ems_units = [inv for inv in all_inverters if inv.type == pb.INVERTER_TYPE_GIVENERGY_EMS] gateway_units = [inv for inv in all_inverters if inv.type == pb.INVERTER_TYPE_GIVENERGY_GATEWAY] - coordinator_types = (pb.INVERTER_TYPE_GIVENERGY_EMS, pb.INVERTER_TYPE_GIVENERGY_GATEWAY) - candidate_aios = [inv for inv in all_inverters if inv.type not in coordinator_types] + candidate_aios = [inv for inv in all_inverters if inv.type not in GATEWAY_COORDINATOR_TYPES] # Filter AIOs to primary units with battery data for planning. any_primary = any(inv.primary for inv in candidate_aios) @@ -1389,16 +1855,30 @@ def automatic_config(self): # Old firmware: no primary flags, use all with battery data aios = [inv for inv in candidate_aios if inv.battery.ByteSize() > 0] + pv_only = [] + gateway_coordinates = len(aios) > 1 + if serial_set: + # gateway_inverter_serial wins: every listed inverter the hub has discovered is + # used, by its discovery record and not by what the hub reports for it right now. + # Judging a listed inverter on its start-up telemetry dropped one the hub could + # not read until the next restart (#5471). One with no battery is a PV source only. + aios = [inv for inv in candidate_aios if self._discovery.get(inv.serial.upper(), {}).get("battery")] + pv_only = [inv for inv in candidate_aios if inv.serial.upper() in self._discovery and not self._discovery[inv.serial.upper()].get("battery")] + gateway_coordinates = gateway_coordinates or len(aios) > 1 + if ems_units: # A Plant EMS is the single control point for the whole system. inverters = ems_units[:1] - elif gateway_units and len(aios) > 1: + elif gateway_units and gateway_coordinates: # Multiple AIOs behind a Gateway: the Gateway is the single control point. # NOTE: control commands are addressed to the Gateway/EMS serial — the firmware # must fan these out to the AIOs (tracked separately in command_handler.cpp). inverters = gateway_units[:1] elif aios: inverters = aios + elif serial_set and not any(inv.serial.upper() in serial_set for inv in all_inverters): + # A serial list naming none of the inverters, reported below + inverters = [] else: # A retained startup frame can contain the discovered inverter list before # battery telemetry has been populated. Treating every discovered unit as @@ -1414,13 +1894,12 @@ def automatic_config(self): # Apply serial filter if configured. A no-match is an error — configuring the # wrong inverter set is worse than not configuring at all. Leave _auto_configured # False so the run loop stays blocked and we retry on the next telemetry. - if self.gateway_inverter_serial: - serial_set = set(s.upper() for s in self.gateway_inverter_serial) + if serial_set: filtered = [inv for inv in inverters if inv.serial.upper() in serial_set] if filtered: inverters = filtered else: - available = [inv.serial for inv in inverters] + available = [inv.serial for inv in (inverters or all_inverters)] self.log(f"Error: GatewayMQTT: gateway_inverter_serial filter {self.gateway_inverter_serial} matched no inverters" f" (available serials: {available}); auto-config aborted — will retry on next telemetry") self._auto_configured = False return @@ -1430,9 +1909,11 @@ def automatic_config(self): # re-discovery (NVS wipe, gateway reboot, repair). PredBat core is positional; # sorting by serial makes the slot index a deterministic function of identity. inverters = sorted(inverters, key=lambda inv: inv.serial) + pv_only = sorted(pv_only, key=lambda inv: inv.serial) num_inverters = len(inverters) self._inverter_slot_serials = [inv.serial for inv in inverters] + self._pv_only_serials = tuple(inv.serial for inv in pv_only) self.log(f"Info: GatewayMQTT: auto-config: {num_inverters} primary inverter(s) of {len(all_inverters)} total") # Set inverter type @@ -1463,6 +1944,7 @@ def automatic_config(self): battery_rate_max_entities = [] inverter_time_entities = [] + slots_without_capacity = [] for inv in inverters: suffix = inv.serial[-6:].lower() base = f"{self.prefix}_gateway_{suffix}" @@ -1487,10 +1969,11 @@ def automatic_config(self): export_limit_entities.append(f"sensor.{base}_export_limit_w") # soc_max: from battery capacity entity - if inv.battery.capacity_wh > 0: + if inv.battery.capacity_wh > 0 or self._discovery.get(inv.serial.upper(), {}).get("capacity_wh"): soc_max_entities.append(f"sensor.{base}_battery_capacity") else: soc_max_entities.append(None) + slots_without_capacity.append(inv.serial) self.log(f"Warn: GatewayMQTT: inverter {inv.serial} has no battery capacity, setting to None for automatic discovery") inverter_limit_entities.append(f"sensor.{base}_inverter_rate_max") @@ -1499,6 +1982,14 @@ def automatic_config(self): # Clock drift detection — uses GatewayStatus.timestamp inverter_time_entities.append(f"sensor.{base}_inverter_time") + # A listed inverter with no battery is a PV source and nothing else: it has no slot, so it + # is never written to, and its generation is counted through the entries past num_inverters + # in pv_power and pv_today, which Predbat sums (as solis.py does for the same case). + pv_only_bases = [f"{self.prefix}_gateway_{_serial_suffix(inv.serial)}" for inv in pv_only] + if pv_only_bases: + self.log(f"Info: GatewayMQTT: inverter(s) {list(self._pv_only_serials)} have no battery — configured as PV only") + pv_power_entities += [f"sensor.{base}_pv_power" for base in pv_only_bases] + # Map entity lists to PredBat args self.set_arg("soc_percent", soc_entities) self.set_arg("soc_max", soc_max_entities) @@ -1535,7 +2026,7 @@ def automatic_config(self): base0 = f"{self.prefix}_gateway_{suffix0}" bases = [f"{self.prefix}_gateway_{inv.serial[-6:].lower()}" for inv in inverters] site_bases = bases[:1] if self.gateway_shared_ct or self.gateway_integrate_power else bases - self.set_arg("pv_today", [f"sensor.{base}_pv_today" for base in bases]) + self.set_arg("pv_today", [f"sensor.{base}_pv_today" for base in bases + pv_only_bases]) self.set_arg("import_today", [f"sensor.{base}_import_today" for base in site_bases]) self.set_arg("export_today", [f"sensor.{base}_export_today" for base in site_bases]) self.set_arg("load_today", [f"sensor.{base}_load_today" for base in site_bases]) @@ -1606,7 +2097,8 @@ def automatic_config(self): self._register_ev_car(status) self._auto_configured = True - self._configured_inverter_serials = frozenset(inv.serial for inv in all_inverters) + self._slots_without_capacity = frozenset(slots_without_capacity) + self._configured_inverter_serials = self._usable_serials(status) | frozenset(self._inverter_slot_serials) self._configured_ev_chargers = frozenset(ev.charge_point_id for ev in status.ev_chargers if ev.charge_point_id) self.log(f"Info: GatewayMQTT: auto-config complete: {num_inverters} inverter(s) registered") return num_inverters @@ -1956,6 +2448,9 @@ async def _check_rate_caps(self): for serial, caps in self._inverter_rate_caps().items(): if self._rate_caps_sent.get(serial) == caps: continue + if _serial_suffix(serial) in self._assumed_ratings: + # The cap comes from a rating the inverter has not reported yet (see _rate_write_held) + continue charge_cap_w, discharge_cap_w = caps await self.publish_command("set_rate_cap", serial=serial, charge_cap_w=charge_cap_w, discharge_cap_w=discharge_cap_w) self._rate_caps_sent[serial] = caps @@ -2187,15 +2682,47 @@ async def number_event(self, entity_id, value): if serial is None: self.log(f"Warn: GatewayMQTT: number_event: cannot resolve serial for entity '{entity_id}' — command not sent") return + if ("_discharge_rate" in entity_id or "_charge_rate" in entity_id) and self._rate_write_held(serial, val): + return if "_discharge_rate" in entity_id: await self._send_control(entity_id, "set_discharge_rate", power_w=val, serial=serial) elif "_charge_rate" in entity_id: await self._send_control(entity_id, "set_charge_rate", power_w=val, serial=serial) elif "_reserve" in entity_id: await self._send_control(entity_id, "set_reserve", target_soc=val, serial=serial) + if val == 0: + # A reserve of 0% asked for here is real, so the next 0 from the hub is not a gap + # to fill with the last value (see _last_rating). Only once the write has gone + # out, and the value is set aside in case the hub refuses it (see _process_ack). + key = (_serial_suffix(serial), "reserve") + self._reserve_before_zero = {suffix: kept for suffix, kept in self._reserve_before_zero.items() if suffix != key[0]} + if key in self._last_ratings: + self._reserve_before_zero = {**self._reserve_before_zero, key[0]: self._last_ratings[key]} + self._last_ratings = {kept_key: kept for kept_key, kept in self._last_ratings.items() if kept_key != key} elif "_target_soc" in entity_id: await self._send_control(entity_id, "set_target_soc", target_soc=val, serial=serial) + def _rate_write_held(self, serial, power_w): + """Whether a charge or discharge rate write must wait for the inverter's rating. + + Until an inverter has reported its maximum rate, the rate Predbat works to is the 6000W + default, and writing that to a 3kW inverter is the fault in #5478. Such a write is not + sent. Predbat reads the unchanged rate back as a write that did not complete and tries + again, so it goes through once the rating is known. A rate of zero is safe on any + inverter and is always sent, so a charge or discharge can still be stopped. + + Args: + serial: The inverter the write is for. + power_w: The rate being written. + + Returns: + bool: True if the write is held. + """ + if power_w <= 0 or _serial_suffix(serial) not in self._assumed_ratings: + return False + self.log(f"Warn: GatewayMQTT: inverter {serial} has not reported its maximum rate — rate write of {power_w}W held until it does") + return True + async def switch_event(self, entity_id, service): """Handle switch entity service calls (charge/discharge enable). @@ -2371,6 +2898,7 @@ def _process_ack(self, data): apply_cache = ok and entry["cached"] is None command = entry["command"] serial = entry["kwargs"].get("serial") + zero_reserve = entry["kwargs"].get("target_soc") == 0 if ok: self.log(f"Info: GatewayMQTT: {command} for {serial} acknowledged by hub ({command_id}) applied={ack.get('applied')}") @@ -2384,6 +2912,13 @@ def _process_ack(self, data): self.log(f"Warn: GatewayMQTT: {command} for {serial} refused by hub: replay ({command_id} already used), {action}") else: self.log(f"Warn: GatewayMQTT: {command} for {serial} refused by hub: {error}") + if command == "set_reserve" and zero_reserve and serial: + # The reserve of 0 was not applied, so a 0 from the hub is a gap again and the + # reserve it had before is still the last real one + key = (_serial_suffix(serial), "reserve") + if key[0] in self._reserve_before_zero and key not in self._last_ratings: + self._last_ratings = {**self._last_ratings, key: self._reserve_before_zero[key[0]]} + restore_status = True if restore_status and self._last_status is not None: self._inject_entities(self._last_status) diff --git a/apps/predbat/tests/test_gateway.py b/apps/predbat/tests/test_gateway.py index e207c81a7..f83d2b08b 100644 --- a/apps/predbat/tests/test_gateway.py +++ b/apps/predbat/tests/test_gateway.py @@ -5,6 +5,8 @@ import sys import os import math +import json +import asyncio import unittest sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..")) @@ -3502,6 +3504,11 @@ def _make_handler_gateway(self): gw.api_started = False gw._last_telemetry_time = 0 gw.update_success_timestamp = MagicMock() + gw.non_fatal_error_occurred = MagicMock() + # Entity states as the gateway published them, so a test can tell stale from rewritten + gw._states = {} + gw.dashboard_item = lambda entity_id, state, attributes=None, app=None: gw._states.__setitem__(entity_id, state) + gw.get_state_wrapper = lambda entity_id=None, default=None, **kwargs: gw._states.get(entity_id, default) return gw def _pv_only_site_status(self, battery_ready): @@ -3562,6 +3569,918 @@ def test_incomplete_battery_telemetry_waits_then_excludes_pv_only_unit(self): assert gw._args["charge_start_time"] == ["select.predbat_gateway_00b002_charge_slot1_start"] assert all("00p001" not in entity for entity in gw._args["charge_start_time"]) + # ------------------------------------------------------------------ + # gateway_inverter_serial is authoritative, and an unusable inverter is picked up later (#5471) + # ------------------------------------------------------------------ + + def _two_ac_status(self, second="ok"): + """Two AC battery inverters on one hub; *second* is how CE2225G400 is reported. + + "ok" is healthy, "no_battery" has no readings at all (the hub is not reading it), "disconnected" also has + connected false, "not_primary" has battery data but no primary flag, and "absent" + leaves it out of the status message. + """ + status = pb.GatewayStatus() + status.device_id = "pbgw_two_ac" + status.firmware = "1.0.34" + status.schema_version = 1 + for serial in ("CE2223G800", "CE2225G400"): + if serial == "CE2225G400" and second == "absent": + continue + inv = status.inverters.add() + inv.type = pb.INVERTER_TYPE_GIVENERGY + inv.serial = serial + inv.primary = True + inv.connected = True + inv.active = True + inv.battery.soc_percent = 50 + inv.battery.capacity_wh = 9500 + inv.battery.rate_max_w = 3000 + if second in ("no_battery", "disconnected"): + status.inverters[1].ClearField("battery") + if second == "disconnected": + status.inverters[1].connected = False + if second == "not_primary": + status.inverters[1].primary = False + return status + + def _errors(self, gw): + """The Error lines the gateway has logged.""" + return [str(c.args[0]) for c in gw.log.call_args_list if str(c.args[0]).startswith("Error")] + + def _logged(self, gw, level, *words): + """Whether a log line of this level holds every one of *words*.""" + return any(str(c.args[0]).startswith(level) and all(word in str(c.args[0]) for word in words) for c in gw.log.call_args_list) + + def _listed_gateway(self, serials, stored=None): + """A handler gateway with a serial list and a storage holding *stored*, loaded as run() does at start-up.""" + gw = self._make_handler_gateway() + gw.gateway_inverter_serial = list(serials) + gw.gateway_device_id = "pbgw_two_ac" + saved = {} + if stored is not None: + saved[("gateway", "discovery_pbgw_two_ac")] = {"inverters": stored} + + class FakeStorage: + """Storage holding what the gateway saved.""" + + async def load(self, module, filename): + """Return the saved data.""" + return saved.get((module, filename)) + + async def save(self, module, filename, data, format="yaml", expiry=None, indent=None): + """Keep the data.""" + saved[(module, filename)] = json.loads(json.dumps(data)) + return True + + gw._saved = saved + gw.base.components.get_component.return_value = FakeStorage() + asyncio.run(gw._load_discovery()) + return gw + + _STORED_400 = {"CE2225G400": {"serial": "CE2225G400", "type": pb.INVERTER_TYPE_GIVENERGY, "battery": True, "capacity_wh": 9500, "rate_max_w": 3000, "inverter_rate_max_w": 0, "export_limit_w": 0, "model": ""}} + + def test_listed_inverter_never_read_is_left_out_until_the_hub_reads_it(self): + """A listed inverter the hub has never read is left out, with a warning and no error, and bound once it is read.""" + for second in ("no_battery", "disconnected", "absent"): + gw = self._listed_gateway(["CE2223G800", "CE2225G400"]) + + gw._process_telemetry(self._two_ac_status(second).SerializeToString()) + + assert gw._args["num_inverters"] == 1, second + assert self._logged(gw, "Warn", "CE2225G400", "not been read by the hub yet"), second + assert self._errors(gw) == [], second + gw.non_fatal_error_occurred.assert_not_called() + + gw._process_telemetry(self._two_ac_status("ok").SerializeToString()) + + assert gw._args["num_inverters"] == 2, second + assert gw._inverter_slot_serials == ["CE2223G800", "CE2225G400"], second + + def test_listed_inverter_is_bound_whatever_its_primary_flag(self): + """The list decides which battery inverters are used, not the hub's primary flag.""" + gw = self._listed_gateway(["CE2223G800", "CE2225G400"]) + + gw._process_telemetry(self._two_ac_status("not_primary").SerializeToString()) + + assert gw._args["num_inverters"] == 2 + assert gw._args["soc_percent"] == ["sensor.predbat_gateway_23g800_soc", "sensor.predbat_gateway_25g400_soc"] + + def test_stored_inverter_unread_at_startup_is_still_bound(self): + """An inverter discovered on an earlier run keeps its slot when the hub cannot read it at start-up (the #5471 incident).""" + for second in ("no_battery", "disconnected", "absent"): + gw = self._listed_gateway(["CE2223G800", "CE2225G400"], stored=self._STORED_400) + + gw._process_telemetry(self._two_ac_status(second).SerializeToString()) + + assert gw._auto_configured is True, second + assert gw._args["num_inverters"] == 2, second + assert gw._inverter_slot_serials == ["CE2223G800", "CE2225G400"], second + assert gw._args["soc_max"] == ["sensor.predbat_gateway_23g800_battery_capacity", "sensor.predbat_gateway_25g400_battery_capacity"], second + # Nothing was published for it before, so it gets default entities and its stored size + assert gw._states["sensor.predbat_gateway_25g400_soc"] == 0, second + assert gw._states["sensor.predbat_gateway_25g400_battery_capacity"] == 9.5, second + assert gw._states["select.predbat_gateway_25g400_charge_slot1_start"] == "00:00:00", second + assert self._logged(gw, "Error", "CE2225G400", "not being read"), second + assert gw.non_fatal_error_occurred.call_count == 1, second + + def _empty_status(self): + """A status in which the hub names no inverter at all.""" + status = pb.GatewayStatus() + status.device_id = "pbgw_two_ac" + status.firmware = "1.0.34" + status.schema_version = 1 + return status + + def test_stored_inverter_is_bound_when_the_hub_reports_no_inverters_at_startup(self): + """A single listed inverter with a stored record keeps its slot when the first status names no inverter.""" + gw = self._listed_gateway(["CE2225G400"], stored=self._STORED_400) + + gw._process_telemetry(self._empty_status().SerializeToString()) + + assert gw._auto_configured is True + assert gw._args["num_inverters"] == 1 + assert gw._inverter_slot_serials == ["CE2225G400"] + assert gw._states["sensor.predbat_gateway_25g400_soc"] == 0 + assert gw._states["sensor.predbat_gateway_25g400_battery_capacity"] == 9.5 + assert self._logged(gw, "Error", "CE2225G400", "not being read") + assert gw.non_fatal_error_occurred.call_count == 1 + # Nothing was read, so the status does not count towards the hub being alive + gw.update_success_timestamp.assert_not_called() + + def test_status_with_no_inverters_leaves_configured_inverters_unread(self): + """Once configured, a status naming no inverter zeroes the powers, keeps the rest and is an error.""" + gw = self._listed_gateway(["CE2223G800", "CE2225G400"]) + healthy = self._two_ac_status("ok") + for inv in healthy.inverters: + inv.battery.power_w = -3000 + gw._process_telemetry(healthy.SerializeToString()) + assert gw._states["sensor.predbat_gateway_23g800_battery_power"] == 3000 + assert self._errors(gw) == [] + + gw._process_telemetry(self._empty_status().SerializeToString()) + + assert gw._args["num_inverters"] == 2 + for suffix in ("23g800", "25g400"): + assert gw._states[f"sensor.predbat_gateway_{suffix}_battery_power"] == 0, suffix + assert gw._states[f"sensor.predbat_gateway_{suffix}_soc"] == 50, suffix + assert self._logged(gw, "Error", "CE2223G800", "CE2225G400", "not being read") + assert gw.non_fatal_error_occurred.call_count == 1 + + gw._process_telemetry(healthy.SerializeToString()) + assert gw._states["sensor.predbat_gateway_23g800_battery_power"] == 3000 + assert self._logged(gw, "Info", "every configured inverter is being read by the hub again") + + def test_status_with_no_inverters_is_ignored_without_a_list_or_a_record(self): + """With no serial list, or no inverter discovered yet, a status naming no inverter is ignored as before.""" + for listed, stored in (([], None), (["CE2225G400"], None)): + gw = self._listed_gateway(listed, stored=stored) + + gw._process_telemetry(self._empty_status().SerializeToString()) + + assert gw._last_status is None, listed + assert gw._auto_configured is False, listed + assert gw._args == {}, listed + + # No list: inverters come and go, so one that goes is not held on to + gw = self._listed_gateway([]) + gw._process_telemetry(self._two_ac_status("ok").SerializeToString()) + healthy = gw._last_status + gw._process_telemetry(self._empty_status().SerializeToString()) + assert gw._last_status is healthy + assert self._errors(gw) == [] + + def test_changed_serial_list_wins_over_stored_discovery(self): + """A stored record for a serial no longer in the list is not used: the list is the setting.""" + gw = self._listed_gateway(["CE2223G800"], stored=self._STORED_400) + assert gw._discovery == {} + + gw._process_telemetry(self._two_ac_status("ok").SerializeToString()) + asyncio.run(gw._save_discovery()) + + assert gw._args["num_inverters"] == 1 + assert gw._inverter_slot_serials == ["CE2223G800"] + assert sorted(gw._saved[("gateway", "discovery_pbgw_two_ac")]["inverters"]) == ["CE2223G800"] + gw.non_fatal_error_occurred.assert_not_called() + + def test_discovery_is_saved_and_not_overwritten_while_an_inverter_is_offline(self): + """Discovery is stored from an inverter the hub is reading, updated when it changes, and left alone while it is offline.""" + gw = self._listed_gateway(["CE2223G800", "CE2225G400"]) + gw._process_telemetry(self._two_ac_status("ok").SerializeToString()) + asyncio.run(gw._save_discovery()) + stored = gw._saved[("gateway", "discovery_pbgw_two_ac")]["inverters"] + assert stored["CE2225G400"] == self._STORED_400["CE2225G400"] + + for second in ("no_battery", "disconnected", "absent"): + gw._process_telemetry(self._two_ac_status(second).SerializeToString()) + assert gw._discovery_dirty is False, second + assert gw._discovery["CE2225G400"] == self._STORED_400["CE2225G400"], second + + bigger = self._two_ac_status("ok") + bigger.inverters[1].battery.capacity_wh = 13500 + gw._process_telemetry(bigger.SerializeToString()) + asyncio.run(gw._save_discovery()) + assert gw._saved[("gateway", "discovery_pbgw_two_ac")]["inverters"]["CE2225G400"]["capacity_wh"] == 13500 + + # A second gateway started from that storage knows both inverters before any telemetry + restarted = self._listed_gateway(["CE2223G800", "CE2225G400"], stored=gw._saved[("gateway", "discovery_pbgw_two_ac")]["inverters"]) + assert sorted(restarted._discovery) == ["CE2223G800", "CE2225G400"] + + def test_failed_discovery_save_is_retried_after_a_wait(self): + """A save the storage refuses is kept pending and tried again after a wait, not on every status message.""" + import gateway + + gw = self._listed_gateway(["CE2223G800", "CE2225G400"]) + storage = gw.base.components.get_component.return_value + real_save = storage.save + attempts = [] + + async def failing_save(module, filename, data, format="yaml", expiry=None, indent=None): + attempts.append(1) + return False + + now = [1000.0] + real_monotonic = gateway._monotonic + gateway._monotonic = lambda: now[0] + try: + storage.save = failing_save + gw._process_telemetry(self._two_ac_status("ok").SerializeToString()) + asyncio.run(gw._save_discovery()) + assert attempts == [1] + assert gw._discovery_dirty is True + assert gw._saved == {} + assert self._logged(gw, "Warn", "failed to save inverter discovery") + + # Still failing, but inside the wait: not tried again + asyncio.run(gw._save_discovery()) + assert attempts == [1] + + storage.save = real_save + now[0] += gateway.GATEWAY_DISCOVERY_SAVE_RETRY_SECONDS + asyncio.run(gw._save_discovery()) + assert gw._discovery_dirty is False + assert sorted(gw._saved[("gateway", "discovery_pbgw_two_ac")]["inverters"]) == ["CE2223G800", "CE2225G400"] + finally: + gateway._monotonic = real_monotonic + + def test_newly_bound_inverter_has_its_entities_on_the_same_status(self): + """An inverter auto-config has just bound is published before the API is declared started. + + One the hub does not flag primary is skipped by the injection that runs ahead of + auto-config, so without a second pass its soc and size entities would not exist + until the next status message. + """ + gw = self._listed_gateway(["CE2223G800", "CE2225G400"]) + + gw._process_telemetry(self._two_ac_status("not_primary").SerializeToString()) + + assert gw.api_started is True + assert gw._args["num_inverters"] == 2 + assert gw._states["sensor.predbat_gateway_25g400_soc"] == 50 + assert gw._states["sensor.predbat_gateway_25g400_battery_capacity"] == 9.5 + + # The same for a PV-only inverter the hub does not flag primary + gw = self._listed_gateway(["FD0000P001", "CH0000B002"]) + status = self._pv_and_battery_status(pv_power=900) + status.inverters[0].primary = False + + gw._process_telemetry(status.SerializeToString()) + + assert gw._pv_only_serials == ("FD0000P001",) + assert gw._states["sensor.predbat_gateway_00p001_pv_power"] == 900 + + def test_no_serial_list_stores_last_known_values_but_inverters_come_and_go(self): + """With no serial list the last known values are stored and used, but a stored inverter is not held on to.""" + first = self._listed_gateway([]) + first._process_telemetry(self._rated_status(rate_max_w=3000, inverter_rate_max_w=3600).SerializeToString()) + asyncio.run(first._save_discovery()) + stored = first._saved[("gateway", "discovery_pbgw_two_ac")]["inverters"] + assert sorted(stored) == ["CE2223G800", "CE2225G400"] + assert stored["CE2225G400"]["rate_max_w"] == 3000 + assert first._args["num_inverters"] == 2 + + # After a restart the hub does not report the second inverter: it is simply not there + gw = self._listed_gateway([], stored=stored) + assert sorted(gw._discovery) == ["CE2223G800", "CE2225G400"] + gw._process_telemetry(self._two_ac_status("absent").SerializeToString()) + assert gw._args["num_inverters"] == 1 + assert gw._inverter_slot_serials == ["CE2223G800"] + assert self._errors(gw) == [] + gw.non_fatal_error_occurred.assert_not_called() + + # Nor is one the hub lists but is not reading + gw = self._listed_gateway([], stored=stored) + gw._process_telemetry(self._two_ac_status("no_battery").SerializeToString()) + assert gw._args["num_inverters"] == 1 + + # But when it is there with its ratings and size unpopulated, the stored ones are used + gw = self._listed_gateway([], stored=stored) + blank = self._rated_status(rate_max_w=0, inverter_rate_max_w=0, export_limit_w=0) + blank.inverters[1].battery.capacity_wh = 0 + gw._process_telemetry(blank.SerializeToString()) + assert gw._args["num_inverters"] == 2 + assert gw._states["sensor.predbat_gateway_25g400_battery_rate_max"] == 3000 + assert gw._states["sensor.predbat_gateway_25g400_inverter_rate_max"] == 3600 + assert gw._states["sensor.predbat_gateway_25g400_export_limit_w"] == 6000 + assert gw._states["sensor.predbat_gateway_25g400_battery_capacity"] == 9.5 + assert gw._args["soc_max"] == ["sensor.predbat_gateway_23g800_battery_capacity", "sensor.predbat_gateway_25g400_battery_capacity"] + + def test_inverter_with_no_serial_is_not_recorded(self): + """A slot the hub reports with an empty serial has nothing to be stored under.""" + gw = self._listed_gateway([]) + status = self._two_ac_status("ok") + status.inverters[1].serial = "" + gw._process_telemetry(status.SerializeToString()) + assert sorted(gw._discovery) == ["CE2223G800"] + + def test_configured_inverter_offline_reports_an_error_until_it_returns(self): + """A configured inverter the hub stops reading is an error on every frame, logged once, and clears when it returns.""" + for second in ("no_battery", "disconnected", "absent"): + gw = self._listed_gateway(["CE2223G800", "CE2225G400"]) + gw._process_telemetry(self._two_ac_status("ok").SerializeToString()) + assert self._errors(gw) == [] + + gw._process_telemetry(self._two_ac_status(second).SerializeToString()) + gw._process_telemetry(self._two_ac_status(second).SerializeToString()) + + unread = [line for line in self._errors(gw) if "not being read" in line] + assert len(unread) == 1 and "CE2225G400" in unread[0] and "CE2223G800" not in unread[0], (second, unread) + assert gw.non_fatal_error_occurred.call_count == 2, second + assert gw._args["num_inverters"] == 2, second + + gw._process_telemetry(self._two_ac_status("ok").SerializeToString()) + + assert gw.non_fatal_error_occurred.call_count == 2, second + assert self._logged(gw, "Info", "being read by the hub again"), second + + def test_configured_inverter_going_offline_keeps_its_slot_and_its_last_data(self): + """An inverter that stops being read stays configured and its entities go stale, they are not zeroed.""" + for listed in (["CE2223G800", "CE2225G400"], []): + gw = self._listed_gateway(listed) + healthy = self._two_ac_status("ok") + healthy.inverters[1].battery.soc_percent = 84 + healthy.inverters[1].battery.power_w = -3000 + healthy.inverters[1].pv.power_w = 400 + gw._process_telemetry(healthy.SerializeToString()) + assert gw._states["sensor.predbat_gateway_25g400_soc"] == 84 + assert gw._states["sensor.predbat_gateway_25g400_battery_power"] == 3000 + + offline = self._two_ac_status("no_battery") + offline.inverters[0].battery.soc_percent = 61 + for _ in range(2): + gw._process_telemetry(offline.SerializeToString()) + + assert gw._args["num_inverters"] == 2, listed + assert gw._states["sensor.predbat_gateway_25g400_soc"] == 84, listed + assert gw._states["sensor.predbat_gateway_25g400_battery_capacity"] == 9.5, listed + assert gw._states["sensor.predbat_gateway_23g800_soc"] == 61, listed + # A power is a rate, so it is not left at the value it had when the inverter dropped out + assert gw._states["sensor.predbat_gateway_25g400_battery_power"] == 0, listed + assert gw._states["sensor.predbat_gateway_25g400_pv_power"] == 0, listed + assert gw._states["sensor.predbat_gateway_25g400_inverter_power"] == 0, listed + + # Gone from the status message altogether: still configured, still stale + healthy.inverters[1].battery.power_w = -2500 + gw._process_telemetry(healthy.SerializeToString()) + assert gw._states["sensor.predbat_gateway_25g400_battery_power"] == 2500, listed + gw._process_telemetry(self._two_ac_status("absent").SerializeToString()) + assert gw._args["num_inverters"] == 2, listed + assert gw._states["sensor.predbat_gateway_25g400_soc"] == 84, listed + assert gw._states["sensor.predbat_gateway_25g400_battery_power"] == 0, listed + assert gw.non_fatal_error_occurred.call_count == (3 if listed else 0), listed + + healthy.inverters[1].battery.soc_percent = 80 + gw._process_telemetry(healthy.SerializeToString()) + assert gw._states["sensor.predbat_gateway_25g400_soc"] == 80, listed + + def test_shared_ct_metering_carries_on_when_the_first_inverter_is_unread(self): + """On a shared clamp the site readings live on the first inverter's entities, and are filled from the next one while it is away.""" + + def status(first): + built = self._two_ac_status("ok") + for inv, grid_w in zip(built.inverters, (-1500, -1500)): + inv.grid.power_w = grid_w + inv.load.power_w = 1800 + inv.energy.grid_import_today_wh = 4200 + inv.energy.grid_export_today_wh = 300 + inv.energy.consumption_today_wh = 6100 + if first == "unread": + for block in ("battery", "pv", "grid", "load", "inverter", "energy"): + built.inverters[0].ClearField(block) + elif first == "absent": + del built.inverters[0] + return built + + for first in ("unread", "absent"): + for listed in (["CE2223G800", "CE2225G400"], []): + gw = self._listed_gateway(listed) + gw._process_telemetry(status("ok").SerializeToString()) + assert gw._args["grid_power"] == ["sensor.predbat_gateway_23g800_grid_power", 0], (first, listed) + + later = status(first) + later.inverters[-1].grid.power_w = -900 + later.inverters[-1].load.power_w = 1100 + later.inverters[-1].energy.grid_import_today_wh = 5000 + later.inverters[-1].energy.consumption_today_wh = 7400 + gw._process_telemetry(later.SerializeToString()) + + assert gw._states["sensor.predbat_gateway_23g800_grid_power"] == -900, (first, listed) + assert gw._states["sensor.predbat_gateway_23g800_load_power"] == 1100, (first, listed) + assert gw._states["sensor.predbat_gateway_23g800_import_today"] == 5.0, (first, listed) + assert gw._states["sensor.predbat_gateway_23g800_export_today"] == 0.3, (first, listed) + assert gw._states["sensor.predbat_gateway_23g800_load_today"] == 7.4, (first, listed) + # Its own battery is still unknown: stale state of charge, no power + assert gw._states["sensor.predbat_gateway_23g800_soc"] == 50, (first, listed) + assert gw._states["sensor.predbat_gateway_23g800_battery_power"] == 0, (first, listed) + + def test_separate_ct_metering_is_not_filled_from_another_inverter(self): + """With a clamp each (gateway_shared_ct off) an unread inverter's grid reading is its own, so it goes to zero.""" + gw = self._listed_gateway(["CE2223G800", "CE2225G400"]) + gw.gateway_shared_ct = False + healthy = self._two_ac_status("ok") + for inv in healthy.inverters: + inv.grid.power_w = -700 + gw._process_telemetry(healthy.SerializeToString()) + + unread = self._two_ac_status("ok") + unread.inverters[1].grid.power_w = -700 + unread.inverters[0].ClearField("battery") + gw._process_telemetry(unread.SerializeToString()) + + assert gw._states["sensor.predbat_gateway_23g800_grid_power"] == 0 + assert gw._states["sensor.predbat_gateway_25g400_grid_power"] == -700 + + def test_unread_inverter_behind_a_coordinator_is_an_error(self): + """Behind a Gateway the Gateway holds the only slot, and one of its listed inverters going unread is still reported.""" + gw = self._listed_gateway(["GW2347G077", "CH2414G318", "CH9999G999"]) + gw._process_telemetry(self._gateway_plus_aios_status(["CH2414G318", "CH9999G999"]).SerializeToString()) + assert self._errors(gw) == [] + + unread = self._gateway_plus_aios_status(["CH2414G318", "CH9999G999"]) + unread.inverters[2].ClearField("battery") + gw._process_telemetry(unread.SerializeToString()) + + assert gw._args["num_inverters"] == 1 + assert self._logged(gw, "Error", "CH9999G999", "not being read") + assert not self._logged(gw, "Error", "CH2414G318") + assert gw.non_fatal_error_occurred.call_count == 1 + + gw._process_telemetry(self._gateway_plus_aios_status(["CH2414G318"]).SerializeToString()) + assert gw.non_fatal_error_occurred.call_count == 2 + + def test_unreadable_stored_discovery_is_ignored(self): + """A stored record that cannot be read is dropped with a warning; it must not stop auto-config.""" + good = self._STORED_400["CE2225G400"] + for bad in ({"serial": "CE2223G800", "type": "GIVENERGY"}, {"serial": "CE2223G800", "capacity_wh": "big"}, {"type": 1}, "CE2223G800", None, {"serial": " "}): + gw = self._listed_gateway(["CE2223G800", "CE2225G400"], stored={"CE2223G800": bad, "CE2225G400": good}) + + assert gw._discovery == {"CE2225G400": good}, bad + assert self._logged(gw, "Warn", "CE2223G800", "unreadable stored discovery"), bad + + gw._process_telemetry(self._two_ac_status("absent").SerializeToString()) + assert gw._args["num_inverters"] == 2, bad + + # A record that says it has a battery in any way but true is not taken to have one + gw = self._listed_gateway(["CE2225G400"], stored={"CE2225G400": {**good, "battery": "false", "capacity_wh": None}}) + assert gw._discovery["CE2225G400"]["battery"] is False + assert gw._discovery["CE2225G400"]["capacity_wh"] == 0 + + def test_stored_pv_only_inverter_missing_at_startup_has_its_pv_entities(self): + """A PV-only inverter known only from its stored record still has the PV entities auto-config binds.""" + stored = {"FD0000P001": {"serial": "FD0000P001", "type": pb.INVERTER_TYPE_GIVENERGY, "battery": False, "capacity_wh": 0, "rate_max_w": 0, "inverter_rate_max_w": 0, "export_limit_w": 0, "model": ""}} + gw = self._listed_gateway(["FD0000P001", "CH0000B002"], stored=stored) + status = self._pv_only_site_status(battery_ready=True) + del status.inverters[0] + + gw._process_telemetry(status.SerializeToString()) + + assert gw._pv_only_serials == ("FD0000P001",) + assert gw._args["pv_power"] == ["sensor.predbat_gateway_00b002_pv_power", "sensor.predbat_gateway_00p001_pv_power"] + assert gw._states["sensor.predbat_gateway_00p001_pv_power"] == 0 + assert gw._states["sensor.predbat_gateway_00p001_pv_today"] == 0 + + def test_coordinator_slot_has_no_battery_size_sensor(self): + """A slot with no battery size known is left for automatic sizing, not pointed at a sensor that is never published.""" + gw = self._listed_gateway(["GW2347G077", "CH2414G318", "CH9999G999"]) + + gw._process_telemetry(self._gateway_plus_aios_status(["CH2414G318", "CH9999G999"]).SerializeToString()) + + assert gw._args["soc_max"] == [None] + assert "sensor.predbat_gateway_47g077_battery_capacity" not in gw._states + + def test_slot_bound_with_no_battery_size_is_rebound_when_the_hub_reports_one(self): + """A battery inverter bound before its size is known gets its size sensor on the first status that carries one.""" + for listed in (["CE2223G800", "CE2225G400"], []): + gw = self._listed_gateway(listed) + unsized = self._two_ac_status("ok") + unsized.inverters[1].battery.capacity_wh = 0 + gw._process_telemetry(unsized.SerializeToString()) + assert gw._args["soc_max"] == ["sensor.predbat_gateway_23g800_battery_capacity", None], listed + + gw._process_telemetry(unsized.SerializeToString()) + assert gw._args["soc_max"][1] is None, listed + + gw._process_telemetry(self._two_ac_status("ok").SerializeToString()) + assert gw._args["soc_max"] == ["sensor.predbat_gateway_23g800_battery_capacity", "sensor.predbat_gateway_25g400_battery_capacity"], listed + assert gw._states["sensor.predbat_gateway_25g400_battery_capacity"] == 9.5, listed + + # ------------------------------------------------------------------ + # A rating the hub stops reporting keeps its last good value (#5478) + # ------------------------------------------------------------------ + + def _rated_status(self, rate_max_w=3000, inverter_rate_max_w=3000, export_limit_w=6000): + """Two 3kW AC inverters; the arguments are what CE2225G400 reports for its ratings.""" + status = self._two_ac_status("ok") + for inv in status.inverters: + inv.inverter.rate_max_w = 3000 + inv.control.export_limit_w = 6000 + inv.control.reserve_soc = 4 + second = status.inverters[1] + second.battery.rate_max_w = rate_max_w + second.inverter.rate_max_w = inverter_rate_max_w + second.control.export_limit_w = export_limit_w + return status + + def test_ratings_the_hub_stops_reporting_keep_their_last_good_value(self): + """A status with the ratings unpopulated must not turn a 3kW inverter into a 6kW one with no export limit.""" + for listed in (["CE2223G800", "CE2225G400"], []): + gw = self._listed_gateway(listed) + gw._process_telemetry(self._rated_status().SerializeToString()) + assert gw._states["sensor.predbat_gateway_25g400_battery_rate_max"] == 3000, listed + assert gw._states["sensor.predbat_gateway_25g400_inverter_rate_max"] == 3000, listed + assert gw._states["sensor.predbat_gateway_25g400_export_limit_w"] == 6000, listed + + gw._process_telemetry(self._rated_status(rate_max_w=0, inverter_rate_max_w=0, export_limit_w=0).SerializeToString()) + + assert gw._states["sensor.predbat_gateway_25g400_battery_rate_max"] == 3000, listed + assert gw._states["sensor.predbat_gateway_25g400_inverter_rate_max"] == 3000, listed + assert gw._states["sensor.predbat_gateway_25g400_export_limit_w"] == 6000, listed + # The other inverter is untouched + assert gw._states["sensor.predbat_gateway_23g800_battery_rate_max"] == 3000, listed + + # A new real value is taken straight away + gw._process_telemetry(self._rated_status(rate_max_w=3600, inverter_rate_max_w=5000, export_limit_w=3680).SerializeToString()) + assert gw._states["sensor.predbat_gateway_25g400_battery_rate_max"] == 3600, listed + assert gw._states["sensor.predbat_gateway_25g400_inverter_rate_max"] == 5000, listed + assert gw._states["sensor.predbat_gateway_25g400_export_limit_w"] == 3680, listed + + def test_zero_export_limit_is_kept_when_the_hub_stops_reporting_it(self): + """A real limit of zero (the hub's value 1) is a limit like any other: a later undefined keeps it.""" + gw = self._listed_gateway([]) + gw._process_telemetry(self._rated_status(export_limit_w=1).SerializeToString()) + assert gw._states["sensor.predbat_gateway_25g400_export_limit_w"] == 0 + + gw._process_telemetry(self._rated_status(export_limit_w=0).SerializeToString()) + assert gw._states["sensor.predbat_gateway_25g400_export_limit_w"] == 0 + + def test_ratings_never_reported_keep_the_existing_defaults(self): + """An inverter that has never reported a rating still gets the 6000W default and no export limit.""" + gw = self._listed_gateway([]) + gw._process_telemetry(self._rated_status(rate_max_w=0, inverter_rate_max_w=0, export_limit_w=0).SerializeToString()) + + assert gw._states["sensor.predbat_gateway_25g400_battery_rate_max"] == 6000 + assert gw._states["sensor.predbat_gateway_25g400_inverter_rate_max"] == 6000 + assert gw._states["sensor.predbat_gateway_25g400_export_limit_w"] == 99999 + + def _writable(self, gw): + """Let *gw* take control writes, recording the commands it publishes.""" + from unittest.mock import MagicMock + + gw._mqtt_connected = True + gw._mqtt_client = MagicMock() + gw._gateway_online = True + gw._rate_caps_sent = {} + gw.is_alive = lambda: True + gw.get_arg = lambda name, default=None, **kwargs: default + gw._published = [] + + async def fake_publish_command(command, **kwargs): + gw._published.append((command, kwargs)) + + gw.publish_command = fake_publish_command + return gw + + def test_rate_writes_are_held_until_the_inverter_reports_its_rating(self): + """With no rating ever reported the 6000W default is only an assumption, so no rate above zero is written on it.""" + gw = self._writable(self._listed_gateway([])) + gw._process_telemetry(self._rated_status(rate_max_w=0, inverter_rate_max_w=0).SerializeToString()) + assert gw._states["sensor.predbat_gateway_25g400_battery_rate_max"] == 6000 + + asyncio.run(gw.number_event("number.predbat_gateway_25g400_discharge_rate", 6000)) + asyncio.run(gw.number_event("number.predbat_gateway_25g400_charge_rate", 6000)) + assert gw._published == [] + assert self._logged(gw, "Warn", "CE2225G400", "has not reported its maximum rate") + + # Stopping a charge or discharge is safe at any rating, as are the other controls and the other inverter + asyncio.run(gw.number_event("number.predbat_gateway_25g400_charge_rate", 0)) + asyncio.run(gw.number_event("number.predbat_gateway_25g400_reserve_soc", 4)) + asyncio.run(gw.number_event("number.predbat_gateway_23g800_discharge_rate", 3000)) + assert gw._published == [ + ("set_charge_rate", {"power_w": 0, "serial": "CE2225G400"}), + ("set_reserve", {"target_soc": 4, "serial": "CE2225G400"}), + ("set_discharge_rate", {"power_w": 3000, "serial": "CE2223G800"}), + ] + + # Once the rating arrives the write goes through, and a later gap in the rating does not hold it again + for rate_max_w in (3000, 0): + gw._published.clear() + gw._process_telemetry(self._rated_status(rate_max_w=rate_max_w).SerializeToString()) + asyncio.run(gw.number_event("number.predbat_gateway_25g400_discharge_rate", 3000)) + assert gw._published == [("set_discharge_rate", {"power_w": 3000, "serial": "CE2225G400"})], rate_max_w + + def test_rate_cap_is_not_sent_on_an_assumed_rating(self): + """The cap sent to the hub comes from the same rating, so it waits for the real one too.""" + from types import SimpleNamespace + + gw = self._writable(self._listed_gateway([])) + gw._process_telemetry(self._rated_status(rate_max_w=0, inverter_rate_max_w=0).SerializeToString()) + gw.base.inverters = [SimpleNamespace(inverter_type="GWMQTT", id=slot, battery_rate_max_charge=0.05, battery_rate_max_discharge=0.05) for slot in (0, 1)] + gw._inverter_rate_caps = lambda: {"CE2223G800": (3000, 3000), "CE2225G400": (6000, 6000)} + + asyncio.run(gw._check_rate_caps()) + assert gw._published == [("set_rate_cap", {"serial": "CE2223G800", "charge_cap_w": 3000, "discharge_cap_w": 3000})] + + gw._process_telemetry(self._rated_status().SerializeToString()) + gw._inverter_rate_caps = lambda: {"CE2223G800": (3000, 3000), "CE2225G400": (3000, 3000)} + asyncio.run(gw._check_rate_caps()) + assert gw._published[1:] == [("set_rate_cap", {"serial": "CE2225G400", "charge_cap_w": 3000, "discharge_cap_w": 3000})] + + def test_coordinator_rate_writes_are_not_held(self): + """A Gateway or Plant EMS is not judged on a battery rating of its own.""" + gw = self._writable(self._listed_gateway([])) + status = self._gateway_plus_aios_status(["CH2414G318", "CH9999G999"]) + status.inverters[0].battery.rate_max_w = 0 + gw._process_telemetry(status.SerializeToString()) + assert gw._inverter_slot_serials == ["GW2347G077"] + + asyncio.run(gw.number_event("number.predbat_gateway_47g077_charge_rate", 6000)) + assert gw._published == [("set_charge_rate", {"power_w": 6000, "serial": "GW2347G077"})] + + def test_stored_ratings_are_used_after_a_restart(self): + """With a serial list the ratings are in the discovery record, so a restart does not fall back to 6000W.""" + first = self._listed_gateway(["CE2223G800", "CE2225G400"]) + first._process_telemetry(self._rated_status(rate_max_w=3000, inverter_rate_max_w=3600).SerializeToString()) + asyncio.run(first._save_discovery()) + stored = first._saved[("gateway", "discovery_pbgw_two_ac")]["inverters"] + assert stored["CE2225G400"]["rate_max_w"] == 3000 + assert stored["CE2225G400"]["inverter_rate_max_w"] == 3600 + assert stored["CE2225G400"]["export_limit_w"] == 6000 + + # Reporting, but with its ratings unpopulated + gw = self._listed_gateway(["CE2223G800", "CE2225G400"], stored=stored) + gw._process_telemetry(self._rated_status(rate_max_w=0, inverter_rate_max_w=0, export_limit_w=0).SerializeToString()) + assert gw._states["sensor.predbat_gateway_25g400_battery_rate_max"] == 3000 + assert gw._states["sensor.predbat_gateway_25g400_inverter_rate_max"] == 3600 + assert gw._states["sensor.predbat_gateway_25g400_export_limit_w"] == 6000 + + # Not being read at all + gw = self._listed_gateway(["CE2223G800", "CE2225G400"], stored=stored) + gw._process_telemetry(self._two_ac_status("no_battery").SerializeToString()) + assert gw._states["sensor.predbat_gateway_25g400_battery_rate_max"] == 3000 + assert gw._states["sensor.predbat_gateway_25g400_inverter_rate_max"] == 3600 + assert gw._states["sensor.predbat_gateway_25g400_export_limit_w"] == 6000 + + def test_stored_zero_export_limit_is_used_after_a_restart(self): + """A stored limit of zero (the hub's value 1) comes back as zero, not as no limit.""" + first = self._listed_gateway(["CE2223G800", "CE2225G400"]) + first._process_telemetry(self._rated_status(export_limit_w=1).SerializeToString()) + asyncio.run(first._save_discovery()) + stored = first._saved[("gateway", "discovery_pbgw_two_ac")]["inverters"] + + gw = self._listed_gateway(["CE2223G800", "CE2225G400"], stored=stored) + gw._process_telemetry(self._rated_status(export_limit_w=0).SerializeToString()) + assert gw._states["sensor.predbat_gateway_25g400_export_limit_w"] == 0 + + def test_reserve_the_hub_stops_reporting_keeps_its_last_good_value(self): + """A status with the reserve unpopulated must not read as a reserve of 0%, which Predbat would then rewrite.""" + gw = self._writable(self._listed_gateway([])) + gw._process_telemetry(self._rated_status().SerializeToString()) + assert gw._states["number.predbat_gateway_25g400_reserve_soc"] == 4 + + blank = self._rated_status() + blank.inverters[1].control.reserve_soc = 0 + gw._process_telemetry(blank.SerializeToString()) + assert gw._states["number.predbat_gateway_25g400_reserve_soc"] == 4 + + changed = self._rated_status() + changed.inverters[1].control.reserve_soc = 20 + gw._process_telemetry(changed.SerializeToString()) + assert gw._states["number.predbat_gateway_25g400_reserve_soc"] == 20 + + # A reserve of 0% that Predbat itself wrote is real, so the next 0 is published as 0 + asyncio.run(gw.number_event("number.predbat_gateway_25g400_reserve_soc", 0)) + assert gw._published == [("set_reserve", {"target_soc": 0, "serial": "CE2225G400"})] + gw._process_telemetry(blank.SerializeToString()) + assert gw._states["number.predbat_gateway_25g400_reserve_soc"] == 0 + # The other inverter's reserve is its own + assert gw._states["number.predbat_gateway_23g800_reserve_soc"] == 4 + + def test_reserve_is_kept_when_the_write_of_zero_fails(self): + """A reserve of 0% that never went out is not real: the hub's 0 is still a gap filled with the last value.""" + gw = self._writable(self._listed_gateway([])) + gw._process_telemetry(self._rated_status().SerializeToString()) + + async def failing_send(entity_id, command, **kwargs): + raise RuntimeError("publish failed") + + gw._send_control = failing_send + try: + asyncio.run(gw.number_event("number.predbat_gateway_25g400_reserve_soc", 0)) + raise AssertionError("the failed publish should have been raised") + except RuntimeError: + pass + + blank = self._rated_status() + blank.inverters[1].control.reserve_soc = 0 + gw._process_telemetry(blank.SerializeToString()) + assert gw._states["number.predbat_gateway_25g400_reserve_soc"] == 4 + + def test_reserve_is_put_back_when_the_hub_refuses_the_write_of_zero(self): + """A refused reserve of 0% was not applied, so the reserve the inverter had is published again.""" + import threading + + gw = self._writable(self._listed_gateway([])) + gw._ack_subscribed = True + gw._acks_seen = True + gw._command_id = 0 + gw._pending_commands = {} + gw._command_lock = threading.Lock() + gw._process_telemetry(self._rated_status().SerializeToString()) + blank = self._rated_status() + blank.inverters[1].control.reserve_soc = 0 + + asyncio.run(gw.number_event("number.predbat_gateway_25g400_reserve_soc", 0)) + command, sent = gw._published[-1] + assert command == "set_reserve" and sent["target_soc"] == 0 + gw._process_telemetry(blank.SerializeToString()) + assert gw._states["number.predbat_gateway_25g400_reserve_soc"] == 0 + + gw._process_ack(json.dumps({"command_id": f"PBAT{sent['command_id']}", "command": "set_reserve", "ok": False, "error": "modbus_write_failed"}).encode("utf-8")) + assert gw._states["number.predbat_gateway_25g400_reserve_soc"] == 4 + # The other inverter is untouched, and an accepted 0 stays 0 + assert gw._states["number.predbat_gateway_23g800_reserve_soc"] == 4 + + asyncio.run(gw.number_event("number.predbat_gateway_23g800_reserve_soc", 0)) + command, sent = gw._published[-1] + blank.inverters[0].control.reserve_soc = 0 + gw._process_telemetry(blank.SerializeToString()) + gw._process_ack(json.dumps({"command_id": f"PBAT{sent['command_id']}", "command": "set_reserve", "ok": True}).encode("utf-8")) + gw._process_telemetry(blank.SerializeToString()) + assert gw._states["number.predbat_gateway_23g800_reserve_soc"] == 0 + + def test_inverter_that_becomes_usable_is_the_integration_baseline_on_that_status(self): + """_process_telemetry re-baselines the integration after auto-config changes the slots.""" + import time + + gw = self._make_handler_gateway() + gw.gateway_integrate_power = True + gw.gateway_shared_ct = True + gw._integrated_energy = None + + gw._process_telemetry(self._two_ac_status("no_battery").SerializeToString()) + assert sorted(gw._integrated_energy["pv"]) == ["CE2223G800"] + + joining = self._two_ac_status("ok") + joining.timestamp = int(time.time()) + 60 + joining.inverters[1].energy.pv_today_wh = 7000 + gw._process_telemetry(joining.SerializeToString()) + assert gw._inverter_slot_serials == ["CE2223G800", "CE2225G400"] + assert gw._integrated_energy["pv"]["CE2225G400"]["base"] == 7000 + + def test_reserve_never_reported_is_published_as_zero(self): + """An inverter that has never reported a reserve still publishes 0, as before.""" + gw = self._listed_gateway([]) + blank = self._rated_status() + blank.inverters[1].control.reserve_soc = 0 + gw._process_telemetry(blank.SerializeToString()) + assert gw._states["number.predbat_gateway_25g400_reserve_soc"] == 0 + + def test_listed_inverters_all_healthy_report_no_error(self): + """A healthy site with a serial list raises nothing.""" + gw = self._listed_gateway(["CE2223G800", "CE2225G400"]) + + gw._process_telemetry(self._two_ac_status("ok").SerializeToString()) + + assert gw._args["num_inverters"] == 2 + assert self._errors(gw) == [] + gw.non_fatal_error_occurred.assert_not_called() + + def test_serial_list_is_not_added_to(self): + """An inverter outside gateway_inverter_serial is never bound, healthy or not.""" + gw = self._listed_gateway(["CE2223G800"]) + + gw._process_telemetry(self._two_ac_status("no_battery").SerializeToString()) + gw._process_telemetry(self._two_ac_status("ok").SerializeToString()) + + assert gw._args["num_inverters"] == 1 + assert gw._inverter_slot_serials == ["CE2223G800"] + gw.non_fatal_error_occurred.assert_not_called() + + def test_serial_list_with_a_coordinator_still_binds_only_the_coordinator(self): + """A list naming a Gateway, with or without its AIOs, binds the Gateway alone: the list never splits a coordinated system.""" + for listed in (["GW2347G077", "CH2414G318", "CH9999G999"], ["GW2347G077"]): + gw = self._listed_gateway(listed) + + gw._process_telemetry(self._gateway_plus_aios_status(["CH2414G318", "CH9999G999"]).SerializeToString()) + + assert gw._args["num_inverters"] == 1, listed + assert gw._args["charge_start_time"] == ["select.predbat_gateway_47g077_charge_slot1_start"], listed + assert gw._pv_only_serials == (), listed + gw.non_fatal_error_occurred.assert_not_called() + + def _pv_and_battery_status(self, pv_power=1200, pv_has_battery=False): + """A PV-only inverter that reports generation, plus one battery inverter.""" + status = self._pv_only_site_status(battery_ready=True) + status.inverters[0].pv.power_w = pv_power + if pv_has_battery: + status.inverters[0].battery.soc_percent = 40 + status.inverters[0].battery.capacity_wh = 5000 + return status + + def test_listed_pv_only_inverter_is_configured_as_pv_only(self): + """A listed inverter with no battery has no slot and is never a write target; its generation is counted.""" + gw = self._listed_gateway(["FD0000P001", "CH0000B002"]) + + for power in (1200, 1500): + gw._process_telemetry(self._pv_and_battery_status(pv_power=power).SerializeToString()) + assert gw._states["sensor.predbat_gateway_00p001_pv_power"] == power + + assert gw._args["num_inverters"] == 1 + assert gw._inverter_slot_serials == ["CH0000B002"] + assert gw._pv_only_serials == ("FD0000P001",) + assert "00p001" not in gw._suffix_to_serial + assert gw._args["pv_power"] == ["sensor.predbat_gateway_00b002_pv_power", "sensor.predbat_gateway_00p001_pv_power"] + assert gw._args["pv_today"] == ["sensor.predbat_gateway_00b002_pv_today", "sensor.predbat_gateway_00p001_pv_today"] + for arg in ("soc_percent", "charge_rate", "discharge_rate", "reserve", "charge_limit", "charge_start_time", "scheduled_charge_enable", "load_power", "grid_power"): + assert len(gw._args[arg]) == 1 and "00p001" not in str(gw._args[arg][0]), arg + assert self._errors(gw) == [] + gw.non_fatal_error_occurred.assert_not_called() + + def test_unlisted_pv_only_inverter_is_left_out_as_before(self): + """With no serial list a PV-only inverter is not configured at all, as before.""" + gw = self._listed_gateway([]) + + gw._process_telemetry(self._pv_and_battery_status().SerializeToString()) + + assert gw._args["num_inverters"] == 1 + assert gw._pv_only_serials == () + assert gw._args["pv_power"] == ["sensor.predbat_gateway_00b002_pv_power"] + + def test_pv_only_inverter_that_reports_a_battery_later_gets_a_slot(self): + """Having a battery is learned when the hub reports it, and re-runs auto-config; it is never unlearned.""" + gw = self._listed_gateway(["FD0000P001", "CH0000B002"]) + gw._process_telemetry(self._pv_and_battery_status().SerializeToString()) + assert gw._args["num_inverters"] == 1 + + gw._process_telemetry(self._pv_and_battery_status(pv_has_battery=True).SerializeToString()) + assert gw._args["num_inverters"] == 2 + assert gw._pv_only_serials == () + assert gw._discovery["FD0000P001"]["battery"] is True + + gw._process_telemetry(self._pv_and_battery_status().SerializeToString()) + assert gw._args["num_inverters"] == 2 + assert gw._discovery["FD0000P001"]["battery"] is True + + def test_unlisted_site_picks_up_an_inverter_that_becomes_usable(self): + """With no serial list, an inverter unusable at first telemetry is registered when it becomes usable, with no restart.""" + for second in ("no_battery", "disconnected", "not_primary"): + gw = self._make_handler_gateway() + + gw._process_telemetry(self._two_ac_status(second).SerializeToString()) + assert gw._args["num_inverters"] == 1, second + assert gw._inverter_slot_serials == ["CE2223G800"], second + + gw._process_telemetry(self._two_ac_status("ok").SerializeToString()) + assert gw._args["num_inverters"] == 2, second + assert gw._inverter_slot_serials == ["CE2223G800", "CE2225G400"], second + + def test_unusable_inverter_does_not_rerun_auto_config_every_frame(self): + """An inverter that stays unusable is not a reason to re-run auto-config, and nor is a steady healthy site.""" + for listed in ([], ["CE2223G800", "CE2225G400"]): + gw = self._listed_gateway(listed) + gw._process_telemetry(self._two_ac_status("no_battery").SerializeToString()) + + runs = [] + real_config = gw.automatic_config + + def counted_config(): + runs.append(1) + return real_config() + + gw.automatic_config = counted_config + for _ in range(3): + gw._process_telemetry(self._two_ac_status("no_battery").SerializeToString()) + assert runs == [], listed + + for _ in range(3): + gw._process_telemetry(self._two_ac_status("ok").SerializeToString()) + assert runs == [1], listed + + # Losing its telemetry again is a removal, which stays ignored: the inverter keeps its slot + gw._process_telemetry(self._two_ac_status("no_battery").SerializeToString()) + assert runs == [1], listed + assert gw._args["num_inverters"] == 2, listed + def test_scenario_second_aio_via_telemetry_moves_control_to_gateway(self): """End-to-end through the telemetry handler (_process_telemetry). @@ -6097,6 +7016,135 @@ def _feed(self, gw, status): def _today(self, gw, name, suffix="23g800"): return gw._dashboard_calls[f"sensor.predbat_gateway_{suffix}_{name}"][0] + def test_midnight_still_resets_the_day_while_an_inverter_is_away(self): + """An inverter kept configured while it is away stops the integration, but not the day ending.""" + gw = self._make_gateway() + self._feed(gw, self._status(self.BASE_TIME, grid_w=-2000)) + self._feed(gw, self._status(self.BASE_TIME + 600, grid_w=-3000)) + assert approx_equal(gw._integrated_energy["import_wh"], 416.67) + + # 16:10 + 8.5h = 00:40 the next day, with the second inverter gone from the status + away = self._status(self.BASE_TIME + 600 + 8 * 3600 + 1800, grid_w=-2000, battery_w=(0,), serials=("CE2223G800",)) + self._feed(gw, away) + + assert gw._inverter_slot_serials == ["CE2223G800", "CE2225G400"] + assert gw._integrated_energy["import_wh"] == 0.0 + assert gw._integrated_energy["load_published_wh"] == 0.0 + assert self._today(gw, "import_today") == 0.0 + assert self._today(gw, "load_today") == 0.0 + + def test_pv_only_inverter_generation_counts_towards_the_load(self): + """A PV-only inverter has no slot, but what it generates is used by the house like any other PV.""" + gw = self._make_gateway() + gw.gateway_inverter_serial = ["CE2223G800", "FD0000P001"] + + def status(timestamp, pv_only_wh, with_pv_only=True): + built = self._status(timestamp, grid_w=0, battery_w=(0,), pv_today_wh=1000, serials=("CE2223G800",)) + if with_pv_only: + inv = built.inverters.add() + inv.type = pb.INVERTER_TYPE_GIVENERGY + inv.serial = "FD0000P001" + inv.primary = True + inv.connected = True + inv.pv.power_w = 1000 + inv.energy.pv_today_wh = pv_only_wh + return built + + self._feed(gw, status(self.BASE_TIME, 3000)) + assert gw._inverter_slot_serials == ["CE2223G800"] + assert gw._pv_only_serials == ("FD0000P001",) + + self._feed(gw, status(self.BASE_TIME + 600, 3200)) + assert approx_equal(gw._integrated_energy["load_wh"], 200.0) + + # Away for a status: nothing added, and the integration carries on for the rest of the site + self._feed(gw, status(self.BASE_TIME + 1200, 0, with_pv_only=False)) + assert approx_equal(gw._integrated_energy["load_wh"], 200.0) + assert gw._integrated_energy["time"] == self.BASE_TIME + 1200 + + self._feed(gw, status(self.BASE_TIME + 1800, 3500)) + assert approx_equal(gw._integrated_energy["load_wh"], 500.0) + + def test_pv_only_inverter_promoted_to_a_slot_keeps_its_generation(self): + """Auto-config re-running and changing the slots does not lose or mix up any inverter's generation.""" + gw = self._make_gateway() + # The PV-only inverter sorts first, so once it has a slot every slot index moves + gw.gateway_inverter_serial = ["AA0000P001", "CE2223G800"] + + def status(timestamp, pv_only_wh, has_battery=False): + built = self._status(timestamp, grid_w=0, battery_w=(0,), pv_today_wh=1000, serials=("CE2223G800",)) + inv = built.inverters.add() + inv.type = pb.INVERTER_TYPE_GIVENERGY + inv.serial = "AA0000P001" + inv.primary = True + inv.connected = True + inv.pv.power_w = 1000 + inv.energy.pv_today_wh = pv_only_wh + if has_battery: + inv.battery.soc_percent = 40 + inv.battery.capacity_wh = 5000 + return built + + self._feed(gw, status(self.BASE_TIME, 3000)) + self._feed(gw, status(self.BASE_TIME + 600, 3200)) + assert gw._pv_only_serials == ("AA0000P001",) + assert approx_equal(gw._integrated_energy["load_wh"], 200.0) + + # Its battery readings arrive: _process_telemetry integrates, then re-runs auto-config + self._feed(gw, status(self.BASE_TIME + 1200, 3300, has_battery=True)) + gw.automatic_config() + assert gw._inverter_slot_serials == ["AA0000P001", "CE2223G800"] + assert gw._pv_only_serials == () + assert approx_equal(gw._integrated_energy["load_wh"], 300.0) + + self._feed(gw, status(self.BASE_TIME + 1800, 3500, has_battery=True)) + assert approx_equal(gw._integrated_energy["load_wh"], 500.0) + + def test_rebase_takes_the_new_fleet_power_and_adds_nothing(self): + """A rebase on a status already integrated adds no energy and starts the next interval from the new slots' power.""" + gw = self._make_gateway(shared_ct=False) + self._feed(gw, self._status(self.BASE_TIME, grid_w=-1000, battery_w=(0,), serials=("CE2225G400",))) + joining = self._status(self.BASE_TIME + 600, grid_w=-1000, battery_w=(0, 0)) + gw._last_status = joining + gw._update_integrated_energy(joining) + before = dict((name, gw._integrated_energy[name]) for name in ("import_wh", "export_wh", "load_wh")) + assert gw._integrated_energy["power"]["import"] == 1000 + + gw.automatic_config() + gw._update_integrated_energy(joining, rebase=True) + + assert dict((name, gw._integrated_energy[name]) for name in before) == before + assert gw._integrated_energy["power"]["import"] == 2000 + assert gw._integrated_energy["time"] == self.BASE_TIME + 600 + + def test_inverter_given_a_slot_later_has_its_generation_counted_from_then(self): + """An inverter that joins the slots after the integration began is counted from its first status, not left out.""" + gw = self._make_gateway() + self._feed(gw, self._status(self.BASE_TIME, grid_w=0, battery_w=(0,), pv_today_wh=1000, serials=("CE2225G400",))) + self._feed(gw, self._status(self.BASE_TIME + 600, grid_w=0, battery_w=(0,), pv_today_wh=1100, serials=("CE2225G400",))) + assert approx_equal(gw._integrated_energy["load_wh"], 100.0) + + def both(timestamp, first_wh, second_wh): + built = self._status(timestamp, grid_w=0, battery_w=(0, 0), pv_today_wh=first_wh) + built.inverters[1].energy.pv_today_wh = second_wh + return built + + # CE2223G800 appears and takes slot 0; its counter already stands at 7000. As in + # _process_telemetry, the status is integrated on the old slots, auto-config re-runs, + # and the status then becomes the baseline for the new ones + joining = both(self.BASE_TIME + 1200, 7000, 1200) + gw._last_status = joining + gw._update_integrated_energy(joining) + gw.automatic_config() + assert gw._inverter_slot_serials == ["CE2223G800", "CE2225G400"] + gw._update_integrated_energy(joining, rebase=True) + assert approx_equal(gw._integrated_energy["load_wh"], 200.0) + assert gw._integrated_energy["pv"]["CE2223G800"]["base"] == 7000 + + # Its generation since the status it joined on is counted, not only from the one after + self._feed(gw, both(self.BASE_TIME + 1800, 7300, 1250)) + assert approx_equal(gw._integrated_energy["load_wh"], 550.0) + def test_import_and_load_are_integrated_from_power(self): """6 kW of import for 6 minutes is 0.6 kWh; 3 kW of that went into a battery, so load is 0.3 kWh.""" gw = self._make_gateway()