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

224 files

Browse Source
This commit is contained in:
Home Assistant Version Control
2026-07-27 13:30:34 +00:00
parent b5723aa856
commit e36b8a1a22
224 changed files with 9967 additions and 2663 deletions
Download Patch File Download Diff File Expand all files Collapse all files
custom_components/maintenance_supporter/helpers/battery_fleet.py
+143 -18
View File
@@ -21,7 +21,7 @@ from __future__ import annotations
from collections import OrderedDict
from dataclasses import dataclass, field
from datetime import date
from datetime import date, timedelta
from typing import Any
from homeassistant.core import HomeAssistant
@@ -61,6 +61,41 @@ NATIVE_LOW_PERCENT = 20
# offline — that's exactly the one you must not hide).
_NO_READING = {"unavailable", "unknown", "none", ""}
# How long a NATIVE battery that was last seen LOW stays in the fleet after
# its entity goes unavailable. Battery Notes covers this case via its retained
# ``battery_low`` attribute; native entities have no equivalent, so without a
# snapshot the battery would vanish at the exact moment it died and took its
# device offline. Bounded so a permanently removed device eventually drops.
_NATIVE_RETENTION = timedelta(hours=48)
# Heuristic (sensors WITHOUT device_class): a %-sensor whose object_id talks
# about a battery — some Zigbee2MQTT/ESPHome devices ship battery levels
# without the device class. Deliberately strict: the exclusion words keep out
# charging electronics and home-storage state-of-charge sensors (a Powerwall
# is not a battery you replace).
_HEURISTIC_EXCLUDE = ("charging", "current", "power", "voltage", "energy", "load", "soc", "state_of_charge", "storage", "temp")
def _is_native_battery_sensor(state: Any) -> bool:
"""Whether a sensor state looks like a replaceable-battery level."""
attrs = state.attributes
if attrs.get("device_class") == "battery":
return True
if attrs.get("unit_of_measurement") != "%":
return False
object_id = state.entity_id.split(".", 1)[1]
if "battery" not in object_id:
return False
return not any(word in object_id for word in _HEURISTIC_EXCLUDE)
def _native_snapshot_cache(hass: HomeAssistant) -> dict[str, dict[str, Any]]:
"""Runtime cache of last-known native battery readings (per entity_id)."""
from ..const import DOMAIN
cache: dict[str, dict[str, Any]] = hass.data.setdefault(DOMAIN, {}).setdefault("battery_fleet_native_cache", {})
return cache
def _norm_type(raw: Any) -> str:
"""Canonicalize a battery-type label for grouping (upper, trimmed)."""
@@ -188,6 +223,46 @@ def _level_of(state_val: str) -> float | None:
return None
def fleet_excluded_entities(hass: HomeAssistant) -> set[str]:
"""Manually excluded battery entity_ids, stored on the fleet object entry.
Inlined lookup (not via battery_fleet_setup.find_fleet_entry) to keep this
module import-cycle-free — setup imports the aggregation, not vice versa.
"""
from ..const import CONF_OBJECT, DOMAIN
for entry in hass.config_entries.async_entries(DOMAIN):
obj = entry.data.get(CONF_OBJECT, {})
if obj.get("battery_fleet"):
return set(obj.get("battery_fleet_excluded") or [])
return set()
def _is_self_charging(hass: HomeAssistant, device_id: str | None) -> bool:
"""Whether a device recharges itself — its battery is never REPLACED.
Issue #107: a Roborock's native battery sensor reads "low" mid-clean, but
nobody swaps its cells. Heuristics (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).
"""
if not device_id:
return False
from homeassistant.helpers import device_registry as dr
from homeassistant.helpers import entity_registry as er
device = dr.async_get(hass).async_get(device_id)
if device and any(domain == "mobile_app" for domain, _ in device.identifiers):
return True
for reg_entry in er.async_entries_for_device(er.async_get(hass), device_id, include_disabled_entities=True):
if reg_entry.domain in ("vacuum", "lawn_mower"):
return True
if reg_entry.domain == "binary_sensor" and (reg_entry.device_class or reg_entry.original_device_class) == "battery_charging":
return True
return False
def read_batteries(hass: HomeAssistant) -> list[Battery]:
"""Read the battery fleet from HA state — Battery Notes AND native.
@@ -197,16 +272,25 @@ def read_batteries(hass: HomeAssistant) -> list[Battery]:
unavailable/unknown but RETAINS its last-known ``battery_low`` — so a
dead battery that took its device offline stays visible.
* **Native** ``device_class: battery`` entities (a %-sensor and/or a
battery-low binary), grouped per device, give a degraded view (type
"Unknown", quantity 1, no forecast). A device already covered by a
Battery Notes note is skipped (dedup by the note's source entity + its
device) so it isn't counted twice.
battery-low binary) — plus %-sensors matching the strict battery-name
heuristic for devices that ship no device class — grouped per device,
give a degraded view (type "Unknown", quantity 1, no forecast). A
device already covered by a Battery Notes note is skipped (dedup by
the note's source entity + its device) so it isn't counted twice;
self-charging devices (vacuums, mowers, phones — see
:func:`_is_self_charging`) are skipped entirely. A native battery
last seen LOW that goes unavailable is retained from a runtime
snapshot for ``_NATIVE_RETENTION`` (the Battery Notes path gets this
for free via its retained ``battery_low`` attribute).
* Manually excluded entity_ids (fleet detail → exclude) are dropped from
BOTH passes.
"""
from homeassistant.helpers import device_registry as dr
from homeassistant.helpers import entity_registry as er
ent_reg = er.async_get(hass)
dev_reg = dr.async_get(hass)
excluded = fleet_excluded_entities(hass)
out: list[Battery] = []
covered_sources: set[str] = set()
@@ -217,17 +301,36 @@ def read_batteries(hass: HomeAssistant) -> list[Battery]:
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' 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)
reg = ent_reg.async_get(state.entity_id)
if reg and reg.device_id:
covered_devices.add(reg.device_id)
level = _level_of(state.state)
available = state.state not in _NO_READING and level is not None
low = bool(attrs.get("battery_low"))
# Offline AND not last-known-low = pure connectivity noise → drop it.
if not available and not low:
# 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(
@@ -237,7 +340,7 @@ def read_batteries(hass: HomeAssistant) -> list[Battery]:
quantity=int(attrs.get("battery_quantity") or 1),
low=low,
level=level,
last_replaced=_parse_last_replaced(attrs.get("battery_last_replaced")),
last_replaced=last_replaced,
available=available,
source="battery_notes",
)
@@ -248,17 +351,25 @@ def read_batteries(hass: HomeAssistant) -> list[Battery]:
native: dict[str, dict[str, Any]] = {}
for domain in ("sensor", "binary_sensor"):
for state in hass.states.async_all(domain):
if state.attributes.get("device_class") != "battery":
# Sensors: device_class battery OR the strict name/% heuristic
# (Zigbee2MQTT/ESPHome levels without a device class). Binaries:
# device_class only — name-guessing booleans is too risky.
if domain == "sensor":
if not _is_native_battery_sensor(state):
continue
elif state.attributes.get("device_class") != "battery":
continue
eid = state.entity_id
if "battery_type" in state.attributes: # Battery Notes battery_plus — handled above
continue
if eid in covered_sources:
if eid in covered_sources or eid in excluded:
continue
reg = ent_reg.async_get(eid)
dev_id = reg.device_id if reg else None
if dev_id and dev_id in covered_devices:
continue
if _is_self_charging(hass, dev_id): # #107: vacuums/mowers/phones
continue
key = dev_id or eid
rec = native.setdefault(
key,
@@ -273,6 +384,8 @@ def read_batteries(hass: HomeAssistant) -> list[Battery]:
if rec["name"] is None and friendly:
rec["name"] = friendly
snapshot_cache = _native_snapshot_cache(hass)
now = dt_util.utcnow()
for rec in native.values():
level = _level_of(rec["level_state"]) if rec["level_state"] is not None else None
low_state = rec["low_state"]
@@ -283,6 +396,19 @@ def read_batteries(hass: HomeAssistant) -> list[Battery]:
low = low_available and str(low_state).lower() in ("on", "true", "1")
else:
low = level is not None and level <= NATIVE_LOW_PERCENT
if available:
# Remember the last real reading — the retention path below needs
# it once the entity goes unavailable.
snapshot_cache[rec["eid"]] = {"low": low, "level": level, "ts": now}
elif not low:
# Native dead-battery retention: an entity that was LOW and then
# went unavailable (the battery died and took the device offline)
# stays visible for _NATIVE_RETENTION instead of vanishing at the
# exact moment it needs replacing.
snap = snapshot_cache.get(rec["eid"])
if snap and snap.get("low") and now - snap["ts"] <= _NATIVE_RETENTION:
low = True
level = snap.get("level")
if not available and not low:
continue
name = rec["name"]
@@ -315,11 +441,9 @@ def has_battery_notes(hass: HomeAssistant) -> bool:
def has_batteries(hass: HomeAssistant) -> bool:
"""Whether ANY battery is trackable — Battery Notes OR native. Gates setup."""
for domain in ("sensor", "binary_sensor"):
for state in hass.states.async_all(domain):
if state.attributes.get("device_class") == "battery":
return True
return False
if any(_is_native_battery_sensor(s) for s in hass.states.async_all("sensor")):
return True
return any(s.attributes.get("device_class") == "battery" for s in hass.states.async_all("binary_sensor"))
def compute_overview(hass: HomeAssistant, *, horizon_days: int = DEFAULT_HORIZON_DAYS) -> BatteryOverview:
@@ -346,6 +470,7 @@ __all__ = [
"build_overview",
"compute_overview",
"discover_battery_types",
"fleet_excluded_entities",
"has_batteries",
"has_battery_notes",
"lifetime_months",