From 2a103f52a53b0a795b7655265e53d0ab642da2a3 Mon Sep 17 00:00:00 2001 From: Darek Date: Wed, 10 Jun 2026 12:28:27 -0500 Subject: [PATCH] Update sub-component batteries from parent HA entity in poller and live sync --- app.py | 14 +++----------- ha_poller.py | 4 ++-- models.py | 7 +++++++ tests/test_ha_integration.py | 31 +++++++++++++++++++++++++++++++ 4 files changed, 43 insertions(+), 13 deletions(-) diff --git a/app.py b/app.py index 68659dd..d778b49 100644 --- a/app.py +++ b/app.py @@ -804,8 +804,8 @@ def create_app(config_object="config"): ha_live_pct = ha_client.get_state(device.ha_entity_id, timeout=1) if ha_live_pct is not None: changed = False - for battery in device.batteries: - if battery.status == "installed" and battery.battery_percentage != ha_live_pct: + for battery in device.installed_batteries(): + if battery.battery_percentage != ha_live_pct: battery.battery_percentage = ha_live_pct db.add(BatteryPctLog( battery_id=battery.id, @@ -1144,15 +1144,7 @@ def create_app(config_object="config"): return redirect(url_for("device_detail", device_id=device_id)) increment = 1 if request.form.get("increment_cycles") else 0 notes = request.form.get("notes", "").strip() or None - if device.has_children(): - installed = [ - b - for child in device.children - for b in child.batteries - if b.status == "installed" - ] - else: - installed = [b for b in device.batteries if b.status == "installed"] + installed = device.installed_batteries() if not installed: flash("No installed batteries to log.", "warning") return redirect(url_for("device_detail", device_id=device_id)) diff --git a/ha_poller.py b/ha_poller.py index 02e8c5e..92792bd 100644 --- a/ha_poller.py +++ b/ha_poller.py @@ -49,8 +49,8 @@ class HaPoller: for device in devices: pct = self._client.get_state(device.ha_entity_id) if pct is not None: - for battery in device.batteries: - if battery.status == "installed" and battery.battery_percentage != pct: + for battery in device.installed_batteries(): + if battery.battery_percentage != pct: battery.battery_percentage = pct session.add(BatteryPctLog( battery_id=battery.id, diff --git a/models.py b/models.py index 7308075..bd2b060 100644 --- a/models.py +++ b/models.py @@ -86,6 +86,13 @@ class Device(Base): return sum(c.battery_slots for c in self.children) return self.battery_slots + def installed_batteries(self): + """Installed batteries on this device, including its sub-components.""" + bats = [b for b in self.batteries if b.status == "installed"] + for child in self.children: + bats.extend(b for b in child.batteries if b.status == "installed") + return bats + def __repr__(self): return f"" diff --git a/tests/test_ha_integration.py b/tests/test_ha_integration.py index e5b7d6e..8f17bd8 100644 --- a/tests/test_ha_integration.py +++ b/tests/test_ha_integration.py @@ -182,6 +182,37 @@ def test_poll_updates_installed_batteries(ha_app, ha_client_f): mock_ha.get_state.assert_called_once_with("sensor.dev_a_battery") +def test_poll_updates_batteries_in_subcomponents(ha_app, ha_client_f): + """Batteries installed in sub-components are updated via the parent's HA entity.""" + ha_client_f.post("/device/add", data={"name": "RC Set", "battery_slots": "0", "battery_size": ""}) + ha_client_f.post("/device/add", data={"name": "Remote", "battery_slots": "2", + "battery_size": "AA", "parent_id": "1"}) + ha_client_f.post("/battery/add", data={"brand": "X", "count": "1", "size": "AA"}) + ha_client_f.post("/battery/1/assign", data={"device_id": "2"}) + ha_client_f.post("/device/1/edit", data={ + "name": "RC Set", "battery_slots": "0", "battery_size": "", + "ha_entity_id": "sensor.rc_set_battery" + }) + + from ha_client import HomeAssistantClient + from ha_poller import HaPoller + from models import Battery, BatteryPctLog + + mock_ha = MagicMock(spec=HomeAssistantClient) + mock_ha.enabled = True + mock_ha.get_state.return_value = 37 + + Session = _make_session_factory(ha_app) + HaPoller(mock_ha, Session, interval=300)._poll_once() + + s = Session() + b = s.get(Battery, 1) + assert b.battery_percentage == 37 + logs = s.query(BatteryPctLog).filter_by(battery_id=1, source="poll").all() + assert len(logs) == 1 and logs[0].percentage == 37 + s.close() + + def test_poll_skips_uninstalled_batteries(ha_app, ha_client_f): """Batteries that are available (not installed) are not updated by the poller.""" ha_client_f.post("/device/add", data={"name": "Dev B", "battery_slots": "1", "battery_size": "AA"})