1594 lines
75 KiB
Python
1594 lines
75 KiB
Python
"""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. A note whose device has NO battery level at all (D#162) gets
|
||
no ``battery_plus`` from Battery Notes — only the ``…_battery_type`` sensor,
|
||
the ``…_battery_last_replaced`` timestamp and the replaced button — and is
|
||
read from the type sensor as a forecast-only row (``no_sensor``).
|
||
|
||
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 collections.abc import Callable
|
||
from dataclasses import dataclass, field
|
||
from datetime import date, timedelta
|
||
from typing import Any, cast
|
||
|
||
from homeassistant.core import HomeAssistant
|
||
from homeassistant.util import dt as dt_util
|
||
|
||
_LOGGER = logging.getLogger(__name__)
|
||
|
||
# The typical-lifetime table, the household overrides and the learned values
|
||
# live in battery_lifetime (D#162 follow-up); the names stay importable here.
|
||
from .battery_lifetime import (
|
||
TYPICAL_LIFETIME_MONTHS,
|
||
LifetimeInfo,
|
||
canonical_type,
|
||
has_type_forecast,
|
||
is_rechargeable_type,
|
||
is_shoppable_type,
|
||
lifetime_resolver,
|
||
observe_replacements,
|
||
table_lifetime_months,
|
||
)
|
||
|
||
# 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 _is_type_note(state: Any) -> bool:
|
||
"""Whether a sensor state is a Battery Notes ``…_battery_type`` sensor.
|
||
|
||
The type sensor carries ``battery_type`` + ``battery_quantity`` but NO
|
||
device class (``battery_plus`` and its low binary carry the same two
|
||
attributes WITH ``device_class: battery``). It exists for every note —
|
||
for a device without a level sensor it is the only trace (D#162).
|
||
"""
|
||
attrs = state.attributes
|
||
return attrs.get("device_class") is None and "battery_type" in attrs and "battery_quantity" in attrs
|
||
|
||
|
||
# Battery Notes unique_id suffixes — every entity of one note shares the
|
||
# subentry's unique_id plus one of these, so siblings resolve through the
|
||
# registry even after the user renamed the entity ids.
|
||
_NOTE_UID_SUFFIXES = (
|
||
"_battery_last_replaced",
|
||
"_battery_replaced_button",
|
||
"_battery_plus",
|
||
"_battery_type",
|
||
"_battery_low",
|
||
)
|
||
|
||
|
||
def note_sibling_entity(hass: HomeAssistant, entity_id: str, *, domain: str, uid_suffix: str) -> str | None:
|
||
"""The registry-resolved Battery Notes sibling of ``entity_id`` — e.g. the
|
||
``button`` / ``_battery_replaced_button`` of a type sensor — or None when
|
||
the entity has no Battery Notes registry entry (state-only) or the
|
||
sibling was never created. Callers fall back to the naming contract."""
|
||
from homeassistant.helpers import entity_registry as er
|
||
|
||
ent_reg = er.async_get(hass)
|
||
reg = ent_reg.async_get(entity_id)
|
||
if reg is None or reg.platform != "battery_notes" or not isinstance(reg.unique_id, str):
|
||
return None
|
||
for own in _NOTE_UID_SUFFIXES:
|
||
if reg.unique_id.endswith(own):
|
||
return ent_reg.async_get_entity_id(domain, "battery_notes", reg.unique_id[: -len(own)] + uid_suffix)
|
||
return None
|
||
|
||
|
||
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 lifetime_months(battery_type: str) -> int:
|
||
"""Typical service life for a battery type — the built-in table value
|
||
(aliases such as LR6/PP3/CR123 folded onto the table key). The forecast
|
||
itself goes through :func:`battery_lifetime.lifetime_resolver`, which
|
||
puts the household's override and the learned value in front of it."""
|
||
return table_lifetime_months(battery_type)
|
||
|
||
|
||
@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)
|
||
# D#162: a Battery Notes note WITHOUT a level sensor — only the type
|
||
# sensor, the last-replaced timestamp and the replaced button exist.
|
||
# Nothing can ever report "still fine": the forecast is the whole signal,
|
||
# and (option, default on) a passed forecast makes it ``low``.
|
||
no_sensor: bool = False
|
||
# D#162 follow-up: Battery Notes minted a replaced button for this note
|
||
# — the roster offers the per-row Replaced action. Decided in
|
||
# read_batteries for EVERY Battery Notes row (a low-only binary row has
|
||
# no level either and needs it just as much as a sensorless one).
|
||
can_mark_replaced: bool = False
|
||
# D#162 follow-up: "manufacturer|model" of the device (device registry)
|
||
# — the pool a learned lifetime is drawn from. A CR2032 in a door sensor
|
||
# and one in a thermostat share nothing but the cell; devices of the same
|
||
# model do. Empty when the device is unknown.
|
||
model_key: str = ""
|
||
# #180: ``low`` is held by the recovery latch — the reading itself is no
|
||
# longer at/below the floor, but it has not risen above
|
||
# ``battery_recovered_percent`` (and no replacement was recorded) since
|
||
# it went low. A level hovering around the floor stays ONE low episode.
|
||
latched: bool = False
|
||
|
||
|
||
@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, months: int | None = None) -> date | None:
|
||
if bat.last_replaced is None:
|
||
return None
|
||
if months is 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,
|
||
lifetime_for: Callable[[Battery], LifetimeInfo] | None = None,
|
||
) -> BatteryOverview:
|
||
"""Aggregate batteries into the fleet view.
|
||
|
||
``lifetime_for`` resolves a battery's lifetime (override > this device's
|
||
own replacements > devices of the same model > table > default, see
|
||
battery_lifetime); without it the table alone is used — the pure-function
|
||
tests and the summary sensors call it that way.
|
||
|
||
* ``low`` = reported low right now (Battery Notes' own threshold) — or a
|
||
sensorless note whose forecast has passed with the due-without-sensor
|
||
option on (D#162; decided in :func:`read_batteries`).
|
||
* ``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()):
|
||
# ONE normalisation for grouping, part ids and the lifetime table
|
||
# (DRY audit 2026-09): before, grouping only upper-cased, so an LR6
|
||
# note minted its own part/chip while the forecast read it as AA.
|
||
t = canonical_type(bat.battery_type)
|
||
types_seen[t] = None
|
||
rechargeable = is_rechargeable_type(bat.battery_type)
|
||
# The shopping groupings list only types a part can stand behind —
|
||
# the same predicate discover_battery_types mints parts by (a low
|
||
# "UNKNOWN" / "Manual" battery used to land in needs_now with no
|
||
# part to buy).
|
||
shoppable = is_shoppable_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)
|
||
info = lifetime_for(bat) if lifetime_for else LifetimeInfo(lifetime_months(bat.battery_type), "table")
|
||
if trend is not None:
|
||
days_raw: int | None = trend[0]
|
||
source, confidence = "trend", trend[1]
|
||
else:
|
||
# No type lifetime for rechargeables (the table describes primary
|
||
# cells) nor for "Manual" / "Irreplaceable" / "Solar" notes.
|
||
pred = None if rechargeable or not has_type_forecast(bat.battery_type) else _predicted_date(bat, info.months)
|
||
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:
|
||
# A battery reported low has no meaningful forecast left to show —
|
||
# except a sensorless note (D#162), where the PASSED forecast IS
|
||
# why it is due: keep the (negative) days and the overdue flag so
|
||
# the roster can show the date the prediction ran out.
|
||
if bat.no_sensor:
|
||
low_row = _row(bat, t, days_raw, source, confidence, rechargeable=rechargeable, forecast_overdue=overdue, lifetime=info)
|
||
else:
|
||
low_row = _row(bat, t, None, rechargeable=rechargeable, lifetime=info)
|
||
ov.low.append(low_row)
|
||
if shoppable:
|
||
ov.needs_now[t] = ov.needs_now.get(t, 0) + bat.quantity
|
||
ov.all.append({**low_row, "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, lifetime=info))
|
||
if shoppable:
|
||
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, lifetime=info), "status": "soon"})
|
||
continue
|
||
ov.all.append({**_row(bat, t, days, source, confidence, rechargeable=rechargeable, lifetime=info), "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,
|
||
lifetime: LifetimeInfo | None = None,
|
||
) -> dict[str, Any]:
|
||
return {
|
||
"entity_id": bat.entity_id,
|
||
# D#162 follow-up: which lifetime the "typical" forecast used and where
|
||
# it came from — the roster tooltip says "18 months, learned from 5
|
||
# replacements" instead of a bare date.
|
||
"lifetime_months": lifetime.months if lifetime else None,
|
||
"lifetime_source": lifetime.source if lifetime else None,
|
||
"lifetime_samples": lifetime.samples if lifetime else 0,
|
||
"model_key": bat.model_key,
|
||
"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,
|
||
# D#162: no level source at all — the roster shows a "No sensor" chip
|
||
# instead of a level bar, and the replaced button is the row's action.
|
||
"no_sensor": bat.no_sensor,
|
||
"can_mark_replaced": bat.can_mark_replaced,
|
||
"last_replaced": bat.last_replaced.isoformat() if bat.last_replaced else None,
|
||
# #180: low only because the recovery latch holds it (the reading is
|
||
# back above the floor but not yet above battery_recovered_percent).
|
||
"latched": bat.latched,
|
||
}
|
||
|
||
|
||
def _quantity_of(attrs: Any) -> int:
|
||
"""``battery_quantity`` as a positive int; anything odd ("2 pcs", "1.0",
|
||
None) is one cell - a template note with a bad value must not raise out
|
||
of read_batteries and blank the whole fleet."""
|
||
raw = attrs.get("battery_quantity") if hasattr(attrs, "get") else None
|
||
try:
|
||
qty = int(float(raw)) if raw not in (None, "") else 1
|
||
except (TypeError, ValueError):
|
||
return 1
|
||
return qty if qty >= 1 else 1
|
||
|
||
|
||
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:
|
||
# An aware stamp (Battery Notes writes UTC) is a LOCAL calendar
|
||
# date: 23:30 UTC on the 3rd is the 4th in Berlin.
|
||
return (dt_util.as_local(parsed) if parsed.tzinfo is not None else 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_object(hass: HomeAssistant) -> dict[str, Any]:
|
||
"""The fleet object dict from its config entry, or ``{}`` when no fleet.
|
||
|
||
Inlined entry loop (not via battery_fleet_setup.find_fleet_entry) to keep
|
||
this module import-cycle-free — setup imports the aggregation, not vice
|
||
versa. The three getters below were three copies of this loop (drift
|
||
audit 2026-08).
|
||
"""
|
||
from ..const import BATTERY_FLEET_OBJECT_FLAG, CONF_OBJECT, DOMAIN
|
||
|
||
for entry in hass.config_entries.async_entries(DOMAIN):
|
||
obj = entry.data.get(CONF_OBJECT, {})
|
||
if obj.get(BATTERY_FLEET_OBJECT_FLAG):
|
||
return cast(dict[str, Any], obj)
|
||
return {}
|
||
|
||
|
||
def fleet_excluded_entities(hass: HomeAssistant) -> set[str]:
|
||
"""Manually excluded battery entity_ids, stored on the fleet object entry."""
|
||
from ..const import BATTERY_FLEET_EXCLUDED
|
||
|
||
return set(_fleet_object(hass).get(BATTERY_FLEET_EXCLUDED) or [])
|
||
|
||
|
||
def fleet_included_entities(hass: HomeAssistant) -> set[str]:
|
||
"""Manually ADDED battery entity_ids (#135) — same storage pattern as
|
||
the exclusions. An include bypasses the discovery heuristic and the
|
||
self-charging filter in BOTH passes (lurisin's Oura Ring carried a
|
||
Battery Notes note, so the native-pass-only bypass never fired);
|
||
Battery-Notes coverage and the exclusion list still win (no duplicate
|
||
rows, exclusion stays king)."""
|
||
from ..const import BATTERY_FLEET_INCLUDED
|
||
|
||
return set(_fleet_object(hass).get(BATTERY_FLEET_INCLUDED) or [])
|
||
|
||
|
||
def fleet_track_self_charging(hass: HomeAssistant) -> bool:
|
||
"""Whether the fleet keeps self-charging devices in the roster (#135).
|
||
|
||
Off (the default) preserves #107: vacuums/phones/rings never enter. On,
|
||
they appear as rechargeables — typed "Rechargeable", labelled
|
||
"— recharge", never counted into the shopping needs — so a low phone or
|
||
smart ring can drive a "charge it" notification. Exclusions still win.
|
||
"""
|
||
from ..const import BATTERY_FLEET_TRACK_SELF_CHARGING
|
||
|
||
return bool(_fleet_object(hass).get(BATTERY_FLEET_TRACK_SELF_CHARGING))
|
||
|
||
|
||
def fleet_due_without_sensor(hass: HomeAssistant) -> bool:
|
||
"""Whether a PASSED forecast on a sensorless note counts as due (D#162).
|
||
|
||
On (the default): a Battery Notes note that has no level sensor — only
|
||
its type, quantity and last-replaced date — goes ``low`` once its
|
||
predicted replacement date has passed, so the fleet task fires for it.
|
||
Nothing else ever could: there is no reading to say "still fine". Off:
|
||
such notes only ever forecast (B1 behaviour — soon, never low). Stored
|
||
as False to switch off; absent means on.
|
||
"""
|
||
from ..const import BATTERY_FLEET_DUE_WITHOUT_SENSOR
|
||
|
||
return _fleet_object(hass).get(BATTERY_FLEET_DUE_WITHOUT_SENSOR) is not False
|
||
|
||
|
||
def get_battery_recovered_percent(hass: HomeAssistant) -> int:
|
||
"""#180: the level a low battery must rise ABOVE to count as recovered
|
||
(replaced). Out-of-range / junk values fall back to the default."""
|
||
from ..const import BATTERY_RECOVERED_PERCENT_RANGE, CONF_BATTERY_RECOVERED_PERCENT, DEFAULT_BATTERY_RECOVERED_PERCENT
|
||
from .global_options import get_global_options
|
||
|
||
raw = get_global_options(hass).get(CONF_BATTERY_RECOVERED_PERCENT, DEFAULT_BATTERY_RECOVERED_PERCENT)
|
||
try:
|
||
value = int(raw)
|
||
except (TypeError, ValueError):
|
||
return DEFAULT_BATTERY_RECOVERED_PERCENT
|
||
lo, hi = BATTERY_RECOVERED_PERCENT_RANGE
|
||
return value if lo <= value <= hi else DEFAULT_BATTERY_RECOVERED_PERCENT
|
||
|
||
|
||
def get_battery_auto_record_recovery(hass: HomeAssistant) -> bool:
|
||
"""#181 follow-up: whether a battery the latch releases BY ITS LEVEL gets
|
||
its replacement recorded automatically (the Battery Notes date + the
|
||
type's cells from stock, see :func:`_schedule_auto_record`). Advanced
|
||
option, off by default."""
|
||
from ..const import CONF_BATTERY_AUTO_RECORD_RECOVERY
|
||
from .global_options import get_global_options
|
||
from .settings_registry import setting_default
|
||
|
||
return bool(get_global_options(hass).get(CONF_BATTERY_AUTO_RECORD_RECOVERY, setting_default(CONF_BATTERY_AUTO_RECORD_RECOVERY)))
|
||
|
||
|
||
# ── #180: the low-recovery latch ─────────────────────────────────────────────
|
||
#
|
||
# A Hue dimmer's level oscillated around the low floor several times a day;
|
||
# the low-count sensor flipped 0 ↔ 1 and the fleet task — auto-complete on
|
||
# recovery — recorded a "completion" on every dip. Once a LEVEL-bearing
|
||
# battery goes low it is latched: it stays counted low (needs_now, the
|
||
# low-count sensor, the Needs-now list) until its level rises ABOVE
|
||
# ``battery_recovered_percent``, or its Battery Notes ``battery_last_replaced``
|
||
# date moves forward (a recorded replacement releases it at any level), or the
|
||
# Replaced action / the record-replacement command releases it directly.
|
||
#
|
||
# Level-less rows (a low-only binary, a native binary battery) are NOT
|
||
# latched: their binary saying "not low" is the all-clear, exactly as before.
|
||
# Sensorless notes (D#162) leave ``low`` only when their forecast re-anchors.
|
||
#
|
||
# Persisted on the fleet task's Store state ({entity_id: {at, last_replaced}})
|
||
# so a restart does not re-open the episode; without a fleet the latch lives
|
||
# in memory (the low-count sensor exists before the fleet does).
|
||
|
||
LOW_LATCH_KEY = "battery_low_latch"
|
||
_LATCH_MEMORY_KEY = "battery_fleet_low_latch_memory"
|
||
|
||
|
||
def _replaced_since(last_replaced: str | None, entry: dict[str, Any]) -> bool:
|
||
"""Whether the battery's current last-replaced date is NEWER than the one
|
||
seen when it latched (a recorded replacement)."""
|
||
if last_replaced is None:
|
||
return False
|
||
seen = entry.get("last_replaced")
|
||
return not isinstance(seen, str) or last_replaced > seen
|
||
|
||
|
||
def apply_low_latch(
|
||
batteries: list[Battery],
|
||
latch: dict[str, Any],
|
||
*,
|
||
recovered: float,
|
||
now_iso: str,
|
||
eligible: set[str] | None = None,
|
||
released_by_level: list[str] | None = None,
|
||
) -> bool:
|
||
"""Pure latch step: mutate ``bat.low``/``bat.latched`` and the ``latch``
|
||
map in place. Returns True when the map changed (the caller persists).
|
||
|
||
``eligible`` = entity ids whose ``low`` is level-driven (default: every
|
||
battery with a level, never a sensorless note). Per eligible battery:
|
||
|
||
* reading low → latch it (or re-latch when a replacement was recorded
|
||
and the fresh reading is STILL low — a new episode);
|
||
* latched + unavailable → stays low (no reading is no proof of recovery);
|
||
* latched + reading not low → released when the level is above
|
||
``recovered`` or a replacement was recorded, else held low.
|
||
|
||
``released_by_level`` (#181 follow-up) collects the ids released by their
|
||
LEVEL alone — no newer replacement date on the note — which is the
|
||
auto-record hook's input: a recorded date already counts as recorded.
|
||
|
||
Entries for batteries no longer in the fleet are dropped.
|
||
"""
|
||
changed = False
|
||
if eligible is None:
|
||
eligible = {b.entity_id for b in batteries if b.level is not None and not b.no_sensor}
|
||
for bat in batteries:
|
||
if bat.entity_id not in eligible:
|
||
continue
|
||
last = bat.last_replaced.isoformat() if bat.last_replaced else None
|
||
raw_entry = latch.get(bat.entity_id)
|
||
entry = raw_entry if isinstance(raw_entry, dict) else None
|
||
if bat.low:
|
||
if entry is None or _replaced_since(last, entry):
|
||
latch[bat.entity_id] = {"at": now_iso, "last_replaced": last}
|
||
changed = True
|
||
continue
|
||
if entry is None:
|
||
continue
|
||
if bat.available and (_replaced_since(last, entry) or (bat.level is not None and bat.level > recovered)):
|
||
if released_by_level is not None and not _replaced_since(last, entry):
|
||
released_by_level.append(bat.entity_id)
|
||
del latch[bat.entity_id]
|
||
changed = True
|
||
continue
|
||
bat.low = True
|
||
bat.latched = True
|
||
known = {b.entity_id for b in batteries}
|
||
for eid in [e for e in latch if e not in known]:
|
||
del latch[eid]
|
||
changed = True
|
||
return changed
|
||
|
||
|
||
def _latch_backend(hass: HomeAssistant) -> tuple[dict[str, Any], Callable[[], None]]:
|
||
"""The latch map + a save callback: the fleet task's Store state when a
|
||
fleet exists, else an in-memory map (nothing to persist to yet)."""
|
||
from ..const import DOMAIN
|
||
from .battery_lifetime import fleet_store_and_task
|
||
|
||
found = fleet_store_and_task(hass)
|
||
if found is not None:
|
||
store, task_id = found
|
||
raw = store.get_task_state(task_id).get(LOW_LATCH_KEY)
|
||
latch: dict[str, Any] = dict(raw) if isinstance(raw, dict) else {}
|
||
|
||
def _save() -> None:
|
||
store.update_task_state(task_id, **{LOW_LATCH_KEY: latch})
|
||
store.async_delay_save()
|
||
|
||
return latch, _save
|
||
memory: dict[str, Any] = hass.data.setdefault(DOMAIN, {}).setdefault(_LATCH_MEMORY_KEY, {})
|
||
return memory, lambda: None
|
||
|
||
|
||
def _apply_low_latch(hass: HomeAssistant, batteries: list[Battery], eligible: set[str]) -> None:
|
||
latch, save = _latch_backend(hass)
|
||
released: list[str] = []
|
||
if apply_low_latch(
|
||
batteries,
|
||
latch,
|
||
recovered=float(get_battery_recovered_percent(hass)),
|
||
now_iso=dt_util.utcnow().isoformat(),
|
||
eligible=eligible,
|
||
released_by_level=released,
|
||
):
|
||
# Persist BEFORE scheduling: the record path re-reads the fleet and
|
||
# must see the entry gone (or it would release — and schedule — again).
|
||
save()
|
||
if released and get_battery_auto_record_recovery(hass):
|
||
_schedule_auto_record(hass, [b for b in batteries if b.entity_id in released])
|
||
|
||
|
||
# ── #181 follow-up: auto-record a level-driven recovery ────────────────────
|
||
#
|
||
# With ``battery_auto_record_recovery`` on, a battery the latch releases
|
||
# BECAUSE ITS LEVEL rose above the recovery threshold — a fresh cell reporting,
|
||
# not a recorded date, which already counts as recorded — is recorded through
|
||
# the same path as the roster's calendar chip: ``async_record_replacement``
|
||
# writes the date to Battery Notes and consumes the type's cells (the note's
|
||
# own quantity) ONCE per day. Per battery, so a partial swap is recorded the
|
||
# moment that battery reports fresh; nothing waits for the fleet task.
|
||
#
|
||
# Scheduled as a task because ``read_batteries`` is sync (it runs inside the
|
||
# low-count sensor's update) and the record path re-reads the fleet itself.
|
||
# Re-entrancy: the latch entry is already gone when the task runs, so the
|
||
# nested read cannot release the same battery again; the in-flight set covers
|
||
# the window until then, and the record path's once-per-day memory makes a
|
||
# bounce that re-latches and recovers on the same day consume nothing twice.
|
||
|
||
_AUTO_RECORD_INFLIGHT_KEY = "battery_fleet_auto_record_inflight"
|
||
|
||
|
||
def _auto_record_inflight(hass: HomeAssistant) -> set[str]:
|
||
from ..const import DOMAIN
|
||
|
||
inflight: set[str] = hass.data.setdefault(DOMAIN, {}).setdefault(_AUTO_RECORD_INFLIGHT_KEY, set())
|
||
return inflight
|
||
|
||
|
||
def _schedule_auto_record(hass: HomeAssistant, batteries: list[Battery]) -> int:
|
||
"""Schedule the auto-record for batteries just released by their level.
|
||
Returns how many were scheduled: a native row has no Battery Notes note
|
||
to record on, and a rechargeable's recovery is a charge, not a swap."""
|
||
import threading
|
||
|
||
inflight = _auto_record_inflight(hass)
|
||
scheduled = 0
|
||
for bat in batteries:
|
||
if bat.source != "battery_notes":
|
||
_LOGGER.debug("Battery %s recovered but has no Battery Notes note - nothing to record on", bat.entity_id)
|
||
continue
|
||
if is_rechargeable_type(bat.battery_type):
|
||
_LOGGER.debug("Battery %s recovered by charging - no replacement to record", bat.entity_id)
|
||
continue
|
||
if bat.entity_id in inflight:
|
||
continue
|
||
inflight.add(bat.entity_id)
|
||
coro = _async_auto_record(hass, bat.entity_id)
|
||
if hass.loop_thread_id != threading.get_ident():
|
||
hass.create_task(coro, "maintenance_supporter_battery_auto_record")
|
||
else:
|
||
hass.async_create_task(coro, "maintenance_supporter_battery_auto_record", eager_start=False)
|
||
scheduled += 1
|
||
return scheduled
|
||
|
||
|
||
async def _async_auto_record(hass: HomeAssistant, entity_id: str) -> None:
|
||
from homeassistant.exceptions import HomeAssistantError
|
||
|
||
from .battery_fleet_setup import async_record_replacement
|
||
|
||
try:
|
||
result = await async_record_replacement(hass, entity_id, dt_util.utcnow().isoformat())
|
||
_LOGGER.info(
|
||
"Battery %s recovered above the threshold: replacement recorded automatically (consumed %s)",
|
||
entity_id,
|
||
result.get("consumed") or {},
|
||
)
|
||
except HomeAssistantError as err:
|
||
# not_available (Battery Notes gone), invalid_device (state-only
|
||
# note), not_found (excluded meanwhile) - nothing to record on.
|
||
_LOGGER.debug("Automatic replacement record for %s skipped: %s", entity_id, err)
|
||
except Exception: # noqa: BLE001 - a hook must never take the fleet down
|
||
_LOGGER.warning("Automatic replacement record for %s failed", entity_id, exc_info=True)
|
||
finally:
|
||
_auto_record_inflight(hass).discard(entity_id)
|
||
|
||
|
||
def release_low_latch(hass: HomeAssistant, entity_ids: list[str]) -> int:
|
||
"""Drop the latch for batteries the user just marked replaced (the
|
||
Replaced action / a recorded replacement) — the count must not wait for
|
||
Battery Notes to echo the new date. Returns how many were released."""
|
||
latch, save = _latch_backend(hass)
|
||
released = 0
|
||
for eid in entity_ids:
|
||
if eid in latch:
|
||
del latch[eid]
|
||
released += 1
|
||
if released:
|
||
save()
|
||
return released
|
||
|
||
|
||
def device_model_key(hass: HomeAssistant, device_id: str | None) -> str:
|
||
""""manufacturer|model" (lowercased) for a device registry id, "" when
|
||
unknown — the learning pool key (see battery_lifetime)."""
|
||
if not device_id:
|
||
return ""
|
||
from homeassistant.helpers import device_registry as dr
|
||
|
||
device = dr.async_get(hass).async_get(device_id)
|
||
if device is None:
|
||
return ""
|
||
# HA 2026.9 may hand back a ChildDeviceEntry, which has no make/model.
|
||
manufacturer = str(getattr(device, "manufacturer", None) or "").strip().lower()
|
||
model = str(getattr(device, "model", None) or getattr(device, "model_id", None) or "").strip().lower()
|
||
if not model or not manufacturer:
|
||
# A generic model without a maker ("Door Sensor") would pool
|
||
# unrelated devices from different integrations.
|
||
return ""
|
||
return f"{manufacturer}|{model}"
|
||
|
||
|
||
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)
|
||
# #127: identifiers are typed (domain, id) but the registry doesn't enforce
|
||
# it — NissanConnect ships 3-element tuples, and strict unpacking took the
|
||
# whole fleet down. Only the domain matters here.
|
||
if device and any(ident[0] == "mobile_app" for ident 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) —
|
||
unless the device still carries a Battery Notes ``…_battery_type``
|
||
sensor (plus entities disabled or hidden, #186): then type, quantity
|
||
and the replacement date come from that note. 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).
|
||
* **Sensorless notes** (D#162, pass 1b): a device with no battery level
|
||
at all gets no ``battery_plus`` from Battery Notes — only the
|
||
``…_battery_type`` sensor (type/quantity), the
|
||
``…_battery_last_replaced`` timestamp and the replaced button. The
|
||
type sensor becomes a forecast-only row (``no_sensor``, never
|
||
offline, no level): last-replaced + the type's typical lifetime
|
||
predicts the swap, and with ``fleet_due_without_sensor`` on a PASSED
|
||
forecast makes the row ``low`` — the only way such a battery can ever
|
||
reach the task. Devices that have any ``battery_plus`` (kept, dropped
|
||
or excluded) or a native row are skipped: the real reading wins.
|
||
* Manually excluded entity_ids (fleet detail → exclude) are dropped from
|
||
ALL passes.
|
||
* Manual includes (#135) act DEVICE-wide: whichever of a device's battery
|
||
entities the user picked, the self-charging filter is lifted for the
|
||
whole device in both passes — so the richest source (a Battery Notes
|
||
note) still wins the row. With ``fleet_track_self_charging`` on, the
|
||
self-charging filter is off entirely and such devices surface as
|
||
rechargeables (native rows typed "Rechargeable", never "Unknown").
|
||
"""
|
||
# #146: the fleet-wide low floor is a household setting (default 20 %).
|
||
from .global_options import get_battery_low_percent
|
||
|
||
floor = float(get_battery_low_percent(hass))
|
||
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)
|
||
included = fleet_included_entities(hass)
|
||
track_self = fleet_track_self_charging(hass)
|
||
# The include acts device-wide (see docstring) — resolve each included
|
||
# entity to its device once.
|
||
included_devices: set[str] = set()
|
||
for _eid in included:
|
||
_reg = ent_reg.async_get(_eid)
|
||
if _reg and _reg.device_id:
|
||
included_devices.add(_reg.device_id)
|
||
# _is_self_charging walks the device's registry entries — cache per device
|
||
# (a device's %-sensor and low-binary would otherwise both pay for it).
|
||
_self_charging_cache: dict[str, bool] = {}
|
||
|
||
def _self_charging(dev_id: str | None) -> bool:
|
||
if not dev_id:
|
||
return False
|
||
if dev_id not in _self_charging_cache:
|
||
_self_charging_cache[dev_id] = _is_self_charging(hass, dev_id)
|
||
return _self_charging_cache[dev_id]
|
||
|
||
out: list[Battery] = []
|
||
covered_sources: set[str] = set()
|
||
covered_devices: set[str] = set()
|
||
# #180: rows whose ``low`` is decided by a LEVEL — the recovery latch
|
||
# applies to these only (a binary's "off" is its own all-clear).
|
||
latch_eligible: 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()
|
||
# Pass 1b dedupe: EVERY battery_plus entity (kept, dropped or excluded)
|
||
# and its device — a device that has a real reading never gets a
|
||
# forecast-only row on top.
|
||
note_binary_ids: set[str] = set()
|
||
note_devices_all: 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)
|
||
else:
|
||
note_binary_ids.add(state.entity_id)
|
||
reg = ent_reg.async_get(state.entity_id)
|
||
dev_id = reg.device_id if reg else None
|
||
if dev_id:
|
||
note_devices_all.add(dev_id)
|
||
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 <= floor)
|
||
latch_eligible.add(state.entity_id)
|
||
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 —
|
||
# unless the user explicitly included THIS entity (#135): a stated
|
||
# battery stays visible even while it reads like noise.
|
||
if not available and not low and last_replaced is None and state.entity_id not in included:
|
||
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. A manual include
|
||
# (this entity or a sibling on the same device, #135) or the
|
||
# track-self-charging option lifts the skip — lurisin's Oura Ring
|
||
# was noted AND self-charging, and the v2.61.0 bypass lived only
|
||
# in the native pass, so including it did nothing.
|
||
if (
|
||
not track_self
|
||
and state.entity_id not in included
|
||
and dev_id not in included_devices
|
||
and _self_charging(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=_quantity_of(attrs),
|
||
model_key=device_model_key(hass, dev_id),
|
||
low=low,
|
||
level=level,
|
||
last_replaced=last_replaced,
|
||
available=available,
|
||
source="battery_notes",
|
||
low_threshold=_note_low_threshold(attrs, floor),
|
||
)
|
||
)
|
||
|
||
# ── 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.
|
||
eid = state.entity_id
|
||
# #135: a manual include bypasses the discovery heuristic (and the
|
||
# self-charging filter below) — the user has stated this IS a
|
||
# battery. Coverage dedupe and the exclusion list still apply.
|
||
if eid not in included:
|
||
if domain == "sensor":
|
||
if not _is_native_battery_sensor(state):
|
||
continue
|
||
elif state.attributes.get("device_class") != "battery":
|
||
continue
|
||
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
|
||
self_charging = _self_charging(dev_id)
|
||
if ( # #107; lifted by an include on the device (#135) or the option
|
||
self_charging
|
||
and not track_self
|
||
and eid not in included
|
||
and dev_id not in included_devices
|
||
):
|
||
continue
|
||
key = dev_id or eid
|
||
rec = native.setdefault(
|
||
key,
|
||
{
|
||
"level_state": None,
|
||
"low_state": None,
|
||
"device_id": dev_id,
|
||
"eid": eid,
|
||
"name": None,
|
||
# A surfaced self-charging device is a RECHARGEABLE, not an
|
||
# "Unknown" cell — the type drives the "— recharge" label
|
||
# and keeps it out of the shopping needs.
|
||
"self_charging": self_charging,
|
||
},
|
||
)
|
||
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
|
||
|
||
# #186: a device whose Battery Notes plus entities are gone (disabled,
|
||
# hidden, or a note that never got a percentage source) still carries the
|
||
# diagnostic ``…_battery_type`` sensor — the native row takes type,
|
||
# quantity and the replacement date from it instead of degrading to
|
||
# "Unknown" (a whole fleet read UNKNOWN although every device page showed
|
||
# its type). Registry device first, Battery Notes' naming contract second.
|
||
type_note_by_device: dict[str, Any] = {}
|
||
type_note_by_base: dict[str, Any] = {}
|
||
for st in hass.states.async_all("sensor"):
|
||
if not _is_type_note(st):
|
||
continue
|
||
t_reg = ent_reg.async_get(st.entity_id)
|
||
if t_reg and t_reg.device_id:
|
||
type_note_by_device.setdefault(t_reg.device_id, st)
|
||
t_obj = st.entity_id.split(".", 1)[1]
|
||
if t_obj.endswith("_battery_type"):
|
||
type_note_by_base.setdefault(t_obj[: -len("_battery_type")], st)
|
||
|
||
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 <= floor
|
||
latch_eligible.add(rec["eid"])
|
||
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
|
||
note = type_note_by_device.get(rec["device_id"]) if rec["device_id"] else None
|
||
if note is None:
|
||
note = type_note_by_base.get(rec["eid"].split(".", 1)[1])
|
||
note_type = str(note.attributes.get("battery_type") or "").strip() if note is not None else ""
|
||
note_last: date | None = None
|
||
if note is not None and note_type:
|
||
n_obj = note.entity_id.split(".", 1)[1]
|
||
last_eid = note_sibling_entity(hass, note.entity_id, domain="sensor", uid_suffix="_battery_last_replaced") or (
|
||
f"sensor.{n_obj[: -len('_battery_type')]}_battery_last_replaced" if n_obj.endswith("_battery_type") else None
|
||
)
|
||
last_state = hass.states.get(last_eid) if last_eid else None
|
||
if last_state and last_state.state not in _NO_READING:
|
||
note_last = _parse_last_replaced(last_state.state)
|
||
out.append(
|
||
Battery(
|
||
entity_id=rec["eid"],
|
||
device_name=name or rec["eid"],
|
||
battery_type="Rechargeable" if rec.get("self_charging") else (note_type or "Unknown"),
|
||
quantity=_quantity_of(note.attributes) if note is not None and note_type else 1,
|
||
model_key=device_model_key(hass, rec.get("device_id")),
|
||
low=low,
|
||
level=level,
|
||
last_replaced=note_last,
|
||
available=available,
|
||
source="native",
|
||
# The household floor decided ``low`` above — the sparkline
|
||
# line, the level-bar colours and the trend regression must
|
||
# cross at the same level, not at the 20 % class default
|
||
# (bug review 2026-09-04).
|
||
low_threshold=floor,
|
||
)
|
||
)
|
||
|
||
# ── Pass 1b: Battery Notes notes WITHOUT a level sensor (D#162) ────────
|
||
# A device that exposes no battery percentage (maisun's 30 Xiaomi/Aqara
|
||
# sensors) gets NO battery_plus from Battery Notes — only the diagnostic
|
||
# ``…_battery_type`` sensor, the ``…_battery_last_replaced`` timestamp
|
||
# and the replaced button. Runs AFTER the other passes so any real
|
||
# reading (a plus of any kind, a native row) wins the device.
|
||
due_without_sensor = fleet_due_without_sensor(hass)
|
||
today = dt_util.now().date()
|
||
# Bug audit 2026-09-12: the "due" decision below must use the SAME lifetime
|
||
# the roster forecasts with (override > learned > table) - it used the bare
|
||
# table, so an override or a learned value moved the roster's ~date while
|
||
# the task kept firing (or never fired) on the table's.
|
||
lifetime_for = lifetime_resolver(hass)
|
||
native_devices = {rec["device_id"] for rec in native.values() if rec["device_id"]}
|
||
native_eids = {rec["eid"] for rec in native.values()}
|
||
for state in hass.states.async_all("sensor"):
|
||
if not _is_type_note(state):
|
||
continue
|
||
eid = state.entity_id
|
||
attrs = state.attributes
|
||
reg = ent_reg.async_get(eid)
|
||
dev_id = reg.device_id if reg else None
|
||
if dev_id and (dev_id in note_devices_all or dev_id in covered_devices or dev_id in native_devices):
|
||
continue
|
||
# State-only entities have no registry entry — Battery Notes' naming
|
||
# contract (``sensor.X_battery_type`` ↔ ``sensor.X_battery_plus`` ↔
|
||
# ``binary_sensor.X_battery_plus_low``) is the fallback dedupe.
|
||
object_id = eid.split(".", 1)[1]
|
||
base = object_id[: -len("_battery_type")] if object_id.endswith("_battery_type") else None
|
||
if base and (f"sensor.{base}_battery_plus" in note_sensor_ids or f"binary_sensor.{base}_battery_plus_low" in note_binary_ids):
|
||
continue
|
||
# #186: the native row on the same base already carries this note's type.
|
||
if base and (f"sensor.{base}" in native_eids or f"binary_sensor.{base}" in native_eids):
|
||
continue
|
||
if eid in excluded:
|
||
continue
|
||
if not track_self and eid not in included and dev_id not in included_devices and _self_charging(dev_id):
|
||
continue
|
||
battery_type = str(attrs.get("battery_type") or "").strip()
|
||
if not battery_type:
|
||
continue
|
||
# The replacement date lives on the sibling timestamp sensor (its
|
||
# state is the ISO timestamp) — registry first, naming second. It may
|
||
# be disabled in Battery Notes: then the row has no forecast at all.
|
||
last_eid = note_sibling_entity(hass, eid, domain="sensor", uid_suffix="_battery_last_replaced")
|
||
if last_eid is None and base:
|
||
last_eid = f"sensor.{base}_battery_last_replaced"
|
||
last_state = hass.states.get(last_eid) if last_eid else None
|
||
last_replaced = (
|
||
_parse_last_replaced(last_state.state) if last_state and last_state.state not in _NO_READING else None
|
||
)
|
||
# The type sensor is named "<device> Battery type" — the device name
|
||
# is the registry's; without a registry entry strip the entity's own
|
||
# name off the friendly name (English fallback for state-only notes).
|
||
name = None
|
||
if dev_id and (dev := dev_reg.async_get(dev_id)):
|
||
name = dev.name_by_user or dev.name
|
||
friendly = str(attrs.get("friendly_name") or "")
|
||
if not name and friendly:
|
||
own_name = (reg.name or reg.original_name) if reg else None
|
||
if own_name and friendly.endswith(own_name):
|
||
name = friendly[: -len(own_name)].strip()
|
||
else:
|
||
name = re.sub(r"\s+battery\s+type$", "", friendly, flags=re.IGNORECASE).strip()
|
||
if not name:
|
||
name = (base or object_id).replace("_", " ").strip().title()
|
||
bat = Battery(
|
||
entity_id=eid,
|
||
device_name=name,
|
||
battery_type=battery_type,
|
||
quantity=_quantity_of(attrs),
|
||
model_key=device_model_key(hass, dev_id),
|
||
low=False,
|
||
level=None,
|
||
last_replaced=last_replaced,
|
||
available=True,
|
||
source="battery_notes",
|
||
low_threshold=floor,
|
||
no_sensor=True,
|
||
)
|
||
# Due = the forecast has PASSED (same arithmetic as build_overview's
|
||
# forecast_overdue). Rechargeables never get a table forecast.
|
||
if due_without_sensor and is_shoppable_type(battery_type):
|
||
pred = _predicted_date(bat, lifetime_for(bat).months)
|
||
bat.low = pred is not None and pred < today
|
||
out.append(bat)
|
||
# #180: hold level-driven rows low until they clearly recover (or a
|
||
# replacement is recorded) — AFTER every pass, so the sensor, the panel
|
||
# overview and the Replaced action all see the same ``low``.
|
||
_apply_low_latch(hass, out, latch_eligible)
|
||
for bat in out:
|
||
if bat.source == "battery_notes":
|
||
bat.can_mark_replaced = has_replaced_button(hass, bat.entity_id)
|
||
return out
|
||
|
||
|
||
def has_replaced_button(hass: HomeAssistant, entity_id: str) -> bool:
|
||
"""Whether Battery Notes minted a replaced button for this note's row
|
||
(naming contract first, registry sibling second — the same lookup
|
||
:func:`battery_fleet_setup.async_mark_replaced` presses)."""
|
||
from .battery_fleet_setup import replaced_button_for
|
||
|
||
if hass.states.get(replaced_button_for(entity_id)) is not None:
|
||
return True
|
||
sibling = note_sibling_entity(hass, entity_id, domain="button", uid_suffix="_battery_replaced_button")
|
||
return sibling is not None and hass.states.get(sibling) is not None
|
||
|
||
|
||
def battery_notes_summary(hass: HomeAssistant) -> dict[str, Any] | None:
|
||
"""What Battery Notes currently reports, for the Settings hint (#146).
|
||
|
||
Derived purely from the ``battery_low_threshold`` attribute the
|
||
``battery_plus`` entities expose — no reach into Battery Notes' own
|
||
configuration. The most common value is their default; devices whose
|
||
value differs are overrides, named (highest first, capped at 5 so a
|
||
large install cannot spam the settings page) and linked via their
|
||
device id when the registry knows one. ``None`` when Battery Notes is
|
||
not present.
|
||
"""
|
||
from collections import Counter
|
||
|
||
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)
|
||
per_device: dict[str, dict[str, Any]] = {}
|
||
for domain in ("sensor", "binary_sensor"):
|
||
for state in hass.states.async_all(domain):
|
||
raw = state.attributes.get("battery_low_threshold")
|
||
if not isinstance(raw, (int, float)) or isinstance(raw, bool):
|
||
continue
|
||
reg = ent_reg.async_get(state.entity_id)
|
||
device_id = reg.device_id if reg else None
|
||
name = state.attributes.get("device_name")
|
||
if device_id and (device := dev_reg.async_get(device_id)):
|
||
name = device.name_by_user or device.name or name
|
||
key = device_id or state.entity_id
|
||
row = per_device.get(key)
|
||
threshold = float(raw)
|
||
if row is None or threshold > row["threshold"]:
|
||
per_device[key] = {
|
||
"name": str(name or state.entity_id),
|
||
"device_id": device_id,
|
||
"threshold": threshold,
|
||
}
|
||
if not per_device:
|
||
return None
|
||
counts = Counter(row["threshold"] for row in per_device.values())
|
||
default = counts.most_common(1)[0][0]
|
||
overrides = sorted(
|
||
(row for row in per_device.values() if row["threshold"] != default),
|
||
key=lambda r: (-r["threshold"], r["name"].lower()),
|
||
)
|
||
return {
|
||
"default": default,
|
||
"devices": len(per_device),
|
||
"overrides": overrides[:5],
|
||
"more": max(0, len(overrides) - 5),
|
||
}
|
||
|
||
|
||
|
||
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 — and so does a
|
||
fleet of sensorless notes (D#162), which only have type sensors.
|
||
"""
|
||
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
|
||
if domain == "sensor" and _is_type_note(state):
|
||
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) or _is_type_note(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()
|
||
batteries = read_batteries(hass)
|
||
observe_replacements(hass, batteries)
|
||
return build_overview(batteries, today=today, horizon_days=horizon_days, lifetime_for=lifetime_resolver(hass))
|
||
|
||
|
||
# ── 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)
|
||
observe_replacements(hass, batteries)
|
||
trends = await async_trend_predictions(hass, batteries)
|
||
return build_overview(
|
||
batteries, today=dt_util.now().date(), horizon_days=horizon_days, trend_predictions=trends, lifetime_for=lifetime_resolver(hass)
|
||
)
|
||
|
||
|
||
# ── 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,
|
||
recovered: float | None = None,
|
||
) -> 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.
|
||
|
||
``recovered`` (#181 noise, the #180 ``battery_recovered_percent``): a rise
|
||
that does not land ABOVE it is a bounce around the low floor, not a
|
||
swap — the chip kept offering "newer dates" for those.
|
||
"""
|
||
from itertools import pairwise
|
||
|
||
if rechargeable:
|
||
return None
|
||
for (_, v_prev), (ts, v) in pairwise(points):
|
||
if v - v_prev < _JUMP_MIN_RISE:
|
||
continue
|
||
if recovered is not None and v <= recovered:
|
||
continue # did not cross the recovery threshold: a bounce, not a swap
|
||
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], floor: float = float(NATIVE_LOW_PERCENT)) -> float:
|
||
"""The Battery-Notes-configured threshold OR the fleet floor — the higher.
|
||
|
||
``floor`` is the household's battery_low_percent (#146); the module
|
||
constant only remains as the fallback default.
|
||
"""
|
||
raw = attrs.get("battery_low_threshold")
|
||
if isinstance(raw, (int, float)):
|
||
return float(max(raw, floor))
|
||
return float(floor)
|
||
|
||
|
||
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]] = {}
|
||
recovered = float(get_battery_recovered_percent(hass))
|
||
|
||
for bat in batteries:
|
||
if bat.no_sensor or (bat.level is None and not bat.low):
|
||
continue # low-only binaries / sensorless notes 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), recovered=recovered
|
||
)
|
||
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):
|
||
# "Irreplaceable" / "Manual" / "Solar" describe the device, not a
|
||
# cell anyone stocks - no part, no reorder threshold. Same predicate
|
||
# as the overview's shopping groupings (is_shoppable_type).
|
||
if not is_shoppable_type(bat.battery_type):
|
||
continue
|
||
t = canonical_type(bat.battery_type)
|
||
totals[t] = totals.get(t, 0) + bat.quantity
|
||
return OrderedDict(sorted(totals.items()))
|
||
|
||
|
||
__all__ = [
|
||
"DEFAULT_HORIZON_DAYS",
|
||
"LOW_LATCH_KEY",
|
||
"NATIVE_LOW_PERCENT",
|
||
"TYPICAL_LIFETIME_MONTHS",
|
||
"Battery",
|
||
"BatteryOverview",
|
||
"apply_low_latch",
|
||
"async_compute_overview",
|
||
"async_level_history",
|
||
"async_trend_predictions",
|
||
"battery_notes_summary",
|
||
"build_overview",
|
||
"canonical_type",
|
||
"compute_overview",
|
||
"discover_battery_types",
|
||
"fleet_due_without_sensor",
|
||
"fleet_excluded_entities",
|
||
"fleet_included_entities",
|
||
"fleet_track_self_charging",
|
||
"get_battery_auto_record_recovery",
|
||
"get_battery_recovered_percent",
|
||
"has_batteries",
|
||
"has_battery_notes",
|
||
"is_rechargeable_type",
|
||
"lifetime_months",
|
||
"note_sibling_entity",
|
||
"read_batteries",
|
||
"release_low_latch",
|
||
]
|