From 2c357535006a8d8177b548d001995c1d9cea4f86 Mon Sep 17 00:00:00 2001 From: Mark Gascoyne Date: Thu, 2 Jul 2026 22:55:04 +0100 Subject: [PATCH 01/14] feat(teslemetry): Tesla Powerwall component - data path --- .cspell/custom-dictionary-workspace.txt | 1 + apps/predbat/teslemetry.py | 161 ++++++++++++++++++++++++ apps/predbat/tests/test_teslemetry.py | 141 +++++++++++++++++++++ apps/predbat/unit_test.py | 2 + 4 files changed, 305 insertions(+) create mode 100644 apps/predbat/teslemetry.py create mode 100644 apps/predbat/tests/test_teslemetry.py diff --git a/.cspell/custom-dictionary-workspace.txt b/.cspell/custom-dictionary-workspace.txt index ef4ded9fb..4c7c33ff2 100644 --- a/.cspell/custom-dictionary-workspace.txt +++ b/.cspell/custom-dictionary-workspace.txt @@ -396,6 +396,7 @@ tdata tdiff temperation Teslemetry +teslemetry testname thirdparty timea diff --git a/apps/predbat/teslemetry.py b/apps/predbat/teslemetry.py new file mode 100644 index 000000000..601012d30 --- /dev/null +++ b/apps/predbat/teslemetry.py @@ -0,0 +1,161 @@ +# ----------------------------------------------------------------------------- +# Predbat Home Battery System +# Copyright Trefor Southwell 2026 - All Rights Reserved +# This application maybe used for personal use only and not for commercial use +# ----------------------------------------------------------------------------- +"""Tesla Powerwall integration via the Teslemetry API. + +Teslemetry (https://api.teslemetry.com) proxies the Tesla Fleet API using the +same URI paths, authenticated with a static customer-supplied token (BYOK). +The base URL is configurable so a direct Fleet API connection can be swapped +in later without changing this component. + +Data path: polls live_status (power flows + SOC), site_info (capacity) and +calendar_history (daily energy) and publishes predbat_teslemetry_* sensors. +Control path: exposes virtual select/number/switch entities driven by the +TESLA service hooks; handlers issue REST commands (operation mode, backup +reserve, grid import/export rules and the export tariff-trick). +""" + +import asyncio + +import aiohttp + +from component_base import ComponentBase + +TESLEMETRY_DEFAULT_URL = "https://api.teslemetry.com" +TESLEMETRY_TIMEOUT = 30 +TESLEMETRY_RETRIES = 3 +LIVE_POLL_SECONDS = 120 +ENERGY_POLL_SECONDS = 300 + +OPERATION_MODES = ["self_consumption", "autonomous", "backup"] +EXPORT_RULES = ["never", "pv_only", "battery_ok"] +TARIFF_MODES = ["normal", "export_now"] + + +class TeslemetryAPI(ComponentBase): + """Tesla Powerwall component using the Teslemetry (or Fleet) REST API.""" + + def initialize(self, key="", site_id="", base_url=TESLEMETRY_DEFAULT_URL, **kwargs): + """Initialise the Teslemetry component from configuration. + + Args: + key: Teslemetry (or Fleet API) bearer token. + site_id: Tesla energy site id to poll and control. + base_url: REST API base URL (Teslemetry by default, swappable for a direct Fleet API connection). + kwargs: Reserved for future control-path configuration (accepted and ignored here). + """ + self.api_key = key + self.site_id = str(site_id) if site_id else "" + self.base_url = base_url + self.api_auth_failed = False + self.last_live_poll = 0 + self.last_energy_poll = 0 + self.log("Info: TeslemetryAPI initialising site_id={}".format(self.site_id)) + + def entity(self, suffix, domain="sensor"): + """Build a prefixed virtual entity id for this component.""" + return "{}.{}_teslemetry_{}".format(domain, self.prefix, suffix) + + async def _request(self, method, path, json_body=None): + """Make a Teslemetry REST request, returning parsed JSON or None on failure.""" + url = "{}{}".format(self.base_url, path) + headers = {"Authorization": "Bearer {}".format(self.api_key), "Content-Type": "application/json"} + timeout = aiohttp.ClientTimeout(total=TESLEMETRY_TIMEOUT) + for attempt in range(TESLEMETRY_RETRIES): + try: + async with aiohttp.ClientSession(timeout=timeout) as session: + async with session.request(method, url, headers=headers, json=json_body) as resp: + if resp.status in (401, 403): + self.api_auth_failed = True + self.log("Warn: Teslemetry auth failed ({}) on {} - token revoked or subscription lapsed".format(resp.status, path)) + return None + if resp.status == 429: + self.log("Warn: Teslemetry rate limited on {} attempt {}".format(path, attempt + 1)) + await asyncio.sleep(2 ** (attempt + 1)) + continue + if resp.status >= 400: + body = await resp.text() + self.log("Warn: Teslemetry HTTP {} on {}: {}".format(resp.status, path, body[:200])) + return None + self.api_auth_failed = False + return await resp.json() + except (aiohttp.ClientError, asyncio.TimeoutError) as err: + self.log("Warn: Teslemetry network error on {} attempt {}: {}".format(path, attempt + 1, err)) + await asyncio.sleep(2**attempt) + return None + + def publish_sensor(self, suffix, state, unit=None, state_class="measurement", friendly=None): + """Publish one virtual sensor via the dashboard.""" + attributes = {"friendly_name": friendly or suffix.replace("_", " ").title(), "state_class": state_class} + if unit: + attributes["unit_of_measurement"] = unit + self.dashboard_item(self.entity(suffix), state, attributes, app="teslemetry") + + async def fetch_live_status(self): + """Fetch live power flows and SOC, publishing power sensors.""" + data = await self._request("GET", "/api/1/energy_sites/{}/live_status".format(self.site_id)) + if not data: + return False + response = data.get("response", {}) + self.publish_sensor("soc", response.get("percentage_charged", 0), unit="%", friendly="Powerwall SOC") + self.publish_sensor("battery_power", response.get("battery_power", 0), unit="W", friendly="Powerwall Battery Power") + self.publish_sensor("grid_power", response.get("grid_power", 0), unit="W", friendly="Powerwall Grid Power") + self.publish_sensor("load_power", response.get("load_power", 0), unit="W", friendly="Powerwall Load Power") + self.publish_sensor("solar_power", response.get("solar_power", 0), unit="W", friendly="Powerwall Solar Power") + self.update_success_timestamp() + return True + + async def fetch_site_info(self): + """Fetch site info, publishing capacity and seeding control entity states.""" + data = await self._request("GET", "/api/1/energy_sites/{}/site_info".format(self.site_id)) + if not data: + return False + response = data.get("response", {}) + nameplate_wh = response.get("nameplate_energy", 0) + if nameplate_wh: + self.publish_sensor("soc_max", round(nameplate_wh / 1000.0, 2), unit="kWh", state_class=None, friendly="Powerwall Capacity") + return True + + async def fetch_energy_today(self): + """Fetch today's cumulative energy, publishing daily kWh sensors.""" + data = await self._request("GET", "/api/1/energy_sites/{}/calendar_history".format(self.site_id)) + if not data: + return False + series = data.get("response", {}).get("time_series", []) + solar = grid_import = grid_export = load = 0.0 + for entry in series: + solar += entry.get("solar_energy_exported", 0) + grid_import += entry.get("grid_energy_imported", 0) + grid_export += entry.get("grid_energy_exported_from_solar", 0) + entry.get("grid_energy_exported_from_battery", 0) + entry.get("grid_energy_exported_from_generator", 0) + load += entry.get("consumer_energy_imported_from_grid", 0) + entry.get("consumer_energy_imported_from_solar", 0) + entry.get("consumer_energy_imported_from_battery", 0) + entry.get("consumer_energy_imported_from_generator", 0) + self.publish_sensor("solar_today", round(solar / 1000.0, 3), unit="kWh", state_class="total_increasing", friendly="Powerwall Solar Today") + self.publish_sensor("import_today", round(grid_import / 1000.0, 3), unit="kWh", state_class="total_increasing", friendly="Powerwall Import Today") + self.publish_sensor("export_today", round(grid_export / 1000.0, 3), unit="kWh", state_class="total_increasing", friendly="Powerwall Export Today") + self.publish_sensor("load_today", round(load / 1000.0, 3), unit="kWh", state_class="total_increasing", friendly="Powerwall Load Today") + return True + + async def run(self, seconds=0, first=False): + """Main component loop: poll data at the configured cadences.""" + if not self.api_key or not self.site_id: + self.log("Warn: Teslemetry key or site_id not configured, skipping run") + return + if first: + await self.fetch_site_info() + if self.api_auth_failed: + # Do not hammer the API with a dead token; only live_status probes recovery + if seconds - self.last_live_poll >= LIVE_POLL_SECONDS: + self.last_live_poll = seconds + await self.fetch_live_status() + return + if first or (seconds - self.last_live_poll >= LIVE_POLL_SECONDS): + self.last_live_poll = seconds + await self.fetch_live_status() + if first or (seconds - self.last_energy_poll >= ENERGY_POLL_SECONDS): + self.last_energy_poll = seconds + await self.fetch_energy_today() + + async def final(self): + """Cleanup on shutdown.""" + self.log("Info: TeslemetryAPI shutdown") diff --git a/apps/predbat/tests/test_teslemetry.py b/apps/predbat/tests/test_teslemetry.py new file mode 100644 index 000000000..57f71c657 --- /dev/null +++ b/apps/predbat/tests/test_teslemetry.py @@ -0,0 +1,141 @@ +# ----------------------------------------------------------------------------- +# Predbat Home Battery System +# Copyright Trefor Southwell 2026 - All Rights Reserved +# This application maybe used for personal use only and not for commercial use +# ----------------------------------------------------------------------------- +"""Unit tests for the TeslemetryAPI component (Tesla Powerwall via Teslemetry).""" + +from tests.test_infra import run_async +from teslemetry import TeslemetryAPI + + +class MockTeslemetryAPI(TeslemetryAPI): + """TeslemetryAPI test double that avoids ComponentBase initialisation.""" + + def __init__(self): + """Initialise the mock with captured state instead of ComponentBase wiring.""" + self.api_key = "test_token" + self.site_id = "123456" + self.base_url = "https://api.teslemetry.com" + self.prefix = "predbat" + self.api_auth_failed = False + self.dashboard_items = {} + self.log_messages = [] + self.entity_states = {} + self.mock_responses = {} + self.requests_made = [] + + def log(self, msg): + """Capture log messages.""" + self.log_messages.append(msg) + + def dashboard_item(self, entity, state, attributes, app=None): + """Capture dashboard items.""" + self.dashboard_items[entity] = {"state": state, "attributes": attributes} + + def set_state_wrapper(self, entity_id, state, attributes={}): + """Capture entity state updates.""" + self.entity_states[entity_id] = state + + def get_state_wrapper(self, entity_id=None, default=None, **kwargs): + """Return captured entity state.""" + return self.entity_states.get(entity_id, default) + + def update_success_timestamp(self): + """No-op for tests.""" + pass + + async def _request(self, method, path, json_body=None): + """Return canned responses instead of real HTTP.""" + self.requests_made.append((method, path, json_body)) + return self.mock_responses.get(path, None) + + +LIVE_STATUS = { + "response": { + "percentage_charged": 55.5, + "battery_power": 1200, + "solar_power": 800, + "load_power": 600, + "grid_power": -1400, + "island_status": "on_grid", + } +} + +SITE_INFO = { + "response": { + "nameplate_energy": 13500, + "default_real_mode": "self_consumption", + "backup_reserve_percent": 20, + } +} + +ENERGY_HISTORY = { + "response": { + "time_series": [ + { + "solar_energy_exported": 4000, + "grid_energy_imported": 2500, + "grid_energy_exported_from_solar": 1000, + "grid_energy_exported_from_battery": 500, + "grid_energy_exported_from_generator": 0, + "consumer_energy_imported_from_grid": 2000, + "consumer_energy_imported_from_solar": 1500, + "consumer_energy_imported_from_battery": 700, + "consumer_energy_imported_from_generator": 0, + } + ] + } +} + + +def test_teslemetry_entity_names(): + """Entity helper builds prefixed ids.""" + api = MockTeslemetryAPI() + assert api.entity("soc") == "sensor.predbat_teslemetry_soc" + assert api.entity("operation_mode", domain="select") == "select.predbat_teslemetry_operation_mode" + + +def test_teslemetry_live_status_publishes_sensors(): + """live_status data is published as power/SOC sensors.""" + api = MockTeslemetryAPI() + api.mock_responses["/api/1/energy_sites/123456/live_status"] = LIVE_STATUS + run_async(api.fetch_live_status()) + assert api.dashboard_items["sensor.predbat_teslemetry_soc"]["state"] == 55.5 + assert api.dashboard_items["sensor.predbat_teslemetry_battery_power"]["state"] == 1200 + assert api.dashboard_items["sensor.predbat_teslemetry_grid_power"]["state"] == -1400 + assert api.dashboard_items["sensor.predbat_teslemetry_solar_power"]["state"] == 800 + assert api.dashboard_items["sensor.predbat_teslemetry_load_power"]["state"] == 600 + + +def test_teslemetry_site_info_publishes_soc_max(): + """site_info nameplate energy is published as soc_max in kWh.""" + api = MockTeslemetryAPI() + api.mock_responses["/api/1/energy_sites/123456/site_info"] = SITE_INFO + run_async(api.fetch_site_info()) + assert api.dashboard_items["sensor.predbat_teslemetry_soc_max"]["state"] == 13.5 + + +def test_teslemetry_energy_today_publishes_kwh(): + """calendar_history energy series is aggregated into daily kWh sensors.""" + api = MockTeslemetryAPI() + api.mock_responses["/api/1/energy_sites/123456/calendar_history"] = ENERGY_HISTORY + run_async(api.fetch_energy_today()) + assert api.dashboard_items["sensor.predbat_teslemetry_solar_today"]["state"] == 4.0 + assert api.dashboard_items["sensor.predbat_teslemetry_import_today"]["state"] == 2.5 + assert api.dashboard_items["sensor.predbat_teslemetry_export_today"]["state"] == 1.5 + assert api.dashboard_items["sensor.predbat_teslemetry_load_today"]["state"] == 4.2 + + +def test_teslemetry(my_predbat=None): + """Run all Teslemetry component tests (registry entry point). + + Args: + my_predbat: Unused; accepted for compatibility with the TEST_REGISTRY calling convention in unit_test.py. + """ + test_teslemetry_entity_names() + test_teslemetry_live_status_publishes_sensors() + test_teslemetry_site_info_publishes_soc_max() + test_teslemetry_energy_today_publishes_kwh() + print("**** Teslemetry tests passed ****") + return 0 diff --git a/apps/predbat/unit_test.py b/apps/predbat/unit_test.py index bcb03096f..8738d1230 100644 --- a/apps/predbat/unit_test.py +++ b/apps/predbat/unit_test.py @@ -50,6 +50,7 @@ from tests.test_saving_session import test_saving_session, test_saving_session_null_octopoints, test_saving_session_notify_config, test_saving_session_default_rate, test_saving_session_axle_conflict, test_saving_session_auto_join_toggle from tests.test_secrets import run_secrets_tests from tests.test_ge_cloud import test_ge_cloud +from tests.test_teslemetry import test_teslemetry from tests.test_compare import test_compare from tests.test_gateway import run_gateway_tests from tests.test_axle import test_axle @@ -289,6 +290,7 @@ def main(): ("balance_inverters", run_balance_inverters_tests, "Balance inverters tests", False), # GE Cloud unit tests ("ge_cloud", test_ge_cloud, "GE Cloud comprehensive tests (API, devices, EVC, inverter ops, events, publishing, config, downloads, cache)", False), + ("teslemetry", test_teslemetry, "Teslemetry Tesla Powerwall component tests (data path, control, tariff)", False), ("integer_config", test_integer_config_entities, "Integer config entities tests", False), ("validate_config", test_validate_config, "APPS_SCHEMA validator tests (string types, sensor boolean states)", False), ("expose_config_integer", test_expose_config_preserves_integer, "Expose config preserves integer tests", False), From 3424893ecbcfa2bd29704971bb1b5278c4c6ddb0 Mon Sep 17 00:00:00 2001 From: Mark Gascoyne Date: Thu, 2 Jul 2026 23:20:16 +0100 Subject: [PATCH 02/14] fix(teslemetry): run() success contract + _request/run tests --- apps/predbat/teslemetry.py | 27 +++-- apps/predbat/tests/test_teslemetry.py | 145 +++++++++++++++++++++++++- 2 files changed, 162 insertions(+), 10 deletions(-) diff --git a/apps/predbat/teslemetry.py b/apps/predbat/teslemetry.py index 601012d30..449316ee7 100644 --- a/apps/predbat/teslemetry.py +++ b/apps/predbat/teslemetry.py @@ -108,7 +108,7 @@ async def fetch_live_status(self): return True async def fetch_site_info(self): - """Fetch site info, publishing capacity and seeding control entity states.""" + """Fetch site info, publishing the battery capacity as soc_max.""" data = await self._request("GET", "/api/1/energy_sites/{}/site_info".format(self.site_id)) if not data: return False @@ -137,24 +137,33 @@ async def fetch_energy_today(self): return True async def run(self, seconds=0, first=False): - """Main component loop: poll data at the configured cadences.""" + """Main component loop: poll data at the configured cadences. + + Returns True on a successful (or gated no-op) cycle and False on failure — + ComponentBase.start() uses this to set api_started, clear the startup flag + and drive its retry backoff, so a failing cycle must return False. + """ if not self.api_key or not self.site_id: self.log("Warn: Teslemetry key or site_id not configured, skipping run") - return - if first: - await self.fetch_site_info() + return False if self.api_auth_failed: - # Do not hammer the API with a dead token; only live_status probes recovery + # Do not hammer the API with a dead token; only live_status probes recovery. + # A successful probe clears the flag in _request; a failed one returns False + # so ComponentBase keeps backing off. if seconds - self.last_live_poll >= LIVE_POLL_SECONDS: self.last_live_poll = seconds - await self.fetch_live_status() - return + return await self.fetch_live_status() + return False + success = True + if first: + await self.fetch_site_info() if first or (seconds - self.last_live_poll >= LIVE_POLL_SECONDS): self.last_live_poll = seconds - await self.fetch_live_status() + success = await self.fetch_live_status() if first or (seconds - self.last_energy_poll >= ENERGY_POLL_SECONDS): self.last_energy_poll = seconds await self.fetch_energy_today() + return success async def final(self): """Cleanup on shutdown.""" diff --git a/apps/predbat/tests/test_teslemetry.py b/apps/predbat/tests/test_teslemetry.py index 57f71c657..fa9f5bcd6 100644 --- a/apps/predbat/tests/test_teslemetry.py +++ b/apps/predbat/tests/test_teslemetry.py @@ -5,7 +5,9 @@ # ----------------------------------------------------------------------------- """Unit tests for the TeslemetryAPI component (Tesla Powerwall via Teslemetry).""" -from tests.test_infra import run_async +from unittest.mock import MagicMock, patch, AsyncMock + +from tests.test_infra import create_aiohttp_mock_response, create_aiohttp_mock_session, run_async from teslemetry import TeslemetryAPI @@ -19,6 +21,8 @@ def __init__(self): self.base_url = "https://api.teslemetry.com" self.prefix = "predbat" self.api_auth_failed = False + self.last_live_poll = 0 + self.last_energy_poll = 0 self.dashboard_items = {} self.log_messages = [] self.entity_states = {} @@ -127,6 +131,139 @@ def test_teslemetry_energy_today_publishes_kwh(): assert api.dashboard_items["sensor.predbat_teslemetry_load_today"]["state"] == 4.2 +def _make_real_request_api(): + """Build a MockTeslemetryAPI with the REAL TeslemetryAPI._request bound back in place of the canned override.""" + api = MockTeslemetryAPI() + api._request = TeslemetryAPI._request.__get__(api) + return api + + +def test_teslemetry_request_auth_failure(): + """A 401 response makes the real _request return None and latch api_auth_failed.""" + + async def test(): + """Drive the real _request against a mocked 401 response.""" + api = _make_real_request_api() + mock_response = create_aiohttp_mock_response(status=401, json_data={"error": "unauthorised"}) + mock_session = create_aiohttp_mock_session(mock_response) + mock_session.request = mock_session.get + with patch("aiohttp.ClientSession") as mock_session_class: + with patch("teslemetry.asyncio.sleep", new_callable=AsyncMock): + mock_session_class.return_value = mock_session + result = await api._request("GET", "/api/1/energy_sites/123456/live_status") + assert result is None + assert api.api_auth_failed is True + assert mock_session.request.call_count == 1 # No retries on auth failure + assert any("auth failed" in msg for msg in api.log_messages) + + run_async(test()) + + +def test_teslemetry_request_rate_limit_retry(): + """A 429 response is retried with backoff and the follow-up 200 JSON is returned.""" + + async def test(): + """Drive the real _request against a mocked 429-then-200 response sequence.""" + api = _make_real_request_api() + resp_429 = create_aiohttp_mock_response(status=429, json_data={"error": "rate limited"}) + resp_200 = create_aiohttp_mock_response(status=200, json_data={"response": {"ok": 1}}) + mock_session = create_aiohttp_mock_session(resp_200) + mock_session.request = MagicMock(side_effect=[resp_429, resp_200]) + with patch("aiohttp.ClientSession") as mock_session_class: + with patch("teslemetry.asyncio.sleep", new_callable=AsyncMock) as mock_sleep: + mock_session_class.return_value = mock_session + result = await api._request("GET", "/api/1/energy_sites/123456/site_info") + assert result == {"response": {"ok": 1}} + assert mock_session.request.call_count == 2 + assert mock_sleep.await_count == 1 + assert api.api_auth_failed is False + + run_async(test()) + + +def test_teslemetry_request_success_clears_auth_flag(): + """A 200 response clears a previously latched api_auth_failed flag.""" + + async def test(): + """Drive the real _request against a mocked 200 response with the auth flag pre-latched.""" + api = _make_real_request_api() + api.api_auth_failed = True + mock_response = create_aiohttp_mock_response(status=200, json_data={"response": {"percentage_charged": 50}}) + mock_session = create_aiohttp_mock_session(mock_response) + mock_session.request = mock_session.get + with patch("aiohttp.ClientSession") as mock_session_class: + with patch("teslemetry.asyncio.sleep", new_callable=AsyncMock): + mock_session_class.return_value = mock_session + result = await api._request("GET", "/api/1/energy_sites/123456/live_status") + assert result == {"response": {"percentage_charged": 50}} + assert api.api_auth_failed is False + + run_async(test()) + + +def _make_run_api_with_fetch_capture(site_info=True, live_status=True, energy_today=True): + """Build a MockTeslemetryAPI whose fetch methods are stubbed to record calls, returning (api, calls list).""" + api = MockTeslemetryAPI() + calls = [] + + async def fake_site_info(): + """Record a site_info fetch.""" + calls.append("site_info") + return site_info + + async def fake_live_status(): + """Record a live_status fetch.""" + calls.append("live_status") + return live_status + + async def fake_energy_today(): + """Record an energy_today fetch.""" + calls.append("energy_today") + return energy_today + + api.fetch_site_info = fake_site_info + api.fetch_live_status = fake_live_status + api.fetch_energy_today = fake_energy_today + return api, calls + + +def test_teslemetry_run_unconfigured_returns_false(): + """run() returns False (not None) when key or site_id is missing, so ComponentBase treats it as a failed cycle.""" + api, calls = _make_run_api_with_fetch_capture() + api.api_key = "" + assert run_async(api.run(0, True)) is False + assert calls == [] + api = MockTeslemetryAPI() + api.site_id = "" + assert run_async(api.run(0, True)) is False + + +def test_teslemetry_run_first_success_returns_true(): + """A successful first run fetches everything and returns True; a gated-out steady-state cycle is a successful no-op.""" + api, calls = _make_run_api_with_fetch_capture() + assert run_async(api.run(0, True)) is True + assert calls == ["site_info", "live_status", "energy_today"] + # Gated-out 60s tick: nothing due yet, but the cycle itself is healthy + calls.clear() + assert run_async(api.run(60, False)) is True + assert calls == [] + # Failed live_status poll propagates as a failed cycle + api, calls = _make_run_api_with_fetch_capture(live_status=False) + assert run_async(api.run(0, True)) is False + + +def test_teslemetry_run_auth_failed_only_probes_live_status(): + """While auth-failed, run() probes ONLY live_status on the poll cadence (no site_info/energy, even when first).""" + api, calls = _make_run_api_with_fetch_capture() + api.api_auth_failed = True + # Gated-out tick: no API traffic at all, cycle reports failure to keep backoff + assert run_async(api.run(60, False)) is False + assert calls == [] + # Poll due (first=True must NOT sneak in site_info while auth-failed) + assert run_async(api.run(120, True)) is True + assert calls == ["live_status"] + + def test_teslemetry(my_predbat=None): """Run all Teslemetry component tests (registry entry point). @@ -137,5 +274,11 @@ def test_teslemetry(my_predbat=None): test_teslemetry_live_status_publishes_sensors() test_teslemetry_site_info_publishes_soc_max() test_teslemetry_energy_today_publishes_kwh() + test_teslemetry_request_auth_failure() + test_teslemetry_request_rate_limit_retry() + test_teslemetry_request_success_clears_auth_flag() + test_teslemetry_run_unconfigured_returns_false() + test_teslemetry_run_first_success_returns_true() + test_teslemetry_run_auth_failed_only_probes_live_status() print("**** Teslemetry tests passed ****") return 0 From 65dcc3b0905d622c2696d1a00f9401ae05ea4b45 Mon Sep 17 00:00:00 2001 From: Mark Gascoyne Date: Thu, 2 Jul 2026 23:39:22 +0100 Subject: [PATCH 03/14] feat(teslemetry): control path - virtual entities and command handlers Implement control path for Tesla Powerwall integration: - Add site_info_done latch to ensure soc_max publishes eventually - Register virtual control entities: operation_mode, backup_reserve, allow_charging_from_grid, allow_export, tariff_mode - Implement command handlers: set_operation_mode, set_backup_reserve, set_grid_charging, set_export_rule - Add set_tariff stub (completed in Task 6) - Implement event handler overrides: select_event, number_event, switch_event - Add 6 control tests + 1 latch test (16 passing tests total) Co-Authored-By: Claude Fable 5 --- apps/predbat/teslemetry.py | 79 ++++++++++++++++++++++++++- apps/predbat/tests/test_teslemetry.py | 71 ++++++++++++++++++++++++ 2 files changed, 148 insertions(+), 2 deletions(-) diff --git a/apps/predbat/teslemetry.py b/apps/predbat/teslemetry.py index 449316ee7..3c4faa916 100644 --- a/apps/predbat/teslemetry.py +++ b/apps/predbat/teslemetry.py @@ -52,7 +52,9 @@ def initialize(self, key="", site_id="", base_url=TESLEMETRY_DEFAULT_URL, **kwar self.api_auth_failed = False self.last_live_poll = 0 self.last_energy_poll = 0 + self.site_info_done = False self.log("Info: TeslemetryAPI initialising site_id={}".format(self.site_id)) + self.register_control_entities() def entity(self, suffix, domain="sensor"): """Build a prefixed virtual entity id for this component.""" @@ -155,8 +157,8 @@ async def run(self, seconds=0, first=False): return await self.fetch_live_status() return False success = True - if first: - await self.fetch_site_info() + if not self.site_info_done: + self.site_info_done = await self.fetch_site_info() if first or (seconds - self.last_live_poll >= LIVE_POLL_SECONDS): self.last_live_poll = seconds success = await self.fetch_live_status() @@ -165,6 +167,79 @@ async def run(self, seconds=0, first=False): await self.fetch_energy_today() return success + def register_control_entities(self): + """Publish the virtual control entities with their initial states.""" + self.dashboard_item(self.entity("operation_mode", domain="select"), "self_consumption", {"friendly_name": "Powerwall Operation Mode", "options": OPERATION_MODES}, app="teslemetry") + self.dashboard_item(self.entity("backup_reserve", domain="number"), 0, {"friendly_name": "Powerwall Backup Reserve", "min": 0, "max": 100, "step": 1, "unit_of_measurement": "%"}, app="teslemetry") + self.dashboard_item(self.entity("allow_charging_from_grid", domain="switch"), "on", {"friendly_name": "Powerwall Allow Grid Charging"}, app="teslemetry") + self.dashboard_item(self.entity("allow_export", domain="select"), "never", {"friendly_name": "Powerwall Allow Export", "options": EXPORT_RULES}, app="teslemetry") + self.dashboard_item(self.entity("tariff_mode", domain="select"), "normal", {"friendly_name": "Powerwall Tariff Mode", "options": TARIFF_MODES}, app="teslemetry") + + async def _command(self, path, body): + """POST a command; return True on success.""" + result = await self._request("POST", "/api/1/energy_sites/{}/{}".format(self.site_id, path), json_body=body) + return result is not None + + async def set_operation_mode(self, mode): + """Set the Powerwall operation mode (self_consumption/autonomous/backup).""" + return await self._command("operation", {"default_real_mode": mode}) + + async def set_backup_reserve(self, percent): + """Set the backup reserve percentage (cloud caps at 80; 80-100 treated as 100 by Predbat).""" + return await self._command("backup", {"backup_reserve_percent": int(percent)}) + + async def set_grid_charging(self, allow): + """Allow or disallow charging the battery from the grid.""" + return await self._command("grid_import_export", {"disallow_charge_from_grid_with_solar_installed": not allow}) + + async def set_export_rule(self, rule): + """Set the grid export rule (never/pv_only/battery_ok).""" + return await self._command("grid_import_export", {"customer_preferred_export_rule": rule}) + + async def set_tariff(self, mode): + """Push the tariff for the requested mode (implemented in the tariff task).""" + self.log("Info: Teslemetry set_tariff({}) stub".format(mode)) + return True + + async def select_event(self, entity_id, value): + """Handle select changes routed from the service-hook loopback.""" + success = False + if entity_id.endswith("_operation_mode") and value in OPERATION_MODES: + success = await self.set_operation_mode(value) + elif entity_id.endswith("_allow_export") and value in EXPORT_RULES: + success = await self.set_export_rule(value) + elif entity_id.endswith("_tariff_mode") and value in TARIFF_MODES: + success = await self.set_tariff(value) + else: + self.log("Warn: Teslemetry unhandled select_event {} = {}".format(entity_id, value)) + return + if success: + self.set_state_wrapper(entity_id, value) + else: + self.log("Warn: Teslemetry command failed for {} = {} (state not updated)".format(entity_id, value)) + + async def number_event(self, entity_id, value): + """Handle number changes routed from the service-hook loopback.""" + if entity_id.endswith("_backup_reserve"): + percent = max(0, min(100, int(float(value)))) + if await self.set_backup_reserve(percent): + self.set_state_wrapper(entity_id, percent) + else: + self.log("Warn: Teslemetry backup_reserve command failed (state not updated)") + else: + self.log("Warn: Teslemetry unhandled number_event {} = {}".format(entity_id, value)) + + async def switch_event(self, entity_id, service): + """Handle switch service calls routed from the service-hook loopback.""" + if entity_id.endswith("_allow_charging_from_grid"): + allow = service != "turn_off" + if await self.set_grid_charging(allow): + self.set_state_wrapper(entity_id, "on" if allow else "off") + else: + self.log("Warn: Teslemetry grid charging command failed (state not updated)") + else: + self.log("Warn: Teslemetry unhandled switch_event {} {}".format(entity_id, service)) + async def final(self): """Cleanup on shutdown.""" self.log("Info: TeslemetryAPI shutdown") diff --git a/apps/predbat/tests/test_teslemetry.py b/apps/predbat/tests/test_teslemetry.py index fa9f5bcd6..aba3fc117 100644 --- a/apps/predbat/tests/test_teslemetry.py +++ b/apps/predbat/tests/test_teslemetry.py @@ -23,6 +23,7 @@ def __init__(self): self.api_auth_failed = False self.last_live_poll = 0 self.last_energy_poll = 0 + self.site_info_done = False self.dashboard_items = {} self.log_messages = [] self.entity_states = {} @@ -264,6 +265,70 @@ def test_teslemetry_run_auth_failed_only_probes_live_status(): assert calls == ["live_status"] +def test_teslemetry_select_operation_mode(): + """select_event on operation_mode POSTs /operation and updates entity state on success.""" + api = MockTeslemetryAPI() + api.mock_responses["/api/1/energy_sites/123456/operation"] = {"response": {"code": 201}} + run_async(api.select_event("select.predbat_teslemetry_operation_mode", "backup")) + assert ("POST", "/api/1/energy_sites/123456/operation", {"default_real_mode": "backup"}) in api.requests_made + assert api.entity_states["select.predbat_teslemetry_operation_mode"] == "backup" + + +def test_teslemetry_select_failure_keeps_state(): + """Failed command does not update entity state (write_and_poll will retry).""" + api = MockTeslemetryAPI() + # No mock response registered -> _request returns None -> command fails + run_async(api.select_event("select.predbat_teslemetry_operation_mode", "backup")) + assert "select.predbat_teslemetry_operation_mode" not in api.entity_states + + +def test_teslemetry_number_backup_reserve(): + """number_event on backup_reserve POSTs /backup with an integer percent.""" + api = MockTeslemetryAPI() + api.mock_responses["/api/1/energy_sites/123456/backup"] = {"response": {"code": 201}} + run_async(api.number_event("number.predbat_teslemetry_backup_reserve", 100)) + assert ("POST", "/api/1/energy_sites/123456/backup", {"backup_reserve_percent": 100}) in api.requests_made + assert api.entity_states["number.predbat_teslemetry_backup_reserve"] == 100 + + +def test_teslemetry_switch_grid_charging(): + """switch_event maps turn_off to disallow grid charging.""" + api = MockTeslemetryAPI() + api.mock_responses["/api/1/energy_sites/123456/grid_import_export"] = {"response": {"code": 201}} + run_async(api.switch_event("switch.predbat_teslemetry_allow_charging_from_grid", "turn_off")) + assert ("POST", "/api/1/energy_sites/123456/grid_import_export", {"disallow_charge_from_grid_with_solar_installed": True}) in api.requests_made + assert api.entity_states["switch.predbat_teslemetry_allow_charging_from_grid"] == "off" + + +def test_teslemetry_select_export_rule(): + """select_event on allow_export POSTs the customer_preferred_export_rule.""" + api = MockTeslemetryAPI() + api.mock_responses["/api/1/energy_sites/123456/grid_import_export"] = {"response": {"code": 201}} + run_async(api.select_event("select.predbat_teslemetry_allow_export", "battery_ok")) + assert ("POST", "/api/1/energy_sites/123456/grid_import_export", {"customer_preferred_export_rule": "battery_ok"}) in api.requests_made + assert api.entity_states["select.predbat_teslemetry_allow_export"] == "battery_ok" + + +def test_teslemetry_site_info_latch(): + """site_info fetch is retried on every cycle until success; latch prevents redundant calls once successful.""" + api, calls = _make_run_api_with_fetch_capture(site_info=False, live_status=True, energy_today=True) + # First cycle: site_info missing, fetch returns False, latch stays False + assert run_async(api.run(0, True)) is True + assert calls == ["site_info", "live_status", "energy_today"] + assert api.site_info_done is False + # Now provide the site_info mock and unbind fetch_site_info so the real one runs + api.mock_responses["/api/1/energy_sites/123456/site_info"] = SITE_INFO + api.fetch_site_info = TeslemetryAPI.fetch_site_info.__get__(api) + # Re-create the fake live_status/energy_today so they still record and return True + api.fetch_live_status = _make_run_api_with_fetch_capture(live_status=True)[0].fetch_live_status + api.fetch_energy_today = _make_run_api_with_fetch_capture(energy_today=True)[0].fetch_energy_today + # Run again: site_info_done latch should trigger fetch_site_info and succeed + assert run_async(api.run(300, False)) is True + # site_info_done should now be True and soc_max should be published + assert api.site_info_done is True + assert api.dashboard_items["sensor.predbat_teslemetry_soc_max"]["state"] == 13.5 + + def test_teslemetry(my_predbat=None): """Run all Teslemetry component tests (registry entry point). @@ -280,5 +345,11 @@ def test_teslemetry(my_predbat=None): test_teslemetry_run_unconfigured_returns_false() test_teslemetry_run_first_success_returns_true() test_teslemetry_run_auth_failed_only_probes_live_status() + test_teslemetry_select_operation_mode() + test_teslemetry_select_failure_keeps_state() + test_teslemetry_number_backup_reserve() + test_teslemetry_switch_grid_charging() + test_teslemetry_select_export_rule() + test_teslemetry_site_info_latch() print("**** Teslemetry tests passed ****") return 0 From 258633c8c0489be193ab70cc48d053058599e8e4 Mon Sep 17 00:00:00 2001 From: Mark Gascoyne Date: Thu, 2 Jul 2026 23:50:06 +0100 Subject: [PATCH 04/14] feat(teslemetry): export tariff-trick and boot reconciliation Implement Task 6 of Tesla Powerwall integration: tariff builder, export-now trick, and boot reconciliation. - Add EXPORT_SELL_RATE constant and current_rates() to read from base.rate_import/export - Implement build_tariff(mode, now) with export_now (current 30-min window ON_PEAK) and normal (flat) modes, returning tariff_content_v2 dict - Replace set_tariff() stub with real implementation POSTing /time_of_use_settings - Add reconcile_on_start() to restore safe state if previous run died mid-export - Wire reconcile_on_start() into run() after site_info latch All 5 new tests pass (build tariff, set tariff, reconcile); regression on ge_cloud also passes (66 tests). Co-Authored-By: Claude Fable 5 --- apps/predbat/teslemetry.py | 102 +++++++++++++++++++++++++- apps/predbat/tests/test_teslemetry.py | 58 +++++++++++++++ 2 files changed, 157 insertions(+), 3 deletions(-) diff --git a/apps/predbat/teslemetry.py b/apps/predbat/teslemetry.py index 3c4faa916..0e568c86a 100644 --- a/apps/predbat/teslemetry.py +++ b/apps/predbat/teslemetry.py @@ -18,6 +18,7 @@ """ import asyncio +from datetime import datetime, timezone import aiohttp @@ -37,6 +38,10 @@ class TeslemetryAPI(ComponentBase): """Tesla Powerwall component using the Teslemetry (or Fleet) REST API.""" + EXPORT_SELL_RATE = 0.50 # GBP/kWh synthetic high sell price to force export now. + DEFAULT_IMPORT_RATE = 0.28 + DEFAULT_EXPORT_RATE = 0.15 + def initialize(self, key="", site_id="", base_url=TESLEMETRY_DEFAULT_URL, **kwargs): """Initialise the Teslemetry component from configuration. @@ -159,6 +164,8 @@ async def run(self, seconds=0, first=False): success = True if not self.site_info_done: self.site_info_done = await self.fetch_site_info() + if first: + await self.reconcile_on_start() if first or (seconds - self.last_live_poll >= LIVE_POLL_SECONDS): self.last_live_poll = seconds success = await self.fetch_live_status() @@ -196,10 +203,99 @@ async def set_export_rule(self, rule): """Set the grid export rule (never/pv_only/battery_ok).""" return await self._command("grid_import_export", {"customer_preferred_export_rule": rule}) + def current_rates(self): + """Return (import_rate, export_rate) in GBP/kWh from Predbat's rate data when available. + + Defaults to class constants if rates are unavailable. Rates from the base are in PENCE + (pence per kWh), so they are divided by 100.0 to convert to GBP. + """ + import_rate = self.DEFAULT_IMPORT_RATE + export_rate = self.DEFAULT_EXPORT_RATE + base = getattr(self, "base", None) + if base is not None: + minutes_now = getattr(base, "minutes_now", None) + rate_import = getattr(base, "rate_import", None) + rate_export = getattr(base, "rate_export", None) + if minutes_now is not None and rate_import: + import_rate = round(rate_import.get(minutes_now, self.DEFAULT_IMPORT_RATE * 100) / 100.0, 4) + if minutes_now is not None and rate_export: + export_rate = round(rate_export.get(minutes_now, self.DEFAULT_EXPORT_RATE * 100) / 100.0, 4) + return import_rate, export_rate + + def build_tariff(self, mode, now=None): + """Build a tariff_content_v2 dict for the requested mode. + + export_now: current 30-minute-aligned window becomes ON_PEAK with a + high sell price so autonomous mode exports immediately; the window + self-expires. normal: flat tariff from the customer's current rates. + """ + if now is None: + now = datetime.now(timezone.utc) + import_rate, export_rate = self.current_rates() + + def charges(off_peak_buy, on_peak_buy): + """Build an energy_charges block.""" + rates = {"SUPER_OFF_PEAK": off_peak_buy} + if mode == "export_now": + rates["ON_PEAK"] = on_peak_buy + return {"ALL": {"rates": {"ALL": 0}}, "AllYear": {"rates": rates}} + + tou_periods = {"SUPER_OFF_PEAK": {"periods": [{"fromDayOfWeek": 0, "toDayOfWeek": 6, "fromHour": 0, "fromMinute": 0, "toHour": 0, "toMinute": 0}]}} + if mode == "export_now": + start = now.replace(minute=0 if now.minute < 30 else 30, second=0, microsecond=0) + end_minute_total = start.hour * 60 + start.minute + 60 + end_hour, end_min = (end_minute_total // 60) % 24, end_minute_total % 60 + window = {"fromDayOfWeek": 0, "toDayOfWeek": 6, "fromHour": start.hour, "fromMinute": start.minute, "toHour": end_hour, "toMinute": end_min} + off_periods = [] + if start.hour > 0 or start.minute > 0: + off_periods.append({"fromDayOfWeek": 0, "toDayOfWeek": 6, "fromHour": 0, "fromMinute": 0, "toHour": start.hour, "toMinute": start.minute}) + if end_hour > 0 or end_min > 0: + off_periods.append({"fromDayOfWeek": 0, "toDayOfWeek": 6, "fromHour": end_hour, "fromMinute": end_min, "toHour": 0, "toMinute": 0}) + tou_periods = {"SUPER_OFF_PEAK": {"periods": off_periods}, "ON_PEAK": {"periods": [window]}} + + seasons = {"AllYear": {"fromMonth": 1, "fromDay": 1, "toMonth": 12, "toDay": 31, "tou_periods": tou_periods}} + sell_rates = charges(export_rate, self.EXPORT_SELL_RATE) + buy_rates = charges(import_rate, import_rate) + common = { + "min_applicable_demand": 0, + "max_applicable_demand": 0, + "monthly_minimum_bill": 0, + "monthly_charges": 0, + "daily_charges": [{"name": "Charge", "amount": 0}], + "demand_charges": {"ALL": {"rates": {"ALL": 0}}, "AllYear": {"rates": {}}}, + } + return { + "version": 1, + "utility": "Predbat", + "code": "PREDBAT-{}".format(mode.upper().replace("_", "-")), + "name": "Predbat ({})".format(mode), + "currency": "GBP", + "daily_demand_charges": {}, + "energy_charges": buy_rates, + "seasons": seasons, + "sell_tariff": {**common, "utility": "Predbat", "energy_charges": sell_rates, "seasons": seasons}, + **common, + } + async def set_tariff(self, mode): - """Push the tariff for the requested mode (implemented in the tariff task).""" - self.log("Info: Teslemetry set_tariff({}) stub".format(mode)) - return True + """Push the tariff for the requested mode via time_of_use_settings.""" + tariff = self.build_tariff(mode) + return await self._command("time_of_use_settings", {"tou_settings": {"tariff_content_v2": tariff}}) + + async def reconcile_on_start(self): + """Restore a safe state if a previous run died mid-export. + + If the tariff mode is left in export_now (indicating a previous run crashed + mid-operation), restores normal tariff and disables export to prevent unintended + grid export. If the tariff mode is already normal, does nothing. + """ + tariff_mode = self.get_state_wrapper(self.entity("tariff_mode", domain="select"), default="normal") + if tariff_mode == "export_now": + self.log("Warn: Teslemetry was left in export_now - restoring normal tariff and disabling export") + if await self.set_tariff("normal"): + self.set_state_wrapper(self.entity("tariff_mode", domain="select"), "normal") + if await self.set_export_rule("never"): + self.set_state_wrapper(self.entity("allow_export", domain="select"), "never") async def select_event(self, entity_id, value): """Handle select changes routed from the service-hook loopback.""" diff --git a/apps/predbat/tests/test_teslemetry.py b/apps/predbat/tests/test_teslemetry.py index aba3fc117..30f94d115 100644 --- a/apps/predbat/tests/test_teslemetry.py +++ b/apps/predbat/tests/test_teslemetry.py @@ -329,6 +329,59 @@ def test_teslemetry_site_info_latch(): assert api.dashboard_items["sensor.predbat_teslemetry_soc_max"]["state"] == 13.5 +def test_teslemetry_build_tariff_export_now(): + """export_now tariff marks the current window ON_PEAK with a high sell rate.""" + from datetime import datetime + + api = MockTeslemetryAPI() + now = datetime(2026, 7, 2, 14, 40) + tariff = api.build_tariff("export_now", now=now) + assert tariff["version"] == 1 + sell = tariff["sell_tariff"]["energy_charges"]["AllYear"]["rates"] + assert sell["ON_PEAK"] > sell["SUPER_OFF_PEAK"] + periods = tariff["seasons"]["AllYear"]["tou_periods"]["ON_PEAK"]["periods"] + assert periods[0]["fromHour"] == 14 and periods[0]["fromMinute"] == 30 + assert periods[0]["toHour"] == 15 and periods[0]["toMinute"] == 30 + + +def test_teslemetry_build_tariff_normal_flat(): + """normal tariff is flat (single ALL rate, no ON_PEAK windows).""" + api = MockTeslemetryAPI() + tariff = api.build_tariff("normal") + assert "ON_PEAK" not in tariff["seasons"]["AllYear"]["tou_periods"] + assert tariff["energy_charges"]["AllYear"]["rates"]["SUPER_OFF_PEAK"] >= 0 + + +def test_teslemetry_set_tariff_posts_tou_settings(): + """set_tariff wraps the tariff in tou_settings and POSTs it.""" + api = MockTeslemetryAPI() + api.mock_responses["/api/1/energy_sites/123456/time_of_use_settings"] = {"response": {"code": 201}} + result = run_async(api.set_tariff("normal")) + assert result is True + method, path, body = api.requests_made[-1] + assert path == "/api/1/energy_sites/123456/time_of_use_settings" + assert "tariff_content_v2" in body["tou_settings"] + + +def test_teslemetry_reconcile_on_start_restores_normal(): + """Boot reconciliation restores normal tariff + never export if left in export_now.""" + api = MockTeslemetryAPI() + api.entity_states["select.predbat_teslemetry_tariff_mode"] = "export_now" + api.mock_responses["/api/1/energy_sites/123456/time_of_use_settings"] = {"response": {"code": 201}} + api.mock_responses["/api/1/energy_sites/123456/grid_import_export"] = {"response": {"code": 201}} + run_async(api.reconcile_on_start()) + assert api.entity_states["select.predbat_teslemetry_tariff_mode"] == "normal" + assert api.entity_states["select.predbat_teslemetry_allow_export"] == "never" + + +def test_teslemetry_reconcile_on_start_noop_when_normal(): + """Boot reconciliation does nothing when tariff mode is already normal.""" + api = MockTeslemetryAPI() + api.entity_states["select.predbat_teslemetry_tariff_mode"] = "normal" + run_async(api.reconcile_on_start()) + assert api.requests_made == [] + + def test_teslemetry(my_predbat=None): """Run all Teslemetry component tests (registry entry point). @@ -351,5 +404,10 @@ def test_teslemetry(my_predbat=None): test_teslemetry_switch_grid_charging() test_teslemetry_select_export_rule() test_teslemetry_site_info_latch() + test_teslemetry_build_tariff_export_now() + test_teslemetry_build_tariff_normal_flat() + test_teslemetry_set_tariff_posts_tou_settings() + test_teslemetry_reconcile_on_start_restores_normal() + test_teslemetry_reconcile_on_start_noop_when_normal() print("**** Teslemetry tests passed ****") return 0 From 87f947e4e96baf4410ad9e75636274da1ed1e24e Mon Sep 17 00:00:00 2001 From: Mark Gascoyne Date: Fri, 3 Jul 2026 00:03:47 +0100 Subject: [PATCH 05/14] fix(teslemetry): midnight-wrap tariff complement + dynamic sell clamp Address review findings on the export tariff-trick: - CRITICAL: export_now windows crossing midnight (e.g. 23:30-00:30) produced overlapping SUPER_OFF_PEAK periods covering the full day. The complement is now a proper circle complement of [start, end): a single wrapped segment when the window crosses midnight, two segments otherwise. - IMPORTANT: ON_PEAK sell is now clamped to max(EXPORT_SELL_RATE, 2x live export rate) so the trick cannot silently invert when the live export rate exceeds 0.50. - MINOR: buy-side ON_PEAK mirrors the high sell rate to discourage grid-charging during the export window. - Tests: midnight-wrap (23:40) and exact-midnight (23:10) cases with a partition-of-day assertion (no overlap, no gap), sell-clamp with a live 0.60 export rate, and a reconcile_on_start partial-failure pin. Co-Authored-By: Claude Fable 5 --- apps/predbat/teslemetry.py | 21 +++++--- apps/predbat/tests/test_teslemetry.py | 69 +++++++++++++++++++++++++++ 2 files changed, 84 insertions(+), 6 deletions(-) diff --git a/apps/predbat/teslemetry.py b/apps/predbat/teslemetry.py index 0e568c86a..737294140 100644 --- a/apps/predbat/teslemetry.py +++ b/apps/predbat/teslemetry.py @@ -232,6 +232,9 @@ def build_tariff(self, mode, now=None): if now is None: now = datetime.now(timezone.utc) import_rate, export_rate = self.current_rates() + # Keep the synthetic ON_PEAK sell strictly above the live export rate, otherwise the + # export trick silently inverts at the highest-value moments. + sell_high = max(self.EXPORT_SELL_RATE, round(export_rate * 2, 4)) def charges(off_peak_buy, on_peak_buy): """Build an energy_charges block.""" @@ -244,18 +247,24 @@ def charges(off_peak_buy, on_peak_buy): if mode == "export_now": start = now.replace(minute=0 if now.minute < 30 else 30, second=0, microsecond=0) end_minute_total = start.hour * 60 + start.minute + 60 + wrapped = end_minute_total >= 24 * 60 end_hour, end_min = (end_minute_total // 60) % 24, end_minute_total % 60 window = {"fromDayOfWeek": 0, "toDayOfWeek": 6, "fromHour": start.hour, "fromMinute": start.minute, "toHour": end_hour, "toMinute": end_min} off_periods = [] - if start.hour > 0 or start.minute > 0: - off_periods.append({"fromDayOfWeek": 0, "toDayOfWeek": 6, "fromHour": 0, "fromMinute": 0, "toHour": start.hour, "toMinute": start.minute}) - if end_hour > 0 or end_min > 0: - off_periods.append({"fromDayOfWeek": 0, "toDayOfWeek": 6, "fromHour": end_hour, "fromMinute": end_min, "toHour": 0, "toMinute": 0}) + if wrapped: + # ON_PEAK crosses midnight, so its circle complement is a single segment from the wrapped end back to the start. + off_periods.append({"fromDayOfWeek": 0, "toDayOfWeek": 6, "fromHour": end_hour, "fromMinute": end_min, "toHour": start.hour, "toMinute": start.minute}) + else: + if start.hour > 0 or start.minute > 0: + off_periods.append({"fromDayOfWeek": 0, "toDayOfWeek": 6, "fromHour": 0, "fromMinute": 0, "toHour": start.hour, "toMinute": start.minute}) + if end_hour > 0 or end_min > 0: + off_periods.append({"fromDayOfWeek": 0, "toDayOfWeek": 6, "fromHour": end_hour, "fromMinute": end_min, "toHour": 0, "toMinute": 0}) tou_periods = {"SUPER_OFF_PEAK": {"periods": off_periods}, "ON_PEAK": {"periods": [window]}} seasons = {"AllYear": {"fromMonth": 1, "fromDay": 1, "toMonth": 12, "toDay": 31, "tou_periods": tou_periods}} - sell_rates = charges(export_rate, self.EXPORT_SELL_RATE) - buy_rates = charges(import_rate, import_rate) + sell_rates = charges(export_rate, sell_high) + # Buy-side ON_PEAK mirrors the high sell rate to discourage grid-charging during the export window. + buy_rates = charges(import_rate, sell_high) common = { "min_applicable_demand": 0, "max_applicable_demand": 0, diff --git a/apps/predbat/tests/test_teslemetry.py b/apps/predbat/tests/test_teslemetry.py index 30f94d115..f1581f2ba 100644 --- a/apps/predbat/tests/test_teslemetry.py +++ b/apps/predbat/tests/test_teslemetry.py @@ -382,6 +382,71 @@ def test_teslemetry_reconcile_on_start_noop_when_normal(): assert api.requests_made == [] +def _assert_tou_periods_partition_day(tou_periods): + """Assert the tou_periods cover every minute of the (circular) day exactly once — no overlaps, no gaps.""" + covered = [0] * (24 * 60) + for group in tou_periods.values(): + for period in group["periods"]: + from_min = period["fromHour"] * 60 + period["fromMinute"] + to_min = period["toHour"] * 60 + period["toMinute"] + if to_min <= from_min: + to_min += 24 * 60 + for minute in range(from_min, to_min): + covered[minute % (24 * 60)] += 1 + assert max(covered) <= 1, "tou_periods overlap" + assert min(covered) >= 1, "tou_periods leave a gap" + + +def test_teslemetry_build_tariff_export_now_midnight_wrap(): + """export_now at 23:40 wraps midnight: ON_PEAK 23:30-00:30 with a single non-overlapping off-peak complement.""" + from datetime import datetime + + api = MockTeslemetryAPI() + tariff = api.build_tariff("export_now", now=datetime(2026, 7, 2, 23, 40)) + tou_periods = tariff["seasons"]["AllYear"]["tou_periods"] + assert tou_periods["ON_PEAK"]["periods"] == [{"fromDayOfWeek": 0, "toDayOfWeek": 6, "fromHour": 23, "fromMinute": 30, "toHour": 0, "toMinute": 30}] + assert tou_periods["SUPER_OFF_PEAK"]["periods"] == [{"fromDayOfWeek": 0, "toDayOfWeek": 6, "fromHour": 0, "fromMinute": 30, "toHour": 23, "toMinute": 30}] + _assert_tou_periods_partition_day(tou_periods) + + +def test_teslemetry_build_tariff_export_now_midnight_exact(): + """export_now at 23:10 ends exactly at midnight: ON_PEAK 23:00-00:00 with a single off-peak complement.""" + from datetime import datetime + + api = MockTeslemetryAPI() + tariff = api.build_tariff("export_now", now=datetime(2026, 7, 2, 23, 10)) + tou_periods = tariff["seasons"]["AllYear"]["tou_periods"] + assert tou_periods["ON_PEAK"]["periods"] == [{"fromDayOfWeek": 0, "toDayOfWeek": 6, "fromHour": 23, "fromMinute": 0, "toHour": 0, "toMinute": 0}] + assert tou_periods["SUPER_OFF_PEAK"]["periods"] == [{"fromDayOfWeek": 0, "toDayOfWeek": 6, "fromHour": 0, "fromMinute": 0, "toHour": 23, "toMinute": 0}] + _assert_tou_periods_partition_day(tou_periods) + + +def test_teslemetry_build_tariff_sell_clamp_above_export_rate(): + """ON_PEAK sell (and buy) stay above the live export rate even when it exceeds the static high rate.""" + from types import SimpleNamespace + + api = MockTeslemetryAPI() + api.base = SimpleNamespace(minutes_now=600, rate_import={600: 30.0}, rate_export={600: 60.0}) + tariff = api.build_tariff("export_now") + sell = tariff["sell_tariff"]["energy_charges"]["AllYear"]["rates"] + assert sell["SUPER_OFF_PEAK"] == 0.6 + assert sell["ON_PEAK"] > 0.6 + assert sell["ON_PEAK"] == 1.2 + # Buy-side ON_PEAK matches the high sell rate to discourage grid-charging during the export window + assert tariff["energy_charges"]["AllYear"]["rates"]["ON_PEAK"] == sell["ON_PEAK"] + + +def test_teslemetry_reconcile_on_start_partial_failure(): + """Partial reconcile: tariff restore succeeds but the export-rule command fails, so only tariff_mode is updated.""" + api = MockTeslemetryAPI() + api.entity_states["select.predbat_teslemetry_tariff_mode"] = "export_now" + api.mock_responses["/api/1/energy_sites/123456/time_of_use_settings"] = {"response": {"code": 201}} + # grid_import_export mock ABSENT -> _request returns None -> set_export_rule fails + run_async(api.reconcile_on_start()) + assert api.entity_states["select.predbat_teslemetry_tariff_mode"] == "normal" + assert "select.predbat_teslemetry_allow_export" not in api.entity_states + + def test_teslemetry(my_predbat=None): """Run all Teslemetry component tests (registry entry point). @@ -409,5 +474,9 @@ def test_teslemetry(my_predbat=None): test_teslemetry_set_tariff_posts_tou_settings() test_teslemetry_reconcile_on_start_restores_normal() test_teslemetry_reconcile_on_start_noop_when_normal() + test_teslemetry_build_tariff_export_now_midnight_wrap() + test_teslemetry_build_tariff_export_now_midnight_exact() + test_teslemetry_build_tariff_sell_clamp_above_export_rate() + test_teslemetry_reconcile_on_start_partial_failure() print("**** Teslemetry tests passed ****") return 0 From 23689d81a47506b2719048d35a91a70803d672e9 Mon Sep 17 00:00:00 2001 From: Mark Gascoyne Date: Fri, 3 Jul 2026 00:15:31 +0100 Subject: [PATCH 06/14] feat(teslemetry): register component and config schema Registers TeslemetryAPI in COMPONENT_LIST (phase 1, event_filter predbat_teslemetry_) and adds teslemetry_key/teslemetry_site_id/ teslemetry_base_url to APPS_SCHEMA, placed alongside the fox_key inverter block (config.py has no deye_key entry to anchor on). --- apps/predbat/components.py | 12 ++++++++++++ apps/predbat/config.py | 3 +++ 2 files changed, 15 insertions(+) diff --git a/apps/predbat/components.py b/apps/predbat/components.py index 4951f79c5..12856c6f6 100644 --- a/apps/predbat/components.py +++ b/apps/predbat/components.py @@ -27,6 +27,7 @@ from axle import AxleAPI from solax import SolaxAPI from sigenergy import SigenergyAPI +from teslemetry import TeslemetryAPI from solis import SolisAPI from alertfeed import AlertFeed from web import WebInterface @@ -366,6 +367,17 @@ "phase": 1, "can_restart": True, }, + "teslemetry": { + "class": TeslemetryAPI, + "name": "Tesla Powerwall (Teslemetry)", + "event_filter": "predbat_teslemetry_", + "args": { + "key": {"required": True, "config": "teslemetry_key"}, + "site_id": {"required": True, "config": "teslemetry_site_id"}, + "base_url": {"required": False, "config": "teslemetry_base_url", "default": "https://api.teslemetry.com"}, + }, + "phase": 1, + }, "solax": { "class": SolaxAPI, "name": "SolaX Cloud API", diff --git a/apps/predbat/config.py b/apps/predbat/config.py index ec826f9c5..a8283129b 100644 --- a/apps/predbat/config.py +++ b/apps/predbat/config.py @@ -2200,6 +2200,9 @@ "fox_auth_method": {"type": "string", "empty": False}, "fox_token_expires_at": {"type": "string", "empty": False}, "fox_token_hash": {"type": "string", "empty": False}, + "teslemetry_key": {"type": "string", "empty": False}, + "teslemetry_site_id": {"type": "string", "empty": False}, + "teslemetry_base_url": {"type": "string", "empty": False}, "octopus_intelligent_slot": {"type": "sensor|sensor_list", "sensor_type": "boolean|action", "entries": "num_cars", "optional_entries": True}, "octopus_ready_time": {"type": "sensor|sensor_list", "sensor_type": "string", "entries": "num_cars", "optional_entries": True}, "octopus_charge_limit": {"type": "sensor|sensor_list", "sensor_type": "float", "entries": "num_cars", "optional_entries": True}, From 55acec2926b247a706cfc7c03c58e65dd861ed97 Mon Sep 17 00:00:00 2001 From: Mark Gascoyne Date: Fri, 3 Jul 2026 12:18:17 +0100 Subject: [PATCH 07/14] fix(teslemetry): device-state boot reconcile + command response validation reconcile_on_start previously read the local tariff_mode entity, which register_control_entities() unconditionally reseeds to "normal" on every boot before run(first=True) fires - making the export_now recovery path dead code (a pod restart mid-export could never self-heal). It now reads the Powerwall's actual tariff via GET tariff_rate and walks the response for a "code" matching build_tariff("export_now"), respecting read-only mode. _command also treated any parsed 2xx body as success; it now fails on an application-level "error" key or response.code >= 400. fetch_site_info seeds the operation_mode/backup_reserve entity states from device data instead of leaving them at hardcoded defaults. Co-Authored-By: Claude Fable 5 --- apps/predbat/teslemetry.py | 99 +++++++++++++++++++++---- apps/predbat/tests/test_teslemetry.py | 102 +++++++++++++++++++++++--- 2 files changed, 178 insertions(+), 23 deletions(-) diff --git a/apps/predbat/teslemetry.py b/apps/predbat/teslemetry.py index 737294140..164838eb7 100644 --- a/apps/predbat/teslemetry.py +++ b/apps/predbat/teslemetry.py @@ -115,7 +115,7 @@ async def fetch_live_status(self): return True async def fetch_site_info(self): - """Fetch site info, publishing the battery capacity as soc_max.""" + """Fetch site info, publishing the battery capacity as soc_max and seeding control entity states.""" data = await self._request("GET", "/api/1/energy_sites/{}/site_info".format(self.site_id)) if not data: return False @@ -123,6 +123,14 @@ async def fetch_site_info(self): nameplate_wh = response.get("nameplate_energy", 0) if nameplate_wh: self.publish_sensor("soc_max", round(nameplate_wh / 1000.0, 2), unit="kWh", state_class=None, friendly="Powerwall Capacity") + # Seed the control entity STATES (display only, no commands) from the device so they reflect + # reality at boot instead of the hardcoded defaults set by register_control_entities(). + default_mode = response.get("default_real_mode") + if default_mode in OPERATION_MODES: + self.set_state_wrapper(self.entity("operation_mode", domain="select"), default_mode) + backup_reserve = response.get("backup_reserve_percent") + if isinstance(backup_reserve, (int, float)): + self.set_state_wrapper(self.entity("backup_reserve", domain="number"), backup_reserve) return True async def fetch_energy_today(self): @@ -183,9 +191,26 @@ def register_control_entities(self): self.dashboard_item(self.entity("tariff_mode", domain="select"), "normal", {"friendly_name": "Powerwall Tariff Mode", "options": TARIFF_MODES}, app="teslemetry") async def _command(self, path, body): - """POST a command; return True on success.""" + """POST a command; return True only if both the transport and the application layer succeeded. + + A 2xx HTTP status with a parsed JSON body is not sufficient on its own: Teslemetry/Fleet API can + return HTTP 200 carrying an application-level failure, so the body is also checked for a truthy + "error" key or a numeric "response.code" of 400 or above. + """ result = await self._request("POST", "/api/1/energy_sites/{}/{}".format(self.site_id, path), json_body=body) - return result is not None + if result is None: + return False + if isinstance(result, dict): + if result.get("error"): + self.log("Warn: Teslemetry command to {} failed (application error): {}".format(path, str(result)[:200])) + return False + response = result.get("response") + if isinstance(response, dict): + code = response.get("code") + if isinstance(code, (int, float)) and not isinstance(code, bool) and code >= 400: + self.log("Warn: Teslemetry command to {} failed (response code {}): {}".format(path, code, str(result)[:200])) + return False + return True async def set_operation_mode(self, mode): """Set the Powerwall operation mode (self_consumption/autonomous/backup).""" @@ -291,20 +316,68 @@ async def set_tariff(self, mode): tariff = self.build_tariff(mode) return await self._command("time_of_use_settings", {"tou_settings": {"tariff_content_v2": tariff}}) + @staticmethod + def _find_tariff_code(node): + """Recursively search a parsed JSON structure for the first string "code" key. + + The Teslemetry/Fleet API tariff_rate response shape is not guaranteed to nest the + code at one fixed path (top-level, under tariff_content_v2, or elsewhere), so this + walks dicts and lists looking for a "code" key rather than assuming a fixed path. + """ + if isinstance(node, dict): + code = node.get("code") + if isinstance(code, str): + return code + for value in node.values(): + found = TeslemetryAPI._find_tariff_code(value) + if found is not None: + return found + elif isinstance(node, list): + for item in node: + found = TeslemetryAPI._find_tariff_code(item) + if found is not None: + return found + return None + + async def get_current_tariff_code(self): + """Fetch the site's current tariff_rate from the device and return its tariff "code", or None on failure.""" + data = await self._request("GET", "/api/1/energy_sites/{}/tariff_rate".format(self.site_id)) + if not data: + return None + return self._find_tariff_code(data) + + def _is_read_only(self): + """Return True if Predbat is configured read-only, so device-write commands must not be sent.""" + base = getattr(self, "base", None) + return bool(getattr(base, "set_read_only", False)) if base is not None else False + async def reconcile_on_start(self): """Restore a safe state if a previous run died mid-export. - If the tariff mode is left in export_now (indicating a previous run crashed - mid-operation), restores normal tariff and disables export to prevent unintended - grid export. If the tariff mode is already normal, does nothing. + Reads the Powerwall's ACTUAL current tariff from the device rather than the local + entity mirror: register_control_entities() unconditionally reseeds the mirror to + "normal" on every boot, so trusting it would make this recovery path unreachable. + If the device's tariff still carries the PREDBAT-EXPORT-NOW marker written by + build_tariff(), a previous run crashed mid-export, so normal tariff and disabled + export are restored (unless Predbat is running read-only, in which case the need + for recovery is logged but no write is sent). If the tariff read itself fails, + this skips silently rather than guessing at the device state. """ - tariff_mode = self.get_state_wrapper(self.entity("tariff_mode", domain="select"), default="normal") - if tariff_mode == "export_now": - self.log("Warn: Teslemetry was left in export_now - restoring normal tariff and disabling export") - if await self.set_tariff("normal"): - self.set_state_wrapper(self.entity("tariff_mode", domain="select"), "normal") - if await self.set_export_rule("never"): - self.set_state_wrapper(self.entity("allow_export", domain="select"), "never") + tariff_code = await self.get_current_tariff_code() + if tariff_code is None: + self.log("Warn: Teslemetry could not read the current tariff from the device on boot - skipping reconcile") + return + export_now_code = self.build_tariff("export_now").get("code") + if tariff_code != export_now_code: + return + if self._is_read_only(): + self.log("Warn: Teslemetry device tariff was left as {} - recovery needed but skipped (read-only mode)".format(tariff_code)) + return + self.log("Warn: Teslemetry device tariff was left as {} - restoring normal tariff and disabling export".format(tariff_code)) + if await self.set_tariff("normal"): + self.set_state_wrapper(self.entity("tariff_mode", domain="select"), "normal") + if await self.set_export_rule("never"): + self.set_state_wrapper(self.entity("allow_export", domain="select"), "never") async def select_event(self, entity_id, value): """Handle select changes routed from the service-hook loopback.""" diff --git a/apps/predbat/tests/test_teslemetry.py b/apps/predbat/tests/test_teslemetry.py index f1581f2ba..85cc67832 100644 --- a/apps/predbat/tests/test_teslemetry.py +++ b/apps/predbat/tests/test_teslemetry.py @@ -75,6 +75,10 @@ async def _request(self, method, path, json_body=None): } } +TARIFF_RATE_EXPORT_NOW = {"response": {"tariff_content_v2": {"version": 1, "utility": "Predbat", "code": "PREDBAT-EXPORT-NOW", "name": "Predbat (export_now)"}}} + +TARIFF_RATE_NORMAL = {"response": {"tariff_content_v2": {"version": 1, "utility": "Predbat", "code": "PREDBAT-NORMAL", "name": "Predbat (normal)"}}} + ENERGY_HISTORY = { "response": { "time_series": [ @@ -121,6 +125,16 @@ def test_teslemetry_site_info_publishes_soc_max(): assert api.dashboard_items["sensor.predbat_teslemetry_soc_max"]["state"] == 13.5 +def test_teslemetry_site_info_seeds_control_entity_states(): + """site_info seeds the operation_mode select and backup_reserve number entity states from the device (display only, no commands).""" + api = MockTeslemetryAPI() + api.mock_responses["/api/1/energy_sites/123456/site_info"] = SITE_INFO + run_async(api.fetch_site_info()) + assert api.entity_states["select.predbat_teslemetry_operation_mode"] == "self_consumption" + assert api.entity_states["number.predbat_teslemetry_backup_reserve"] == 20 + assert api.requests_made == [("GET", "/api/1/energy_sites/123456/site_info", None)] + + def test_teslemetry_energy_today_publishes_kwh(): """calendar_history energy series is aggregated into daily kWh sensors.""" api = MockTeslemetryAPI() @@ -282,6 +296,36 @@ def test_teslemetry_select_failure_keeps_state(): assert "select.predbat_teslemetry_operation_mode" not in api.entity_states +def test_teslemetry_command_success_on_low_response_code(): + """_command treats a response.code below 400 as success.""" + api = MockTeslemetryAPI() + api.mock_responses["/api/1/energy_sites/123456/operation"] = {"response": {"code": 201}} + assert run_async(api._command("operation", {"default_real_mode": "backup"})) is True + + +def test_teslemetry_command_failure_on_application_error_key(): + """_command treats a body carrying an "error" key as a failure even though _request returned parsed JSON.""" + api = MockTeslemetryAPI() + api.mock_responses["/api/1/energy_sites/123456/operation"] = {"error": "invalid_argument"} + assert run_async(api._command("operation", {"default_real_mode": "backup"})) is False + assert any("failed" in msg for msg in api.log_messages) + + +def test_teslemetry_command_failure_on_high_response_code(): + """_command treats an application-level response.code >= 400 as a failure despite a 2xx transport status.""" + api = MockTeslemetryAPI() + api.mock_responses["/api/1/energy_sites/123456/operation"] = {"response": {"code": 401}} + assert run_async(api._command("operation", {"default_real_mode": "backup"})) is False + assert any("failed" in msg for msg in api.log_messages) + + +def test_teslemetry_command_none_response_is_failure(): + """_command returns False when _request itself returns None (unchanged transport-failure behaviour).""" + api = MockTeslemetryAPI() + # No mock response registered -> _request returns None + assert run_async(api._command("operation", {"default_real_mode": "backup"})) is False + + def test_teslemetry_number_backup_reserve(): """number_event on backup_reserve POSTs /backup with an integer percent.""" api = MockTeslemetryAPI() @@ -363,23 +407,54 @@ def test_teslemetry_set_tariff_posts_tou_settings(): assert "tariff_content_v2" in body["tou_settings"] -def test_teslemetry_reconcile_on_start_restores_normal(): - """Boot reconciliation restores normal tariff + never export if left in export_now.""" +def test_teslemetry_reconcile_on_start_device_marker_restores_normal(): + """Boot reconciliation reads the ACTUAL device tariff (not the local entity mirror); a PREDBAT-EXPORT-NOW + marker on the device restores normal tariff + never export and updates the mirror entities on success.""" api = MockTeslemetryAPI() - api.entity_states["select.predbat_teslemetry_tariff_mode"] = "export_now" + # The entity mirror is always reseeded to "normal" by register_control_entities() on boot, so it must + # be irrelevant here - only the device response drives the outcome. + api.entity_states["select.predbat_teslemetry_tariff_mode"] = "normal" + api.mock_responses["/api/1/energy_sites/123456/tariff_rate"] = TARIFF_RATE_EXPORT_NOW api.mock_responses["/api/1/energy_sites/123456/time_of_use_settings"] = {"response": {"code": 201}} api.mock_responses["/api/1/energy_sites/123456/grid_import_export"] = {"response": {"code": 201}} run_async(api.reconcile_on_start()) + assert ("GET", "/api/1/energy_sites/123456/tariff_rate", None) in api.requests_made assert api.entity_states["select.predbat_teslemetry_tariff_mode"] == "normal" assert api.entity_states["select.predbat_teslemetry_allow_export"] == "never" + assert any("PREDBAT-EXPORT-NOW" in msg for msg in api.log_messages) -def test_teslemetry_reconcile_on_start_noop_when_normal(): - """Boot reconciliation does nothing when tariff mode is already normal.""" +def test_teslemetry_reconcile_on_start_device_normal_is_noop(): + """Boot reconciliation issues zero command requests when the device tariff is not the export_now marker.""" api = MockTeslemetryAPI() - api.entity_states["select.predbat_teslemetry_tariff_mode"] = "normal" + api.mock_responses["/api/1/energy_sites/123456/tariff_rate"] = TARIFF_RATE_NORMAL + run_async(api.reconcile_on_start()) + assert [req for req in api.requests_made if req[0] == "POST"] == [] + assert api.entity_states == {} + + +def test_teslemetry_reconcile_on_start_read_failure_skips_without_crash(): + """Boot reconciliation skips silently (no writes, no exception) when the device tariff read fails.""" + api = MockTeslemetryAPI() + # No mock response registered for tariff_rate -> _request returns None + run_async(api.reconcile_on_start()) + assert [req for req in api.requests_made if req[0] == "POST"] == [] + assert api.entity_states == {} + + +def test_teslemetry_reconcile_on_start_read_only_mode_skips_writes(): + """In read-only mode, a device tariff still marked PREDBAT-EXPORT-NOW is logged but not written back.""" + from types import SimpleNamespace + + api = MockTeslemetryAPI() + api.base = SimpleNamespace(set_read_only=True) + api.mock_responses["/api/1/energy_sites/123456/tariff_rate"] = TARIFF_RATE_EXPORT_NOW + api.mock_responses["/api/1/energy_sites/123456/time_of_use_settings"] = {"response": {"code": 201}} + api.mock_responses["/api/1/energy_sites/123456/grid_import_export"] = {"response": {"code": 201}} run_async(api.reconcile_on_start()) - assert api.requests_made == [] + assert [req for req in api.requests_made if req[0] == "POST"] == [] + assert api.entity_states == {} + assert any("read-only" in msg.lower() for msg in api.log_messages) def _assert_tou_periods_partition_day(tou_periods): @@ -439,7 +514,7 @@ def test_teslemetry_build_tariff_sell_clamp_above_export_rate(): def test_teslemetry_reconcile_on_start_partial_failure(): """Partial reconcile: tariff restore succeeds but the export-rule command fails, so only tariff_mode is updated.""" api = MockTeslemetryAPI() - api.entity_states["select.predbat_teslemetry_tariff_mode"] = "export_now" + api.mock_responses["/api/1/energy_sites/123456/tariff_rate"] = TARIFF_RATE_EXPORT_NOW api.mock_responses["/api/1/energy_sites/123456/time_of_use_settings"] = {"response": {"code": 201}} # grid_import_export mock ABSENT -> _request returns None -> set_export_rule fails run_async(api.reconcile_on_start()) @@ -456,6 +531,7 @@ def test_teslemetry(my_predbat=None): test_teslemetry_entity_names() test_teslemetry_live_status_publishes_sensors() test_teslemetry_site_info_publishes_soc_max() + test_teslemetry_site_info_seeds_control_entity_states() test_teslemetry_energy_today_publishes_kwh() test_teslemetry_request_auth_failure() test_teslemetry_request_rate_limit_retry() @@ -465,6 +541,10 @@ def test_teslemetry(my_predbat=None): test_teslemetry_run_auth_failed_only_probes_live_status() test_teslemetry_select_operation_mode() test_teslemetry_select_failure_keeps_state() + test_teslemetry_command_success_on_low_response_code() + test_teslemetry_command_failure_on_application_error_key() + test_teslemetry_command_failure_on_high_response_code() + test_teslemetry_command_none_response_is_failure() test_teslemetry_number_backup_reserve() test_teslemetry_switch_grid_charging() test_teslemetry_select_export_rule() @@ -472,8 +552,10 @@ def test_teslemetry(my_predbat=None): test_teslemetry_build_tariff_export_now() test_teslemetry_build_tariff_normal_flat() test_teslemetry_set_tariff_posts_tou_settings() - test_teslemetry_reconcile_on_start_restores_normal() - test_teslemetry_reconcile_on_start_noop_when_normal() + test_teslemetry_reconcile_on_start_device_marker_restores_normal() + test_teslemetry_reconcile_on_start_device_normal_is_noop() + test_teslemetry_reconcile_on_start_read_failure_skips_without_crash() + test_teslemetry_reconcile_on_start_read_only_mode_skips_writes() test_teslemetry_build_tariff_export_now_midnight_wrap() test_teslemetry_build_tariff_export_now_midnight_exact() test_teslemetry_build_tariff_sell_clamp_above_export_rate() From 1f2adf263f93b1e5e408334894bcb7f5f8ac11b3 Mon Sep 17 00:00:00 2001 From: Mark Gascoyne Date: Fri, 3 Jul 2026 12:46:47 +0100 Subject: [PATCH 08/14] fix(teslemetry): read set_read_only via get_arg at boot _is_read_only() read base.set_read_only directly, but the PredBat constructor defaults that attribute to True (predbat.py:504) and it is only refreshed from config in the fetch cycle (fetch.py:2401), which runs AFTER phase-1 components start - so at reconcile_on_start time the attribute was always True and the recovery write was unreachable again, with a misleading read-only log. load_user_config(load_config=True) runs before phase-1 start (predbat.py:1574 vs 1577), so reading via get_arg("set_read_only", False) returns the real configured value. Also split get_current_tariff_code's failure logging: "read failed" (no response) vs "no tariff code" (response without a code key). Co-Authored-By: Claude Fable 5 --- apps/predbat/teslemetry.py | 32 +++++++++++++++------ apps/predbat/tests/test_teslemetry.py | 41 +++++++++++++++++++++++++-- 2 files changed, 62 insertions(+), 11 deletions(-) diff --git a/apps/predbat/teslemetry.py b/apps/predbat/teslemetry.py index 164838eb7..6e956cbc9 100644 --- a/apps/predbat/teslemetry.py +++ b/apps/predbat/teslemetry.py @@ -340,16 +340,32 @@ def _find_tariff_code(node): return None async def get_current_tariff_code(self): - """Fetch the site's current tariff_rate from the device and return its tariff "code", or None on failure.""" + """Fetch the site's current tariff_rate from the device and return its tariff "code", or None on failure. + + Logs distinct warnings for the two failure modes: the read itself failing (network/HTTP error, empty + body) versus a successful read whose response simply carries no "code" key anywhere. + """ data = await self._request("GET", "/api/1/energy_sites/{}/tariff_rate".format(self.site_id)) if not data: + self.log("Warn: Teslemetry tariff_rate read failed - no response from the device") return None - return self._find_tariff_code(data) + code = self._find_tariff_code(data) + if code is None: + self.log("Warn: Teslemetry tariff_rate response contained no tariff code: {}".format(str(data)[:200])) + return code def _is_read_only(self): - """Return True if Predbat is configured read-only, so device-write commands must not be sent.""" - base = getattr(self, "base", None) - return bool(getattr(base, "set_read_only", False)) if base is not None else False + """Return True if Predbat is configured read-only, so device-write commands must not be sent. + + Reads the configuration via get_arg rather than the base's set_read_only attribute: the attribute + defaults to True in the PredBat constructor and is only refreshed from config during the fetch cycle, + which runs AFTER phase-1 components start - so at reconcile_on_start time the attribute would always + read True even for read-write instances. load_user_config(load_config=True) runs before phase-1 + component start, so get_arg returns the real configured value here. + """ + if getattr(self, "base", None) is None: + return False + return bool(self.get_arg("set_read_only", False)) async def reconcile_on_start(self): """Restore a safe state if a previous run died mid-export. @@ -360,12 +376,12 @@ async def reconcile_on_start(self): If the device's tariff still carries the PREDBAT-EXPORT-NOW marker written by build_tariff(), a previous run crashed mid-export, so normal tariff and disabled export are restored (unless Predbat is running read-only, in which case the need - for recovery is logged but no write is sent). If the tariff read itself fails, - this skips silently rather than guessing at the device state. + for recovery is logged but no write is sent). If no tariff code can be determined + (read failure or code-less response, each logged distinctly by + get_current_tariff_code), this skips rather than guessing at the device state. """ tariff_code = await self.get_current_tariff_code() if tariff_code is None: - self.log("Warn: Teslemetry could not read the current tariff from the device on boot - skipping reconcile") return export_now_code = self.build_tariff("export_now").get("code") if tariff_code != export_now_code: diff --git a/apps/predbat/tests/test_teslemetry.py b/apps/predbat/tests/test_teslemetry.py index 85cc67832..97e38d4d5 100644 --- a/apps/predbat/tests/test_teslemetry.py +++ b/apps/predbat/tests/test_teslemetry.py @@ -434,20 +434,33 @@ def test_teslemetry_reconcile_on_start_device_normal_is_noop(): def test_teslemetry_reconcile_on_start_read_failure_skips_without_crash(): - """Boot reconciliation skips silently (no writes, no exception) when the device tariff read fails.""" + """Boot reconciliation skips (no writes, no exception) when the device tariff read fails, logging a read failure.""" api = MockTeslemetryAPI() # No mock response registered for tariff_rate -> _request returns None run_async(api.reconcile_on_start()) assert [req for req in api.requests_made if req[0] == "POST"] == [] assert api.entity_states == {} + assert any("read failed" in msg for msg in api.log_messages) + assert not any("no tariff code" in msg for msg in api.log_messages) + + +def test_teslemetry_reconcile_on_start_no_code_in_response_skips(): + """Boot reconciliation skips with a DISTINCT log when the tariff read succeeds but the response carries no code key.""" + api = MockTeslemetryAPI() + api.mock_responses["/api/1/energy_sites/123456/tariff_rate"] = {"response": {"tariff_content_v2": {"version": 1, "utility": "SomeUtility"}}} + run_async(api.reconcile_on_start()) + assert [req for req in api.requests_made if req[0] == "POST"] == [] + assert api.entity_states == {} + assert any("no tariff code" in msg for msg in api.log_messages) + assert not any("read failed" in msg for msg in api.log_messages) def test_teslemetry_reconcile_on_start_read_only_mode_skips_writes(): - """In read-only mode, a device tariff still marked PREDBAT-EXPORT-NOW is logged but not written back.""" + """When get_arg reports read-only, a device tariff still marked PREDBAT-EXPORT-NOW is logged but not written back.""" from types import SimpleNamespace api = MockTeslemetryAPI() - api.base = SimpleNamespace(set_read_only=True) + api.base = SimpleNamespace(set_read_only=True, get_arg=lambda arg, default=None, **kwargs: True if arg == "set_read_only" else default) api.mock_responses["/api/1/energy_sites/123456/tariff_rate"] = TARIFF_RATE_EXPORT_NOW api.mock_responses["/api/1/energy_sites/123456/time_of_use_settings"] = {"response": {"code": 201}} api.mock_responses["/api/1/energy_sites/123456/grid_import_export"] = {"response": {"code": 201}} @@ -457,6 +470,26 @@ def test_teslemetry_reconcile_on_start_read_only_mode_skips_writes(): assert any("read-only" in msg.lower() for msg in api.log_messages) +def test_teslemetry_reconcile_on_start_ignores_boot_default_read_only_attribute(): + """The recovery write must NOT be gated on base.set_read_only: the constructor defaults that attribute to True + and it is only refreshed from config in the fetch cycle AFTER phase-1 components start, so at reconcile time it + is always True. With get_arg (the real config source) reporting False, the writes must go ahead regardless.""" + from types import SimpleNamespace + + api = MockTeslemetryAPI() + # Simulate the real boot condition: stale constructor default True on the attribute, real config value False. + api.base = SimpleNamespace(set_read_only=True, get_arg=lambda arg, default=None, **kwargs: False if arg == "set_read_only" else default) + api.mock_responses["/api/1/energy_sites/123456/tariff_rate"] = TARIFF_RATE_EXPORT_NOW + api.mock_responses["/api/1/energy_sites/123456/time_of_use_settings"] = {"response": {"code": 201}} + api.mock_responses["/api/1/energy_sites/123456/grid_import_export"] = {"response": {"code": 201}} + run_async(api.reconcile_on_start()) + assert any(req[1].endswith("/time_of_use_settings") for req in api.requests_made if req[0] == "POST") + assert any(req[1].endswith("/grid_import_export") for req in api.requests_made if req[0] == "POST") + assert api.entity_states["select.predbat_teslemetry_tariff_mode"] == "normal" + assert api.entity_states["select.predbat_teslemetry_allow_export"] == "never" + assert not any("read-only" in msg.lower() for msg in api.log_messages) + + def _assert_tou_periods_partition_day(tou_periods): """Assert the tou_periods cover every minute of the (circular) day exactly once — no overlaps, no gaps.""" covered = [0] * (24 * 60) @@ -555,7 +588,9 @@ def test_teslemetry(my_predbat=None): test_teslemetry_reconcile_on_start_device_marker_restores_normal() test_teslemetry_reconcile_on_start_device_normal_is_noop() test_teslemetry_reconcile_on_start_read_failure_skips_without_crash() + test_teslemetry_reconcile_on_start_no_code_in_response_skips() test_teslemetry_reconcile_on_start_read_only_mode_skips_writes() + test_teslemetry_reconcile_on_start_ignores_boot_default_read_only_attribute() test_teslemetry_build_tariff_export_now_midnight_wrap() test_teslemetry_build_tariff_export_now_midnight_exact() test_teslemetry_build_tariff_sell_clamp_above_export_rate() From 213f36ab16e4d6cb0421a3123b52480772eb084e Mon Sep 17 00:00:00 2001 From: "pre-commit-ci-lite[bot]" <117423508+pre-commit-ci-lite[bot]@users.noreply.github.com> Date: Fri, 3 Jul 2026 12:48:54 +0000 Subject: [PATCH 09/14] [pre-commit.ci lite] apply automatic fixes --- .cspell/custom-dictionary-workspace.txt | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.cspell/custom-dictionary-workspace.txt b/.cspell/custom-dictionary-workspace.txt index 4c7c33ff2..2146b7805 100644 --- a/.cspell/custom-dictionary-workspace.txt +++ b/.cspell/custom-dictionary-workspace.txt @@ -395,8 +395,8 @@ TCWORLD tdata tdiff temperation -Teslemetry teslemetry +Teslemetry testname thirdparty timea From 8443eac351eccec4a96ec538b371894a9f5f0275 Mon Sep 17 00:00:00 2001 From: Mark Gascoyne Date: Mon, 6 Jul 2026 11:12:44 +0100 Subject: [PATCH 10/14] =?UTF-8?q?fix(teslemetry):=20xhigh=20review=20?= =?UTF-8?q?=E2=80=94=20local-time=20tariff,=20reconcile=20latch,=20soc=5Fm?= =?UTF-8?q?ax=20retry,=20calendar=5Fhistory=20query,=20control=20attrs,=20?= =?UTF-8?q?number=20guard?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Fixes six code-review findings, each with a regression test confirmed red before/green after: - build_tariff now derives `now` from the base's local time (base.now_utc.astimezone(base.local_tz)) instead of raw UTC, so the export_now ON_PEAK window lands on the Powerwall's actual local wall-clock boundary instead of being off by the site's UTC offset. - run() gates reconcile_on_start() on a new `reconcile_done` latch (mirroring site_info_done) instead of the `first` flag, so a boot that 401s (consuming `first`) and later recovers still reconciles a device stuck exporting. reconcile_on_start() now returns whether it actually got a usable tariff read, so the latch only sets once that happened. - fetch_site_info() only returns True once soc_max was actually published; when nameplate_energy is absent/zero it still seeds the other control-entity fields but returns False, so site_info_done stays False and it retries. - fetch_energy_today() now requests calendar_history with kind=energy&period=day, which the Fleet/Teslemetry endpoint requires. - All control-entity writes (select/number/switch events, reconcile's restore, site_info seeding) now route through a new publish_control() helper that re-applies the options/min/max/friendly_name register_control_entities recorded at init, instead of wiping them via a bare set_state_wrapper call. - number_event() guards the int(float(value)) conversion against a null value (e.g. from a number.increment/decrement service call) instead of raising, matching gateway.py's pattern. Co-Authored-By: Claude Opus 4.8 (1M context) --- apps/predbat/teslemetry.py | 121 ++++++++++++++++----- apps/predbat/tests/test_teslemetry.py | 147 +++++++++++++++++++++++++- 2 files changed, 240 insertions(+), 28 deletions(-) diff --git a/apps/predbat/teslemetry.py b/apps/predbat/teslemetry.py index 6e956cbc9..c5c192092 100644 --- a/apps/predbat/teslemetry.py +++ b/apps/predbat/teslemetry.py @@ -58,6 +58,7 @@ def initialize(self, key="", site_id="", base_url=TESLEMETRY_DEFAULT_URL, **kwar self.last_live_poll = 0 self.last_energy_poll = 0 self.site_info_done = False + self.reconcile_done = False self.log("Info: TeslemetryAPI initialising site_id={}".format(self.site_id)) self.register_control_entities() @@ -115,27 +116,39 @@ async def fetch_live_status(self): return True async def fetch_site_info(self): - """Fetch site info, publishing the battery capacity as soc_max and seeding control entity states.""" + """Fetch site info, publishing the battery capacity as soc_max and seeding control entity states. + + Returns True only once soc_max has actually been published. If nameplate_energy is absent or + zero, control-entity seeding still happens with whatever fields are present, but this returns + False so the run()-level site_info_done latch stays False and the fetch is retried on a later + cycle instead of permanently giving up on soc_max. + """ data = await self._request("GET", "/api/1/energy_sites/{}/site_info".format(self.site_id)) if not data: return False response = data.get("response", {}) nameplate_wh = response.get("nameplate_energy", 0) + soc_max_published = False if nameplate_wh: self.publish_sensor("soc_max", round(nameplate_wh / 1000.0, 2), unit="kWh", state_class=None, friendly="Powerwall Capacity") + soc_max_published = True # Seed the control entity STATES (display only, no commands) from the device so they reflect # reality at boot instead of the hardcoded defaults set by register_control_entities(). default_mode = response.get("default_real_mode") if default_mode in OPERATION_MODES: - self.set_state_wrapper(self.entity("operation_mode", domain="select"), default_mode) + self.publish_control(self.entity("operation_mode", domain="select"), default_mode) backup_reserve = response.get("backup_reserve_percent") if isinstance(backup_reserve, (int, float)): - self.set_state_wrapper(self.entity("backup_reserve", domain="number"), backup_reserve) - return True + self.publish_control(self.entity("backup_reserve", domain="number"), backup_reserve) + return soc_max_published async def fetch_energy_today(self): - """Fetch today's cumulative energy, publishing daily kWh sensors.""" - data = await self._request("GET", "/api/1/energy_sites/{}/calendar_history".format(self.site_id)) + """Fetch today's cumulative energy, publishing daily kWh sensors. + + The Fleet/Teslemetry calendar_history endpoint requires a "kind" and a "period" query + parameter; without them it does not reliably return the daily energy time_series. + """ + data = await self._request("GET", "/api/1/energy_sites/{}/calendar_history?kind=energy&period=day".format(self.site_id)) if not data: return False series = data.get("response", {}).get("time_series", []) @@ -157,6 +170,12 @@ async def run(self, seconds=0, first=False): Returns True on a successful (or gated no-op) cycle and False on failure — ComponentBase.start() uses this to set api_started, clear the startup flag and drive its retry backoff, so a failing cycle must return False. + + Boot reconciliation is gated on the `reconcile_done` latch (mirroring the `site_info_done` + pattern) rather than on `first`: if the very first cycle 401s, the auth-failed fast-path + above returns before reconcile ever runs, but `first` is still consumed by that cycle. Gating + on a latch instead means reconcile still runs on the first cycle that is NOT auth-failed, + however many cycles that takes to arrive - a boot that 401s then recovers still reconciles. """ if not self.api_key or not self.site_id: self.log("Warn: Teslemetry key or site_id not configured, skipping run") @@ -172,8 +191,8 @@ async def run(self, seconds=0, first=False): success = True if not self.site_info_done: self.site_info_done = await self.fetch_site_info() - if first: - await self.reconcile_on_start() + if not self.reconcile_done: + self.reconcile_done = await self.reconcile_on_start() if first or (seconds - self.last_live_poll >= LIVE_POLL_SECONDS): self.last_live_poll = seconds success = await self.fetch_live_status() @@ -183,12 +202,37 @@ async def run(self, seconds=0, first=False): return success def register_control_entities(self): - """Publish the virtual control entities with their initial states.""" - self.dashboard_item(self.entity("operation_mode", domain="select"), "self_consumption", {"friendly_name": "Powerwall Operation Mode", "options": OPERATION_MODES}, app="teslemetry") - self.dashboard_item(self.entity("backup_reserve", domain="number"), 0, {"friendly_name": "Powerwall Backup Reserve", "min": 0, "max": 100, "step": 1, "unit_of_measurement": "%"}, app="teslemetry") - self.dashboard_item(self.entity("allow_charging_from_grid", domain="switch"), "on", {"friendly_name": "Powerwall Allow Grid Charging"}, app="teslemetry") - self.dashboard_item(self.entity("allow_export", domain="select"), "never", {"friendly_name": "Powerwall Allow Export", "options": EXPORT_RULES}, app="teslemetry") - self.dashboard_item(self.entity("tariff_mode", domain="select"), "normal", {"friendly_name": "Powerwall Tariff Mode", "options": TARIFF_MODES}, app="teslemetry") + """Publish the virtual control entities with their initial states. + + Also records each entity's display attributes (options/min/max/friendly_name) in + self.control_attrs, so publish_control can re-apply them on every subsequent write instead + of a bare value-only write silently replacing them with an empty attributes dict. + """ + controls = ( + ("operation_mode", "select", "self_consumption", {"friendly_name": "Powerwall Operation Mode", "options": OPERATION_MODES}), + ("backup_reserve", "number", 0, {"friendly_name": "Powerwall Backup Reserve", "min": 0, "max": 100, "step": 1, "unit_of_measurement": "%"}), + ("allow_charging_from_grid", "switch", "on", {"friendly_name": "Powerwall Allow Grid Charging"}), + ("allow_export", "select", "never", {"friendly_name": "Powerwall Allow Export", "options": EXPORT_RULES}), + ("tariff_mode", "select", "normal", {"friendly_name": "Powerwall Tariff Mode", "options": TARIFF_MODES}), + ) + self.control_attrs = {} + for suffix, domain, initial_state, attributes in controls: + entity_id = self.entity(suffix, domain=domain) + self.control_attrs[entity_id] = attributes + self.dashboard_item(entity_id, initial_state, attributes, app="teslemetry") + + def publish_control(self, entity_id, value): + """Write a control entity's new state, re-applying the attributes register_control_entities + recorded for it at init (options/min/max/friendly_name). + + A bare set_state_wrapper(entity, value) write defaults attributes to {}, which silently + replaces the options/min/max/friendly_name published once at startup and never resent, so + every select/number/switch write - and the reconcile/site_info writes that mirror device + state into these same entities - must route through here instead of calling + set_state_wrapper directly. + """ + attrs = getattr(self, "control_attrs", {}).get(entity_id, {}) + self.dashboard_item(entity_id, value, attrs, app="teslemetry") async def _command(self, path, body): """POST a command; return True only if both the transport and the application layer succeeded. @@ -253,9 +297,22 @@ def build_tariff(self, mode, now=None): export_now: current 30-minute-aligned window becomes ON_PEAK with a high sell price so autonomous mode exports immediately; the window self-expires. normal: flat tariff from the customer's current rates. + + The Powerwall applies tou_periods windows in the SITE'S LOCAL wall-clock time (the same + clock Predbat schedules against), not UTC, so when `now` is not injected (production use) + it defaults to the base's local time - base.now_utc converted to base.local_tz - rather than + a bare UTC clock read, which would misplace the window by the site's UTC offset. Falls back + to system UTC only when no base is wired up (e.g. some unit tests), since there is then no + site-local timezone to derive from. """ if now is None: - now = datetime.now(timezone.utc) + base = getattr(self, "base", None) + base_now_utc = getattr(base, "now_utc", None) if base is not None else None + if base_now_utc is not None: + local_tz = getattr(base, "local_tz", None) or getattr(self, "local_tz", None) or timezone.utc + now = base_now_utc.astimezone(local_tz) + else: + now = datetime.now(timezone.utc) import_rate, export_rate = self.current_rates() # Keep the synthetic ON_PEAK sell strictly above the live export rate, otherwise the # export trick silently inverts at the highest-value moments. @@ -379,21 +436,27 @@ async def reconcile_on_start(self): for recovery is logged but no write is sent). If no tariff code can be determined (read failure or code-less response, each logged distinctly by get_current_tariff_code), this skips rather than guessing at the device state. + + Returns True once this has actually executed against a live API (the tariff read + returned a usable code, whether or not a restore was needed), so run() can latch + reconcile_done and stop retrying. Returns False when the tariff read itself failed to + return anything usable, so run() retries on a later cycle instead of latching a no-op. """ tariff_code = await self.get_current_tariff_code() if tariff_code is None: - return + return False export_now_code = self.build_tariff("export_now").get("code") if tariff_code != export_now_code: - return + return True if self._is_read_only(): self.log("Warn: Teslemetry device tariff was left as {} - recovery needed but skipped (read-only mode)".format(tariff_code)) - return + return True self.log("Warn: Teslemetry device tariff was left as {} - restoring normal tariff and disabling export".format(tariff_code)) if await self.set_tariff("normal"): - self.set_state_wrapper(self.entity("tariff_mode", domain="select"), "normal") + self.publish_control(self.entity("tariff_mode", domain="select"), "normal") if await self.set_export_rule("never"): - self.set_state_wrapper(self.entity("allow_export", domain="select"), "never") + self.publish_control(self.entity("allow_export", domain="select"), "never") + return True async def select_event(self, entity_id, value): """Handle select changes routed from the service-hook loopback.""" @@ -408,16 +471,24 @@ async def select_event(self, entity_id, value): self.log("Warn: Teslemetry unhandled select_event {} = {}".format(entity_id, value)) return if success: - self.set_state_wrapper(entity_id, value) + self.publish_control(entity_id, value) else: self.log("Warn: Teslemetry command failed for {} = {} (state not updated)".format(entity_id, value)) async def number_event(self, entity_id, value): - """Handle number changes routed from the service-hook loopback.""" + """Handle number changes routed from the service-hook loopback. + + Guards the int(float(value)) conversion: a number.increment/decrement service call can + forward value=None, which would otherwise raise a TypeError out of the dispatch loop. + """ if entity_id.endswith("_backup_reserve"): - percent = max(0, min(100, int(float(value)))) + try: + percent = max(0, min(100, int(float(value)))) + except (ValueError, TypeError): + self.log("Warn: Teslemetry invalid number_event value for {}: {}".format(entity_id, value)) + return if await self.set_backup_reserve(percent): - self.set_state_wrapper(entity_id, percent) + self.publish_control(entity_id, percent) else: self.log("Warn: Teslemetry backup_reserve command failed (state not updated)") else: @@ -428,7 +499,7 @@ async def switch_event(self, entity_id, service): if entity_id.endswith("_allow_charging_from_grid"): allow = service != "turn_off" if await self.set_grid_charging(allow): - self.set_state_wrapper(entity_id, "on" if allow else "off") + self.publish_control(entity_id, "on" if allow else "off") else: self.log("Warn: Teslemetry grid charging command failed (state not updated)") else: diff --git a/apps/predbat/tests/test_teslemetry.py b/apps/predbat/tests/test_teslemetry.py index 97e38d4d5..1ad74080f 100644 --- a/apps/predbat/tests/test_teslemetry.py +++ b/apps/predbat/tests/test_teslemetry.py @@ -8,7 +8,7 @@ from unittest.mock import MagicMock, patch, AsyncMock from tests.test_infra import create_aiohttp_mock_response, create_aiohttp_mock_session, run_async -from teslemetry import TeslemetryAPI +from teslemetry import TeslemetryAPI, OPERATION_MODES class MockTeslemetryAPI(TeslemetryAPI): @@ -24,6 +24,7 @@ def __init__(self): self.last_live_poll = 0 self.last_energy_poll = 0 self.site_info_done = False + self.reconcile_done = False self.dashboard_items = {} self.log_messages = [] self.entity_states = {} @@ -35,8 +36,10 @@ def log(self, msg): self.log_messages.append(msg) def dashboard_item(self, entity, state, attributes, app=None): - """Capture dashboard items.""" + """Capture dashboard items (mirrors production dashboard_item, which also calls through to + set_state_wrapper - so entity_states reflects the latest write here too).""" self.dashboard_items[entity] = {"state": state, "attributes": attributes} + self.entity_states[entity] = state def set_state_wrapper(self, entity_id, state, attributes={}): """Capture entity state updates.""" @@ -138,7 +141,7 @@ def test_teslemetry_site_info_seeds_control_entity_states(): def test_teslemetry_energy_today_publishes_kwh(): """calendar_history energy series is aggregated into daily kWh sensors.""" api = MockTeslemetryAPI() - api.mock_responses["/api/1/energy_sites/123456/calendar_history"] = ENERGY_HISTORY + api.mock_responses["/api/1/energy_sites/123456/calendar_history?kind=energy&period=day"] = ENERGY_HISTORY run_async(api.fetch_energy_today()) assert api.dashboard_items["sensor.predbat_teslemetry_solar_today"]["state"] == 4.0 assert api.dashboard_items["sensor.predbat_teslemetry_import_today"]["state"] == 2.5 @@ -555,6 +558,136 @@ def test_teslemetry_reconcile_on_start_partial_failure(): assert "select.predbat_teslemetry_allow_export" not in api.entity_states +def test_teslemetry_build_tariff_uses_site_local_time_not_utc(): + """build_tariff's export_now window must be expressed in the SITE'S LOCAL wall-clock time (the + same clock Predbat schedules in), not raw UTC - the Powerwall applies tou_periods in local time. + + A base whose local time is 15:40 during BST (UTC+1) must yield an ON_PEAK window starting at + 15:30 local. A naive UTC-only implementation would instead read whatever the system clock (or, + as sabotaged here, a mocked bogus UTC time) reports, producing the wrong window boundaries. + """ + import pytz + from datetime import datetime as real_datetime + from types import SimpleNamespace + + api = MockTeslemetryAPI() + london = pytz.timezone("Europe/London") + local_now = london.localize(real_datetime(2026, 7, 2, 15, 40)) + api.base = SimpleNamespace(now_utc=local_now, local_tz=london, minutes_now=None, rate_import=None, rate_export=None) + + # Sabotage a naive implementation: if build_tariff ignores the base and calls datetime.now(timezone.utc) + # directly, it will see this deliberately-wrong bogus time (03:05) instead of the real 15:40 local time above. + with patch("teslemetry.datetime") as mock_datetime: + mock_datetime.now.return_value = real_datetime(2026, 7, 2, 3, 5) + tariff = api.build_tariff("export_now") + + periods = tariff["seasons"]["AllYear"]["tou_periods"]["ON_PEAK"]["periods"] + assert periods[0]["fromHour"] == 15 and periods[0]["fromMinute"] == 30 + + +def test_teslemetry_reconcile_latch_survives_auth_failed_first_cycle(): + """A first cycle that 401s (auth-failed, consuming `first`) must not permanently skip reconcile: + once the API recovers on a later cycle, reconcile_on_start still runs and restores a stuck export + state, because reconcile is gated on a `reconcile_done` latch rather than on `first`.""" + api, calls = _make_run_api_with_fetch_capture() + api.api_auth_failed = True # Simulate: the boot cycle already found the token dead. + assert api.reconcile_done is False + # Cycle 1 (first=True): auth-failed fast path, poll not yet due -> no HTTP traffic at all, no reconcile chance. + assert run_async(api.run(0, True)) is False + assert calls == [] + assert api.reconcile_done is False + + # Auth recovers (e.g. a live_status probe on an intervening cycle succeeded and cleared the flag). + api.api_auth_failed = False + api.mock_responses["/api/1/energy_sites/123456/tariff_rate"] = TARIFF_RATE_EXPORT_NOW + api.mock_responses["/api/1/energy_sites/123456/time_of_use_settings"] = {"response": {"code": 201}} + api.mock_responses["/api/1/energy_sites/123456/grid_import_export"] = {"response": {"code": 201}} + # Cycle 2 (first=False): reconcile must still run even though `first` was already consumed in cycle 1. + assert run_async(api.run(300, False)) is True + assert api.entity_states["select.predbat_teslemetry_tariff_mode"] == "normal" + assert api.entity_states["select.predbat_teslemetry_allow_export"] == "never" + assert api.reconcile_done is True + + +def test_teslemetry_reconcile_latch_runs_once_on_healthy_boot(): + """A healthy first cycle runs reconcile exactly once; the latch prevents a redundant reconcile + call (and its tariff_rate read) on every subsequent cycle.""" + api, calls = _make_run_api_with_fetch_capture() + api.mock_responses["/api/1/energy_sites/123456/tariff_rate"] = TARIFF_RATE_NORMAL + assert run_async(api.run(0, True)) is True + assert api.reconcile_done is True + tariff_reads = [req for req in api.requests_made if req[1].endswith("/tariff_rate")] + assert len(tariff_reads) == 1 + # A later cycle must not repeat the reconcile tariff_rate read. + assert run_async(api.run(600, False)) is True + tariff_reads = [req for req in api.requests_made if req[1].endswith("/tariff_rate")] + assert len(tariff_reads) == 1 + + +def test_teslemetry_site_info_returns_false_without_nameplate_so_soc_max_retries(): + """fetch_site_info must return False when nameplate_energy is absent/zero so run()'s + site_info_done latch stays False and soc_max is retried on a later cycle - even though other + fields (control-entity seeding) were published from the same response.""" + api = MockTeslemetryAPI() + site_info_no_nameplate = {"response": {"nameplate_energy": 0, "default_real_mode": "self_consumption", "backup_reserve_percent": 20}} + api.mock_responses["/api/1/energy_sites/123456/site_info"] = site_info_no_nameplate + assert run_async(api.fetch_site_info()) is False + assert "sensor.predbat_teslemetry_soc_max" not in api.dashboard_items + # Control-entity seeding still happens even though soc_max could not be published. + assert api.entity_states["select.predbat_teslemetry_operation_mode"] == "self_consumption" + + # Nameplate becomes available on a later cycle - now it must succeed and return True. + api.mock_responses["/api/1/energy_sites/123456/site_info"] = SITE_INFO + assert run_async(api.fetch_site_info()) is True + assert api.dashboard_items["sensor.predbat_teslemetry_soc_max"]["state"] == 13.5 + + +def test_teslemetry_run_site_info_latch_stays_false_without_nameplate(): + """At the run() level, a site_info response without nameplate_energy must leave site_info_done + False so the fetch is retried on a later cycle, rather than latching done with no soc_max ever + published.""" + api = MockTeslemetryAPI() + api.fetch_site_info = TeslemetryAPI.fetch_site_info.__get__(api) + api.mock_responses["/api/1/energy_sites/123456/site_info"] = {"response": {"nameplate_energy": 0}} + run_async(api.run(0, True)) + assert api.site_info_done is False + + +def test_teslemetry_energy_today_requests_kind_and_period(): + """fetch_energy_today must query calendar_history with kind=energy&period=day - the Fleet/Teslemetry + endpoint requires both, and omitting them (the previous bare-path request) produces an invalid or + empty response.""" + api = MockTeslemetryAPI() + api.mock_responses["/api/1/energy_sites/123456/calendar_history?kind=energy&period=day"] = ENERGY_HISTORY + run_async(api.fetch_energy_today()) + assert api.requests_made == [("GET", "/api/1/energy_sites/123456/calendar_history?kind=energy&period=day", None)] + assert api.dashboard_items["sensor.predbat_teslemetry_solar_today"]["state"] == 4.0 + + +def test_teslemetry_select_event_preserves_control_attributes(): + """select_event must not wipe the options/min/max/friendly_name attributes that + register_control_entities set at init - every control write must re-apply them via + publish_control, verified here through the dashboard_items capture (which records attributes, + unlike the bare entity_states mirror).""" + api = MockTeslemetryAPI() + api.register_control_entities() + api.mock_responses["/api/1/energy_sites/123456/operation"] = {"response": {"code": 201}} + run_async(api.select_event("select.predbat_teslemetry_operation_mode", "backup")) + item = api.dashboard_items["select.predbat_teslemetry_operation_mode"] + assert item["state"] == "backup" + assert item["attributes"].get("options") == OPERATION_MODES + + +def test_teslemetry_number_event_guards_null_value(): + """number_event must not crash on a null value (e.g. forwarded from a number.increment/decrement + service call without an explicit value) - it should log a warning and post no command.""" + api = MockTeslemetryAPI() + run_async(api.number_event("number.predbat_teslemetry_backup_reserve", None)) + assert api.requests_made == [] + assert "number.predbat_teslemetry_backup_reserve" not in api.entity_states + assert any("invalid" in msg.lower() for msg in api.log_messages) + + def test_teslemetry(my_predbat=None): """Run all Teslemetry component tests (registry entry point). @@ -595,5 +728,13 @@ def test_teslemetry(my_predbat=None): test_teslemetry_build_tariff_export_now_midnight_exact() test_teslemetry_build_tariff_sell_clamp_above_export_rate() test_teslemetry_reconcile_on_start_partial_failure() + test_teslemetry_build_tariff_uses_site_local_time_not_utc() + test_teslemetry_reconcile_latch_survives_auth_failed_first_cycle() + test_teslemetry_reconcile_latch_runs_once_on_healthy_boot() + test_teslemetry_site_info_returns_false_without_nameplate_so_soc_max_retries() + test_teslemetry_run_site_info_latch_stays_false_without_nameplate() + test_teslemetry_energy_today_requests_kind_and_period() + test_teslemetry_select_event_preserves_control_attributes() + test_teslemetry_number_event_guards_null_value() print("**** Teslemetry tests passed ****") return 0 From d5f404801def42fb100c6c066766cae9576a2bb6 Mon Sep 17 00:00:00 2001 From: "pre-commit-ci-lite[bot]" <117423508+pre-commit-ci-lite[bot]@users.noreply.github.com> Date: Mon, 6 Jul 2026 11:04:10 +0000 Subject: [PATCH 11/14] [pre-commit.ci lite] apply automatic fixes --- .cspell/custom-dictionary-workspace.txt | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.cspell/custom-dictionary-workspace.txt b/.cspell/custom-dictionary-workspace.txt index 2146b7805..4c7c33ff2 100644 --- a/.cspell/custom-dictionary-workspace.txt +++ b/.cspell/custom-dictionary-workspace.txt @@ -395,8 +395,8 @@ TCWORLD tdata tdiff temperation -teslemetry Teslemetry +teslemetry testname thirdparty timea From 1026e6d84f14f39c33508c216cee70a9ccbcb2d5 Mon Sep 17 00:00:00 2001 From: Mark Gascoyne Date: Mon, 6 Jul 2026 21:50:45 +0100 Subject: [PATCH 12/14] feat(teslemetry): dedupe control commands (write-on-change) to cut Teslemetry command-credit spend Route the 5 control writes (operation_mode, backup_reserve, grid_charging, export_rule, tariff) through a per-key _last_sent signature cache so a POST only fires when the value actually changed, instead of every 5-minute loopback cycle from the repeat:true service hooks. The cache is only updated on a confirmed-successful send, so a failed command still retries next cycle, and fetch_site_info refreshes operation_mode/backup_reserve to the device's actual state so external drift (e.g. changed in the Tesla app) is detected and re-asserted. reconcile_on_start forces its tariff and export-rule recovery writes so a future cache-preseed can never make recovery look like a no-op. Co-Authored-By: Claude Opus 4.8 (1M context) --- .cspell/custom-dictionary-workspace.txt | 3 + apps/predbat/teslemetry.py | 92 +++++++++++++---- apps/predbat/tests/test_teslemetry.py | 127 ++++++++++++++++++++++++ 3 files changed, 205 insertions(+), 17 deletions(-) diff --git a/.cspell/custom-dictionary-workspace.txt b/.cspell/custom-dictionary-workspace.txt index 4c7c33ff2..d1f605224 100644 --- a/.cspell/custom-dictionary-workspace.txt +++ b/.cspell/custom-dictionary-workspace.txt @@ -85,6 +85,7 @@ dayname daynumber daysymbol dedup +dedupe dend denorm derating @@ -306,6 +307,8 @@ predai predbat predbatt predheat +preseed +preseeded prevs psum pvbat diff --git a/apps/predbat/teslemetry.py b/apps/predbat/teslemetry.py index c5c192092..adf3ae272 100644 --- a/apps/predbat/teslemetry.py +++ b/apps/predbat/teslemetry.py @@ -15,9 +15,18 @@ Control path: exposes virtual select/number/switch entities driven by the TESLA service hooks; handlers issue REST commands (operation mode, backup reserve, grid import/export rules and the export tariff-trick). + +Command dedupe: the TESLA service hooks carry `repeat: true`, so Predbat fires +each hook's loopback event every cycle (~5 minutes) even when nothing has +changed. Every control write is routed through `_apply_command`, which only +POSTs when the value actually differs from the last CONFIRMED-SENT value, +cutting Teslemetry command-credit spend without losing failure-retry (a +failed POST never updates the cache) or external-drift correction (see +`fetch_site_info`). """ import asyncio +import json from datetime import datetime, timezone import aiohttp @@ -59,6 +68,7 @@ def initialize(self, key="", site_id="", base_url=TESLEMETRY_DEFAULT_URL, **kwar self.last_energy_poll = 0 self.site_info_done = False self.reconcile_done = False + self._last_sent = {} self.log("Info: TeslemetryAPI initialising site_id={}".format(self.site_id)) self.register_control_entities() @@ -137,9 +147,15 @@ async def fetch_site_info(self): default_mode = response.get("default_real_mode") if default_mode in OPERATION_MODES: self.publish_control(self.entity("operation_mode", domain="select"), default_mode) + # Drift correction: refresh the dedupe cache to the device's ACTUAL mode. If it matches what + # Predbat last sent, the cache is unchanged (no spurious resend); if the user changed it + # externally (e.g. the Tesla app), the cache now reflects that drift, so the next assertion + # of Predbat's desired mode finds a mismatch and re-POSTs instead of being silently skipped. + self._last_sent["operation_mode"] = default_mode backup_reserve = response.get("backup_reserve_percent") if isinstance(backup_reserve, (int, float)): self.publish_control(self.entity("backup_reserve", domain="number"), backup_reserve) + self._last_sent["backup_reserve"] = int(backup_reserve) # Same drift-correction rationale as operation_mode above. return soc_max_published async def fetch_energy_today(self): @@ -256,21 +272,50 @@ async def _command(self, path, body): return False return True - async def set_operation_mode(self, mode): - """Set the Powerwall operation mode (self_consumption/autonomous/backup).""" - return await self._command("operation", {"default_real_mode": mode}) + async def _apply_command(self, key, signature, sender, force=False): + """Send a control command only when its value has actually changed, to cut Teslemetry command-credit spend. + + `key` identifies which control this is ("operation_mode", "backup_reserve", "grid_charging", + "export_rule" or "tariff") and `signature` is a comparable representation of the value being + asserted (e.g. the mode string, or a canonical JSON serialisation of a built tariff body). + When `signature` already matches the last confirmed-sent value for `key`, `sender` is not + called at all - this is the actual command-credit saving, since the TESLA service hooks fire + their loopback event every cycle regardless of whether anything changed. + + Invariant (failure-retry): the dedupe cache is updated ONLY after a confirmed-successful send + (a True return from `sender`), never on a skip or a failure. A dropped/failed command therefore + leaves the cache stale, so the very next cycle's loopback event re-sends the same value instead + of being silently skipped forever - failure-retry is preserved even though every-cycle sends of + an unchanged value are deduped away. + + `force=True` bypasses the cache check and always calls `sender`, still refreshing the cache on + success. This is used by `reconcile_on_start` so its corrective writes can never be silently + skipped by a (present or future) pre-seeded cache entry that happens to already match what the + recovery is about to send. + """ + if not force and self._last_sent.get(key) == signature: + return True + ok = await sender() + if ok: + self._last_sent[key] = signature + return ok + + async def set_operation_mode(self, mode, force=False): + """Set the Powerwall operation mode (self_consumption/autonomous/backup), deduped on write-on-change.""" + return await self._apply_command("operation_mode", mode, lambda: self._command("operation", {"default_real_mode": mode}), force=force) - async def set_backup_reserve(self, percent): - """Set the backup reserve percentage (cloud caps at 80; 80-100 treated as 100 by Predbat).""" - return await self._command("backup", {"backup_reserve_percent": int(percent)}) + async def set_backup_reserve(self, percent, force=False): + """Set the backup reserve percentage (cloud caps at 80; 80-100 treated as 100 by Predbat), deduped on write-on-change.""" + percent = int(percent) + return await self._apply_command("backup_reserve", percent, lambda: self._command("backup", {"backup_reserve_percent": percent}), force=force) - async def set_grid_charging(self, allow): - """Allow or disallow charging the battery from the grid.""" - return await self._command("grid_import_export", {"disallow_charge_from_grid_with_solar_installed": not allow}) + async def set_grid_charging(self, allow, force=False): + """Allow or disallow charging the battery from the grid, deduped on write-on-change.""" + return await self._apply_command("grid_charging", bool(allow), lambda: self._command("grid_import_export", {"disallow_charge_from_grid_with_solar_installed": not allow}), force=force) - async def set_export_rule(self, rule): - """Set the grid export rule (never/pv_only/battery_ok).""" - return await self._command("grid_import_export", {"customer_preferred_export_rule": rule}) + async def set_export_rule(self, rule, force=False): + """Set the grid export rule (never/pv_only/battery_ok), deduped on write-on-change.""" + return await self._apply_command("export_rule", rule, lambda: self._command("grid_import_export", {"customer_preferred_export_rule": rule}), force=force) def current_rates(self): """Return (import_rate, export_rate) in GBP/kWh from Predbat's rate data when available. @@ -368,10 +413,18 @@ def charges(off_peak_buy, on_peak_buy): **common, } - async def set_tariff(self, mode): - """Push the tariff for the requested mode via time_of_use_settings.""" + async def set_tariff(self, mode, force=False): + """Push the tariff for the requested mode via time_of_use_settings, deduped on write-on-change. + + The dedupe signature is a canonical (sort_keys) JSON serialisation of the BUILT tariff body + rather than just `mode`: the export_now window and sell price are time-varying (the ON_PEAK + window advances every 30 minutes and the sell price tracks the live export rate), so the same + `mode` string can legitimately need re-sending when the underlying built tariff has changed, + while an unchanged build (the common case every 5-minute cycle) is skipped. + """ tariff = self.build_tariff(mode) - return await self._command("time_of_use_settings", {"tou_settings": {"tariff_content_v2": tariff}}) + signature = json.dumps(tariff, sort_keys=True) + return await self._apply_command("tariff", signature, lambda: self._command("time_of_use_settings", {"tou_settings": {"tariff_content_v2": tariff}}), force=force) @staticmethod def _find_tariff_code(node): @@ -441,6 +494,11 @@ async def reconcile_on_start(self): returned a usable code, whether or not a restore was needed), so run() can latch reconcile_done and stop retrying. Returns False when the tariff read itself failed to return anything usable, so run() retries on a later cycle instead of latching a no-op. + + The restore writes below pass force=True to `set_tariff`/`set_export_rule` so this recovery + path can NEVER be silently skipped by the write-on-change dedupe cache - at boot the cache is + empty so this is naturally a no-op today, but forcing makes that explicit and future-proof + against a cache-preseed feature that might otherwise make recovery look like a no-op. """ tariff_code = await self.get_current_tariff_code() if tariff_code is None: @@ -452,9 +510,9 @@ async def reconcile_on_start(self): self.log("Warn: Teslemetry device tariff was left as {} - recovery needed but skipped (read-only mode)".format(tariff_code)) return True self.log("Warn: Teslemetry device tariff was left as {} - restoring normal tariff and disabling export".format(tariff_code)) - if await self.set_tariff("normal"): + if await self.set_tariff("normal", force=True): self.publish_control(self.entity("tariff_mode", domain="select"), "normal") - if await self.set_export_rule("never"): + if await self.set_export_rule("never", force=True): self.publish_control(self.entity("allow_export", domain="select"), "never") return True diff --git a/apps/predbat/tests/test_teslemetry.py b/apps/predbat/tests/test_teslemetry.py index 1ad74080f..6a39b78f3 100644 --- a/apps/predbat/tests/test_teslemetry.py +++ b/apps/predbat/tests/test_teslemetry.py @@ -30,6 +30,7 @@ def __init__(self): self.entity_states = {} self.mock_responses = {} self.requests_made = [] + self._last_sent = {} def log(self, msg): """Capture log messages.""" @@ -688,6 +689,124 @@ def test_teslemetry_number_event_guards_null_value(): assert any("invalid" in msg.lower() for msg in api.log_messages) +def test_teslemetry_dedupe_operation_mode_skips_repeat_post(): + """Sending the same operation_mode value twice via select_event posts only once (write-on-change).""" + api = MockTeslemetryAPI() + api.mock_responses["/api/1/energy_sites/123456/operation"] = {"response": {"code": 201}} + run_async(api.select_event("select.predbat_teslemetry_operation_mode", "backup")) + run_async(api.select_event("select.predbat_teslemetry_operation_mode", "backup")) + posts = [req for req in api.requests_made if req[0] == "POST"] + assert len(posts) == 1 + assert api.entity_states["select.predbat_teslemetry_operation_mode"] == "backup" + + +def test_teslemetry_dedupe_operation_mode_resends_on_change(): + """A changed operation_mode value must still POST (dedupe only skips an unchanged value).""" + api = MockTeslemetryAPI() + api.mock_responses["/api/1/energy_sites/123456/operation"] = {"response": {"code": 201}} + run_async(api.select_event("select.predbat_teslemetry_operation_mode", "backup")) + run_async(api.select_event("select.predbat_teslemetry_operation_mode", "autonomous")) + posts = [req for req in api.requests_made if req[0] == "POST"] + assert len(posts) == 2 + assert api.entity_states["select.predbat_teslemetry_operation_mode"] == "autonomous" + + +def test_teslemetry_dedupe_failed_post_not_cached_so_retries(): + """A failed POST must NOT update the dedupe cache, so the identical value is re-sent next cycle + (failure-retry survives the write-on-change dedupe).""" + api = MockTeslemetryAPI() + # No mock response registered for /operation -> _request returns None -> command fails. + run_async(api.select_event("select.predbat_teslemetry_operation_mode", "backup")) + assert api.requests_made == [("POST", "/api/1/energy_sites/123456/operation", {"default_real_mode": "backup"})] + assert "select.predbat_teslemetry_operation_mode" not in api.entity_states + assert "operation_mode" not in api._last_sent + # API recovers; the SAME value must still be sent - the cache was never populated on failure. + api.mock_responses["/api/1/energy_sites/123456/operation"] = {"response": {"code": 201}} + run_async(api.select_event("select.predbat_teslemetry_operation_mode", "backup")) + posts = [req for req in api.requests_made if req[0] == "POST"] + assert len(posts) == 2 + assert api.entity_states["select.predbat_teslemetry_operation_mode"] == "backup" + + +def test_teslemetry_dedupe_tariff_identical_body_skips_repeat_post(): + """Two normal-tariff builds with an identical body POST only once.""" + api = MockTeslemetryAPI() + api.mock_responses["/api/1/energy_sites/123456/time_of_use_settings"] = {"response": {"code": 201}} + run_async(api.set_tariff("normal")) + run_async(api.set_tariff("normal")) + posts = [req for req in api.requests_made if req[0] == "POST"] + assert len(posts) == 1 + + +def test_teslemetry_dedupe_tariff_resends_when_rates_change(): + """A tariff rebuild whose sell/buy price actually changed must re-POST rather than be deduped.""" + from types import SimpleNamespace + + api = MockTeslemetryAPI() + api.mock_responses["/api/1/energy_sites/123456/time_of_use_settings"] = {"response": {"code": 201}} + api.base = SimpleNamespace(minutes_now=600, rate_import={600: 30.0}, rate_export={600: 15.0}, now_utc=None, local_tz=None) + run_async(api.set_tariff("normal")) + api.base.rate_export = {600: 60.0} # Export rate jumps, so the sell-side signature changes. + run_async(api.set_tariff("normal")) + posts = [req for req in api.requests_made if req[0] == "POST"] + assert len(posts) == 2 + + +def test_teslemetry_drift_correction_refreshes_cache_and_reasserts(): + """A site_info poll reporting a device mode that differs from what Predbat last sent must refresh + the dedupe cache to the ACTUAL device value, so the next assertion of Predbat's desired mode is not + skipped - this is how externally-changed (Tesla app) drift gets corrected rather than silently kept.""" + api = MockTeslemetryAPI() + api.mock_responses["/api/1/energy_sites/123456/operation"] = {"response": {"code": 201}} + run_async(api.select_event("select.predbat_teslemetry_operation_mode", "backup")) + assert api._last_sent["operation_mode"] == "backup" + # SITE_INFO reports default_real_mode = self_consumption, i.e. the user changed it externally. + api.mock_responses["/api/1/energy_sites/123456/site_info"] = SITE_INFO + run_async(api.fetch_site_info()) + assert api._last_sent["operation_mode"] == "self_consumption" + # Predbat's desired mode is still "backup" - the drifted cache means this must re-POST. + run_async(api.select_event("select.predbat_teslemetry_operation_mode", "backup")) + posts = [req for req in api.requests_made if req[0] == "POST" and req[1].endswith("/operation")] + assert len(posts) == 2 + + +def test_teslemetry_drift_correction_no_spurious_resend_when_matching(): + """When site_info reports the SAME mode Predbat last sent, the dedupe cache is unchanged and the + next assertion of that mode is still skipped - no spurious re-send from the drift-refresh itself.""" + api = MockTeslemetryAPI() + api.mock_responses["/api/1/energy_sites/123456/operation"] = {"response": {"code": 201}} + run_async(api.select_event("select.predbat_teslemetry_operation_mode", "self_consumption")) + assert api._last_sent["operation_mode"] == "self_consumption" + # SITE_INFO's default_real_mode is also self_consumption - no drift. + api.mock_responses["/api/1/energy_sites/123456/site_info"] = SITE_INFO + run_async(api.fetch_site_info()) + assert api._last_sent["operation_mode"] == "self_consumption" + run_async(api.select_event("select.predbat_teslemetry_operation_mode", "self_consumption")) + posts = [req for req in api.requests_made if req[0] == "POST" and req[1].endswith("/operation")] + assert len(posts) == 1 + + +def test_teslemetry_reconcile_forces_write_even_if_cache_preseeded(): + """reconcile_on_start's corrective writes must not be silently skipped by the dedupe cache even if + it already holds a value matching what recovery is about to send (e.g. a future cache-preseed + feature) - the recovery path always forces the POST regardless of cache state.""" + import json as _json + + api = MockTeslemetryAPI() + api.mock_responses["/api/1/energy_sites/123456/tariff_rate"] = TARIFF_RATE_EXPORT_NOW + api.mock_responses["/api/1/energy_sites/123456/time_of_use_settings"] = {"response": {"code": 201}} + api.mock_responses["/api/1/energy_sites/123456/grid_import_export"] = {"response": {"code": 201}} + # Pre-seed the dedupe cache as if "normal"/"never" had already been confirmed-sent - a naive + # dedupe would treat the recovery writes below as no-ops and skip them entirely. + api._last_sent["tariff"] = _json.dumps(api.build_tariff("normal"), sort_keys=True) + api._last_sent["export_rule"] = "never" + run_async(api.reconcile_on_start()) + assert any(req[1].endswith("/time_of_use_settings") for req in api.requests_made if req[0] == "POST") + assert any(req[1].endswith("/grid_import_export") for req in api.requests_made if req[0] == "POST") + assert api.entity_states["select.predbat_teslemetry_tariff_mode"] == "normal" + assert api.entity_states["select.predbat_teslemetry_allow_export"] == "never" + + def test_teslemetry(my_predbat=None): """Run all Teslemetry component tests (registry entry point). @@ -736,5 +855,13 @@ def test_teslemetry(my_predbat=None): test_teslemetry_energy_today_requests_kind_and_period() test_teslemetry_select_event_preserves_control_attributes() test_teslemetry_number_event_guards_null_value() + test_teslemetry_dedupe_operation_mode_skips_repeat_post() + test_teslemetry_dedupe_operation_mode_resends_on_change() + test_teslemetry_dedupe_failed_post_not_cached_so_retries() + test_teslemetry_dedupe_tariff_identical_body_skips_repeat_post() + test_teslemetry_dedupe_tariff_resends_when_rates_change() + test_teslemetry_drift_correction_refreshes_cache_and_reasserts() + test_teslemetry_drift_correction_no_spurious_resend_when_matching() + test_teslemetry_reconcile_forces_write_even_if_cache_preseeded() print("**** Teslemetry tests passed ****") return 0 From 9accebd8e18fcb382a458360d237379b17b1a557 Mon Sep 17 00:00:00 2001 From: Mark Gascoyne Date: Tue, 7 Jul 2026 00:12:43 +0100 Subject: [PATCH 13/14] test(teslemetry): cover tariff window-advance + backup_reserve drift; document transition-based drift limitation --- apps/predbat/teslemetry.py | 20 ++++++ apps/predbat/tests/test_teslemetry.py | 90 +++++++++++++++++++++++++++ 2 files changed, 110 insertions(+) diff --git a/apps/predbat/teslemetry.py b/apps/predbat/teslemetry.py index adf3ae272..81d29923e 100644 --- a/apps/predbat/teslemetry.py +++ b/apps/predbat/teslemetry.py @@ -70,6 +70,7 @@ def initialize(self, key="", site_id="", base_url=TESLEMETRY_DEFAULT_URL, **kwar self.reconcile_done = False self._last_sent = {} self.log("Info: TeslemetryAPI initialising site_id={}".format(self.site_id)) + self.log("Info: Teslemetry control drift-correction is transition-based (self-heals when Predbat's own desired value changes, plus a one-off refresh at boot) - full periodic device-state reconciliation is a pilot follow-up") self.register_control_entities() def entity(self, suffix, domain="sensor"): @@ -144,6 +145,17 @@ async def fetch_site_info(self): soc_max_published = True # Seed the control entity STATES (display only, no commands) from the device so they reflect # reality at boot instead of the hardcoded defaults set by register_control_entities(). + # + # Known limitation (accepted, not an oversight): fetch_site_info is only ever driven to + # completion ONCE per process - run()'s site_info_done latch stops calling it again once it + # returns True - so the operation_mode/backup_reserve drift-refresh below corrects drift at + # BOOT only. If the device drifts again mid-run (e.g. the customer changes mode/reserve via + # the Tesla app after startup), that drift is not detected again until the process restarts. + # In the meantime it self-heals only on a Predbat "transition" - i.e. whenever Predbat's OWN + # desired value actually changes, since a changed target never matches the stale cache + # regardless of the device's true state. A repeated assertion of the SAME target while the + # device has silently drifted away stays stuck until the next boot. This is a ship-now, + # close-in-pilot decision; continuous periodic device-state reconciliation is a follow-up. default_mode = response.get("default_real_mode") if default_mode in OPERATION_MODES: self.publish_control(self.entity("operation_mode", domain="select"), default_mode) @@ -292,6 +304,14 @@ async def _apply_command(self, key, signature, sender, force=False): success. This is used by `reconcile_on_start` so its corrective writes can never be silently skipped by a (present or future) pre-seeded cache entry that happens to already match what the recovery is about to send. + + Drift-refresh limitation (accepted, not an oversight): unlike "operation_mode" and + "backup_reserve" (refreshed once per process from the device's actual state - see the NOTE in + `fetch_site_info`), "grid_charging" and "export_rule" are NEVER device-drift-refreshed at all, + because the site_info response this component reads does not expose either field's actual + device state. If the device drifts on these two externally, the cache is never proactively + corrected; they self-heal only when Predbat's own desired value changes (a transition), since a + changed target never matches the stale cache irrespective of the device's true state. """ if not force and self._last_sent.get(key) == signature: return True diff --git a/apps/predbat/tests/test_teslemetry.py b/apps/predbat/tests/test_teslemetry.py index 6a39b78f3..067d3440c 100644 --- a/apps/predbat/tests/test_teslemetry.py +++ b/apps/predbat/tests/test_teslemetry.py @@ -786,6 +786,93 @@ def test_teslemetry_drift_correction_no_spurious_resend_when_matching(): assert len(posts) == 1 +def test_teslemetry_dedupe_tariff_resends_on_window_advance(): + """A tariff rebuild whose live rate is UNCHANGED but whose export_now ON_PEAK window has + advanced (a later base local time crosses into the next 30-minute-aligned window) must still + re-POST - the dedupe signature is the full built tariff body (which embeds the window), not + just the mode string, so a window-only move is not silently skipped. + + Would be RED under a hypothetical mode-string-keyed signature: if `set_tariff` deduped on + `_apply_command("tariff", mode, ...)` instead of the built-tariff JSON, both calls below pass + the identical mode "export_now", so `self._last_sent["tariff"] == "export_now"` would already + match on the second call and the POST would be skipped (posts == 1) even though the device's + ON_PEAK window moved from 14:30-15:30 to 15:00-16:00 - the customer would be left on a STALE, + now-expired export window. GREEN as-built: the signature is + `json.dumps(build_tariff(mode), sort_keys=True)`, which differs between the two builds purely + because fromHour/fromMinute/toHour/toMinute moved, so both calls POST. + """ + from datetime import datetime, timezone + from types import SimpleNamespace + + api = MockTeslemetryAPI() + api.mock_responses["/api/1/energy_sites/123456/time_of_use_settings"] = {"response": {"code": 201}} + # Live import/export rates held FIXED across both calls (same minutes_now, same rate tables) so + # only the base local time - and therefore the window - differs between the two builds. + api.base = SimpleNamespace( + now_utc=datetime(2026, 7, 2, 14, 40, tzinfo=timezone.utc), + local_tz=timezone.utc, + minutes_now=600, + rate_import={600: 28.0}, + rate_export={600: 15.0}, + ) + run_async(api.set_tariff("export_now")) + + # Advance base local time into the NEXT 30-minute window; rates untouched. + api.base.now_utc = datetime(2026, 7, 2, 15, 10, tzinfo=timezone.utc) + run_async(api.set_tariff("export_now")) + + posts = [req for req in api.requests_made if req[0] == "POST"] + assert len(posts) == 2 + first_tariff = posts[0][2]["tou_settings"]["tariff_content_v2"] + second_tariff = posts[1][2]["tou_settings"]["tariff_content_v2"] + first_window = first_tariff["seasons"]["AllYear"]["tou_periods"]["ON_PEAK"]["periods"][0] + second_window = second_tariff["seasons"]["AllYear"]["tou_periods"]["ON_PEAK"]["periods"][0] + assert (first_window["fromHour"], first_window["fromMinute"]) == (14, 30) + assert (second_window["fromHour"], second_window["fromMinute"]) == (15, 0) + # Confirm the live rate really was constant - only the window differs between the two builds. + first_on_peak_sell = first_tariff["sell_tariff"]["energy_charges"]["AllYear"]["rates"]["ON_PEAK"] + second_on_peak_sell = second_tariff["sell_tariff"]["energy_charges"]["AllYear"]["rates"]["ON_PEAK"] + assert first_on_peak_sell == second_on_peak_sell + + +def test_teslemetry_backup_reserve_drift_correction_refreshes_cache_and_reasserts(): + """A site_info poll reporting a device backup_reserve_percent that differs from what Predbat + last sent must refresh the dedupe cache to the ACTUAL device value, so the next assertion of + Predbat's desired reserve is not skipped - mirrors + test_teslemetry_drift_correction_refreshes_cache_and_reasserts (operation_mode) but exercises + the backup_reserve refresh path in fetch_site_info, which previously had no direct test.""" + api = MockTeslemetryAPI() + api.mock_responses["/api/1/energy_sites/123456/backup"] = {"response": {"code": 201}} + run_async(api.number_event("number.predbat_teslemetry_backup_reserve", 20)) + assert api._last_sent["backup_reserve"] == 20 + # Device now reports backup_reserve_percent = 50, i.e. the user changed it externally. + drifted_site_info = {"response": {"nameplate_energy": 13500, "default_real_mode": "self_consumption", "backup_reserve_percent": 50}} + api.mock_responses["/api/1/energy_sites/123456/site_info"] = drifted_site_info + run_async(api.fetch_site_info()) + assert api._last_sent["backup_reserve"] == 50 + # Predbat's desired reserve is still 20 - the drifted cache means this must re-POST. + run_async(api.number_event("number.predbat_teslemetry_backup_reserve", 20)) + posts = [req for req in api.requests_made if req[0] == "POST" and req[1].endswith("/backup")] + assert len(posts) == 2 + + +def test_teslemetry_backup_reserve_drift_correction_no_spurious_resend_when_matching(): + """When site_info reports the SAME backup_reserve Predbat last sent, the dedupe cache is + unchanged and the next assertion of that value is still skipped - no spurious re-send from the + drift-refresh itself.""" + api = MockTeslemetryAPI() + api.mock_responses["/api/1/energy_sites/123456/backup"] = {"response": {"code": 201}} + run_async(api.number_event("number.predbat_teslemetry_backup_reserve", 20)) + assert api._last_sent["backup_reserve"] == 20 + # SITE_INFO's backup_reserve_percent is also 20 - no drift. + api.mock_responses["/api/1/energy_sites/123456/site_info"] = SITE_INFO + run_async(api.fetch_site_info()) + assert api._last_sent["backup_reserve"] == 20 + run_async(api.number_event("number.predbat_teslemetry_backup_reserve", 20)) + posts = [req for req in api.requests_made if req[0] == "POST" and req[1].endswith("/backup")] + assert len(posts) == 1 + + def test_teslemetry_reconcile_forces_write_even_if_cache_preseeded(): """reconcile_on_start's corrective writes must not be silently skipped by the dedupe cache even if it already holds a value matching what recovery is about to send (e.g. a future cache-preseed @@ -862,6 +949,9 @@ def test_teslemetry(my_predbat=None): test_teslemetry_dedupe_tariff_resends_when_rates_change() test_teslemetry_drift_correction_refreshes_cache_and_reasserts() test_teslemetry_drift_correction_no_spurious_resend_when_matching() + test_teslemetry_dedupe_tariff_resends_on_window_advance() + test_teslemetry_backup_reserve_drift_correction_refreshes_cache_and_reasserts() + test_teslemetry_backup_reserve_drift_correction_no_spurious_resend_when_matching() test_teslemetry_reconcile_forces_write_even_if_cache_preseeded() print("**** Teslemetry tests passed ****") return 0 From 8e5c8d214c53fdc6d01f232c63fa1e6ea7490a45 Mon Sep 17 00:00:00 2001 From: "pre-commit-ci-lite[bot]" <117423508+pre-commit-ci-lite[bot]@users.noreply.github.com> Date: Mon, 6 Jul 2026 23:15:50 +0000 Subject: [PATCH 14/14] [pre-commit.ci lite] apply automatic fixes --- .cspell/custom-dictionary-workspace.txt | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.cspell/custom-dictionary-workspace.txt b/.cspell/custom-dictionary-workspace.txt index d1f605224..ad3a2510c 100644 --- a/.cspell/custom-dictionary-workspace.txt +++ b/.cspell/custom-dictionary-workspace.txt @@ -398,8 +398,8 @@ TCWORLD tdata tdiff temperation -Teslemetry teslemetry +Teslemetry testname thirdparty timea