From 9a4dc6257842b8cf33e2036931d8e960a2af9ba9 Mon Sep 17 00:00:00 2001 From: Trefor Southwell Date: Wed, 19 Aug 2026 20:38:04 +0100 Subject: [PATCH] fix(teslemetry): assert optimization_strategy=economics on every tariff push Powerwall's Time-Based Control silently stays in Balanced mode - which only offsets house load and never exports stored energy for price - because set_tariff() never sent the Fleet API's optimization_strategy dial. Every device command succeeded and Predbat reported "Exporting", but no energy ever left the battery. "economics" is the strategy that actually exports for price; it's now included in the dedupe signature so the standalone CLI harness gained OAuth-mode support (auth_method/token_expires_at/token_hash/ user_id, mirroring fox.py) plus an --apps flag to load config straight from a Predbat apps.yaml, which is what surfaced and let us confirm the fix live. Fixes #4600 Co-Authored-By: Claude Sonnet 5 --- apps/predbat/teslemetry.py | 119 ++++++++++++++++- apps/predbat/tests/test_teslemetry.py | 185 ++++++++++++++++++++++++++ 2 files changed, 297 insertions(+), 7 deletions(-) diff --git a/apps/predbat/teslemetry.py b/apps/predbat/teslemetry.py index 5cfea585e..d569a7a0c 100644 --- a/apps/predbat/teslemetry.py +++ b/apps/predbat/teslemetry.py @@ -27,6 +27,7 @@ import asyncio import copy import json +import os import sys from datetime import datetime, timezone @@ -56,6 +57,12 @@ OPERATION_MODES = ["self_consumption", "autonomous", "backup"] EXPORT_RULES = ["never", "pv_only", "battery_ok"] +# tou_settings.optimization_strategy: the OTHER dial the Fleet API exposes alongside the tariff itself, +# deciding whether Time-Based Control actually acts on price. "balanced" (the device default, and what +# is left in place if this is never sent) only discharges to offset house load and never exports stored +# energy for price, no matter how attractive the pushed sell rate is; "economics" is the strategy that +# exports for price. See GH#4600. +TARIFF_OPTIMIZATION_STRATEGY = "economics" REAL_TIERS = ["SUPER_OFF_PEAK", "OFF_PEAK", "PARTIAL_PEAK"] BOOST_TIER = "ON_PEAK" @@ -1146,13 +1153,20 @@ def _assemble_tariff(self, code, buy_charges, buy_periods, sell_charges, sell_pe } async def set_tariff(self, tariff, force=False): - """Push a prebuilt tariff via time_of_use_settings, deduped on the serialised tariff body. + """Push a prebuilt tariff via time_of_use_settings, deduped on the serialised tou_settings body. + + Also asserts optimization_strategy=TARIFF_OPTIMIZATION_STRATEGY on every push (GH#4600): + without it, Time-Based Control can silently stay in Balanced mode and never export, and there + is no read endpoint to detect that drift, so reasserting it alongside the tariff is the only + available mitigation. It is included in the dedupe signature so any future change to the + strategy forces a re-push rather than being masked by an unrelated-looking tariff match. Prices are rounded to whole pence upstream so per-cycle rate nudges do not change the JSON; a re-push therefore fires only on a genuine band, boost-window or day-of-week change. """ - 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) + body = {"optimization_strategy": TARIFF_OPTIMIZATION_STRATEGY, "tariff_content_v2": tariff} + signature = json.dumps(body, sort_keys=True) + return await self._apply_command("tariff", signature, lambda: self._command("time_of_use_settings", {"tou_settings": body}), force=force) async def sync_tariff(self): """Build the tariff from current rates + committed discharge window and push it (deduped). @@ -1237,7 +1251,7 @@ async def final(self): self.log("Info: TeslemetryAPI shutdown") -async def test_teslemetry_api(key, site_id=None, base_url=None, control=False): +async def test_teslemetry_api(key, site_id=None, base_url=None, control=False, auth_method=None, token_expires_at=None, token_hash=None, user_id=None, supabase_url=None, supabase_key=None): """Run a standalone test of the Teslemetry component against the live API. site_id is optional and acts as a filter over the sites discovered from /api/1/products; when @@ -1246,20 +1260,42 @@ async def test_teslemetry_api(key, site_id=None, base_url=None, control=False): emulator and any crash-recovery writes are suppressed via set_read_only. Pass control=True to let the component send commands. + auth_method="oauth" exercises the direct-Fleet-API OAuth path (mirrors Fox/Kraken/Solis) instead + of the default static api_key mode: key is then the OAuth access token, token_expires_at/token_hash + seed refresh state, and user_id is written into MockBase.args (as OAuthMixin's _do_refresh reads + instance_id from self.base.args). A live refresh additionally needs a still-valid refresh token + held server-side, and SUPABASE_URL/SUPABASE_KEY: supabase_url/supabase_key are exported into + os.environ here (mirrors fox.py's test_fox_api) because OAuthMixin reads them from the environment + rather than accepting them as component arguments - passing them straight to TeslemetryAPI would + silently do nothing. + Returns True only if the run connected and completed successfully. A failed connection - an auth failure (401/403) or an unsuccessful run() - returns False so main() can exit non-zero instead of a broken connection looking like a pass. """ + if supabase_url: + os.environ["SUPABASE_URL"] = supabase_url + if supabase_key: + os.environ["SUPABASE_KEY"] = supabase_key + mode = "READ-WRITE (controls may change)" if control else "READ-ONLY (status only, no controls changed)" - print("Testing Teslemetry API for site {} - {}".format(site_id or "auto-discover", mode)) + print("Testing Teslemetry API for site {} - {} - auth={}".format(site_id or "auto-discover", mode, auth_method or "api_key")) mock_base = MockBase() # Read-only by default so a bare test run only reports status and never changes the Powerwall. mock_base.args["set_read_only"] = not control + if user_id: + mock_base.args["user_id"] = user_id arg_dict = {"key": key, "site_id": site_id or "", "automatic": True} if base_url: arg_dict["base_url"] = base_url + if auth_method: + arg_dict["auth_method"] = auth_method + if token_expires_at: + arg_dict["token_expires_at"] = token_expires_at + if token_hash: + arg_dict["token_hash"] = token_hash api = TeslemetryAPI(mock_base, **arg_dict) print("Calling run() once...") @@ -1276,16 +1312,85 @@ async def test_teslemetry_api(key, site_id=None, base_url=None, control=False): return True +APPS_YAML_ROOT_KEY = "pred_bat" +# Maps a Predbat apps.yaml config item name to the test_teslemetry_api()/main() argument it feeds. +# user_id is deliberately not teslemetry-prefixed: it is the shared predbat.com instance id consulted +# by OAuthMixin's refresh for every OAuth-based component, not a Teslemetry-specific setting. +APPS_YAML_ARG_KEYS = { + "teslemetry_key": "key", + "teslemetry_site_id": "site_id", + "teslemetry_base_url": "base_url", + "teslemetry_auth_method": "auth_method", + "teslemetry_token_expires_at": "token_expires_at", + "teslemetry_token_hash": "token_hash", + "user_id": "user_id", + "supabase_url": "supabase_url", + "supabase_key": "supabase_key", +} + + +def load_teslemetry_args_from_apps_yaml(path): + """Load Teslemetry CLI arguments from a Predbat apps.yaml's pred_bat: section. + + Lets the standalone CLI harness be pointed at a real config file (--apps path) instead of every + value - including a long OAuth access token - being pasted onto the command line and left sitting + in shell history. Only keys actually present (and non-empty) in the file are returned, so an + absent item is omitted rather than defaulted to None/empty and overriding a CLI flag or argparse + default with a null. + """ + from ruamel.yaml import YAML + + yaml = YAML(typ="safe") + with open(path, "r") as f: + data = yaml.load(f) or {} + section = data.get(APPS_YAML_ROOT_KEY, data) or {} + return {arg_name: section[config_name] for config_name, arg_name in APPS_YAML_ARG_KEYS.items() if section.get(config_name) not in (None, "")} + + def main(): # pragma: no cover """Command line entry point to test the Teslemetry component against the live API.""" parser = argparse.ArgumentParser(description="Test Teslemetry Tesla Powerwall API") - parser.add_argument("--key", required=True, help="Teslemetry (or Fleet API) bearer token") + parser.add_argument("--apps", default=None, help="Path to a Predbat apps.yaml (or an exported copy) to load teslemetry_key/site_id/base_url/auth_method/token_expires_at/token_hash/user_id from; any of --key etc below still override the file's values") + parser.add_argument("--key", default=None, help="Teslemetry (or Fleet API) bearer token") parser.add_argument("--site-id", default=None, help="Optional Tesla energy site id to filter the sites discovered from /api/1/products (default: use the first site on the account)") parser.add_argument("--base-url", default=None, help="REST API base URL (default {})".format(TESLEMETRY_DEFAULT_URL)) parser.add_argument("--control", action="store_true", help="Allow control commands to be sent (default is read-only: report status only, change nothing)") + parser.add_argument("--auth-method", default=None, choices=["api_key", "oauth"], help="'api_key' (default) for a static Teslemetry token, or 'oauth' for a direct Fleet API OAuth access token") + parser.add_argument("--token-expires-at", default=None, help="OAuth access token expiry (ISO timestamp) - oauth mode only") + parser.add_argument("--token-hash", default=None, help="Server-computed OAuth token hash for refresh dedup - oauth mode only") + parser.add_argument("--user-id", default=None, help="predbat.com instance/user id, required by OAuthMixin to refresh - oauth mode only") + parser.add_argument("--supabase-url", default=None, help="Supabase URL for OAuth token refresh - oauth mode only") + parser.add_argument("--supabase-key", default=None, help="Supabase service key for OAuth token refresh - oauth mode only") args = parser.parse_args() - ok = asyncio.run(test_teslemetry_api(args.key, args.site_id, base_url=args.base_url, control=args.control)) + file_args = load_teslemetry_args_from_apps_yaml(args.apps) if args.apps else {} + + key = args.key or file_args.get("key") + if not key: + parser.error("--key is required (directly, or via --apps pointing at a file with teslemetry_key set)") + site_id = args.site_id or file_args.get("site_id") + base_url = args.base_url or file_args.get("base_url") + auth_method = args.auth_method or file_args.get("auth_method") + token_expires_at = args.token_expires_at or file_args.get("token_expires_at") + token_hash = args.token_hash or file_args.get("token_hash") + user_id = args.user_id or file_args.get("user_id") + supabase_url = args.supabase_url or file_args.get("supabase_url") + supabase_key = args.supabase_key or file_args.get("supabase_key") + + ok = asyncio.run( + test_teslemetry_api( + key, + site_id, + base_url=base_url, + control=args.control, + auth_method=auth_method, + token_expires_at=token_expires_at, + token_hash=token_hash, + user_id=user_id, + supabase_url=supabase_url, + supabase_key=supabase_key, + ) + ) sys.exit(0 if ok else 1) diff --git a/apps/predbat/tests/test_teslemetry.py b/apps/predbat/tests/test_teslemetry.py index 9316412d4..55a8dd0cd 100644 --- a/apps/predbat/tests/test_teslemetry.py +++ b/apps/predbat/tests/test_teslemetry.py @@ -755,6 +755,24 @@ def test_teslemetry_set_tariff_posts_tou_settings(): assert "tariff_content_v2" in body["tou_settings"] +def test_teslemetry_set_tariff_asserts_optimization_strategy_economics(): + """set_tariff must assert optimization_strategy=economics on every push (GH#4600). + + Without this, the Fleet API tou_settings.optimization_strategy dial is left untouched (whatever + the customer's Tesla app happens to hold), so Time-Based Control can silently stay in Balanced + mode - which only offsets house load and never exports stored energy, no matter how high the + pushed sell price is. "economics" is the strategy that actually exports for price. + """ + api = MockTeslemetryAPI() + api.base = _rate_base(import_p=28.0, export_p=15.0) + api.mock_responses["/api/1/energy_sites/123456/time_of_use_settings"] = {"response": {"code": 201}} + t = api.build_tariff(None) + result = run_async(api.set_tariff(t)) + assert result is True + method, path, body = api.requests_made[-1] + assert body["tou_settings"]["optimization_strategy"] == "economics" + + def test_teslemetry_sync_tariff_dedupes_unchanged(): """Two syncs with identical inputs push the tariff exactly once (monthly API-call budget).""" api = MockTeslemetryAPI() @@ -1657,6 +1675,168 @@ async def healthy_request(self, method, path, json_body=None): teslemetry.TeslemetryAPI._request = original +def test_teslemetry_cli_harness_wires_oauth_args(): + """test_teslemetry_api must forward auth_method/token_expires_at/token_hash/user_id through to + the component (and user_id into MockBase.args, where OAuthMixin's refresh reads instance_id from) + so the standalone CLI harness can exercise OAuth mode, not just the default static api_key mode. + A live end-to-end refresh still needs SUPABASE_URL/SUPABASE_KEY and a real refresh token server-side + (GH#4600 follow-up) - this only confirms the harness plumbs the arguments through correctly.""" + import io + import contextlib + from datetime import datetime + import teslemetry + + captured = {} + original_run = teslemetry.TeslemetryAPI.run + + async def capturing_run(self, seconds=0, first=False): + """Capture OAuth wiring instead of performing a real run.""" + captured["auth_method"] = self.auth_method + captured["token_expires_at"] = self.token_expires_at + captured["token_hash"] = self.token_hash + captured["user_id"] = self.base.args.get("user_id") + self.api_auth_failed = True + return False + + teslemetry.TeslemetryAPI.run = capturing_run + try: + with contextlib.redirect_stdout(io.StringIO()): + run_async( + teslemetry.test_teslemetry_api( + "oauth-access-token", + "site123", + auth_method="oauth", + token_expires_at="2026-08-17T03:02:31+00:00", + token_hash="abc123", + user_id="user-456", + ) + ) + finally: + teslemetry.TeslemetryAPI.run = original_run + + assert captured["auth_method"] == "oauth" + assert captured["token_expires_at"] == datetime.fromisoformat("2026-08-17T03:02:31+00:00").timestamp() + assert captured["token_hash"] == "abc123" + assert captured["user_id"] == "user-456" + + +def test_teslemetry_api_sets_supabase_env_vars_for_oauth_refresh(): + """test_teslemetry_api must export SUPABASE_URL/SUPABASE_KEY into the environment when given + (mirrors fox.py's test_fox_api), because OAuthMixin._do_refresh reads them via os.environ rather + than through any argument the component itself accepts. Without this, supabase_url/supabase_key + loaded from --apps (or passed directly) had no way to actually reach the refresh call - the exact + gap that surfaced testing the CLI harness live against a real oauth apps.yaml (GH#4600 follow-up).""" + import io + import contextlib + import os + import teslemetry + + original_run = teslemetry.TeslemetryAPI.run + captured = {} + + async def capturing_run(self, seconds=0, first=False): + """Capture the environment instead of performing a real run.""" + captured["SUPABASE_URL"] = os.environ.get("SUPABASE_URL") + captured["SUPABASE_KEY"] = os.environ.get("SUPABASE_KEY") + self.api_auth_failed = True + return False + + saved_url = os.environ.pop("SUPABASE_URL", None) + saved_key = os.environ.pop("SUPABASE_KEY", None) + teslemetry.TeslemetryAPI.run = capturing_run + try: + with contextlib.redirect_stdout(io.StringIO()): + run_async( + teslemetry.test_teslemetry_api( + "oauth-access-token", + "site123", + auth_method="oauth", + supabase_url="https://example.supabase.co", + supabase_key="service-key", + ) + ) + finally: + teslemetry.TeslemetryAPI.run = original_run + if saved_url is None: + os.environ.pop("SUPABASE_URL", None) + else: + os.environ["SUPABASE_URL"] = saved_url + if saved_key is None: + os.environ.pop("SUPABASE_KEY", None) + else: + os.environ["SUPABASE_KEY"] = saved_key + + assert captured["SUPABASE_URL"] == "https://example.supabase.co" + assert captured["SUPABASE_KEY"] == "service-key" + + +def test_teslemetry_load_args_from_apps_yaml_extracts_teslemetry_section(): + """load_teslemetry_args_from_apps_yaml pulls the teslemetry_* keys (plus the top-level user_id) + out of a real Predbat apps.yaml's pred_bat: section, so the CLI harness can be pointed at a config + file (--apps path) instead of pasting a long OAuth token on the command line (GH#4600 follow-up).""" + import tempfile + import os + import teslemetry + + content = """ +pred_bat: + supabase_url: https://example.supabase.co + supabase_key: the-supabase-service-key + teslemetry_auth_method: oauth + teslemetry_automatic: true + teslemetry_base_url: https://fleet-api.prd.eu.vn.cloud.tesla.com + teslemetry_key: the-access-token + teslemetry_site_id: '1689257309996718' + teslemetry_token_expires_at: '2026-08-17T03:02:31.016+00:00' + teslemetry_token_hash: the-token-hash + user_id: 80e510e4-8f58-4b66-b6ca-10f08ba16682 + some_other_unrelated_key: ignored +""" + fd, path = tempfile.mkstemp(suffix=".yaml") + try: + with os.fdopen(fd, "w") as f: + f.write(content) + result = teslemetry.load_teslemetry_args_from_apps_yaml(path) + finally: + os.remove(path) + + assert result == { + "key": "the-access-token", + "site_id": "1689257309996718", + "base_url": "https://fleet-api.prd.eu.vn.cloud.tesla.com", + "auth_method": "oauth", + "token_expires_at": "2026-08-17T03:02:31.016+00:00", + "token_hash": "the-token-hash", + "user_id": "80e510e4-8f58-4b66-b6ca-10f08ba16682", + "supabase_url": "https://example.supabase.co", + "supabase_key": "the-supabase-service-key", + } + + +def test_teslemetry_load_args_from_apps_yaml_omits_missing_keys(): + """Absent teslemetry_* items (e.g. a static api_key setup with no oauth fields) must be omitted + from the result entirely, not defaulted to None/empty - so main() falls through to its own + argparse defaults / --key requirement rather than being overridden with a null.""" + import tempfile + import os + import teslemetry + + content = """ +pred_bat: + teslemetry_key: the-access-token + teslemetry_site_id: '12345' +""" + fd, path = tempfile.mkstemp(suffix=".yaml") + try: + with os.fdopen(fd, "w") as f: + f.write(content) + result = teslemetry.load_teslemetry_args_from_apps_yaml(path) + finally: + os.remove(path) + + assert result == {"key": "the-access-token", "site_id": "12345"} + + def test_teslemetry_quantise_flat_single_tier(): """A flat rate collapses to one tier priced in GBP whole pence, all 48 slots the same.""" rates = {m: 28.0 for m in range(0, 2880)} # 28p flat @@ -1927,6 +2107,7 @@ def test_teslemetry(my_predbat=None): test_teslemetry_day_runs_groups_replicated_days() test_teslemetry_day_runs_all_identical_single_run() test_teslemetry_set_tariff_posts_tou_settings() + test_teslemetry_set_tariff_asserts_optimization_strategy_economics() test_teslemetry_sync_tariff_dedupes_unchanged() test_teslemetry_sync_tariff_pushes_on_window_change() test_teslemetry_sync_tariff_read_only_no_push() @@ -1975,6 +2156,10 @@ def test_teslemetry(my_predbat=None): test_teslemetry_automatic_config_skips_unpublished_rate_sensors() test_teslemetry_mock_base_get_arg_consults_args() test_teslemetry_cli_harness_signals_failure_on_auth_error() + test_teslemetry_cli_harness_wires_oauth_args() + test_teslemetry_api_sets_supabase_env_vars_for_oauth_refresh() + test_teslemetry_load_args_from_apps_yaml_extracts_teslemetry_section() + test_teslemetry_load_args_from_apps_yaml_omits_missing_keys() test_teslemetry_discover_site_uses_first_and_filters() test_teslemetry_discover_site_no_match_returns_false() test_teslemetry_run_discovers_site_before_polling()