@@ -6,11 +6,18 @@ 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 the single
``battery_plus`` sensor (device_class ``battery``): ``battery_type``,
``battery_quantity``, ``battery_low``, ``battery_low_threshold``,
``battery_last_replaced``. We read that one sensor kind — no dependency on the
(optional, often-disabled) battery-low binary.
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 wi th
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``
@@ -19,6 +26,8 @@ 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
@@ -27,6 +36,8 @@ 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;
@@ -103,6 +114,24 @@ def _norm_type(raw: Any) -> str:
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 )
@@ -122,6 +151,11 @@ class Battery:
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
@@ -166,6 +200,7 @@ def build_overview(
* ,
today : date ,
horizon_days : int = DEFAULT_HORIZON_DAYS ,
trend_predictions : dict [ str , tuple [ int , str ] ] | None = None ,
) - > BatteryOverview :
""" Aggregate batteries into the fleet view.
@@ -174,7 +209,8 @@ def build_overview(
``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× × " ).
grouping ( " 2× × " ). 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.
"""
@@ -184,20 +220,37 @@ def build_overview(
for bat in sorted ( batteries , key = lambda b : b . device_name . lower ( ) ) :
t = _norm_type ( bat . battery_type )
types_seen [ t ] = None
pred = _predicted_date ( bat )
days = ( pred - today ) . days if pred is not None else 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 : int | None = max ( 0 , trend [ 0 ] )
source , confidence = " trend " , trend [ 1 ]
else :
pred = None if rechargeable else _predicted_date ( bat )
days = ( pred - today ) . days if pred is not None else None
source , confidence = " typical " , None
if bat . low :
ov . low . append ( _row ( bat , t , None ) )
ov . needs_now [ t ] = ov . needs_now . get ( t , 0 ) + bat . quantity
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 ) , " status " : " low " } )
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 ) )
ov . needs_soon [ t ] = ov . needs_soon . get ( t , 0 ) + bat . quantity
ov . all . append ( { * * _row ( bat , t , days ) , " status " : " soon " } )
ov . soon . append ( _row ( bat , t , days , source , confidence , rechargeable = rechargeable ) )
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 ) , " status " : " soon " } )
continue
ov . all . append ( { * * _row ( bat , t , days ) , " status " : " ok " } )
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 )
@@ -206,7 +259,15 @@ def build_overview(
return ov
def _row ( bat : Battery , canon_type : str , days_until : int | None ) - > dict [ str , Any ] :
def _row (
bat : Battery ,
canon_type : str ,
days_until : int | None ,
predicted_source : str = " typical " ,
prediction_confidence : str | None = None ,
* ,
rechargeable : bool = False ,
) - > dict [ str , Any ] :
return {
" entity_id " : bat . entity_id ,
" device_name " : bat . device_name ,
@@ -215,6 +276,15 @@ def _row(bat: Battery, canon_type: str, days_until: int | None) -> dict[str, Any
" 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 ,
}
@@ -256,10 +326,15 @@ 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 (native pickup only — an explicit
Battery Notes note always wins): the device also has a vacuum/lawn_mower
entity, exposes a ``battery_charging`` binary, or is a Companion-app
phone/tablet (``mobile_app`` identifiers).
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
@@ -272,7 +347,10 @@ def _is_self_charging(hass: HomeAssistant, device_id: str | None) -> bool:
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 " :
if (
reg_entry . domain == " binary_sensor "
and ( reg_entry . device_class or reg_entry . original_device_class ) == " battery_charging "
) :
return True
return False
@@ -284,7 +362,17 @@ def read_batteries(hass: HomeAssistant) -> list[Battery]:
``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.
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 %-s ensor and/or a
battery-low binary) — plus %-s ensors matching the strict battery-name
heuristic for devices that ship no device class — grouped per device,
@@ -311,54 +399,97 @@ def read_batteries(hass: HomeAssistant) -> list[Battery]:
covered_devices : set [ str ] = set ( )
# ── Pass 1: Battery Notes battery_plus ──────────────────────────────────
for state in hass . states . async_all ( " sensor " ) :
attrs = state . attributes
if attrs . get ( " device_class " ) != " battery " or " battery_type " not in attrs :
continue
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' ow n 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 s tat e 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 )
reg = ent_reg . async_get ( state . entity_id )
if reg and reg . device_id :
covered_devices . add ( reg . device_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
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 " ,
# 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 percen tag e note counts as sibling coverage —
# kept, dropped or excluded: it s 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": ...}}
@@ -445,11 +576,16 @@ def read_batteries(hass: HomeAssistant) -> list[Battery]:
def has_battery_notes ( hass : HomeAssistant ) - > bool :
""" Whether the Battery Notes integration is present (any battery_plus). """
for state in hass . states . async_all ( " sensor " ) :
a = state . attributes
if a . get ( " device_class " ) == " battery " and " battery_type " in a :
return True
""" 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
@@ -461,16 +597,229 @@ def has_batteries(hass: HomeAssistant) -> bool:
def compute_overview ( hass : HomeAssistant , * , horizon_days : int = DEFAULT_HORIZON_DAYS ) - > BatteryOverview :
""" Read + aggregate in one call (HA-side entry point). """
""" 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 % f or 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. """
""" 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 ( ) ) )
@@ -481,12 +830,16 @@ __all__ = [
" 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 " ,
]
Home Assistant Version Control