francop/HomeAssistantVS
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

278 files

Browse Source
This commit is contained in:
Home Assistant Version Control committed 2026-09-13 20:38:58 +00:00
1 parent d7a0a08372
commit 1b4f5f68c6
278 files changed
+36892 -7150
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

custom_components/maintenance_supporter/helpers/battery_fleet.py
+121 -68
View File
@@ -32,6 +32,7 @@ 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
@@ -41,26 +42,19 @@ 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
# 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
@@ -152,33 +146,12 @@ def _native_snapshot_cache(hass: HomeAssistant) -> dict[str, dict[str, Any]]:
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)
"""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
@@ -210,6 +183,11 @@ class Battery:
# 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 = ""
@dataclass
@@ -237,10 +215,11 @@ class BatteryOverview:
return len(self.low)
def _predicted_date(bat: Battery) -> date | None:
def _predicted_date(bat: Battery, months: int | None = None) -> date | None:
if bat.last_replaced is None:
return None
months = lifetime_months(bat.battery_type)
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
@@ -255,9 +234,15 @@ def build_overview(
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`).
@@ -274,9 +259,17 @@ def build_overview(
types_seen: OrderedDict[str, None] = OrderedDict()
for bat in sorted(batteries, key=lambda b: b.device_name.lower()):
t = _norm_type(bat.battery_type)
# 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
@@ -286,11 +279,14 @@ def build_overview(
# "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:
pred = None if rechargeable else _predicted_date(bat)
# 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
@@ -308,21 +304,21 @@ def build_overview(
# 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)
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)
low_row = _row(bat, t, None, rechargeable=rechargeable, lifetime=info)
ov.low.append(low_row)
if not rechargeable:
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))
if not rechargeable:
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), "status": "soon"})
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), "status": "ok"})
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)
@@ -340,9 +336,17 @@ def _row(
*,
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,
@@ -371,13 +375,27 @@ def _row(
}
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:
return parsed.date()
# 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
@@ -454,6 +472,26 @@ def fleet_due_without_sensor(hass: HomeAssistant) -> bool:
return _fleet_object(hass).get(BATTERY_FLEET_DUE_WITHOUT_SENSOR) is not False
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.
@@ -675,7 +713,8 @@ def read_batteries(hass: HomeAssistant) -> list[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),
quantity=_quantity_of(attrs),
model_key=device_model_key(hass, dev_id),
low=low,
level=level,
last_replaced=last_replaced,
@@ -779,6 +818,7 @@ def read_batteries(hass: HomeAssistant) -> list[Battery]:
device_name=name or rec["eid"],
battery_type="Rechargeable" if rec.get("self_charging") else "Unknown",
quantity=1,
model_key=device_model_key(hass, rec.get("device_id")),
low=low,
level=level,
last_replaced=None,
@@ -800,6 +840,11 @@ def read_batteries(hass: HomeAssistant) -> list[Battery]:
# 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"]}
for state in hass.states.async_all("sensor"):
if not _is_type_note(state):
@@ -853,7 +898,8 @@ def read_batteries(hass: HomeAssistant) -> list[Battery]:
entity_id=eid,
device_name=name,
battery_type=battery_type,
quantity=int(attrs.get("battery_quantity") or 1),
quantity=_quantity_of(attrs),
model_key=device_model_key(hass, dev_id),
low=False,
level=None,
last_replaced=last_replaced,
@@ -864,8 +910,8 @@ def read_batteries(hass: HomeAssistant) -> list[Battery]:
)
# Due = the forecast has PASSED (same arithmetic as build_overview's
# forecast_overdue). Rechargeables never get a table forecast.
if due_without_sensor and not is_rechargeable_type(battery_type):
pred = _predicted_date(bat)
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)
for bat in out:
@@ -973,7 +1019,9 @@ def compute_overview(hass: HomeAssistant, *, horizon_days: int = DEFAULT_HORIZON
:func:`async_compute_overview` instead.
"""
today = dt_util.now().date()
return build_overview(read_batteries(hass), today=today, horizon_days=horizon_days)
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) ───────────────────────────────
@@ -1048,8 +1096,11 @@ async def async_trend_predictions(hass: HomeAssistant, batteries: list[Battery])
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)
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 ────────────────────────────────
@@ -1188,11 +1239,12 @@ def discover_battery_types(hass: HomeAssistant) -> OrderedDict[str, int]:
"""
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":
# "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()))
@@ -1208,6 +1260,7 @@ __all__ = [
"async_trend_predictions",
"battery_notes_summary",
"build_overview",
"canonical_type",
"compute_overview",
"discover_battery_types",
"fleet_due_without_sensor",