"""Battery-fleet aggregation over the Battery Notes integration. The user does NOT want one maintenance task per battery (30-70+ devices would bury the task list). Instead this aggregates every Battery Notes ``battery_plus`` sensor into ONE fleet view: which batteries are low now, grouped by battery type (so you know *what to buy*), plus a simple deterministic forecast of what will be needed soon (so you can order in time). Battery Notes exposes everything we need as ATTRIBUTES on its entities (device_class ``battery``): ``battery_type``, ``battery_quantity``, ``battery_low``, ``battery_low_threshold``, ``battery_last_replaced``. The percentage sensor is the primary source; LOW-ONLY sources (a Matter lock with just a battery-low binary, #121) are read from their ``…_battery_plus_low`` binary instead. Forecast (#114 + follow-up): the ~replacement date comes from the DISCHARGE TREND where recorder data supports it (``async_trend_predictions`` — the SensorPredictor regression asking "when does the level fall below the low threshold?", medium/high confidence only, cached 6 h) and falls back to ``battery_last_replaced`` + the type-lifetime table everywhere else. The pure builder ``build_overview`` takes plain battery dicts + an injected ``today`` so the forecast is unit-testable with synthetic dates; ``read_batteries`` is the thin HA-reading adapter. """ from __future__ import annotations import logging import re from collections import OrderedDict from dataclasses import dataclass, field from datetime import date, timedelta from typing import Any from homeassistant.core import HomeAssistant from homeassistant.util import dt as dt_util _LOGGER = logging.getLogger(__name__) # Editorial typical service life per battery type, in MONTHS — the forecast # anchor (battery_last_replaced + lifetime = predicted replacement). These are # deliberately conservative sensor-use estimates and are meant to be tunable; # unknown types fall back to DEFAULT_LIFETIME_MONTHS. TYPICAL_LIFETIME_MONTHS: dict[str, int] = { "AAAA": 10, "AAA": 10, "AA": 12, "C": 18, "D": 24, "9V": 12, "CR2": 18, "CR123A": 18, "CR2032": 18, "CR2450": 24, "CR2477": 24, "CR2016": 18, "CR2025": 18, } DEFAULT_LIFETIME_MONTHS = 12 # How far ahead "needed soon" looks by default (days). DEFAULT_HORIZON_DAYS = 28 # A native battery %-sensor without a dedicated low binary is treated as low # at or below this level (editorial — HA has no universal low threshold). NATIVE_LOW_PERCENT = 20 # States that mean "no reading" — a removed device leaves nothing; an offline # one leaves these. We keep an OFFLINE battery visible only when its last-known # low flag says it needs attention (a dead battery often takes its device # offline — that's exactly the one you must not hide). _NO_READING = {"unavailable", "unknown", "none", ""} # How long a NATIVE battery that was last seen LOW stays in the fleet after # its entity goes unavailable. Battery Notes covers this case via its retained # ``battery_low`` attribute; native entities have no equivalent, so without a # snapshot the battery would vanish at the exact moment it died and took its # device offline. Bounded so a permanently removed device eventually drops. _NATIVE_RETENTION = timedelta(hours=48) # Heuristic (sensors WITHOUT device_class): a %-sensor whose object_id talks # about a battery — some Zigbee2MQTT/ESPHome devices ship battery levels # without the device class. Deliberately strict: the exclusion words keep out # charging electronics and home-storage state-of-charge sensors (a Powerwall # is not a battery you replace). _HEURISTIC_EXCLUDE = ("charging", "current", "power", "voltage", "energy", "load", "soc", "state_of_charge", "storage", "temp") def _is_native_battery_sensor(state: Any) -> bool: """Whether a sensor state looks like a replaceable-battery level.""" attrs = state.attributes if attrs.get("device_class") == "battery": return True if attrs.get("unit_of_measurement") != "%": return False object_id = state.entity_id.split(".", 1)[1] if "battery" not in object_id: return False return not any(word in object_id for word in _HEURISTIC_EXCLUDE) def _native_snapshot_cache(hass: HomeAssistant) -> dict[str, dict[str, Any]]: """Runtime cache of last-known native battery readings (per entity_id).""" from ..const import DOMAIN cache: dict[str, dict[str, Any]] = hass.data.setdefault(DOMAIN, {}).setdefault("battery_fleet_native_cache", {}) return cache def _norm_type(raw: Any) -> str: """Canonicalize a battery-type label for grouping (upper, trimmed).""" s = str(raw or "").strip() return s.upper() if s else "UNKNOWN" # Battery Notes' library labels rechargeable packs with type strings like # "Rechargeable", "Nuki Battery Pack" or li-ion cell names. Such a battery is # CHARGED, never bought — so it must not enter the shopping groupings, and the # type-lifetime table (a primary-cell prior) has nothing honest to say about # it. Low tracking and the discharge-trend forecast stay: "charge the lock in # ~20 days" is exactly what the roster is for. _RECHARGEABLE_TYPE_RE = re.compile( r"rechargeable|akku|accu|li[- ]?ion|li[- ]?po|lifepo|ni[- ]?mh|nicd|18650|21700|" r"power ?pack|battery ?pack|built[- ]?in", re.IGNORECASE, ) def is_rechargeable_type(battery_type: Any) -> bool: """Whether a battery-type label describes a rechargeable pack/cell.""" return bool(_RECHARGEABLE_TYPE_RE.search(str(battery_type or ""))) def lifetime_months(battery_type: str) -> int: """Typical service life for a (canonicalized) battery type.""" return TYPICAL_LIFETIME_MONTHS.get(_norm_type(battery_type), DEFAULT_LIFETIME_MONTHS) @dataclass class Battery: """One battery-powered device — from Battery Notes (rich) or a native ``device_class: battery`` entity (degraded: no type/quantity/forecast).""" entity_id: str device_name: str battery_type: str quantity: int low: bool level: float | None last_replaced: date | None available: bool = True source: str = "battery_notes" # The level at which THIS battery counts low: Battery Notes' configured # threshold or the fleet-wide floor, whichever is higher (the one that # crosses first on the way down). One field feeds the trend regression, # the sparkline threshold line and the level-bar colors alike. low_threshold: float = float(NATIVE_LOW_PERCENT) @dataclass class BatteryOverview: """The aggregated fleet view backing the single fleet task + its detail.""" total: int = 0 low: list[dict[str, Any]] = field(default_factory=list) soon: list[dict[str, Any]] = field(default_factory=list) # EVERY tracked battery, each tagged low / soon / ok, device-name sorted. # # low and soon answer "what needs doing"; this answers "what is the fleet # watching". Without it a device that is perfectly healthy appears nowhere # in the payload, so it could only be excluded once it had already gone low # — which is exactly when it is being noisy, and after the fleet task may # have auto-completed and dropped it from the list again (discussion #113). all: list[dict[str, Any]] = field(default_factory=list) # Grouped quantities by canonical type. needs_now: OrderedDict[str, int] = field(default_factory=OrderedDict) needs_soon: OrderedDict[str, int] = field(default_factory=OrderedDict) types: list[str] = field(default_factory=list) @property def low_count(self) -> int: return len(self.low) def _predicted_date(bat: Battery) -> date | None: if bat.last_replaced is None: return None months = lifetime_months(bat.battery_type) # Month arithmetic without dateutil: add whole months, clamp the day. y, m = bat.last_replaced.year, bat.last_replaced.month + months y += (m - 1) // 12 m = (m - 1) % 12 + 1 day = min(bat.last_replaced.day, 28) return date(y, m, day) def build_overview( batteries: list[Battery], *, today: date, horizon_days: int = DEFAULT_HORIZON_DAYS, trend_predictions: dict[str, tuple[int, str]] | None = None, ) -> BatteryOverview: """Aggregate batteries into the fleet view. * ``low`` = reported low right now (Battery Notes' own threshold). * ``soon`` = NOT low yet but predicted to reach end-of-life within ``horizon_days`` (deterministic last_replaced + typical-lifetime forecast). A battery already low is never double-counted into soon. * ``needs_now`` / ``needs_soon`` = summed quantities per type — the shopping grouping ("2× AA, 4× AAA"). Rechargeable types never enter it: a low rechargeable means "charge it", not "buy one". * ``all`` = every battery with its status, so a healthy device can be excluded BEFORE it ever becomes noisy. """ ov = BatteryOverview(total=len(batteries)) types_seen: OrderedDict[str, None] = OrderedDict() for bat in sorted(batteries, key=lambda b: b.device_name.lower()): t = _norm_type(bat.battery_type) types_seen[t] = None rechargeable = is_rechargeable_type(bat.battery_type) # Blend (#114 follow-up): the DISCHARGE TREND wins where the recorder # data supports it (medium/high confidence, filtered upstream) — it is # device-specific and usage-aware; the type's typical lifetime is the # prior everything else falls back to. For rechargeables the table is # no prior at all (its lifetimes describe primary cells, and Battery # Notes seeds last_replaced at note creation — a real fleet showed # "replace the vacuum's pack" dated from the day the device was added), # so they get a ~date only when the trend has earned one. trend = (trend_predictions or {}).get(bat.entity_id) if trend is not None: days_raw: int | None = trend[0] source, confidence = "trend", trend[1] else: pred = None if rechargeable else _predicted_date(bat) days_raw = (pred - today).days if pred is not None else None source, confidence = "typical", None # B1 (decided 2026-08): a PASSED prediction while the battery still # reports healthy is "forecast overdue" — it stays in `soon` (clamped # to 0 days, no negative countdowns) and NEVER escalates into `low` # or the task trigger: the forecast has error bars, the sensor says # fine, and the usual cause is a swap that was never recorded (the # roster's unrecorded-swap hint covers exactly that). Reality fires # the task; predictions only shop. overdue = days_raw is not None and days_raw < 0 days = max(0, days_raw) if days_raw is not None else None if bat.low: ov.low.append(_row(bat, t, None, rechargeable=rechargeable)) if not rechargeable: ov.needs_now[t] = ov.needs_now.get(t, 0) + bat.quantity # A battery reported low has no meaningful forecast left to show. ov.all.append({**_row(bat, t, None, rechargeable=rechargeable), "status": "low"}) continue if days is not None and days <= horizon_days: ov.soon.append(_row(bat, t, days, source, confidence, rechargeable=rechargeable, forecast_overdue=overdue)) if not rechargeable: ov.needs_soon[t] = ov.needs_soon.get(t, 0) + bat.quantity ov.all.append({**_row(bat, t, days, source, confidence, rechargeable=rechargeable, forecast_overdue=overdue), "status": "soon"}) continue ov.all.append({**_row(bat, t, days, source, confidence, rechargeable=rechargeable), "status": "ok"}) ov.soon.sort(key=lambda r: r["days_until"] if r["days_until"] is not None else 1 << 30) ov.types = sorted(types_seen) ov.needs_now = OrderedDict(sorted(ov.needs_now.items())) ov.needs_soon = OrderedDict(sorted(ov.needs_soon.items())) return ov def _row( bat: Battery, canon_type: str, days_until: int | None, predicted_source: str = "typical", prediction_confidence: str | None = None, *, rechargeable: bool = False, forecast_overdue: bool = False, ) -> dict[str, Any]: return { "entity_id": bat.entity_id, "device_name": bat.device_name, "battery_type": canon_type, "quantity": bat.quantity, "level": bat.level, "days_until": days_until, "available": bat.available, # #114 follow-up: where the ~date comes from — "trend" (discharge # regression, with confidence) or "typical" (type-lifetime table). "predicted_source": predicted_source, "prediction_confidence": prediction_confidence, # Charged, never bought: low means "recharge", and the row never # contributes to the shopping groupings. "rechargeable": rechargeable, # This battery's own low threshold — the level bars color against it. "low_threshold": bat.low_threshold, # B1: the predicted date has PASSED while the battery still reports # healthy. Deliberately NOT low and NOT the task trigger — a forecast # carries error bars and the sensor says fine — but the roster shows # the discrepancy (common cause: a swap that was never recorded). "forecast_overdue": forecast_overdue, } def _parse_last_replaced(raw: Any) -> date | None: if not raw: return None try: parsed = dt_util.parse_datetime(str(raw)) if parsed is not None: return parsed.date() return date.fromisoformat(str(raw)[:10]) except (ValueError, TypeError): return None def _level_of(state_val: str) -> float | None: try: return float(state_val) except (ValueError, TypeError): return None def fleet_excluded_entities(hass: HomeAssistant) -> set[str]: """Manually excluded battery entity_ids, stored on the fleet object entry. Inlined lookup (not via battery_fleet_setup.find_fleet_entry) to keep this module import-cycle-free — setup imports the aggregation, not vice versa. """ from ..const import CONF_OBJECT, DOMAIN for entry in hass.config_entries.async_entries(DOMAIN): obj = entry.data.get(CONF_OBJECT, {}) if obj.get("battery_fleet"): return set(obj.get("battery_fleet_excluded") or []) return set() def _is_self_charging(hass: HomeAssistant, device_id: str | None) -> bool: """Whether a device recharges itself — its battery is never REPLACED. Issue #107: a Roborock's native battery sensor reads "low" mid-clean, but nobody swaps its cells. Heuristics: the device also has a vacuum/lawn_mower entity, exposes a ``battery_charging`` binary, or is a Companion-app phone/tablet (``mobile_app`` identifiers). Applied to BOTH passes. This originally spared Battery Notes entries on the theory that an explicit note is deliberate intent — but Battery Notes auto-discovery proposes notes for vacuums straight from its library (type "Rechargeable"), so a real fleet ended up telling its owner to buy a "RECHARGEABLE" for the vacuum. """ if not device_id: return False from homeassistant.helpers import device_registry as dr from homeassistant.helpers import entity_registry as er device = dr.async_get(hass).async_get(device_id) if device and any(domain == "mobile_app" for domain, _ in device.identifiers): return True for reg_entry in er.async_entries_for_device(er.async_get(hass), device_id, include_disabled_entities=True): if reg_entry.domain in ("vacuum", "lawn_mower"): return True if ( reg_entry.domain == "binary_sensor" and (reg_entry.device_class or reg_entry.original_device_class) == "battery_charging" ): return True return False def read_batteries(hass: HomeAssistant) -> list[Battery]: """Read the battery fleet from HA state — Battery Notes AND native. * **Battery Notes** ``battery_plus`` sensors (device_class ``battery`` + a ``battery_type`` attribute) give the rich view: type, quantity, low, last-replaced. When the source goes offline the sensor reads unavailable/unknown but RETAINS its last-known ``battery_low`` — so a dead battery that took its device offline stays visible. A device whose source reports no percentage at all (a Matter lock with only a battery-low binary, #121) gets NO percentage sensor from Battery Notes — its metadata lives solely on the ``…_battery_plus_low`` BINARY, so a second sweep picks those up for devices the sensor sweep did not cover. Devices with BOTH stay one row (the binary carries the same attributes and would otherwise duplicate every battery and dodge exclusions). Self-charging devices (vacuums, mowers, phones — see :func:`_is_self_charging`) are skipped here too: Battery Notes auto-discovery notes them from its library, so a note is no proof of intent to track a replaceable cell. * **Native** ``device_class: battery`` entities (a %-sensor and/or a battery-low binary) — plus %-sensors matching the strict battery-name heuristic for devices that ship no device class — grouped per device, give a degraded view (type "Unknown", quantity 1, no forecast). A device already covered by a Battery Notes note is skipped (dedup by the note's source entity + its device) so it isn't counted twice; self-charging devices (vacuums, mowers, phones — see :func:`_is_self_charging`) are skipped entirely. A native battery last seen LOW that goes unavailable is retained from a runtime snapshot for ``_NATIVE_RETENTION`` (the Battery Notes path gets this for free via its retained ``battery_low`` attribute). * Manually excluded entity_ids (fleet detail → exclude) are dropped from BOTH passes. """ from homeassistant.helpers import device_registry as dr from homeassistant.helpers import entity_registry as er ent_reg = er.async_get(hass) dev_reg = dr.async_get(hass) excluded = fleet_excluded_entities(hass) out: list[Battery] = [] covered_sources: set[str] = set() covered_devices: set[str] = set() # ── Pass 1: Battery Notes battery_plus ────────────────────────────────── # Percentage SENSORS first, then LOW-ONLY BINARIES (#121): a source with # no percentage (a Matter lock's plain battery-low binary) gets no # ``battery_plus`` sensor from Battery Notes, so the type/quantity/ # last-replaced metadata exists only on the ``…_battery_plus_low`` binary. # The binary sweep is restricted to devices the sensor sweep did NOT # cover: a percentage note's own low binary carries the SAME attributes, # and taking it too would put every battery in the roster twice — and let # an exclusion set on the sensor row resurrect through the binary. note_sensor_ids: set[str] = set() for domain, binary_pass in (("sensor", False), ("binary_sensor", True)): for state in hass.states.async_all(domain): attrs = state.attributes if attrs.get("device_class") != "battery" or "battery_type" not in attrs: continue if not binary_pass: # EVERY matching percentage note counts as sibling coverage — # kept, dropped or excluded: its low binary describes the same # battery and must never become a second (or resurrected) row. note_sensor_ids.add(state.entity_id) reg = ent_reg.async_get(state.entity_id) dev_id = reg.device_id if reg else None if binary_pass: if dev_id and dev_id in covered_devices: continue # Registry-based dedupe is not enough on its own (caught live: # state-only entities have no registry entry, and every fleet # battery doubled). Two fallbacks: the shared source entity, # and Battery Notes' naming contract — # ``sensor.X_battery_plus`` ↔ ``binary_sensor.X_battery_plus_low``. src_attr = attrs.get("source_entity_id") if src_attr and src_attr in covered_sources: continue object_id = state.entity_id.split(".", 1)[1] if object_id.endswith("_low") and f"sensor.{object_id[: -len('_low')]}" in note_sensor_ids: continue # No percentage to read — the binary state IS the low signal. level = None available = state.state not in _NO_READING low = bool(attrs.get("battery_low")) or str(state.state).lower() == "on" else: level = _level_of(state.state) available = state.state not in _NO_READING and level is not None # B2 (roadmap 2026-07-22 audit): ONE low floor across both # passes. Battery Notes' own threshold (default 10 %) still # counts via its battery_low flag, but the fleet-wide # NATIVE_LOW_PERCENT floor is OR-ed in — a CR2032 at 11.5 % # was "healthy" here while the same level counted low in the # native pass. A HIGHER Battery Notes threshold (e.g. 30 %) # still wins through battery_low. low = bool(attrs.get("battery_low")) or (level is not None and level <= NATIVE_LOW_PERCENT) last_replaced = _parse_last_replaced(attrs.get("battery_last_replaced")) # B1 (roadmap 2026-07-22 audit): a forecast-only note — no level # sensor, so the state reads unknown forever — must SURVIVE when it # carries a replacement date: that date is all `_predicted_date` # needs, and dropping these hid 11 overdue batteries in a live fleet. # Offline AND not low AND no date = pure connectivity noise → drop. if not available and not low and last_replaced is None: continue # B3: only a KEPT note covers its source/device — a dropped dead note # must not suppress the native fallback for its own device (a device # with a dead note and a working level sensor was invisible in BOTH # passes). src = attrs.get("source_entity_id") if src: covered_sources.add(src) if dev_id: covered_devices.add(dev_id) # An EXCLUDED note still covers (above): exclusion hides the battery — # it must not resurrect as a degraded native "Unknown" row. if state.entity_id in excluded: continue # #107 follow-up: the skip covers noted devices too (it covers # above for the same reason exclusion does). Battery Notes # auto-discovers vacuums/phones from its library, so a note is # not evidence anyone means to swap cells there. if _is_self_charging(hass, dev_id): continue out.append( Battery( entity_id=state.entity_id, device_name=attrs.get("device_name") or attrs.get("friendly_name") or state.entity_id, battery_type=str(attrs.get("battery_type") or "Unknown"), quantity=int(attrs.get("battery_quantity") or 1), low=low, level=level, last_replaced=last_replaced, available=available, source="battery_notes", low_threshold=_note_low_threshold(attrs), ) ) # ── Pass 2: native battery entities, grouped per device ───────────────── # {group_key: {"level_state": s, "low_state": s, "name": ..., "device_id": ..., "eid": ...}} native: dict[str, dict[str, Any]] = {} for domain in ("sensor", "binary_sensor"): for state in hass.states.async_all(domain): # Sensors: device_class battery OR the strict name/% heuristic # (Zigbee2MQTT/ESPHome levels without a device class). Binaries: # device_class only — name-guessing booleans is too risky. if domain == "sensor": if not _is_native_battery_sensor(state): continue elif state.attributes.get("device_class") != "battery": continue eid = state.entity_id if "battery_type" in state.attributes: # Battery Notes battery_plus — handled above continue if eid in covered_sources or eid in excluded: continue reg = ent_reg.async_get(eid) dev_id = reg.device_id if reg else None if dev_id and dev_id in covered_devices: continue if _is_self_charging(hass, dev_id): # #107: vacuums/mowers/phones continue key = dev_id or eid rec = native.setdefault( key, {"level_state": None, "low_state": None, "device_id": dev_id, "eid": eid, "name": None}, ) friendly = state.attributes.get("friendly_name") if domain == "sensor": rec["level_state"] = state.state rec["eid"] = eid else: rec["low_state"] = state.state if rec["name"] is None and friendly: rec["name"] = friendly snapshot_cache = _native_snapshot_cache(hass) now = dt_util.utcnow() for rec in native.values(): level = _level_of(rec["level_state"]) if rec["level_state"] is not None else None low_state = rec["low_state"] level_available = rec["level_state"] not in _NO_READING if rec["level_state"] is not None else False low_available = low_state not in _NO_READING if low_state is not None else False available = level_available or low_available if low_state is not None: low = low_available and str(low_state).lower() in ("on", "true", "1") else: low = level is not None and level <= NATIVE_LOW_PERCENT if available: # Remember the last real reading — the retention path below needs # it once the entity goes unavailable. snapshot_cache[rec["eid"]] = {"low": low, "level": level, "ts": now} elif not low: # Native dead-battery retention: an entity that was LOW and then # went unavailable (the battery died and took the device offline) # stays visible for _NATIVE_RETENTION instead of vanishing at the # exact moment it needs replacing. snap = snapshot_cache.get(rec["eid"]) if snap and snap.get("low") and now - snap["ts"] <= _NATIVE_RETENTION: low = True level = snap.get("level") if not available and not low: continue name = rec["name"] if not name and rec["device_id"] and (dev := dev_reg.async_get(rec["device_id"])): name = dev.name_by_user or dev.name out.append( Battery( entity_id=rec["eid"], device_name=name or rec["eid"], battery_type="Unknown", quantity=1, low=low, level=level, last_replaced=None, available=available, source="native", ) ) return out def has_battery_notes(hass: HomeAssistant) -> bool: """Whether the Battery Notes integration is present (any battery_plus). Binaries count too (#121): an install whose only noted devices are low-only sources has no ``battery_plus`` sensor at all. """ for domain in ("sensor", "binary_sensor"): for state in hass.states.async_all(domain): a = state.attributes if a.get("device_class") == "battery" and "battery_type" in a: return True return False def has_batteries(hass: HomeAssistant) -> bool: """Whether ANY battery is trackable — Battery Notes OR native. Gates setup.""" if any(_is_native_battery_sensor(s) for s in hass.states.async_all("sensor")): return True return any(s.attributes.get("device_class") == "battery" for s in hass.states.async_all("binary_sensor")) def compute_overview(hass: HomeAssistant, *, horizon_days: int = DEFAULT_HORIZON_DAYS) -> BatteryOverview: """Read + aggregate in one call (SYNC entry point — table forecast only). The summary sensors call this from their update path; recorder-backed trend regression stays out of it deliberately. The panel goes through :func:`async_compute_overview` instead. """ today = dt_util.now().date() return build_overview(read_batteries(hass), today=today, horizon_days=horizon_days) # ── discharge-trend forecast (#114 follow-up) ─────────────────────────────── _TREND_CACHE_KEY = "maintenance_supporter_battery_trend_cache" _TREND_CACHE_TTL = timedelta(hours=6) _TREND_MIN_CONFIDENCE = ("medium", "high") # Beyond this the regression extrapolates >12x its 30 d observation window — # a real prod evaluation produced "empty in 1142 d" at medium confidence for a # barely-draining motion sensor, where the type table is the honest answer. _TREND_MAX_DAYS = 365 # Reject a series whose level ROSE by more than this (percent points) after a # minimum inside the window: real discharges are monotone-ish, big recoveries # mean the percentage tracks something else (cold-dip voltage bounce on a # CR2032 is the classic). Small relaxation bounces (+3-4 %, seen on a real # LYWSD03MMC) stay below it. _TREND_MAX_RECOVERY_PCT = 10.0 async def async_trend_predictions(hass: HomeAssistant, batteries: list[Battery]) -> dict[str, tuple[int, str]]: """Per-battery discharge-trend forecast: {entity_id: (days_until, confidence)}. Reuses the SensorPredictor's recorder regression, asking "when does this level sensor fall below its low threshold?". Only batteries with a live percentage reading are analysed (low-only binaries have no level to regress); low-confidence, non-falling, and far-out trends (beyond ``_TREND_MAX_DAYS``) are dropped so the caller can fall back to the type-lifetime table. Cached for 6 h per entity (misses included) — batteries drain over weeks, and the overview is fetched on every panel visit; 30+ recorder regressions per click would be waste. """ from .sensor_predictor import SensorPredictor cache: dict[str, tuple[Any, tuple[int, str] | None]] = hass.data.setdefault(_TREND_CACHE_KEY, {}) now = dt_util.utcnow() predictor = SensorPredictor(hass) out: dict[str, tuple[int, str]] = {} for bat in batteries: if bat.level is None or not bat.available or bat.low: continue cached = cache.get(bat.entity_id) if cached is not None and now - cached[0] < _TREND_CACHE_TTL: if cached[1] is not None: out[bat.entity_id] = cached[1] continue # The replacement moment is the fleet's low signal — the battery's # own low_threshold (shared with the sparkline and the level bars). threshold = bat.low_threshold result: tuple[int, str] | None = None try: pred = await predictor.async_predict_below(bat.entity_id, threshold, max_recovery=_TREND_MAX_RECOVERY_PCT) if ( pred is not None and pred.days_until_threshold is not None and pred.confidence in _TREND_MIN_CONFIDENCE and pred.days_until_threshold <= _TREND_MAX_DAYS ): result = (int(pred.days_until_threshold), pred.confidence) except Exception: # noqa: BLE001 - a recorder hiccup must never break the overview _LOGGER.debug("Trend prediction failed for %s", bat.entity_id, exc_info=True) cache[bat.entity_id] = (now, result) if result is not None: out[bat.entity_id] = result return out async def async_compute_overview(hass: HomeAssistant, *, horizon_days: int = DEFAULT_HORIZON_DAYS) -> BatteryOverview: """Read + trend-enrich + aggregate (the panel's entry point).""" batteries = read_batteries(hass) trends = await async_trend_predictions(hass, batteries) return build_overview(batteries, today=dt_util.now().date(), horizon_days=horizon_days, trend_predictions=trends) # ── level history for the roster sparklines ──────────────────────────────── _HISTORY_CACHE_KEY = "maintenance_supporter_battery_history_cache" _HISTORY_CACHE_TTL = timedelta(hours=6) # ~60 points draw a smooth 30 d line; hourly stats would be 720. _HISTORY_MAX_POINTS = 60 def _downsample(points: list[tuple[float, float]], max_points: int = _HISTORY_MAX_POINTS) -> list[tuple[float, float]]: """Bucket-mean a point series down to at most ``max_points``. Mean per bucket (not every-Nth) so a short voltage dip still leaves a visible dent instead of being skipped entirely. """ if len(points) <= max_points: return points size = (len(points) + max_points - 1) // max_points out: list[tuple[float, float]] = [] for i in range(0, len(points), size): bucket = points[i : i + size] out.append((bucket[-1][0], sum(v for _, v in bucket) / len(bucket))) return out # A real cell swap shows as a large upward step between adjacent 12 h buckets # (+40..+90 typically); relaxation bounces stay under ~5. Between them: 25. _JUMP_MIN_RISE = 25.0 # A jump already recorded within this many days of battery_last_replaced is # NOT flagged — the user pressed the button, nothing to fix. _JUMP_RECORDED_SLACK_DAYS = 2 def _detect_unrecorded_jump( points: list[tuple[float, float]], last_replaced: date | None, *, rechargeable: bool = False, ) -> dict[str, Any] | None: """An upward level step that looks like a swap nobody recorded. A real fleet had a sensor sit at 16 % for three weeks, get fresh cells and jump to 100 % — while ``battery_last_replaced`` stayed 21 months old, silently anchoring the type-lifetime forecast to the DEAD battery. The step is unmistakable in the recorder, so surface it and offer to record it. Rechargeables are exempt: their packs jump on every routine charge. """ from itertools import pairwise if rechargeable: return None for (_, v_prev), (ts, v) in pairwise(points): if v - v_prev < _JUMP_MIN_RISE: continue jump_date = dt_util.utc_from_timestamp(ts).date() if last_replaced is not None and abs((jump_date - last_replaced).days) <= _JUMP_RECORDED_SLACK_DAYS: continue # already recorded return {"at": round(ts), "from": round(v_prev, 1), "to": round(v, 1)} return None def _note_low_threshold(attrs: dict[str, Any]) -> float: """The Battery-Notes-configured threshold OR the fleet floor — the higher.""" raw = attrs.get("battery_low_threshold") if isinstance(raw, (int, float)): return float(max(raw, NATIVE_LOW_PERCENT)) return float(NATIVE_LOW_PERCENT) async def async_level_history(hass: HomeAssistant, batteries: list[Battery]) -> dict[str, dict[str, Any]]: """Per-battery downsampled level history: {entity_id: {points, threshold}}. Feeds the roster sparklines. Same 30 d recorder window the trend regression sees (so the drawn line IS what the forecast reasoned about), same 6 h cache-including-misses discipline as the trend — the roster is opened per panel visit and batteries drain over weeks. Low batteries are included (unlike the trend): the dive INTO low is exactly what the sparkline should show. """ from .sensor_predictor import SensorPredictor cache: dict[str, tuple[Any, list[tuple[float, float]]]] = hass.data.setdefault(_HISTORY_CACHE_KEY, {}) now = dt_util.utcnow() predictor = SensorPredictor(hass) out: dict[str, dict[str, Any]] = {} for bat in batteries: if bat.level is None and not bat.low: continue # low-only binaries have no level series to draw cached = cache.get(bat.entity_id) if cached is not None and now - cached[0] < _HISTORY_CACHE_TTL: points = cached[1] else: try: # Deliberate reuse of the predictor's fetch so the sparkline # and the regression see the same series. points = _downsample(await predictor._async_fetch_statistics_points(bat.entity_id, 30)) except Exception: # noqa: BLE001 - a recorder hiccup must never break the roster _LOGGER.debug("Level history failed for %s", bat.entity_id, exc_info=True) points = [] cache[bat.entity_id] = (now, points) if points: entry: dict[str, Any] = { "points": [[round(ts), round(v, 1)] for ts, v in points], "threshold": bat.low_threshold, } jump = _detect_unrecorded_jump(points, bat.last_replaced, rechargeable=is_rechargeable_type(bat.battery_type)) if jump is not None: # The Battery Notes service that records a replacement takes # the DEVICE — resolve it here so the panel's one-click fix # doesn't need a registry lookup of its own. from homeassistant.helpers import entity_registry as er reg = er.async_get(hass).async_get(bat.entity_id) if reg and reg.device_id: entry["jump"] = {**jump, "device_id": reg.device_id} out[bat.entity_id] = entry return out def discover_battery_types(hass: HomeAssistant) -> OrderedDict[str, int]: """Battery types present across the fleet → total quantity, for part setup. Rechargeable types are left out: nobody stocks a "RECHARGEABLE" spare, so setup must not mint a part (with a reorder threshold!) for one. The UNKNOWN bucket is left out for the same reason — native batteries without a type once minted an "UNKNOWN battery" part whose buy link was an Amazon search for the literal word UNKNOWN (seen on a real fleet at 0 of 22). Give the battery a type (a Battery Notes note) and it gets a real part. """ totals: OrderedDict[str, int] = OrderedDict() for bat in read_batteries(hass): if is_rechargeable_type(bat.battery_type): continue t = _norm_type(bat.battery_type) if t == "UNKNOWN": continue totals[t] = totals.get(t, 0) + bat.quantity return OrderedDict(sorted(totals.items())) __all__ = [ "DEFAULT_HORIZON_DAYS", "NATIVE_LOW_PERCENT", "TYPICAL_LIFETIME_MONTHS", "Battery", "BatteryOverview", "async_compute_overview", "async_level_history", "async_trend_predictions", "build_overview", "compute_overview", "discover_battery_types", "fleet_excluded_entities", "has_batteries", "has_battery_notes", "is_rechargeable_type", "lifetime_months", "read_batteries", ]